// 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(
¤t,
component,
OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
Mode::empty(),
)
.map_err(invalid)?,
);
}
validate_owned_directory(&fstat(¤t).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(¬ification) {
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")));
}
}