openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge refreshed PSEC-4922 base into approval integrity

Browse Source
This commit is contained in:
Chris Bookholt
2026-07-02 19:09:32 -07:00
parent bfb7344449 9f9db61ddf
commit 4a312592ba
64 changed files with 2459 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MODULE.bazel.lock
Generated
+4 -3
View File
File diff suppressed because one or more lines are too long
README.md
+1 -1
View File
@@ -21,7 +21,7 @@ curl -fsSL https://chatgpt.com/codex/install.sh | sh
Run the following on Windows to install Codex CLI:
```
```shell
powershell -ExecutionPolicy ByPass -c "irm https://chatgpt.com/codex/install.ps1 | iex"
```
codex-rs/.cargo/audit.toml
+3 -1
View File
@@ -1,5 +1,5 @@
[advisories]
# Reviewed 2026-06-07. Keep this list in sync with ../deny.toml.
# Reviewed 2026-07-02. Keep this list in sync with ../deny.toml.
ignore = [
"RUSTSEC-2024-0388", # derivative 2.2.0 via starlark/starlark_syntax; upstream crate is unmaintained
"RUSTSEC-2025-0057", # fxhash 0.2.1 via starlark_map/bm25; upstream crate is unmaintained
@@ -10,4 +10,6 @@ ignore = [
"RUSTSEC-2026-0118", # hickory-proto via rama-dns/rama-tcp; remove when rama updates to hickory 0.26.1 or hickory-net
"RUSTSEC-2026-0119", # hickory-proto via rama-dns/rama-tcp; remove when rama updates to hickory 0.26.1 or hickory-net
"RUSTSEC-2026-0173", # proc-macro-error2 via i18n-embed-fl/age/codex-secrets; remove when local secrets storage migrates off age or age drops i18n-embed-fl
"RUSTSEC-2026-0194", # quick-xml via plist/syntect and wayland-scanner; trusted inputs only; remove when rust-plist#191 and wayland-rs#938 are released
"RUSTSEC-2026-0195", # quick-xml via plist/syntect and wayland-scanner; trusted inputs only; remove when rust-plist#191 and wayland-rs#938 are released
]
codex-rs/Cargo.lock
Generated
+18 -9
View File
@@ -3712,7 +3712,7 @@ dependencies = [
"icu_provider",
"landlock",
"pretty_assertions",
"quick-xml",
"quick-xml 0.41.0",
"reqwest 0.12.28",
"schemars 0.8.22",
"seccompiler",
@@ -10272,13 +10272,13 @@ checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
[[package]]
name = "plist"
version = "1.8.0"
version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "740ebea15c5d1428f910cd1a5f52cebf8d25006245ed8ade92702f4943d91e07"
checksum = "092791278e026273c1b65bbdcfbba3a300f2994c896bd01ab01da613c29c46f1"
dependencies = [
"base64 0.22.1",
"indexmap 2.14.0",
"quick-xml",
"quick-xml 0.39.4",
"serde",
"time",
]
@@ -10711,9 +10711,18 @@ checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
[[package]]
name = "quick-xml"
version = "0.38.4"
version = "0.39.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e"
dependencies = [
"memchr",
]
[[package]]
name = "quick-xml"
version = "0.41.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1"
dependencies = [
"memchr",
"serde",
@@ -14674,12 +14683,12 @@ dependencies = [
[[package]]
name = "wayland-scanner"
version = "0.31.8"
version = "0.31.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5423e94b6a63e68e439803a3e153a9252d5ead12fd853334e2ad33997e3889e3"
checksum = "9c324a910fd86ebdc364a3e61ec1f11737d3b1d6c273c0239ee8ff4bc0d24b4a"
dependencies = [
"proc-macro2",
"quick-xml",
"quick-xml 0.39.4",
"quote",
]
codex-rs/Cargo.toml
+1 -1
View File
@@ -362,7 +362,7 @@ portable-pty = "0.9.0"
predicates = "3"
pretty_assertions = "1.4.1"
pulldown-cmark = "0.10"
quick-xml = "0.38.4"
quick-xml = "0.41.0"
rand = "0.9"
ratatui = "0.29.0"
ratatui-macros = "0.6.0"
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+23 -7
View File
@@ -1786,13 +1786,29 @@
"type": "object"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+23 -7
View File
@@ -2713,13 +2713,29 @@
"type": "object"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+23 -7
View File
@@ -12860,13 +12860,29 @@
"type": "object"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+23 -7
View File
@@ -9264,13 +9264,29 @@
"type": "object"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json
Generated
+23 -7
View File
@@ -681,13 +681,29 @@
]
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json
Generated
+23 -7
View File
@@ -681,13 +681,29 @@
]
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/ThreadSettingsUpdatedNotification.json
Generated
+23 -7
View File
@@ -108,13 +108,29 @@
"type": "string"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json
Generated
+23 -7
View File
@@ -195,13 +195,29 @@
"type": "string"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"Personality": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json
Generated
+23 -7
View File
@@ -681,13 +681,29 @@
]
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json
Generated
+23 -7
View File
@@ -141,13 +141,29 @@
"type": "string"
},
"MultiAgentMode": {
"description": "Controls whether the model receives multi-agent delegation instructions and, when it does, whether it should only spawn sub-agents after an explicit user request or may delegate proactively when doing so would help. `none` leaves the multi-agent tools available without injecting delegation instructions.",
"enum": [
"none",
"explicitRequestOnly",
"proactive"
],
"type": "string"
"description": "Controls the effective multi-agent delegation instructions for a turn. `custom` means the configured mode hint defines the policy instead of a built-in policy.",
"oneOf": [
{
"enum": [
"explicitRequestOnly",
"proactive"
],
"type": "string"
},
{
"additionalProperties": false,
"properties": {
"custom": {
"type": "string"
}
},
"required": [
"custom"
],
"title": "CustomMultiAgentMode",
"type": "object"
}
]
},
"NetworkAccess": {
"enum": [
codex-rs/app-server-protocol/schema/typescript/MultiAgentMode.ts
Generated
+3 -5
View File
@@ -3,9 +3,7 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
/**
* Controls whether the model receives multi-agent delegation instructions and,
* when it does, whether it should only spawn sub-agents after an explicit user
* request or may delegate proactively when doing so would help. `none` leaves
* the multi-agent tools available without injecting delegation instructions.
* Controls the effective multi-agent delegation instructions for a turn. `custom` means the
* configured mode hint defines the policy instead of a built-in policy.
*/
export type MultiAgentMode = "none" | "explicitRequestOnly" | "proactive";
export type MultiAgentMode = { "custom": string } | "explicitRequestOnly" | "proactive";
codex-rs/app-server/tests/common/json_logging.rs
+94 -12
View File
@@ -1,11 +1,79 @@
use std::path::Path;
use std::process::Command;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use serde_json::Value;
use serde_json::json;
use tokio::sync::Notify;
#[derive(Clone, Default)]
pub(crate) struct JsonLogCapture {
lines: Arc<Mutex<Vec<String>>>,
updated: Arc<Notify>,
}
impl JsonLogCapture {
pub(crate) fn record(&self, line: String) {
self.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(line);
self.updated.notify_one();
}
pub(crate) async fn wait_for_event(&self, event_name: &str) -> Result<Value> {
let mut events = self.wait_for_events(event_name, /*count*/ 1).await?;
Ok(events.remove(0))
}
pub(crate) async fn wait_for_events(
&self,
event_name: &str,
count: usize,
) -> Result<Vec<Value>> {
let result = tokio::time::timeout(Duration::from_secs(10), async {
loop {
let updated = self.updated.notified();
let events = self
.events()?
.into_iter()
.filter(|event| event["fields"]["event.name"].as_str() == Some(event_name))
.collect::<Vec<_>>();
if events.len() >= count {
return Ok(events);
}
updated.await;
}
})
.await;
match result {
Ok(result) => result,
Err(_) => {
let lines = self
.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.join("\n");
anyhow::bail!(
"timed out waiting for {count} JSON log event(s) named `{event_name}`; captured stderr:\n{lines}"
)
}
}
}
pub(crate) fn events(&self) -> Result<Vec<Value>> {
let lines = self
.lines
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
json_log_events(lines.iter().map(String::as_str))
}
}
pub fn app_server_json_shutdown_event(
binary: &str,
@@ -31,25 +99,39 @@ pub fn app_server_json_shutdown_event(
let stderr = String::from_utf8(output.stderr)?;
anyhow::ensure!(output.status.success(), "app-server failed: {stderr}");
let events = stderr
.lines()
.filter(|line| !line.is_empty())
.map(serde_json::from_str::<Value>)
.collect::<serde_json::Result<Vec<_>>>()
.with_context(|| format!("app-server stderr was not JSONL: {stderr}"))?;
let events = json_log_events(stderr.lines())
.with_context(|| format!("app-server stderr was not valid JSONL: {stderr}"))?;
let event = events
.iter()
.find(|event| event["fields"]["message"] == "processor task exited")
.context("missing INFO shutdown event in app-server JSON logs")?;
let timestamp = event["timestamp"]
.as_str()
.context("shutdown event did not include a timestamp")?;
chrono::DateTime::parse_from_rfc3339(timestamp)
.with_context(|| format!("shutdown event timestamp was not RFC 3339: {timestamp}"))?;
Ok(json!({
"level": event["level"],
"fields": event["fields"],
"target": event["target"],
}))
}
fn json_log_events<'a>(lines: impl IntoIterator<Item = &'a str>) -> Result<Vec<Value>> {
lines
.into_iter()
.filter(|line| !line.is_empty())
.map(|line| {
let event = serde_json::from_str::<Value>(line)
.with_context(|| format!("log line was not JSON: {line}"))?;
anyhow::ensure!(
event["level"].is_string()
&& event["fields"].is_object()
&& event["target"].is_string(),
"JSON log event did not include level, fields, and target: {line}"
);
let timestamp = event["timestamp"]
.as_str()
.with_context(|| format!("JSON log event did not include a timestamp: {line}"))?;
chrono::DateTime::parse_from_rfc3339(timestamp).with_context(|| {
format!("JSON log event timestamp was not RFC 3339: {timestamp}")
})?;
Ok(event)
})
.collect()
}
codex-rs/app-server/tests/common/test_app_server.rs
+56 -1
View File
@@ -123,6 +123,8 @@ use core_test_support::test_codex::TestEnv;
use core_test_support::test_codex::test_env;
use tokio::process::Command;
use crate::json_logging::JsonLogCapture;
pub struct TestAppServer {
next_request_id: AtomicI64,
/// Retain this child process until the client is dropped. The Tokio runtime
@@ -134,6 +136,7 @@ pub struct TestAppServer {
stdout: BufReader<ChildStdout>,
pending_messages: VecDeque<JSONRPCMessage>,
auto_env: Option<TestEnv>,
json_logs: JsonLogCapture,
}
pub const DEFAULT_CLIENT_NAME: &str = "codex-app-server-tests";
@@ -160,6 +163,31 @@ impl TestAppServer {
/// URL-based configuration, this helper rejects a `codex_home` containing
/// that file.
pub async fn new_with_auto_env(codex_home: &Path) -> anyhow::Result<Self> {
Self::new_with_auto_env_and_env(codex_home, &[]).await
}
/// Starts an auto-environment app server that emits JSON logs.
///
/// `rust_log` is the value to use for the `RUST_LOG` environment variable.
pub async fn new_with_auto_env_and_json_logging(
codex_home: &Path,
rust_log: impl Into<String>,
) -> anyhow::Result<Self> {
let rust_log = rust_log.into();
Self::new_with_auto_env_and_env(
codex_home,
&[
("LOG_FORMAT", Some("json")),
("RUST_LOG", Some(rust_log.as_str())),
],
)
.await
}
async fn new_with_auto_env_and_env(
codex_home: &Path,
extra_env_overrides: &[(&str, Option<&str>)],
) -> anyhow::Result<Self> {
let environments_toml = codex_home.join("environments.toml");
ensure!(
!environments_toml
@@ -172,7 +200,7 @@ impl TestAppServer {
let auto_env = test_env().await?;
// Noise registry configuration takes precedence over the URL-based
// provider, so clear inherited values to keep the selection hermetic.
let env_overrides = [
let mut env_overrides = vec![
(
CODEX_EXEC_SERVER_URL_ENV_VAR,
auto_env.environment().exec_server_url(),
@@ -182,6 +210,7 @@ impl TestAppServer {
(CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, None),
(CODEX_EXEC_SERVER_NOISE_CHATGPT_ACCOUNT_ID_ENV_VAR, None),
];
env_overrides.extend_from_slice(extra_env_overrides);
let mut app_server = Self::new_with_env(codex_home, &env_overrides).await?;
app_server.auto_env = Some(auto_env);
Ok(app_server)
@@ -202,6 +231,28 @@ impl TestAppServer {
})
}
/// Waits for a JSON stderr event whose structured `event.name` field matches.
pub async fn wait_for_json_log_event(
&self,
event_name: &str,
) -> anyhow::Result<serde_json::Value> {
self.json_logs.wait_for_event(event_name).await
}
/// Waits for the requested number of JSON stderr events with the same `event.name` field.
pub async fn wait_for_json_log_events(
&self,
event_name: &str,
count: usize,
) -> anyhow::Result<Vec<serde_json::Value>> {
self.json_logs.wait_for_events(event_name, count).await
}
/// Returns every stderr line parsed and validated as a JSON log event.
pub fn json_log_events(&self) -> anyhow::Result<Vec<serde_json::Value>> {
self.json_logs.events()
}
pub async fn new_without_managed_config(codex_home: &Path) -> anyhow::Result<Self> {
Self::new_with_env(codex_home, &[(DISABLE_MANAGED_CONFIG_ENV_VAR, Some("1"))]).await
}
@@ -322,10 +373,13 @@ impl TestAppServer {
// Forward child's stderr to our stderr so failures are visible even
// when stdout/stderr are captured by the test harness.
let json_logs = JsonLogCapture::default();
if let Some(stderr) = process.stderr.take() {
let json_logs = json_logs.clone();
let mut stderr_reader = BufReader::new(stderr).lines();
tokio::spawn(async move {
while let Ok(Some(line)) = stderr_reader.next_line().await {
json_logs.record(line.clone());
eprintln!("[mcp stderr] {line}");
}
});
@@ -337,6 +391,7 @@ impl TestAppServer {
stdout,
pending_messages: VecDeque::new(),
auto_env: None,
json_logs,
})
}
codex-rs/app-server/tests/suite/logging.rs
+145
View File
@@ -1,8 +1,30 @@
use anyhow::Context;
use anyhow::Result;
use app_test_support::TestAppServer;
use app_test_support::app_server_json_shutdown_event;
use app_test_support::create_exec_command_sse_response;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::to_response;
use app_test_support::write_mock_responses_config_toml;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use codex_features::Feature;
use core_test_support::skip_if_no_network;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::collections::BTreeMap;
use tempfile::TempDir;
use tokio::time::Duration;
use tokio::time::timeout;
const READ_TIMEOUT: Duration = Duration::from_secs(10);
#[test]
fn standalone_app_server_emits_json_info_events() -> Result<()> {
@@ -25,3 +47,126 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn app_server_emits_structured_tool_call_timing_event() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = create_mock_responses_server_sequence(vec![
create_exec_command_sse_response("exec-call-1")?,
create_final_assistant_message_sse_response("done")?,
])
.await;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml(
codex_home.path(),
&server.uri(),
&BTreeMap::from([(Feature::UnifiedExec, true)]),
/*auto_compact_limit*/ 100_000,
/*requires_openai_auth*/ None,
"mock_provider",
"compact",
)?;
let mut app_server = TestAppServer::new_with_auto_env_and_json_logging(
codex_home.path(),
"warn,codex_core::tools::parallel=info",
)
.await?;
timeout(READ_TIMEOUT, app_server.initialize()).await??;
let thread_start_id = app_server
.send_thread_start_request_with_auto_env(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let thread_start_response: JSONRPCResponse = timeout(
READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?;
let turn_start_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "run a command".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_start_response: JSONRPCResponse = timeout(
READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
)
.await??;
let TurnStartResponse { turn } = to_response(turn_start_response)?;
timeout(
READ_TIMEOUT,
app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
let mut tool_call = app_server
.wait_for_json_log_event("codex.tool_call")
.await?;
let tool_call_object = tool_call
.as_object_mut()
.context("tool call log event must be an object")?;
// JsonLogCapture already validates the timestamp as RFC 3339.
tool_call_object
.remove("timestamp")
.context("tool call log event must include a timestamp")?;
let fields = tool_call_object
.get_mut("fields")
.and_then(Value::as_object_mut)
.context("tool call log event fields must be an object")?;
let trace_id = fields
.remove("trace_id")
.context("tool call log event must include trace_id")?;
anyhow::ensure!(trace_id.is_string(), "trace_id must be a string");
let dispatch_duration_ms = fields
.remove("dispatch_duration_ms")
.and_then(|duration| duration.as_u64())
.context("dispatch_duration_ms must be a nonnegative integer")?;
let handler_duration_ms = fields
.remove("handler_duration_ms")
.and_then(|duration| duration.as_u64())
.context("handler_duration_ms must be a nonnegative integer")?;
let total_duration_ms = fields
.remove("total_duration_ms")
.and_then(|duration| duration.as_u64())
.context("total_duration_ms must be a nonnegative integer")?;
let accounted_duration_ms = dispatch_duration_ms
.checked_add(handler_duration_ms)
.context("dispatch and handler durations must not overflow")?;
anyhow::ensure!(
total_duration_ms >= accounted_duration_ms
&& total_duration_ms - accounted_duration_ms <= 1,
"dispatch and handler durations must account for total duration within integer truncation"
);
assert_eq!(
tool_call,
json!({
"level": "INFO",
"fields": {
"message": "tool call completed",
"event.name": "codex.tool_call",
"conversation.id": thread.id,
"turn_id": turn.id,
"tool_name": "exec_command",
"call_id": "exec-call-1",
"tool_source": "direct",
"execution_started": true,
},
"target": "codex_core::tools::parallel",
})
);
Ok(())
}
codex-rs/codex-api/src/endpoint/responses_websocket.rs
-3
View File
@@ -43,7 +43,6 @@ use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::instrument;
use tracing::trace;
use tungstenite::extensions::ExtensionsConfig;
use tungstenite::extensions::compression::deflate::DeflateConfig;
use tungstenite::protocol::WebSocketConfig;
@@ -782,8 +781,6 @@ async fn send_websocket_request(
telemetry: Option<&Arc<dyn WebsocketTelemetry>>,
connection_reused: bool,
) -> Result<(), ApiError> {
trace!("websocket request: {request_text}");
let request_start = Instant::now();
let result = tokio::time::timeout(
idle_timeout,
codex-rs/core/config.schema.json
+3
View File
@@ -1750,6 +1750,9 @@
"minimum": 0.0,
"type": "integer"
},
"multi_agent_mode_hint_text": {
"type": "string"
},
"non_code_mode_only": {
"type": "boolean"
},
codex-rs/core/src/agent/control.rs
+78 -37
View File
@@ -4,6 +4,8 @@ use crate::agent::registry::AgentRegistry;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::resolve_role_config;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::codex_thread::ThreadConfigSnapshot;
use crate::config::Config;
use crate::config::RolloutBudgetConfig;
@@ -139,12 +141,12 @@ impl AgentControl {
pub(crate) async fn send_input(
&self,
agent_id: ThreadId,
initial_operation: Op,
input: Vec<UserInput>,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_op(agent_id, &initial_operation)
self.ensure_execution_capacity_for_turn_start(agent_id, /*starts_turn*/ true)
.await?;
self.send_input_after_capacity_check(agent_id, &state, initial_operation)
self.send_input_after_capacity_check(agent_id, &state, input)
.await
}
@@ -152,19 +154,14 @@ impl AgentControl {
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
initial_operation: Op,
input: Vec<UserInput>,
) -> CodexResult<String> {
let last_task_message = match &initial_operation {
Op::InterAgentCommunication { communication } => {
last_task_message_from_communication(communication)
}
_ => non_empty_task_message(render_input_preview(&initial_operation)),
};
let last_task_message = non_empty_task_message(render_input_preview(&input));
let result = self
.handle_thread_request_result(
agent_id,
state,
state.send_op(agent_id, initial_operation).await,
state.send_op(agent_id, input.into()).await,
)
.await;
if result.is_ok() {
@@ -182,14 +179,60 @@ impl AgentControl {
&self,
agent_id: ThreadId,
communication: InterAgentCommunication,
agent_communication_context: AgentCommunicationContext,
) -> CodexResult<String> {
let state = self.upgrade()?;
self.ensure_execution_capacity_for_turn_start(agent_id, communication.trigger_turn)
.await?;
self.send_inter_agent_communication_after_capacity_check(
agent_id,
&state,
communication,
agent_communication_context,
)
.await
}
async fn send_inter_agent_communication_after_capacity_check(
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
self.submit_inter_agent_communication(agent_id, state, communication, context)
.await
}
async fn submit_inter_agent_communication(
&self,
agent_id: ThreadId,
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
) -> CodexResult<String> {
let last_task_message = last_task_message_from_communication(&communication);
let state = self.upgrade()?;
let op = Op::InterAgentCommunication { communication };
self.ensure_execution_capacity_for_op(agent_id, &op).await?;
let communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let result = self
.handle_thread_request_result(agent_id, &state, state.send_op(agent_id, op).await)
.handle_thread_request_result(
agent_id,
state,
state
.send_op(agent_id, Op::InterAgentCommunication { communication })
.await,
)
.await;
if let (Some(communication), Ok(communication_id)) =
(communication_for_log, result.as_ref())
{
crate::agent_communication::emit_agent_communication_send(
communication_id,
&context,
&communication,
agent_id,
);
}
if result.is_ok() {
match last_task_message {
Some(last_task_message) => self
@@ -480,8 +523,10 @@ impl AgentControl {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id);
let _ = control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await;
return;
}
@@ -706,27 +751,23 @@ fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> b
})
}
pub(crate) fn render_input_preview(initial_operation: &Op) -> String {
match initial_operation {
Op::UserInput { items, .. } => items
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n"),
Op::InterAgentCommunication { communication } => communication.content.clone(),
_ => String::new(),
}
pub(crate) fn render_input_preview(input: &[UserInput]) -> String {
input
.iter()
.map(|item| match item {
UserInput::Text { text, .. } => text.clone(),
UserInput::Image { .. } => "[image]".to_string(),
UserInput::LocalImage { path, .. } => {
format!("[local_image:{}]", path.display())
}
UserInput::Skill { name, path, .. } => {
format!("[skill:${name}]({})", path.display())
}
UserInput::Mention { name, path, .. } => format!("[mention:${name}]({path})"),
_ => "[input]".to_string(),
})
.collect::<Vec<_>>()
.join("\n")
}
fn last_task_message_from_communication(communication: &InterAgentCommunication) -> Option<String> {
codex-rs/core/src/agent/control/execution.rs
+10 -1
View File
@@ -32,7 +32,16 @@ impl AgentControl {
thread_id: ThreadId,
op: &Op,
) -> CodexResult<()> {
if !op_starts_turn(op) {
self.ensure_execution_capacity_for_turn_start(thread_id, op_starts_turn(op))
.await
}
pub(super) async fn ensure_execution_capacity_for_turn_start(
&self,
thread_id: ThreadId,
starts_turn: bool,
) -> CodexResult<()> {
if !starts_turn {
return Ok(());
}
let state = self.upgrade()?;
codex-rs/core/src/agent/control/spawn.rs
+54 -8
View File
@@ -9,6 +9,17 @@ struct SpawnAgentThreadInheritance {
exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
}
/// Initial input delivered after a spawned agent acquires execution capacity.
///
/// V2 communication spawns keep the communication and its context paired so centralized
/// submission and lifecycle logging cannot receive one without the other. Other spawn sources
/// provide user input directly, making an uncontextualized inter-agent communication
/// unrepresentable.
enum SpawnInitialInput {
UserInput(Vec<UserInput>),
InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext),
}
fn default_agent_nickname_list() -> Vec<&'static str> {
AGENT_NAMES
.lines()
@@ -89,12 +100,12 @@ impl AgentControl {
pub(crate) async fn spawn_agent(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
session_source: Option<SessionSource>,
) -> CodexResult<ThreadId> {
let spawned_agent = Box::pin(self.spawn_agent_internal(
config,
initial_operation,
SpawnInitialInput::UserInput(initial_input),
session_source,
SpawnAgentOptions::default(),
))
@@ -106,12 +117,34 @@ impl AgentControl {
pub(crate) async fn spawn_agent_with_metadata(
&self,
config: Config,
initial_operation: Op,
initial_input: Vec<UserInput>,
session_source: Option<SessionSource>,
options: SpawnAgentOptions, // TODO(jif) drop with new fork.
) -> CodexResult<LiveAgent> {
Box::pin(self.spawn_agent_internal(config, initial_operation, session_source, options))
.await
Box::pin(self.spawn_agent_internal(
config,
SpawnInitialInput::UserInput(initial_input),
session_source,
options,
))
.await
}
pub(crate) async fn spawn_agent_with_communication(
&self,
config: Config,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
session_source: Option<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
Box::pin(self.spawn_agent_internal(
config,
SpawnInitialInput::InterAgentCommunication(communication, context),
session_source,
options,
))
.await
}
pub(crate) async fn ensure_v2_agent_loaded(
@@ -197,7 +230,7 @@ impl AgentControl {
async fn spawn_agent_internal(
&self,
config: Config,
initial_operation: Op,
initial_input: SpawnInitialInput,
session_source: Option<SessionSource>,
options: SpawnAgentOptions,
) -> CodexResult<LiveAgent> {
@@ -356,8 +389,21 @@ impl AgentControl {
)
.await;
self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation)
.await?;
match initial_input {
SpawnInitialInput::UserInput(input) => {
self.send_input_after_capacity_check(new_thread.thread_id, &state, input)
.await?;
}
SpawnInitialInput::InterAgentCommunication(communication, context) => {
self.send_inter_agent_communication_after_capacity_check(
new_thread.thread_id,
&state,
communication,
context,
)
.await?;
}
}
if multi_agent_version != MultiAgentVersion::V2 {
let child_reference = agent_metadata
.agent_path
codex-rs/core/src/agent/control_tests.rs
+21 -11
View File
@@ -3,6 +3,8 @@ use crate::CodexThread;
use crate::StateDbHandle;
use crate::ThreadManager;
use crate::agent::agent_status_from_event;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::config::AgentRoleConfig;
use crate::config::Config;
use crate::config::ConfigBuilder;
@@ -65,12 +67,11 @@ async fn test_config() -> (TempDir, Config) {
test_config_with_cli_overrides(Vec::new()).await
}
fn text_input(text: &str) -> Op {
fn text_input(text: &str) -> Vec<UserInput> {
vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}]
.into()
}
fn assistant_message(text: &str, phase: Option<MessagePhase>) -> ResponseItem {
@@ -311,8 +312,7 @@ async fn send_input_errors_when_manager_dropped() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail without a manager");
@@ -423,8 +423,7 @@ async fn send_input_errors_when_thread_missing() {
vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect_err("send_input should fail for missing thread");
@@ -490,8 +489,7 @@ async fn send_input_submits_user_message() {
vec![UserInput::Text {
text: "hello from tests".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
)
.await
.expect("send_input should succeed");
@@ -531,7 +529,11 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
let submission_id = harness
.control
.send_inter_agent_communication(thread_id, communication.clone())
.send_inter_agent_communication(
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
assert!(!submission_id.is_empty());
@@ -656,7 +658,11 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication.clone())
.send_inter_agent_communication(
spawned_agent.thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed after reload");
let expected = (
@@ -803,7 +809,11 @@ async fn encrypted_inter_agent_communication_clears_existing_last_task_message()
);
harness
.control
.send_inter_agent_communication(spawned_agent.thread_id, communication)
.send_inter_agent_communication(
spawned_agent.thread_id,
communication,
AgentCommunicationContext::new(AgentCommunicationKind::Followup, ThreadId::new()),
)
.await
.expect("send_inter_agent_communication should succeed");
codex-rs/core/src/agent_communication.rs
+79
View File
@@ -0,0 +1,79 @@
use codex_protocol::ThreadId;
use codex_protocol::protocol::InterAgentCommunication;
const AGENT_COMMUNICATION_TARGET: &str = "codex_otel.agent_communication";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum AgentCommunicationKind {
Spawn,
Message,
Followup,
Result,
}
impl AgentCommunicationKind {
fn as_str(self) -> &'static str {
match self {
Self::Spawn => "spawn",
Self::Message => "message",
Self::Followup => "followup",
Self::Result => "result",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct AgentCommunicationContext {
kind: AgentCommunicationKind,
sender_thread_id: ThreadId,
}
impl AgentCommunicationContext {
pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self {
Self {
kind,
sender_thread_id,
}
}
}
pub(crate) fn logging_enabled() -> bool {
tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO)
}
pub(crate) fn emit_agent_communication_send(
communication_id: &str,
context: &AgentCommunicationContext,
communication: &InterAgentCommunication,
receiver_thread_id: ThreadId,
) {
tracing::info!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
kind = context.kind.as_str(),
state = "send",
sender_thread_id = %context.sender_thread_id,
receiver_thread_id = %receiver_thread_id,
content = if communication.content.is_empty() {
communication.encrypted_content.as_deref().unwrap_or_default()
} else {
communication.content.as_str()
},
},
"agent communication"
);
}
pub(crate) fn emit_agent_communication_receive(communication_id: &str) {
tracing::info!(
target: AGENT_COMMUNICATION_TARGET,
{
event.name = "codex.agent_communication",
communication_id,
state = "receive",
},
"agent communication"
);
}
codex-rs/core/src/client.rs
+33 -20
View File
@@ -1123,42 +1123,48 @@ impl ModelClientSession {
}
}
/// Checks whether the current request is an incremental extension of the previous request.
/// We only reuse an incremental input delta when non-input request fields are unchanged and
/// `input` is a strict extension of the previous known input. Server-returned output items
/// are treated as part of the baseline so we do not resend them.
fn get_incremental_items(
&self,
request: &ResponsesApiRequest,
last_response: Option<&LastResponse>,
allow_empty_delta: bool,
) -> Option<Vec<ResponseItem>> {
// Checks whether the current request is an incremental extension of the previous request.
// We only reuse an incremental input delta when non-input request fields are unchanged and
// `input` is a strict
// extension of the previous known input. Server-returned output items are treated as part
// of the baseline so we do not resend them.
let previous_request = self.websocket_session.last_request.as_ref()?;
if !responses_request_properties_match(previous_request, request) {
trace!("incremental request failed, websocket reuse properties didn't match");
return None;
}
let Some(after_previous_input) = request
.input
.strip_prefix(previous_request.input.as_slice())
else {
trace!("incremental request failed, items didn't match");
return None;
};
let mut response_items =
last_response.map_or_else(Vec::new, |response| response.items_added.clone());
if !self.client.state.provider.info().is_openai() {
response_items
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
// To compare the inputs, we concatenate the previous request items with the response items,
// then compare that against the equivalent slice of request items, ignoring metadata. If
// they match, we can consider the remaining items the incremental request.
let mut previous_items = previous_request.input.clone();
if let Some(response) = last_response {
previous_items.extend_from_slice(&response.items_added);
}
let Some(incremental_items) = after_previous_input.strip_prefix(response_items.as_slice())
previous_items
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
let Some((request_items_to_compare, incremental_items)) =
request.input.split_at_checked(previous_items.len())
else {
trace!("incremental request failed, items didn't match");
trace!("incremental request failed, incompatible request length");
return None;
};
let mut request_prefix = request_items_to_compare.to_vec();
request_prefix
.iter_mut()
.for_each(ResponseItem::clear_internal_chat_message_metadata_passthrough);
if previous_items != request_prefix {
trace!("incremental request failed, items didn't match");
return None;
}
if !allow_empty_delta && incremental_items.is_empty() {
return None;
}
@@ -1912,6 +1918,7 @@ where
let mut logged_error = false;
let mut tx_last_response = Some(tx_last_response);
let mut items_added: Vec<ResponseItem> = Vec::new();
let (request_start, mut ttft_ms) = (Instant::now(), None);
let mut api_stream = api_stream;
let upstream_request_id = upstream_request_id.as_deref();
if let Some(upstream_request_id) = upstream_request_id {
@@ -1961,6 +1968,7 @@ where
Some(usage.cached_input_tokens),
Some(usage.reasoning_output_tokens),
usage.total_tokens,
ttft_ms,
);
}
inference_trace_attempt.record_completed(
@@ -1988,6 +1996,11 @@ where
}
}
Ok(event) => {
if matches!(&event, ResponseEvent::OutputItemAdded(_)) && ttft_ms.is_none() {
ttft_ms = Some(
i64::try_from(request_start.elapsed().as_millis()).unwrap_or(i64::MAX),
);
}
if tx_event.send(Ok(event)).await.is_err() {
inference_trace_attempt.record_cancelled(
STREAM_DROPPED_REASON,
codex-rs/core/src/config/config_tests.rs
+21
View File
@@ -10307,6 +10307,7 @@ default_wait_timeout_ms = 30000
usage_hint_text = "Custom delegation guidance."
root_agent_usage_hint_text = "Root guidance."
subagent_usage_hint_text = "Subagent guidance."
multi_agent_mode_hint_text = "Custom mode guidance."
tool_namespace = "agents"
hide_spawn_agent_metadata = true
non_code_mode_only = true
@@ -10343,6 +10344,10 @@ non_code_mode_only = true
config.multi_agent_v2.subagent_usage_hint_text.as_deref(),
Some("Subagent guidance.")
);
assert_eq!(
config.multi_agent_v2.multi_agent_mode_hint_text.as_deref(),
Some("Custom mode guidance.")
);
assert_eq!(
config.multi_agent_v2.tool_namespace.as_deref(),
Some("agents")
@@ -10405,6 +10410,22 @@ max_concurrent_threads_per_session = 17
);
}
#[test]
fn multi_agent_v2_preserves_empty_mode_hint_override() {
let config_toml = toml::from_str(
r#"[features.multi_agent_v2]
multi_agent_mode_hint_text = ""
"#,
)
.expect("multi-agent v2 config should parse");
let expected = MultiAgentV2Config {
multi_agent_mode_hint_text: Some(String::new()),
..Default::default()
};
assert_eq!(resolve_multi_agent_v2_config(&config_toml), expected);
}
#[tokio::test]
async fn multi_agent_v2_empty_usage_hint_overrides_clear_default_hints() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
codex-rs/core/src/config/mod.rs
+7
View File
@@ -1144,6 +1144,7 @@ pub struct MultiAgentV2Config {
pub usage_hint_text: Option<String>,
pub root_agent_usage_hint_text: Option<String>,
pub subagent_usage_hint_text: Option<String>,
pub multi_agent_mode_hint_text: Option<String>,
pub tool_namespace: Option<String>,
pub hide_spawn_agent_metadata: bool,
pub non_code_mode_only: bool,
@@ -1165,6 +1166,7 @@ impl MultiAgentV2Config {
DEFAULT_MULTI_AGENT_V2_SUBAGENT_USAGE_HINT_TEXT,
max_concurrent_threads_per_session,
)),
multi_agent_mode_hint_text: None,
tool_namespace: Some(DEFAULT_MULTI_AGENT_V2_TOOL_NAMESPACE.to_string()),
hide_spawn_agent_metadata: true,
non_code_mode_only: true,
@@ -2501,6 +2503,10 @@ fn resolve_multi_agent_v2_config(config_toml: &ConfigToml) -> MultiAgentV2Config
base.map(|config| &config.subagent_usage_hint_text),
default.subagent_usage_hint_text,
);
let multi_agent_mode_hint_text = base
.and_then(|config| config.multi_agent_mode_hint_text.as_ref())
.cloned()
.or(default.multi_agent_mode_hint_text);
let tool_namespace = base
.and_then(|config| config.tool_namespace.as_ref())
.cloned()
@@ -2520,6 +2526,7 @@ fn resolve_multi_agent_v2_config(config_toml: &ConfigToml) -> MultiAgentV2Config
usage_hint_text,
root_agent_usage_hint_text,
subagent_usage_hint_text,
multi_agent_mode_hint_text,
tool_namespace,
hide_spawn_agent_metadata,
non_code_mode_only,
codex-rs/core/src/context/multi_agent_mode_instructions.rs
+2 -3
View File
@@ -4,7 +4,6 @@ use codex_protocol::protocol::MULTI_AGENT_MODE_CLOSE_TAG;
use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG;
const EXPLICIT_REQUEST_ONLY_MULTI_AGENT_MODE_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const NO_MULTI_AGENT_MODE_TEXT: &str = "Multi-agent delegation mode instructions are inactive. Any earlier multi-agent mode developer message no longer applies.";
const PROACTIVE_MULTI_AGENT_MODE_TEXT: &str = "Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.";
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -32,8 +31,8 @@ impl ContextualUserFragment for MultiAgentModeInstructions {
}
fn body(&self) -> String {
match self.multi_agent_mode {
MultiAgentMode::None => NO_MULTI_AGENT_MODE_TEXT.to_string(),
match &self.multi_agent_mode {
MultiAgentMode::Custom(hint_text) => hint_text.clone(),
MultiAgentMode::ExplicitRequestOnly => {
EXPLICIT_REQUEST_ONLY_MULTI_AGENT_MODE_TEXT.to_string()
}
codex-rs/core/src/context_manager/updates.rs
-3
View File
@@ -85,9 +85,6 @@ fn build_multi_agent_mode_update_item(
}
match effective_multi_agent_mode {
Some(MultiAgentMode::None) => {
Some(MultiAgentModeInstructions::new(MultiAgentMode::None).render())
}
Some(multi_agent_mode) => Some(MultiAgentModeInstructions::new(multi_agent_mode).render()),
None if previous.multi_agent_mode == Some(MultiAgentMode::Proactive) => {
Some(MultiAgentModeInstructions::new(MultiAgentMode::ExplicitRequestOnly).render())
codex-rs/core/src/lib.rs
+1
View File
@@ -31,6 +31,7 @@ pub use codex_thread::TryStartTurnIfIdleError;
pub use codex_thread::TryStartTurnIfIdleRejectionReason;
pub use session::turn_context::TurnContext;
mod agent;
mod agent_communication;
mod attestation;
mod codex_delegate;
mod command_canonicalization;
codex-rs/core/src/session/handlers.rs
+1
View File
@@ -285,6 +285,7 @@ pub async fn inter_agent_communication(
sess.input_queue
.enqueue_mailbox_communication(communication)
.await;
crate::agent_communication::emit_agent_communication_receive(&sub_id);
if trigger_turn {
sess.maybe_start_turn_for_pending_work_with_sub_id(sub_id)
.await;
codex-rs/core/src/session/mod.rs
+9 -12
View File
@@ -14,6 +14,8 @@ use crate::agent::AgentControl;
use crate::agent::AgentStatus;
use crate::agent::agent_status_from_event;
use crate::agent::status::is_final;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::attestation::AttestationProvider;
use crate::build_available_skills;
use crate::compact;
@@ -92,7 +94,6 @@ use codex_protocol::approvals::NetworkPolicyRuleAction;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::AutoCompactTokenLimitScope;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::MultiAgentMode;
use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::WebSearchMode;
@@ -1889,10 +1890,12 @@ impl Session {
message,
/*trigger_turn*/ false,
);
let context =
AgentCommunicationContext::new(AgentCommunicationKind::Result, self.thread_id);
if let Err(err) = self
.services
.agent_control
.send_inter_agent_communication(parent_thread_id, communication)
.send_inter_agent_communication(parent_thread_id, communication, context)
.await
{
debug!("failed to notify parent thread {parent_thread_id}: {err}");
@@ -3505,16 +3508,10 @@ impl Session {
{
items.push(usage_hint_message);
}
match multi_agents::effective_multi_agent_mode(turn_context) {
Some(
multi_agent_mode
@ (MultiAgentMode::ExplicitRequestOnly | MultiAgentMode::Proactive),
) => {
items.push(ContextualUserFragment::into(
MultiAgentModeInstructions::new(multi_agent_mode),
));
}
Some(MultiAgentMode::None) | None => {}
if let Some(multi_agent_mode) = multi_agents::effective_multi_agent_mode(turn_context) {
items.push(ContextualUserFragment::into(
MultiAgentModeInstructions::new(multi_agent_mode),
));
}
if let Some(contextual_user_message) =
crate::context_manager::updates::build_contextual_user_message(contextual_user_sections)
codex-rs/core/src/session/multi_agents.rs
+12 -3
View File
@@ -41,9 +41,18 @@ pub(crate) fn effective_multi_agent_mode(turn_context: &TurnContext) -> Option<M
return None;
}
let multi_agent_mode = match turn_context.effective_reasoning_effort() {
Some(ReasoningEffort::Ultra) => MultiAgentMode::Proactive,
_ => MultiAgentMode::ExplicitRequestOnly,
// A configured hint, including an empty string, defines a custom policy instead of an
// effort-derived built-in policy.
let multi_agent_mode = match &turn_context
.config
.multi_agent_v2
.multi_agent_mode_hint_text
{
Some(hint_text) => MultiAgentMode::Custom(hint_text.clone()),
None => match turn_context.effective_reasoning_effort() {
Some(ReasoningEffort::Ultra) => MultiAgentMode::Proactive,
_ => MultiAgentMode::ExplicitRequestOnly,
},
};
match &turn_context.session_source {
codex-rs/core/src/tools/handlers/agent_jobs.rs
+1 -1
View File
@@ -203,7 +203,7 @@ async fn run_agent_job_loop(
.agent_control
.spawn_agent_with_metadata(
options.spawn_config.clone(),
items.into(),
items,
Some(SessionSource::SubAgent(SubAgentSource::Other(format!(
"agent_job:{job_id}"
)))),
codex-rs/core/src/tools/handlers/multi_agents_common.rs
+3 -5
View File
@@ -18,7 +18,6 @@ use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::openai_models::ReasoningEffortPreset;
use codex_protocol::protocol::CollabAgentRef;
use codex_protocol::protocol::CollabAgentStatusEntry;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
@@ -163,7 +162,7 @@ pub(crate) fn thread_spawn_source(
pub(crate) fn parse_collab_input(
message: Option<String>,
items: Option<Vec<UserInput>>,
) -> Result<Op, FunctionCallError> {
) -> Result<Vec<UserInput>, FunctionCallError> {
match (message, items) {
(Some(_), Some(_)) => Err(FunctionCallError::RespondToModel(
"Provide either message or items, but not both".to_string(),
@@ -180,8 +179,7 @@ pub(crate) fn parse_collab_input(
Ok(vec![UserInput::Text {
text: message,
text_elements: Vec::new(),
}]
.into())
}])
}
(None, Some(items)) => {
if items.is_empty() {
@@ -189,7 +187,7 @@ pub(crate) fn parse_collab_input(
"Items can't be empty".to_string(),
));
}
Ok(items.into())
Ok(items)
}
}
}
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+6 -12
View File
@@ -1377,8 +1377,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1454,8 +1453,7 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1626,8 +1624,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
vec![UserInput::Text {
text: "research".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -1647,8 +1644,7 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
vec![UserInput::Text {
text: "build".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 2,
@@ -4097,8 +4093,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_id() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
@@ -4165,8 +4160,7 @@ async fn multi_agent_v2_interrupt_agent_rejects_self_target_by_task_name() {
vec![UserInput::Text {
text: "inspect this repo".to_string(),
text_elements: Vec::new(),
}]
.into(),
}],
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: root.thread_id,
depth: 1,
codex-rs/core/src/tools/handlers/multi_agents_v2.rs
-1
View File
@@ -19,7 +19,6 @@ use codex_protocol::protocol::CollabWaitingEndEvent;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::SubAgentActivityEvent;
use codex_protocol::protocol::SubAgentActivityKind;
use codex_protocol::user_input::UserInput;
use codex_tools::ToolName;
use serde::Deserialize;
use serde::Serialize;
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+9 -2
View File
@@ -4,6 +4,8 @@
//! resulting `InterAgentCommunication` should wake the target immediately.
use super::*;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::context::FunctionToolOutput;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::protocol::InterAgentCommunication;
@@ -46,7 +48,7 @@ pub(crate) struct FollowupTaskArgs {
pub(crate) message: String,
}
fn message_content(message: String) -> Result<String, FunctionCallError> {
pub(super) fn message_content(message: String) -> Result<String, FunctionCallError> {
if message.trim().is_empty() {
return Err(FunctionCallError::RespondToModel(
"Empty message can't be sent to an agent".to_string(),
@@ -101,10 +103,15 @@ pub(crate) async fn handle_message_string_tool(
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, receiver_agent_path.clone(), message);
let kind = match mode {
MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message,
MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup,
};
let context = AgentCommunicationContext::new(kind, session.thread_id);
let result = session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication))
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication), context)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
result?;
codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs
+25 -29
View File
@@ -4,11 +4,13 @@ use crate::agent::control::SpawnAgentOptions;
use crate::agent::next_thread_spawn_depth;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::apply_role_to_config;
use crate::agent_communication::AgentCommunicationContext;
use crate::agent_communication::AgentCommunicationKind;
use crate::tools::handlers::multi_agents_spec::SpawnAgentToolOptions;
use crate::tools::handlers::multi_agents_spec::create_spawn_agent_tool_v2;
use crate::tools::handlers::multi_agents_v2::message_tool::message_content;
use crate::turn_timing::now_unix_timestamp_ms;
use codex_protocol::AgentPath;
use codex_protocol::protocol::Op;
use codex_tools::ToolSpec;
#[derive(Default)]
@@ -55,8 +57,7 @@ async fn handle_spawn_agent(
.map(str::trim)
.filter(|role| !role.is_empty());
let message = args.message.clone();
let initial_operation = parse_collab_input(Some(args.message), /*items*/ None)?;
let message = message_content(args.message)?;
let session_source = turn.session_source.clone();
let child_depth = next_thread_spawn_depth(&session_source);
let mut config =
@@ -104,33 +105,28 @@ async fn handle_spawn_agent(
"spawned agent is missing a canonical task name".to_string(),
)
})?;
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication = communication_from_tool_message(author, new_agent_path.clone(), message);
let context = AgentCommunicationContext::new(AgentCommunicationKind::Spawn, session.thread_id);
let spawned_agent = Box::pin(
session.services.agent_control.spawn_agent_with_metadata(
config,
match initial_operation {
Op::UserInput { items, .. }
if items
.iter()
.all(|item| matches!(item, UserInput::Text { .. })) =>
{
let author = turn
.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root);
let communication =
communication_from_tool_message(author, new_agent_path.clone(), message);
Op::InterAgentCommunication { communication }
}
initial_operation => initial_operation,
},
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
),
session
.services
.agent_control
.spawn_agent_with_communication(
config,
communication,
context,
Some(spawn_source),
SpawnAgentOptions {
fork_parent_spawn_call_id: fork_mode.as_ref().map(|_| call_id.clone()),
fork_mode,
parent_thread_id: Some(session.thread_id),
environments: Some(turn.environments.to_selections()),
},
),
)
.await
.map_err(collab_spawn_error)?;
codex-rs/core/src/tools/parallel.rs
+260 -8
View File
@@ -1,4 +1,5 @@
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Instant;
@@ -9,6 +10,7 @@ use tokio_util::either::Either;
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;
use tracing::Instrument;
use tracing::info;
use tracing::instrument;
use tracing::trace_span;
@@ -27,6 +29,15 @@ use crate::tools::router::ToolRouter;
use codex_protocol::error::CodexErr;
use codex_protocol::models::ResponseInputItem;
struct ToolCallTimingGuard {
started_at: Instant,
execution_started_at: Arc<OnceLock<Instant>>,
conversation_id: String,
turn_id: String,
call_id: String,
tool_name: codex_tools::ToolName,
}
#[derive(Clone)]
pub(crate) struct ToolCallRuntime {
router: Arc<ToolRouter>,
@@ -96,6 +107,11 @@ impl ToolCallRuntime {
let invocation_cancellation_token = cancellation_token.clone();
let wait_for_runtime_cancellation = self.router.tool_waits_for_runtime_cancellation(&call);
let started = Instant::now();
let tool_call_timing_guard =
ToolCallTimingGuard::capture(started, &session.thread_id, &turn.sub_id, &call, &source);
let execution_started_at = tool_call_timing_guard
.as_ref()
.map(|timing| Arc::clone(&timing.execution_started_at));
let abort_session = Arc::clone(&session);
let abort_source = source.clone();
let abort_turn = Arc::clone(&turn);
@@ -112,13 +128,18 @@ impl ToolCallRuntime {
);
let abort_dispatch_span = dispatch_span.clone();
let mut handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
let mut dispatch_handle: AbortOnDropHandle<Result<AnyToolResult, FunctionCallError>> =
AbortOnDropHandle::new(tokio::spawn(async move {
let _guard = if supports_parallel {
Either::Left(lock.read().await)
} else {
Either::Right(lock.write().await)
};
// Admission through the parallel-execution gate marks the end
// of dispatch waiting and the start of handler execution.
if let Some(execution_started_at) = execution_started_at {
let _ = execution_started_at.set(Instant::now());
}
router
.dispatch_tool_call_with_terminal_outcome(
@@ -135,28 +156,29 @@ impl ToolCallRuntime {
}));
async move {
let _tool_call_timing_guard = tool_call_timing_guard;
tokio::select! {
res = &mut handle => res.map_err(Self::tool_task_join_error)?,
res = &mut dispatch_handle => res.map_err(Self::tool_task_join_error)?,
_ = cancellation_token.cancelled() => {
if terminal_outcome_reached.load(Ordering::Acquire) || handle.is_finished() {
handle.await.map_err(Self::tool_task_join_error)?
if terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() {
dispatch_handle.await.map_err(Self::tool_task_join_error)?
} else {
let secs = started.elapsed().as_secs_f32().max(0.1);
abort_dispatch_span.record("aborted", true);
if wait_for_runtime_cancellation {
if terminal_outcome_reached.swap(true, Ordering::AcqRel) {
return handle.await.map_err(Self::tool_task_join_error)?;
return dispatch_handle.await.map_err(Self::tool_task_join_error)?;
}
// The abort owns the terminal outcome; await only so
// the runtime can finish process teardown.
match handle.await {
match dispatch_handle.await {
Ok(_) => {}
Err(err) if err.is_cancelled() => {}
Err(err) => return Err(Self::tool_task_join_error(err)),
}
} else {
handle.abort();
match handle.await {
dispatch_handle.abort();
match dispatch_handle.await {
Ok(result) => return result,
Err(err) if err.is_cancelled() => {}
Err(err) => return Err(Self::tool_task_join_error(err)),
@@ -237,6 +259,88 @@ impl ToolCallRuntime {
}
}
impl ToolCallTimingGuard {
fn capture(
started_at: Instant,
conversation_id: &impl std::fmt::Display,
turn_id: &str,
call: &ToolCall,
source: &ToolCallSource,
) -> Option<Self> {
// Code-mode calls are nested within a direct code-mode tool call whose
// timing already includes them. Suppress nested guards so consumers do
// not mistake overlapping events for independent tool-call latency.
if !matches!(source, ToolCallSource::Direct) || !tracing::enabled!(tracing::Level::INFO) {
return None;
}
Some(Self {
started_at,
execution_started_at: Arc::new(OnceLock::new()),
conversation_id: conversation_id.to_string(),
turn_id: turn_id.to_string(),
call_id: call.call_id.clone(),
tool_name: call.tool_name.clone(),
})
}
}
impl Drop for ToolCallTimingGuard {
fn drop(&mut self) {
let completed_at = Instant::now();
// Snapshot once so a concurrently-starting dispatch cannot make one
// event internally inconsistent.
let execution_started_at = self
.execution_started_at
.get()
.copied()
.filter(|execution_started_at| *execution_started_at <= completed_at);
let duration_ms = |duration: std::time::Duration| u64::try_from(duration.as_millis()).ok();
let total_duration_ms = duration_ms(completed_at.duration_since(self.started_at));
let dispatch_duration_ms = execution_started_at.map_or_else(
|| total_duration_ms,
|execution_started_at| {
duration_ms(execution_started_at.duration_since(self.started_at))
},
);
let handler_duration_ms = execution_started_at.map_or(Some(0), |execution_started_at| {
duration_ms(completed_at.duration_since(execution_started_at))
});
macro_rules! log_tool_call {
($dispatch_duration_ms:expr, $handler_duration_ms:expr, $total_duration_ms:expr) => {
info!(
event.name = "codex.tool_call",
trace_id = %codex_otel::current_span_trace_id().unwrap_or_default(),
conversation.id = %self.conversation_id,
turn_id = %self.turn_id,
tool_name = %self.tool_name,
call_id = %self.call_id,
tool_source = "direct",
execution_started = execution_started_at.is_some(),
dispatch_duration_ms = $dispatch_duration_ms,
handler_duration_ms = $handler_duration_ms,
total_duration_ms = $total_duration_ms,
"tool call completed"
);
};
}
match (dispatch_duration_ms, handler_duration_ms, total_duration_ms) {
(Some(dispatch_duration_ms), Some(handler_duration_ms), Some(total_duration_ms)) => {
log_tool_call!(dispatch_duration_ms, handler_duration_ms, total_duration_ms);
}
_ => {
log_tool_call!(
tracing::field::Empty,
tracing::field::Empty,
tracing::field::Empty
);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -255,6 +359,154 @@ mod tests {
use pretty_assertions::assert_eq;
use tokio::sync::Notify;
use tokio::sync::oneshot;
use tracing_test::internal::MockWriter;
#[test]
fn tool_call_timing_guard_ignores_code_mode_source() {
let subscriber = tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.finish();
tracing::subscriber::with_default(subscriber, || {
let call = ToolCall {
tool_name: codex_tools::ToolName::plain("test_tool"),
call_id: "call-1".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
};
let direct_guard = ToolCallTimingGuard::capture(
Instant::now(),
&"conversation-id",
"turn-id",
&call,
&ToolCallSource::Direct,
);
assert!(
direct_guard.is_some(),
"direct tool calls should create a timing guard"
);
drop(direct_guard);
let code_mode_guard = ToolCallTimingGuard::capture(
Instant::now(),
&"conversation-id",
"turn-id",
&call,
&ToolCallSource::CodeMode {
cell_id: "cell-1".to_string(),
runtime_tool_call_id: "runtime-call-1".to_string(),
},
);
assert!(
code_mode_guard.is_none(),
"nested code-mode calls should not create overlapping timing events"
);
});
}
#[tokio::test]
async fn cancellation_before_dispatch_admission_logs_dispatch_only_timing() -> anyhow::Result<()>
{
let (session, turn_context) = crate::session::tests::make_session_and_context().await;
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let tool_name = codex_tools::ToolName::plain("test_tool");
let handler = Arc::new(ImmediateHandler {
tool_name: tool_name.clone(),
}) as Arc<dyn CoreToolRuntime>;
let step_context = StepContext::for_test(Arc::clone(&turn_context));
let router = Arc::new(ToolRouter::from_parts(
ToolRegistry::from_tools([handler]),
Vec::new(),
));
let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new()));
let runtime = ToolCallRuntime::new(router, session, step_context, tracker);
let execution_gate = Arc::clone(&runtime.parallel_execution);
let execution_gate_guard = execution_gate
.try_write_owned()
.expect("execution gate should be available before dispatch starts");
let (release_execution_gate_tx, release_execution_gate_rx) = std::sync::mpsc::channel();
let execution_gate_task = tokio::task::spawn_blocking(move || {
let _execution_gate_guard = execution_gate_guard;
release_execution_gate_rx
.recv()
.expect("test should release the execution gate");
});
let buffer: &'static std::sync::Mutex<Vec<u8>> =
Box::leak(Box::new(std::sync::Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(tracing::Level::INFO)
.with_writer(MockWriter::new(buffer))
.finish();
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
let cancellation_token = CancellationToken::new();
let call = ToolCall {
tool_name,
call_id: "call-1".to_string(),
payload: ToolPayload::Function {
arguments: "{}".to_string(),
},
};
let response_task =
tokio::spawn(runtime.handle_tool_call(call, cancellation_token.clone()));
cancellation_token.cancel();
tokio::time::timeout(Duration::from_secs(1), response_task)
.await
.expect("timed out waiting for cancelled tool response")
.expect("cancelled tool response task should join")
.expect("cancelled tool call should produce a response");
let logs = String::from_utf8(
buffer
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone(),
)?;
let timing_events = logs
.lines()
.filter(|line| line.contains("event.name=\"codex.tool_call\""))
.collect::<Vec<_>>();
assert_eq!(
timing_events.len(),
1,
"cancelled tool call should emit exactly one timing event; logs:\n{logs}"
);
let timing_event = timing_events[0];
assert!(
timing_event.contains("execution_started=false"),
"tool cancelled before admission should not report execution started: {timing_event}"
);
assert!(
timing_event.contains("handler_duration_ms=0"),
"tool cancelled before admission should report zero handler duration: {timing_event}"
);
let duration_field = |name: &str| {
timing_event.split_whitespace().find_map(|field| {
field
.strip_prefix(&format!("{name}="))
.and_then(|value| value.parse::<u64>().ok())
})
};
let dispatch_duration_ms = duration_field("dispatch_duration_ms")
.expect("timing event should include dispatch_duration_ms");
let total_duration_ms = duration_field("total_duration_ms")
.expect("timing event should include total_duration_ms");
assert_eq!(
dispatch_duration_ms, total_duration_ms,
"tool cancelled before admission should attribute all elapsed time to dispatch: {timing_event}"
);
release_execution_gate_tx
.send(())
.expect("execution gate task should remain available");
execution_gate_task
.await
.expect("execution gate task should join");
Ok(())
}
struct ImmediateHandler {
tool_name: codex_tools::ToolName,
codex-rs/core/tests/suite/client_websockets.rs
+63 -12
View File
@@ -966,35 +966,77 @@ async fn responses_websocket_v2_requests_use_v2_when_provider_supports_websocket
async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
skip_if_no_network!();
let mut assistant_output_with_turn_id = ev_assistant_message("msg-1", "assistant output");
assistant_output_with_turn_id["item"]["internal_chat_message_metadata_passthrough"] =
json!({ "turn_id": "turn-1" });
let assistant_output_without_turn_id = ev_assistant_message("msg-2", "second assistant output");
let server = start_websocket_server(vec![vec![
vec![
ev_response_created("resp-1"),
ev_assistant_message("msg-1", "assistant output"),
assistant_output_with_turn_id,
ev_completed("resp-1"),
],
vec![ev_response_created("resp-2"), ev_completed("resp-2")],
vec![
ev_response_created("resp-2"),
assistant_output_without_turn_id,
ev_completed("resp-2"),
],
vec![ev_response_created("resp-3"), ev_completed("resp-3")],
]])
.await;
let harness = websocket_harness_with_options(&server, /*runtime_metrics_enabled*/ false).await;
let prompt_one = prompt_with_input(vec![message_item("hello")]);
let prompt_two = prompt_with_input(vec![
message_item("hello"),
assistant_message_item("msg-1", "assistant output"),
message_item("second"),
]);
// use OpenAI provider to check metadata logic
let mut provider = websocket_provider(&server);
provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name;
let harness =
websocket_harness_with_provider_options(provider, /*runtime_metrics_enabled*/ false).await;
// Turn one: initiate
let prompt_one = prompt_with_input(vec![message_item("hello")]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_one).await;
}
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_two).await;
// Turn two: the response and reconstructed history have matching metadata.
let mut first_assistant_output = assistant_message_item("msg-1", "assistant output");
first_assistant_output.set_turn_id_if_missing("turn-1");
let prompt_two = prompt_with_input(vec![
message_item("hello"),
first_assistant_output.clone(),
message_item("second"),
]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_two).await;
}
// Turn three: the reconstructed history has metadata that the response omitted.
let mut second_assistant_output = assistant_message_item("msg-2", "second assistant output");
second_assistant_output.set_turn_id_if_missing("turn-2");
let prompt_three = prompt_with_input(vec![
message_item("hello"),
first_assistant_output,
message_item("second"),
second_assistant_output,
message_item("third"),
]);
{
let mut client_session = harness.client.new_session();
stream_until_complete(&mut client_session, &harness, &prompt_three).await;
}
assert_eq!(server.handshakes().len(), 1);
let connection = server.single_connection();
assert_eq!(connection.len(), 2);
assert_eq!(connection.len(), 3);
// validate second turn
let second = connection.get(1).expect("missing request").body_json();
assert_eq!(second["type"].as_str(), Some("response.create"));
assert_eq!(second["previous_response_id"].as_str(), Some("resp-1"));
@@ -1003,6 +1045,15 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
serde_json::to_value(&prompt_two.input[2..]).unwrap()
);
// validate third turn
let third = connection.get(2).expect("missing request").body_json();
assert_eq!(third["type"].as_str(), Some("response.create"));
assert_eq!(third["previous_response_id"].as_str(), Some("resp-2"));
assert_eq!(
third["input"],
serde_json::to_value(&prompt_three.input[4..]).unwrap()
);
server.shutdown().await;
}
codex-rs/core/tests/suite/multi_agent_mode.rs
+107
View File
@@ -20,9 +20,11 @@ use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
const NO_SPAWN_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const PROACTIVE_TEXT: &str = "Proactive multi-agent delegation is active.";
const CUSTOM_MODE_HINT_TEXT: &str = "Use the configured delegation policy.";
fn add_ultra_reasoning(model_info: &mut ModelInfo) {
model_info.supports_reasoning_summaries = true;
@@ -41,6 +43,12 @@ fn configure_multi_agent_v2(config: &mut Config) {
.expect("test config should allow feature update");
}
// Configuring a custom mode hint also enables multi-agent V2 for the test.
fn configure_custom_mode_hint(config: &mut Config) {
configure_multi_agent_v2(config);
config.multi_agent_v2.multi_agent_mode_hint_text = Some(CUSTOM_MODE_HINT_TEXT.to_string());
}
fn configure_ultra(config: &mut Config) {
configure_multi_agent_v2(config);
config.model_reasoning_effort = Some(ReasoningEffort::Ultra);
@@ -120,6 +128,105 @@ async fn ultra_reasoning_uses_max_and_proactive_mode() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn configured_mode_hint_uses_custom_mode_across_reasoning_efforts() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let responses = mount_sse_sequence(
&server,
(1..=2)
.map(|index| {
sse(vec![
ev_response_created(&format!("resp-{index}")),
ev_completed(&format!("resp-{index}")),
])
})
.collect(),
)
.await;
let test = test_codex()
.with_model_info_override("gpt-5.4", add_ultra_reasoning)
.with_config(configure_custom_mode_hint)
.build(&server)
.await?;
let rollout_path = test
.session_configured
.rollout_path
.clone()
.expect("rollout path");
submit_turn(&test.codex, "explicit", Some(ReasoningEffort::High)).await?;
submit_turn(&test.codex, "proactive", Some(ReasoningEffort::Ultra)).await?;
let requests = responses.requests();
let first_input = requests[0].input();
let first_texts = developer_texts(&first_input);
let second_input = requests[1].input();
let second_texts = developer_texts(&second_input);
let instruction_counts = |texts: &[&str]| {
(
count_containing(texts, CUSTOM_MODE_HINT_TEXT),
count_containing(texts, NO_SPAWN_TEXT),
count_containing(texts, PROACTIVE_TEXT),
)
};
assert_eq!(instruction_counts(&first_texts), (1, 0, 0));
assert_eq!(instruction_counts(&second_texts), (1, 0, 0));
let rollout_values = std::fs::read_to_string(rollout_path)?
.lines()
.map(serde_json::from_str::<Value>)
.collect::<serde_json::Result<Vec<_>>>()?;
let recorded_modes = rollout_values
.iter()
.filter(|value| value.get("type").and_then(Value::as_str) == Some("turn_context"))
.filter_map(|value| value.pointer("/payload/multi_agent_mode").cloned())
.collect::<Vec<_>>();
assert_eq!(
recorded_modes,
[
json!({"custom": CUSTOM_MODE_HINT_TEXT}),
json!({"custom": CUSTOM_MODE_HINT_TEXT}),
]
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_configured_mode_hint_suppresses_builtin_text() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response = mount_sse_once(
&server,
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
)
.await;
let test = test_codex()
.with_config(|config| {
configure_multi_agent_v2(config);
config.multi_agent_v2.multi_agent_mode_hint_text = Some(String::new());
})
.build(&server)
.await?;
submit_turn(&test.codex, "hello", Some(ReasoningEffort::High)).await?;
let input = response.single_request().input();
let texts = developer_texts(&input);
assert_eq!(
(
count_containing(&texts, MULTI_AGENT_MODE_OPEN_TAG),
count_containing(&texts, NO_SPAWN_TEXT),
count_containing(&texts, PROACTIVE_TEXT),
),
(1, 0, 0)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn leaving_ultra_after_cold_resume_emits_explicit_mode() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/otel.rs
+18 -12
View File
@@ -567,19 +567,22 @@ async fn process_sse_emits_completed_telemetry() {
mount_sse_once(
&server,
sse(vec![serde_json::json!({
"type": "response.completed",
"response": {
"id": "resp1",
"usage": {
"input_tokens": 3,
"input_tokens_details": { "cached_tokens": 1 },
"output_tokens": 5,
"output_tokens_details": { "reasoning_tokens": 2 },
"total_tokens": 9
sse(vec![
ev_reasoning_item_added("reasoning-1", &[]),
serde_json::json!({
"type": "response.completed",
"response": {
"id": "resp1",
"usage": {
"input_tokens": 3,
"input_tokens_details": { "cached_tokens": 1 },
"output_tokens": 5,
"output_tokens_details": { "reasoning_tokens": 2 },
"total_tokens": 9
}
}
}
})]),
}),
]),
)
.await;
@@ -612,6 +615,9 @@ async fn process_sse_emits_completed_telemetry() {
&& line.contains("cached_token_count=1")
&& line.contains("reasoning_token_count=2")
&& line.contains("tool_token_count=9")
&& extract_log_field(line, "ttft_ms")
.and_then(|ttft_ms| ttft_ms.parse::<i64>().ok())
.is_some()
})
.map(|_| Ok(()))
.unwrap_or(Err("missing response.completed telemetry".to_string()))
codex-rs/core/tests/suite/subagent_notifications.rs
+55 -1
View File
@@ -39,10 +39,13 @@ use serde_json::Value;
use serde_json::json;
use std::fs;
use std::path::Path;
use std::sync::Mutex;
use std::time::Duration;
use test_case::test_case;
use tokio::time::Instant;
use tokio::time::sleep;
use tracing::Level;
use tracing_test::internal::MockWriter;
use wiremock::MockServer;
const SPAWN_CALL_ID: &str = "spawn-call-1";
@@ -96,6 +99,13 @@ fn decoded_body(req: &wiremock::Request) -> Option<Vec<u8>> {
}
}
fn log_field<'a>(line: &'a str, name: &str) -> Option<&'a str> {
let prefix = format!("{name}=");
line.split_ascii_whitespace()
.find_map(|field| field.strip_prefix(&prefix))
.map(|value| value.trim_matches('"'))
}
fn has_subagent_notification(req: &ResponsesRequest) -> bool {
req.message_input_texts("user")
.iter()
@@ -1042,8 +1052,16 @@ async fn spawned_multi_agent_v2_child_inherits_parent_developer_context() -> Res
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[tokio::test]
async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result<()> {
let output: &'static Mutex<Vec<u8>> = Box::leak(Box::new(Mutex::new(Vec::new())));
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.with_max_level(Level::INFO)
.with_writer(MockWriter::new(output))
.finish();
let _guard = tracing::subscriber::set_default(subscriber);
let server = start_mock_server().await;
let encrypted_message = "opaque-encrypted-message";
let spawn_args = serde_json::to_string(&json!({
@@ -1098,6 +1116,7 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result
.expect("test config should allow feature update");
});
let test = builder.build(&server).await?;
let root_thread_id = test.session_configured.thread_id;
test.submit_turn(TURN_1_PROMPT).await?;
@@ -1124,6 +1143,41 @@ async fn encrypted_multi_agent_v2_spawn_sends_agent_message_to_child() -> Result
})])
);
let child_thread_id = test
.thread_manager
.list_thread_ids()
.await
.into_iter()
.find(|thread_id| *thread_id != root_thread_id)
.expect("child thread ID");
let logs = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let logs = String::from_utf8(output.lock().expect("buffer lock").clone())
.expect("logs should be UTF-8");
if logs.contains("kind=\"spawn\"") && logs.contains("state=\"receive\"") {
break logs;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("spawn communication logs should be emitted");
let send = logs
.lines()
.find(|line| line.contains("kind=\"spawn\"") && line.contains("state=\"send\""))
.expect("spawn send event");
assert!(send.contains(&format!("sender_thread_id={root_thread_id}")));
assert!(send.contains(&format!("receiver_thread_id={child_thread_id}")));
assert!(send.contains(&format!("content=\"{encrypted_message}\"")));
let communication_id = log_field(send, "communication_id").expect("communication ID");
logs.lines()
.find(|line| {
line.contains("state=\"receive\"")
&& log_field(line, "communication_id") == Some(communication_id)
})
.expect("correlated receive event");
Ok(())
}
codex-rs/deny.toml
+3 -1
View File
@@ -70,7 +70,7 @@ feature-depth = 1
# A list of advisory IDs to ignore. Note that ignored advisories will still
# output a note when they are encountered.
ignore = [
# Reviewed 2026-06-07. Keep this list in sync with .cargo/audit.toml.
# Reviewed 2026-07-02. Keep this list in sync with .cargo/audit.toml.
# Each exception must identify the dependency path and removal condition.
{ id = "RUSTSEC-2024-0388", reason = "derivative is unmaintained; pulled in via starlark/starlark_syntax v0.14.2 used by execpolicy/cli/core; no fixed starlark release yet" },
{ id = "RUSTSEC-2025-0057", reason = "fxhash is unmaintained; pulled in via starlark_map under starlark v0.14.2 used by execpolicy/cli/core and bm25 used by core; remove when upstream dependencies drop fxhash" },
@@ -82,6 +82,8 @@ ignore = [
{ id = "RUSTSEC-2026-0118", reason = "hickory-proto v0.25.2 is pulled in via rama-dns/rama-tcp used by codex-network-proxy; DNSSEC features are not enabled; remove when rama updates to hickory 0.26.1 or hickory-net" },
{ id = "RUSTSEC-2026-0119", reason = "hickory-proto v0.25.2 is pulled in via rama-dns/rama-tcp used by codex-network-proxy; no fixed rama release is available yet; remove when rama updates to hickory 0.26.1 or hickory-net" },
{ id = "RUSTSEC-2026-0173", reason = "proc-macro-error2 is unmaintained; pulled in via i18n-embed-fl/age used by codex-secrets local storage; remove when codex-secrets migrates off age or age drops i18n-embed-fl" },
{ id = "RUSTSEC-2026-0194", reason = "quick-xml v0.39.4 remains via plist/syntect and wayland-scanner; plist does not exercise the affected APIs and wayland-scanner parses trusted build-time XML; remove when rust-plist#191 and wayland-rs#938 are released" },
{ id = "RUSTSEC-2026-0195", reason = "quick-xml v0.39.4 remains via plist/syntect and wayland-scanner; plist does not exercise the affected APIs and wayland-scanner parses trusted build-time XML; remove when rust-plist#191 and wayland-rs#938 are released" },
]
# If this is true, then cargo deny will use the git executable to fetch advisory database.
# If this is false, then it uses a built-in git library.
codex-rs/exec-server/src/client_transport.rs
+3 -1
View File
@@ -274,7 +274,9 @@ impl ExecServerClient {
connect_async_with_config(
request,
Some(noise_relay_websocket_config()),
/*disable_nagle*/ false,
// Rendezvous sends small, latency-sensitive frames, so avoid Nagle's coalescing delay.
/*disable_nagle*/
true,
),
)
.await
codex-rs/exec-server/src/lib.rs
+1
View File
@@ -29,6 +29,7 @@ mod sandboxed_file_system;
mod server;
mod telemetry;
mod trace_context;
mod websocket_pong_watchdog;
use codex_exec_server_protocol as protocol;
codex-rs/exec-server/src/noise_relay/harness.rs
+249 -65
View File
@@ -6,6 +6,7 @@
//! normal `JsonRpcConnection`. Outbound JSON-RPC is framed and split into Noise
//! records; inbound records are reordered before decryption and reassembly.
use futures::FutureExt;
use futures::Sink;
use futures::SinkExt;
use futures::Stream;
@@ -24,6 +25,7 @@ use crate::connection::CHANNEL_CAPACITY;
use crate::connection::JsonRpcConnection;
use crate::connection::JsonRpcConnectionEvent;
use crate::connection::JsonRpcTransport;
use crate::connection::WEBSOCKET_KEEPALIVE_INTERVAL;
use crate::noise_channel::InitiatorHandshake;
use crate::noise_channel::NoiseChannelIdentity;
use crate::noise_channel::NoiseChannelPublicKey;
@@ -39,6 +41,9 @@ use crate::relay::decode_relay_message_frame;
use crate::relay::encode_relay_message_frame;
use crate::relay_proto::RelayData;
use crate::relay_proto::RelayMessageFrame;
use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT;
use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT_REASON;
use crate::websocket_pong_watchdog::WebSocketPongWatchdog;
/// Values that bind one harness websocket to the intended executor registration.
///
@@ -58,6 +63,8 @@ pub(crate) struct NoiseHarnessConnectionArgs {
// Preserve the availability signal while replacing attacker-controlled reason
// text before it reaches disconnect diagnostics.
const NOISE_RELAY_RESET_DISCONNECT_REASON: &str = "Noise relay stream reset";
// Give a Pong already queued behind data a bounded chance to reach the reader.
const MAX_FRAMES_DRAINED_AFTER_PONG_DEADLINE: usize = 32;
/// Adapt one harness rendezvous websocket into an authenticated JSON-RPC connection.
///
@@ -112,8 +119,7 @@ where
&incoming_tx,
&disconnected_tx,
format!("failed to start Noise relay handshake: {error}"),
)
.await;
);
return;
}
};
@@ -147,8 +153,7 @@ where
&incoming_tx,
&disconnected_tx,
"Noise relay websocket ended during handshake".to_string(),
)
.await;
);
return;
};
let message = match incoming_message {
@@ -158,8 +163,7 @@ where
&incoming_tx,
&disconnected_tx,
"Noise relay websocket received close frame during handshake".to_string(),
)
.await;
);
return;
}
Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_)) => continue,
@@ -168,8 +172,7 @@ where
&incoming_tx,
&disconnected_tx,
"Noise relay transport expects binary protobuf frames".to_string(),
)
.await;
);
return;
}
Err(error) => {
@@ -179,8 +182,7 @@ where
format!(
"failed to read Noise relay websocket from {connection_label}: {error}"
),
)
.await;
);
return;
}
};
@@ -191,8 +193,7 @@ where
&incoming_tx,
&disconnected_tx,
format!("failed to parse Noise relay frame: {error}"),
)
.await;
);
return;
}
};
@@ -209,8 +210,7 @@ where
&incoming_tx,
&disconnected_tx,
format!("invalid Noise relay handshake response: {error}"),
)
.await;
);
return;
}
};
@@ -228,8 +228,7 @@ where
&incoming_tx,
&disconnected_tx,
format!("Noise relay handshake failed: {error}"),
)
.await;
);
return;
}
}
@@ -239,8 +238,7 @@ where
&incoming_tx,
&disconnected_tx,
NOISE_RELAY_RESET_DISCONNECT_REASON.to_string(),
)
.await;
);
return;
}
Ok(
@@ -253,8 +251,7 @@ where
&incoming_tx,
&disconnected_tx,
"Noise relay received data before handshake completion".to_string(),
)
.await;
);
return;
}
}
@@ -263,57 +260,177 @@ where
// After the handshake, each relay sequence maps to exactly one Noise
// transport record. Outbound records are encrypted once; inbound
// records are reordered and deduplicated before decryption.
let mut websocket = websocket.peekable();
let mut next_outbound_seq = 0u32;
let mut inbound_ciphertexts = OrderedCiphertextFrames::default();
let mut inbound_decoder = JsonRpcMessageDecoder::default();
let mut keepalive = tokio::time::interval_at(
tokio::time::Instant::now() + WEBSOCKET_KEEPALIVE_INTERVAL,
WEBSOCKET_KEEPALIVE_INTERVAL,
);
keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut pong_watchdog = WebSocketPongWatchdog::new(WEBSOCKET_PONG_TIMEOUT);
let pong_deadline = tokio::time::sleep(WEBSOCKET_PONG_TIMEOUT);
tokio::pin!(pong_deadline);
// Keep one framed message as a cursor. Sending one Noise record per loop
// creates a scheduling point for keepalive and inbound control frames
// without splitting the WebSocket reader and writer.
let mut pending_outbound: Option<(Vec<u8>, usize)> = None;
let mut force_incoming = false;
let mut frames_drained_after_pong_deadline = 0usize;
'relay: loop {
// Consume a due tick before the always-ready record arm below can win
// another select iteration and postpone the keepalive.
if pong_watchdog.deadline().is_none()
&& keepalive.tick().now_or_never().is_some()
{
if let Err(error) = send_keepalive_ping(
&mut websocket,
&mut pong_watchdog,
pong_deadline.as_mut(),
)
.await
{
warn!("failed to write Noise relay keepalive ping: {error}");
break;
}
frames_drained_after_pong_deadline = 0;
continue;
}
let pong_deadline_expired = pong_watchdog
.deadline()
.is_some_and(|deadline| tokio::time::Instant::now() >= deadline);
// After expiry, inspect only frames already queued. Forcing the peeked
// item through next() makes the 32-frame grace deterministic.
if pong_deadline_expired && !force_incoming {
if frames_drained_after_pong_deadline
< MAX_FRAMES_DRAINED_AFTER_PONG_DEADLINE
&& std::pin::Pin::new(&mut websocket)
.peek()
.now_or_never()
.is_some()
{
force_incoming = true;
} else {
warn!(
noise_reason = WEBSOCKET_PONG_TIMEOUT_REASON,
"Noise harness rendezvous websocket disconnected"
);
send_disconnected(
&incoming_tx,
&disconnected_tx,
WEBSOCKET_PONG_TIMEOUT_REASON.to_string(),
);
return;
}
}
// While a Pong is outstanding, drain already-queued inbound traffic
// before the next fragment so a queued Pong cannot sit behind writes.
if !force_incoming
&& pong_watchdog.deadline().is_some()
&& pending_outbound.is_some()
&& std::pin::Pin::new(&mut websocket)
.peek()
.now_or_never()
.is_some()
{
force_incoming = true;
}
tokio::select! {
maybe_message = outgoing_rx.recv() => {
maybe_message = outgoing_rx.recv(), if pending_outbound.is_none() && !force_incoming && !pong_deadline_expired => {
let Some(message) = maybe_message else {
break;
};
let framed = match frame_jsonrpc_message(&message) {
Ok(framed) => framed,
pending_outbound = Some(match frame_jsonrpc_message(&message) {
Ok(framed) => (framed, 0),
Err(error) => {
warn!("failed to frame JSON-RPC payload for Noise relay: {error}");
break;
}
});
}
_ = std::future::ready(()), if pending_outbound.is_some() && !force_incoming && !pong_deadline_expired => {
let seq = match take_next_sequence(&mut next_outbound_seq) {
Ok(seq) => seq,
Err(error) => {
warn!("Noise relay sequence exhausted: {error}");
break 'relay;
}
};
for plaintext_record in framed.chunks(NOISE_RECORD_PLAINTEXT_LEN) {
let seq = match take_next_sequence(&mut next_outbound_seq) {
Ok(seq) => seq,
Err(error) => {
warn!("Noise relay sequence exhausted: {error}");
break 'relay;
}
let (ciphertext, next_offset, message_complete) = {
let Some((framed, offset)) = pending_outbound.as_ref() else {
continue;
};
let ciphertext = match transport.encrypt(plaintext_record) {
let next_offset = (*offset + NOISE_RECORD_PLAINTEXT_LEN).min(framed.len());
let ciphertext = match transport.encrypt(&framed[*offset..next_offset]) {
Ok(ciphertext) => ciphertext,
Err(error) => {
warn!("failed to encrypt JSON-RPC payload for Noise relay: {error}");
break 'relay;
}
};
let frame = RelayMessageFrame::data(stream_id.clone(), seq, ciphertext);
if let Err(error) = websocket
.send(Message::Binary(encode_relay_message_frame(&frame).into()))
.await
{
warn!("failed to write Noise relay websocket: {error}");
break 'relay;
}
(ciphertext, next_offset, next_offset == framed.len())
};
let frame = RelayMessageFrame::data(stream_id.clone(), seq, ciphertext);
// A Pong can arrive after the readiness check while this write owns the
// combined sink and stream. A single bounded record can therefore hit the
// deadline and disconnect with that Pong queued. Treat that as write
// backpressure; this loop yields only between records.
if let Err(error) = send_websocket_message(
&mut websocket,
Message::Binary(encode_relay_message_frame(&frame).into()),
pong_watchdog.write_deadline(tokio::time::Instant::now()),
)
.await
{
warn!("failed to write Noise relay websocket: {error}");
break 'relay;
}
if message_complete {
pending_outbound = None;
} else if let Some((_framed, offset)) = pending_outbound.as_mut() {
*offset = next_offset;
}
}
_ = &mut pong_deadline, if pong_watchdog.deadline().is_some() && !force_incoming => {
continue;
}
_ = keepalive.tick(), if pong_watchdog.deadline().is_none() => {
if let Err(error) = send_keepalive_ping(
&mut websocket,
&mut pong_watchdog,
pong_deadline.as_mut(),
)
.await
{
warn!("failed to write Noise relay keepalive ping: {error}");
break;
}
frames_drained_after_pong_deadline = 0;
}
incoming_message = websocket.next() => {
force_incoming = false;
let Some(incoming_message) = incoming_message else {
break;
};
// Count each completed read after expiry. If only the deadline arm
// advanced this counter, reads won by a simultaneously ready incoming
// arm would not count toward the 32-frame cap.
if pong_watchdog
.deadline()
.is_some_and(|deadline| tokio::time::Instant::now() >= deadline)
{
frames_drained_after_pong_deadline += 1;
}
match incoming_message {
Ok(Message::Binary(payload)) => {
let frame = match decode_relay_message_frame(payload.as_ref()) {
Ok(frame) => frame,
Err(error) => {
send_malformed(&incoming_tx, error.to_string()).await;
send_malformed(&incoming_tx, error.to_string());
break;
}
};
@@ -325,7 +442,7 @@ where
let data = match frame.into_data() {
Ok(data) => data,
Err(error) => {
send_malformed(&incoming_tx, error.to_string()).await;
send_malformed(&incoming_tx, error.to_string());
break;
}
};
@@ -334,22 +451,26 @@ where
&mut transport,
&mut inbound_decoder,
data,
pong_watchdog.write_deadline(tokio::time::Instant::now()),
&incoming_tx,
)
.await
{
send_malformed(&incoming_tx, error.to_string()).await;
if matches!(error, ExecServerError::Closed) {
break;
}
send_malformed(&incoming_tx, error.to_string());
break;
}
}
Ok(RelayFrameBodyKind::Reset) => {
let _ = incoming_tx
.send(JsonRpcConnectionEvent::Disconnected {
let _ = incoming_tx.try_send(
JsonRpcConnectionEvent::Disconnected {
reason: Some(
NOISE_RELAY_RESET_DISCONNECT_REASON.to_string(),
),
})
.await;
},
);
break;
}
Ok(
@@ -361,20 +482,22 @@ where
send_malformed(
&incoming_tx,
"Noise relay received invalid post-handshake frame".to_string(),
)
.await;
);
break;
}
}
}
Ok(Message::Close(_)) => break,
Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_)) => {}
Ok(Message::Pong(_)) => {
pong_watchdog.received_pong();
frames_drained_after_pong_deadline = 0;
}
Ok(Message::Ping(_) | Message::Frame(_)) => {}
Ok(Message::Text(_)) => {
send_malformed(
&incoming_tx,
"Noise relay transport expects binary protobuf frames".to_string(),
)
.await;
);
break;
}
Err(error) => {
@@ -398,6 +521,46 @@ where
}
}
async fn send_websocket_message<T, E>(
websocket: &mut T,
message: Message,
deadline: tokio::time::Instant,
) -> Result<(), String>
where
T: Sink<Message, Error = E> + Unpin,
E: std::fmt::Display,
{
match tokio::time::timeout_at(deadline, websocket.send(message)).await {
Ok(Ok(())) => Ok(()),
Ok(Err(error)) => Err(error.to_string()),
Err(_) => Err("websocket write timed out".to_string()),
}
}
async fn send_keepalive_ping<T, E>(
websocket: &mut T,
pong_watchdog: &mut WebSocketPongWatchdog,
pong_deadline: std::pin::Pin<&mut tokio::time::Sleep>,
) -> Result<(), String>
where
T: Sink<Message, Error = E> + Unpin,
E: std::fmt::Display,
{
send_websocket_message(
websocket,
Message::Ping(Vec::new().into()),
pong_watchdog.write_deadline(tokio::time::Instant::now()),
)
.await?;
// Start the response clock after the Ping flushes; waiting for sink capacity
// is governed by the write deadline above.
pong_watchdog.ping_sent(tokio::time::Instant::now());
if let Some(deadline) = pong_watchdog.deadline() {
pong_deadline.reset(deadline);
}
Ok(())
}
/// Order and decrypt one relay frame, then emit any complete JSON-RPC messages.
/// Relay records and JSON-RPC messages do not share boundaries, so reassembly
/// happens after decryption.
@@ -406,6 +569,7 @@ async fn receive_data(
transport: &mut NoiseTransport,
decoder: &mut JsonRpcMessageDecoder,
data: RelayData,
delivery_deadline: tokio::time::Instant,
incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
) -> Result<(), ExecServerError> {
// Ordering must happen before decryption because Noise transport nonces are
@@ -419,30 +583,50 @@ async fn receive_data(
// The authenticated byte stream can carry partial or multiple JSON-RPC
// messages; emit only complete, successfully parsed messages.
for message in decoder.push(&plaintext)? {
incoming_tx
.send(JsonRpcConnectionEvent::Message(message))
.await
.map_err(|_| ExecServerError::Closed)?;
send_incoming_event(
incoming_tx,
JsonRpcConnectionEvent::Message(message),
delivery_deadline,
)
.await?;
}
}
Ok(())
}
async fn send_malformed(incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>, reason: String) {
let _ = incoming_tx
.send(JsonRpcConnectionEvent::MalformedMessage { reason })
.await;
async fn send_incoming_event(
incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
event: JsonRpcConnectionEvent,
deadline: tokio::time::Instant,
) -> Result<(), ExecServerError> {
match tokio::time::timeout_at(deadline, incoming_tx.send(event)).await {
Ok(Ok(())) => Ok(()),
Ok(Err(_)) => Err(ExecServerError::Closed),
Err(_) => {
warn!(
noise_reason = "application_backpressure",
"Noise harness application event delivery timed out"
);
Err(ExecServerError::Closed)
}
}
}
async fn send_disconnected(
fn send_malformed(incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>, reason: String) {
let _ = incoming_tx.try_send(JsonRpcConnectionEvent::MalformedMessage { reason });
}
fn send_disconnected(
incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
disconnected_tx: &watch::Sender<bool>,
reason: String,
) {
let _ = disconnected_tx.send(true);
let _ = incoming_tx
.send(JsonRpcConnectionEvent::Disconnected {
reason: Some(reason),
})
.await;
let _ = incoming_tx.try_send(JsonRpcConnectionEvent::Disconnected {
reason: Some(reason),
});
}
#[cfg(test)]
#[path = "harness_tests.rs"]
mod tests;
codex-rs/exec-server/src/noise_relay/harness_tests.rs
+442
View File
@@ -0,0 +1,442 @@
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::task::Context as TaskContext;
use std::task::Poll;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use codex_exec_server_protocol::JSONRPCMessage;
use codex_exec_server_protocol::JSONRPCRequest;
use codex_exec_server_protocol::RequestId;
use futures::Sink;
use futures::SinkExt;
use futures::StreamExt;
use futures::channel::mpsc as futures_mpsc;
use pretty_assertions::assert_eq;
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use tokio::time::Instant;
use tokio::time::timeout;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
use super::*;
use crate::connection::JsonRpcConnectionEvent;
use crate::noise_channel::PendingResponderHandshake;
const ENVIRONMENT_ID: &str = "environment-1";
const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
#[tokio::test]
async fn fragmented_writes_yield_to_keepalive_and_queued_pong() -> Result<()> {
let (connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
connection
.outgoing_tx
.send(JSONRPCMessage::Request(JSONRPCRequest {
id: RequestId::Integer(1),
method: "large".to_string(),
params: Some(serde_json::json!({
"payload": "x".repeat(NOISE_RECORD_PLAINTEXT_LEN * 3),
})),
trace: None,
}))
.await?;
control.wait_for_blocked_write(/*expected*/ 1).await?;
tokio::time::sleep(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
control.grant_writes(/*count*/ 1);
let first_data = read_outbound_data(&mut outbound_rx).await?;
assert_eq!(first_data.seq, 0);
control.wait_for_blocked_write(/*expected*/ 2).await?;
control.grant_writes(/*count*/ 1);
let Message::Ping(ping_payload) = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed before sending keepalive")?
else {
anyhow::bail!("expected keepalive between fragmented writes");
};
control.wait_for_blocked_write(/*expected*/ 3).await?;
control.send_inbound(Message::Pong(ping_payload))?;
tokio::time::sleep(WEBSOCKET_KEEPALIVE_INTERVAL + Duration::from_millis(10)).await;
control.grant_writes(/*count*/ 1);
let second_data = read_outbound_data(&mut outbound_rx).await?;
assert_eq!(second_data.seq, 1);
control.wait_for_blocked_write(/*expected*/ 4).await?;
control.grant_writes(/*count*/ 1);
let next_message = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed after receiving queued Pong")?;
assert!(matches!(next_message, Message::Ping(_)));
for task in &connection.task_handles {
task.abort();
}
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn post_deadline_drain_stops_before_frame_33() -> Result<()> {
let (mut connection, mut control, mut outbound_rx) = connected_controlled_harness().await?;
control.wait_for_blocked_write(/*expected*/ 1).await?;
control.grant_writes(/*count*/ 1);
let Message::Ping(ping_payload) = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed before sending keepalive")?
else {
anyhow::bail!("expected keepalive ping");
};
let reads_before_deadline = control.inbound_reads();
let unrelated_frame =
encode_relay_message_frame(&RelayMessageFrame::resume("unrelated-stream".to_string()));
for _ in 0..MAX_FRAMES_DRAINED_AFTER_PONG_DEADLINE {
control.send_inbound(Message::Binary(unrelated_frame.clone().into()))?;
}
control.send_inbound(Message::Pong(ping_payload))?;
// Keep the current-thread runtime from consuming the queued frames until the
// Pong deadline and every frame are ready together.
std::thread::sleep(WEBSOCKET_PONG_TIMEOUT + Duration::from_millis(10));
let event = timeout(Duration::from_secs(1), connection.incoming_rx.recv()).await?;
let Some(JsonRpcConnectionEvent::Disconnected { reason }) = event else {
anyhow::bail!("expected Pong timeout, got {event:?}");
};
assert_eq!(reason.as_deref(), Some(WEBSOCKET_PONG_TIMEOUT_REASON));
assert_eq!(
control.inbound_reads() - reads_before_deadline,
MAX_FRAMES_DRAINED_AFTER_PONG_DEADLINE
);
Ok(())
}
#[tokio::test]
async fn pong_keeps_harness_alive_until_peer_stops_responding() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let websocket_url = format!("ws://{}", listener.local_addr()?);
let harness_connection = tokio::spawn(connect_async(websocket_url));
let (socket, _peer_addr) = listener.accept().await?;
let mut executor_websocket = accept_async(socket).await?;
let (harness_websocket, _response) = harness_connection.await??;
let executor_identity = NoiseChannelIdentity::generate()?;
let mut connection = noise_harness_connection_from_websocket(
harness_websocket,
NoiseHarnessConnectionArgs {
connection_label: "test rendezvous".to_string(),
environment_id: ENVIRONMENT_ID.to_string(),
executor_registration_id: EXECUTOR_REGISTRATION_ID.to_string(),
identity: NoiseChannelIdentity::generate()?,
responder_public_key: executor_identity.public_key(),
harness_key_authorization: "authorization".to_string(),
},
);
let resume_message = timeout(Duration::from_secs(1), executor_websocket.next())
.await?
.context("harness closed before sending resume")??;
let Message::Binary(resume_payload) = resume_message else {
anyhow::bail!("expected resume frame, got {resume_message:?}");
};
let resume = decode_relay_message_frame(resume_payload.as_ref())?;
assert_eq!(resume.validate()?, RelayFrameBodyKind::Resume);
let handshake_message = timeout(Duration::from_secs(1), executor_websocket.next())
.await?
.context("harness closed before sending handshake")??;
let Message::Binary(handshake_payload) = handshake_message else {
anyhow::bail!("expected handshake frame, got {handshake_message:?}");
};
let handshake = decode_relay_message_frame(handshake_payload.as_ref())?;
assert_eq!(handshake.stream_id, resume.stream_id);
let stream_id = handshake.stream_id.clone();
let prologue =
noise_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id.as_str());
let pending = PendingResponderHandshake::read_request(
&executor_identity,
&prologue,
&handshake.into_handshake_payload()?,
)?;
let (_transport, response) = pending.complete()?;
let response = RelayMessageFrame::handshake(stream_id, response);
executor_websocket
.send(Message::Binary(
encode_relay_message_frame(&response).into(),
))
.await?;
let mut pings = 0;
while pings < 6 {
let message = timeout(Duration::from_secs(1), executor_websocket.next())
.await?
.context("harness disconnected before six keepalive pings")??;
match message {
Message::Ping(payload) => {
executor_websocket.send(Message::Pong(payload)).await?;
pings += 1;
}
Message::Pong(_) | Message::Frame(_) => {}
message => anyhow::bail!("expected keepalive ping, got {message:?}"),
}
}
// Keep non-Pong traffic flowing after responses stop. It must not defeat
// the bounded grace for a Pong already queued behind data.
let unrelated_frame =
encode_relay_message_frame(&RelayMessageFrame::resume("unrelated-stream".to_string()));
let traffic_task = tokio::spawn(async move {
loop {
if executor_websocket
.send(Message::Binary(unrelated_frame.clone().into()))
.await
.is_err()
{
break;
}
tokio::time::sleep(Duration::from_millis(5)).await;
}
});
let event = timeout(Duration::from_secs(1), connection.incoming_rx.recv()).await?;
traffic_task.abort();
let _ = traffic_task.await;
let Some(JsonRpcConnectionEvent::Disconnected { reason }) = event else {
anyhow::bail!("expected pong timeout, got {event:?}");
};
assert_eq!(reason.as_deref(), Some(WEBSOCKET_PONG_TIMEOUT_REASON));
Ok(())
}
#[tokio::test]
async fn application_event_delivery_is_bounded() -> Result<()> {
let (incoming_tx, _incoming_rx) = mpsc::channel(1);
incoming_tx
.send(JsonRpcConnectionEvent::MalformedMessage {
reason: "fill queue".to_string(),
})
.await?;
let result = send_incoming_event(
&incoming_tx,
JsonRpcConnectionEvent::MalformedMessage {
reason: "blocked event".to_string(),
},
Instant::now() + Duration::from_millis(10),
)
.await;
assert!(matches!(result, Err(ExecServerError::Closed)));
Ok(())
}
async fn read_outbound_data(
outbound_rx: &mut futures_mpsc::UnboundedReceiver<Message>,
) -> Result<RelayData> {
let Message::Binary(payload) = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed before sending data")?
else {
anyhow::bail!("expected relay data frame");
};
let frame = decode_relay_message_frame(payload.as_ref())?;
assert_eq!(frame.validate()?, RelayFrameBodyKind::Data);
frame.into_data().map_err(anyhow::Error::from)
}
async fn connected_controlled_harness() -> Result<(
JsonRpcConnection,
ControlledWebSocketHandle,
futures_mpsc::UnboundedReceiver<Message>,
)> {
let (websocket, control, mut outbound_rx) = ControlledWebSocket::new(/*write_permits*/ 2);
let executor_identity = NoiseChannelIdentity::generate()?;
let connection = noise_harness_connection_from_websocket(
websocket,
NoiseHarnessConnectionArgs {
connection_label: "test rendezvous".to_string(),
environment_id: ENVIRONMENT_ID.to_string(),
executor_registration_id: EXECUTOR_REGISTRATION_ID.to_string(),
identity: NoiseChannelIdentity::generate()?,
responder_public_key: executor_identity.public_key(),
harness_key_authorization: "authorization".to_string(),
},
);
let Message::Binary(resume_payload) = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed before sending resume")?
else {
anyhow::bail!("expected resume frame");
};
let resume = decode_relay_message_frame(resume_payload.as_ref())?;
let Message::Binary(handshake_payload) = timeout(Duration::from_secs(1), outbound_rx.next())
.await?
.context("harness closed before sending handshake")?
else {
anyhow::bail!("expected handshake frame");
};
let handshake = decode_relay_message_frame(handshake_payload.as_ref())?;
let stream_id = handshake.stream_id.clone();
assert_eq!(stream_id, resume.stream_id);
let prologue =
noise_channel_prologue(ENVIRONMENT_ID, EXECUTOR_REGISTRATION_ID, stream_id.as_str());
let pending = PendingResponderHandshake::read_request(
&executor_identity,
&prologue,
&handshake.into_handshake_payload()?,
)?;
let (_transport, response) = pending.complete()?;
control.send_inbound(Message::Binary(
encode_relay_message_frame(&RelayMessageFrame::handshake(stream_id, response)).into(),
))?;
Ok((connection, control, outbound_rx))
}
struct ControlledWebSocket {
inbound_rx: futures_mpsc::UnboundedReceiver<Result<Message, std::convert::Infallible>>,
outbound_tx: futures_mpsc::UnboundedSender<Message>,
write_permit_rx: futures_mpsc::UnboundedReceiver<()>,
blocked_write_tx: futures_mpsc::UnboundedSender<usize>,
write_waiting: bool,
blocked_writes: usize,
inbound_reads: Arc<AtomicUsize>,
}
struct ControlledWebSocketHandle {
inbound_tx: futures_mpsc::UnboundedSender<Result<Message, std::convert::Infallible>>,
write_permit_tx: futures_mpsc::UnboundedSender<()>,
blocked_write_rx: futures_mpsc::UnboundedReceiver<usize>,
inbound_reads: Arc<AtomicUsize>,
}
impl ControlledWebSocket {
fn new(
write_permits: usize,
) -> (
Self,
ControlledWebSocketHandle,
futures_mpsc::UnboundedReceiver<Message>,
) {
let (inbound_tx, inbound_rx) = futures_mpsc::unbounded();
let (outbound_tx, outbound_rx) = futures_mpsc::unbounded();
let (write_permit_tx, write_permit_rx) = futures_mpsc::unbounded();
let (blocked_write_tx, blocked_write_rx) = futures_mpsc::unbounded();
for _ in 0..write_permits {
write_permit_tx
.unbounded_send(())
.expect("test write permit receiver should stay open");
}
let inbound_reads = Arc::new(AtomicUsize::new(0));
(
Self {
inbound_rx,
outbound_tx,
write_permit_rx,
blocked_write_tx,
write_waiting: false,
blocked_writes: 0,
inbound_reads: Arc::clone(&inbound_reads),
},
ControlledWebSocketHandle {
inbound_tx,
write_permit_tx,
blocked_write_rx,
inbound_reads,
},
outbound_rx,
)
}
}
impl ControlledWebSocketHandle {
fn send_inbound(&self, message: Message) -> Result<()> {
self.inbound_tx
.unbounded_send(Ok(message))
.map_err(anyhow::Error::from)
}
fn grant_writes(&self, count: usize) {
for _ in 0..count {
self.write_permit_tx
.unbounded_send(())
.expect("test write permit receiver should stay open");
}
}
fn inbound_reads(&self) -> usize {
self.inbound_reads.load(Ordering::Acquire)
}
async fn wait_for_blocked_write(&mut self, expected: usize) -> Result<()> {
let actual = timeout(Duration::from_secs(1), self.blocked_write_rx.next())
.await?
.context("websocket closed before blocking the expected write")?;
assert_eq!(actual, expected);
Ok(())
}
}
impl Sink<Message> for ControlledWebSocket {
type Error = std::convert::Infallible;
fn poll_ready(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Result<(), Self::Error>> {
let this = self.get_mut();
match Pin::new(&mut this.write_permit_rx).poll_next(cx) {
Poll::Ready(Some(())) => {
this.write_waiting = false;
Poll::Ready(Ok(()))
}
Poll::Ready(None) | Poll::Pending => {
if !this.write_waiting {
this.write_waiting = true;
this.blocked_writes += 1;
this.blocked_write_tx
.unbounded_send(this.blocked_writes)
.expect("test blocked-write receiver should stay open");
}
Poll::Pending
}
}
}
fn start_send(self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
self.outbound_tx
.unbounded_send(item)
.expect("test outbound receiver should stay open");
Ok(())
}
fn poll_flush(
self: Pin<&mut Self>,
_cx: &mut TaskContext<'_>,
) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(
self: Pin<&mut Self>,
_cx: &mut TaskContext<'_>,
) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
}
impl futures::Stream for ControlledWebSocket {
type Item = Result<Message, std::convert::Infallible>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Option<Self::Item>> {
let result = Pin::new(&mut self.inbound_rx).poll_next(cx);
if matches!(result, Poll::Ready(Some(_))) {
self.inbound_reads.fetch_add(1, Ordering::Release);
}
result
}
}
codex-rs/exec-server/src/relay.rs
+92 -10
View File
@@ -41,6 +41,9 @@ use crate::relay_proto::RelayReset;
use crate::relay_proto::RelayResume;
use crate::relay_proto::relay_message_frame;
use crate::server::ConnectionProcessor;
use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT;
use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT_REASON;
use crate::websocket_pong_watchdog::WebSocketPongWatchdog;
const RELAY_MESSAGE_FRAME_VERSION: u32 = 1;
const MAX_ACTIVE_NOISE_RELAY_STREAMS: usize = 128;
@@ -49,6 +52,27 @@ const MAX_HARNESS_KEY_AUTHORIZATION_BYTES: usize = 4096;
const MAX_PENDING_HANDSHAKE_VALIDATIONS: usize = 32;
const HARNESS_KEY_VALIDATION_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub(crate) enum RendezvousDisconnectReason {
PeerClose,
ReadError,
WriteError,
PongTimeout,
LocalShutdown,
}
impl RendezvousDisconnectReason {
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::PeerClose => "peer_close",
Self::ReadError => "read_error",
Self::WriteError => "write_error",
Self::PongTimeout => WEBSOCKET_PONG_TIMEOUT_REASON,
Self::LocalShutdown => "local_shutdown",
}
}
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub(crate) enum RelayFrameBodyKind {
Data,
@@ -448,7 +472,8 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
executor_registration_id: String,
identity: NoiseChannelIdentity,
validator: V,
) where
) -> RendezvousDisconnectReason
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
V: HarnessKeyValidator + Clone + 'static,
{
@@ -461,6 +486,7 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
mpsc::channel::<Vec<u8>>(CHANNEL_CAPACITY);
let (closed_stream_tx, mut closed_stream_rx) =
mpsc::channel::<ClosedNoiseVirtualStream>(MAX_ACTIVE_NOISE_RELAY_STREAMS);
let (pong_tx, mut pong_rx) = mpsc::channel(1);
// Use a separate writer so this loop never waits on the channel it drains.
let mut physical_writer_task = tokio::spawn(async move {
let mut keepalive = tokio::time::interval_at(
@@ -468,19 +494,61 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
WEBSOCKET_KEEPALIVE_INTERVAL,
);
keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut pong_watchdog = WebSocketPongWatchdog::new(WEBSOCKET_PONG_TIMEOUT);
let pong_deadline = tokio::time::sleep(WEBSOCKET_PONG_TIMEOUT);
tokio::pin!(pong_deadline);
loop {
let message = tokio::select! {
pong = pong_rx.recv() => {
let Some(()) = pong else {
break RendezvousDisconnectReason::LocalShutdown;
};
pong_watchdog.received_pong();
continue;
}
_ = &mut pong_deadline, if pong_watchdog.deadline().is_some() => {
match pong_rx.try_recv() {
Ok(()) => {
pong_watchdog.received_pong();
continue;
}
Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
break RendezvousDisconnectReason::PongTimeout;
}
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
break RendezvousDisconnectReason::LocalShutdown;
}
}
}
_ = keepalive.tick(), if pong_watchdog.deadline().is_none() => {
Message::Ping(Vec::new().into())
}
encoded = physical_outgoing_rx.recv() => {
let Some(encoded) = encoded else {
break;
break RendezvousDisconnectReason::LocalShutdown;
};
Message::Binary(encoded.into())
}
_ = keepalive.tick() => Message::Ping(Vec::new().into()),
};
if let Err(error) = websocket_sink.send(message).await {
warn!("Noise multiplexed environment websocket write failed: {error}");
break;
let is_keepalive_ping = matches!(message, Message::Ping(_));
let write_deadline = pong_watchdog.write_deadline(tokio::time::Instant::now());
match tokio::time::timeout_at(write_deadline, websocket_sink.send(message)).await {
Ok(Ok(())) => {
if is_keepalive_ping {
pong_watchdog.ping_sent(tokio::time::Instant::now());
if let Some(deadline) = pong_watchdog.deadline() {
pong_deadline.as_mut().reset(deadline);
}
}
}
Ok(Err(error)) => {
warn!("Noise multiplexed environment websocket write failed: {error}");
break RendezvousDisconnectReason::WriteError;
}
Err(_) => {
warn!("Noise multiplexed environment websocket write timed out");
break RendezvousDisconnectReason::WriteError;
}
}
}
});
@@ -489,13 +557,18 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
let mut validation_tasks: JoinSet<HarnessKeyValidationResult> = JoinSet::new();
let mut failed_handshakes = 0usize;
let mut next_validation_id = 0u64;
let mut disconnect_reason = RendezvousDisconnectReason::LocalShutdown;
loop {
// Registry calls run separately so a slow check does not block the relay.
let frame = tokio::select! {
writer_result = &mut physical_writer_task => {
if let Err(error) = writer_result {
warn!("Noise multiplexed environment websocket writer failed: {error}");
match writer_result {
Ok(reason) => disconnect_reason = reason,
Err(error) => {
warn!("Noise multiplexed environment websocket writer failed: {error}");
disconnect_reason = RendezvousDisconnectReason::LocalShutdown;
}
}
break;
}
@@ -621,14 +694,22 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
continue;
}
},
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_))) => continue,
Some(Ok(Message::Close(_))) | None => {
disconnect_reason = RendezvousDisconnectReason::PeerClose;
break;
}
Some(Ok(Message::Pong(_))) => {
let _ = pong_tx.try_send(());
continue;
}
Some(Ok(Message::Ping(_) | Message::Frame(_))) => continue,
Some(Ok(Message::Text(_))) => {
warn!("dropping non-binary Noise relay frame from harness");
continue;
}
Some(Err(error)) => {
debug!("Noise multiplexed environment websocket read failed: {error}");
disconnect_reason = RendezvousDisconnectReason::ReadError;
break;
}
}
@@ -802,6 +883,7 @@ pub(crate) async fn run_multiplexed_environment<S, V>(
physical_writer_task.abort();
let _ = physical_writer_task.await;
}
disconnect_reason
}
/// Charge one failed authenticated-channel attempt to this physical relay.
codex-rs/exec-server/src/relay_noise_tests.rs
+81
View File
@@ -17,6 +17,7 @@ use tokio_tungstenite::tungstenite::Message;
use super::HarnessKeyValidator;
use super::MAX_FAILED_NOISE_HANDSHAKES;
use super::MAX_HARNESS_KEY_AUTHORIZATION_BYTES;
use super::RendezvousDisconnectReason;
use super::run_multiplexed_environment;
use crate::ExecServerError;
use crate::ExecServerRuntimePaths;
@@ -33,6 +34,86 @@ use crate::server::ConnectionProcessor;
const ENVIRONMENT_ID: &str = "environment-1";
const EXECUTOR_REGISTRATION_ID: &str = "registration-1";
#[tokio::test]
async fn missing_pong_disconnects_physical_relay() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let websocket_url = format!("ws://{}", listener.local_addr()?);
let harness_connection = tokio::spawn(connect_async(websocket_url));
let (socket, _peer_addr) = listener.accept().await?;
let environment_websocket = accept_async(socket).await?;
let (_harness_websocket, _response) = harness_connection.await??;
let environment_task = tokio::spawn(run_multiplexed_environment(
environment_websocket,
ConnectionProcessor::new(ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?),
ENVIRONMENT_ID.to_string(),
EXECUTOR_REGISTRATION_ID.to_string(),
NoiseChannelIdentity::generate()?,
BlockingValidator {
calls: Arc::new(AtomicUsize::new(0)),
release: Arc::new(Notify::new()),
},
));
assert_eq!(
timeout(Duration::from_secs(1), environment_task).await??,
RendezvousDisconnectReason::PongTimeout
);
Ok(())
}
#[tokio::test]
async fn pong_keeps_physical_relay_connected() -> Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let websocket_url = format!("ws://{}", listener.local_addr()?);
let harness_connection = tokio::spawn(connect_async(websocket_url));
let (socket, _peer_addr) = listener.accept().await?;
let environment_websocket = accept_async(socket).await?;
let (mut harness_websocket, _response) = harness_connection.await??;
let environment_task = tokio::spawn(run_multiplexed_environment(
environment_websocket,
ConnectionProcessor::new(ExecServerRuntimePaths::new(
std::env::current_exe()?,
/*codex_linux_sandbox_exe*/ None,
)?),
ENVIRONMENT_ID.to_string(),
EXECUTOR_REGISTRATION_ID.to_string(),
NoiseChannelIdentity::generate()?,
BlockingValidator {
calls: Arc::new(AtomicUsize::new(0)),
release: Arc::new(Notify::new()),
},
));
timeout(Duration::from_secs(1), async {
let mut pings = 0;
while pings < 6 {
match harness_websocket.next().await {
Some(Ok(Message::Ping(payload))) => {
harness_websocket.send(Message::Pong(payload)).await?;
pings += 1;
}
Some(Ok(Message::Pong(_) | Message::Frame(_))) => {}
Some(Ok(message)) => anyhow::bail!("expected keepalive ping, got {message:?}"),
Some(Err(error)) => return Err(error.into()),
None => anyhow::bail!("environment disconnected before six keepalive pings"),
}
}
Ok::<_, anyhow::Error>(())
})
.await??;
harness_websocket.close(None).await?;
assert_eq!(
timeout(Duration::from_secs(1), environment_task).await??,
RendezvousDisconnectReason::PeerClose
);
Ok(())
}
#[derive(Clone)]
struct BlockingValidator {
calls: Arc<AtomicUsize>,
codex-rs/exec-server/src/remote.rs
+13 -3
View File
@@ -492,7 +492,7 @@ pub async fn run_remote_environment(
noise_outcome = "ok",
"Noise executor connected to rendezvous"
);
run_multiplexed_environment(
let disconnect_reason = run_multiplexed_environment(
websocket,
processor.clone(),
response.environment_id.clone(),
@@ -505,7 +505,15 @@ pub async fn run_remote_environment(
},
)
.await;
config.telemetry.remote_reconnect("disconnected");
info!(
noise_event = "rendezvous_connection",
noise_outcome = "disconnected",
noise_reason = disconnect_reason.as_str(),
"Noise executor disconnected from rendezvous"
);
config
.telemetry
.remote_reconnect(disconnect_reason.as_str());
}
Err(error) => {
let registration_rejected = matches!(
@@ -561,7 +569,9 @@ async fn connect_rendezvous(
connect_async_with_config(
request,
Some(noise_relay_websocket_config()),
/*disable_nagle*/ false,
// Rendezvous sends small, latency-sensitive frames, so avoid Nagle's coalescing delay.
/*disable_nagle*/
true,
)
.await
.map(|(websocket, _)| websocket)
codex-rs/exec-server/src/websocket_pong_watchdog.rs
+44
View File
@@ -0,0 +1,44 @@
use std::time::Duration;
use tokio::time::Instant;
#[cfg(test)]
pub(crate) const WEBSOCKET_PONG_TIMEOUT: Duration = Duration::from_millis(100);
#[cfg(not(test))]
pub(crate) const WEBSOCKET_PONG_TIMEOUT: Duration = Duration::from_secs(60);
pub(crate) const WEBSOCKET_PONG_TIMEOUT_REASON: &str = "pong_timeout";
/// Tracks whether a WebSocket peer has acknowledged a keepalive ping.
pub(crate) struct WebSocketPongWatchdog {
timeout: Duration,
deadline: Option<Instant>,
}
impl WebSocketPongWatchdog {
pub(crate) fn new(timeout: Duration) -> Self {
Self {
timeout,
deadline: None,
}
}
pub(crate) fn ping_sent(&mut self, now: Instant) {
self.deadline.get_or_insert(now + self.timeout);
}
pub(crate) fn deadline(&self) -> Option<Instant> {
self.deadline
}
pub(crate) fn write_deadline(&self, now: Instant) -> Instant {
self.deadline.unwrap_or(now + self.timeout)
}
pub(crate) fn received_pong(&mut self) {
self.deadline = None;
}
}
#[cfg(test)]
#[path = "websocket_pong_watchdog_tests.rs"]
mod tests;
codex-rs/exec-server/src/websocket_pong_watchdog_tests.rs
+34
View File
@@ -0,0 +1,34 @@
use std::time::Duration;
use pretty_assertions::assert_eq;
use tokio::time::Instant;
use super::WebSocketPongWatchdog;
#[test]
fn repeated_ping_does_not_extend_deadline() {
let started_at = Instant::now();
let timeout = Duration::from_secs(2);
let mut watchdog = WebSocketPongWatchdog::new(timeout);
watchdog.ping_sent(started_at);
watchdog.ping_sent(started_at + Duration::from_secs(1));
assert_eq!(
watchdog.write_deadline(started_at + Duration::from_secs(1)),
started_at + timeout
);
assert_eq!(watchdog.deadline(), Some(started_at + timeout));
}
#[test]
fn pong_starts_a_fresh_deadline() {
let started_at = Instant::now();
let timeout = Duration::from_secs(2);
let mut watchdog = WebSocketPongWatchdog::new(timeout);
watchdog.ping_sent(started_at);
watchdog.received_pong();
assert_eq!(watchdog.deadline(), None);
watchdog.ping_sent(started_at + timeout);
assert_eq!(watchdog.deadline(), Some(started_at + timeout + timeout));
}
codex-rs/features/src/feature_configs.rs
+2
View File
@@ -56,6 +56,8 @@ pub struct MultiAgentV2ConfigToml {
#[serde(skip_serializing_if = "Option::is_none")]
pub subagent_usage_hint_text: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub multi_agent_mode_hint_text: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
#[schemars(length(min = 1, max = 64), regex(pattern = r"^[a-zA-Z0-9_-]+$"))]
pub tool_namespace: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
codex-rs/features/src/tests.rs
+2
View File
@@ -610,6 +610,7 @@ usage_hint_enabled = false
usage_hint_text = "Custom delegation guidance."
root_agent_usage_hint_text = "Root guidance."
subagent_usage_hint_text = "Subagent guidance."
multi_agent_mode_hint_text = "Custom mode guidance."
tool_namespace = "agents"
hide_spawn_agent_metadata = true
non_code_mode_only = true
@@ -633,6 +634,7 @@ non_code_mode_only = true
usage_hint_text: Some("Custom delegation guidance.".to_string()),
root_agent_usage_hint_text: Some("Root guidance.".to_string()),
subagent_usage_hint_text: Some("Subagent guidance.".to_string()),
multi_agent_mode_hint_text: Some("Custom mode guidance.".to_string()),
tool_namespace: Some("agents".to_string()),
hide_spawn_agent_metadata: Some(true),
non_code_mode_only: Some(true),
codex-rs/model-provider/src/amazon_bedrock/catalog.rs
+11
View File
@@ -61,6 +61,8 @@ fn gpt_5_bedrock_model(openai_slug: &str, bedrock_slug: &str, priority: i32) ->
model.priority = priority;
model.context_window = Some(GPT_5_BEDROCK_CONTEXT_WINDOW);
model.max_context_window = Some(GPT_5_BEDROCK_CONTEXT_WINDOW);
model.availability_nux = None;
model.upgrade = None;
model
}
@@ -127,6 +129,15 @@ mod tests {
}
}
#[test]
fn gpt_5_bedrock_models_do_not_include_availability_nux_or_upgrade() {
let catalog = static_model_catalog();
for model in catalog.models {
assert_eq!((model.availability_nux, model.upgrade), (None, None));
}
}
#[test]
fn gpt_5_6_bedrock_models_clone_gpt_5_5_config_with_max_reasoning_effort() {
let catalog = static_model_catalog();
codex-rs/otel/src/events/session_telemetry.rs
+2
View File
@@ -910,6 +910,7 @@ impl SessionTelemetry {
cached_token_count: Option<i64>,
reasoning_token_count: Option<i64>,
tool_token_count: i64,
ttft_ms: Option<i64>,
) {
log_and_trace_event!(
self,
@@ -921,6 +922,7 @@ impl SessionTelemetry {
cached_token_count = cached_token_count,
reasoning_token_count = reasoning_token_count,
tool_token_count = %tool_token_count,
ttft_ms = ttft_ms,
service_tier = self.metadata.service_tier.as_deref(),
model_reasoning_effort = self.metadata.model_reasoning_effort.as_deref(),
},
codex-rs/protocol/src/config_types.rs
+25 -9
View File
@@ -296,23 +296,39 @@ pub enum Personality {
Pragmatic,
}
/// Controls whether the model receives multi-agent delegation instructions and,
/// when it does, whether it should only spawn sub-agents after an explicit user
/// request or may delegate proactively when doing so would help. `none` leaves
/// the multi-agent tools available without injecting delegation instructions.
#[derive(
Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Display, JsonSchema, TS, Default,
)]
#[serde(rename_all = "camelCase")]
/// Controls the effective multi-agent delegation instructions for a turn. `custom` means the
/// configured mode hint defines the policy instead of a built-in policy.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Display, JsonSchema, TS, Default)]
#[serde(rename_all = "camelCase", from = "MultiAgentModeWire")]
#[ts(rename_all = "camelCase")]
#[strum(serialize_all = "camelCase")]
pub enum MultiAgentMode {
None,
Custom(String),
#[default]
ExplicitRequestOnly,
Proactive,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
enum MultiAgentModeWire {
None,
Custom(String),
ExplicitRequestOnly,
Proactive,
}
impl From<MultiAgentModeWire> for MultiAgentMode {
fn from(value: MultiAgentModeWire) -> Self {
match value {
MultiAgentModeWire::None => Self::Custom(String::new()),
MultiAgentModeWire::Custom(hint_text) => Self::Custom(hint_text),
MultiAgentModeWire::ExplicitRequestOnly => Self::ExplicitRequestOnly,
MultiAgentModeWire::Proactive => Self::Proactive,
}
}
}
#[derive(
Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Display, JsonSchema, TS, Default,
)]
codex-rs/protocol/src/protocol.rs
+20 -1
View File
@@ -2668,7 +2668,7 @@ impl InitialHistory {
| RolloutItem::WorldState(_)
| RolloutItem::EventMsg(_) => None,
})
.and_then(|turn_context| turn_context.multi_agent_mode)
.and_then(|turn_context| turn_context.multi_agent_mode.clone())
}
pub fn get_resumed_session_sources(&self) -> Option<(SessionSource, Option<ThreadSource>)> {
@@ -5741,6 +5741,25 @@ mod tests {
Ok(())
}
#[test]
fn latest_effective_multi_agent_mode_maps_legacy_none_to_empty_custom() -> Result<()> {
let value = json!({
"cwd": test_path_buf("/tmp"),
"approval_policy": "never",
"sandbox_policy": { "type": "danger-full-access" },
"model": "gpt-5",
"multi_agent_mode": "none",
"summary": "auto",
});
let item = RolloutItem::TurnContext(serde_json::from_value(value)?);
assert_eq!(
InitialHistory::Forked(vec![item]).get_latest_effective_multi_agent_mode(),
Some(MultiAgentMode::Custom(String::new()))
);
Ok(())
}
#[test]
fn turn_context_item_serializes_network_when_present() -> Result<()> {
let item = TurnContextItem {
codex-rs/tui/src/history_cell/notices.rs
+1 -1
View File
@@ -96,7 +96,7 @@ const SAFETY_ACCESS_BLOCK_LEARN_MORE_URL: &str = "https://help.openai.com/en/art
pub(crate) fn new_safety_access_block_event() -> SafetyAccessBlockCell {
SafetyAccessBlockCell {
body: "We take extra caution with requests involving biological research and applications that could pose safety risks. Eligible researchers can apply for Trusted Access. If you’re a researcher at an approved organization, you may be able to apply for Trusted Access.",
body: "We take extra caution with requests involving biological research and applications that could pose safety risks. Eligible researchers can apply for Trusted Access.",
trusted_access_url: "https://www.openai.com/form/trusted-access-for-biology-research/",
}
}
codex-rs/tui/src/history_cell/snapshots/codex_tui__history_cell__tests__safety_access_block_event_snapshot.snap
+1 -2
View File
@@ -5,8 +5,7 @@ expression: rendered
ⓘ This content can't be shown
We take extra caution with requests involving biological research and
applications that could pose safety risks. Eligible researchers can apply
for Trusted Access. If you’re a researcher at an approved organization, you
may be able to apply for Trusted Access.
for Trusted Access.
Trusted Access: https://www.openai.com/form/trusted-access-for-biology-
research/
Learn more: https://help.openai.com/en/articles/20001326