openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-02 19:43:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge bb4f97d8e9 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2025-09-08 08:30:29 -07:00
committed by GitHub
parent 3a2f4350b6 bb4f97d8e9
commit c2857ea033
35 changed files with 525 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+61 -24
View File
@@ -547,9 +547,9 @@ checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
[[package]]
name = "clap"
version = "4.5.45"
version = "4.5.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fc0e74a703892159f5ae7d3aac52c8e6c392f5ae5f359c70b5881d60aaac318"
checksum = "7eac00902d9d136acd712710d71823fb8ac8004ca445a89e73a41d45aa712931"
dependencies = [
"clap_builder",
"clap_derive",
@@ -557,9 +557,9 @@ dependencies = [
[[package]]
name = "clap_builder"
version = "4.5.44"
version = "4.5.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b3e7f4214277f3c7aa526a59dd3fbe306a370daee1f8b7b8c987069cd8e888a8"
checksum = "2ad9bbf750e73b5884fb8a211a9424a1906c1e156724260fdae972f31d70e1d6"
dependencies = [
"anstream",
"anstyle",
@@ -579,9 +579,9 @@ dependencies = [
[[package]]
name = "clap_derive"
version = "4.5.45"
version = "4.5.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "14cb31bb0a7d536caef2639baa7fad459e15c3144efefa6dbd1c84562c4739f6"
checksum = "bbfd7eae0b0f1a6e63d4b13c9c478de77c2eb546fba158ad50b4203dc24b9f9c"
dependencies = [
"heck",
"proc-macro2",
@@ -1707,6 +1707,26 @@ version = "2.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
[[package]]
name = "fax"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f05de7d48f37cd6730705cbca900770cab77a89f413d23e100ad7fad7795a0ab"
dependencies = [
"fax_derive",
]
[[package]]
name = "fax_derive"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a0aca10fb742cb43f9e7bb8467c91aa9bcb8e3ffbc6a6f7389bb93ffc920577d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.104",
]
[[package]]
name = "fd-lock"
version = "4.0.4"
@@ -2361,9 +2381,9 @@ dependencies = [
[[package]]
name = "image"
version = "0.25.6"
version = "0.25.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "db35664ce6b9810857a38a906215e75a9c879f0696556a39f59c62829710251a"
checksum = "529feb3e6769d234375c4cf1ee2ce713682b8e76538cb13f9fc23e1400a591e7"
dependencies = [
"bytemuck",
"byteorder-lite",
@@ -2371,6 +2391,7 @@ dependencies = [
"exr",
"gif",
"image-webp",
"moxcms",
"num-traits",
"png",
"qoi",
@@ -2434,9 +2455,9 @@ checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd"
[[package]]
name = "insta"
version = "1.43.1"
version = "1.43.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "154934ea70c58054b556dd430b99a98c2a7ff5309ac9891597e339b5c28f4371"
checksum = "46fdb647ebde000f43b5b53f773c30cf9b0cb4300453208713fa38b2c70935a0"
dependencies = [
"console",
"once_cell",
@@ -2624,12 +2645,6 @@ dependencies = [
"libc",
]
[[package]]
name = "jpeg-decoder"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "00810f1d8b74be64b13dbf3db89ac67740615d6c891f0e7b6179326533011a07"
[[package]]
name = "js-sys"
version = "0.3.77"
@@ -2928,6 +2943,16 @@ dependencies = [
"windows-sys 0.59.0",
]
[[package]]
name = "moxcms"
version = "0.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ddd32fa8935aeadb8a8a6b6b351e40225570a37c43de67690383d87ef170cd08"
dependencies = [
"num-traits",
"pxfm",
]
[[package]]
name = "multimap"
version = "0.10.1"
@@ -3436,11 +3461,11 @@ dependencies = [
[[package]]
name = "png"
version = "0.17.16"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "82151a2fc869e011c153adc57cf2789ccb8d9906ce52c0b39a6b5697749d7526"
checksum = "97baced388464909d42d89643fe4361939af9b7ce7a31ee32a168f832a70f2a0"
dependencies = [
"bitflags 1.3.2",
"bitflags 2.9.1",
"crc32fast",
"fdeflate",
"flate2",
@@ -3609,6 +3634,15 @@ version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "007d8adb5ddab6f8e3f491ac63566a7d5002cc7ed73901f72057943fa71ae1ae"
[[package]]
name = "pxfm"
version = "0.1.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f55f4fedc84ed39cb7a489322318976425e42a147e2be79d8f878e2884f94e84"
dependencies = [
"num-traits",
]
[[package]]
name = "qoi"
version = "0.4.1"
@@ -4976,13 +5010,16 @@ dependencies = [
[[package]]
name = "tiff"
version = "0.9.1"
version = "0.10.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba1310fcea54c6a9a4fd1aad794ecc02c31682f6bfbecdf460bf19533eed1e3e"
checksum = "af9605de7fee8d9551863fd692cce7637f548dbd9db9180fcc07ccc6d26c336f"
dependencies = [
"fax",
"flate2",
"jpeg-decoder",
"half",
"quick-error",
"weezl",
"zune-jpeg",
]
[[package]]
@@ -5354,9 +5391,9 @@ dependencies = [
[[package]]
name = "tree-sitter"
version = "0.25.8"
version = "0.25.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d7b8994f367f16e6fa14b5aebbcb350de5d7cbea82dc5b00ae997dd71680dd2"
checksum = "ccd2a058a86cfece0bf96f7cce1021efef9c8ed0e892ab74639173e5ed7a34fa"
dependencies = [
"cc",
"regex",
codex-rs/apply-patch/Cargo.toml
+1 -1
View File
@@ -18,7 +18,7 @@ workspace = true
anyhow = "1"
similar = "2.7.0"
thiserror = "2.0.16"
tree-sitter = "0.25.8"
tree-sitter = "0.25.9"
tree-sitter-bash = "0.25.0"
once_cell = "1"
codex-rs/core/Cargo.toml
+1 -1
View File
@@ -53,7 +53,7 @@ tokio-util = "0.7.16"
toml = "0.9.5"
toml_edit = "0.23.4"
tracing = { version = "0.1.41", features = ["log"] }
tree-sitter = "0.25.8"
tree-sitter = "0.25.9"
tree-sitter-bash = "0.25.0"
uuid = { version = "1", features = ["serde", "v4"] }
whoami = "1.6.1"
codex-rs/core/src/client.rs
+16 -8
View File
@@ -6,6 +6,7 @@ use std::time::Duration;
use crate::AuthManager;
use bytes::Bytes;
use codex_protocol::mcp_protocol::AuthMode;
use codex_protocol::mcp_protocol::ConversationId;
use eventsource_stream::Eventsource;
use futures::prelude::*;
use regex_lite::Regex;
@@ -19,7 +20,6 @@ use tokio_util::io::ReaderStream;
use tracing::debug;
use tracing::trace;
use tracing::warn;
use uuid::Uuid;
use crate::chat_completions::AggregateStreamExt;
use crate::chat_completions::stream_chat_completions;
@@ -70,7 +70,7 @@ pub struct ModelClient {
auth_manager: Option<Arc<AuthManager>>,
client: reqwest::Client,
provider: ModelProviderInfo,
session_id: Uuid,
conversation_id: ConversationId,
effort: ReasoningEffortConfig,
summary: ReasoningSummaryConfig,
}
@@ -82,7 +82,7 @@ impl ModelClient {
provider: ModelProviderInfo,
effort: ReasoningEffortConfig,
summary: ReasoningSummaryConfig,
session_id: Uuid,
conversation_id: ConversationId,
) -> Self {
let client = create_client(&config.responses_originator_header);
@@ -91,7 +91,7 @@ impl ModelClient {
auth_manager,
client,
provider,
session_id,
conversation_id,
effort,
summary,
}
@@ -197,7 +197,7 @@ impl ModelClient {
store: false,
stream: true,
include,
prompt_cache_key: Some(self.session_id.to_string()),
prompt_cache_key: Some(self.conversation_id.to_string()),
text,
};
@@ -223,7 +223,9 @@ impl ModelClient {
req_builder = req_builder
.header("OpenAI-Beta", "responses=experimental")
.header("session_id", self.session_id.to_string())
// Send session_id for compatibility.
.header("conversation_id", self.conversation_id.to_string())
.header("session_id", self.conversation_id.to_string())
.header(reqwest::header::ACCEPT, "text/event-stream")
.json(&payload);
@@ -398,9 +400,15 @@ impl From<ResponseCompletedUsage> for TokenUsage {
fn from(val: ResponseCompletedUsage) -> Self {
TokenUsage {
input_tokens: val.input_tokens,
cached_input_tokens: val.input_tokens_details.map(|d| d.cached_tokens),
cached_input_tokens: val
.input_tokens_details
.map(|d| d.cached_tokens)
.unwrap_or(0),
output_tokens: val.output_tokens,
reasoning_output_tokens: val.output_tokens_details.map(|d| d.reasoning_tokens),
reasoning_output_tokens: val
.output_tokens_details
.map(|d| d.reasoning_tokens)
.unwrap_or(0),
total_tokens: val.total_tokens,
}
}
codex-rs/core/src/codex.rs
+50 -28
View File
@@ -15,6 +15,7 @@ use async_channel::Sender;
use codex_apply_patch::ApplyPatchAction;
use codex_apply_patch::MaybeApplyPatchVerified;
use codex_apply_patch::maybe_parse_apply_patch_verified;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::protocol::ConversationHistoryResponseEvent;
use codex_protocol::protocol::TaskStartedEvent;
use codex_protocol::protocol::TurnAbortReason;
@@ -99,6 +100,7 @@ use crate::protocol::SessionConfiguredEvent;
use crate::protocol::StreamErrorEvent;
use crate::protocol::Submission;
use crate::protocol::TaskCompleteEvent;
use crate::protocol::TokenUsageInfo;
use crate::protocol::TurnDiffEvent;
use crate::protocol::WebSearchBeginEvent;
use crate::rollout::RolloutRecorder;
@@ -148,7 +150,7 @@ pub struct Codex {
/// unique session id.
pub struct CodexSpawnOk {
pub codex: Codex,
pub session_id: Uuid,
pub conversation_id: ConversationId,
}
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
@@ -204,7 +206,7 @@ impl Codex {
session
.record_initial_history(&turn_context, conversation_history)
.await;
let session_id = session.session_id;
let conversation_id = session.conversation_id;
// This task will run until Op::Shutdown is received.
tokio::spawn(submission_loop(
@@ -219,7 +221,10 @@ impl Codex {
rx_event,
};
Ok(CodexSpawnOk { codex, session_id })
Ok(CodexSpawnOk {
codex,
conversation_id,
})
}
/// Submit the `op` wrapped in a `Submission` with a unique ID.
@@ -261,13 +266,14 @@ struct State {
pending_approvals: HashMap<String, oneshot::Sender<ReviewDecision>>,
pending_input: Vec<ResponseInputItem>,
history: ConversationHistory,
token_info: Option<TokenUsageInfo>,
}
/// Context for an initialized model agent
///
/// A session has at most 1 running task at a time, and can be interrupted by user input.
pub(crate) struct Session {
session_id: Uuid,
conversation_id: ConversationId,
tx_event: Sender<Event>,
/// Manager for external MCP servers/tools.
@@ -356,7 +362,7 @@ impl Session {
tx_event: Sender<Event>,
initial_history: InitialHistory,
) -> anyhow::Result<(Arc<Self>, TurnContext)> {
let session_id = Uuid::new_v4();
let conversation_id = ConversationId::from(Uuid::new_v4());
let ConfigureSession {
provider,
model,
@@ -383,7 +389,7 @@ impl Session {
// - spin up MCP connection manager
// - perform default shell discovery
// - load history metadata
let rollout_fut = RolloutRecorder::new(&config, session_id, user_instructions.clone());
let rollout_fut = RolloutRecorder::new(&config, conversation_id, user_instructions.clone());
let mcp_fut = McpConnectionManager::new(config.mcp_servers.clone());
let default_shell_fut = shell::default_user_shell();
@@ -429,7 +435,7 @@ impl Session {
}
}
// Now that `session_id` is final (may have been updated by resume),
// Now that the conversation id is final (may have been updated by resume),
// construct the model client.
let client = ModelClient::new(
config.clone(),
@@ -437,7 +443,7 @@ impl Session {
provider.clone(),
model_reasoning_effort,
model_reasoning_summary,
session_id,
conversation_id,
);
let turn_context = TurnContext {
client,
@@ -459,7 +465,7 @@ impl Session {
cwd,
};
let sess = Arc::new(Session {
session_id,
conversation_id,
tx_event: tx_event.clone(),
mcp_connection_manager,
session_manager: ExecSessionManager::default(),
@@ -481,7 +487,7 @@ impl Session {
let events = std::iter::once(Event {
id: INITIAL_SUBMIT_ID.to_owned(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id,
session_id: conversation_id,
model,
history_log_id,
history_entry_count,
@@ -1082,7 +1088,7 @@ async fn submission_loop(
provider,
effective_effort,
effective_summary,
sess.session_id,
sess.conversation_id,
);
let new_approval_policy = approval_policy.unwrap_or(prev.approval_policy);
@@ -1170,7 +1176,7 @@ async fn submission_loop(
provider,
effort,
summary,
sess.session_id,
sess.conversation_id,
);
let fresh_turn_context = TurnContext {
@@ -1213,7 +1219,7 @@ async fn submission_loop(
other => sess.notify_approval(&id, other),
},
Op::AddToHistory { text } => {
let id = sess.session_id;
let id = sess.conversation_id;
let config = config.clone();
tokio::spawn(async move {
if let Err(e) = crate::message_history::append_entry(&text, &id, &config).await
@@ -1244,7 +1250,7 @@ async fn submission_loop(
log_id,
entry: entry_opt.map(|e| {
codex_protocol::message_history::HistoryEntry {
session_id: e.session_id,
conversation_id: e.session_id,
ts: e.ts,
text: e.text,
}
@@ -1350,7 +1356,7 @@ async fn submission_loop(
let event = Event {
id: sub_id.clone(),
msg: EventMsg::ConversationHistory(ConversationHistoryResponseEvent {
conversation_id: sess.session_id,
conversation_id: sess.conversation_id,
entries: sess.state.lock_unchecked().history.contents(),
}),
};
@@ -1767,15 +1773,23 @@ async fn try_run_turn(
response_id: _,
token_usage,
} => {
if let Some(token_usage) = token_usage {
sess.tx_event
.send(Event {
id: sub_id.to_string(),
msg: EventMsg::TokenCount(token_usage),
})
.await
.ok();
}
let info = {
let mut st = sess.state.lock_unchecked();
let info = TokenUsageInfo::new_or_append(
&st.token_info,
&token_usage,
turn_context.client.get_model_context_window(),
);
st.token_info = info.clone();
info
};
sess.tx_event
.send(Event {
id: sub_id.to_string(),
msg: EventMsg::TokenCount(crate::protocol::TokenCountEvent { info }),
})
.await
.ok();
let unified_diff = turn_diff_tracker.get_unified_diff();
if let Ok(Some(unified_diff)) = unified_diff {
@@ -2841,13 +2855,21 @@ async fn drain_to_completed(
response_id: _,
token_usage,
}) => {
// some providers don't return token usage, so we default
// TODO: consider approximate token usage
let token_usage = token_usage.unwrap_or_default();
let info = {
let mut st = sess.state.lock_unchecked();
let info = TokenUsageInfo::new_or_append(
&st.token_info,
&token_usage,
turn_context.client.get_model_context_window(),
);
st.token_info = info.clone();
info
};
sess.tx_event
.send(Event {
id: sub_id.to_string(),
msg: EventMsg::TokenCount(token_usage),
msg: EventMsg::TokenCount(crate::protocol::TokenCountEvent { info }),
})
.await
.ok();
codex-rs/core/src/config_types.rs
+4
View File
@@ -18,6 +18,10 @@ pub struct McpServerConfig {
#[serde(default)]
pub env: Option<HashMap<String, String>>,
/// Startup timeout in milliseconds for initializing MCP server & initially listing tools.
#[serde(default)]
pub startup_timeout_ms: Option<u64>,
}
#[derive(Deserialize, Debug, Copy, Clone, PartialEq)]
codex-rs/core/src/conversation_manager.rs
+10 -10
View File
@@ -4,8 +4,8 @@ use std::sync::Arc;
use crate::AuthManager;
use crate::CodexAuth;
use codex_protocol::mcp_protocol::ConversationId;
use tokio::sync::RwLock;
use uuid::Uuid;
use crate::codex::Codex;
use crate::codex::CodexSpawnOk;
@@ -29,7 +29,7 @@ pub enum InitialHistory {
/// Represents a newly created Codex conversation, including the first event
/// (which is [`EventMsg::SessionConfigured`]).
pub struct NewConversation {
pub conversation_id: Uuid,
pub conversation_id: ConversationId,
pub conversation: Arc<CodexConversation>,
pub session_configured: SessionConfiguredEvent,
}
@@ -37,7 +37,7 @@ pub struct NewConversation {
/// [`ConversationManager`] is responsible for creating conversations and
/// maintaining them in memory.
pub struct ConversationManager {
conversations: Arc<RwLock<HashMap<Uuid, Arc<CodexConversation>>>>,
conversations: Arc<RwLock<HashMap<ConversationId, Arc<CodexConversation>>>>,
auth_manager: Arc<AuthManager>,
}
@@ -70,13 +70,13 @@ impl ConversationManager {
let initial_history = RolloutRecorder::get_rollout_history(resume_path).await?;
let CodexSpawnOk {
codex,
session_id: conversation_id,
conversation_id,
} = Codex::spawn(config, auth_manager, initial_history).await?;
self.finalize_spawn(codex, conversation_id).await
} else {
let CodexSpawnOk {
codex,
session_id: conversation_id,
conversation_id,
} = { Codex::spawn(config, auth_manager, InitialHistory::New).await? };
self.finalize_spawn(codex, conversation_id).await
}
@@ -85,7 +85,7 @@ impl ConversationManager {
async fn finalize_spawn(
&self,
codex: Codex,
conversation_id: Uuid,
conversation_id: ConversationId,
) -> CodexResult<NewConversation> {
// The first event must be `SessionInitialized`. Validate and forward it
// to the caller so that they can display it in the conversation
@@ -116,7 +116,7 @@ impl ConversationManager {
pub async fn get_conversation(
&self,
conversation_id: Uuid,
conversation_id: ConversationId,
) -> CodexResult<Arc<CodexConversation>> {
let conversations = self.conversations.read().await;
conversations
@@ -134,12 +134,12 @@ impl ConversationManager {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
let CodexSpawnOk {
codex,
session_id: conversation_id,
conversation_id,
} = Codex::spawn(config, auth_manager, initial_history).await?;
self.finalize_spawn(codex, conversation_id).await
}
pub async fn remove_conversation(&self, conversation_id: Uuid) {
pub async fn remove_conversation(&self, conversation_id: ConversationId) {
self.conversations.write().await.remove(&conversation_id);
}
@@ -161,7 +161,7 @@ impl ConversationManager {
let auth_manager = self.auth_manager.clone();
let CodexSpawnOk {
codex,
session_id: conversation_id,
conversation_id,
} = Codex::spawn(config, auth_manager, history).await?;
self.finalize_spawn(codex, conversation_id).await
codex-rs/core/src/error.rs
+2 -2
View File
@@ -1,10 +1,10 @@
use codex_protocol::mcp_protocol::ConversationId;
use reqwest::StatusCode;
use serde_json;
use std::io;
use std::time::Duration;
use thiserror::Error;
use tokio::task::JoinError;
use uuid::Uuid;
pub type Result<T> = std::result::Result<T, CodexErr>;
@@ -49,7 +49,7 @@ pub enum CodexErr {
Stream(String, Option<Duration>),
#[error("no conversation with id: {0}")]
ConversationNotFound(Uuid),
ConversationNotFound(ConversationId),
#[error("session configured event was not the first event in the stream")]
SessionConfiguredNotFirstEvent,
codex-rs/core/src/mcp_connection_manager.rs
+38 -19
View File
@@ -9,6 +9,7 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::ffi::OsString;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Context;
@@ -36,8 +37,8 @@ use crate::config_types::McpServerConfig;
const MCP_TOOL_NAME_DELIMITER: &str = "__";
const MAX_TOOL_NAME_LENGTH: usize = 64;
/// Timeout for the `tools/list` request.
const LIST_TOOLS_TIMEOUT: Duration = Duration::from_secs(10);
/// Default timeout for initializing MCP server & initially listing tools.
const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(10);
/// Map that holds a startup error for every MCP server that could **not** be
/// spawned successfully.
@@ -81,6 +82,11 @@ struct ToolInfo {
tool: Tool,
}
struct ManagedClient {
client: Arc<McpClient>,
startup_timeout: Duration,
}
/// A thin wrapper around a set of running [`McpClient`] instances.
#[derive(Default)]
pub(crate) struct McpConnectionManager {
@@ -88,7 +94,7 @@ pub(crate) struct McpConnectionManager {
///
/// The server name originates from the keys of the `mcp_servers` map in
/// the user configuration.
clients: HashMap<String, std::sync::Arc<McpClient>>,
clients: HashMap<String, ManagedClient>,
/// Fully qualified tool name -> tool instance.
tools: HashMap<String, ToolInfo>,
@@ -126,8 +132,15 @@ impl McpConnectionManager {
continue;
}
let startup_timeout = cfg
.startup_timeout_ms
.map(Duration::from_millis)
.unwrap_or(DEFAULT_STARTUP_TIMEOUT);
join_set.spawn(async move {
let McpServerConfig { command, args, env } = cfg;
let McpServerConfig {
command, args, env, ..
} = cfg;
let client_res = McpClient::new_stdio_client(
command.into(),
args.into_iter().map(OsString::from).collect(),
@@ -154,12 +167,15 @@ impl McpConnectionManager {
protocol_version: mcp_types::MCP_SCHEMA_VERSION.to_owned(),
};
let initialize_notification_params = None;
let timeout = Some(Duration::from_secs(10));
match client
.initialize(params, initialize_notification_params, timeout)
.initialize(
params,
initialize_notification_params,
Some(startup_timeout),
)
.await
{
Ok(_response) => (server_name, Ok(client)),
Ok(_response) => (server_name, Ok((client, startup_timeout))),
Err(e) => (server_name, Err(e)),
}
}
@@ -168,15 +184,20 @@ impl McpConnectionManager {
});
}
let mut clients: HashMap<String, std::sync::Arc<McpClient>> =
HashMap::with_capacity(join_set.len());
let mut clients: HashMap<String, ManagedClient> = HashMap::with_capacity(join_set.len());
while let Some(res) = join_set.join_next().await {
let (server_name, client_res) = res?; // JoinError propagation
match client_res {
Ok(client) => {
clients.insert(server_name, std::sync::Arc::new(client));
Ok((client, startup_timeout)) => {
clients.insert(
server_name,
ManagedClient {
client: Arc::new(client),
startup_timeout,
},
);
}
Err(e) => {
errors.insert(server_name, e);
@@ -212,6 +233,7 @@ impl McpConnectionManager {
.clients
.get(server)
.ok_or_else(|| anyhow!("unknown MCP server '{server}'"))?
.client
.clone();
client
@@ -229,21 +251,18 @@ impl McpConnectionManager {
/// Query every server for its available tools and return a single map that
/// contains **all** tools. Each key is the fully-qualified name for the tool.
async fn list_all_tools(
clients: &HashMap<String, std::sync::Arc<McpClient>>,
) -> Result<Vec<ToolInfo>> {
async fn list_all_tools(clients: &HashMap<String, ManagedClient>) -> Result<Vec<ToolInfo>> {
let mut join_set = JoinSet::new();
// Spawn one task per server so we can query them concurrently. This
// keeps the overall latency roughly at the slowest server instead of
// the cumulative latency.
for (server_name, client) in clients {
for (server_name, managed_client) in clients {
let server_name_cloned = server_name.clone();
let client_clone = client.clone();
let client_clone = managed_client.client.clone();
let startup_timeout = managed_client.startup_timeout;
join_set.spawn(async move {
let res = client_clone
.list_tools(None, Some(LIST_TOOLS_TIMEOUT))
.await;
let res = client_clone.list_tools(None, Some(startup_timeout)).await;
(server_name_cloned, res)
});
}
codex-rs/core/src/message_history.rs
+10 -5
View File
@@ -5,7 +5,7 @@
//! JSON-Lines tooling. Each record has the following schema:
//!
//! ````text
//! {"session_id":"<uuid>","ts":<unix_seconds>,"text":"<message>"}
//! {"conversation_id":"<uuid>","ts":<unix_seconds>,"text":"<message>"}
//! ````
//!
//! To minimise the chance of interleaved writes when multiple processes are
@@ -22,10 +22,11 @@ use std::path::PathBuf;
use serde::Deserialize;
use serde::Serialize;
use codex_protocol::mcp_protocol::ConversationId;
use std::time::Duration;
use tokio::fs;
use tokio::io::AsyncReadExt;
use uuid::Uuid;
use crate::config::Config;
use crate::config_types::HistoryPersistence;
@@ -54,10 +55,14 @@ fn history_filepath(config: &Config) -> PathBuf {
path
}
/// Append a `text` entry associated with `session_id` to the history file. Uses
/// Append a `text` entry associated with `conversation_id` to the history file. Uses
/// advisory file locking to ensure that concurrent writes do not interleave,
/// which entails a small amount of blocking I/O internally.
pub(crate) async fn append_entry(text: &str, session_id: &Uuid, config: &Config) -> Result<()> {
pub(crate) async fn append_entry(
text: &str,
conversation_id: &ConversationId,
config: &Config,
) -> Result<()> {
match config.history.persistence {
HistoryPersistence::SaveAll => {
// Save everything: proceed.
@@ -84,7 +89,7 @@ pub(crate) async fn append_entry(text: &str, session_id: &Uuid, config: &Config)
// Construct the JSON line first so we can write it in a single syscall.
let entry = HistoryEntry {
session_id: session_id.to_string(),
session_id: conversation_id.to_string(),
ts,
text: text.to_string(),
};
codex-rs/core/src/rollout/recorder.rs
+13 -10
View File
@@ -5,6 +5,7 @@ use std::fs::{self};
use std::io::Error as IoError;
use std::path::Path;
use codex_protocol::mcp_protocol::ConversationId;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
@@ -17,7 +18,6 @@ use tokio::sync::mpsc::{self};
use tokio::sync::oneshot;
use tracing::info;
use tracing::warn;
use uuid::Uuid;
use super::SESSIONS_SUBDIR;
use super::list::ConversationsPage;
@@ -32,7 +32,7 @@ use codex_protocol::models::ResponseItem;
#[derive(Serialize, Deserialize, Clone, Default)]
pub struct SessionMeta {
pub id: Uuid,
pub id: ConversationId,
pub timestamp: String,
pub instructions: Option<String>,
}
@@ -55,7 +55,7 @@ pub struct SavedSession {
pub items: Vec<ResponseItem>,
#[serde(default)]
pub state: SessionStateSnapshot,
pub session_id: Uuid,
pub session_id: ConversationId,
}
/// Records all [`ResponseItem`]s for a session and flushes them to disk after
@@ -94,14 +94,14 @@ impl RolloutRecorder {
/// error so the caller can decide whether to disable persistence.
pub async fn new(
config: &Config,
uuid: Uuid,
conversation_id: ConversationId,
instructions: Option<String>,
) -> std::io::Result<Self> {
let LogFileInfo {
file,
session_id,
conversation_id: session_id,
timestamp,
} = create_log_file(config, uuid)?;
} = create_log_file(config, conversation_id)?;
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
@@ -227,13 +227,16 @@ struct LogFileInfo {
file: File,
/// Session ID (also embedded in filename).
session_id: Uuid,
conversation_id: ConversationId,
/// Timestamp for the start of the session.
timestamp: OffsetDateTime,
}
fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFileInfo> {
fn create_log_file(
config: &Config,
conversation_id: ConversationId,
) -> std::io::Result<LogFileInfo> {
// Resolve ~/.codex/sessions/YYYY/MM/DD and create it if missing.
let timestamp = OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
@@ -252,7 +255,7 @@ fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFile
.format(format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let filename = format!("rollout-{date_str}-{session_id}.jsonl");
let filename = format!("rollout-{date_str}-{conversation_id}.jsonl");
let path = dir.join(filename);
let file = std::fs::OpenOptions::new()
@@ -262,7 +265,7 @@ fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFile
Ok(LogFileInfo {
file,
session_id,
conversation_id,
timestamp,
})
}
codex-rs/core/tests/chat_completions_payload.rs
+2 -2
View File
@@ -11,11 +11,11 @@ use codex_core::ReasoningItemContent;
use codex_core::ResponseItem;
use codex_core::WireApi;
use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_protocol::mcp_protocol::ConversationId;
use core_test_support::load_default_config_for_test;
use futures::StreamExt;
use serde_json::Value;
use tempfile::TempDir;
use uuid::Uuid;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
@@ -76,7 +76,7 @@ async fn run_request(input: Vec<ResponseItem>) -> Value {
provider,
effort,
summary,
Uuid::new_v4(),
ConversationId::new(),
);
let mut prompt = Prompt::default();
codex-rs/core/tests/chat_completions_sse.rs
+2 -2
View File
@@ -8,10 +8,10 @@ use codex_core::ResponseEvent;
use codex_core::ResponseItem;
use codex_core::WireApi;
use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_protocol::mcp_protocol::ConversationId;
use core_test_support::load_default_config_for_test;
use futures::StreamExt;
use tempfile::TempDir;
use uuid::Uuid;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
@@ -69,7 +69,7 @@ async fn run_stream(sse_body: &str) -> Vec<ResponseEvent> {
provider,
effort,
summary,
Uuid::new_v4(),
ConversationId::new(),
);
let mut prompt = Prompt::default();
codex-rs/core/tests/suite/client.rs
+5 -5
View File
@@ -242,7 +242,7 @@ async fn resume_includes_initial_messages_and_sends_prior_items() {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_session_id_and_model_headers_in_request() {
async fn includes_conversation_id_and_model_headers_in_request() {
if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
@@ -299,12 +299,12 @@ async fn includes_session_id_and_model_headers_in_request() {
// get request from the server
let request = &server.received_requests().await.unwrap()[0];
let request_session_id = request.headers.get("session_id").unwrap();
let request_conversation_id = request.headers.get("conversation_id").unwrap();
let request_authorization = request.headers.get("authorization").unwrap();
let request_originator = request.headers.get("originator").unwrap();
assert_eq!(
request_session_id.to_str().unwrap(),
request_conversation_id.to_str().unwrap(),
conversation_id.to_string()
);
assert_eq!(request_originator.to_str().unwrap(), "codex_cli_rs");
@@ -477,14 +477,14 @@ async fn chatgpt_auth_sends_correct_request() {
// get request from the server
let request = &server.received_requests().await.unwrap()[0];
let request_session_id = request.headers.get("session_id").unwrap();
let request_conversation_id = request.headers.get("conversation_id").unwrap();
let request_authorization = request.headers.get("authorization").unwrap();
let request_originator = request.headers.get("originator").unwrap();
let request_chatgpt_account_id = request.headers.get("chatgpt-account-id").unwrap();
let request_body = request.body_json::<serde_json::Value>().unwrap();
assert_eq!(
request_session_id.to_str().unwrap(),
request_conversation_id.to_str().unwrap(),
conversation_id.to_string()
);
assert_eq!(request_originator.to_str().unwrap(), "codex_cli_rs");
codex-rs/exec/src/event_processor_with_human_output.rs
+10 -4
View File
@@ -189,8 +189,14 @@ impl EventProcessor for EventProcessorWithHumanOutput {
}
return CodexStatus::InitiateShutdown;
}
EventMsg::TokenCount(token_usage) => {
ts_println!(self, "tokens used: {}", token_usage.blended_total());
EventMsg::TokenCount(ev) => {
if let Some(usage_info) = ev.info {
ts_println!(
self,
"tokens used: {}",
usage_info.total_token_usage.blended_total()
);
}
}
EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
if !self.answer_started {
@@ -511,7 +517,7 @@ impl EventProcessor for EventProcessorWithHumanOutput {
}
EventMsg::SessionConfigured(session_configured_event) => {
let SessionConfiguredEvent {
session_id,
session_id: conversation_id,
model,
history_log_id: _,
history_entry_count: _,
@@ -522,7 +528,7 @@ impl EventProcessor for EventProcessorWithHumanOutput {
self,
"{} {}",
"codex session".style(self.magenta).style(self.bold),
session_id.to_string().style(self.dimmed)
conversation_id.to_string().style(self.dimmed)
);
ts_println!(self, "model: {}", model);
codex-rs/mcp-server/src/codex_message_processor.rs
+11 -11
View File
@@ -99,7 +99,7 @@ pub(crate) struct CodexMessageProcessor {
conversation_listeners: HashMap<Uuid, oneshot::Sender<()>>,
active_login: Arc<Mutex<Option<ActiveLogin>>>,
// Queue of pending interrupt requests per conversation. We reply when TurnAborted arrives.
pending_interrupts: Arc<Mutex<HashMap<Uuid, Vec<RequestId>>>>,
pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<RequestId>>>>,
}
impl CodexMessageProcessor {
@@ -511,7 +511,7 @@ impl CodexMessageProcessor {
..
} = conversation_id;
let response = NewConversationResponse {
conversation_id: ConversationId(conversation_id),
conversation_id,
model: session_configured.model,
};
self.outgoing.send_response(request_id, response).await;
@@ -632,7 +632,7 @@ impl CodexMessageProcessor {
// Reply with conversation id + model and initial messages (when present)
let response = codex_protocol::mcp_protocol::ResumeConversationResponse {
conversation_id: ConversationId(conversation_id),
conversation_id,
model: session_configured.model.clone(),
initial_messages: session_configured.initial_messages.clone(),
};
@@ -656,7 +656,7 @@ impl CodexMessageProcessor {
} = params;
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.get_conversation(conversation_id)
.await
else {
let error = JSONRPCErrorError {
@@ -704,7 +704,7 @@ impl CodexMessageProcessor {
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.get_conversation(conversation_id)
.await
else {
let error = JSONRPCErrorError {
@@ -750,7 +750,7 @@ impl CodexMessageProcessor {
let InterruptConversationParams { conversation_id } = params;
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.get_conversation(conversation_id)
.await
else {
let error = JSONRPCErrorError {
@@ -765,7 +765,7 @@ impl CodexMessageProcessor {
// Record the pending interrupt so we can reply when TurnAborted arrives.
{
let mut map = self.pending_interrupts.lock().await;
map.entry(conversation_id.0).or_default().push(request_id);
map.entry(conversation_id).or_default().push(request_id);
}
// Submit the interrupt; we'll respond upon TurnAborted.
@@ -780,12 +780,12 @@ impl CodexMessageProcessor {
let AddConversationListenerParams { conversation_id } = params;
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.get_conversation(conversation_id)
.await
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("conversation not found: {}", conversation_id.0),
message: format!("conversation not found: {conversation_id}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
@@ -898,7 +898,7 @@ async fn apply_bespoke_event_handling(
conversation_id: ConversationId,
conversation: Arc<CodexConversation>,
outgoing: Arc<OutgoingMessageSender>,
pending_interrupts: Arc<Mutex<HashMap<Uuid, Vec<RequestId>>>>,
pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<RequestId>>>>,
) {
let Event { id: event_id, msg } = event;
match msg {
@@ -951,7 +951,7 @@ async fn apply_bespoke_event_handling(
EventMsg::TurnAborted(turn_aborted_event) => {
let pending = {
let mut map = pending_interrupts.lock().await;
map.remove(&conversation_id.0).unwrap_or_default()
map.remove(&conversation_id).unwrap_or_default()
};
if !pending.is_empty() {
let response = InterruptConversationResponse {
codex-rs/mcp-server/src/codex_tool_config.rs
+10 -9
View File
@@ -181,8 +181,8 @@ impl CodexToolCallParam {
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct CodexToolCallReplyParam {
/// The *session id* for this conversation.
pub session_id: String,
/// The conversation id for this Codex session.
pub conversation_id: String,
/// The *next user prompt* to continue the Codex conversation.
pub prompt: String,
@@ -213,7 +213,8 @@ pub(crate) fn create_tool_for_codex_tool_call_reply_param() -> Tool {
input_schema: tool_input_schema,
output_schema: None,
description: Some(
"Continue a Codex session by providing the session id and prompt.".to_string(),
"Continue a Codex conversation by providing the conversation id and prompt."
.to_string(),
),
annotations: None,
}
@@ -308,21 +309,21 @@ mod tests {
let tool = create_tool_for_codex_tool_call_reply_param();
let tool_json = serde_json::to_value(&tool).expect("tool serializes");
let expected_tool_json = serde_json::json!({
"description": "Continue a Codex session by providing the session id and prompt.",
"description": "Continue a Codex conversation by providing the conversation id and prompt.",
"inputSchema": {
"properties": {
"conversationId": {
"description": "The conversation id for this Codex session.",
"type": "string"
},
"prompt": {
"description": "The *next user prompt* to continue the Codex conversation.",
"type": "string"
},
"sessionId": {
"description": "The *session id* for this conversation.",
"type": "string"
},
},
"required": [
"conversationId",
"prompt",
"sessionId",
],
"type": "object",
},
codex-rs/mcp-server/src/codex_tool_runner.rs
+6 -6
View File
@@ -18,13 +18,13 @@ use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::Submission;
use codex_core::protocol::TaskCompleteEvent;
use codex_protocol::mcp_protocol::ConversationId;
use mcp_types::CallToolResult;
use mcp_types::ContentBlock;
use mcp_types::RequestId;
use mcp_types::TextContent;
use serde_json::json;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::exec_approval::handle_exec_approval_request;
use crate::outgoing_message::OutgoingMessageSender;
@@ -43,7 +43,7 @@ pub async fn run_codex_tool_session(
config: CodexConfig,
outgoing: Arc<OutgoingMessageSender>,
conversation_manager: Arc<ConversationManager>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ConversationId>>>,
) {
let NewConversation {
conversation_id,
@@ -119,13 +119,13 @@ pub async fn run_codex_tool_session_reply(
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
prompt: String,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
session_id: Uuid,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ConversationId>>>,
conversation_id: ConversationId,
) {
running_requests_id_to_codex_uuid
.lock()
.await
.insert(request_id.clone(), session_id);
.insert(request_id.clone(), conversation_id);
if let Err(e) = conversation
.submit(Op::UserInput {
items: vec![InputItem::Text { text: prompt }],
@@ -154,7 +154,7 @@ async fn run_codex_tool_session_inner(
codex: Arc<CodexConversation>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ConversationId>>>,
) {
let request_id_str = match &request_id {
RequestId::String(s) => s.clone(),
codex-rs/mcp-server/src/message_processor.rs
+30 -18
View File
@@ -9,6 +9,7 @@ use crate::codex_tool_config::create_tool_for_codex_tool_call_reply_param;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::outgoing_message::OutgoingMessageSender;
use codex_protocol::mcp_protocol::ClientRequest;
use codex_protocol::mcp_protocol::ConversationId;
use codex_core::AuthManager;
use codex_core::ConversationManager;
@@ -41,7 +42,7 @@ pub(crate) struct MessageProcessor {
initialized: bool,
codex_linux_sandbox_exe: Option<PathBuf>,
conversation_manager: Arc<ConversationManager>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, ConversationId>>>,
}
impl MessageProcessor {
@@ -436,7 +437,10 @@ impl MessageProcessor {
tracing::info!("tools/call -> params: {:?}", arguments);
// parse arguments
let CodexToolCallReplyParam { session_id, prompt } = match arguments {
let CodexToolCallReplyParam {
conversation_id,
prompt,
} = match arguments {
Some(json_val) => match serde_json::from_value::<CodexToolCallReplyParam>(json_val) {
Ok(params) => params,
Err(e) => {
@@ -457,12 +461,12 @@ impl MessageProcessor {
},
None => {
tracing::error!(
"Missing arguments for codex-reply tool-call; the `session_id` and `prompt` fields are required."
"Missing arguments for codex-reply tool-call; the `conversation_id` and `prompt` fields are required."
);
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_owned(),
text: "Missing arguments for codex-reply tool-call; the `session_id` and `prompt` fields are required.".to_owned(),
text: "Missing arguments for codex-reply tool-call; the `conversation_id` and `prompt` fields are required.".to_owned(),
annotations: None,
})],
is_error: Some(true),
@@ -473,14 +477,14 @@ impl MessageProcessor {
return;
}
};
let session_id = match Uuid::parse_str(&session_id) {
Ok(id) => id,
let conversation_id = match Uuid::parse_str(&conversation_id) {
Ok(id) => ConversationId::from(id),
Err(e) => {
tracing::error!("Failed to parse session_id: {e}");
tracing::error!("Failed to parse conversation_id: {e}");
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_owned(),
text: format!("Failed to parse session_id: {e}"),
text: format!("Failed to parse conversation_id: {e}"),
annotations: None,
})],
is_error: Some(true),
@@ -496,14 +500,18 @@ impl MessageProcessor {
let outgoing = self.outgoing.clone();
let running_requests_id_to_codex_uuid = self.running_requests_id_to_codex_uuid.clone();
let codex = match self.conversation_manager.get_conversation(session_id).await {
let codex = match self
.conversation_manager
.get_conversation(conversation_id)
.await
{
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for session_id: {session_id}");
tracing::warn!("Session not found for conversation_id: {conversation_id}");
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_owned(),
text: format!("Session not found for session_id: {session_id}"),
text: format!("Session not found for conversation_id: {conversation_id}"),
annotations: None,
})],
is_error: Some(true),
@@ -528,7 +536,7 @@ impl MessageProcessor {
request_id,
prompt,
running_requests_id_to_codex_uuid,
session_id,
conversation_id,
)
.await;
}
@@ -564,24 +572,28 @@ impl MessageProcessor {
RequestId::Integer(i) => i.to_string(),
};
// Obtain the session_id while holding the first lock, then release.
let session_id = {
// Obtain the conversation id while holding the first lock, then release.
let conversation_id = {
let map_guard = self.running_requests_id_to_codex_uuid.lock().await;
match map_guard.get(&request_id) {
Some(id) => *id, // Uuid is Copy
Some(id) => *id,
None => {
tracing::warn!("Session not found for request_id: {}", request_id_string);
return;
}
}
};
tracing::info!("session_id: {session_id}");
tracing::info!("conversation_id: {conversation_id}");
// Obtain the Codex conversation from the server.
let codex_arc = match self.conversation_manager.get_conversation(session_id).await {
let codex_arc = match self
.conversation_manager
.get_conversation(conversation_id)
.await
{
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for session_id: {session_id}");
tracing::warn!("Session not found for conversation_id: {conversation_id}");
return;
}
};
codex-rs/mcp-server/src/outgoing_message.rs
+5 -2
View File
@@ -258,6 +258,7 @@ pub(crate) struct OutgoingError {
mod tests {
use codex_core::protocol::EventMsg;
use codex_core::protocol::SessionConfiguredEvent;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::mcp_protocol::LoginChatGptCompleteNotification;
use pretty_assertions::assert_eq;
use serde_json::json;
@@ -270,10 +271,11 @@ mod tests {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let conversation_id = ConversationId::new();
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: Uuid::new_v4(),
session_id: conversation_id,
model: "gpt-4o".to_string(),
history_log_id: 1,
history_entry_count: 1000,
@@ -302,8 +304,9 @@ mod tests {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let conversation_id = ConversationId::new();
let session_configured_event = SessionConfiguredEvent {
session_id: Uuid::new_v4(),
session_id: conversation_id,
model: "gpt-4o".to_string(),
history_log_id: 1,
history_entry_count: 1000,
codex-rs/mcp-server/tests/suite/list_resume.rs
+1 -1
View File
@@ -142,7 +142,7 @@ async fn test_list_and_resume_conversations() {
} = to_response::<ResumeConversationResponse>(resume_resp)
.expect("deserialize resumeConversation response");
// conversation id should be a valid UUID
let _ = uuid::Uuid::from_bytes(conversation_id.0.into_bytes());
let _: uuid::Uuid = conversation_id.into();
}
fn create_fake_rollout(codex_home: &Path, filename_ts: &str, meta_rfc3339: &str, preview: &str) {
codex-rs/protocol/src/mcp_protocol.rs
+19 -1
View File
@@ -19,16 +19,34 @@ use strum_macros::Display;
use ts_rs::TS;
use uuid::Uuid;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS, Hash, Default)]
#[ts(type = "string")]
pub struct ConversationId(pub Uuid);
impl ConversationId {
pub fn new() -> Self {
Self(Uuid::new_v4())
}
}
impl Display for ConversationId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
impl From<Uuid> for ConversationId {
fn from(value: Uuid) -> Self {
Self(value)
}
}
impl From<ConversationId> for Uuid {
fn from(value: ConversationId) -> Self {
value.0
}
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, TS)]
#[ts(type = "string")]
pub struct GitSha(pub String);
codex-rs/protocol/src/message_history.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use serde::Serialize;
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct HistoryEntry {
pub session_id: String,
pub conversation_id: String,
pub ts: u64,
pub text: String,
}
codex-rs/protocol/src/protocol.rs
+75 -24
View File
@@ -10,7 +10,14 @@ use std::path::PathBuf;
use std::str::FromStr;
use std::time::Duration;
use crate::config_types::ReasoningEffort as ReasoningEffortConfig;
use crate::config_types::ReasoningSummary as ReasoningSummaryConfig;
use crate::custom_prompts::CustomPrompt;
use crate::mcp_protocol::ConversationId;
use crate::message_history::HistoryEntry;
use crate::models::ResponseItem;
use crate::parse_command::ParsedCommand;
use crate::plan_tool::UpdatePlanArgs;
use mcp_types::CallToolResult;
use mcp_types::Tool as McpTool;
use serde::Deserialize;
@@ -18,14 +25,6 @@ use serde::Serialize;
use serde_with::serde_as;
use strum_macros::Display;
use ts_rs::TS;
use uuid::Uuid;
use crate::config_types::ReasoningEffort as ReasoningEffortConfig;
use crate::config_types::ReasoningSummary as ReasoningSummaryConfig;
use crate::message_history::HistoryEntry;
use crate::models::ResponseItem;
use crate::parse_command::ParsedCommand;
use crate::plan_tool::UpdatePlanArgs;
/// Open/close tags for special user-input blocks. Used across crates to avoid
/// duplicated hardcoded strings.
@@ -417,9 +416,9 @@ pub enum EventMsg {
/// Agent has completed all actions
TaskComplete(TaskCompleteEvent),
/// Token count event, sent periodically to report the number of tokens
/// used in the current session.
TokenCount(TokenUsage),
/// Usage update for the current session, including totals and last turn.
/// Optional means unknown — UIs should not display when `None`.
TokenCount(TokenCountEvent),
/// Agent text output message
AgentMessage(AgentMessageEvent),
@@ -521,12 +520,54 @@ pub struct TaskStartedEvent {
#[derive(Debug, Clone, Deserialize, Serialize, Default)]
pub struct TokenUsage {
pub input_tokens: u64,
pub cached_input_tokens: Option<u64>,
pub cached_input_tokens: u64,
pub output_tokens: u64,
pub reasoning_output_tokens: Option<u64>,
pub reasoning_output_tokens: u64,
pub total_tokens: u64,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TokenUsageInfo {
pub total_token_usage: TokenUsage,
pub last_token_usage: TokenUsage,
pub model_context_window: Option<u64>,
}
impl TokenUsageInfo {
pub fn new_or_append(
info: &Option<TokenUsageInfo>,
last: &Option<TokenUsage>,
model_context_window: Option<u64>,
) -> Option<Self> {
if info.is_none() && last.is_none() {
return None;
}
let mut info = match info {
Some(info) => info.clone(),
None => Self {
total_token_usage: TokenUsage::default(),
last_token_usage: TokenUsage::default(),
model_context_window,
},
};
if let Some(last) = last {
info.append_last_usage(last);
}
Some(info)
}
pub fn append_last_usage(&mut self, last: &TokenUsage) {
self.total_token_usage.add_assign(last);
self.last_token_usage = last.clone();
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TokenCountEvent {
pub info: Option<TokenUsageInfo>,
}
// Includes prompts, tools and space to call compact.
const BASELINE_TOKENS: u64 = 12000;
@@ -536,7 +577,7 @@ impl TokenUsage {
}
pub fn cached_input(&self) -> u64 {
self.cached_input_tokens.unwrap_or(0)
self.cached_input_tokens
}
pub fn non_cached_input(&self) -> u64 {
@@ -554,7 +595,7 @@ impl TokenUsage {
/// This will be off for the current turn and pending function calls.
pub fn tokens_in_context_window(&self) -> u64 {
self.total_tokens
.saturating_sub(self.reasoning_output_tokens.unwrap_or(0))
.saturating_sub(self.reasoning_output_tokens)
}
/// Estimate the remaining user-controllable percentage of the model's context window.
@@ -579,6 +620,15 @@ impl TokenUsage {
let remaining = effective_window.saturating_sub(used);
((remaining as f32 / effective_window as f32) * 100.0).clamp(0.0, 100.0) as u8
}
/// In-place element-wise sum of token counts.
pub fn add_assign(&mut self, other: &TokenUsage) {
self.input_tokens += other.input_tokens;
self.cached_input_tokens += other.cached_input_tokens;
self.output_tokens += other.output_tokens;
self.reasoning_output_tokens += other.reasoning_output_tokens;
self.total_tokens += other.total_tokens;
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -606,10 +656,11 @@ impl fmt::Display for FinalOutput {
String::new()
},
token_usage.output_tokens,
token_usage
.reasoning_output_tokens
.map(|r| format!(" (reasoning {r})"))
.unwrap_or_default()
if token_usage.reasoning_output_tokens > 0 {
format!(" (reasoning {})", token_usage.reasoning_output_tokens)
} else {
String::new()
}
)
}
}
@@ -739,7 +790,7 @@ pub struct WebSearchEndEvent {
/// in-memory transcript.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ConversationHistoryResponseEvent {
pub conversation_id: Uuid,
pub conversation_id: ConversationId,
pub entries: Vec<ResponseItem>,
}
@@ -879,8 +930,8 @@ pub struct ListCustomPromptsResponseEvent {
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
pub struct SessionConfiguredEvent {
/// Unique id for this session.
pub session_id: Uuid,
/// Name left as session_id instead of conversation_id for backwards compatibility.
pub session_id: ConversationId,
/// Tell the client what model is being queried.
pub model: String,
@@ -962,11 +1013,11 @@ mod tests {
/// amount of nesting.
#[test]
fn serialize_event() {
let session_id: Uuid = uuid::uuid!("67e55044-10b1-426f-9247-bb680e5fe0c8");
let conversation_id = ConversationId(uuid::uuid!("67e55044-10b1-426f-9247-bb680e5fe0c8"));
let event = Event {
id: "1234".to_string(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id,
session_id: conversation_id,
model: "codex-mini-latest".to_string(),
history_log_id: 0,
history_entry_count: 0,
codex-rs/tui/Cargo.toml
+2 -2
View File
@@ -44,7 +44,7 @@ crossterm = { version = "0.28.1", features = [
"event-stream",
] }
diffy = "0.4.2"
image = { version = "^0.25.6", default-features = false, features = [
image = { version = "^0.25.8", default-features = false, features = [
"jpeg",
"png",
] }
@@ -100,7 +100,7 @@ arboard = "3"
[dev-dependencies]
chrono = { version = "0.4", features = ["serde"] }
insta = "1.43.1"
insta = "1.43.2"
pretty_assertions = "1"
rand = "0.9"
vt100 = "0.16.2"
codex-rs/tui/src/app_backtrack.rs
+6 -5
View File
@@ -4,6 +4,7 @@ use crate::pager_overlay::Overlay;
use crate::tui;
use crate::tui::TuiEvent;
use codex_core::protocol::ConversationHistoryResponseEvent;
use codex_protocol::mcp_protocol::ConversationId;
use color_eyre::eyre::Result;
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
@@ -14,13 +15,13 @@ pub(crate) struct BacktrackState {
/// True when Esc has primed backtrack mode in the main view.
pub(crate) primed: bool,
/// Session id of the base conversation to fork from.
pub(crate) base_id: Option<uuid::Uuid>,
pub(crate) base_id: Option<ConversationId>,
/// Current step count (Nth last user message).
pub(crate) count: usize,
/// True when the transcript overlay is showing a backtrack preview.
pub(crate) overlay_preview_active: bool,
/// Pending fork request: (base_id, drop_count, prefill).
pub(crate) pending: Option<(uuid::Uuid, usize, String)>,
pub(crate) pending: Option<(ConversationId, usize, String)>,
}
impl App {
@@ -91,7 +92,7 @@ impl App {
pub(crate) fn request_backtrack(
&mut self,
prefill: String,
base_id: uuid::Uuid,
base_id: ConversationId,
drop_last_messages: usize,
) {
self.backtrack.pending = Some((base_id, drop_last_messages, prefill));
@@ -135,7 +136,7 @@ impl App {
fn prime_backtrack(&mut self) {
self.backtrack.primed = true;
self.backtrack.count = 0;
self.backtrack.base_id = self.chat_widget.session_id();
self.backtrack.base_id = self.chat_widget.conversation_id();
self.chat_widget.show_esc_backtrack_hint();
}
@@ -151,7 +152,7 @@ impl App {
/// When overlay is already open, begin preview mode and select latest user message.
fn begin_overlay_backtrack_preview(&mut self, tui: &mut tui::Tui) {
self.backtrack.primed = true;
self.backtrack.base_id = self.chat_widget.session_id();
self.backtrack.base_id = self.chat_widget.conversation_id();
self.backtrack.overlay_preview_active = true;
let sel = self.compute_backtrack_selection(tui, 1);
self.apply_backtrack_selection(sel);
codex-rs/tui/src/bottom_pane/bottom_pane_view.rs
+1 -1
View File
@@ -19,7 +19,7 @@ pub(crate) trait BottomPaneView {
/// Handle Ctrl-C while this view is active.
fn on_ctrl_c(&mut self, _pane: &mut BottomPane) -> CancellationEvent {
CancellationEvent::Ignored
CancellationEvent::NotHandled
}
/// Return the desired height of the view.
codex-rs/tui/src/bottom_pane/chat_composer.rs
+24 -23
View File
@@ -1,4 +1,4 @@
use codex_core::protocol::TokenUsage;
use codex_core::protocol::TokenUsageInfo;
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyEventKind;
@@ -63,12 +63,6 @@ struct AttachedImage {
path: PathBuf,
}
struct TokenUsageInfo {
total_token_usage: TokenUsage,
last_token_usage: TokenUsage,
model_context_window: Option<u64>,
}
pub(crate) struct ChatComposer {
textarea: TextArea,
textarea_state: RefCell<TextAreaState>,
@@ -85,6 +79,7 @@ pub(crate) struct ChatComposer {
has_focus: bool,
attached_images: Vec<AttachedImage>,
placeholder_text: String,
is_task_running: bool,
// Non-bracketed paste burst tracker.
paste_burst: PasteBurst,
// When true, disables paste-burst logic and inserts characters immediately.
@@ -125,6 +120,7 @@ impl ChatComposer {
has_focus: has_input_focus,
attached_images: Vec::new(),
placeholder_text,
is_task_running: false,
paste_burst: PasteBurst::default(),
disable_paste_burst: false,
custom_prompts: Vec::new(),
@@ -166,17 +162,8 @@ impl ChatComposer {
/// Update the cached *context-left* percentage and refresh the placeholder
/// text. The UI relies on the placeholder to convey the remaining
/// context when the composer is empty.
pub(crate) fn set_token_usage(
&mut self,
total_token_usage: TokenUsage,
last_token_usage: TokenUsage,
model_context_window: Option<u64>,
) {
self.token_usage_info = Some(TokenUsageInfo {
total_token_usage,
last_token_usage,
model_context_window,
});
pub(crate) fn set_token_usage(&mut self, token_info: Option<TokenUsageInfo>) {
self.token_usage_info = token_info;
}
/// Record the history metadata advertised by `SessionConfiguredEvent` so
@@ -1220,6 +1207,10 @@ impl ChatComposer {
self.has_focus = has_focus;
}
pub fn set_task_running(&mut self, running: bool) {
self.is_task_running = running;
}
pub(crate) fn set_esc_backtrack_hint(&mut self, show: bool) {
self.esc_backtrack_hint = show;
}
@@ -1244,11 +1235,16 @@ impl WidgetRef for ChatComposer {
ActivePopup::None => {
let bottom_line_rect = popup_rect;
let mut hint: Vec<Span<'static>> = if self.ctrl_c_quit_hint {
let ctrl_c_followup = if self.is_task_running {
" to interrupt"
} else {
" to quit"
};
vec![
" ".into(),
key_hint::ctrl('C'),
" again".into(),
" to quit".into(),
ctrl_c_followup.into(),
]
} else {
let newline_hint_key = if self.use_shift_enter_hint {
@@ -1290,11 +1286,16 @@ impl WidgetRef for ChatComposer {
} else {
100
};
let context_style = if percent_remaining < 20 {
Style::default().fg(Color::Yellow)
} else {
Style::default().add_modifier(Modifier::DIM)
};
hint.push(" ".into());
hint.push(
Span::from(format!("{percent_remaining}% context left"))
.style(Style::default().add_modifier(Modifier::DIM)),
);
hint.push(Span::styled(
format!("{percent_remaining}% context left"),
context_style,
));
}
}
codex-rs/tui/src/bottom_pane/mod.rs
+24 -20
View File
@@ -5,7 +5,7 @@ use crate::app_event_sender::AppEventSender;
use crate::tui::FrameRequester;
use crate::user_approval_widget::ApprovalRequest;
use bottom_pane_view::BottomPaneView;
use codex_core::protocol::TokenUsage;
use codex_core::protocol::TokenUsageInfo;
use codex_file_search::FileMatch;
use crossterm::event::KeyEvent;
use ratatui::buffer::Buffer;
@@ -30,8 +30,8 @@ mod textarea;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CancellationEvent {
Ignored,
Handled,
NotHandled,
}
pub(crate) use chat_composer::ChatComposer;
@@ -195,7 +195,15 @@ impl BottomPane {
pub(crate) fn on_ctrl_c(&mut self) -> CancellationEvent {
let mut view = match self.active_view.take() {
Some(view) => view,
None => return CancellationEvent::Ignored,
None => {
return if self.composer_is_empty() {
CancellationEvent::NotHandled
} else {
self.set_composer_text(String::new());
self.show_ctrl_c_quit_hint();
CancellationEvent::Handled
};
}
};
let event = view.on_ctrl_c(self);
@@ -208,7 +216,7 @@ impl BottomPane {
}
self.show_ctrl_c_quit_hint();
}
CancellationEvent::Ignored => {
CancellationEvent::NotHandled => {
self.active_view = Some(view);
}
}
@@ -267,6 +275,7 @@ impl BottomPane {
}
}
#[cfg(test)]
pub(crate) fn ctrl_c_quit_hint_visible(&self) -> bool {
self.ctrl_c_quit_hint
}
@@ -289,6 +298,7 @@ impl BottomPane {
pub fn set_task_running(&mut self, running: bool) {
self.is_task_running = running;
self.composer.set_task_running(running);
if running {
if self.status.is_none() {
@@ -358,14 +368,8 @@ impl BottomPane {
/// Update the *context-window remaining* indicator in the composer. This
/// is forwarded directly to the underlying `ChatComposer`.
pub(crate) fn set_token_usage(
&mut self,
total_token_usage: TokenUsage,
last_token_usage: TokenUsage,
model_context_window: Option<u64>,
) {
self.composer
.set_token_usage(total_token_usage, last_token_usage, model_context_window);
pub(crate) fn set_token_usage(&mut self, token_info: Option<TokenUsageInfo>) {
self.composer.set_token_usage(token_info);
self.request_redraw();
}
@@ -510,7 +514,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -519,7 +523,7 @@ mod tests {
pane.push_approval_request(exec_request());
assert_eq!(CancellationEvent::Handled, pane.on_ctrl_c());
assert!(pane.ctrl_c_quit_hint_visible());
assert_eq!(CancellationEvent::Ignored, pane.on_ctrl_c());
assert_eq!(CancellationEvent::NotHandled, pane.on_ctrl_c());
}
// live ring removed; related tests deleted.
@@ -530,7 +534,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -561,7 +565,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx.clone(),
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -589,7 +593,7 @@ mod tests {
// Render and ensure the top row includes the Working header and a composer line below.
// Give the animation thread a moment to tick.
std::thread::sleep(std::time::Duration::from_millis(120));
std::thread::sleep(Duration::from_millis(120));
let area = Rect::new(0, 0, 40, 6);
let mut buf = Buffer::empty(area);
(&pane).render_ref(area, &mut buf);
@@ -629,7 +633,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -660,7 +664,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -711,7 +715,7 @@ mod tests {
let tx = AppEventSender::new(tx_raw);
let mut pane = BottomPane::new(BottomPaneParams {
app_event_tx: tx,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
codex-rs/tui/src/chatwidget.rs
+41 -69
View File
@@ -29,6 +29,7 @@ use codex_core::protocol::PatchApplyBeginEvent;
use codex_core::protocol::StreamErrorEvent;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::protocol::TokenUsage;
use codex_core::protocol::TokenUsageInfo;
use codex_core::protocol::TurnAbortReason;
use codex_core::protocol::TurnDiffEvent;
use codex_core::protocol::UserMessageEvent;
@@ -84,7 +85,7 @@ use codex_core::protocol::AskForApproval;
use codex_core::protocol::SandboxPolicy;
use codex_core::protocol_config_types::ReasoningEffort as ReasoningEffortConfig;
use codex_file_search::FileMatch;
use uuid::Uuid;
use codex_protocol::mcp_protocol::ConversationId;
// Track information about an in-flight exec command.
struct RunningCommand {
@@ -109,8 +110,7 @@ pub(crate) struct ChatWidget {
active_exec_cell: Option<ExecCell>,
config: Config,
initial_user_message: Option<UserMessage>,
total_token_usage: TokenUsage,
last_token_usage: TokenUsage,
token_info: Option<TokenUsageInfo>,
// Stream lifecycle controller
stream: StreamController,
running_commands: HashMap<String, RunningCommand>,
@@ -121,7 +121,7 @@ pub(crate) struct ChatWidget {
reasoning_buffer: String,
// Accumulates full reasoning content for transcript-only recording
full_reasoning_buffer: String,
session_id: Option<Uuid>,
conversation_id: Option<ConversationId>,
frame_requester: FrameRequester,
// Whether to include the initial welcome banner on session configured
show_welcome_banner: bool,
@@ -163,7 +163,7 @@ impl ChatWidget {
fn on_session_configured(&mut self, event: codex_core::protocol::SessionConfiguredEvent) {
self.bottom_pane
.set_history_metadata(event.history_log_id, event.history_entry_count);
self.session_id = Some(event.session_id);
self.conversation_id = Some(event.session_id);
let initial_messages = event.initial_messages.clone();
if let Some(messages) = initial_messages {
self.replay_initial_messages(messages);
@@ -259,16 +259,10 @@ impl ChatWidget {
self.maybe_send_next_queued_input();
}
fn on_token_count(&mut self, token_usage: TokenUsage) {
self.total_token_usage = add_token_usage(&self.total_token_usage, &token_usage);
self.last_token_usage = token_usage;
self.bottom_pane.set_token_usage(
self.total_token_usage.clone(),
self.last_token_usage.clone(),
self.config.model_context_window,
);
pub(crate) fn set_token_info(&mut self, info: Option<TokenUsageInfo>) {
self.bottom_pane.set_token_usage(info.clone());
self.token_info = info;
}
/// Finalize any active exec as failed, push an error message into history,
/// and stop/clear running UI state.
fn finalize_turn_with_error_message(&mut self, message: String) {
@@ -659,15 +653,14 @@ impl ChatWidget {
initial_prompt.unwrap_or_default(),
initial_images,
),
total_token_usage: TokenUsage::default(),
last_token_usage: TokenUsage::default(),
token_info: None,
stream: StreamController::new(config),
running_commands: HashMap::new(),
task_complete_pending: false,
interrupts: InterruptManager::new(),
reasoning_buffer: String::new(),
full_reasoning_buffer: String::new(),
session_id: None,
conversation_id: None,
queued_user_messages: VecDeque::new(),
show_welcome_banner: true,
suppress_session_configured_redraw: false,
@@ -712,15 +705,14 @@ impl ChatWidget {
initial_prompt.unwrap_or_default(),
initial_images,
),
total_token_usage: TokenUsage::default(),
last_token_usage: TokenUsage::default(),
token_info: None,
stream: StreamController::new(config),
running_commands: HashMap::new(),
task_complete_pending: false,
interrupts: InterruptManager::new(),
reasoning_buffer: String::new(),
full_reasoning_buffer: String::new(),
session_id: None,
conversation_id: None,
queued_user_messages: VecDeque::new(),
show_welcome_banner: false,
suppress_session_configured_redraw: true,
@@ -1050,7 +1042,7 @@ impl ChatWidget {
EventMsg::AgentReasoningSectionBreak(_) => self.on_reasoning_section_break(),
EventMsg::TaskStarted(_) => self.on_task_started(),
EventMsg::TaskComplete(TaskCompleteEvent { .. }) => self.on_task_complete(),
EventMsg::TokenCount(token_usage) => self.on_token_count(token_usage),
EventMsg::TokenCount(ev) => self.set_token_info(ev.info),
EventMsg::Error(ErrorEvent { message }) => self.on_error(message),
EventMsg::TurnAborted(ev) => match ev.reason {
TurnAbortReason::Interrupted => {
@@ -1157,10 +1149,17 @@ impl ChatWidget {
}
pub(crate) fn add_status_output(&mut self) {
let default_usage;
let usage_ref = if let Some(ti) = &self.token_info {
&ti.total_token_usage
} else {
default_usage = TokenUsage::default();
&default_usage
};
self.add_to_history(history_cell::new_status_output(
&self.config,
&self.total_token_usage,
&self.session_id,
usage_ref,
&self.conversation_id,
));
}
@@ -1289,15 +1288,17 @@ impl ChatWidget {
/// Handle Ctrl-C key press.
fn on_ctrl_c(&mut self) {
if self.bottom_pane.on_ctrl_c() == CancellationEvent::Ignored {
if self.bottom_pane.is_task_running() {
self.submit_op(Op::Interrupt);
} else if self.bottom_pane.ctrl_c_quit_hint_visible() {
self.submit_op(Op::Shutdown);
} else {
self.bottom_pane.show_ctrl_c_quit_hint();
}
if self.bottom_pane.on_ctrl_c() == CancellationEvent::Handled {
return;
}
if self.bottom_pane.is_task_running() {
self.bottom_pane.show_ctrl_c_quit_hint();
self.submit_op(Op::Interrupt);
return;
}
self.submit_op(Op::Shutdown);
}
pub(crate) fn composer_is_empty(&self) -> bool {
@@ -1352,12 +1353,15 @@ impl ChatWidget {
self.submit_user_message(text.into());
}
pub(crate) fn token_usage(&self) -> &TokenUsage {
&self.total_token_usage
pub(crate) fn token_usage(&self) -> TokenUsage {
self.token_info
.as_ref()
.map(|ti| ti.total_token_usage.clone())
.unwrap_or_default()
}
pub(crate) fn session_id(&self) -> Option<Uuid> {
self.session_id
pub(crate) fn conversation_id(&self) -> Option<ConversationId> {
self.conversation_id
}
/// Return a reference to the widget's current config (includes any
@@ -1367,12 +1371,8 @@ impl ChatWidget {
}
pub(crate) fn clear_token_usage(&mut self) {
self.total_token_usage = TokenUsage::default();
self.bottom_pane.set_token_usage(
self.total_token_usage.clone(),
self.last_token_usage.clone(),
self.config.model_context_window,
);
self.token_info = None;
self.bottom_pane.set_token_usage(None);
}
pub fn cursor_pos(&self, area: Rect) -> Option<(u16, u16)> {
@@ -1405,34 +1405,6 @@ const EXAMPLE_PROMPTS: [&str; 6] = [
"Improve documentation in @filename",
];
fn add_token_usage(current_usage: &TokenUsage, new_usage: &TokenUsage) -> TokenUsage {
let cached_input_tokens = match (
current_usage.cached_input_tokens,
new_usage.cached_input_tokens,
) {
(Some(current), Some(new)) => Some(current + new),
(Some(current), None) => Some(current),
(None, Some(new)) => Some(new),
(None, None) => None,
};
let reasoning_output_tokens = match (
current_usage.reasoning_output_tokens,
new_usage.reasoning_output_tokens,
) {
(Some(current), Some(new)) => Some(current + new),
(Some(current), None) => Some(current),
(None, Some(new)) => Some(new),
(None, None) => None,
};
TokenUsage {
input_tokens: current_usage.input_tokens + new_usage.input_tokens,
cached_input_tokens,
output_tokens: current_usage.output_tokens + new_usage.output_tokens,
reasoning_output_tokens,
total_tokens: current_usage.total_tokens + new_usage.total_tokens,
}
}
// Extract the first bold (Markdown) element in the form **...** from `s`.
// Returns the inner text if found; otherwise `None`.
fn extract_first_bold(s: &str) -> Option<String> {
codex-rs/tui/src/chatwidget/tests.rs
+30 -30
View File
@@ -25,6 +25,7 @@ use codex_core::protocol::PatchApplyEndEvent;
use codex_core::protocol::StreamErrorEvent;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::protocol::TaskStartedEvent;
use codex_protocol::mcp_protocol::ConversationId;
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
@@ -35,11 +36,10 @@ use std::io::BufRead;
use std::io::BufReader;
use std::path::PathBuf;
use tokio::sync::mpsc::unbounded_channel;
use uuid::Uuid;
fn test_config() -> Config {
// Use base defaults to avoid depending on host state.
codex_core::config::Config::load_from_base_config_with_overrides(
Config::load_from_base_config_with_overrides(
ConfigToml::default(),
ConfigOverrides::default(),
std::env::temp_dir(),
@@ -79,7 +79,7 @@ fn final_answer_without_newline_is_flushed_immediately() {
// Set up a VT100 test terminal to capture ANSI visual output
let width: u16 = 80;
let height: u16 = 2000;
let viewport = ratatui::layout::Rect::new(0, height - 1, width, 1);
let viewport = Rect::new(0, height - 1, width, 1);
let backend = ratatui::backend::TestBackend::new(width, height);
let mut terminal = crate::custom_terminal::Terminal::with_options(backend)
.expect("failed to construct terminal");
@@ -132,13 +132,15 @@ fn final_answer_without_newline_is_flushed_immediately() {
fn resumed_initial_messages_render_history() {
let (mut chat, mut rx, _ops) = make_chatwidget_manual();
let conversation_id = ConversationId::new();
let configured = codex_core::protocol::SessionConfiguredEvent {
session_id: Uuid::nil(),
session_id: conversation_id,
model: "test-model".to_string(),
history_log_id: 0,
history_entry_count: 0,
initial_messages: Some(vec![
EventMsg::UserMessage(codex_core::protocol::UserMessageEvent {
EventMsg::UserMessage(UserMessageEvent {
message: "hello from user".to_string(),
kind: Some(InputMessageKind::Plain),
}),
@@ -185,7 +187,7 @@ async fn helpers_are_available_and_do_not_panic() {
)));
let init = ChatWidgetInit {
config: cfg,
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
app_event_tx: tx,
initial_prompt: None,
initial_images: Vec::new(),
@@ -208,7 +210,7 @@ fn make_chatwidget_manual() -> (
let cfg = test_config();
let bottom = BottomPane::new(BottomPaneParams {
app_event_tx: app_event_tx.clone(),
frame_requester: crate::tui::FrameRequester::test_dummy(),
frame_requester: FrameRequester::test_dummy(),
has_input_focus: true,
enhanced_keys_supported: false,
placeholder_text: "Ask Codex to do anything".to_string(),
@@ -221,18 +223,17 @@ fn make_chatwidget_manual() -> (
active_exec_cell: None,
config: cfg.clone(),
initial_user_message: None,
total_token_usage: TokenUsage::default(),
last_token_usage: TokenUsage::default(),
token_info: None,
stream: StreamController::new(cfg),
running_commands: HashMap::new(),
task_complete_pending: false,
interrupts: InterruptManager::new(),
reasoning_buffer: String::new(),
full_reasoning_buffer: String::new(),
session_id: None,
frame_requester: crate::tui::FrameRequester::test_dummy(),
conversation_id: None,
frame_requester: FrameRequester::test_dummy(),
show_welcome_banner: true,
queued_user_messages: std::collections::VecDeque::new(),
queued_user_messages: VecDeque::new(),
suppress_session_configured_redraw: false,
};
(widget, rx, op_rx)
@@ -368,11 +369,10 @@ fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) {
// Build the full command vec and parse it using core's parser,
// then convert to protocol variants for the event payload.
let command = vec!["bash".to_string(), "-lc".to_string(), raw_cmd.to_string()];
let parsed_cmd: Vec<codex_protocol::parse_command::ParsedCommand> =
codex_core::parse_command::parse_command(&command)
.into_iter()
.map(Into::into)
.collect();
let parsed_cmd: Vec<ParsedCommand> = codex_core::parse_command::parse_command(&command)
.into_iter()
.map(Into::into)
.collect();
chat.handle_codex_event(Event {
id: call_id.to_string(),
msg: EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
@@ -413,7 +413,7 @@ fn active_blob(chat: &ChatWidget) -> String {
lines_to_single_string(&lines)
}
fn open_fixture(name: &str) -> std::fs::File {
fn open_fixture(name: &str) -> File {
// 1) Prefer fixtures within this crate
{
let mut p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
@@ -621,7 +621,7 @@ async fn binary_size_transcript_snapshot() {
// Set up a VT100 test terminal to capture ANSI visual output
let width: u16 = 80;
let height: u16 = 2000;
let viewport = ratatui::layout::Rect::new(0, height - 1, width, 1);
let viewport = Rect::new(0, height - 1, width, 1);
let backend = ratatui::backend::TestBackend::new(width, height);
let mut terminal = crate::custom_terminal::Terminal::with_options(backend)
.expect("failed to construct terminal");
@@ -806,7 +806,7 @@ fn approval_modal_exec_snapshot() {
// Build a chat widget with manual channels to avoid spawning the agent.
let (mut chat, _rx, _op_rx) = make_chatwidget_manual();
// Ensure policy allows surfacing approvals explicitly (not strictly required for direct event).
chat.config.approval_policy = codex_core::protocol::AskForApproval::OnRequest;
chat.config.approval_policy = AskForApproval::OnRequest;
// Inject an exec approval request to display the approval modal.
let ev = ExecApprovalRequestEvent {
call_id: "call-approve-cmd".into(),
@@ -836,7 +836,7 @@ fn approval_modal_exec_snapshot() {
#[test]
fn approval_modal_exec_without_reason_snapshot() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual();
chat.config.approval_policy = codex_core::protocol::AskForApproval::OnRequest;
chat.config.approval_policy = AskForApproval::OnRequest;
let ev = ExecApprovalRequestEvent {
call_id: "call-approve-cmd-noreason".into(),
@@ -862,10 +862,10 @@ fn approval_modal_exec_without_reason_snapshot() {
#[test]
fn approval_modal_patch_snapshot() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual();
chat.config.approval_policy = codex_core::protocol::AskForApproval::OnRequest;
chat.config.approval_policy = AskForApproval::OnRequest;
// Build a small changeset and a reason/grant_root to exercise the prompt text.
let mut changes = std::collections::HashMap::new();
let mut changes = HashMap::new();
changes.insert(
PathBuf::from("README.md"),
FileChange::Add {
@@ -911,7 +911,7 @@ fn interrupt_restores_queued_messages_into_composer() {
chat.handle_codex_event(Event {
id: "turn-1".into(),
msg: EventMsg::TurnAborted(codex_core::protocol::TurnAbortedEvent {
reason: codex_core::protocol::TurnAbortReason::Interrupted,
reason: TurnAbortReason::Interrupted,
}),
});
@@ -1345,7 +1345,7 @@ fn apply_patch_full_flow_integration_like() {
fn apply_patch_untrusted_shows_approval_modal() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual();
// Ensure approval policy is untrusted (OnRequest)
chat.config.approval_policy = codex_core::protocol::AskForApproval::OnRequest;
chat.config.approval_policy = AskForApproval::OnRequest;
// Simulate a patch approval request from backend
let mut changes = HashMap::new();
@@ -1364,8 +1364,8 @@ fn apply_patch_untrusted_shows_approval_modal() {
});
// Render and ensure the approval modal title is present
let area = ratatui::layout::Rect::new(0, 0, 80, 12);
let mut buf = ratatui::buffer::Buffer::empty(area);
let area = Rect::new(0, 0, 80, 12);
let mut buf = Buffer::empty(area);
(&chat).render_ref(area, &mut buf);
let mut contains_title = false;
@@ -1390,7 +1390,7 @@ fn apply_patch_request_shows_diff_summary() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual();
// Ensure we are in OnRequest so an approval is surfaced
chat.config.approval_policy = codex_core::protocol::AskForApproval::OnRequest;
chat.config.approval_policy = AskForApproval::OnRequest;
// Simulate backend asking to apply a patch adding two lines to README.md
let mut changes = HashMap::new();
@@ -1692,7 +1692,7 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() {
let width: u16 = 80;
let ui_height: u16 = chat.desired_height(width);
let vt_height: u16 = 40;
let viewport = ratatui::layout::Rect::new(0, vt_height - ui_height, width, ui_height);
let viewport = Rect::new(0, vt_height - ui_height, width, ui_height);
// Use TestBackend for the terminal (no real ANSI emitted by drawing),
// but capture VT100 escape stream for history insertion with a separate writer.
@@ -1707,7 +1707,7 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() {
}
// 2) Render the ChatWidget UI into an off-screen buffer using WidgetRef directly
let mut ui_buf = ratatui::buffer::Buffer::empty(viewport);
let mut ui_buf = Buffer::empty(viewport);
(&chat).render_ref(viewport, &mut ui_buf);
// 3) Build VT100 visual from the captured ANSI
codex-rs/tui/src/history_cell.rs
+4 -5
View File
@@ -27,6 +27,7 @@ use codex_core::protocol::McpInvocation;
use codex_core::protocol::SandboxPolicy;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::protocol::TokenUsage;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::parse_command::ParsedCommand;
use image::DynamicImage;
use image::ImageReader;
@@ -49,7 +50,6 @@ use std::time::Duration;
use std::time::Instant;
use tracing::error;
use unicode_width::UnicodeWidthStr;
use uuid::Uuid;
#[derive(Clone, Debug)]
pub(crate) struct CommandOutput {
@@ -821,7 +821,7 @@ pub(crate) fn new_completed_mcp_tool_call(
pub(crate) fn new_status_output(
config: &Config,
usage: &TokenUsage,
session_id: &Option<Uuid>,
session_id: &Option<ConversationId>,
) -> PlainHistoryCell {
let mut lines: Vec<Line<'static>> = Vec::new();
lines.push("/status".magenta().into());
@@ -966,9 +966,8 @@ pub(crate) fn new_status_output(
" • Input: ".into(),
usage.non_cached_input().to_string().into(),
];
if let Some(cached) = usage.cached_input_tokens
&& cached > 0
{
if usage.cached_input_tokens > 0 {
let cached = usage.cached_input_tokens;
input_line_spans.push(format!(" (+ {cached} cached)").into());
}
lines.push(Line::from(input_line_spans));