From 2f29b90eebd8fc1bbdc07cdd22ac4883f16849a5 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 26 Aug 2026 06:17:20 +0200 Subject: [PATCH] Reserve connector capacity for every caller --- src/daemon.rs | 77 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 17 ++---------- 2 files changed, 80 insertions(+), 14 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index 982a86e..ed48823 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -110,6 +110,10 @@ impl CodexDaemonConfig { { return Err(CodexDaemonError::InvalidConfig("runtime bounds")); } + let required_connections = required_connection_capacity(&self.callers)?; + if self.max_connections < required_connections { + return Err(CodexDaemonError::InvalidConfig("connection capacity")); + } let mut caller_ids = HashSet::new(); for caller in &self.callers { if caller.caller_runtime_id.trim().is_empty() @@ -132,6 +136,37 @@ impl CodexDaemonConfig { } Ok(bind) } + + pub fn admit_caller(&mut self, caller: CodexCallerConfig) -> Result<(), CodexDaemonError> { + let mut candidate = self.clone(); + if candidate + .callers + .iter() + .any(|existing| existing.caller_runtime_id == caller.caller_runtime_id) + { + return Err(CodexDaemonError::InvalidConfig("duplicate caller")); + } + candidate.max_frame_bytes = candidate.max_frame_bytes.max( + caller + .max_payload_bytes + .saturating_add(FRAME_OVERHEAD_BUDGET), + ); + candidate.callers.push(caller); + candidate.max_connections = required_connection_capacity(&candidate.callers)?; + candidate.validate()?; + *self = candidate; + Ok(()) + } +} + +fn required_connection_capacity(callers: &[CodexCallerConfig]) -> Result { + callers + .iter() + .try_fold(0_usize, |total, caller| { + total.checked_add(caller.max_concurrent_requests) + }) + .map(|total| total.max(8)) + .ok_or(CodexDaemonError::InvalidConfig("connection capacity")) } pub fn load_daemon_config(path: &Path) -> Result { @@ -494,6 +529,48 @@ mod tests { config.validate().unwrap(); } + #[test] + fn caller_admission_reserves_every_independently_admitted_request_slot() { + let root = Path::new("/srv/codex-connector"); + let first = CodexCallerConfig { + caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(), + connection_key_file: root.join("ghostlight.key"), + connection_key_epoch: 1, + allowed_models: vec!["gpt-5.6-luna".to_string()], + max_concurrent_requests: 8, + max_payload_bytes: 1_048_576, + max_output_tokens: 16_384, + }; + let mut config = CodexDaemonConfig::single_caller( + "127.0.0.1:4103", + root.join("codex"), + [7; 32], + root.join("codex-home"), + root.join("replay.cc"), + first, + ); + config + .admit_caller(CodexCallerConfig { + caller_runtime_id: "epiphany-model-runtime".to_string(), + connection_key_file: root.join("epiphany.key"), + connection_key_epoch: 1, + allowed_models: vec!["gpt-5.6-luna".to_string()], + max_concurrent_requests: 8, + max_payload_bytes: 1_048_576, + max_output_tokens: 16_384, + }) + .unwrap(); + + assert_eq!(config.max_connections, 16); + config.validate().unwrap(); + + config.max_connections = 15; + assert!(matches!( + config.validate(), + Err(CodexDaemonError::InvalidConfig("connection capacity")) + )); + } + #[test] fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() { let mut encoded = Vec::new(); diff --git a/src/main.rs b/src/main.rs index c926a6c..8ef76b6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -82,13 +82,6 @@ fn admit_caller_config(args: &[OsString]) -> Result<(), String> { let mut config = codex_connector::load_daemon_config(&PathBuf::from(source)) .map_err(|error| error.to_string())?; let caller_runtime_id = os_text(caller_runtime_id, "caller runtime")?; - if config - .callers - .iter() - .any(|caller| caller.caller_runtime_id == caller_runtime_id) - { - return Err(format!("caller {caller_runtime_id} is already admitted")); - } let caller = CodexCallerConfig { caller_runtime_id, connection_key_file: PathBuf::from(connection_key_file), @@ -98,13 +91,9 @@ fn admit_caller_config(args: &[OsString]) -> Result<(), String> { max_payload_bytes: parse_number(max_payload_bytes, "max payload bytes")?, max_output_tokens: parse_number(max_output_tokens, "max output tokens")?, }; - config.max_frame_bytes = config - .max_frame_bytes - .max(caller.max_payload_bytes.saturating_add(4096)); - config.max_connections = config - .max_connections - .max(caller.max_concurrent_requests.max(8)); - config.callers.push(caller); + config + .admit_caller(caller) + .map_err(|error| error.to_string())?; codex_connector::write_daemon_config(&PathBuf::from(destination), &config) .map_err(|error| error.to_string()) }