harness_codex/

app_server_process.rs

// SPDX-License-Identifier: AGPL-3.0-or-later

//! Owned child-process transport for the Codex whole-agent app-server seam.
//!
//! The transport starts a new process and communicates only through its piped
//! stdio. It never discovers or attaches to an existing app-server or Desktop
//! process and never reads authentication files, keychains, or token values.

use crate::{
    CURRENT_DESKTOP_CODEX_VERSION, CURRENT_STANDALONE_CLI_VERSION, CodexAgentRuntime,
    CodexToolPolicy, ManagedAccountProjection, ModelPageProjection, RateLimitsProjection,
    ReasoningEffort, RuntimeError, RuntimeEvent, TerminalStatus, TurnCompletionReceipt,
};
use rustix::fs::{AtFlags, FileType, Mode, OFlags, Stat, fstat, open, openat, statat};
use serde::Serialize;
use sha2::Digest as _;
use std::collections::VecDeque;
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
#[cfg(unix)]
use std::os::unix::fs::MetadataExt as _;
use std::path::{Component, Path, PathBuf};
use std::process::{Child, ChildStdin, Command, ExitStatus, Stdio};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use thiserror::Error;

const DEFAULT_MAX_LINE_BYTES: usize = 256 * 1024;
const FRAME_CHANNEL_CAPACITY: usize = 64;
const MAX_BUFFERED_EVENTS: usize = 256;
const MAX_IGNORED_NON_SEMANTIC_NOTIFICATIONS: usize = 64;
const MAX_DISCOVERY_PAGES: usize = 8;
const MAX_MCP_SERVER_CONFIG_BYTES: usize = 1024 * 1024;
const MAX_MCP_SERVER_KEYS: usize = 32;
const MAX_MCP_SERVER_KEY_BYTES: usize = 64;
const FIXED_PRODUCTION_ARG_COUNT: usize = 19;
const MAX_PROCESS_ARGS: usize = FIXED_PRODUCTION_ARG_COUNT + (MAX_MCP_SERVER_KEYS * 2);
const MAX_ARG_BYTES: usize = 4096;
const MIN_TIMEOUT: Duration = Duration::from_millis(10);
const MAX_TIMEOUT: Duration = Duration::from_secs(120);
pub const APP_SERVER_DISCOVERY_SCHEMA_VERSION: &str = "1.0.0";
pub const CODEX_APP_SERVER_PROTOCOL_VERSION: &str = "2026-07-29.app-server-jsonl";
const MANAGED_CHILD_ENV_ALLOWLIST: [&str; 9] = [
    "HOME",
    "CODEX_HOME",
    "PATH",
    "TMPDIR",
    "LANG",
    "LC_ALL",
    "LC_CTYPE",
    "SSL_CERT_FILE",
    "SSL_CERT_DIR",
];

/// Fixed, content-free reasons for a fail-closed owned-child configuration
/// boundary.  These are intentionally classes rather than filesystem or
/// configuration details so live evidence can locate the boundary without
/// projecting paths, TOML keys, values, or digests.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PreflightConfigurationRefusal {
    CodexHomeStructural,
    ConfigOpenTypeOwnerModeLink,
    ConfigFileIdentity,
    ConfigContentDigest,
    ConfigSemanticKeys,
    ConfigOverrideDigest,
    ConfigArgCap,
    Other,
}

impl PreflightConfigurationRefusal {
    const fn as_str(self) -> &'static str {
        match self {
            Self::CodexHomeStructural => "codex-home-structural",
            Self::ConfigOpenTypeOwnerModeLink => "config-open-type-owner-mode-link",
            Self::ConfigFileIdentity => "config-file-identity",
            Self::ConfigContentDigest => "config-content-digest",
            Self::ConfigSemanticKeys => "config-semantic-keys",
            Self::ConfigOverrideDigest => "config-override-digest",
            Self::ConfigArgCap => "config-arg-cap",
            Self::Other => "config-other",
        }
    }
}

/// Content-free evidence that the owned child disables every structurally
/// observed Codex MCP server. Server names are intentionally not exposed.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct OwnedChildMcpIsolationAttestation {
    pub configured_server_count: usize,
    pub config_sha256: String,
    pub override_set_sha256: String,
}

/// A bounded line transport consumed by [`CodexAppServerClient`].
pub trait AppServerLineTransport {
    fn send_line(&mut self, line: &str) -> Result<(), AppServerProcessError>;
    fn recv_line(&mut self, timeout: Duration) -> Result<Vec<u8>, AppServerProcessError>;
    fn shutdown(&mut self) -> Result<ShutdownReceipt, AppServerProcessError>;
}

/// Configuration for a newly owned `codex app-server` process.
pub struct OwnedAppServerConfig {
    program: PathBuf,
    executable_identity: ExecutableIdentity,
    expected_executable_sha256: Option<String>,
    args: Vec<OsString>,
    required_app_server_version: Option<String>,
    operation_timeout: Duration,
    shutdown_timeout: Duration,
    max_line_bytes: usize,
    mcp_isolation: Option<McpServerIsolation>,
}

struct McpServerIsolation {
    codex_home: PathBuf,
    codex_home_directory: fs::File,
    // `CODEX_HOME` is a private directory whose metadata legitimately changes
    // when a separately authorized zero-turn qualification writes unrelated
    // state.  Keep its structural identity pinned, but reserve full byte and
    // timestamp pinning for the security-relevant `config.toml` below.
    codex_home_identity: PinnedDirectoryIdentity,
    config_identity: PinnedFilesystemIdentity,
    config_sha256: String,
    configured_server_count: usize,
    server_keys: Vec<String>,
    override_set_sha256: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct PinnedFilesystemIdentity {
    device: u64,
    inode: u64,
    owner_uid: u32,
    mode: u32,
    nlink: u64,
    size: u64,
    modified_seconds: i64,
    modified_nanoseconds: i64,
}

/// The immutable security boundary for the `CODEX_HOME` directory itself.
///
/// Directory size and modification time change when an owned qualification
/// writes an unrelated, owner-private state file.  They are therefore not
/// admission facts for the later owned-child spawn.  The directory descriptor,
/// canonical path, ownership, type/mode, device, and inode remain
/// pinned; `config.toml` retains the full identity and byte-digest binding.
#[derive(Clone, Debug, PartialEq, Eq)]
struct PinnedDirectoryIdentity {
    device: u64,
    inode: u64,
    owner_uid: u32,
    mode: u32,
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct ExecutableIdentity {
    sha256: String,
    #[cfg(unix)]
    device: u64,
    #[cfg(unix)]
    inode: u64,
    #[cfg(unix)]
    owner_uid: u32,
    #[cfg(unix)]
    mode: u32,
    #[cfg(unix)]
    nlink: u64,
    #[cfg(unix)]
    size: u64,
    #[cfg(unix)]
    modified_seconds: i64,
    #[cfg(unix)]
    modified_nanoseconds: i64,
}

impl fmt::Debug for OwnedAppServerConfig {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OwnedAppServerConfig")
            .field("program", &self.program)
            .field("args", &"[REDACTED]")
            .field("operation_timeout", &self.operation_timeout)
            .field("shutdown_timeout", &self.shutdown_timeout)
            .field("max_line_bytes", &self.max_line_bytes)
            .finish()
    }
}

impl OwnedAppServerConfig {
    /// Creates the production owned-child command. Authentication remains under
    /// the child Codex process's normal managed credential lifecycle. The
    /// child always resolves the managed ChatGPT account against the OpenAI
    /// provider so Phase-1 discovery is deterministic even when the user's
    /// `~/.codex/config.toml` defaults `model_provider` to the continuity
    /// router (an OpenAI-compatible local provider that has no ChatGPT
    /// account to report from `account/read`).
    pub fn codex(
        executable: impl Into<PathBuf>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let codex_home = codex_home_from_environment()?;
        Self::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            operation_timeout,
            shutdown_timeout,
        )
    }

    /// Creates the production owned-child command while pinning the exact
    /// Codex configuration directory from which MCP server names are read.
    /// This is public so platform callers can bind the same managed
    /// `CODEX_HOME` directory they pass to the child, without reading any
    /// command, environment, or credential value from its configuration.
    pub fn codex_with_mcp_server_isolation(
        executable: impl Into<PathBuf>,
        codex_home: impl Into<PathBuf>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let isolation = McpServerIsolation::read(codex_home.into())?;
        let mut args = vec![
            OsString::from("app-server"),
            OsString::from("--listen"),
            OsString::from("stdio://"),
            OsString::from("-c"),
            OsString::from("model_provider=\"openai\""),
            OsString::from("-c"),
            OsString::from("web_search=\"disabled\""),
            OsString::from("--disable"),
            OsString::from("hooks"),
            OsString::from("--disable"),
            OsString::from("apps"),
            OsString::from("--disable"),
            OsString::from("multi_agent"),
            OsString::from("--disable"),
            OsString::from("shell_tool"),
            OsString::from("--disable"),
            OsString::from("unified_exec"),
            OsString::from("--disable"),
            OsString::from("standalone_web_search"),
        ];
        args.extend(isolation.server_keys.iter().flat_map(|key| {
            [
                OsString::from("-c"),
                OsString::from(format!("mcp_servers.{key}.enabled=false")),
            ]
        }));
        let mut config = Self::new(
            executable.into(),
            args,
            operation_timeout,
            shutdown_timeout,
            DEFAULT_MAX_LINE_BYTES,
            Some(isolation),
        )?;
        config.required_app_server_version = Some(CURRENT_STANDALONE_CLI_VERSION.to_owned());
        Ok(config)
    }

    /// Creates an owned Codex child with the caller-provided executable digest
    /// as an explicit boundary. The digest is checked at construction and
    /// again immediately before process creation.
    pub fn codex_pinned(
        executable: impl Into<PathBuf>,
        expected_executable_sha256: &str,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let mut config = Self::codex(executable, operation_timeout, shutdown_timeout)?;
        config.require_expected_executable_sha256(expected_executable_sha256)?;
        Ok(config)
    }

    /// Pinned variant of [`Self::codex_with_mcp_server_isolation`].
    pub fn codex_pinned_with_mcp_server_isolation(
        executable: impl Into<PathBuf>,
        expected_executable_sha256: &str,
        codex_home: impl Into<PathBuf>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let mut config = Self::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            operation_timeout,
            shutdown_timeout,
        )?;
        config.require_expected_executable_sha256(expected_executable_sha256)?;
        Ok(config)
    }

    /// Creates the bounded owned-child command for the separately qualified
    /// ChatGPT Desktop app-server executable. The process remains a fresh
    /// owned child; this never attaches to a running Desktop process.
    pub fn codex_desktop(
        executable: impl Into<PathBuf>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let mut config = Self::codex(executable, operation_timeout, shutdown_timeout)?;
        config.required_app_server_version = Some(CURRENT_DESKTOP_CODEX_VERSION.to_owned());
        Ok(config)
    }

    /// Desktop variant of [`Self::codex_pinned`], retaining the separately
    /// reviewed Desktop app-server version boundary.
    pub fn codex_desktop_pinned(
        executable: impl Into<PathBuf>,
        expected_executable_sha256: &str,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let mut config = Self::codex_desktop(executable, operation_timeout, shutdown_timeout)?;
        config.require_expected_executable_sha256(expected_executable_sha256)?;
        Ok(config)
    }

    /// Desktop variant of [`Self::codex_pinned_with_mcp_server_isolation`],
    /// retaining the separately reviewed Desktop app-server version boundary.
    pub fn codex_desktop_pinned_with_mcp_server_isolation(
        executable: impl Into<PathBuf>,
        expected_executable_sha256: &str,
        codex_home: impl Into<PathBuf>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
    ) -> Result<Self, AppServerProcessError> {
        let mut config = Self::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            operation_timeout,
            shutdown_timeout,
        )?;
        config.required_app_server_version = Some(CURRENT_DESKTOP_CODEX_VERSION.to_owned());
        config.require_expected_executable_sha256(expected_executable_sha256)?;
        Ok(config)
    }

    fn new(
        program: PathBuf,
        args: Vec<OsString>,
        operation_timeout: Duration,
        shutdown_timeout: Duration,
        max_line_bytes: usize,
        mcp_isolation: Option<McpServerIsolation>,
    ) -> Result<Self, AppServerProcessError> {
        if !program.is_absolute()
            || args.len() > MAX_PROCESS_ARGS
            || args.iter().any(|arg| !valid_arg(arg))
            || !valid_timeout(operation_timeout)
            || !valid_timeout(shutdown_timeout)
            || !(1024..=DEFAULT_MAX_LINE_BYTES).contains(&max_line_bytes)
        {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        let program = program
            .canonicalize()
            .map_err(|_| AppServerProcessError::InvalidConfiguration)?;
        let executable_identity = capture_executable_identity(&program)?;
        Ok(Self {
            program,
            executable_identity,
            expected_executable_sha256: None,
            args,
            required_app_server_version: None,
            operation_timeout,
            shutdown_timeout,
            max_line_bytes,
            mcp_isolation,
        })
    }

    #[cfg(test)]
    fn fixture(program: &std::path::Path, args: Vec<OsString>, timeout: Duration) -> Self {
        Self::new(
            program.to_path_buf(),
            args,
            timeout,
            Duration::from_millis(250),
            DEFAULT_MAX_LINE_BYTES,
            None,
        )
        .expect("valid fixture child configuration")
    }

    pub const fn operation_timeout(&self) -> Duration {
        self.operation_timeout
    }

    /// The exact standalone CLI protocol version reviewed for this adapter.
    /// Callers must reject a different installed version before live use.
    pub const fn compatible_cli_version(&self) -> &'static str {
        CURRENT_STANDALONE_CLI_VERSION
    }

    fn require_expected_executable_sha256(
        &mut self,
        expected_executable_sha256: &str,
    ) -> Result<(), AppServerProcessError> {
        if !valid_sha256(expected_executable_sha256)
            || self.executable_identity.sha256 != expected_executable_sha256
        {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        self.expected_executable_sha256 = Some(expected_executable_sha256.to_owned());
        Ok(())
    }

    fn revalidate_before_spawn(&self) -> Result<(), AppServerProcessError> {
        let actual = capture_executable_identity(&self.program)?;
        if actual != self.executable_identity
            || self
                .expected_executable_sha256
                .as_deref()
                .is_some_and(|expected| actual.sha256 != expected)
        {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        if let Some(isolation) = &self.mcp_isolation {
            isolation.revalidate()?;
        }
        Ok(())
    }

    /// Returns only sealed MCP isolation facts. Names and configuration values
    /// never cross this boundary.
    pub fn mcp_isolation_attestation(&self) -> Option<OwnedChildMcpIsolationAttestation> {
        self.mcp_isolation
            .as_ref()
            .map(|isolation| isolation.attestation())
    }

    fn managed_codex_home(&self) -> Option<&Path> {
        self.mcp_isolation
            .as_ref()
            .map(|isolation| isolation.codex_home.as_path())
    }
}

impl McpServerIsolation {
    fn read(codex_home: PathBuf) -> Result<Self, AppServerProcessError> {
        let codex_home_directory = open_owned_directory(&codex_home).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;
        let codex_home_stat = fstat(&codex_home_directory).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;
        validate_owned_directory(&codex_home_stat).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;
        let codex_home_identity = pinned_directory_identity(&codex_home_stat).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;

        let config = fs::File::from(
            openat(
                &codex_home_directory,
                "config.toml",
                OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                Mode::empty(),
            )
            .map_err(|_| {
                configuration_refusal(PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink)
            })?,
        );
        let before = fstat(&config).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink)
        })?;
        let config_identity = pinned_identity(&before).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink)
        })?;
        validate_owned_regular_file(&before, Some(MAX_MCP_SERVER_CONFIG_BYTES)).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink)
        })?;
        let named_before = statat(
            &codex_home_directory,
            "config.toml",
            AtFlags::SYMLINK_NOFOLLOW,
        )
        .map_err(|_| configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity))?;
        if pinned_identity(&named_before)
            .map_err(|_| configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity))?
            != config_identity
        {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigFileIdentity,
            ));
        }

        let mut bytes = Vec::with_capacity(usize::try_from(before.st_size).unwrap_or(0));
        (&config)
            .take((MAX_MCP_SERVER_CONFIG_BYTES + 1) as u64)
            .read_to_end(&mut bytes)
            .map_err(|_| {
                configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity)
            })?;
        let after = fstat(&config).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity)
        })?;
        let named_after = statat(
            &codex_home_directory,
            "config.toml",
            AtFlags::SYMLINK_NOFOLLOW,
        )
        .map_err(|_| configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity))?;
        if bytes.len() > MAX_MCP_SERVER_CONFIG_BYTES
            || pinned_identity(&after).map_err(|_| {
                configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity)
            })? != config_identity
            || pinned_identity(&named_after).map_err(|_| {
                configuration_refusal(PreflightConfigurationRefusal::ConfigFileIdentity)
            })? != config_identity
        {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigFileIdentity,
            ));
        }

        let server_keys = parse_mcp_server_keys(&bytes)?;
        let configured_server_count = server_keys.len();
        let config_sha256 = sha256_bytes(&bytes);
        let override_set_sha256 = mcp_override_set_sha256(&server_keys);
        Ok(Self {
            codex_home,
            codex_home_directory,
            codex_home_identity,
            config_identity,
            config_sha256,
            configured_server_count,
            server_keys,
            override_set_sha256,
        })
    }

    fn revalidate(&self) -> Result<(), AppServerProcessError> {
        let directory = fstat(&self.codex_home_directory).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;
        validate_owned_directory(&directory).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })?;
        if pinned_directory_identity(&directory).map_err(|_| {
            configuration_refusal(PreflightConfigurationRefusal::CodexHomeStructural)
        })? != self.codex_home_identity
        {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::CodexHomeStructural,
            ));
        }

        let fresh = Self::read(self.codex_home.clone())?;
        if fresh.codex_home_identity != self.codex_home_identity {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::CodexHomeStructural,
            ));
        }
        if fresh.config_identity != self.config_identity {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigFileIdentity,
            ));
        }
        if fresh.config_sha256 != self.config_sha256 {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigContentDigest,
            ));
        }
        if fresh.server_keys != self.server_keys
            || fresh.configured_server_count != self.configured_server_count
        {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigSemanticKeys,
            ));
        }
        if fresh.override_set_sha256 != self.override_set_sha256 {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigOverrideDigest,
            ));
        }
        Ok(())
    }

    fn attestation(&self) -> OwnedChildMcpIsolationAttestation {
        OwnedChildMcpIsolationAttestation {
            configured_server_count: self.configured_server_count,
            config_sha256: self.config_sha256.clone(),
            override_set_sha256: self.override_set_sha256.clone(),
        }
    }
}

fn codex_home_from_environment() -> Result<PathBuf, AppServerProcessError> {
    let configured = std::env::var_os("CODEX_HOME").filter(|value| !value.is_empty());
    let home = match configured {
        Some(path) => PathBuf::from(path),
        None => PathBuf::from(
            std::env::var_os("HOME").ok_or(AppServerProcessError::InvalidConfiguration)?,
        )
        .join(".codex"),
    };
    if !home.is_absolute() {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    Ok(home)
}

fn open_owned_directory(path: &Path) -> Result<fs::File, AppServerProcessError> {
    if !path.is_absolute()
        || path
            .components()
            .any(|component| matches!(component, Component::ParentDir | Component::CurDir))
    {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    let mut current = fs::File::from(
        open(
            "/",
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
            Mode::empty(),
        )
        .map_err(invalid)?,
    );
    for component in path.components() {
        let Component::Normal(component) = component else {
            continue;
        };
        current = fs::File::from(
            openat(
                &current,
                component,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                Mode::empty(),
            )
            .map_err(invalid)?,
        );
    }
    validate_owned_directory(&fstat(&current).map_err(invalid)?)?;
    Ok(current)
}

fn validate_owned_directory(stat: &Stat) -> Result<(), AppServerProcessError> {
    #[cfg(unix)]
    if FileType::from_raw_mode(stat.st_mode) != FileType::Directory
        || stat.st_uid != rustix::process::geteuid().as_raw()
        || stat.st_mode & 0o022 != 0
    {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    #[cfg(not(unix))]
    return Err(AppServerProcessError::InvalidConfiguration);
    Ok(())
}

fn validate_owned_regular_file(
    stat: &Stat,
    maximum_size: Option<usize>,
) -> Result<(), AppServerProcessError> {
    #[cfg(unix)]
    if FileType::from_raw_mode(stat.st_mode) != FileType::RegularFile
        || stat.st_uid != rustix::process::geteuid().as_raw()
        || stat.st_mode & 0o022 != 0
        || stat.st_nlink != 1
        || maximum_size.is_some_and(|maximum| stat.st_size > maximum as _)
    {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    #[cfg(not(unix))]
    return Err(AppServerProcessError::InvalidConfiguration);
    Ok(())
}

// rustix Stat field widths differ by platform: Linux reports device as u64,
// mode as u32, and link count as u64, while macOS reports i32, u16, and u16.
// The pinned identity commits fixed widths (u64/u32/u64).  These helpers
// absorb the platform width so the direct-width platform takes no conversion
// (a `try_from`/`from` that is a no-op is a clippy `useless_conversion`).

#[cfg(target_os = "linux")]
fn identity_device(stat: &Stat) -> Result<u64, AppServerProcessError> {
    Ok(stat.st_dev)
}

#[cfg(not(target_os = "linux"))]
fn identity_device(stat: &Stat) -> Result<u64, AppServerProcessError> {
    u64::try_from(stat.st_dev).map_err(|_| AppServerProcessError::InvalidConfiguration)
}

#[cfg(target_os = "linux")]
fn identity_mode(stat: &Stat) -> u32 {
    stat.st_mode
}

#[cfg(not(target_os = "linux"))]
fn identity_mode(stat: &Stat) -> u32 {
    u32::from(stat.st_mode)
}

#[cfg(target_os = "linux")]
fn identity_nlink(stat: &Stat) -> u64 {
    stat.st_nlink
}

#[cfg(not(target_os = "linux"))]
fn identity_nlink(stat: &Stat) -> u64 {
    u64::from(stat.st_nlink)
}

fn pinned_identity(stat: &Stat) -> Result<PinnedFilesystemIdentity, AppServerProcessError> {
    // rustix types st_mtime_nsec as u64 on Linux and i64 on macOS; the value
    // is always a sub-second nanosecond count, so the conversion is lossless.
    #[cfg(target_os = "linux")]
    let modified_nanoseconds = i64::try_from(stat.st_mtime_nsec)
        .map_err(|_| AppServerProcessError::InvalidConfiguration)?;
    #[cfg(not(target_os = "linux"))]
    let modified_nanoseconds = stat.st_mtime_nsec;
    Ok(PinnedFilesystemIdentity {
        device: identity_device(stat)?,
        inode: stat.st_ino,
        owner_uid: stat.st_uid,
        mode: identity_mode(stat),
        nlink: identity_nlink(stat),
        size: u64::try_from(stat.st_size)
            .map_err(|_| AppServerProcessError::InvalidConfiguration)?,
        modified_seconds: stat.st_mtime,
        modified_nanoseconds,
    })
}

fn pinned_directory_identity(
    stat: &Stat,
) -> Result<PinnedDirectoryIdentity, AppServerProcessError> {
    Ok(PinnedDirectoryIdentity {
        device: identity_device(stat)?,
        inode: stat.st_ino,
        owner_uid: stat.st_uid,
        // The raw mode includes the directory file-type bits.  Callers first
        // validate that this is a private directory, so equality commits both
        // type and permissions without treating mutable timestamps as identity.
        mode: identity_mode(stat),
    })
}

fn parse_mcp_server_keys(bytes: &[u8]) -> Result<Vec<String>, AppServerProcessError> {
    let text = std::str::from_utf8(bytes)
        .map_err(|_| configuration_refusal(PreflightConfigurationRefusal::ConfigSemanticKeys))?;
    // Parse the already descriptor-pinned bytes using the same TOML grammar
    // accepted by Codex. A literal-header scan misses dotted assignments,
    // inline parent tables, quoted keys, and whitespace-equivalent headers.
    // We inspect only `mcp_servers` shape and keys; command/env/credential
    // values are never projected into an argument, receipt, or diagnostic.
    let root = toml::from_str::<toml::Value>(text)
        .map_err(|_| configuration_refusal(PreflightConfigurationRefusal::ConfigSemanticKeys))?;
    let root = root
        .as_table()
        .ok_or_else(|| configuration_refusal(PreflightConfigurationRefusal::ConfigSemanticKeys))?;
    let Some(servers) = root.get("mcp_servers") else {
        return Ok(Vec::new());
    };
    let servers = servers
        .as_table()
        .ok_or_else(|| configuration_refusal(PreflightConfigurationRefusal::ConfigSemanticKeys))?;
    if servers.len() > MAX_MCP_SERVER_KEYS {
        return Err(configuration_refusal(
            PreflightConfigurationRefusal::ConfigArgCap,
        ));
    }
    let mut keys = Vec::with_capacity(servers.len());
    for (key, server) in servers {
        if !valid_mcp_server_key(key) || !server.is_table() {
            return Err(configuration_refusal(
                PreflightConfigurationRefusal::ConfigSemanticKeys,
            ));
        }
        keys.push(key.clone());
    }
    keys.sort_unstable();
    keys.dedup();
    Ok(keys)
}

fn valid_mcp_server_key(key: &str) -> bool {
    !key.is_empty()
        && key.len() <= MAX_MCP_SERVER_KEY_BYTES
        && key
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
}

fn mcp_override_set_sha256(server_keys: &[String]) -> String {
    let mut digest = sha2::Sha256::new();
    digest.update(b"steadymux.owned-child.mcp-disable-overrides.v1\0");
    for key in server_keys {
        digest.update(key.as_bytes());
        digest.update([0]);
    }
    hex_digest(digest.finalize())
}

fn sha256_bytes(bytes: &[u8]) -> String {
    hex_digest(sha2::Sha256::digest(bytes))
}

fn hex_digest(digest: impl AsRef<[u8]>) -> String {
    let mut output = String::with_capacity(64);
    for byte in digest.as_ref() {
        use std::fmt::Write as _;
        write!(&mut output, "{byte:02x}").expect("writing to String cannot fail");
    }
    output
}

fn invalid(_: impl std::fmt::Debug) -> AppServerProcessError {
    AppServerProcessError::InvalidConfiguration
}

fn configuration_refusal(class: PreflightConfigurationRefusal) -> AppServerProcessError {
    AppServerProcessError::PreflightConfiguration(class)
}

fn capture_executable_identity(
    program: &std::path::Path,
) -> Result<ExecutableIdentity, AppServerProcessError> {
    let before = fs::metadata(program).map_err(|_| AppServerProcessError::InvalidConfiguration)?;
    if !before.is_file() {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    #[cfg(unix)]
    {
        let current_uid = rustix::process::geteuid().as_raw();
        if before.nlink() != 1
            || (before.uid() != 0 && before.uid() != current_uid)
            || before.mode() & 0o022 != 0
            || before.mode() & 0o111 == 0
        {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
    }
    #[cfg(not(unix))]
    {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    let sha256 = sha256_file(program)?;
    let after = fs::metadata(program).map_err(|_| AppServerProcessError::InvalidConfiguration)?;
    #[cfg(unix)]
    if before.dev() != after.dev()
        || before.ino() != after.ino()
        || before.uid() != after.uid()
        || before.mode() != after.mode()
        || before.nlink() != after.nlink()
        || before.len() != after.len()
        || before.mtime() != after.mtime()
        || before.mtime_nsec() != after.mtime_nsec()
    {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    Ok(ExecutableIdentity {
        sha256,
        #[cfg(unix)]
        device: after.dev(),
        #[cfg(unix)]
        inode: after.ino(),
        #[cfg(unix)]
        owner_uid: after.uid(),
        #[cfg(unix)]
        mode: after.mode(),
        #[cfg(unix)]
        nlink: after.nlink(),
        #[cfg(unix)]
        size: after.len(),
        #[cfg(unix)]
        modified_seconds: after.mtime(),
        #[cfg(unix)]
        modified_nanoseconds: after.mtime_nsec(),
    })
}

fn valid_sha256(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
}

fn sha256_file(program: &std::path::Path) -> Result<String, AppServerProcessError> {
    let mut file =
        fs::File::open(program).map_err(|_| AppServerProcessError::InvalidConfiguration)?;
    let mut digest = sha2::Sha256::new();
    let mut buffer = [0_u8; 64 * 1024];
    loop {
        let read = file
            .read(&mut buffer)
            .map_err(|_| AppServerProcessError::InvalidConfiguration)?;
        if read == 0 {
            break;
        }
        digest.update(&buffer[..read]);
    }
    let mut output = String::with_capacity(64);
    for byte in digest.finalize() {
        use std::fmt::Write as _;
        write!(&mut output, "{byte:02x}").expect("writing to String cannot fail");
    }
    Ok(output)
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ShutdownReceipt {
    pub forced: bool,
    pub success: bool,
}

#[derive(Debug, Error, PartialEq, Eq)]
pub enum AppServerProcessError {
    #[error("the owned app-server configuration is invalid")]
    InvalidConfiguration,
    #[error("the owned app-server configuration was refused before spawn")]
    PreflightConfiguration(PreflightConfigurationRefusal),
    #[error("the owned app-server process could not be started")]
    SpawnFailed,
    #[error("the owned app-server executable was unavailable")]
    SpawnExecutableUnavailable,
    #[error("the owned app-server executable was not permitted")]
    SpawnExecutablePermissionDenied,
    #[error("the owned app-server stdio could not be established")]
    SpawnStdioUnavailable,
    #[error("the owned app-server reader task could not be established")]
    SpawnReaderUnavailable,
    #[error("the Codex executable version is not supported by this adapter")]
    UnsupportedExecutableVersion,
    #[error("the owned app-server transport failed")]
    TransportFailed,
    #[error("the owned app-server emitted an oversized frame")]
    FrameTooLarge,
    #[error("the owned app-server closed its stream unexpectedly")]
    UnexpectedEof,
    #[error("the owned app-server operation timed out")]
    Timeout,
    #[error("the owned app-server exited before completing the operation")]
    ProcessExited,
    #[error("the turn outcome is ambiguous after a request write and replay is forbidden")]
    AmbiguousAfterWriteNoReplay,
    #[error("the app-server rejected the JSON-RPC request after write; replay is forbidden")]
    PostWriteJsonRpcRejectedNoReplay,
    #[error("the app-server emitted a side-effecting item after write; replay is forbidden")]
    PostWriteSideEffectRejectedNoReplay,
    #[error(
        "the app-server notification or protocol state mismatched after write; replay is forbidden"
    )]
    PostWriteNotificationOrStateMismatchNoReplay,
    #[error("the app-server timed out after request write; replay is forbidden")]
    PostWriteTimeoutNoReplay,
    #[error("the app-server stream ended after request write; replay is forbidden")]
    PostWriteUnexpectedEofNoReplay,
    #[error("the app-server transport failed after request write; replay is forbidden")]
    PostWriteTransportNoReplay,
    #[error("the app-server capacity bound was exceeded after request write; replay is forbidden")]
    PostWriteCapacityExceededNoReplay,
    #[error("the app-server protocol state failed")]
    Runtime(#[from] RuntimeError),
    #[error("the app-server selected a different model than requested")]
    ModelRerouted,
    #[error("the owned app-server shutdown could not be completed")]
    ShutdownFailed,
    #[error("the app-server discovery response exceeded configured page bounds")]
    DiscoveryPageLimit,
    #[error("the app-server event buffer is exhausted")]
    EventCapacityExceeded,
    #[error("the app-server terminal status does not match the requested operation")]
    UnexpectedTerminalStatus,
}

impl AppServerProcessError {
    /// Fixed, content-free class for the owned-child preflight boundary.
    ///
    /// This deliberately never includes a path, argv element, operating-system
    /// error text, child output, or configuration fact.  It exists solely to
    /// distinguish the fail-closed boundaries needed to diagnose an owned
    /// child that could not reach JSON-RPC initialization.
    pub const fn preflight_refusal_class(&self) -> &'static str {
        match self {
            Self::InvalidConfiguration => PreflightConfigurationRefusal::Other.as_str(),
            Self::PreflightConfiguration(class) => class.as_str(),
            Self::SpawnExecutableUnavailable => "command-not-found",
            Self::SpawnExecutablePermissionDenied => "command-permission",
            Self::SpawnFailed => "command-other",
            Self::SpawnStdioUnavailable => "stdio",
            Self::SpawnReaderUnavailable => "reader",
            Self::UnsupportedExecutableVersion => "version",
            Self::ProcessExited | Self::UnexpectedEof => "child-exit",
            Self::Timeout => "timeout",
            Self::TransportFailed | Self::FrameTooLarge => "transport",
            Self::EventCapacityExceeded => "capacity",
            Self::ShutdownFailed => "shutdown",
            Self::Runtime(_) => "protocol",
            Self::AmbiguousAfterWriteNoReplay
            | Self::PostWriteJsonRpcRejectedNoReplay
            | Self::PostWriteSideEffectRejectedNoReplay
            | Self::PostWriteNotificationOrStateMismatchNoReplay
            | Self::PostWriteTimeoutNoReplay
            | Self::PostWriteUnexpectedEofNoReplay
            | Self::PostWriteTransportNoReplay
            | Self::PostWriteCapacityExceededNoReplay
            | Self::ModelRerouted
            | Self::DiscoveryPageLimit
            | Self::UnexpectedTerminalStatus => "postwrite-or-terminal",
        }
    }
}

enum ProcessFrame {
    Line(Vec<u8>),
    Eof,
    TooLarge,
    Failed,
}

/// A newly spawned app-server child with exclusively owned stdio handles.
pub struct OwnedAppServerTransport {
    child: Option<Child>,
    stdin: Option<ChildStdin>,
    frames: Receiver<ProcessFrame>,
    stdout_task: Option<JoinHandle<()>>,
    stderr_task: Option<JoinHandle<()>>,
    shutdown_timeout: Duration,
    max_line_bytes: usize,
    mcp_isolation_attestation: Option<OwnedChildMcpIsolationAttestation>,
    shutdown_receipt: Option<ShutdownReceipt>,
}

impl fmt::Debug for OwnedAppServerTransport {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("OwnedAppServerTransport")
            .field("child_id", &self.child.as_ref().map(Child::id))
            .field("stdio", &"[OWNED]")
            .field("stopped", &self.shutdown_receipt.is_some())
            .finish()
    }
}

impl OwnedAppServerTransport {
    pub fn spawn(config: &OwnedAppServerConfig) -> Result<Self, AppServerProcessError> {
        config.revalidate_before_spawn()?;
        let mcp_isolation_attestation = config.mcp_isolation_attestation();
        let mut command = Command::new(&config.program);
        command
            .args(&config.args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        configure_managed_child_env(&mut command, config.managed_codex_home());

        let mut child = command.spawn().map_err(|error| match error.kind() {
            std::io::ErrorKind::NotFound => AppServerProcessError::SpawnExecutableUnavailable,
            std::io::ErrorKind::PermissionDenied => {
                AppServerProcessError::SpawnExecutablePermissionDenied
            }
            _ => AppServerProcessError::SpawnFailed,
        })?;
        let Some(stdin) = child.stdin.take() else {
            kill_and_reap(&mut child);
            return Err(AppServerProcessError::SpawnStdioUnavailable);
        };
        let Some(stdout) = child.stdout.take() else {
            kill_and_reap(&mut child);
            return Err(AppServerProcessError::SpawnStdioUnavailable);
        };
        let Some(stderr) = child.stderr.take() else {
            kill_and_reap(&mut child);
            return Err(AppServerProcessError::SpawnStdioUnavailable);
        };
        let (sender, frames) = mpsc::sync_channel(FRAME_CHANNEL_CAPACITY);
        let max_line_bytes = config.max_line_bytes;
        let stdout_task = match thread::Builder::new()
            .name("steadymux-codex-app-server-stdout".to_owned())
            .spawn(move || read_stdout(stdout, sender, max_line_bytes))
        {
            Ok(task) => task,
            Err(_) => {
                kill_and_reap(&mut child);
                return Err(AppServerProcessError::SpawnReaderUnavailable);
            }
        };
        let stderr_task = match thread::Builder::new()
            .name("steadymux-codex-app-server-stderr".to_owned())
            .spawn(move || discard_stderr(stderr))
        {
            Ok(task) => task,
            Err(_) => {
                kill_and_reap(&mut child);
                let _ = stdout_task.join();
                return Err(AppServerProcessError::SpawnReaderUnavailable);
            }
        };
        Ok(Self {
            child: Some(child),
            stdin: Some(stdin),
            frames,
            stdout_task: Some(stdout_task),
            stderr_task: Some(stderr_task),
            shutdown_timeout: config.shutdown_timeout,
            max_line_bytes: config.max_line_bytes,
            mcp_isolation_attestation,
            shutdown_receipt: None,
        })
    }

    /// Returns the sealed MCP isolation state revalidated immediately before
    /// this owned child was created. It intentionally excludes server names
    /// and all Codex configuration values.
    pub fn mcp_isolation_attestation(&self) -> Option<&OwnedChildMcpIsolationAttestation> {
        self.mcp_isolation_attestation.as_ref()
    }

    pub fn child_id(&self) -> Option<u32> {
        self.child.as_ref().map(Child::id)
    }

    fn child_exited(&mut self) -> Result<bool, AppServerProcessError> {
        self.child
            .as_mut()
            .ok_or(AppServerProcessError::ProcessExited)?
            .try_wait()
            .map(|status| status.is_some())
            .map_err(|_| AppServerProcessError::TransportFailed)
    }

    fn stop_child(&mut self) -> Result<ShutdownReceipt, AppServerProcessError> {
        if let Some(receipt) = self.shutdown_receipt {
            return Ok(receipt);
        }
        self.stdin.take();
        let deadline = Instant::now() + self.shutdown_timeout;
        let mut forced = false;
        let status = loop {
            let child = self
                .child
                .as_mut()
                .ok_or(AppServerProcessError::ShutdownFailed)?;
            match child
                .try_wait()
                .map_err(|_| AppServerProcessError::ShutdownFailed)?
            {
                Some(status) => break status,
                None if Instant::now() < deadline => thread::sleep(Duration::from_millis(5)),
                None => {
                    forced = true;
                    child
                        .kill()
                        .map_err(|_| AppServerProcessError::ShutdownFailed)?;
                    break child
                        .wait()
                        .map_err(|_| AppServerProcessError::ShutdownFailed)?;
                }
            }
        };
        self.child.take();
        let reader_deadline = Instant::now() + self.shutdown_timeout;
        let stdout_stopped = join_reader_until(self.stdout_task.take(), reader_deadline);
        let stderr_stopped = join_reader_until(self.stderr_task.take(), reader_deadline);
        let readers_stopped = stdout_stopped && stderr_stopped;
        let receipt = ShutdownReceipt {
            forced,
            success: exit_is_success_or_forced(status, forced) && readers_stopped,
        };
        self.shutdown_receipt = Some(receipt);
        Ok(receipt)
    }
}

impl AppServerLineTransport for OwnedAppServerTransport {
    fn send_line(&mut self, line: &str) -> Result<(), AppServerProcessError> {
        let Some(body) = line.strip_suffix('\n') else {
            return Err(AppServerProcessError::TransportFailed);
        };
        if self.shutdown_receipt.is_some()
            || body.is_empty()
            || body.len() > self.max_line_bytes
            || body.bytes().any(|byte| matches!(byte, b'\n' | b'\r' | 0))
        {
            return Err(AppServerProcessError::TransportFailed);
        }
        if self.child_exited()? {
            return Err(AppServerProcessError::ProcessExited);
        }
        let stdin = self
            .stdin
            .as_mut()
            .ok_or(AppServerProcessError::ProcessExited)?;
        stdin
            .write_all(line.as_bytes())
            .and_then(|()| stdin.flush())
            .map_err(|_| AppServerProcessError::TransportFailed)
    }

    fn recv_line(&mut self, timeout: Duration) -> Result<Vec<u8>, AppServerProcessError> {
        if timeout.is_zero() || timeout > MAX_TIMEOUT {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        match self.frames.recv_timeout(timeout) {
            Ok(ProcessFrame::Line(line)) => Ok(line),
            Ok(ProcessFrame::Eof) => Err(AppServerProcessError::UnexpectedEof),
            Ok(ProcessFrame::TooLarge) => Err(AppServerProcessError::FrameTooLarge),
            Ok(ProcessFrame::Failed) => Err(AppServerProcessError::TransportFailed),
            Err(RecvTimeoutError::Timeout) => Err(AppServerProcessError::Timeout),
            Err(RecvTimeoutError::Disconnected) => Err(AppServerProcessError::UnexpectedEof),
        }
    }

    fn shutdown(&mut self) -> Result<ShutdownReceipt, AppServerProcessError> {
        self.stop_child()
    }
}

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

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct AppServerDiscoveryReceipt {
    pub schema_version: String,
    pub required_cli_version: String,
    pub protocol_version: String,
    pub account: ManagedAccountProjection,
    pub model_pages: Vec<ModelPageProjection>,
    pub rate_limits: RateLimitsProjection,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ActiveCodexTurn {
    pub thread_id: String,
    pub turn_id: String,
    pub model: String,
    pub effort: ReasoningEffort,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TurnRunReceipt {
    pub events: Vec<RuntimeEvent>,
    pub completion: TurnCompletionReceipt,
}

/// Drives the pure transcript state machine through an injected line transport.
pub struct CodexAppServerClient<T: AppServerLineTransport> {
    runtime: CodexAgentRuntime,
    transport: T,
    operation_timeout: Duration,
    buffered: VecDeque<RuntimeEvent>,
    required_cli_version: Option<String>,
    #[cfg(test)]
    turn_operation_timeout: Option<Duration>,
}

impl<T: AppServerLineTransport> fmt::Debug for CodexAppServerClient<T> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("CodexAppServerClient")
            .field("runtime", &self.runtime)
            .field("transport", &"[REDACTED]")
            .field("operation_timeout", &self.operation_timeout)
            .field("buffered_events", &self.buffered.len())
            .finish()
    }
}

impl CodexAppServerClient<OwnedAppServerTransport> {
    pub fn spawn(config: &OwnedAppServerConfig) -> Result<Self, AppServerProcessError> {
        Self::spawn_with_tool_policy(config, CodexToolPolicy::RejectAll)
    }

    pub fn spawn_with_tool_policy(
        config: &OwnedAppServerConfig,
        tool_policy: CodexToolPolicy,
    ) -> Result<Self, AppServerProcessError> {
        let transport = OwnedAppServerTransport::spawn(config)?;
        let mut client =
            Self::new_with_tool_policy(transport, config.operation_timeout, tool_policy)?;
        if let Some(version) = &config.required_app_server_version {
            client.required_cli_version = Some(version.clone());
        }
        Ok(client)
    }
}

impl<T: AppServerLineTransport> CodexAppServerClient<T> {
    pub fn new(transport: T, operation_timeout: Duration) -> Result<Self, AppServerProcessError> {
        Self::new_with_tool_policy(transport, operation_timeout, CodexToolPolicy::RejectAll)
    }

    pub fn new_with_tool_policy(
        transport: T,
        operation_timeout: Duration,
        tool_policy: CodexToolPolicy,
    ) -> Result<Self, AppServerProcessError> {
        if !valid_timeout(operation_timeout) {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        Ok(Self {
            runtime: CodexAgentRuntime::with_tool_policy(tool_policy),
            transport,
            operation_timeout,
            buffered: VecDeque::new(),
            required_cli_version: None,
            #[cfg(test)]
            turn_operation_timeout: None,
        })
    }

    /// Require the exact reviewed standalone CLI version during initialize.
    /// Injected transcript transports use this same guard as owned processes.
    pub fn require_reviewed_cli_version(&mut self) {
        self.required_cli_version = Some(CURRENT_STANDALONE_CLI_VERSION.to_owned());
    }

    /// Requires the already pinned executable's reviewed app-server version.
    /// This is intentionally a closed string supplied by the owned-child
    /// configuration, never by a provider response or request payload.
    pub fn require_app_server_version(
        &mut self,
        version: &str,
    ) -> Result<(), AppServerProcessError> {
        if version.is_empty() || version.len() > 128 || version.chars().any(char::is_control) {
            return Err(AppServerProcessError::InvalidConfiguration);
        }
        self.required_cli_version = Some(version.to_owned());
        Ok(())
    }

    pub fn initialize(&mut self) -> Result<(), AppServerProcessError> {
        let request = self.runtime.initialize_request()?;
        self.send_and_wait(&request, |event| {
            matches!(event, RuntimeEvent::InitializeAccepted)
        })?;
        if self
            .required_cli_version
            .as_deref()
            .is_some_and(|required| self.runtime.initialized_server_version() != Some(required))
        {
            return self.fail(AppServerProcessError::UnsupportedExecutableVersion);
        }
        let notification = self.runtime.initialized_notification()?;
        match self.transport.send_line(&notification) {
            Ok(()) => Ok(()),
            Err(error) => self.fail(error),
        }
    }

    pub fn discover(&mut self) -> Result<AppServerDiscoveryReceipt, AppServerProcessError> {
        let request = self.runtime.account_read_request()?;
        let account = match self.send_and_wait(&request, |event| {
            matches!(event, RuntimeEvent::ManagedAccount(_))
        })? {
            RuntimeEvent::ManagedAccount(account) => account,
            _ => {
                return Err(AppServerProcessError::Runtime(
                    RuntimeError::ProtocolFailure,
                ));
            }
        };

        let mut model_pages = Vec::new();
        let mut cursor = None;
        loop {
            if model_pages.len() >= MAX_DISCOVERY_PAGES {
                return self.fail(AppServerProcessError::DiscoveryPageLimit);
            }
            let request = self.runtime.model_list_request(cursor.as_deref())?;
            let page = match self.send_and_wait(&request, |event| {
                matches!(event, RuntimeEvent::ModelPage(_))
            })? {
                RuntimeEvent::ModelPage(page) => page,
                _ => {
                    return Err(AppServerProcessError::Runtime(
                        RuntimeError::ProtocolFailure,
                    ));
                }
            };
            cursor.clone_from(&page.next_cursor);
            model_pages.push(page);
            if cursor.is_none() {
                break;
            }
        }

        let request = self.runtime.rate_limits_read_request()?;
        let rate_limits = match self.send_and_wait(&request, |event| {
            matches!(event, RuntimeEvent::RateLimits(_))
        })? {
            RuntimeEvent::RateLimits(rate_limits) => rate_limits,
            _ => {
                return Err(AppServerProcessError::Runtime(
                    RuntimeError::ProtocolFailure,
                ));
            }
        };
        Ok(AppServerDiscoveryReceipt {
            schema_version: APP_SERVER_DISCOVERY_SCHEMA_VERSION.to_owned(),
            // The initialize response is the only version value that has
            // actually crossed the owned transport and passed the configured
            // version guard. Never report a global default here: Desktop and
            // standalone owned children have distinct reviewed versions.
            required_cli_version: self
                .runtime
                .initialized_server_version()
                .ok_or(AppServerProcessError::Runtime(
                    RuntimeError::ProtocolFailure,
                ))?
                .to_owned(),
            protocol_version: CODEX_APP_SERVER_PROTOCOL_VERSION.to_owned(),
            account,
            model_pages,
            rate_limits,
        })
    }

    pub fn start_turn(
        &mut self,
        client_message_id: &str,
        text: &str,
        model: &str,
        effort: ReasoningEffort,
    ) -> Result<ActiveCodexTurn, AppServerProcessError> {
        self.start_turn_with_instructions(client_message_id, text, None, None, model, effort)
    }

    pub fn start_turn_with_instructions(
        &mut self,
        client_message_id: &str,
        text: &str,
        base_instructions: Option<&str>,
        developer_instructions: Option<&str>,
        model: &str,
        effort: ReasoningEffort,
    ) -> Result<ActiveCodexTurn, AppServerProcessError> {
        eprintln!("steadymux-owned-child: stage=thread-request");
        let request = self.runtime.thread_start_request_with_instructions(
            model,
            effort,
            base_instructions,
            developer_instructions,
        )?;
        eprintln!("steadymux-owned-child: stage=thread-wait");
        let thread_id = match self.send_and_wait(&request, |event| {
            matches!(event, RuntimeEvent::ThreadStarted { .. })
        })? {
            RuntimeEvent::ThreadStarted { thread_id } => thread_id,
            _ => {
                return Err(AppServerProcessError::Runtime(
                    RuntimeError::ProtocolFailure,
                ));
            }
        };
        eprintln!("steadymux-owned-child: stage=thread-accepted");
        // Test seam: timeout fixtures run thread/start under the fixture's
        // generous setup window, then clamp the operation timeout to the tight
        // value only for the turn/start wait. This keeps the wall-clock setup
        // latency entirely outside the timeout under test.
        #[cfg(test)]
        if let Some(turn_timeout) = self.turn_operation_timeout {
            self.operation_timeout = turn_timeout;
        }
        eprintln!("steadymux-owned-child: stage=turn-request");
        let request =
            self.runtime
                .turn_start_request(&thread_id, client_message_id, text, model, effort)?;
        eprintln!("steadymux-owned-child: stage=turn-wait");
        let turn_response = match self.send_and_wait(&request, |event| {
            matches!(event, RuntimeEvent::TurnAccepted { .. })
        }) {
            Ok(event) => event,
            Err(error) => return self.fail(classify_post_write_failure(error)),
        };
        let turn_id = match turn_response {
            RuntimeEvent::TurnAccepted {
                thread_id: response_thread,
                turn_id,
            } if response_thread == thread_id => turn_id,
            _ => {
                return self
                    .fail(AppServerProcessError::PostWriteNotificationOrStateMismatchNoReplay);
            }
        };
        Ok(ActiveCodexTurn {
            thread_id,
            turn_id,
            model: model.to_owned(),
            effort,
        })
    }

    pub fn wait_for_terminal(
        &mut self,
        turn: &ActiveCodexTurn,
    ) -> Result<TurnRunReceipt, AppServerProcessError> {
        let mut events = Vec::new();
        let deadline = Instant::now() + self.operation_timeout;
        loop {
            let event = match self.pop_buffered_for_turn(turn) {
                Some(event) => event,
                None => match self.receive_event_until(deadline) {
                    Ok(event) => event,
                    Err(error) => {
                        return self.fail(classify_post_write_failure(error));
                    }
                },
            };
            if event_is_reroute_for(&event, turn) {
                return self.fail(AppServerProcessError::ModelRerouted);
            }
            if event_matches_turn(&event, turn) {
                if let RuntimeEvent::TurnCompleted(completion) = &event {
                    let completion = completion.clone();
                    events.push(event);
                    return Ok(TurnRunReceipt { events, completion });
                }
                if events.len() >= MAX_BUFFERED_EVENTS {
                    return self.fail(AppServerProcessError::PostWriteCapacityExceededNoReplay);
                }
                events.push(event);
            } else {
                if let Err(error) = self.push_buffered(event) {
                    return self.fail(classify_post_write_failure(error));
                }
            }
        }
    }

    /// Wait for a terminal while honoring the caller's one cancellation
    /// authority. Cancellation after turn/start sends `turn/interrupt` on the
    /// same owned client and joins the terminal lifecycle before returning.
    pub fn wait_for_terminal_with_cancellation(
        &mut self,
        turn: &ActiveCodexTurn,
        cancelled: impl Fn() -> bool,
    ) -> Result<TurnRunReceipt, AppServerProcessError> {
        if cancelled() {
            return self.interrupt(turn);
        }
        let mut events = Vec::new();
        let deadline = Instant::now() + self.operation_timeout;
        loop {
            if cancelled() {
                return self.interrupt(turn);
            }
            let event = match self.pop_buffered_for_turn(turn) {
                Some(event) => event,
                None => match self.receive_event_until(deadline) {
                    Ok(event) => event,
                    Err(error) => return self.fail(classify_post_write_failure(error)),
                },
            };
            if event_is_reroute_for(&event, turn) {
                return self.fail(AppServerProcessError::ModelRerouted);
            }
            if event_matches_turn(&event, turn) {
                if let RuntimeEvent::TurnCompleted(completion) = &event {
                    let completion = completion.clone();
                    events.push(event);
                    return Ok(TurnRunReceipt { events, completion });
                }
                if events.len() >= MAX_BUFFERED_EVENTS {
                    return self.fail(AppServerProcessError::PostWriteCapacityExceededNoReplay);
                }
                events.push(event);
            } else if let Err(error) = self.push_buffered(event) {
                return self.fail(classify_post_write_failure(error));
            }
        }
    }

    pub fn interrupt(
        &mut self,
        turn: &ActiveCodexTurn,
    ) -> Result<TurnRunReceipt, AppServerProcessError> {
        let request = self
            .runtime
            .turn_interrupt_request(&turn.thread_id, &turn.turn_id)?;
        if let Err(error) = self.send_and_wait(&request, |event| {
            matches!(
                event,
                RuntimeEvent::InterruptAcknowledged {
                    thread_id,
                    turn_id
                } if thread_id == &turn.thread_id && turn_id == &turn.turn_id
            )
        }) {
            return self.fail(classify_post_write_failure(error));
        }
        let receipt = self.wait_for_terminal(turn)?;
        if receipt.completion.status != TerminalStatus::Interrupted {
            return self.fail(AppServerProcessError::UnexpectedTerminalStatus);
        }
        Ok(receipt)
    }

    pub fn shutdown(&mut self) -> Result<ShutdownReceipt, AppServerProcessError> {
        self.transport.shutdown()
    }

    pub fn into_transport(self) -> T {
        self.transport
    }

    fn send_and_wait(
        &mut self,
        request: &str,
        predicate: impl Fn(&RuntimeEvent) -> bool,
    ) -> Result<RuntimeEvent, AppServerProcessError> {
        if let Err(error) = self.transport.send_line(request) {
            return self.fail(error);
        }
        let deadline = Instant::now() + self.operation_timeout;
        loop {
            let event = self.receive_event_until(deadline)?;
            if predicate(&event) {
                return Ok(event);
            }
            self.push_buffered(event)?;
        }
    }

    fn pop_buffered_for_turn(&mut self, turn: &ActiveCodexTurn) -> Option<RuntimeEvent> {
        let position = self
            .buffered
            .iter()
            .position(|event| event_matches_turn(event, turn));
        position.and_then(|position| self.buffered.remove(position))
    }

    fn receive_event_until(
        &mut self,
        deadline: Instant,
    ) -> Result<RuntimeEvent, AppServerProcessError> {
        let mut ignored_non_semantic_notifications = 0usize;
        loop {
            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
                return self.fail(AppServerProcessError::Timeout);
            };
            if remaining.is_zero() {
                return self.fail(AppServerProcessError::Timeout);
            }
            let line = match self.transport.recv_line(remaining) {
                Ok(line) => line,
                Err(error) => return self.fail(error),
            };
            match self.runtime.ingest_line(&line) {
                Ok(RuntimeEvent::NonSemanticNotificationIgnored) => {
                    let Some(next_count) = ignored_non_semantic_notifications.checked_add(1) else {
                        return self.fail(AppServerProcessError::EventCapacityExceeded);
                    };
                    ignored_non_semantic_notifications = next_count;
                    if ignored_non_semantic_notifications > MAX_IGNORED_NON_SEMANTIC_NOTIFICATIONS {
                        return self.fail(AppServerProcessError::EventCapacityExceeded);
                    }
                }
                Ok(event) => return Ok(event),
                Err(error) => {
                    log_runtime_failure(&error);
                    return self.fail(AppServerProcessError::Runtime(error));
                }
            }
        }
    }

    fn push_buffered(&mut self, event: RuntimeEvent) -> Result<(), AppServerProcessError> {
        if self.buffered.len() >= MAX_BUFFERED_EVENTS {
            return self.fail(AppServerProcessError::EventCapacityExceeded);
        }
        self.buffered.push_back(event);
        Ok(())
    }

    fn fail<R>(&mut self, error: AppServerProcessError) -> Result<R, AppServerProcessError> {
        let _ = self.transport.shutdown();
        Err(error)
    }
}

fn classify_post_write_failure(error: AppServerProcessError) -> AppServerProcessError {
    match error {
        AppServerProcessError::Runtime(RuntimeError::JsonRpcRejected) => {
            AppServerProcessError::PostWriteJsonRpcRejectedNoReplay
        }
        AppServerProcessError::Runtime(RuntimeError::SideEffectingItemRejected) => {
            AppServerProcessError::PostWriteSideEffectRejectedNoReplay
        }
        AppServerProcessError::Runtime(_) => {
            AppServerProcessError::PostWriteNotificationOrStateMismatchNoReplay
        }
        AppServerProcessError::Timeout => AppServerProcessError::PostWriteTimeoutNoReplay,
        AppServerProcessError::UnexpectedEof => {
            AppServerProcessError::PostWriteUnexpectedEofNoReplay
        }
        AppServerProcessError::TransportFailed
        | AppServerProcessError::FrameTooLarge
        | AppServerProcessError::ProcessExited => AppServerProcessError::PostWriteTransportNoReplay,
        AppServerProcessError::EventCapacityExceeded => {
            AppServerProcessError::PostWriteCapacityExceededNoReplay
        }
        _ => AppServerProcessError::PostWriteNotificationOrStateMismatchNoReplay,
    }
}

fn log_runtime_failure(error: &RuntimeError) {
    let class = match error {
        RuntimeError::UnsupportedNotificationMethod(method) => {
            let method_sha256 = sha2::Sha256::digest(method.as_bytes())
                .iter()
                .map(|byte| format!("{byte:02x}"))
                .collect::<String>();
            eprintln!(
                "steadymux-owned-child: refused=runtime class=unsupported-notification method-sha256={method_sha256}"
            );
            return;
        }
        RuntimeError::JsonRpcRejected => "json-rpc-rejected",
        RuntimeError::SensitiveInputRejected => "sensitive-input",
        RuntimeError::SideEffectingItemRejected => "side-effecting-item",
        RuntimeError::CorrelationMismatch => "correlation-mismatch",
        RuntimeError::ProtocolFailure => "protocol-failure",
        RuntimeError::InvalidInput => "invalid-input",
        RuntimeError::PendingCapacityExceeded => "pending-capacity",
        RuntimeError::OutputLimitExceeded => "output-limit",
        _ => "other",
    };
    eprintln!("steadymux-owned-child: refused=runtime class={class}");
}

fn event_matches_turn(event: &RuntimeEvent, turn: &ActiveCodexTurn) -> bool {
    match event {
        RuntimeEvent::TurnAccepted { thread_id, turn_id }
        | RuntimeEvent::TurnStarted { thread_id, turn_id }
        | RuntimeEvent::ItemStarted {
            thread_id, turn_id, ..
        }
        | RuntimeEvent::ItemDelta {
            thread_id, turn_id, ..
        }
        | RuntimeEvent::ItemCompleted {
            thread_id, turn_id, ..
        }
        | RuntimeEvent::ToolCallStarted {
            thread_id, turn_id, ..
        }
        | RuntimeEvent::ToolCallCompleted {
            thread_id, turn_id, ..
        }
        | RuntimeEvent::InterruptAcknowledged { thread_id, turn_id }
        | RuntimeEvent::ModelReroutedFailure {
            thread_id, turn_id, ..
        } => thread_id == &turn.thread_id && turn_id == &turn.turn_id,
        RuntimeEvent::TurnCompleted(receipt) => {
            receipt.thread_id == turn.thread_id && receipt.turn_id == turn.turn_id
        }
        RuntimeEvent::InitializeAccepted
        | RuntimeEvent::NonSemanticNotificationIgnored
        | RuntimeEvent::ManagedAccount(_)
        | RuntimeEvent::ModelPage(_)
        | RuntimeEvent::RateLimits(_)
        | RuntimeEvent::ThreadStarted { .. } => false,
    }
}

fn event_is_reroute_for(event: &RuntimeEvent, turn: &ActiveCodexTurn) -> bool {
    matches!(
        event,
        RuntimeEvent::ModelReroutedFailure {
            thread_id,
            turn_id,
            ..
        } if thread_id == &turn.thread_id && turn_id == &turn.turn_id
    )
}

fn valid_arg(value: &OsStr) -> bool {
    let Some(value) = value.to_str() else {
        return false;
    };
    !value.is_empty()
        && value.len() <= MAX_ARG_BYTES
        && !value.contains('\0')
        && !contains_token_marker(value)
}

fn valid_timeout(value: Duration) -> bool {
    (MIN_TIMEOUT..=MAX_TIMEOUT).contains(&value)
}

fn configure_managed_child_env(command: &mut Command, pinned_codex_home: Option<&Path>) {
    // The child resolves managed authentication from its own HOME/CODEX_HOME.
    // Start from an empty environment so no raw token or unrelated secret can
    // cross this process boundary, then copy only non-secret runtime paths and
    // locale settings required by the CLI.
    command.env_clear();
    for name in MANAGED_CHILD_ENV_ALLOWLIST {
        if name == "CODEX_HOME" && pinned_codex_home.is_some() {
            continue;
        }
        if let Some(value) = std::env::var_os(name) {
            command.env(name, value);
        }
    }
    if let Some(codex_home) = pinned_codex_home {
        command.env("CODEX_HOME", codex_home);
    }
}

fn contains_token_marker(value: &str) -> bool {
    let value = value.to_ascii_lowercase();
    [
        "access_token",
        "refresh_token",
        "authorization:",
        "bearer ",
        "api_key",
        "apikey",
        "sk-",
    ]
    .iter()
    .any(|marker| value.contains(marker))
}

fn read_stdout(stdout: impl Read, sender: SyncSender<ProcessFrame>, max_line_bytes: usize) {
    let mut reader = BufReader::new(stdout);
    loop {
        let frame = match read_bounded_line(&mut reader, max_line_bytes) {
            Ok(Some(line)) => ProcessFrame::Line(line),
            Ok(None) => ProcessFrame::Eof,
            Err(LineReadError::TooLarge) => ProcessFrame::TooLarge,
            Err(LineReadError::Io) => ProcessFrame::Failed,
        };
        let terminal = !matches!(frame, ProcessFrame::Line(_));
        if sender.send(frame).is_err() || terminal {
            return;
        }
    }
}

fn discard_stderr(mut stderr: impl Read) {
    let mut buffer = [0_u8; 4096];
    loop {
        match stderr.read(&mut buffer) {
            Ok(0) | Err(_) => return,
            Ok(_) => {}
        }
    }
}

enum LineReadError {
    TooLarge,
    Io,
}

fn read_bounded_line(
    reader: &mut impl BufRead,
    max_line_bytes: usize,
) -> Result<Option<Vec<u8>>, LineReadError> {
    let mut line = Vec::new();
    loop {
        let available = reader.fill_buf().map_err(|_| LineReadError::Io)?;
        if available.is_empty() {
            return if line.is_empty() {
                Ok(None)
            } else {
                Err(LineReadError::Io)
            };
        }
        if let Some(newline) = available.iter().position(|byte| *byte == b'\n') {
            if line.len().saturating_add(newline) > max_line_bytes {
                return Err(LineReadError::TooLarge);
            }
            line.extend_from_slice(&available[..newline]);
            reader.consume(newline + 1);
            return Ok(Some(line));
        }
        if line.len().saturating_add(available.len()) > max_line_bytes {
            return Err(LineReadError::TooLarge);
        }
        let consumed = available.len();
        line.extend_from_slice(available);
        reader.consume(consumed);
    }
}

fn join_reader_until(task: Option<JoinHandle<()>>, deadline: Instant) -> bool {
    let Some(task) = task else {
        return true;
    };
    while !task.is_finished() && Instant::now() < deadline {
        thread::sleep(Duration::from_millis(5));
    }
    task.is_finished() && task.join().is_ok()
}

fn kill_and_reap(child: &mut Child) {
    let _ = child.kill();
    let _ = child.wait();
}

fn exit_is_success_or_forced(status: ExitStatus, forced: bool) -> bool {
    status.success() || forced
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use tempfile::TempDir;

    fn reviewed_live_executable(sentinel: &str) -> PathBuf {
        assert_eq!(
            std::env::var(sentinel).as_deref(),
            Ok("1"),
            "set the test-specific live authorization sentinel explicitly"
        );
        std::env::var_os("STEADYMUX_CODEX_EXECUTABLE")
            .map(PathBuf::from)
            .expect("set an absolute reviewed Codex executable path")
    }

    #[test]
    fn owned_child_preflight_classes_are_fixed_and_content_free() {
        let cases = [
            (AppServerProcessError::InvalidConfiguration, "config-other"),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::CodexHomeStructural,
                ),
                "codex-home-structural",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink,
                ),
                "config-open-type-owner-mode-link",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigFileIdentity,
                ),
                "config-file-identity",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigContentDigest,
                ),
                "config-content-digest",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigSemanticKeys,
                ),
                "config-semantic-keys",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigOverrideDigest,
                ),
                "config-override-digest",
            ),
            (
                AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigArgCap,
                ),
                "config-arg-cap",
            ),
            (
                AppServerProcessError::SpawnExecutableUnavailable,
                "command-not-found",
            ),
            (
                AppServerProcessError::SpawnExecutablePermissionDenied,
                "command-permission",
            ),
            (AppServerProcessError::SpawnFailed, "command-other"),
            (AppServerProcessError::SpawnStdioUnavailable, "stdio"),
            (AppServerProcessError::SpawnReaderUnavailable, "reader"),
            (AppServerProcessError::ProcessExited, "child-exit"),
            (AppServerProcessError::UnexpectedEof, "child-exit"),
        ];
        for (error, class) in cases {
            assert_eq!(error.preflight_refusal_class(), class);
            assert!(
                error
                    .preflight_refusal_class()
                    .bytes()
                    .all(|byte| byte.is_ascii_lowercase() || byte == b'-'),
                "diagnostic class is fixed and does not project process data"
            );
        }
    }

    /// Generous window for owned-fixture setup (spawn, initialize, discover,
    /// thread/start). The child process and its reader threads must come up
    /// and complete several round-trips inside this window; 2s absorbs the
    /// wall-clock scheduling latency that a tight timeout would race under
    /// parallel test load.
    #[cfg(unix)]
    const TIMEOUT_FIXTURE_SETUP_WINDOW: Duration = Duration::from_secs(2);

    #[cfg(unix)]
    fn shell_fixture(script: &str, timeout: Duration) -> (TempDir, OwnedAppServerConfig) {
        let directory = TempDir::new().expect("temporary fake app-server directory");
        let script_path = directory.path().join("fake-app-server.sh");
        fs::write(&script_path, script).expect("write fake app-server script");
        let shell = std::path::Path::new("/bin/sh")
            .canonicalize()
            .expect("canonical shell");
        let config =
            OwnedAppServerConfig::fixture(&shell, vec![script_path.into_os_string()], timeout);
        (directory, config)
    }

    #[cfg(unix)]
    fn instruction_trace_fixture() -> (TempDir, OwnedAppServerConfig, PathBuf) {
        let directory = TempDir::new().expect("temporary instruction fixture directory");
        let script_path = directory.path().join("instruction-app-server.sh");
        let trace = directory.path().join("thread-start.jsonl");
        let script = format!(
            r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{{"id":1,"result":{{"userAgent":"steadymux/0.148.0"}}}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{{"id":2,"result":{{"account":{{"type":"chatgpt","planType":"pro"}},"requiresOpenaiAuth":true}}}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{{"id":3,"result":{{"data":[{{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{{"reasoningEffort":"medium"}}]}}],"nextCursor":null}}}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{{"id":4,"result":{{"rateLimits":{{"primary":null,"secondary":null,"rateLimitReachedType":null}}}}}}' ;;
    *'"method":"thread/start"'*)
      printf '%s\n' "$line" > '{}'
      printf '%s\n' '{{"id":5,"result":{{"thread":{{"id":"thr-instructions"}}}}}}' ;;
    *'"method":"turn/start"'*) printf '%s\n' '{{"id":6,"result":{{"turn":{{"id":"turn-instructions","status":"inProgress","items":[],"error":null}}}}}}' ;;
  esac
done
"#,
            trace.display()
        );
        fs::write(&script_path, script).expect("write instruction fixture script");
        let shell = std::path::Path::new("/bin/sh")
            .canonicalize()
            .expect("canonical shell");
        let config = OwnedAppServerConfig::fixture(
            &shell,
            vec![script_path.into_os_string()],
            Duration::from_secs(2),
        );
        (directory, config, trace)
    }

    #[cfg(unix)]
    fn attested_executable_fixture(version: &str) -> (TempDir, PathBuf, PathBuf) {
        use std::os::unix::fs::PermissionsExt;

        let directory = TempDir::new().expect("temporary executable directory");
        let executable = directory.path().join("codex-fixture");
        let trace = directory.path().join("invocations.log");
        let script = format!(
            "#!/bin/sh\nprintf 'argv:%s\\n' \"$*\" >> '{}'\nwhile IFS= read -r line; do\n  printf '%s\\n' \"$line\" >> '{}'\n  case \"$line\" in\n    *'\"method\":\"initialize\"'*) printf '%s\\n' '{{\"id\":1,\"result\":{{\"userAgent\":\"steadymux/{} (fixture)\"}}}}' ;;\n    *'\"method\":\"model/list\"'*) printf '%s\\n' '{{\"id\":2,\"result\":{{\"data\":[],\"nextCursor\":null}}}}' ;;\n  esac\ndone\n",
            trace.display(),
            trace.display(),
            version
        );
        fs::write(&executable, script).expect("write executable fixture");
        let mut permissions = fs::metadata(&executable).unwrap().permissions();
        permissions.set_mode(0o700);
        fs::set_permissions(&executable, permissions).unwrap();
        (directory, executable, trace)
    }

    #[cfg(unix)]
    fn mcp_config_home(directory: &TempDir, tables: &[&str]) -> PathBuf {
        use std::os::unix::fs::PermissionsExt as _;

        let home = directory.path().join("codex-home");
        fs::create_dir(&home).expect("private Codex home");
        let mut home_permissions = fs::metadata(&home)
            .expect("Codex home metadata")
            .permissions();
        home_permissions.set_mode(0o700);
        fs::set_permissions(&home, home_permissions).expect("private Codex home mode");
        let config = tables.join("\n");
        fs::write(home.join("config.toml"), config).expect("MCP config fixture");
        let mut config_permissions = fs::metadata(home.join("config.toml"))
            .expect("MCP config metadata")
            .permissions();
        config_permissions.set_mode(0o600);
        fs::set_permissions(home.join("config.toml"), config_permissions)
            .expect("private MCP config mode");
        home.canonicalize().expect("canonical private Codex home")
    }

    #[cfg(unix)]
    const COMPLETE_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","email":"private@example.test","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":{"usedPercent":10,"windowDurationMins":15,"resetsAt":1730947200},"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-fixture"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-fixture","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-fixture","turn":{"id":"turn-fixture","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-fixture","turnId":"turn-fixture","item":{"type":"agentMessage","id":"item-fixture","text":""}}}'
      printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thr-fixture","turnId":"turn-fixture","itemId":"item-fixture","delta":"draft"}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-fixture","turnId":"turn-fixture","item":{"type":"agentMessage","id":"item-fixture","text":"authoritative"}}}'
      printf '%s\n' '{"method":"thread/tokenUsage/updated","params":{"threadId":"thr-fixture","turnId":"turn-fixture","tokenUsage":{"last":{"cachedInputTokens":3,"inputTokens":21,"outputTokens":8,"reasoningOutputTokens":2,"totalTokens":29},"total":{"cachedInputTokens":103,"inputTokens":121,"outputTokens":108,"reasoningOutputTokens":102,"totalTokens":229},"modelContextWindow":128000}}}'
      printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thr-fixture","turn":{"id":"turn-fixture","status":"completed","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const CONTROL_NOTIFICATION_DISCOVERY_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*)
      printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}'
      printf '%s\n' '{"method":"remoteControl/status/changed","params":{"status":"disabled","serverName":"private-server-canary","installationId":"private-installation-canary","environmentId":null},"emittedAtMs":1}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const CONTROL_NOTIFICATION_FLOOD_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*)
      printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}'
      ignored=0
      while [ "$ignored" -lt 65 ]; do
        printf '%s\n' '{"method":"remoteControl/status/changed","params":{"status":"disabled","serverName":"private-server-canary","installationId":"private-installation-canary","environmentId":null},"emittedAtMs":1}'
        ignored=$((ignored + 1))
      done ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const CONTROL_NOTIFICATION_THRESHOLD_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*)
      printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}'
      ignored=0
      while [ "$ignored" -lt 64 ]; do
        printf '%s\n' '{"method":"remoteControl/status/changed","params":{"status":"disabled","serverName":"private-server-canary","installationId":"private-installation-canary","environmentId":null},"emittedAtMs":1}'
        ignored=$((ignored + 1))
      done ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const NON_SEMANTIC_NO_TOOLS_TURN_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-luna","displayName":"Luna","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"low"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*)
      printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-notification"}}}'
      printf '%s\n' '{"method":"thread/started","params":{"thread":{"id":"thr-notification","preview":"private-preview-canary","ephemeral":true,"modelProvider":"openai","createdAt":1,"updatedAt":1,"status":{"type":"idle"},"cwd":"/private-cwd-canary","cliVersion":"0.148.0","source":"appServer","sessionId":"private-session-canary","turns":[],"agentNickname":null,"agentRole":null,"canAcceptDirectInput":true,"extra":null,"forkedFromId":null,"gitInfo":null,"historyMode":"legacy","name":null,"parentThreadId":null,"path":null,"recencyAt":1,"section":null,"sectionEnteredAt":null,"threadSource":null}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-notification","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":[]}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-notification","turn":{"id":"turn-notification","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-notification","turnId":"turn-notification","startedAtMs":1,"item":{"type":"reasoning","id":"reasoning-notification","summary":[],"content":[]}}}'
      printf '%s\n' '{"method":"item/reasoning/summaryPartAdded","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0}}'
      printf '%s\n' '{"method":"item/reasoning/summaryTextDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0,"delta":"private-summary-canary"}}'
      printf '%s\n' '{"method":"item/reasoning/textDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","contentIndex":0,"delta":"private-reasoning-canary"}}'
      printf '%s\n' '{"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"total":{"cachedInputTokens":11,"inputTokens":12,"outputTokens":13,"reasoningOutputTokens":14,"totalTokens":50},"modelContextWindow":128000}}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-notification","turnId":"turn-notification","completedAtMs":2,"item":{"type":"reasoning","id":"reasoning-notification","summary":[],"content":[]}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-notification","turnId":"turn-notification","startedAtMs":3,"item":{"type":"agentMessage","id":"item-notification","text":""}}}'
      printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"item-notification","delta":"answer"}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-notification","turnId":"turn-notification","completedAtMs":4,"item":{"type":"agentMessage","id":"item-notification","text":"answer"}}}'
      printf '%s\n' '{"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"idle"}}}'
      printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thr-notification","turn":{"id":"turn-notification","status":"completed","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const TOOL_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-tool"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-tool","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-tool","turn":{"id":"turn-tool","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-tool","turnId":"turn-tool","item":{"type":"mcpToolCall","id":"call-tool","server":"fixture-server","tool":"fixture-read","status":"inProgress","arguments":{"query":"private-argument"},"appContext":null,"pluginId":null,"result":null,"error":null,"durationMs":null}}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-tool","turnId":"turn-tool","completedAtMs":1,"item":{"type":"mcpToolCall","id":"call-tool","server":"fixture-server","tool":"fixture-read","status":"completed","arguments":{"query":"private-argument"},"appContext":null,"pluginId":null,"result":{"content":[{"type":"text","text":"private-result"}]},"error":null,"durationMs":1}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-tool","turnId":"turn-tool","item":{"type":"agentMessage","id":"item-tool-answer","text":""}}}'
      printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thr-tool","turnId":"turn-tool","itemId":"item-tool-answer","delta":"draft"}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-tool","turnId":"turn-tool","item":{"type":"agentMessage","id":"item-tool-answer","text":"authoritative"}}}'
      printf '%s\n' '{"method":"thread/tokenUsage/updated","params":{"threadId":"thr-tool","turnId":"turn-tool","tokenUsage":{"last":{"cachedInputTokens":5,"inputTokens":31,"outputTokens":4,"reasoningOutputTokens":1,"totalTokens":35},"total":{"cachedInputTokens":205,"inputTokens":231,"outputTokens":204,"reasoningOutputTokens":201,"totalTokens":435},"modelContextWindow":128000}}}'
      printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thr-tool","turn":{"id":"turn-tool","status":"completed","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const INTERRUPT_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-interrupt"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-interrupt","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-interrupt","turn":{"id":"turn-interrupt","status":"inProgress","items":[],"error":null}}}' ;;
    *'"method":"turn/interrupt"'*)
      printf '%s\n' '{"id":7,"result":{}}'
      printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thr-interrupt","turn":{"id":"turn-interrupt","status":"interrupted","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const WAIT_FAILURE_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-failure"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-failure","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-failure","turn":{"id":"turn-failure","status":"inProgress","items":[],"error":null}}}'
      exit 0 ;;
  esac
done
"#;

    #[cfg(unix)]
    const INTERRUPT_PROTOCOL_FAILURE_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-failure"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-failure","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-failure","turn":{"id":"turn-failure","status":"inProgress","items":[],"error":null}}}' ;;
    *'"method":"turn/interrupt"'*) printf '%s\n' '{"method":"unexpected","params":{"content":"private-unsupported-canary"}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const SIDE_EFFECT_REJECTION_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-side-effect"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-side-effect","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-side-effect","turn":{"id":"turn-side-effect","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-side-effect","turnId":"turn-side-effect","item":{"type":"commandExecution","id":"item-side-effect","command":"private-command-canary"}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const TURN_START_JSON_RPC_REJECTION_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-rejected"}}}' ;;
    *'"method":"turn/start"'*) printf '%s\n' '{"id":6,"error":{"code":-32603,"message":"private-json-rpc-message-canary","data":{"content":"private-json-rpc-data-canary"}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const TURN_STARTED_BEFORE_RESPONSE_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-notification-first"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-notification-first","turn":{"id":"turn-notification-first","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/started","params":{"threadId":"thr-notification-first","turnId":"turn-notification-first","item":{"type":"agentMessage","id":"item-notification-first","text":""}}}'
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-notification-first","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thr-notification-first","turnId":"turn-notification-first","itemId":"item-notification-first","delta":"answer"}}'
      printf '%s\n' '{"method":"item/completed","params":{"threadId":"thr-notification-first","turnId":"turn-notification-first","item":{"type":"agentMessage","id":"item-notification-first","text":"answer"}}}'
      printf '%s\n' '{"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification-first","turnId":"turn-notification-first","tokenUsage":{"last":{"cachedInputTokens":0,"inputTokens":2,"outputTokens":1,"reasoningOutputTokens":0,"totalTokens":3},"total":{"cachedInputTokens":0,"inputTokens":2,"outputTokens":1,"reasoningOutputTokens":0,"totalTokens":3},"modelContextWindow":128000}}}'
      printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thr-notification-first","turn":{"id":"turn-notification-first","status":"completed","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const TURN_STARTED_WITHOUT_RESPONSE_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-notification-timeout"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-notification-timeout","turn":{"id":"turn-notification-timeout","status":"inProgress","items":[],"error":null}}}' ;;
  esac
done
"#;

    #[cfg(unix)]
    const INTERRUPT_TIMEOUT_SCRIPT: &str = r#"
while IFS= read -r line; do
  case "$line" in
    *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{"userAgent":"steadymux/0.148.0"}}' ;;
    *'"method":"initialized"'*) ;;
    *'"method":"account/read"'*) printf '%s\n' '{"id":2,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}}' ;;
    *'"method":"model/list"'*) printf '%s\n' '{"id":3,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}}' ;;
    *'"method":"account/rateLimits/read"'*) printf '%s\n' '{"id":4,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":null}}}' ;;
    *'"method":"thread/start"'*) printf '%s\n' '{"id":5,"result":{"thread":{"id":"thr-timeout"}}}' ;;
    *'"method":"turn/start"'*)
      printf '%s\n' '{"id":6,"result":{"turn":{"id":"turn-timeout","status":"inProgress","items":[],"error":null}}}'
      printf '%s\n' '{"method":"turn/started","params":{"threadId":"thr-timeout","turn":{"id":"turn-timeout","status":"inProgress","items":[],"error":null}}}' ;;
    *'"method":"turn/interrupt"'*) ;;
  esac
done
"#;

    /// Spawns an owned fixture turn. Spawn and every setup operation
    /// (initialize, discover, thread/start, turn/start) run under the generous
    /// setup window so wall-clock scheduling latency can never race a tight
    /// timeout. The caller-supplied timeout is applied to the owned client
    /// afterwards so that only the deliberately-failing post-write phase
    /// (the turn/interrupt wait in the timeout fixtures) sees it.
    #[cfg(unix)]
    fn start_owned_fixture_turn_with_timeout(
        script: &str,
        turn_phase_timeout: Duration,
    ) -> (
        TempDir,
        CodexAppServerClient<OwnedAppServerTransport>,
        ActiveCodexTurn,
    ) {
        let (directory, config) = shell_fixture(script, TIMEOUT_FIXTURE_SETUP_WINDOW);
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn failure fixture");
        client.initialize().expect("initialize failure fixture");
        client.discover().expect("discover failure fixture");
        let turn = client
            .start_turn(
                "failure-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect("start failure fixture turn");
        client.operation_timeout = turn_phase_timeout;
        (directory, client, turn)
    }

    #[cfg(unix)]
    fn start_owned_fixture_turn(
        script: &str,
    ) -> (
        TempDir,
        CodexAppServerClient<OwnedAppServerTransport>,
        ActiveCodexTurn,
    ) {
        start_owned_fixture_turn_with_timeout(script, Duration::from_secs(2))
    }

    #[test]
    #[cfg(unix)]
    fn owned_fake_process_completes_discovery_and_authoritative_turn() {
        let (_directory, config) = shell_fixture(COMPLETE_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn owned fixture child");
        client.initialize().expect("initialize fixture client");
        let discovery = client.discover().expect("bounded discovery");
        assert_eq!(
            discovery.schema_version,
            APP_SERVER_DISCOVERY_SCHEMA_VERSION
        );
        assert_eq!(
            discovery.required_cli_version,
            CURRENT_STANDALONE_CLI_VERSION
        );
        assert_eq!(
            discovery.protocol_version,
            CODEX_APP_SERVER_PROTOCOL_VERSION
        );
        assert_eq!(discovery.account.plan_type.as_deref(), Some("pro"));
        assert_eq!(discovery.model_pages.len(), 1);
        assert_eq!(discovery.model_pages[0].models[0].id, "gpt-5.6-terra");
        let turn = client
            .start_turn(
                "fixture-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect("start exact selected turn");
        let receipt = client
            .wait_for_terminal(&turn)
            .expect("terminal completion");
        assert_eq!(receipt.completion.status, TerminalStatus::Completed);
        assert_eq!(receipt.completion.output.as_deref(), Some("authoritative"));
        assert_eq!(receipt.completion.input_tokens, Some(21));
        assert_eq!(receipt.completion.output_tokens, Some(8));
        assert!(client.shutdown().expect("shutdown fixture").success);
        assert!(!format!("{client:?}").contains("private@example.test"));
    }

    #[test]
    #[cfg(unix)]
    fn desktop_version_guard_is_reported_from_verified_initialize_and_mismatch_refuses() {
        let desktop_script = COMPLETE_SCRIPT.replace(
            "\"userAgent\":\"steadymux/0.148.0\"",
            &format!("\"userAgent\":\"steadymux/{CURRENT_DESKTOP_CODEX_VERSION}\""),
        );
        let (_directory, config) = shell_fixture(&desktop_script, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("desktop transcript client");
        client
            .require_app_server_version(CURRENT_DESKTOP_CODEX_VERSION)
            .expect("closed desktop version guard");
        client.initialize().expect("verified desktop initialize");
        let discovery = client.discover().expect("desktop discovery");
        assert_eq!(
            discovery.required_cli_version,
            CURRENT_DESKTOP_CODEX_VERSION
        );
        assert!(client.shutdown().expect("desktop shutdown").success);

        let (_directory, config) = shell_fixture(&desktop_script, Duration::from_secs(2));
        let mut mismatch =
            CodexAppServerClient::spawn(&config).expect("mismatch transcript client");
        mismatch
            .require_app_server_version(CURRENT_STANDALONE_CLI_VERSION)
            .expect("closed standalone version guard");
        assert_eq!(
            mismatch.initialize(),
            Err(AppServerProcessError::UnsupportedExecutableVersion)
        );
    }

    #[test]
    #[cfg(unix)]
    fn owned_process_sends_exact_redacted_thread_instruction_shape() {
        let (_directory, config, trace) = instruction_trace_fixture();
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn instruction fixture");
        client.initialize().expect("initialize instruction fixture");
        client.discover().expect("discover instruction fixture");
        let base = "private-base-process-canary\nsecond line";
        let developer = "private-developer-process-canary\nsecond line";
        let turn = client
            .start_turn_with_instructions(
                "instruction-message",
                "bounded fixture prompt",
                Some(base),
                Some(developer),
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect("start turn with bounded instructions");
        assert_eq!(turn.thread_id, "thr-instructions");
        assert_eq!(turn.turn_id, "turn-instructions");

        let request: serde_json::Value = serde_json::from_str(
            fs::read_to_string(&trace)
                .expect("read exact thread request trace")
                .trim_end(),
        )
        .expect("parse exact thread request trace");
        assert_eq!(
            request["params"],
            serde_json::json!({
                "model": "gpt-5.6-terra",
                "modelProvider": "openai",
                "approvalPolicy": "never",
                "sandbox": "read-only",
                "ephemeral": true,
                "baseInstructions": base,
                "developerInstructions": developer
            })
        );
        let debug = format!("{client:?} {turn:?}");
        assert!(!debug.contains(base));
        assert!(!debug.contains(developer));

        let secret = "sk-privateprocessinstruction123456";
        let error = client
            .start_turn_with_instructions(
                "rejected-instruction-message",
                "bounded fixture prompt",
                None,
                Some(secret),
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect_err("secret-shaped instruction must fail before transport write");
        assert_eq!(
            error,
            AppServerProcessError::Runtime(RuntimeError::SensitiveInputRejected)
        );
        assert!(!format!("{error:?} {error} {client:?}").contains(secret));
        assert_eq!(
            fs::read_to_string(&trace)
                .expect("thread trace remains a single request")
                .lines()
                .count(),
            1
        );
        assert!(
            client
                .shutdown()
                .expect("shutdown instruction fixture")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn reviewed_control_notification_is_discarded_without_buffering() {
        let (_directory, config) = shell_fixture(
            CONTROL_NOTIFICATION_DISCOVERY_SCRIPT,
            Duration::from_secs(2),
        );
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn control fixture");
        client.initialize().expect("initialize control fixture");
        let discovery = client
            .discover()
            .expect("control notification must not block discovery");
        assert!(
            discovery
                .model_pages
                .iter()
                .all(|page| page.models.is_empty())
        );
        assert!(client.buffered.is_empty());
        let debug = format!("{client:?}");
        assert!(!debug.contains("private-server-canary"));
        assert!(!debug.contains("private-installation-canary"));
        assert!(client.shutdown().expect("shutdown control fixture").success);
    }

    #[test]
    #[cfg(unix)]
    fn ignored_control_notification_flood_is_bounded() {
        let (_directory, config) =
            shell_fixture(CONTROL_NOTIFICATION_FLOOD_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn flood fixture");
        client.initialize().expect("initialize flood fixture");
        assert_eq!(
            client.discover(),
            Err(AppServerProcessError::EventCapacityExceeded)
        );
        assert!(
            client
                .shutdown()
                .expect("flood child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn ignored_control_notification_exact_threshold_is_accepted() {
        let (_directory, config) = shell_fixture(
            CONTROL_NOTIFICATION_THRESHOLD_SCRIPT,
            Duration::from_secs(2),
        );
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn threshold fixture");
        client.initialize().expect("initialize threshold fixture");
        client
            .discover()
            .expect("exactly 64 ignored notifications remain within the bound");
        assert!(client.buffered.is_empty());
        assert!(
            client
                .shutdown()
                .expect("shutdown threshold fixture")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn no_tools_turn_discards_reviewed_non_semantic_notifications() {
        let (_directory, config) =
            shell_fixture(NON_SEMANTIC_NO_TOOLS_TURN_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn notification fixture");
        client
            .initialize()
            .expect("initialize notification fixture");
        client.discover().expect("discover notification fixture");
        let turn = client
            .start_turn(
                "notification-message",
                "bounded fixture prompt",
                "gpt-5.6-luna",
                ReasoningEffort::Low,
            )
            .expect("start notification fixture turn");
        let receipt = client
            .wait_for_terminal(&turn)
            .expect("reviewed notifications must not make the turn ambiguous");
        assert_eq!(receipt.completion.output.as_deref(), Some("answer"));
        assert_eq!(receipt.completion.input_tokens, Some(2));
        assert_eq!(receipt.completion.output_tokens, Some(3));
        assert_eq!(receipt.events.len(), 7);
        assert!(client.buffered.is_empty());
        let debug = format!("{receipt:?} {client:?}");
        assert!(!debug.contains("private-summary-canary"));
        assert!(!debug.contains("private-reasoning-canary"));
        assert!(
            client
                .shutdown()
                .expect("shutdown notification fixture")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn owned_fake_process_preserves_canonical_tool_call_result_lifecycle() {
        let (_directory, config) = shell_fixture(TOOL_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn_with_tool_policy(
            &config,
            CodexToolPolicy::CanonicalMcpLifecycle,
        )
        .expect("spawn owned tool fixture child");
        client.initialize().unwrap();
        client.discover().unwrap();
        let turn = client
            .start_turn(
                "fixture-tool-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .unwrap();
        let receipt = client.wait_for_terminal(&turn).unwrap();
        assert!(matches!(
            receipt.events.as_slice(),
            [
                RuntimeEvent::TurnStarted { .. },
                RuntimeEvent::ToolCallStarted { item_id: started, .. },
                RuntimeEvent::ToolCallCompleted { item_id: completed, .. },
                RuntimeEvent::ItemStarted { .. },
                RuntimeEvent::ItemDelta { .. },
                RuntimeEvent::ItemCompleted { .. },
                RuntimeEvent::TurnCompleted(_),
            ] if started == "call-tool" && completed == "call-tool"
        ));
        assert_eq!(
            receipt.completion.replay,
            crate::ReplayDisposition::DeniedToolOrSideEffect
        );
        assert_eq!(receipt.completion.output.as_deref(), Some("authoritative"));
        assert_eq!(receipt.completion.input_tokens, Some(31));
        assert_eq!(receipt.completion.output_tokens, Some(4));
        assert!(!format!("{receipt:?}").contains("private-argument"));
        assert!(!format!("{receipt:?}").contains("private-result"));
        assert!(client.shutdown().unwrap().success);
    }

    #[test]
    #[cfg(unix)]
    fn interrupt_requires_ack_and_terminal_interrupted_before_success() {
        let (_directory, config) = shell_fixture(INTERRUPT_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn interrupt fixture");
        client.initialize().unwrap();
        client.discover().unwrap();
        let turn = client
            .start_turn(
                "fixture-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .unwrap();
        let receipt = client.interrupt(&turn).expect("correlated interruption");
        assert_eq!(receipt.completion.status, TerminalStatus::Interrupted);
        assert_eq!(receipt.completion.input_tokens, None);
        assert_eq!(receipt.completion.output_tokens, None);
        assert!(client.shutdown().unwrap().success);
    }

    #[test]
    #[cfg(unix)]
    fn wait_transport_failure_after_turn_write_is_explicitly_non_replayable() {
        let (_directory, mut client, turn) = start_owned_fixture_turn(WAIT_FAILURE_SCRIPT);
        assert_eq!(
            client.wait_for_terminal(&turn),
            Err(AppServerProcessError::PostWriteUnexpectedEofNoReplay)
        );
        assert!(
            client
                .shutdown()
                .expect("failed child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn unsupported_notification_after_turn_write_has_static_non_replayable_code() {
        let (_directory, mut client, turn) =
            start_owned_fixture_turn(INTERRUPT_PROTOCOL_FAILURE_SCRIPT);
        let error = client
            .interrupt(&turn)
            .expect_err("unsupported notification must fail closed without replay");
        assert_eq!(
            error,
            AppServerProcessError::PostWriteNotificationOrStateMismatchNoReplay
        );
        assert!(!format!("{error:?} {error}").contains("private-unsupported-canary"));
        assert!(
            client
                .shutdown()
                .expect("failed child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn side_effect_rejection_after_turn_write_has_static_content_free_code() {
        let (_directory, mut client, turn) = start_owned_fixture_turn(SIDE_EFFECT_REJECTION_SCRIPT);
        let error = client
            .wait_for_terminal(&turn)
            .expect_err("side-effecting item must fail closed without replay");
        assert_eq!(
            error,
            AppServerProcessError::PostWriteSideEffectRejectedNoReplay
        );
        let debug = format!("{error:?} {error} {client:?}");
        assert!(!debug.contains("private-command-canary"));
        assert!(
            client
                .shutdown()
                .expect("side-effect rejection child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn json_rpc_rejection_after_turn_write_has_static_content_free_code() {
        let (_directory, config) =
            shell_fixture(TURN_START_JSON_RPC_REJECTION_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn rejection fixture");
        client.initialize().expect("initialize rejection fixture");
        client.discover().expect("discover rejection fixture");
        let error = client
            .start_turn(
                "rejected-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect_err("JSON-RPC rejection must fail closed without replay");
        assert_eq!(
            error,
            AppServerProcessError::PostWriteJsonRpcRejectedNoReplay
        );
        let debug = format!("{error:?} {error} {client:?}");
        assert!(!debug.contains("private-json-rpc-message-canary"));
        assert!(!debug.contains("private-json-rpc-data-canary"));
        assert!(
            client
                .shutdown()
                .expect("rejected child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn turn_scoped_notification_before_turn_start_response_completes() {
        let (_directory, config) =
            shell_fixture(TURN_STARTED_BEFORE_RESPONSE_SCRIPT, Duration::from_secs(2));
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn ordering fixture");
        client.initialize().expect("initialize ordering fixture");
        client.discover().expect("discover ordering fixture");
        let turn = client
            .start_turn(
                "ordering-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            )
            .expect("exactly correlated notification-first turn");
        let receipt = client
            .wait_for_terminal(&turn)
            .expect("notification-first turn must complete");
        assert_eq!(receipt.completion.output.as_deref(), Some("answer"));
        assert_eq!(receipt.completion.input_tokens, Some(2));
        assert_eq!(receipt.completion.output_tokens, Some(1));
        assert!(
            client
                .shutdown()
                .expect("ordering failure child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn notification_first_missing_response_times_out_without_replay() {
        let (_directory, config) = shell_fixture(
            TURN_STARTED_WITHOUT_RESPONSE_SCRIPT,
            TIMEOUT_FIXTURE_SETUP_WINDOW,
        );
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn timeout fixture");
        client.initialize().expect("initialize timeout fixture");
        client.discover().expect("discover timeout fixture");
        // thread/start runs under the generous setup window; the tight timeout
        // is clamped on only for the turn/start wait, whose terminal state is
        // a guaranteed timeout (the fixture never responds and never exits).
        client.turn_operation_timeout = Some(Duration::from_millis(50));
        assert_eq!(
            client.start_turn(
                "ordering-timeout-message",
                "bounded fixture prompt",
                "gpt-5.6-terra",
                ReasoningEffort::Medium,
            ),
            Err(AppServerProcessError::PostWriteTimeoutNoReplay)
        );
        assert!(
            client
                .shutdown()
                .expect("timed out child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn timeout_after_turn_write_is_distinct_from_eof_and_non_replayable() {
        let (_directory, mut client, turn) = start_owned_fixture_turn_with_timeout(
            INTERRUPT_TIMEOUT_SCRIPT,
            Duration::from_millis(50),
        );
        assert_eq!(
            client.interrupt(&turn),
            Err(AppServerProcessError::PostWriteTimeoutNoReplay)
        );
        assert!(
            client
                .shutdown()
                .expect("timed out child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn timeout_forces_owned_child_reap() {
        let sleep = std::path::Path::new("/bin/sleep")
            .canonicalize()
            .expect("canonical sleep fixture");
        let config = OwnedAppServerConfig::fixture(
            &sleep,
            vec![OsString::from("60")],
            Duration::from_millis(25),
        );
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn stalled fixture child");
        assert!(client.transport.child_id().is_some());
        assert_eq!(client.initialize(), Err(AppServerProcessError::Timeout));
        assert!(client.transport.child_id().is_none());
        let receipt = client
            .shutdown()
            .expect("timeout path already reaped child");
        assert!(receipt.success);
    }

    #[test]
    #[cfg(unix)]
    fn production_owned_child_constructor_disables_hooks_shell_and_web_search() {
        let (directory, executable, _trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(
            &directory,
            &[
                "[mcp_servers.zeta]",
                "command = \"private-command\"",
                "[mcp_servers.zeta.env]",
                "PRIVATE_ENV = \"private-value\"",
                "[mcp_servers.alpha]",
            ],
        );
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            Duration::from_secs(5),
            Duration::from_secs(1),
        )
        .expect("synthetic owned-child configuration");
        let args = config
            .args
            .iter()
            .map(|argument| argument.to_string_lossy().into_owned())
            .collect::<Vec<_>>();

        assert_eq!(
            args,
            vec![
                "app-server",
                "--listen",
                "stdio://",
                "-c",
                "model_provider=\"openai\"",
                "-c",
                "web_search=\"disabled\"",
                "--disable",
                "hooks",
                "--disable",
                "apps",
                "--disable",
                "multi_agent",
                "--disable",
                "shell_tool",
                "--disable",
                "unified_exec",
                "--disable",
                "standalone_web_search",
                "-c",
                "mcp_servers.alpha.enabled=false",
                "-c",
                "mcp_servers.zeta.enabled=false",
            ],
            "production owned children must disable apps, multi-agent tools, and every configured MCP server as well as hooks, stable shell executors, and standalone web search"
        );
        let attestation = config
            .mcp_isolation_attestation()
            .expect("production Codex config is MCP-isolated");
        assert_eq!(attestation.configured_server_count, 2);
        assert!(valid_sha256(&attestation.config_sha256));
        assert!(valid_sha256(&attestation.override_set_sha256));
        assert!(!format!("{config:?}").contains("alpha"));
    }

    #[test]
    #[cfg(unix)]
    fn mcp_server_toml_representations_produce_exact_disable_overrides() {
        let cases = [
            (
                "mcp_servers.dotted = { command = \"private-command\" }",
                "dotted",
            ),
            (
                "mcp_servers = { inline = { command = \"private-command\" } }",
                "inline",
            ),
            (
                "[ mcp_servers . whitespace ]\ncommand = \"private-command\"",
                "whitespace",
            ),
        ];
        for (toml, expected_key) in cases {
            let (directory, executable, _trace) = attested_executable_fixture("0.145.0");
            let codex_home = mcp_config_home(&directory, &[toml]);
            let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
                executable,
                codex_home,
                Duration::from_secs(1),
                Duration::from_secs(1),
            )
            .expect("TOML-equivalent direct MCP server table");
            let overrides = config
                .args
                .windows(2)
                .filter(|arguments| arguments[0] == "-c")
                .map(|arguments| arguments[1].to_string_lossy().into_owned())
                .filter(|argument| argument.starts_with("mcp_servers."))
                .collect::<Vec<_>>();
            assert_eq!(
                overrides,
                vec![format!("mcp_servers.{expected_key}.enabled=false")],
                "every supported TOML representation must produce one exact configured-server disable"
            );
        }
    }

    #[test]
    #[cfg(unix)]
    fn configured_codex_server_is_disabled_once_without_synthetic_builtin_override() {
        let (directory, executable, _trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.codex]"]);
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("managed Codex server configuration");
        let overrides = config
            .args
            .windows(2)
            .filter(|arguments| arguments[0] == "-c")
            .map(|arguments| arguments[1].to_string_lossy().into_owned())
            .filter(|argument| argument.starts_with("mcp_servers."))
            .collect::<Vec<_>>();
        assert_eq!(overrides, vec!["mcp_servers.codex.enabled=false"]);
        assert_eq!(
            config
                .mcp_isolation_attestation()
                .expect("production child has isolation attestation")
                .configured_server_count,
            1,
            "the attestation reports the configured Codex server exactly once"
        );
    }

    #[test]
    #[cfg(unix)]
    fn maximum_configured_mcp_servers_and_builtin_bridge_fit_the_process_arg_cap() {
        let (directory, executable, _trace) = attested_executable_fixture("0.145.0");
        let tables = (0..MAX_MCP_SERVER_KEYS)
            .map(|index| format!("[mcp_servers.s{index:02}]"))
            .collect::<Vec<_>>();
        let table_refs = tables.iter().map(String::as_str).collect::<Vec<_>>();
        let codex_home = mcp_config_home(&directory, &table_refs);
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("the maximum configured MCP server count must remain spawnable");
        let overrides = config
            .args
            .windows(2)
            .filter(|arguments| arguments[0] == "-c")
            .map(|arguments| arguments[1].to_string_lossy().into_owned())
            .filter(|argument| argument.starts_with("mcp_servers."))
            .collect::<Vec<_>>();
        assert_eq!(
            overrides.len(),
            MAX_MCP_SERVER_KEYS,
            "all 32 configured servers must receive exact disable overrides"
        );
        assert_eq!(config.args.len(), MAX_PROCESS_ARGS);
        let attestation = config
            .mcp_isolation_attestation()
            .expect("production child has MCP isolation attestation");
        assert_eq!(attestation.configured_server_count, MAX_MCP_SERVER_KEYS);
        assert_eq!(
            attestation.override_set_sha256,
            "5a08526fa80dc6f3652a8684200dcc87f62b1038033b9c90fd916eff07db1f72",
            "the complete sorted configured-server disable set is committed"
        );
    }

    #[test]
    #[cfg(unix)]
    fn more_than_maximum_configured_mcp_servers_refuse_before_process_construction() {
        let (directory, executable, _trace) = attested_executable_fixture("0.145.0");
        let tables = (0..=MAX_MCP_SERVER_KEYS)
            .map(|index| format!("[mcp_servers.s{index:02}]"))
            .collect::<Vec<_>>();
        let table_refs = tables.iter().map(String::as_str).collect::<Vec<_>>();
        let codex_home = mcp_config_home(&directory, &table_refs);
        assert!(
            matches!(
                OwnedAppServerConfig::codex_with_mcp_server_isolation(
                    executable,
                    codex_home,
                    Duration::from_secs(1),
                    Duration::from_secs(1),
                ),
                Err(AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigArgCap,
                ))
            ),
            "a 33rd configured MCP server must refuse before child argv construction"
        );
    }

    #[test]
    fn mcp_server_toml_non_table_or_unsafe_key_refuses() {
        for bytes in [
            b"mcp_servers = { scalar = \"private-command\" }".as_slice(),
            b"[mcp_servers.\"unsafe.key\"]\ncommand = \"private-command\"".as_slice(),
            b"mcp_servers = \"not-a-table\"".as_slice(),
        ] {
            assert_eq!(
                parse_mcp_server_keys(bytes),
                Err(AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigSemanticKeys,
                )),
                "non-table and unsafe MCP server representations cannot reach child argv"
            );
        }
    }

    #[test]
    #[cfg(unix)]
    fn mcp_server_isolation_allows_unrelated_private_codex_home_state_after_snapshot() {
        use std::os::unix::fs::PermissionsExt as _;

        let (directory, executable, trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.approved]"]);
        let qualification_state = codex_home.join("qualification-state.json");
        fs::write(
            &qualification_state,
            b"initial content-free qualification state",
        )
        .expect("write initial unrelated qualification state");
        let mut permissions = fs::metadata(&qualification_state)
            .expect("qualification state metadata")
            .permissions();
        permissions.set_mode(0o600);
        fs::set_permissions(&qualification_state, permissions)
            .expect("qualification state remains owner-private");
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home.clone(),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("isolated config snapshot");

        // A zero-turn qualification may write unrelated owner-private state
        // under CODEX_HOME.  That changes the directory timestamp but must not
        // weaken the separately full-pinned config.toml boundary.
        fs::write(
            &qualification_state,
            b"refreshed content-free qualification state",
        )
        .expect("write unrelated qualification state");
        // A qualification may also create and later remove an owner-private
        // subdirectory.  That changes the directory link count while leaving
        // its pinned structural identity and config.toml untouched.
        let qualification_directory = codex_home.join("qualification-cache");
        fs::create_dir(&qualification_directory).expect("create unrelated private directory");
        fs::remove_dir(&qualification_directory).expect("remove unrelated private directory");

        let original_isolation = config
            .mcp_isolation
            .as_ref()
            .expect("production config keeps the original isolation snapshot");
        let refreshed_isolation = McpServerIsolation::read(codex_home.clone())
            .expect("unrelated state must leave the pinned config readable");
        assert_eq!(
            refreshed_isolation.codex_home_identity, original_isolation.codex_home_identity,
            "unrelated entries may alter directory timestamps but never structural identity"
        );
        assert_eq!(
            refreshed_isolation.config_identity, original_isolation.config_identity,
            "config.toml retains its complete file identity"
        );
        assert_eq!(
            refreshed_isolation.config_sha256, original_isolation.config_sha256,
            "config.toml retains its exact content binding"
        );

        let mut transport = OwnedAppServerTransport::spawn(&config)
            .expect("unrelated CODEX_HOME state must not invalidate the pinned config");
        // The owned child writes its argv trace only after exec completes.  On
        // a heavily loaded machine the exec may be delayed well past the
        // shutdown timeout, so poll-wait for the trace with a generous
        // deadline and assert it exists before shutting down.  The shutdown
        // assertion stays separate so a load-induced exec delay can never be
        // mistaken for a failed revalidation.
        let trace_deadline = Instant::now() + Duration::from_secs(10);
        while !trace.exists() && Instant::now() < trace_deadline {
            thread::sleep(Duration::from_millis(5));
        }
        assert!(
            trace.exists(),
            "the revalidated configuration must reach the owned-child spawn"
        );
        assert!(
            transport
                .shutdown()
                .expect("owned fixture must shut down after the spawn proof")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn mcp_isolation_revalidation_uses_fixed_drift_classes() {
        let (directory, _executable, _trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.approved]"]);

        let missing_home = directory.path().join("missing-codex-home");
        assert!(matches!(
            McpServerIsolation::read(missing_home),
            Err(AppServerProcessError::PreflightConfiguration(
                PreflightConfigurationRefusal::CodexHomeStructural,
            ))
        ));

        use std::os::unix::fs::PermissionsExt as _;
        let blocked_directory = TempDir::new().expect("blocked MCP fixture directory");
        let blocked_config_home = mcp_config_home(&blocked_directory, &["[mcp_servers.blocked]"]);
        fs::set_permissions(
            blocked_config_home.join("config.toml"),
            fs::Permissions::from_mode(0o000),
        )
        .expect("block config open");
        assert!(matches!(
            McpServerIsolation::read(blocked_config_home),
            Err(AppServerProcessError::PreflightConfiguration(
                PreflightConfigurationRefusal::ConfigOpenTypeOwnerModeLink,
            ))
        ));

        let mut file_identity = McpServerIsolation::read(codex_home.clone()).expect("snapshot");
        file_identity.config_identity.inode = file_identity.config_identity.inode.saturating_add(1);
        assert_eq!(
            file_identity
                .revalidate()
                .expect_err("identity drift is refused")
                .preflight_refusal_class(),
            "config-file-identity"
        );

        let mut content = McpServerIsolation::read(codex_home.clone()).expect("snapshot");
        content.config_sha256 = "0".repeat(64);
        assert_eq!(
            content
                .revalidate()
                .expect_err("digest drift is refused")
                .preflight_refusal_class(),
            "config-content-digest"
        );

        let mut semantic = McpServerIsolation::read(codex_home.clone()).expect("snapshot");
        semantic.server_keys.push("additional".to_owned());
        assert_eq!(
            semantic
                .revalidate()
                .expect_err("semantic drift is refused")
                .preflight_refusal_class(),
            "config-semantic-keys"
        );

        let mut override_digest = McpServerIsolation::read(codex_home).expect("snapshot");
        override_digest.override_set_sha256 = "0".repeat(64);
        assert_eq!(
            override_digest
                .revalidate()
                .expect_err("override drift is refused")
                .preflight_refusal_class(),
            "config-override-digest"
        );
    }

    #[test]
    #[cfg(unix)]
    fn mcp_server_config_substitution_or_unsafe_key_refuses_before_spawn() {
        use std::os::unix::fs::PermissionsExt as _;

        let (directory, executable, trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.approved]"]);
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home.clone(),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("isolated config");
        fs::write(
            codex_home.join("config.toml"),
            "[mcp_servers.approved]\n[mcp_servers.uncovered]",
        )
        .expect("substitute config after constructor pin");
        let mut permissions = fs::metadata(codex_home.join("config.toml"))
            .expect("substituted config metadata")
            .permissions();
        permissions.set_mode(0o600);
        fs::set_permissions(codex_home.join("config.toml"), permissions)
            .expect("restore config mode");
        assert!(
            matches!(
                OwnedAppServerTransport::spawn(&config),
                Err(AppServerProcessError::PreflightConfiguration(
                    PreflightConfigurationRefusal::ConfigFileIdentity,
                ))
            ),
            "an uncovered server added after construction must prevent child spawn"
        );
        assert!(
            !trace.exists(),
            "configuration drift must be rejected before the owned child receives argv"
        );

        let unsafe_directory = TempDir::new().expect("unsafe MCP fixture directory");
        let unsafe_home = mcp_config_home(&unsafe_directory, &["[mcp_servers.\"unsafe.key\"]"]);
        let error = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            directory.path().join("codex-fixture"),
            unsafe_home,
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect_err("quoted or dotted MCP keys cannot become child command arguments");
        assert_eq!(
            error,
            AppServerProcessError::PreflightConfiguration(
                PreflightConfigurationRefusal::ConfigSemanticKeys,
            )
        );
    }

    #[test]
    #[cfg(unix)]
    fn mcp_server_config_replacement_refuses_before_spawn() {
        use std::os::unix::fs::PermissionsExt as _;

        let (directory, executable, trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.approved]"]);
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home.clone(),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("isolated config");

        let replacement = codex_home.join("replacement.toml");
        fs::write(&replacement, "[mcp_servers.approved]").expect("replacement config");
        let mut permissions = fs::metadata(&replacement)
            .expect("replacement metadata")
            .permissions();
        permissions.set_mode(0o600);
        fs::set_permissions(&replacement, permissions).expect("replacement private mode");
        fs::rename(&replacement, codex_home.join("config.toml"))
            .expect("atomically replace pinned config");

        assert!(matches!(
            OwnedAppServerTransport::spawn(&config),
            Err(AppServerProcessError::PreflightConfiguration(
                PreflightConfigurationRefusal::ConfigFileIdentity,
            ))
        ));
        assert!(
            !trace.exists(),
            "a replacement config inode must be rejected before child spawn"
        );
    }

    #[test]
    #[cfg(unix)]
    fn mismatched_owned_child_attestation_blocks_discovery_and_turns() {
        let (directory, executable, trace) = attested_executable_fixture("0.145.0");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.attested]"]);
        // Generous operation and shutdown windows: the fixture responds with a
        // version mismatch the moment it execs, so under heavy parallel load a
        // delayed child exec must time out with `UnsupportedExecutableVersion`
        // (the semantic under test) rather than a transport timeout.
        let config = OwnedAppServerConfig::codex_with_mcp_server_isolation(
            executable,
            codex_home,
            Duration::from_secs(30),
            Duration::from_secs(30),
        )
        .expect("synthetic owned-child configuration");
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn exact owned child");
        assert_eq!(
            client
                .transport
                .mcp_isolation_attestation()
                .expect("spawn records revalidated MCP isolation")
                .configured_server_count,
            1
        );
        let error = client
            .initialize()
            .expect_err("mismatched same-child attestation must fail closed");
        assert_eq!(error, AppServerProcessError::UnsupportedExecutableVersion);
        let trace = fs::read_to_string(trace).expect("owned-child invocation trace");
        assert!(trace.contains("argv:app-server --listen stdio://"));
        assert!(
            trace.contains("model_provider=\"openai\""),
            "owned child must pin the OpenAI provider for managed-account discovery"
        );
        assert!(
            trace.contains("--disable hooks"),
            "owned child must disable hooks before managed-account execution"
        );
        assert!(
            trace.contains("--disable apps"),
            "owned child must disable the stable Apps capability before managed-account execution"
        );
        assert!(
            trace.contains("--disable multi_agent"),
            "owned child must disable the stable multi-agent capability before managed-account execution"
        );
        assert!(
            trace.contains("--disable shell_tool"),
            "owned child must disable the legacy shell executor"
        );
        assert!(
            trace.contains("--disable unified_exec"),
            "owned child must disable the current shell executor"
        );
        assert!(
            trace.contains("-c web_search=\"disabled\"")
                && trace.contains("--disable standalone_web_search"),
            "owned child must pin configuration and capability shutdown for web search"
        );
        assert!(
            trace.contains("-c mcp_servers.attested.enabled=false"),
            "the actual child argv must disable every server in the pinned MCP config"
        );
        assert!(trace.contains("\"method\":\"initialize\""));
        assert!(
            !trace.contains("\"method\":\"initialized\"")
                && !trace.contains("\"method\":\"model/list\"")
                && !trace.contains("\"method\":\"thread/start\"")
                && !trace.contains("\"method\":\"turn/start\"")
        );
        assert!(
            client
                .shutdown()
                .expect("mismatch child already reaped")
                .success
        );
    }

    #[test]
    #[cfg(unix)]
    fn pinned_executable_substitution_refuses_before_spawn() {
        use std::os::unix::fs::PermissionsExt as _;
        let directory = TempDir::new().expect("temporary owned-child directory");
        let codex_home = mcp_config_home(&directory, &["[mcp_servers.approved]"]);
        let executable = directory.path().join("codex-app-server");
        fs::write(&executable, "#!/bin/sh\nexit 0\n").expect("fixture executable");
        let mut permissions = fs::metadata(&executable)
            .expect("fixture metadata")
            .permissions();
        permissions.set_mode(0o700);
        fs::set_permissions(&executable, permissions).expect("private executable mode");
        let expected = sha256_file(&executable).expect("fixture digest");
        let config = OwnedAppServerConfig::codex_pinned_with_mcp_server_isolation(
            executable.clone(),
            &expected,
            codex_home.clone(),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("pinned configuration");
        let desktop_config = OwnedAppServerConfig::codex_desktop_pinned_with_mcp_server_isolation(
            executable.clone(),
            &expected,
            codex_home,
            Duration::from_secs(1),
            Duration::from_secs(1),
        )
        .expect("pinned desktop configuration");
        let replacement = directory.path().join("replacement");
        fs::write(&replacement, "#!/bin/sh\nexit 1\n").expect("replacement executable");
        let mut replacement_permissions = fs::metadata(&replacement)
            .expect("replacement metadata")
            .permissions();
        replacement_permissions.set_mode(0o700);
        fs::set_permissions(&replacement, replacement_permissions).expect("replacement mode");
        fs::rename(&replacement, &executable).expect("atomic substitution");
        assert!(matches!(
            OwnedAppServerTransport::spawn(&config),
            Err(AppServerProcessError::InvalidConfiguration)
        ));
        assert!(matches!(
            OwnedAppServerTransport::spawn(&desktop_config),
            Err(AppServerProcessError::InvalidConfiguration)
        ));
    }

    /// Read-only compatibility probe for the explicitly reviewed Codex CLI.
    /// This test never starts a thread or turn and is both ignored and gated by
    /// an explicit environment sentinel to prevent accidental account access.
    #[test]
    #[ignore = "requires explicit live managed-account discovery authorization"]
    fn live_discovery_conforms_for_reviewed_cli_without_inference() {
        let executable = reviewed_live_executable("STEADYMUX_CODEX_APP_SERVER_LIVE_DISCOVERY");

        let config = OwnedAppServerConfig::codex(
            executable,
            Duration::from_secs(10),
            Duration::from_secs(2),
        )
        .expect("reviewed owned-child configuration");
        assert_eq!(
            config.compatible_cli_version(),
            CURRENT_STANDALONE_CLI_VERSION
        );
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn reviewed app-server");
        client.initialize().expect("initialize reviewed app-server");
        let discovery = client.discover().expect("read-only live discovery");
        assert!(discovery.account.plan_type.is_some());
        for expected in ["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna"] {
            assert!(
                discovery
                    .model_pages
                    .iter()
                    .flat_map(|page| &page.models)
                    .any(|model| model.id == expected),
                "reviewed model catalog entry is missing"
            );
        }
        client.shutdown().expect("reap reviewed app-server");
    }

    /// Bounded one-turn conformance for a separately authorized managed
    /// subscription proof. The model and effort are exact operator inputs; the
    /// prompt is fixed and carries no user content. The runtime itself fixes
    /// `approvalPolicy=never`, `sandbox=read-only`, and `ephemeral=true`.
    #[test]
    #[ignore = "requires explicit live managed-account turn authorization"]
    fn live_single_turn_conforms_for_exact_model_and_effort() {
        let executable = reviewed_live_executable("STEADYMUX_CODEX_APP_SERVER_LIVE_TURN");
        assert_eq!(
            std::env::var("STEADYMUX_CODEX_APP_SERVER_LIVE_TURN_AUTHORIZED").as_deref(),
            Ok("1"),
            "acknowledge the single live subscription turn explicitly"
        );
        let model = std::env::var("STEADYMUX_CODEX_LIVE_MODEL")
            .expect("select one exact discovered model ID");
        let effort = match std::env::var("STEADYMUX_CODEX_LIVE_EFFORT").as_deref() {
            Ok("minimal") => ReasoningEffort::Minimal,
            Ok("low") => ReasoningEffort::Low,
            Ok("medium") => ReasoningEffort::Medium,
            Ok("high") => ReasoningEffort::High,
            Ok("xhigh") => ReasoningEffort::Xhigh,
            Ok("max") => ReasoningEffort::Max,
            Ok("ultra") => ReasoningEffort::Ultra,
            _ => panic!("select one exact supported reasoning effort"),
        };
        let config = OwnedAppServerConfig::codex(
            executable,
            Duration::from_secs(120),
            Duration::from_secs(2),
        )
        .expect("reviewed owned-child configuration");
        let mut client = CodexAppServerClient::spawn(&config).expect("spawn reviewed app-server");
        let run = (|| -> Result<TurnRunReceipt, AppServerProcessError> {
            client.initialize()?;
            client.discover()?;
            let turn = client.start_turn(
                "steadymux-live-conformance",
                "Return the single token STEADYMUX_CANARY.",
                &model,
                effort,
            )?;
            client.wait_for_terminal(&turn)
        })();
        let shutdown = client.shutdown();

        let receipt = run.expect("one bounded live turn must reach a terminal result");
        let shutdown = shutdown.expect("owned live app-server must be reaped");
        assert!(
            shutdown.success,
            "owned live app-server shutdown must succeed"
        );
        assert_eq!(receipt.completion.status, TerminalStatus::Completed);
        assert_eq!(receipt.completion.selected_model, model);
        assert_eq!(receipt.completion.selected_effort, effort);
        assert!(
            receipt.completion.output.as_deref().is_some_and(
                |output| !output.trim().is_empty() && output.len() <= DEFAULT_MAX_LINE_BYTES
            ),
            "completed live turn must return nonempty bounded output"
        );
    }

    #[test]
    fn bounded_reader_rejects_oversized_and_unterminated_frames() {
        let mut oversized = BufReader::new(std::io::Cursor::new(vec![b'x'; 17]));
        assert!(matches!(
            read_bounded_line(&mut oversized, 16),
            Err(LineReadError::TooLarge)
        ));
        let mut unterminated = BufReader::new(std::io::Cursor::new(b"{}".to_vec()));
        assert!(matches!(
            read_bounded_line(&mut unterminated, 16),
            Err(LineReadError::Io)
        ));
    }

    #[test]
    #[cfg(unix)]
    fn command_arguments_cannot_carry_token_material() {
        let shell = std::path::Path::new("/bin/sh")
            .canonicalize()
            .expect("canonical shell");
        let error = OwnedAppServerConfig::new(
            shell,
            vec![OsString::from("Bearer secret-canary")],
            Duration::from_secs(1),
            Duration::from_secs(1),
            DEFAULT_MAX_LINE_BYTES,
            None,
        )
        .expect_err("token-shaped arguments are rejected");
        assert_eq!(error, AppServerProcessError::InvalidConfiguration);
        assert!(!format!("{error:?} {error}").contains("secret-canary"));
    }

    #[test]
    fn child_environment_is_explicitly_allowlisted_without_raw_tokens() {
        let mut command = Command::new(std::env::current_exe().expect("current test executable"));
        command.env("OPENAI_API_KEY", "secret-canary");
        configure_managed_child_env(&mut command, None);
        let configured_names = command
            .get_envs()
            .filter_map(|(name, value)| value.map(|_| name.to_string_lossy().into_owned()))
            .collect::<Vec<_>>();
        assert!(configured_names.iter().all(|name| {
            MANAGED_CHILD_ENV_ALLOWLIST
                .iter()
                .any(|allowed| name == allowed)
        }));
        assert!(!configured_names.iter().any(|name| name == "OPENAI_API_KEY"));
        assert!(!format!("{command:?}").contains("secret-canary"));
    }

    #[test]
    fn os_argument_validation_is_bounded() {
        assert!(!valid_arg(OsStr::new("")));
        assert!(valid_arg(OsStr::new("app-server")));
    }
}