Merge origin/main and keep V8 exec runtime behavior

This commit is contained in:
pakrym-oai
2026-03-10 19:29:53 -07:00
50 changed files with 1409 additions and 433 deletions
+1
View File
@@ -385,6 +385,7 @@ impl Codex {
if let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) = session_source
&& depth >= config.agent_max_depth
{
let _ = config.features.disable(Feature::SpawnCsv);
let _ = config.features.disable(Feature::Collab);
}
+32
View File
@@ -3965,6 +3965,7 @@ fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> {
personality: Some(Personality::Pragmatic),
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
realtime_audio: RealtimeAudioConfig::default(),
experimental_realtime_start_instructions: None,
experimental_realtime_ws_base_url: None,
experimental_realtime_ws_model: None,
experimental_realtime_ws_backend_prompt: None,
@@ -4100,6 +4101,7 @@ fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> {
personality: Some(Personality::Pragmatic),
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
realtime_audio: RealtimeAudioConfig::default(),
experimental_realtime_start_instructions: None,
experimental_realtime_ws_base_url: None,
experimental_realtime_ws_model: None,
experimental_realtime_ws_backend_prompt: None,
@@ -4233,6 +4235,7 @@ fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> {
personality: Some(Personality::Pragmatic),
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
realtime_audio: RealtimeAudioConfig::default(),
experimental_realtime_start_instructions: None,
experimental_realtime_ws_base_url: None,
experimental_realtime_ws_model: None,
experimental_realtime_ws_backend_prompt: None,
@@ -4352,6 +4355,7 @@ fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> {
personality: Some(Personality::Pragmatic),
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
realtime_audio: RealtimeAudioConfig::default(),
experimental_realtime_start_instructions: None,
experimental_realtime_ws_base_url: None,
experimental_realtime_ws_model: None,
experimental_realtime_ws_backend_prompt: None,
@@ -5261,6 +5265,34 @@ async fn feature_requirements_reject_legacy_aliases() {
);
}
#[test]
fn experimental_realtime_start_instructions_load_from_config_toml() -> std::io::Result<()> {
let cfg: ConfigToml = toml::from_str(
r#"
experimental_realtime_start_instructions = "start instructions from config"
"#,
)
.expect("TOML deserialization should succeed");
assert_eq!(
cfg.experimental_realtime_start_instructions.as_deref(),
Some("start instructions from config")
);
let codex_home = TempDir::new()?;
let config = Config::load_from_base_config_with_overrides(
cfg,
ConfigOverrides::default(),
codex_home.path().to_path_buf(),
)?;
assert_eq!(
config.experimental_realtime_start_instructions.as_deref(),
Some("start instructions from config")
);
Ok(())
}
#[test]
fn experimental_realtime_ws_base_url_loads_from_config_toml() -> std::io::Result<()> {
let cfg: ConfigToml = toml::from_str(
+9
View File
@@ -470,6 +470,10 @@ pub struct Config {
/// context appended to websocket session instructions. An empty string
/// disables startup context injection entirely.
pub experimental_realtime_ws_startup_context: Option<String>,
/// Experimental / do not use. Replaces the built-in realtime start
/// instructions inserted into developer messages when realtime becomes
/// active.
pub experimental_realtime_start_instructions: Option<String>,
/// When set, restricts ChatGPT login to a specific workspace identifier.
pub forced_chatgpt_workspace_id: Option<String>,
@@ -1241,6 +1245,10 @@ pub struct ConfigToml {
/// context appended to websocket session instructions. An empty string
/// disables startup context injection entirely.
pub experimental_realtime_ws_startup_context: Option<String>,
/// Experimental / do not use. Replaces the built-in realtime start
/// instructions inserted into developer messages when realtime becomes
/// active.
pub experimental_realtime_start_instructions: Option<String>,
pub projects: Option<HashMap<String, ProjectConfig>>,
/// Controls the web search tool mode: disabled, cached, or live.
@@ -2426,6 +2434,7 @@ impl Config {
experimental_realtime_ws_model: cfg.experimental_realtime_ws_model,
experimental_realtime_ws_backend_prompt: cfg.experimental_realtime_ws_backend_prompt,
experimental_realtime_ws_startup_context: cfg.experimental_realtime_ws_startup_context,
experimental_realtime_start_instructions: cfg.experimental_realtime_start_instructions,
forced_chatgpt_workspace_id,
forced_login_method,
include_apply_patch_tool: include_apply_patch_tool_flag,
+11 -1
View File
@@ -75,7 +75,17 @@ pub(crate) fn build_realtime_update_item(
next.realtime_active,
) {
(Some(true), false) => Some(DeveloperInstructions::realtime_end_message("inactive")),
(Some(false), true) | (None, true) => Some(DeveloperInstructions::realtime_start_message()),
(Some(false), true) | (None, true) => Some(
if let Some(instructions) = next
.config
.experimental_realtime_start_instructions
.as_deref()
{
DeveloperInstructions::realtime_start_message_with_instructions(instructions)
} else {
DeveloperInstructions::realtime_start_message()
},
),
(Some(true), true) | (Some(false), false) => None,
(None, false) => previous_turn_settings
.and_then(|settings| settings.realtime_active)
+32
View File
@@ -138,6 +138,8 @@ pub enum Feature {
EnableRequestCompression,
/// Enable collab tools.
Collab,
/// Enable CSV-backed agent job tools.
SpawnCsv,
/// Enable apps.
Apps,
/// Enable plugins.
@@ -414,6 +416,9 @@ impl Features {
}
pub(crate) fn normalize_dependencies(&mut self) {
if self.enabled(Feature::SpawnCsv) && !self.enabled(Feature::Collab) {
self.enable(Feature::Collab);
}
if self.enabled(Feature::JsReplToolsOnly) && !self.enabled(Feature::JsRepl) {
tracing::warn!("js_repl_tools_only requires js_repl; disabling js_repl_tools_only");
self.disable(Feature::JsReplToolsOnly);
@@ -693,6 +698,12 @@ pub const FEATURES: &[FeatureSpec] = &[
},
default_enabled: false,
},
FeatureSpec {
id: Feature::SpawnCsv,
key: "spawn_csv",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::Apps,
key: "apps",
@@ -997,6 +1008,27 @@ mod tests {
assert_eq!(feature_for_key("collab"), Some(Feature::Collab));
}
#[test]
fn spawn_csv_is_under_development() {
assert_eq!(Feature::SpawnCsv.stage(), Stage::UnderDevelopment);
assert_eq!(Feature::SpawnCsv.default_enabled(), false);
}
#[test]
fn spawn_csv_normalization_enables_multi_agent_one_way() {
let mut spawn_csv_features = Features::with_defaults();
spawn_csv_features.enable(Feature::SpawnCsv);
spawn_csv_features.normalize_dependencies();
assert_eq!(spawn_csv_features.enabled(Feature::SpawnCsv), true);
assert_eq!(spawn_csv_features.enabled(Feature::Collab), true);
let mut collab_features = Features::with_defaults();
collab_features.enable(Feature::Collab);
collab_features.normalize_dependencies();
assert_eq!(collab_features.enabled(Feature::Collab), true);
assert_eq!(collab_features.enabled(Feature::SpawnCsv), false);
}
#[test]
fn apps_require_feature_flag_and_chatgpt_auth() {
let mut features = Features::with_defaults();
+1
View File
@@ -687,6 +687,7 @@ fn build_guardian_subagent_config(
)?);
}
for feature in [
Feature::SpawnCsv,
Feature::Collab,
Feature::WebSearchRequest,
Feature::WebSearchCached,
+1
View File
@@ -270,6 +270,7 @@ mod agent {
// Approval policy
agent_config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
// Consolidation runs as an internal sub-agent and must not recursively delegate.
let _ = agent_config.features.disable(Feature::SpawnCsv);
let _ = agent_config.features.disable(Feature::Collab);
// Sandbox policy
+1
View File
@@ -100,6 +100,7 @@ async fn start_review_conversation(
{
panic!("by construction Constrained<WebSearchMode> must always support Disabled: {err}");
}
let _ = sub_agent_config.features.disable(Feature::SpawnCsv);
let _ = sub_agent_config.features.disable(Feature::Collab);
// Set explicit review rubric for the sub-agent
+39 -17
View File
@@ -1,5 +1,6 @@
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use crate::client_common::tools::ToolSpec;
use crate::codex::Session;
@@ -8,6 +9,7 @@ use crate::config::Config;
use crate::features::Feature;
use crate::function_tool::FunctionCallError;
use crate::tools::ToolRouter;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::context::ToolPayload;
use crate::tools::router::ToolCall;
@@ -50,7 +52,7 @@ pub(crate) fn instructions(config: &Config) -> Option<String> {
));
section.push_str("- Import nested tools from `tools.js`, for example `import { exec_command } from \"tools.js\"` or `import { tools } from \"tools.js\"`. Namespaced tools are also available from `tools/<namespace...>.js`; MCP tools use `tools/mcp/<server>.js`, for example `import { append_notebook_logs_chart } from \"tools/mcp/ologs.js\"`. `tools[name]` and identifier wrappers like `await exec_command(args)` remain available for compatibility. Nested tool calls resolve to their code-mode result values.\n");
section.push_str(&format!(
"- Import `{{ output_text, output_image, set_max_output_tokens_per_exec_call, store, load }}` from `@openai/code_mode` (or `\"openai/code_mode\"`). `output_text(value)` surfaces text back to the model and stringifies non-string objects with `JSON.stringify(...)` when possible. `output_image(imageUrl)` appends an `input_image` content item for `http(s)` or `data:` URLs. `store(key, value)` persists JSON-serializable values across `{PUBLIC_TOOL_NAME}` calls in the current session, and `load(key)` returns a cloned stored value or `undefined`. `set_max_output_tokens_per_exec_call(value)` sets the token budget used to truncate the final Rust-side result of the current `{PUBLIC_TOOL_NAME}` execution; the default is `10000`. This guards the overall `{PUBLIC_TOOL_NAME}` output, not individual nested tool invocations. When truncation happens, the final text uses the unified-exec style `Original token count:` / `Output:` wrapper and the usual `…N tokens truncated…` marker.\n",
"- Import `{{ output_text, output_image, set_max_output_tokens_per_exec_call, store, load }}` from `@openai/code_mode` (or `\"openai/code_mode\"`). `output_text(value)` surfaces text back to the model and stringifies non-string objects with `JSON.stringify(...)` when possible. `output_image(imageUrl)` appends an `input_image` content item for `http(s)` or `data:` URLs. `store(key, value)` persists JSON-serializable values across `{PUBLIC_TOOL_NAME}` calls in the current session, and `load(key)` returns a cloned stored value or `undefined`. `set_max_output_tokens_per_exec_call(value)` sets the token budget used to truncate the final Rust-side result of the current `{PUBLIC_TOOL_NAME}` execution; the default is `10000`. This guards the overall `{PUBLIC_TOOL_NAME}` output, not individual nested tool invocations. The returned content starts with a separate `Script completed` or `Script failed` text item that includes wall time. When truncation happens, the final text may include `Total output lines:` and the usual `…N tokens truncated…` marker.\n",
));
section.push_str(
"- Function tools require JSON object arguments. Freeform tools require raw strings.\n",
@@ -66,7 +68,7 @@ pub(crate) async fn execute(
turn: Arc<TurnContext>,
tracker: SharedTurnDiffTracker,
code: String,
) -> Result<Vec<FunctionCallOutputContentItem>, FunctionCallError> {
) -> Result<FunctionToolOutput, FunctionCallError> {
let exec = ExecContext {
session,
turn,
@@ -74,6 +76,7 @@ pub(crate) async fn execute(
};
let enabled_tools = build_enabled_tools(&exec).await;
let stored_values = exec.session.services.code_mode_store.stored_values().await;
let started_at = std::time::Instant::now();
let callback_exec = exec.clone();
let result = execute_code_mode(
code,
@@ -87,38 +90,57 @@ pub(crate) async fn execute(
.code_mode_store
.replace_stored_values(result.stored_values)
.await;
let items = output_content_items_from_json_values(result.content_items)
let mut items = output_content_items_from_json_values(result.content_items)
.map_err(FunctionCallError::RespondToModel)?;
Ok(truncate_code_mode_result(
if !result.success {
let error_text = result
.error_text
.unwrap_or_else(|| "JavaScript execution failed".to_string());
items.push(FunctionCallOutputContentItem::InputText {
text: format!("Script error:\n{error_text}"),
});
}
let mut items = truncate_code_mode_result(items, Some(result.max_output_tokens_per_exec_call));
prepend_script_status(&mut items, result.success, started_at.elapsed());
Ok(FunctionToolOutput::from_content(
items,
Some(result.max_output_tokens_per_exec_call),
Some(result.success),
))
}
fn prepend_script_status(
content_items: &mut Vec<FunctionCallOutputContentItem>,
success: bool,
wall_time: Duration,
) {
let wall_time_seconds = ((wall_time.as_secs_f32()) * 10.0).round() / 10.0;
let header = format!(
"{}\nWall time {wall_time_seconds:.1} seconds\nOutput:\n",
if success {
"Script completed"
} else {
"Script failed"
}
);
content_items.insert(0, FunctionCallOutputContentItem::InputText { text: header });
}
fn truncate_code_mode_result(
items: Vec<FunctionCallOutputContentItem>,
max_output_tokens_per_exec_call: Option<usize>,
) -> Vec<FunctionCallOutputContentItem> {
let max_output_tokens = resolve_max_tokens(max_output_tokens_per_exec_call);
let policy = TruncationPolicy::Tokens(max_output_tokens);
if items
.iter()
.all(|item| matches!(item, FunctionCallOutputContentItem::InputText { .. }))
{
let (mut truncated_items, original_token_count) =
formatted_truncate_text_content_items_with_policy(
&items,
TruncationPolicy::Tokens(max_output_tokens),
);
if let Some(original_token_count) = original_token_count
&& let Some(FunctionCallOutputContentItem::InputText { text }) =
truncated_items.first_mut()
{
*text = format!("Original token count: {original_token_count}\nOutput:\n{text}");
}
let (truncated_items, _) =
formatted_truncate_text_content_items_with_policy(&items, policy);
return truncated_items;
}
truncate_function_output_items_with_policy(&items, TruncationPolicy::Tokens(max_output_tokens))
truncate_function_output_items_with_policy(&items, policy)
}
async fn build_enabled_tools(exec: &ExecContext) -> Vec<EnabledTool> {
+3 -3
View File
@@ -159,7 +159,7 @@ pub struct ExecCommandToolOutput {
/// Raw bytes returned for this unified exec call before any truncation.
pub raw_output: Vec<u8>,
pub max_output_tokens: Option<usize>,
pub process_id: Option<String>,
pub process_id: Option<i32>,
pub exit_code: Option<i32>,
pub original_token_count: Option<usize>,
pub session_command: Option<Vec<String>>,
@@ -194,7 +194,7 @@ impl ToolOutput for ExecCommandToolOutput {
#[serde(skip_serializing_if = "Option::is_none")]
exit_code: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
session_id: Option<String>,
session_id: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
original_token_count: Option<usize>,
output: String,
@@ -204,7 +204,7 @@ impl ToolOutput for ExecCommandToolOutput {
chunk_id: (!self.chunk_id.is_empty()).then(|| self.chunk_id.clone()),
wall_time_seconds: self.wall_time.as_secs_f64(),
exit_code: self.exit_code,
session_id: self.process_id.clone(),
session_id: self.process_id,
original_token_count: self.original_token_count,
output: self.truncated_output(),
};
@@ -51,7 +51,6 @@ impl ToolHandler for CodeModeHandler {
}
};
let content_items = code_mode::execute(session, turn, tracker, code).await?;
Ok(FunctionToolOutput::from_content(content_items, Some(true)))
code_mode::execute(session, turn, tracker, code).await
}
}
@@ -157,6 +157,8 @@ mod spawn {
call_id: call_id.clone(),
sender_thread_id: session.conversation_id,
prompt: prompt.clone(),
model: args.model.clone().unwrap_or_default(),
reasoning_effort: args.reasoning_effort.unwrap_or_default(),
}
.into(),
)
@@ -972,6 +974,7 @@ fn apply_spawn_agent_runtime_overrides(
fn apply_spawn_agent_overrides(config: &mut Config, child_depth: i32) {
if child_depth >= config.agent_max_depth {
let _ = config.features.disable(Feature::SpawnCsv);
let _ = config.features.disable(Feature::Collab);
}
}
@@ -190,7 +190,7 @@ impl ToolHandler for UnifiedExecHandler {
)
{
let approval_policy = context.turn.approval_policy.value();
manager.release_process_id(&process_id).await;
manager.release_process_id(process_id).await;
return Err(FunctionCallError::RespondToModel(format!(
"approval policy is {approval_policy:?}; reject command — you cannot ask for escalated permissions if the approval policy is {approval_policy:?}"
)));
@@ -211,7 +211,7 @@ impl ToolHandler for UnifiedExecHandler {
) {
Ok(normalized) => normalized,
Err(err) => {
manager.release_process_id(&process_id).await;
manager.release_process_id(process_id).await;
return Err(FunctionCallError::RespondToModel(err));
}
};
@@ -228,7 +228,7 @@ impl ToolHandler for UnifiedExecHandler {
)
.await?
{
manager.release_process_id(&process_id).await;
manager.release_process_id(process_id).await;
return Ok(ExecCommandToolOutput {
event_call_id: String::new(),
chunk_id: String::new(),
@@ -271,7 +271,7 @@ impl ToolHandler for UnifiedExecHandler {
let args: WriteStdinArgs = parse_arguments(&arguments)?;
let response = manager
.write_stdin(WriteStdinRequest {
process_id: &args.session_id.to_string(),
process_id: args.session_id,
input: &args.chars,
yield_time_ms: args.yield_time_ms,
max_output_tokens: args.max_output_tokens,
+1 -1
View File
@@ -1621,7 +1621,7 @@ source: /[\s\S]+/
enabled_tool_names.join(", ")
};
let description = format!(
"Runs JavaScript in an embedded V8 runtime. This is a freeform tool: send raw JavaScript source text (no JSON/quotes/markdown fences). Direct tool calls remain available while `{PUBLIC_TOOL_NAME}` is enabled. Inside JavaScript, import nested tools from `tools.js`, for example `import {{ exec_command }} from \"tools.js\"` or `import {{ tools }} from \"tools.js\"`. Namespaced tools are also available from `tools/<namespace...>.js`; MCP tools use `tools/mcp/<server>.js`, for example `import {{ append_notebook_logs_chart }} from \"tools/mcp/ologs.js\"`. `tools[name]` and identifier wrappers like `await shell(args)` remain available for compatibility when the tool name is a valid JS identifier. Nested tool calls resolve to their code-mode result values. Import `{{ output_text, output_image, set_max_output_tokens_per_exec_call, store, load }}` from `\"@openai/code_mode\"` (or `\"openai/code_mode\"`); `output_text(value)` surfaces text back to the model and stringifies non-string objects when possible, `output_image(imageUrl)` appends an `input_image` content item for `http(s)` or `data:` URLs, `store(key, value)` persists JSON-serializable values across `{PUBLIC_TOOL_NAME}` calls in the current session, `load(key)` returns a cloned stored value or `undefined`, and `set_max_output_tokens_per_exec_call(value)` sets the token budget used to truncate the final Rust-side result of the current `{PUBLIC_TOOL_NAME}` execution. The default is `10000`. This guards the overall `{PUBLIC_TOOL_NAME}` output, not individual nested tool invocations. When truncation happens, the final text uses the unified-exec style `Original token count:` / `Output:` wrapper and the usual `…N tokens truncated…` marker. Function tools require JSON object arguments. Freeform tools require raw strings. `add_content(value)` remains available for compatibility with a content item, content-item array, or string. Structured nested-tool results should be converted to text first, for example with `JSON.stringify(...)`. Only content passed to `output_text(...)`, `output_image(...)`, or `add_content(value)` is surfaced back to the model. Enabled nested tools: {enabled_list}."
"Runs JavaScript in an embedded V8 runtime. This is a freeform tool: send raw JavaScript source text (no JSON/quotes/markdown fences). Direct tool calls remain available while `{PUBLIC_TOOL_NAME}` is enabled. Inside JavaScript, import nested tools from `tools.js`, for example `import {{ exec_command }} from \"tools.js\"` or `import {{ tools }} from \"tools.js\"`. Namespaced tools are also available from `tools/<namespace...>.js`; MCP tools use `tools/mcp/<server>.js`, for example `import {{ append_notebook_logs_chart }} from \"tools/mcp/ologs.js\"`. `tools[name]` and identifier wrappers like `await shell(args)` remain available for compatibility when the tool name is a valid JS identifier. Nested tool calls resolve to their code-mode result values. Import `{{ output_text, output_image, set_max_output_tokens_per_exec_call, store, load }}` from `\"@openai/code_mode\"` (or `\"openai/code_mode\"`); `output_text(value)` surfaces text back to the model and stringifies non-string objects when possible, `output_image(imageUrl)` appends an `input_image` content item for `http(s)` or `data:` URLs, `store(key, value)` persists JSON-serializable values across `{PUBLIC_TOOL_NAME}` calls in the current session, `load(key)` returns a cloned stored value or `undefined`, and `set_max_output_tokens_per_exec_call(value)` sets the token budget used to truncate the final Rust-side result of the current `{PUBLIC_TOOL_NAME}` execution. The default is `10000`. This guards the overall `{PUBLIC_TOOL_NAME}` output, not individual nested tool invocations. The returned content starts with a separate `Script completed` or `Script failed` text item that includes wall time. When truncation happens, the final text may include `Total output lines:` and the usual `…N tokens truncated…` marker. Function tools require JSON object arguments. Freeform tools require raw strings. `add_content(value)` remains available for compatibility with a content item, content-item array, or string. Structured nested-tool results should be converted to text first, for example with `JSON.stringify(...)`. Only content passed to `output_text(...)`, `output_image(...)`, or `add_content(value)` is surfaced back to the model. Enabled nested tools: {enabled_list}."
);
ToolSpec::Freeform(FreeformTool {
@@ -110,7 +110,7 @@ pub(crate) fn spawn_exit_watcher(
call_id: String,
command: Vec<String>,
cwd: PathBuf,
process_id: String,
process_id: i32,
transcript: Arc<Mutex<HeadTailBuffer>>,
started_at: Instant,
) {
@@ -129,7 +129,7 @@ pub(crate) fn spawn_exit_watcher(
call_id,
command,
cwd,
Some(process_id),
Some(process_id.to_string()),
transcript,
String::new(),
exit_code,
+1 -1
View File
@@ -7,7 +7,7 @@ pub(crate) enum UnifiedExecError {
CreateProcess { message: String },
// The model is trained on `session_id`, but internally we track a `process_id`.
#[error("Unknown process id {process_id}")]
UnknownProcessId { process_id: String },
UnknownProcessId { process_id: i32 },
#[error("failed to write to stdin")]
WriteToStdin,
#[error(
+15 -35
View File
@@ -86,7 +86,7 @@ impl UnifiedExecContext {
#[derive(Debug)]
pub(crate) struct ExecCommandRequest {
pub command: Vec<String>,
pub process_id: String,
pub process_id: i32,
pub yield_time_ms: u64,
pub max_output_tokens: Option<usize>,
pub workdir: Option<PathBuf>,
@@ -101,7 +101,7 @@ pub(crate) struct ExecCommandRequest {
#[derive(Debug)]
pub(crate) struct WriteStdinRequest<'a> {
pub process_id: &'a str,
pub process_id: i32,
pub input: &'a str,
pub yield_time_ms: u64,
pub max_output_tokens: Option<usize>,
@@ -109,14 +109,14 @@ pub(crate) struct WriteStdinRequest<'a> {
#[derive(Default)]
pub(crate) struct ProcessStore {
processes: HashMap<String, ProcessEntry>,
reserved_process_ids: HashSet<String>,
processes: HashMap<i32, ProcessEntry>,
reserved_process_ids: HashSet<i32>,
}
impl ProcessStore {
fn remove(&mut self, process_id: &str) -> Option<ProcessEntry> {
self.reserved_process_ids.remove(process_id);
self.processes.remove(process_id)
fn remove(&mut self, process_id: i32) -> Option<ProcessEntry> {
self.reserved_process_ids.remove(&process_id);
self.processes.remove(&process_id)
}
}
@@ -144,7 +144,7 @@ impl Default for UnifiedExecProcessManager {
struct ProcessEntry {
process: Arc<UnifiedExecProcess>,
call_id: String,
process_id: String,
process_id: i32,
command: Vec<String>,
tty: bool,
network_approval_id: Option<String>,
@@ -238,7 +238,7 @@ mod tests {
async fn write_stdin(
session: &Arc<Session>,
process_id: &str,
process_id: i32,
input: &str,
yield_time_ms: u64,
) -> Result<ExecCommandToolOutput, UnifiedExecError> {
@@ -294,11 +294,7 @@ mod tests {
let (session, turn) = test_session_and_turn().await;
let open_shell = exec_command(&session, &turn, "bash -i", 2_500).await?;
let process_id = open_shell
.process_id
.as_ref()
.expect("expected process_id")
.as_str();
let process_id = open_shell.process_id.expect("expected process_id");
write_stdin(
&session,
@@ -330,15 +326,11 @@ mod tests {
let (session, turn) = test_session_and_turn().await;
let shell_a = exec_command(&session, &turn, "bash -i", 2_500).await?;
let session_a = shell_a
.process_id
.as_ref()
.expect("expected process id")
.clone();
let session_a = shell_a.process_id.expect("expected process id");
write_stdin(
&session,
session_a.as_str(),
session_a,
"export CODEX_INTERACTIVE_SHELL_VAR=codex\n",
2_500,
)
@@ -358,11 +350,7 @@ mod tests {
let out_3 = write_stdin(
&session,
shell_a
.process_id
.as_ref()
.expect("expected process id")
.as_str(),
shell_a.process_id.expect("expected process id"),
"echo $CODEX_INTERACTIVE_SHELL_VAR\n",
2_500,
)
@@ -384,11 +372,7 @@ mod tests {
let (session, turn) = test_session_and_turn().await;
let open_shell = exec_command(&session, &turn, "bash -i", 2_500).await?;
let process_id = open_shell
.process_id
.as_ref()
.expect("expected process id")
.as_str();
let process_id = open_shell.process_id.expect("expected process id");
write_stdin(
&session,
@@ -501,11 +485,7 @@ mod tests {
let (session, turn) = test_session_and_turn().await;
let open_shell = exec_command(&session, &turn, "bash -i", 2_500).await?;
let process_id = open_shell
.process_id
.as_ref()
.expect("expected process id")
.as_str();
let process_id = open_shell.process_id.expect("expected process id");
write_stdin(&session, process_id, "exit\n", 2_500).await?;
@@ -98,41 +98,39 @@ struct PreparedProcessHandles {
cancellation_token: CancellationToken,
pause_state: Option<watch::Receiver<bool>>,
command: Vec<String>,
process_id: String,
process_id: i32,
tty: bool,
}
impl UnifiedExecProcessManager {
pub(crate) async fn allocate_process_id(&self) -> String {
pub(crate) async fn allocate_process_id(&self) -> i32 {
loop {
let mut store = self.process_store.lock().await;
let process_id = if should_use_deterministic_process_ids() {
// test or deterministic mode
let next = store
store
.reserved_process_ids
.iter()
.filter_map(|s| s.parse::<i32>().ok())
.copied()
.max()
.map(|m| std::cmp::max(m, 999) + 1)
.unwrap_or(1000);
next.to_string()
.unwrap_or(1000)
} else {
// production mode → random
rand::rng().random_range(1_000..100_000).to_string()
rand::rng().random_range(1_000..100_000)
};
if store.reserved_process_ids.contains(&process_id) {
continue;
}
store.reserved_process_ids.insert(process_id.clone());
store.reserved_process_ids.insert(process_id);
return process_id;
}
}
pub(crate) async fn release_process_id(&self, process_id: &str) {
pub(crate) async fn release_process_id(&self, process_id: i32) {
let removed = {
let mut store = self.process_store.lock().await;
store.remove(process_id)
@@ -172,7 +170,7 @@ impl UnifiedExecProcessManager {
(Arc::new(process), deferred_network_approval)
}
Err(err) => {
self.release_process_id(&request.process_id).await;
self.release_process_id(request.process_id).await;
return Err(err);
}
};
@@ -188,7 +186,7 @@ impl UnifiedExecProcessManager {
&request.command,
cwd.clone(),
ExecCommandSource::UnifiedExecStartup,
Some(request.process_id.clone()),
Some(request.process_id.to_string()),
);
emitter.emit(event_ctx, ToolEventStage::Begin).await;
@@ -227,7 +225,7 @@ impl UnifiedExecProcessManager {
let exit_code = process.exit_code();
let has_exited = process.has_exited() || exit_code.is_some();
let chunk_id = generate_chunk_id();
let process_id = request.process_id.clone();
let process_id = request.process_id;
if has_exited {
// Short‑lived command: emit ExecCommandEnd immediately using the
@@ -240,7 +238,7 @@ impl UnifiedExecProcessManager {
context.call_id.clone(),
request.command.clone(),
cwd.clone(),
Some(process_id),
Some(process_id.to_string()),
Arc::clone(&transcript),
text.clone(),
exit,
@@ -248,7 +246,7 @@ impl UnifiedExecProcessManager {
)
.await;
self.release_process_id(&request.process_id).await;
self.release_process_id(request.process_id).await;
finish_deferred_network_approval(
context.session.as_ref(),
deferred_network_approval.take(),
@@ -287,7 +285,7 @@ impl UnifiedExecProcessManager {
process_id: if has_exited {
None
} else {
Some(request.process_id.clone())
Some(request.process_id)
},
exit_code,
original_token_count: Some(original_token_count),
@@ -301,7 +299,7 @@ impl UnifiedExecProcessManager {
&self,
request: WriteStdinRequest<'_>,
) -> Result<ExecCommandToolOutput, UnifiedExecError> {
let process_id = request.process_id.to_string();
let process_id = request.process_id;
let PreparedProcessHandles {
writer_tx,
@@ -315,7 +313,7 @@ impl UnifiedExecProcessManager {
process_id,
tty,
..
} = self.prepare_process_handles(process_id.as_str()).await?;
} = self.prepare_process_handles(process_id).await?;
if !request.input.is_empty() {
if !tty {
@@ -359,7 +357,7 @@ impl UnifiedExecProcessManager {
// still alive or has exited and been removed from the store; we thread
// that through so the handler can tag TerminalInteraction with an
// appropriate process_id and exit_code.
let status = self.refresh_process_state(process_id.as_str()).await;
let status = self.refresh_process_state(process_id).await;
let (process_id, exit_code, event_call_id) = match status {
ProcessStatus::Alive {
exit_code,
@@ -372,7 +370,7 @@ impl UnifiedExecProcessManager {
}
ProcessStatus::Unknown => {
return Err(UnifiedExecError::UnknownProcessId {
process_id: request.process_id.to_string(),
process_id: request.process_id,
});
}
};
@@ -392,18 +390,18 @@ impl UnifiedExecProcessManager {
Ok(response)
}
async fn refresh_process_state(&self, process_id: &str) -> ProcessStatus {
async fn refresh_process_state(&self, process_id: i32) -> ProcessStatus {
let status = {
let mut store = self.process_store.lock().await;
let Some(entry) = store.processes.get(process_id) else {
let Some(entry) = store.processes.get(&process_id) else {
return ProcessStatus::Unknown;
};
let exit_code = entry.process.exit_code();
let process_id = entry.process_id.clone();
let process_id = entry.process_id;
if entry.process.has_exited() {
let Some(entry) = store.remove(&process_id) else {
let Some(entry) = store.remove(process_id) else {
return ProcessStatus::Unknown;
};
ProcessStatus::Exited {
@@ -426,16 +424,13 @@ impl UnifiedExecProcessManager {
async fn prepare_process_handles(
&self,
process_id: &str,
process_id: i32,
) -> Result<PreparedProcessHandles, UnifiedExecError> {
let mut store = self.process_store.lock().await;
let entry =
store
.processes
.get_mut(process_id)
.ok_or(UnifiedExecError::UnknownProcessId {
process_id: process_id.to_string(),
})?;
let entry = store
.processes
.get_mut(&process_id)
.ok_or(UnifiedExecError::UnknownProcessId { process_id })?;
entry.last_used = Instant::now();
let OutputHandles {
output_buffer,
@@ -458,7 +453,7 @@ impl UnifiedExecProcessManager {
cancellation_token,
pause_state,
command: entry.command.clone(),
process_id: entry.process_id.clone(),
process_id: entry.process_id,
tty: entry.tty,
})
}
@@ -481,7 +476,7 @@ impl UnifiedExecProcessManager {
command: &[String],
cwd: PathBuf,
started_at: Instant,
process_id: String,
process_id: i32,
tty: bool,
network_approval_id: Option<String>,
transcript: Arc<tokio::sync::Mutex<HeadTailBuffer>>,
@@ -489,7 +484,7 @@ impl UnifiedExecProcessManager {
let entry = ProcessEntry {
process: Arc::clone(&process),
call_id: context.call_id.clone(),
process_id: process_id.clone(),
process_id,
command: command.to_vec(),
tty,
network_approval_id,
@@ -499,7 +494,7 @@ impl UnifiedExecProcessManager {
let (number_processes, pruned_entry) = {
let mut store = self.process_store.lock().await;
let pruned_entry = Self::prune_processes_if_needed(&mut store);
store.processes.insert(process_id.clone(), entry);
store.processes.insert(process_id, entry);
(store.processes.len(), pruned_entry)
};
// prune_processes_if_needed runs while holding process_store; do async
@@ -526,7 +521,7 @@ impl UnifiedExecProcessManager {
context.call_id.clone(),
command.to_vec(),
cwd,
process_id.clone(),
process_id,
transcript,
started_at,
);
@@ -759,31 +754,31 @@ impl UnifiedExecProcessManager {
return None;
}
let meta: Vec<(String, Instant, bool)> = store
let meta: Vec<(i32, Instant, bool)> = store
.processes
.iter()
.map(|(id, entry)| (id.clone(), entry.last_used, entry.process.has_exited()))
.map(|(id, entry)| (*id, entry.last_used, entry.process.has_exited()))
.collect();
if let Some(process_id) = Self::process_id_to_prune_from_meta(&meta) {
return store.remove(&process_id);
return store.remove(process_id);
}
None
}
// Centralized pruning policy so we can easily swap strategies later.
fn process_id_to_prune_from_meta(meta: &[(String, Instant, bool)]) -> Option<String> {
fn process_id_to_prune_from_meta(meta: &[(i32, Instant, bool)]) -> Option<i32> {
if meta.is_empty() {
return None;
}
let mut by_recency = meta.to_vec();
by_recency.sort_by_key(|(_, last_used, _)| Reverse(*last_used));
let protected: HashSet<String> = by_recency
let protected: HashSet<i32> = by_recency
.iter()
.take(8)
.map(|(process_id, _, _)| process_id.clone())
.map(|(process_id, _, _)| *process_id)
.collect();
let mut lru = meta.to_vec();
@@ -793,7 +788,7 @@ impl UnifiedExecProcessManager {
.iter()
.find(|(process_id, _, exited)| !protected.contains(process_id) && *exited)
{
return Some(process_id.clone());
return Some(*process_id);
}
lru.into_iter()
@@ -824,7 +819,7 @@ enum ProcessStatus {
Alive {
exit_code: Option<i32>,
call_id: String,
process_id: String,
process_id: i32,
},
Exited {
exit_code: Option<i32>,
@@ -874,67 +869,64 @@ mod tests {
#[test]
fn pruning_prefers_exited_processes_outside_recently_used() {
let now = Instant::now();
let id = |n: i32| n.to_string();
let meta = vec![
(id(1), now - Duration::from_secs(40), false),
(id(2), now - Duration::from_secs(30), true),
(id(3), now - Duration::from_secs(20), false),
(id(4), now - Duration::from_secs(19), false),
(id(5), now - Duration::from_secs(18), false),
(id(6), now - Duration::from_secs(17), false),
(id(7), now - Duration::from_secs(16), false),
(id(8), now - Duration::from_secs(15), false),
(id(9), now - Duration::from_secs(14), false),
(id(10), now - Duration::from_secs(13), false),
(1, now - Duration::from_secs(40), false),
(2, now - Duration::from_secs(30), true),
(3, now - Duration::from_secs(20), false),
(4, now - Duration::from_secs(19), false),
(5, now - Duration::from_secs(18), false),
(6, now - Duration::from_secs(17), false),
(7, now - Duration::from_secs(16), false),
(8, now - Duration::from_secs(15), false),
(9, now - Duration::from_secs(14), false),
(10, now - Duration::from_secs(13), false),
];
let candidate = UnifiedExecProcessManager::process_id_to_prune_from_meta(&meta);
assert_eq!(candidate, Some(id(2)));
assert_eq!(candidate, Some(2));
}
#[test]
fn pruning_falls_back_to_lru_when_no_exited() {
let now = Instant::now();
let id = |n: i32| n.to_string();
let meta = vec![
(id(1), now - Duration::from_secs(40), false),
(id(2), now - Duration::from_secs(30), false),
(id(3), now - Duration::from_secs(20), false),
(id(4), now - Duration::from_secs(19), false),
(id(5), now - Duration::from_secs(18), false),
(id(6), now - Duration::from_secs(17), false),
(id(7), now - Duration::from_secs(16), false),
(id(8), now - Duration::from_secs(15), false),
(id(9), now - Duration::from_secs(14), false),
(id(10), now - Duration::from_secs(13), false),
(1, now - Duration::from_secs(40), false),
(2, now - Duration::from_secs(30), false),
(3, now - Duration::from_secs(20), false),
(4, now - Duration::from_secs(19), false),
(5, now - Duration::from_secs(18), false),
(6, now - Duration::from_secs(17), false),
(7, now - Duration::from_secs(16), false),
(8, now - Duration::from_secs(15), false),
(9, now - Duration::from_secs(14), false),
(10, now - Duration::from_secs(13), false),
];
let candidate = UnifiedExecProcessManager::process_id_to_prune_from_meta(&meta);
assert_eq!(candidate, Some(id(1)));
assert_eq!(candidate, Some(1));
}
#[test]
fn pruning_protects_recent_processes_even_if_exited() {
let now = Instant::now();
let id = |n: i32| n.to_string();
let meta = vec![
(id(1), now - Duration::from_secs(40), false),
(id(2), now - Duration::from_secs(30), false),
(id(3), now - Duration::from_secs(20), true),
(id(4), now - Duration::from_secs(19), false),
(id(5), now - Duration::from_secs(18), false),
(id(6), now - Duration::from_secs(17), false),
(id(7), now - Duration::from_secs(16), false),
(id(8), now - Duration::from_secs(15), false),
(id(9), now - Duration::from_secs(14), false),
(id(10), now - Duration::from_secs(13), true),
(1, now - Duration::from_secs(40), false),
(2, now - Duration::from_secs(30), false),
(3, now - Duration::from_secs(20), true),
(4, now - Duration::from_secs(19), false),
(5, now - Duration::from_secs(18), false),
(6, now - Duration::from_secs(17), false),
(7, now - Duration::from_secs(16), false),
(8, now - Duration::from_secs(15), false),
(9, now - Duration::from_secs(14), false),
(10, now - Duration::from_secs(13), true),
];
let candidate = UnifiedExecProcessManager::process_id_to_prune_from_meta(&meta);
// (10) is exited but among the last 8; we should drop the LRU outside that set.
assert_eq!(candidate, Some(id(1)));
assert_eq!(candidate, Some(1));
}
}