From fd5ba462eb06b484037d3f5ec8ba64eff4df9a4f Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Mon, 10 Aug 2026 19:09:58 +0200 Subject: [PATCH] Add controlled Codex app-server binding --- Cargo.lock | 166 +++++- Cargo.toml | 1 + src/codex_app_server.rs | 1067 +++++++++++++++++++++++++++++++++++++ src/eval_run.rs | 6 +- src/lib.rs | 1 + src/main.rs | 29 +- src/reconcile.rs | 97 +++- src/run.rs | 6 +- tests/codex_app_server.rs | 103 ++++ 9 files changed, 1465 insertions(+), 11 deletions(-) create mode 100644 src/codex_app_server.rs create mode 100644 tests/codex_app_server.rs diff --git a/Cargo.lock b/Cargo.lock index 069c3c65..3f95298e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -91,12 +91,38 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core", +] + [[package]] name = "clap" version = "4.6.3" @@ -152,6 +178,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -161,6 +193,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -171,14 +212,40 @@ dependencies = [ "typenum", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "digest" version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer", - "crypto-common", + "block-buffer 0.10.4", + "crypto-common 0.1.7", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer 0.12.1", + "const-oid", + "crypto-common 0.2.2", ] [[package]] @@ -231,6 +298,7 @@ dependencies = [ "cfg-if", "libc", "r-efi", + "rand_core", ] [[package]] @@ -245,6 +313,31 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -437,6 +530,23 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rustix" version = "1.1.4" @@ -511,6 +621,17 @@ dependencies = [ "serde_core", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", +] + [[package]] name = "sha2" version = "0.10.9" @@ -518,8 +639,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.2.17", + "digest 0.10.7", ] [[package]] @@ -538,6 +659,7 @@ dependencies = [ "sha2", "tempfile", "toml", + "tungstenite", ] [[package]] @@ -589,6 +711,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.2", +] + [[package]] name = "toml" version = "0.9.12+spec-1.1.0" @@ -628,6 +770,22 @@ version = "1.1.2+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" +[[package]] +name = "tungstenite" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e48ac77174b19c110a50ab2128b24215ac9cb40e0e12e093fb602d175c569d22" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/Cargo.toml b/Cargo.toml index 3e32ee82..e0220b25 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,6 +35,7 @@ serde_json = "1" sha2 = "0.10" tempfile = "3" toml = "0.9" +tungstenite = "0.30" [dev-dependencies] libc = "0.2" diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs new file mode 100644 index 00000000..8c94a923 --- /dev/null +++ b/src/codex_app_server.rs @@ -0,0 +1,1067 @@ +//! Controlled Codex app-server launch and persistent thread ownership. +//! +//! Native delivery cannot infer a thread from cwd, process, PTY, or `thread/list`. This module +//! starts a dedicated provider daemon, initializes an observer connection before the interactive +//! client starts, and binds a typed start or successful-resume event to the exact wrapper process +//! incarnation that owns the PTY launch. Message watching and delivery are deliberately later +//! layers; this module establishes only the topology and identity boundary they consume. + +use std::fs::{self, File, OpenOptions}; +use std::io::{Read as _, Write}; +use std::net::Shutdown; +use std::os::unix::ffi::OsStrExt as _; +use std::os::unix::fs::{FileTypeExt as _, OpenOptionsExt as _, PermissionsExt as _}; +use std::os::unix::io::AsRawFd as _; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::sync::mpsc::{self, Receiver, Sender}; +use std::thread; +use std::time::{Duration, Instant}; + +use anyhow::{Context as _, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use sha2::{Digest as _, Sha256}; +use tungstenite::{Message, WebSocket}; + +pub const SUPPORTED_CODEX_CLI_VERSION: &str = "codex-cli 0.145.0"; +const RUNTIME_SCHEMA: &str = "st2.codex-runtime.v1"; +const BINDING_SCHEMA: &str = "st2.codex-thread-binding.v1"; +const STARTUP_TIMEOUT: Duration = Duration::from_secs(30); +const CONTROL_POLL: Duration = Duration::from_millis(100); +const SOCKET_PATH_BUDGET: usize = 96; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct CodexRuntime { + schema: String, + agent: String, + runtime_id: String, + incarnation: String, +} + +impl CodexRuntime { + fn fresh(agent: String, runtime_id: String) -> Result { + Ok(Self { + schema: RUNTIME_SCHEMA.to_string(), + agent, + runtime_id, + incarnation: random_token()?, + }) + } + + pub fn agent(&self) -> &str { + &self.agent + } + + pub fn runtime_id(&self) -> &str { + &self.runtime_id + } + + pub fn incarnation(&self) -> &str { + &self.incarnation + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct CodexThreadBinding { + schema: String, + agent: String, + runtime_id: String, + runtime_incarnation: String, + thread_id: String, +} + +impl CodexThreadBinding { + fn new(runtime: &CodexRuntime, thread_id: String) -> Self { + Self { + schema: BINDING_SCHEMA.to_string(), + agent: runtime.agent.clone(), + runtime_id: runtime.runtime_id.clone(), + runtime_incarnation: runtime.incarnation.clone(), + thread_id, + } + } + + pub fn thread_id(&self) -> &str { + &self.thread_id + } + + pub fn runtime_incarnation(&self) -> &str { + &self.runtime_incarnation + } +} + +/// Run one authored Codex argv behind a dedicated app server and initialized control connection. +pub fn run_controlled( + catalog_root: &Path, + identity: String, + runtime_id: String, + codex_argv: Vec, +) -> Result<()> { + anyhow::ensure!( + !codex_argv.is_empty(), + "Codex controlled launch argv is empty" + ); + ensure_supported_version(&codex_argv[0])?; + + let state_dir = state_dir(catalog_root, &identity); + secure_dir(&state_dir)?; + let _owner_lock = acquire_owner_lock(&state_dir)?; + let binding_path = state_dir.join("binding.json"); + let resume_thread = load_resume_thread(&binding_path, &identity, &runtime_id)?; + + let socket_path = socket_path(catalog_root, &identity)?; + let socket_dir = socket_path + .parent() + .context("Codex app-server socket has no parent")?; + secure_dir(socket_dir)?; + match fs::symlink_metadata(&socket_path) { + Ok(metadata) => { + anyhow::ensure!( + metadata.file_type().is_socket(), + "Codex app-server path already exists and is not a socket: {}", + socket_path.display() + ); + match UnixStream::connect(&socket_path) { + Ok(_) => anyhow::bail!( + "Codex app-server socket {} is already live; refusing a second control owner", + socket_path.display() + ), + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound + ) => + { + fs::remove_file(&socket_path).with_context(|| { + format!("removing stale Codex socket {}", socket_path.display()) + })?; + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "checking existing Codex socket {} before launch", + socket_path.display() + ) + }); + } + } + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error) + .with_context(|| format!("checking Codex socket path {}", socket_path.display())); + } + } + + // Publish a new incarnation only after this process holds the owner lock and has proved that no + // older daemon is live. A rejected second owner must not invalidate the first owner's binding. + let runtime = CodexRuntime::fresh(identity, runtime_id)?; + atomic_json(&state_dir.join("runtime.json"), &runtime)?; + + let log = OpenOptions::new() + .create(true) + .append(true) + .mode(0o600) + .open(state_dir.join("app-server.log"))?; + let endpoint = format!("unix://{}", socket_path.display()); + let mut server = Command::new(&codex_argv[0]) + .args(["app-server", "--listen", &endpoint]) + .stdin(Stdio::null()) + .stdout(log.try_clone()?) + .stderr(log) + .spawn() + .with_context(|| format!("starting {} app-server", codex_argv[0]))?; + + let result = run_connected( + &mut server, + &socket_path, + &endpoint, + &state_dir, + &runtime, + &codex_argv, + resume_thread.as_deref(), + ); + terminate_child(&mut server); + let _ = fs::remove_file(&socket_path); + result +} + +fn run_connected( + server: &mut Child, + socket_path: &Path, + endpoint: &str, + state_dir: &Path, + runtime: &CodexRuntime, + codex_argv: &[String], + resume_thread: Option<&str>, +) -> Result<()> { + let control = connect_control(server, socket_path, STARTUP_TIMEOUT)?; + let shutdown = control.try_clone()?; + let websocket = initialize_control(control)?; + let (events_tx, events_rx) = mpsc::channel(); + let binding_path = state_dir.join("binding.json"); + let runtime_for_reader = runtime.clone(); + let expected_resume = resume_thread.map(str::to_owned); + let event_thread = thread::spawn(move || { + pump_control( + websocket, + &binding_path, + &runtime_for_reader, + expected_resume.as_deref(), + events_tx, + ) + }); + + // The initialized observer is already reading before this child can issue thread/start or + // thread/resume. Insert the remote endpoint as a global Codex option and preserve every authored + // argument after the provider executable. + let mut tui_command = Command::new(&codex_argv[0]); + tui_command.args(controlled_tui_args( + endpoint, + &codex_argv[1..], + resume_thread, + )?); + let mut tui = tui_command + .stdin(Stdio::inherit()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .with_context(|| format!("starting controlled {} TUI", codex_argv[0]))?; + + let result = wait_for_binding(&mut tui, &events_rx, STARTUP_TIMEOUT) + .and_then(|_| monitor_bound_tui(&mut tui, &events_rx)); + if result.is_err() { + terminate_child(&mut tui); + } + let _ = shutdown.shutdown(Shutdown::Both); + let _ = event_thread.join(); + result +} + +fn controlled_tui_args( + endpoint: &str, + authored_args: &[String], + resume_thread: Option<&str>, +) -> Result> { + let mut args = vec!["--remote".to_string(), endpoint.to_string()]; + let Some(thread_id) = resume_thread else { + args.extend_from_slice(authored_args); + return Ok(args); + }; + let Some(insertion) = resume_insertion_index(authored_args)? else { + args.extend_from_slice(authored_args); + return Ok(args); + }; + args.push("resume".to_string()); + // Codex models these flags on the `resume` command as well as the root command. Keep them + // before SESSION_ID so clap does not treat a following flag as the optional prompt. + args.extend_from_slice(&authored_args[..insertion]); + args.push(thread_id.to_string()); + args.extend_from_slice(&authored_args[insertion..]); + Ok(args) +} + +/// Find where a pinned Codex 0.145.0 interactive argv begins its prompt or subcommand. +/// +/// Automatic resume must insert `resume ` after global options and before the authored +/// prompt. Unknown options fail closed because guessing can turn an option value into a prompt or a +/// prompt into a session selector. `--image` is variadic, so automatic resume requires an explicit +/// `--` boundary when that option is present. +fn resume_insertion_index(authored_args: &[String]) -> Result> { + let delimiter = authored_args.iter().position(|arg| arg == "--"); + let mut index = 0; + while index < authored_args.len() { + let argument = authored_args[index].as_str(); + if argument == "--" { + return Ok(Some(index)); + } + if !argument.starts_with('-') || argument == "-" { + return if matches!(argument, "resume" | "fork") { + Ok(None) + } else { + Ok(Some(index)) + }; + } + + if matches!( + argument, + "--strict-config" + | "--oss" + | "--dangerously-bypass-approvals-and-sandbox" + | "--dangerously-bypass-hook-trust" + | "--search" + | "--no-alt-screen" + ) { + index += 1; + continue; + } + anyhow::ensure!( + !matches!(argument, "-h" | "--help" | "-V" | "--version"), + "cannot automatically resume a Codex help or version invocation" + ); + + let exact_value_option = matches!( + argument, + "-c" | "--config" + | "--enable" + | "--disable" + | "--remote-auth-token-env" + | "-m" + | "--model" + | "--local-provider" + | "-p" + | "--profile" + | "-s" + | "--sandbox" + | "-C" + | "--cd" + | "--add-dir" + | "-a" + | "--ask-for-approval" + ); + if exact_value_option { + anyhow::ensure!( + index + 1 < authored_args.len(), + "Codex option '{argument}' has no value" + ); + index += 2; + continue; + } + if matches!(argument, "-i" | "--image") + || argument.starts_with("-i=") + || argument.starts_with("--image=") + { + let boundary = delimiter.context( + "automatic Codex resume with variadic --image requires an explicit `--` prompt boundary", + )?; + return Ok(Some(boundary)); + } + + let long_value = [ + "--config=", + "--enable=", + "--disable=", + "--remote-auth-token-env=", + "--model=", + "--local-provider=", + "--profile=", + "--sandbox=", + "--cd=", + "--add-dir=", + "--ask-for-approval=", + ] + .iter() + .any(|prefix| argument.starts_with(prefix)); + let short_value = ["-c", "-m", "-p", "-s", "-C", "-a"] + .iter() + .any(|prefix| argument.starts_with(prefix) && argument.len() > prefix.len()); + anyhow::ensure!( + long_value || short_value, + "cannot automatically resume through unknown Codex option '{argument}'" + ); + index += 1; + } + Ok(Some(authored_args.len())) +} + +fn connect_control( + server: &mut Child, + socket_path: &Path, + timeout: Duration, +) -> Result { + let deadline = Instant::now() + timeout; + loop { + match UnixStream::connect(socket_path) { + Ok(stream) => return Ok(stream), + Err(error) if Instant::now() < deadline => { + if let Some(status) = server.try_wait()? { + anyhow::bail!("Codex app-server exited before control connected: {status}"); + } + if error.kind() != std::io::ErrorKind::NotFound + && error.kind() != std::io::ErrorKind::ConnectionRefused + { + return Err(error).with_context(|| { + format!("connecting Codex control socket {}", socket_path.display()) + }); + } + thread::sleep(Duration::from_millis(50)); + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "Codex control socket {} was not ready within {}s", + socket_path.display(), + timeout.as_secs() + ) + }); + } + } + } +} + +fn initialize_control(stream: UnixStream) -> Result> { + stream.set_read_timeout(Some(STARTUP_TIMEOUT))?; + let (mut websocket, response) = tungstenite::client("ws://localhost/", stream) + .map_err(|error| anyhow::anyhow!("Codex WebSocket handshake failed: {error}"))?; + anyhow::ensure!( + response.status().as_u16() == 101, + "Codex WebSocket handshake returned {}", + response.status() + ); + write_json_message( + &mut websocket, + &json!({ + "method": "initialize", + "id": 0, + "params": { + "clientInfo": { + "name": "st2", + "title": "st2", + "version": env!("CARGO_PKG_VERSION") + }, + "capabilities": { "experimentalApi": true } + } + }), + )?; + + loop { + let message = read_json_message(&mut websocket)? + .context("Codex app-server closed the control connection during initialize")?; + if message.get("id") != Some(&Value::from(0)) { + continue; + } + if let Some(error) = message.get("error") { + anyhow::bail!("Codex app-server rejected initialize: {error}"); + } + anyhow::ensure!( + message.get("result").is_some(), + "Codex app-server initialize response has no result" + ); + break; + } + write_json_message( + &mut websocket, + &json!({ "method": "initialized", "params": {} }), + )?; + websocket.get_ref().set_read_timeout(None)?; + Ok(websocket) +} + +#[derive(Debug)] +enum ControlEvent { + Bound, + Closed, + Failed(String), +} + +fn pump_control( + mut websocket: WebSocket, + binding_path: &Path, + runtime: &CodexRuntime, + expected_resume: Option<&str>, + events: Sender, +) { + let result = (|| -> Result<()> { + let mut bound_thread: Option = None; + loop { + let Some(message) = read_json_message(&mut websocket)? else { + let _ = events.send(ControlEvent::Closed); + return Ok(()); + }; + let thread_id = match message.get("method").and_then(Value::as_str) { + Some("thread/started") => message + .pointer("/params/thread/id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .context("thread/started has no non-empty params.thread.id")?, + Some("thread/status/changed") if expected_resume.is_some() => { + let thread_id = message + .pointer("/params/threadId") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .context("thread/status/changed has no non-empty params.threadId")?; + let status = message + .pointer("/params/status/type") + .and_then(Value::as_str) + .context("thread/status/changed has no params.status.type")?; + if Some(thread_id) != expected_resume || !matches!(status, "idle" | "active") { + continue; + } + thread_id + } + _ => continue, + }; + match bound_thread.as_deref() { + None => { + atomic_json( + binding_path, + &CodexThreadBinding::new(runtime, thread_id.to_string()), + )?; + bound_thread = Some(thread_id.to_string()); + let _ = events.send(ControlEvent::Bound); + } + Some(bound) if bound == thread_id => {} + // A dedicated daemon can emit secondary thread starts for review/fork flows. The + // first TUI-owned thread remains the binding; never silently rebind it. + Some(_) => {} + } + } + })(); + if let Err(error) = result { + let _ = events.send(ControlEvent::Failed(format!("{error:#}"))); + } +} + +fn wait_for_binding( + tui: &mut Child, + events: &Receiver, + timeout: Duration, +) -> Result<()> { + let deadline = Instant::now() + timeout; + loop { + if let Some(status) = tui.try_wait()? { + anyhow::bail!("controlled Codex TUI exited before thread binding: {status}"); + } + let wait = deadline + .saturating_duration_since(Instant::now()) + .min(CONTROL_POLL); + if wait.is_zero() { + anyhow::bail!( + "controlled Codex TUI did not establish typed thread ownership within {}s", + timeout.as_secs() + ); + } + match events.recv_timeout(wait) { + Ok(ControlEvent::Bound) => return Ok(()), + Ok(ControlEvent::Closed) => { + anyhow::bail!("Codex control connection closed before thread binding") + } + Ok(ControlEvent::Failed(error)) => { + anyhow::bail!("Codex control failed before thread binding: {error}") + } + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => { + anyhow::bail!("Codex control observer ended before thread binding") + } + } + } +} + +fn monitor_bound_tui(tui: &mut Child, events: &Receiver) -> Result<()> { + loop { + if let Some(status) = tui.try_wait()? { + return completed_tui(status); + } + match events.recv_timeout(CONTROL_POLL) { + Ok(ControlEvent::Bound) => {} + Ok(ControlEvent::Closed) => { + anyhow::bail!("Codex control connection closed while the TUI was live") + } + Ok(ControlEvent::Failed(error)) => { + anyhow::bail!("Codex control failed while the TUI was live: {error}") + } + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => { + anyhow::bail!("Codex control observer ended while the TUI was live") + } + } + } +} + +fn completed_tui(status: ExitStatus) -> Result<()> { + anyhow::ensure!( + status.success(), + "controlled Codex TUI exited with {status}" + ); + Ok(()) +} + +fn ensure_supported_version(codex: &str) -> Result<()> { + let output = Command::new(codex) + .arg("--version") + .output() + .with_context(|| format!("reading Codex version from {codex}"))?; + anyhow::ensure!( + output.status.success(), + "{codex} --version failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + let actual = String::from_utf8(output.stdout) + .context("Codex version output is not UTF-8")? + .trim() + .to_string(); + anyhow::ensure!( + actual == SUPPORTED_CODEX_CLI_VERSION, + "unsupported Codex app-server protocol version '{actual}' (expected '{SUPPORTED_CODEX_CLI_VERSION}')" + ); + Ok(()) +} + +pub fn state_dir(catalog_root: &Path, identity: &str) -> PathBuf { + let base = std::env::var_os("XDG_STATE_HOME") + .map(PathBuf::from) + .or_else(|| std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".local/state"))) + .unwrap_or_else(|| PathBuf::from("/tmp")); + state_dir_in(&base, catalog_root, identity) +} + +fn state_dir_in(base: &Path, catalog_root: &Path, identity: &str) -> PathBuf { + base.join("st2") + .join("codex") + .join(runtime_key(catalog_root, identity)) +} + +fn socket_path(catalog_root: &Path, identity: &str) -> Result { + let key = runtime_key(catalog_root, identity); + let preferred = std::env::var_os("XDG_RUNTIME_DIR") + .map(PathBuf::from) + .filter(|path| path.is_absolute()) + .map(|base| base.join("st2-codex").join(format!("{key}.sock"))); + if let Some(path) = preferred + && path.as_os_str().as_bytes().len() <= SOCKET_PATH_BUDGET + { + return Ok(path); + } + let path = PathBuf::from("/tmp") + .join(format!("st2-{}", unsafe { libc::geteuid() })) + .join("codex") + .join(format!("{key}.sock")); + anyhow::ensure!( + path.as_os_str().as_bytes().len() <= SOCKET_PATH_BUDGET, + "Codex app-server socket path is too long: {}", + path.display() + ); + Ok(path) +} + +fn runtime_key(catalog_root: &Path, identity: &str) -> String { + let mut hash = Sha256::new(); + for value in [catalog_root.as_os_str().as_bytes(), identity.as_bytes()] { + hash.update((value.len() as u64).to_be_bytes()); + hash.update(value); + } + let digest = format!("{:x}", hash.finalize()); + digest[..24].to_string() +} + +fn secure_dir(path: &Path) -> Result<()> { + fs::create_dir_all(path)?; + fs::set_permissions(path, fs::Permissions::from_mode(0o700))?; + Ok(()) +} + +fn acquire_owner_lock(state_dir: &Path) -> Result { + let path = state_dir.join("owner.lock"); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .mode(0o600) + .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW) + .open(&path) + .with_context(|| format!("opening Codex runtime owner lock {}", path.display()))?; + // SAFETY: `file` owns this descriptor until the returned guard is dropped. `flock` does not + // access Rust memory, and closing the descriptor releases the process-scoped lock after crash. + let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; + if result != 0 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("Codex runtime already has an owner at {}", path.display())); + } + Ok(file) +} + +fn atomic_json(path: &Path, value: &impl Serialize) -> Result<()> { + let parent = path.parent().context("state file has no parent")?; + secure_dir(parent)?; + let temp = parent.join(format!( + ".{}.{}.tmp", + path.file_name().unwrap().to_string_lossy(), + random_token()? + )); + let result = (|| -> Result<()> { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&temp)?; + serde_json::to_writer_pretty(&mut file, value)?; + file.write_all(b"\n")?; + file.sync_all()?; + fs::rename(&temp, path)?; + File::open(parent)?.sync_all()?; + Ok(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&temp); + } + result +} + +pub fn load_current_binding( + path: &Path, + runtime: &CodexRuntime, +) -> Result> { + let bytes = match fs::read(path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let binding: CodexThreadBinding = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + binding.schema == BINDING_SCHEMA, + "unsupported Codex binding schema" + ); + anyhow::ensure!( + binding.agent == runtime.agent + && binding.runtime_id == runtime.runtime_id + && binding.runtime_incarnation == runtime.incarnation, + "Codex thread binding belongs to a different runtime incarnation" + ); + Ok(Some(binding)) +} + +fn load_resume_thread(path: &Path, agent: &str, runtime_id: &str) -> Result> { + let bytes = match fs::read(path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let binding: CodexThreadBinding = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + binding.schema == BINDING_SCHEMA, + "unsupported Codex binding schema" + ); + anyhow::ensure!( + binding.agent == agent && binding.runtime_id == runtime_id, + "Codex resume binding belongs to a different agent runtime" + ); + anyhow::ensure!( + !binding.thread_id.is_empty(), + "Codex resume binding has an empty thread id" + ); + Ok(Some(binding.thread_id)) +} + +fn random_token() -> Result { + let mut bytes = [0_u8; 16]; + File::open("/dev/urandom")?.read_exact(&mut bytes)?; + Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect()) +} + +fn write_json_message(websocket: &mut WebSocket, value: &Value) -> Result<()> { + websocket.send(Message::Text(value.to_string().into()))?; + Ok(()) +} + +fn read_json_message(websocket: &mut WebSocket) -> Result> { + loop { + let message = match websocket.read() { + Ok(message) => message, + Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => { + return Ok(None); + } + Err(error) => return Err(error.into()), + }; + match message { + Message::Text(text) => { + let value = serde_json::from_str(&text) + .context("decoding Codex app-server WebSocket JSON")?; + return Ok(Some(value)); + } + Message::Close(_) => return Ok(None), + Message::Ping(_) | Message::Pong(_) => continue, + Message::Binary(_) | Message::Frame(_) => { + anyhow::bail!("Codex app-server sent a non-text WebSocket message") + } + } + } +} + +fn terminate_child(child: &mut Child) { + match child.try_wait() { + Ok(Some(_)) => {} + _ => { + let _ = child.kill(); + let _ = child.wait(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::net::UnixListener; + + #[test] + fn control_initializes_before_recording_the_first_thread_only() { + let tmp = tempfile::tempdir().unwrap(); + let socket = tmp.path().join("server.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut websocket = tungstenite::accept(stream).unwrap(); + let initialize = read_json_message(&mut websocket).unwrap().unwrap(); + assert_eq!(initialize["method"], "initialize"); + assert_eq!(initialize["params"]["clientInfo"]["name"], "st2"); + write_json_message( + &mut websocket, + &json!({ "id": 0, "result": { "userAgent": "fake" } }), + ) + .unwrap(); + let initialized = read_json_message(&mut websocket).unwrap().unwrap(); + assert_eq!(initialized["method"], "initialized"); + write_json_message( + &mut websocket, + &json!({ "method": "thread/started", "params": { "thread": { "id": "thread-main" } } }), + ) + .unwrap(); + write_json_message( + &mut websocket, + &json!({ "method": "thread/started", "params": { "thread": { "id": "thread-review" } } }), + ) + .unwrap(); + }); + + let stream = UnixStream::connect(&socket).unwrap(); + let shutdown = stream.try_clone().unwrap(); + let websocket = initialize_control(stream).unwrap(); + let state = tmp.path().join("state"); + let binding_path = state.join("binding.json"); + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let (tx, rx) = mpsc::channel(); + let runtime_for_pump = runtime.clone(); + let binding_for_pump = binding_path.clone(); + let pump = thread::spawn(move || { + pump_control(websocket, &binding_for_pump, &runtime_for_pump, None, tx) + }); + assert!(matches!( + rx.recv_timeout(Duration::from_secs(2)).unwrap(), + ControlEvent::Bound + )); + server.join().unwrap(); + let _ = shutdown.shutdown(Shutdown::Both); + pump.join().unwrap(); + + let binding = load_current_binding(&binding_path, &runtime) + .unwrap() + .unwrap(); + assert_eq!(binding.thread_id(), "thread-main"); + } + + #[test] + fn a_successfully_loaded_expected_resume_is_bound_to_the_new_incarnation() { + let tmp = tempfile::tempdir().unwrap(); + let socket = tmp.path().join("server.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut websocket = tungstenite::accept(stream).unwrap(); + let initialize = read_json_message(&mut websocket).unwrap().unwrap(); + assert_eq!(initialize["method"], "initialize"); + write_json_message( + &mut websocket, + &json!({ "id": 0, "result": { "userAgent": "fake" } }), + ) + .unwrap(); + let initialized = read_json_message(&mut websocket).unwrap().unwrap(); + assert_eq!(initialized["method"], "initialized"); + write_json_message( + &mut websocket, + &json!({ + "method": "thread/status/changed", + "params": { + "threadId": "thread-unrelated", + "status": { "type": "active", "activeFlags": [] } + } + }), + ) + .unwrap(); + write_json_message( + &mut websocket, + &json!({ + "method": "thread/status/changed", + "params": { + "threadId": "thread-prior", + "status": { "type": "idle" } + } + }), + ) + .unwrap(); + }); + + let stream = UnixStream::connect(&socket).unwrap(); + let shutdown = stream.try_clone().unwrap(); + let websocket = initialize_control(stream).unwrap(); + let binding_path = tmp.path().join("state/binding.json"); + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let (tx, rx) = mpsc::channel(); + let runtime_for_pump = runtime.clone(); + let binding_for_pump = binding_path.clone(); + let pump = thread::spawn(move || { + pump_control( + websocket, + &binding_for_pump, + &runtime_for_pump, + Some("thread-prior"), + tx, + ) + }); + assert!(matches!( + rx.recv_timeout(Duration::from_secs(2)).unwrap(), + ControlEvent::Bound + )); + server.join().unwrap(); + let _ = shutdown.shutdown(Shutdown::Both); + pump.join().unwrap(); + + let binding = load_current_binding(&binding_path, &runtime) + .unwrap() + .unwrap(); + assert_eq!(binding.thread_id(), "thread-prior"); + } + + #[test] + fn a_binding_from_another_runtime_incarnation_is_rejected() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join("binding.json"); + let prior = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let current = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + atomic_json( + &path, + &CodexThreadBinding::new(&prior, "thread-prior".into()), + ) + .unwrap(); + assert_eq!( + load_resume_thread(&path, "h.worker", "h.worker").unwrap(), + Some("thread-prior".into()), + "a validated prior binding may select resume but must not become current ownership" + ); + let error = load_current_binding(&path, ¤t).unwrap_err(); + assert!(error.to_string().contains("different runtime incarnation")); + } + + #[test] + fn controlled_tui_resumes_a_prior_binding_without_overriding_authored_selection() { + let authored = vec!["--model".into(), "gpt-test".into(), "boot".into()]; + assert_eq!( + controlled_tui_args("unix:///server.sock", &authored, None).unwrap(), + [ + "--remote", + "unix:///server.sock", + "--model", + "gpt-test", + "boot" + ] + ); + assert_eq!( + controlled_tui_args("unix:///server.sock", &authored, Some("thread-prior")).unwrap(), + [ + "--remote", + "unix:///server.sock", + "resume", + "--model", + "gpt-test", + "thread-prior", + "boot" + ] + ); + assert_eq!( + controlled_tui_args( + "unix:///server.sock", + &["resume".into(), "thread-explicit".into()], + Some("thread-prior") + ) + .unwrap(), + [ + "--remote", + "unix:///server.sock", + "resume", + "thread-explicit" + ] + ); + + let fork = vec![ + "--dangerously-bypass-hook-trust".into(), + "fork".into(), + "thread-explicit".into(), + ]; + assert_eq!( + controlled_tui_args("unix:///server.sock", &fork, Some("thread-prior")).unwrap(), + [ + "--remote", + "unix:///server.sock", + "--dangerously-bypass-hook-trust", + "fork", + "thread-explicit" + ] + ); + } + + #[test] + fn controlled_tui_resume_fails_closed_at_ambiguous_option_boundaries() { + let unknown = controlled_tui_args( + "unix:///server.sock", + &["--future-option".into(), "value".into(), "prompt".into()], + Some("thread-prior"), + ) + .unwrap_err(); + assert!(unknown.to_string().contains("unknown Codex option")); + + let image = controlled_tui_args( + "unix:///server.sock", + &["--image".into(), "one.png".into(), "prompt".into()], + Some("thread-prior"), + ) + .unwrap_err(); + assert!(image.to_string().contains("explicit `--`")); + + assert_eq!( + controlled_tui_args( + "unix:///server.sock", + &[ + "--image".into(), + "one.png".into(), + "--".into(), + "prompt".into(), + ], + Some("thread-prior"), + ) + .unwrap(), + [ + "--remote", + "unix:///server.sock", + "resume", + "--image", + "one.png", + "thread-prior", + "--", + "prompt" + ] + ); + } + + #[test] + fn state_key_is_path_and_identity_specific_without_embedding_either() { + let base = Path::new("/state"); + let first = state_dir_in(base, Path::new("/catalog/a"), "h.worker"); + let second = state_dir_in(base, Path::new("/catalog/b"), "h.worker"); + assert_ne!(first, second); + assert!(first.starts_with("/state/st2/codex")); + assert!(!first.display().to_string().contains("worker")); + assert!(!first.display().to_string().contains("catalog/a")); + } + + #[test] + fn runtime_owner_lock_is_nonblocking_and_released_on_close() { + let tmp = tempfile::tempdir().unwrap(); + let first = acquire_owner_lock(tmp.path()).unwrap(); + let error = acquire_owner_lock(tmp.path()).unwrap_err(); + assert!(error.to_string().contains("already has an owner")); + drop(first); + acquire_owner_lock(tmp.path()).unwrap(); + } +} diff --git a/src/eval_run.rs b/src/eval_run.rs index a758aabb..e3c0b9f0 100644 --- a/src/eval_run.rs +++ b/src/eval_run.rs @@ -16,7 +16,9 @@ use crate::eval_spec::{ }; use crate::expand::expand_catalog; use crate::flapping::FlappingCap; -use crate::reconcile::{TaskCompileContext, compile_generated_ding_tasks, reconcile}; +use crate::reconcile::{TaskCompileContext, compile_generated_tasks, reconcile}; +#[cfg(test)] +use crate::reconcile::compile_generated_ding_tasks; use crate::run::{Runner, SystemRunner, UpReport, detect_host, execute}; use agent_spec::spec::{AgentDesiredState, AgentSpec, JobType, Task, TaskKind, TaskLifecycle}; @@ -1094,7 +1096,7 @@ fn run_eval_inner( .collect::>(); (specs, runtime_tasks, participants, None) }; - compile_generated_ding_tasks(&mut specs, host, task_context)?; + compile_generated_tasks(&mut specs, host, task_context)?; let task_ids = runtime_tasks .iter() .map(|task| task.runtime_id.clone()) diff --git a/src/lib.rs b/src/lib.rs index 0dd18146..31541c65 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,6 +10,7 @@ pub mod agents; pub mod catalog; pub mod catalog_lock; pub mod catalog_transaction; +pub mod codex_app_server; pub mod context; pub mod ding; pub mod eval_run; diff --git a/src/main.rs b/src/main.rs index 32d1c77f..e7a570c8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -112,6 +112,19 @@ enum Command { #[arg(long, default_value_t = 1000)] interval: u64, }, + /// Internal controlled Codex launch. Generated only for `deliver "app-server"` tasks. + #[command(hide = true)] + CodexAppServer { + /// Exact agent bus identity that owns the controlled thread. + #[arg(long)] + identity: String, + /// Exact reconciled PTY task identity for this runtime. + #[arg(long)] + runtime_id: String, + /// Original structured Codex invocation, including its provider executable. + #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] + codex_argv: Vec, + }, /// Get or set an agent's presence status. No `--set` prints the status; no identity means yours /// (`$ST_AGENT`). Settable: offline | available | busy | away | dnd (`unknown` is derived). Status { @@ -779,6 +792,20 @@ fn main() -> Result<()> { host, interval, } => ding_cmd(session, identity, root, host, interval), + Command::CodexAppServer { + identity, + runtime_id, + codex_argv, + } => { + let catalog = catalog_arg(None)?; + let catalog = catalog.canonicalize().unwrap_or(catalog); + st2::codex_app_server::run_controlled( + &catalog, + identity, + runtime_id, + codex_argv, + ) + } Command::Status { identity, set, ctx } => status_cmd(identity, set, ctx), Command::Rename(args) => presentation_cmd(st2::agent_author::PresentationField::Name, args), Command::Describe(args) => { @@ -2514,7 +2541,7 @@ fn up_spec_fleet(spec_file: &Path, host: Option, once: bool, interval: u let task_context = st2::reconcile::TaskCompileContext::current(root.clone())?; st2::eval_run::prepare_spawn_env(task_context.st2_executable()); let mut specs = st2::eval_run::spec_to_agent_specs(&spec.agents, &this_host, &root); - st2::reconcile::compile_generated_ding_tasks(&mut specs, &this_host, &task_context)?; + st2::reconcile::compile_generated_tasks(&mut specs, &this_host, &task_context)?; let runner = SystemRunner::new(root.clone(), exec_state_dir(&this_host)); // One supervisor per (spec dir, host) — the same host-lock discipline as the catalog path. diff --git a/src/reconcile.rs b/src/reconcile.rs index 4688778a..d2d0b45a 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -15,7 +15,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; -use agent_spec::spec::{AgentSpec, TaskKind, TaskLifecycle}; +use agent_spec::spec::{AgentSpec, DeliveryTransport, TaskKind, TaskLifecycle}; /// Immutable inputs captured once before generated tasks are compiled. #[derive(Debug, Clone, PartialEq, Eq)] @@ -73,6 +73,20 @@ impl TaskCompileContext { pub fn st2_executable(&self) -> &Path { &self.st2_executable } + + pub fn catalog_root(&self) -> &Path { + &self.catalog_root + } +} + +/// Compile every runner-owned launch marker into an exact invocation of this st2 binary. +pub fn compile_generated_tasks( + specs: &mut [AgentSpec], + this_host: &str, + context: &TaskCompileContext, +) -> Result<()> { + compile_generated_ding_tasks(specs, this_host, context)?; + compile_app_server_agent_tasks(specs, this_host, context) } /// Replace only runner-generated DING markers with exact direct argv. Authored tasks never carry @@ -122,6 +136,87 @@ pub fn compile_generated_ding_tasks( Ok(()) } +/// Route an explicitly selected Codex native transport through st2's controlled-launch wrapper. +/// +/// The wrapper owns the provider daemon and its control connection, so it can complete the +/// initialize handshake before the interactive client is allowed to create or resume a thread. +/// App-server delivery therefore requires structured argv: rewriting opaque shell source would be +/// unsound, and an already-remote launch would have two competing control owners. +pub fn compile_app_server_agent_tasks( + specs: &mut [AgentSpec], + this_host: &str, + context: &TaskCompileContext, +) -> Result<()> { + let st2_executable = context + .st2_executable + .to_str() + .context("running st2 executable path is not UTF-8")? + .to_owned(); + let catalog_root = context + .catalog_root + .to_str() + .context("catalog root is not UTF-8")? + .to_owned(); + + for spec in specs { + if spec.delivery != Some(DeliveryTransport::AppServer) { + continue; + } + let bus_id = spec.bus_id(this_host); + let mut candidates = spec + .tasks + .iter_mut() + .filter(|task| !task.derived && task.name == "agent"); + let task = candidates.next().with_context(|| { + format!( + "agent '{bus_id}' selects `deliver \"app-server\"` but has no canonical `agent` task" + ) + })?; + anyhow::ensure!( + candidates.next().is_none(), + "agent '{bus_id}' selects `deliver \"app-server\"` with more than one canonical `agent` task" + ); + anyhow::ensure!( + task.kind == TaskKind::Pty, + "agent '{bus_id}' selects `deliver \"app-server\"` for a non-PTY canonical task" + ); + let authored = task.argv.clone().with_context(|| { + format!( + "agent '{bus_id}' selects `deliver \"app-server\"`; its canonical task must use structured `argv`, not shell `command`" + ) + })?; + anyhow::ensure!( + !authored.is_empty(), + "agent '{bus_id}' selects `deliver \"app-server\"` with an empty canonical argv" + ); + anyhow::ensure!( + !authored + .iter() + .any(|arg| arg == "--remote" || arg.starts_with("--remote=")), + "agent '{bus_id}' selects `deliver \"app-server\"` but its canonical argv already declares `--remote`" + ); + let runtime_id = task + .id + .clone() + .unwrap_or_else(|| format!("{bus_id}.{}", task.name)); + let mut argv = vec![ + st2_executable.clone(), + "--catalog".to_string(), + catalog_root.clone(), + "codex-app-server".to_string(), + "--identity".to_string(), + bus_id, + "--runtime-id".to_string(), + runtime_id, + "--".to_string(), + ]; + argv.extend(authored); + task.command = None; + task.argv = Some(argv); + } + Ok(()) +} + /// ACTUAL state: one running/known task as st2 observes it (unioned across backends). #[derive(Debug, Clone, PartialEq, Eq)] pub struct Session { diff --git a/src/run.rs b/src/run.rs index 3ec307c3..3f596daf 100644 --- a/src/run.rs +++ b/src/run.rs @@ -32,7 +32,7 @@ use crate::flapping::FlappingCap; use crate::message; use crate::reconcile::{ PtyPresentation, ReconcilePlan, Session, TaskCompileContext, TaskLaunch, TaskTarget, - compile_generated_ding_tasks, + compile_generated_tasks, }; use crate::task_inventory::{ DesiredRuntime, ObservationBatch, ObservedState, RuntimeGeneration, RuntimeObservation, @@ -1419,7 +1419,7 @@ fn reconcile_pass( .filter(|spec| !materialized.failed_agents.contains(&spec.bus_id(this_host))) .cloned() .collect(); - if let Err(error) = compile_generated_ding_tasks(&mut eligible_specs, this_host, task_context) { + if let Err(error) = compile_generated_tasks(&mut eligible_specs, this_host, task_context) { report.skipped = true; report .errors @@ -1700,7 +1700,7 @@ where crate::reconcile::validate_task_identities(specs, this_host)?; let task_context = TaskCompileContext::current(catalog_root.to_path_buf())?; let mut compiled_specs = specs.to_vec(); - compile_generated_ding_tasks(&mut compiled_specs, this_host, &task_context)?; + compile_generated_tasks(&mut compiled_specs, this_host, &task_context)?; let sessions = runner .list_sessions() .map_err(|e| anyhow::anyhow!("list sessions: {e}"))?; diff --git a/tests/codex_app_server.rs b/tests/codex_app_server.rs new file mode 100644 index 00000000..b0f7d2f9 --- /dev/null +++ b/tests/codex_app_server.rs @@ -0,0 +1,103 @@ +use std::fs; +use std::path::Path; + +use st2::DeliveryTransport; +use st2::reconcile::{TaskCompileContext, compile_generated_tasks}; + +fn write(path: &Path, body: &str) { + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, body).unwrap(); +} + +fn context(root: &Path) -> TaskCompileContext { + let executable = root.join("bin/st2"); + write(&executable, "test binary"); + TaskCompileContext::new(root.to_path_buf(), executable).unwrap() +} + +#[test] +fn app_server_selector_wraps_the_canonical_argv_with_exact_owner_inputs() { + let tmp = tempfile::tempdir().unwrap(); + let declaration = tmp.path().join("agents/h/worker/agent.kdl"); + write( + &declaration, + r#"agent "worker" { + host "h" + deliver "app-server" + argv "codex" "--model" "gpt-test" "boot" +} +"#, + ); + let mut found = st2::discover(tmp.path()); + assert!(found.errors.is_empty(), "{:?}", found.errors); + + compile_generated_tasks(&mut found.specs, "h", &context(tmp.path())).unwrap(); + + let spec = &found.specs[0]; + assert_eq!(spec.delivery, Some(DeliveryTransport::AppServer)); + let task = spec.tasks.iter().find(|task| task.name == "agent").unwrap(); + assert_eq!(task.command, None); + assert_eq!( + task.argv.as_deref(), + Some( + [ + tmp.path().join("bin/st2").display().to_string(), + "--catalog".into(), + tmp.path().display().to_string(), + "codex-app-server".into(), + "--identity".into(), + "h.worker".into(), + "--runtime-id".into(), + "h.worker".into(), + "--".into(), + "codex".into(), + "--model".into(), + "gpt-test".into(), + "boot".into(), + ] + .as_slice() + ) + ); +} + +#[test] +fn app_server_selector_rejects_shell_and_pre_remote_launches_without_mutating_them() { + for (name, launch, expected) in [ + ( + "shell", + "command \"exec codex\"", + "must use structured `argv`", + ), + ( + "remote", + "argv \"codex\" \"--remote\" \"unix:///other.sock\"", + "already declares `--remote`", + ), + ] { + let tmp = tempfile::tempdir().unwrap(); + write( + &tmp.path().join(format!("agents/h/{name}/agent.kdl")), + &format!("agent \"{name}\" {{ host \"h\"; deliver \"app-server\"; {launch} }}"), + ); + let mut found = st2::discover(tmp.path()); + assert!(found.errors.is_empty(), "{name}: {:?}", found.errors); + let before = found.specs.clone(); + let error = + compile_generated_tasks(&mut found.specs, "h", &context(tmp.path())).unwrap_err(); + assert!(error.to_string().contains(expected), "{error:#}"); + assert_eq!(found.specs, before, "{name} compile failure mutated source"); + } +} + +#[test] +fn mcp_selector_does_not_rewrite_the_authored_launch() { + let tmp = tempfile::tempdir().unwrap(); + write( + &tmp.path().join("agents/h/worker/agent.kdl"), + r#"agent "worker" { host "h"; deliver "mcp"; argv "claude" "boot" }"#, + ); + let mut found = st2::discover(tmp.path()); + let before = found.specs.clone(); + compile_generated_tasks(&mut found.specs, "h", &context(tmp.path())).unwrap(); + assert_eq!(found.specs, before); +}