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

Merge remote-tracking branch 'origin/main' into jif/sandbox-1

Browse Source 
# Conflicts:
#	codex-rs/core/src/codex.rs
This commit is contained in:
jimmyfraiture
2025-09-30 12:45:23 +01:00
parent 43c0abb31e 5b038135de
commit 4533dceafa
198 changed files with 20572 additions and 1538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/apply_patch.rs
+7 -6
View File
@@ -46,12 +46,13 @@ pub(crate) async fn apply_patch(
&turn_context.sandbox_policy,
&turn_context.cwd,
) {
SafetyCheck::AutoApprove { .. } => {
InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
action,
user_explicitly_approved_this_action: false,
})
}
SafetyCheck::AutoApprove {
user_explicitly_approved,
..
} => InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
action,
user_explicitly_approved_this_action: user_explicitly_approved,
}),
SafetyCheck::AskUser => {
// Compute a readable summary of path changes to include in the
// approval request so the user can make an informed decision.
codex-rs/core/src/chat_completions.rs
+27 -18
View File
@@ -1,6 +1,19 @@
use std::time::Duration;
use crate::ModelProviderInfo;
use crate::client_common::Prompt;
use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::error::CodexErr;
use crate::error::Result;
use crate::model_family::ModelFamily;
use crate::openai_tools::create_tools_json_for_chat_completions_api;
use crate::util::backoff;
use bytes::Bytes;
use codex_otel::otel_event_manager::OtelEventManager;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ReasoningItemContent;
use codex_protocol::models::ResponseItem;
use eventsource_stream::Eventsource;
use futures::Stream;
use futures::StreamExt;
@@ -15,25 +28,13 @@ use tokio::time::timeout;
use tracing::debug;
use tracing::trace;
use crate::ModelProviderInfo;
use crate::client_common::Prompt;
use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::error::CodexErr;
use crate::error::Result;
use crate::model_family::ModelFamily;
use crate::openai_tools::create_tools_json_for_chat_completions_api;
use crate::util::backoff;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ReasoningItemContent;
use codex_protocol::models::ResponseItem;
/// Implementation for the classic Chat Completions API.
pub(crate) async fn stream_chat_completions(
prompt: &Prompt,
model_family: &ModelFamily,
client: &reqwest::Client,
provider: &ModelProviderInfo,
otel_event_manager: &OtelEventManager,
) -> Result<ResponseStream> {
if prompt.output_schema.is_some() {
return Err(CodexErr::UnsupportedOperation(
@@ -294,10 +295,13 @@ pub(crate) async fn stream_chat_completions(
let req_builder = provider.create_request_builder(client, &None).await?;
let res = req_builder
.header(reqwest::header::ACCEPT, "text/event-stream")
.json(&payload)
.send()
let res = otel_event_manager
.log_request(attempt, || {
req_builder
.header(reqwest::header::ACCEPT, "text/event-stream")
.json(&payload)
.send()
})
.await;
match res {
@@ -308,6 +312,7 @@ pub(crate) async fn stream_chat_completions(
stream,
tx_event,
provider.stream_idle_timeout(),
otel_event_manager.clone(),
));
return Ok(ResponseStream { rx_event });
}
@@ -351,6 +356,7 @@ async fn process_chat_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
otel_event_manager: OtelEventManager,
) where
S: Stream<Item = Result<Bytes>> + Unpin,
{
@@ -374,7 +380,10 @@ async fn process_chat_sse<S>(
let mut reasoning_text = String::new();
loop {
let sse = match timeout(idle_timeout, stream.next()).await {
let sse = match otel_event_manager
.log_sse_event(|| timeout(idle_timeout, stream.next()))
.await
{
Ok(Some(Ok(ev))) => ev,
Ok(Some(Err(e))) => {
let _ = tx_event
codex-rs/core/src/client.rs
+96 -17
View File
@@ -47,6 +47,7 @@ use crate::protocol::RateLimitWindow;
use crate::protocol::TokenUsage;
use crate::token_data::PlanType;
use crate::util::backoff;
use codex_otel::otel_event_manager::OtelEventManager;
use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::models::ResponseItem;
@@ -73,6 +74,7 @@ struct Error {
pub struct ModelClient {
config: Arc<Config>,
auth_manager: Option<Arc<AuthManager>>,
otel_event_manager: OtelEventManager,
client: reqwest::Client,
provider: ModelProviderInfo,
conversation_id: ConversationId,
@@ -84,6 +86,7 @@ impl ModelClient {
pub fn new(
config: Arc<Config>,
auth_manager: Option<Arc<AuthManager>>,
otel_event_manager: OtelEventManager,
provider: ModelProviderInfo,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
@@ -94,6 +97,7 @@ impl ModelClient {
Self {
config,
auth_manager,
otel_event_manager,
client,
provider,
conversation_id,
@@ -127,6 +131,7 @@ impl ModelClient {
&self.config.model_family,
&self.client,
&self.provider,
&self.otel_event_manager,
)
.await?;
@@ -163,7 +168,12 @@ impl ModelClient {
if let Some(path) = &*CODEX_RS_SSE_FIXTURE {
// short circuit for tests
warn!(path, "Streaming from fixture");
return stream_from_fixture(path, self.provider.clone()).await;
return stream_from_fixture(
path,
self.provider.clone(),
self.otel_event_manager.clone(),
)
.await;
}
let auth_manager = self.auth_manager.clone();
@@ -233,7 +243,7 @@ impl ModelClient {
let max_attempts = self.provider.request_max_retries();
for attempt in 0..=max_attempts {
match self
.attempt_stream_responses(&payload_json, &auth_manager)
.attempt_stream_responses(attempt, &payload_json, &auth_manager)
.await
{
Ok(stream) => {
@@ -258,6 +268,7 @@ impl ModelClient {
/// Single attempt to start a streaming Responses API call.
async fn attempt_stream_responses(
&self,
attempt: u64,
payload_json: &Value,
auth_manager: &Option<Arc<AuthManager>>,
) -> std::result::Result<ResponseStream, StreamAttemptError> {
@@ -291,7 +302,11 @@ impl ModelClient {
req_builder = req_builder.header("chatgpt-account-id", account_id);
}
let res = req_builder.send().await;
let res = self
.otel_event_manager
.log_request(attempt, || req_builder.send())
.await;
if let Ok(resp) = &res {
trace!(
"Response status: {}, cf-ray: {}",
@@ -322,6 +337,7 @@ impl ModelClient {
stream,
tx_event,
self.provider.stream_idle_timeout(),
self.otel_event_manager.clone(),
));
Ok(ResponseStream { rx_event })
@@ -399,6 +415,10 @@ impl ModelClient {
self.provider.clone()
}
pub fn get_otel_event_manager(&self) -> OtelEventManager {
self.otel_event_manager.clone()
}
/// Returns the currently configured model slug.
pub fn get_model(&self) -> String {
self.config.model.clone()
@@ -605,6 +625,7 @@ async fn process_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
otel_event_manager: OtelEventManager,
) where
S: Stream<Item = Result<Bytes>> + Unpin,
{
@@ -616,7 +637,10 @@ async fn process_sse<S>(
let mut response_error: Option<CodexErr> = None;
loop {
let sse = match timeout(idle_timeout, stream.next()).await {
let sse = match otel_event_manager
.log_sse_event(|| timeout(idle_timeout, stream.next()))
.await
{
Ok(Some(Ok(sse))) => sse,
Ok(Some(Err(e))) => {
debug!("SSE Error: {e:#}");
@@ -630,6 +654,21 @@ async fn process_sse<S>(
id: response_id,
usage,
}) => {
if let Some(token_usage) = &usage {
otel_event_manager.sse_event_completed(
token_usage.input_tokens,
token_usage.output_tokens,
token_usage
.input_tokens_details
.as_ref()
.map(|d| d.cached_tokens),
token_usage
.output_tokens_details
.as_ref()
.map(|d| d.reasoning_tokens),
token_usage.total_tokens,
);
}
let event = ResponseEvent::Completed {
response_id,
token_usage: usage.map(Into::into),
@@ -637,12 +676,13 @@ async fn process_sse<S>(
let _ = tx_event.send(Ok(event)).await;
}
None => {
let _ = tx_event
.send(Err(response_error.unwrap_or(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
))))
.await;
let error = response_error.unwrap_or(CodexErr::Stream(
"stream closed before response.completed".into(),
None,
));
otel_event_manager.see_event_completed_failed(&error);
let _ = tx_event.send(Err(error)).await;
}
}
return;
@@ -746,7 +786,9 @@ async fn process_sse<S>(
response_error = Some(CodexErr::Stream(message, delay));
}
Err(e) => {
debug!("failed to parse ErrorResponse: {e}");
let error = format!("failed to parse ErrorResponse: {e}");
debug!(error);
response_error = Some(CodexErr::Stream(error, None))
}
}
}
@@ -760,7 +802,9 @@ async fn process_sse<S>(
response_completed = Some(r);
}
Err(e) => {
debug!("failed to parse ResponseCompleted: {e}");
let error = format!("failed to parse ResponseCompleted: {e}");
debug!(error);
response_error = Some(CodexErr::Stream(error, None));
continue;
}
};
@@ -807,6 +851,7 @@ async fn process_sse<S>(
async fn stream_from_fixture(
path: impl AsRef<Path>,
provider: ModelProviderInfo,
otel_event_manager: OtelEventManager,
) -> Result<ResponseStream> {
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
let f = std::fs::File::open(path.as_ref())?;
@@ -825,6 +870,7 @@ async fn stream_from_fixture(
stream,
tx_event,
provider.stream_idle_timeout(),
otel_event_manager,
));
Ok(ResponseStream { rx_event })
}
@@ -880,6 +926,7 @@ mod tests {
async fn collect_events(
chunks: &[&[u8]],
provider: ModelProviderInfo,
otel_event_manager: OtelEventManager,
) -> Vec<Result<ResponseEvent>> {
let mut builder = IoBuilder::new();
for chunk in chunks {
@@ -889,7 +936,12 @@ mod tests {
let reader = builder.build();
let stream = ReaderStream::new(reader).map_err(CodexErr::Io);
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(16);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
tokio::spawn(process_sse(
stream,
tx,
provider.stream_idle_timeout(),
otel_event_manager,
));
let mut events = Vec::new();
while let Some(ev) = rx.recv().await {
@@ -903,6 +955,7 @@ mod tests {
async fn run_sse(
events: Vec<serde_json::Value>,
provider: ModelProviderInfo,
otel_event_manager: OtelEventManager,
) -> Vec<ResponseEvent> {
let mut body = String::new();
for e in events {
@@ -919,7 +972,12 @@ mod tests {
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(8);
let stream = ReaderStream::new(std::io::Cursor::new(body)).map_err(CodexErr::Io);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
tokio::spawn(process_sse(
stream,
tx,
provider.stream_idle_timeout(),
otel_event_manager,
));
let mut out = Vec::new();
while let Some(ev) = rx.recv().await {
@@ -928,6 +986,18 @@ mod tests {
out
}
fn otel_event_manager() -> OtelEventManager {
OtelEventManager::new(
ConversationId::new(),
"test",
"test",
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
)
}
// ────────────────────────────
// Tests from `implement-test-for-responses-api-sse-parser`
// ────────────────────────────
@@ -979,9 +1049,12 @@ mod tests {
requires_openai_auth: false,
};
let otel_event_manager = otel_event_manager();
let events = collect_events(
&[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()],
provider,
otel_event_manager,
)
.await;
@@ -1039,7 +1112,9 @@ mod tests {
requires_openai_auth: false,
};
let events = collect_events(&[sse1.as_bytes()], provider).await;
let otel_event_manager = otel_event_manager();
let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
assert_eq!(events.len(), 2);
@@ -1073,7 +1148,9 @@ mod tests {
requires_openai_auth: false,
};
let events = collect_events(&[sse1.as_bytes()], provider).await;
let otel_event_manager = otel_event_manager();
let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
assert_eq!(events.len(), 1);
@@ -1178,7 +1255,9 @@ mod tests {
requires_openai_auth: false,
};
let out = run_sse(evs, provider).await;
let otel_event_manager = otel_event_manager();
let out = run_sse(evs, provider, otel_event_manager).await;
assert_eq!(out.len(), case.expected_len, "case {}", case.name);
assert!(
(case.expect_first)(&out[0]),
codex-rs/core/src/codex.rs
+148 -31
View File
@@ -1,5 +1,7 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::fmt::Debug;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
@@ -9,6 +11,7 @@ use crate::client_common::REVIEW_PROMPT;
use crate::event_mapping::map_response_item_to_event_messages;
use crate::function_tool::FunctionCallError;
use crate::review_format::format_review_findings_block;
use crate::terminal;
use crate::user_notification::UserNotifier;
use async_channel::Receiver;
use async_channel::Sender;
@@ -121,6 +124,8 @@ use crate::unified_exec::UnifiedExecSessionManager;
use crate::user_instructions::UserInstructions;
use crate::user_notification::UserNotification;
use crate::util::backoff;
use codex_otel::otel_event_manager::OtelEventManager;
use codex_otel::otel_event_manager::ToolDecisionSource;
use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::custom_prompts::CustomPrompt;
@@ -418,11 +423,35 @@ impl Session {
}
}
let otel_event_manager = OtelEventManager::new(
conversation_id,
config.model.as_str(),
config.model_family.slug.as_str(),
auth_manager.auth().and_then(|a| a.get_account_id()),
auth_manager.auth().map(|a| a.mode),
config.otel.log_user_prompt,
terminal::user_agent(),
);
otel_event_manager.conversation_starts(
config.model_provider.name.as_str(),
config.model_reasoning_effort,
config.model_reasoning_summary,
config.model_context_window,
config.model_max_output_tokens,
config.model_auto_compact_token_limit,
config.approval_policy,
config.sandbox_policy.clone(),
config.mcp_servers.keys().map(String::as_str).collect(),
config.active_profile.clone(),
);
// Now that the conversation id is final (may have been updated by resume),
// construct the model client.
let client = ModelClient::new(
config.clone(),
Some(auth_manager.clone()),
otel_event_manager,
provider.clone(),
model_reasoning_effort,
model_reasoning_summary,
@@ -1108,9 +1137,15 @@ async fn submission_loop(
updated_config.model_context_window = Some(model_info.context_window);
}
let otel_event_manager = prev.client.get_otel_event_manager().with_model(
updated_config.model.as_str(),
updated_config.model_family.slug.as_str(),
);
let client = ModelClient::new(
Arc::new(updated_config),
auth_manager,
otel_event_manager,
provider,
effective_effort,
effective_summary,
@@ -1162,6 +1197,10 @@ async fn submission_loop(
}
}
Op::UserInput { items } => {
turn_context
.client
.get_otel_event_manager()
.user_prompt(&items);
// attempt to inject input into current task
if let Err(items) = sess.inject_input(items).await {
// no current task, spawn a new one
@@ -1179,6 +1218,10 @@ async fn submission_loop(
summary,
final_output_json_schema,
} => {
turn_context
.client
.get_otel_event_manager()
.user_prompt(&items);
// attempt to inject input into current task
if let Err(items) = sess.inject_input(items).await {
// Derive a fresh TurnContext for this turn using the provided overrides.
@@ -1197,11 +1240,18 @@ async fn submission_loop(
per_turn_config.model_context_window = Some(model_info.context_window);
}
let otel_event_manager =
turn_context.client.get_otel_event_manager().with_model(
per_turn_config.model.as_str(),
per_turn_config.model_family.slug.as_str(),
);
// Build a new client with per‑turn reasoning settings.
// Reuse the same provider and session id; auth defaults to env/API key.
let client = ModelClient::new(
Arc::new(per_turn_config),
auth_manager,
otel_event_manager,
provider,
effort,
summary,
@@ -1458,10 +1508,19 @@ async fn spawn_review_thread(
per_turn_config.model_context_window = Some(model_info.context_window);
}
let otel_event_manager = parent_turn_context
.client
.get_otel_event_manager()
.with_model(
per_turn_config.model.as_str(),
per_turn_config.model_family.slug.as_str(),
);
let per_turn_config = Arc::new(per_turn_config);
let client = ModelClient::new(
per_turn_config.clone(),
auth_manager,
otel_event_manager,
provider,
per_turn_config.model_reasoning_effort,
per_turn_config.model_reasoning_summary,
@@ -2126,16 +2185,21 @@ async fn handle_response_item(
.await;
Some(resp)
} else {
let result = handle_function_call(
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
name,
arguments,
call_id.clone(),
)
.await;
let result = turn_context
.client
.get_otel_event_manager()
.log_tool_result(name.as_str(), call_id.as_str(), arguments.as_str(), || {
handle_function_call(
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
name.to_owned(),
arguments.to_owned(),
call_id.clone(),
)
})
.await;
let output = match result {
Ok(content) => FunctionCallOutputPayload {
@@ -2156,6 +2220,7 @@ async fn handle_response_item(
status: _,
action,
} => {
let name = "local_shell";
let LocalShellAction::Exec(action) = action;
tracing::info!("LocalShellCall: {action:?}");
let params = ShellToolCallParams {
@@ -2169,11 +2234,18 @@ async fn handle_response_item(
(Some(call_id), _) => call_id,
(None, Some(id)) => id,
(None, None) => {
error!("LocalShellCall without call_id or id");
let error_message = "LocalShellCall without call_id or id";
turn_context
.client
.get_otel_event_manager()
.log_tool_failed(name, error_message);
error!(error_message);
return Ok(Some(ResponseInputItem::FunctionCallOutput {
call_id: "".to_string(),
output: FunctionCallOutputPayload {
content: "LocalShellCall without call_id or id".to_string(),
content: error_message.to_string(),
success: None,
},
}));
@@ -2182,15 +2254,26 @@ async fn handle_response_item(
let exec_params = to_exec_params(params, turn_context);
{
let result = handle_container_exec_with_params(
exec_params,
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
effective_call_id.clone(),
)
.await;
let result = turn_context
.client
.get_otel_event_manager()
.log_tool_result(
name,
effective_call_id.as_str(),
exec_params.command.join(" ").as_str(),
|| {
handle_container_exec_with_params(
name,
exec_params,
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
effective_call_id.clone(),
)
},
)
.await;
let output = match result {
Ok(content) => FunctionCallOutputPayload {
@@ -2215,16 +2298,21 @@ async fn handle_response_item(
input,
status: _,
} => {
let result = handle_custom_tool_call(
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
name,
input,
call_id.clone(),
)
.await;
let result = turn_context
.client
.get_otel_event_manager()
.log_tool_result(name.as_str(), call_id.as_str(), input.as_str(), || {
handle_custom_tool_call(
sess,
turn_context,
turn_diff_tracker,
sub_id.to_string(),
name.to_owned(),
input.to_owned(),
call_id.clone(),
)
})
.await;
let output = match result {
Ok(content) => content,
@@ -2330,6 +2418,7 @@ async fn handle_function_call(
"container.exec" | "shell" => {
let params = parse_container_exec_arguments(arguments, turn_context, &call_id)?;
handle_container_exec_with_params(
name.as_str(),
params,
sess,
turn_context,
@@ -2393,6 +2482,7 @@ async fn handle_function_call(
justification: None,
};
handle_container_exec_with_params(
name.as_str(),
exec_params,
sess,
turn_context,
@@ -2465,6 +2555,7 @@ async fn handle_custom_tool_call(
};
handle_container_exec_with_params(
name.as_str(),
exec_params,
sess,
turn_context,
@@ -2507,6 +2598,7 @@ fn parse_container_exec_arguments(
}
async fn handle_container_exec_with_params(
tool_name: &str,
params: ExecParams,
sess: &Session,
turn_context: &TurnContext,
@@ -2514,6 +2606,8 @@ async fn handle_container_exec_with_params(
sub_id: String,
call_id: String,
) -> Result<String, FunctionCallError> {
let otel_event_manager = turn_context.client.get_otel_event_manager();
if params.with_escalated_permissions.unwrap_or(false)
&& !matches!(turn_context.approval_policy, AskForApproval::OnRequest)
{
@@ -2889,13 +2983,17 @@ mod tests {
use super::*;
use crate::config::ConfigOverrides;
use crate::config::ConfigToml;
use crate::protocol::CompactedItem;
use crate::protocol::InitialHistory;
use crate::protocol::ResumedHistory;
use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
use codex_protocol::mcp_protocol::AuthMode;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use mcp_types::ContentBlock;
use mcp_types::TextContent;
use pretty_assertions::assert_eq;
@@ -3130,6 +3228,18 @@ mod tests {
})
}
fn otel_event_manager(conversation_id: ConversationId, config: &Config) -> OtelEventManager {
OtelEventManager::new(
conversation_id,
config.model.as_str(),
config.model_family.slug.as_str(),
None,
Some(AuthMode::ChatGPT),
false,
"test".to_string(),
)
}
pub(crate) fn make_session_and_context() -> (Session, TurnContext) {
let (tx_event, _rx_event) = async_channel::unbounded();
let codex_home = tempfile::tempdir().expect("create temp dir");
@@ -3141,9 +3251,11 @@ mod tests {
.expect("load default test config");
let config = Arc::new(config);
let conversation_id = ConversationId::default();
let otel_event_manager = otel_event_manager(conversation_id, config.as_ref());
let client = ModelClient::new(
config.clone(),
None,
otel_event_manager,
config.model_provider.clone(),
config.model_reasoning_effort,
config.model_reasoning_summary,
@@ -3212,9 +3324,11 @@ mod tests {
.expect("load default test config");
let config = Arc::new(config);
let conversation_id = ConversationId::default();
let otel_event_manager = otel_event_manager(conversation_id, config.as_ref());
let client = ModelClient::new(
config.clone(),
None,
otel_event_manager,
config.model_provider.clone(),
config.model_reasoning_effort,
config.model_reasoning_summary,
@@ -3509,10 +3623,12 @@ mod tests {
let mut turn_diff_tracker = TurnDiffTracker::new();
let tool_name = "shell";
let sub_id = "test-sub".to_string();
let call_id = "test-call".to_string();
let resp = handle_container_exec_with_params(
tool_name,
params,
&session,
&turn_context,
@@ -3538,6 +3654,7 @@ mod tests {
turn_context.sandbox_policy = SandboxPolicy::DangerFullAccess;
let resp2 = handle_container_exec_with_params(
tool_name,
params2,
&session,
&turn_context,
codex-rs/core/src/config.rs
+28 -1
View File
@@ -1,8 +1,12 @@
use crate::config_profile::ConfigProfile;
use crate::config_types::DEFAULT_OTEL_ENVIRONMENT;
use crate::config_types::History;
use crate::config_types::McpServerConfig;
use crate::config_types::McpServerTransportConfig;
use crate::config_types::Notifications;
use crate::config_types::OtelConfig;
use crate::config_types::OtelConfigToml;
use crate::config_types::OtelExporterKind;
use crate::config_types::ReasoningSummaryFormat;
use crate::config_types::SandboxWorkspaceWrite;
use crate::config_types::ShellEnvironmentPolicy;
@@ -199,6 +203,9 @@ pub struct Config {
/// All characters are inserted as they are received, and no buffering
/// or placeholder replacement will occur for fast keypress bursts.
pub disable_paste_burst: bool,
/// OTEL configuration (exporter type, endpoint, headers, etc.).
pub otel: crate::config_types::OtelConfig,
}
impl Config {
@@ -426,7 +433,7 @@ fn set_project_trusted_inner(doc: &mut DocumentMut, project_path: &Path) -> anyh
.get_mut(project_key.as_str())
.and_then(|i| i.as_table_mut())
else {
return Err(anyhow::anyhow!("project table missing for {}", project_key));
return Err(anyhow::anyhow!("project table missing for {project_key}"));
};
proj_tbl.set_implicit(false);
proj_tbl["trust_level"] = toml_edit::value("trusted");
@@ -719,6 +726,9 @@ pub struct ConfigToml {
/// All characters are inserted as they are received, and no buffering
/// or placeholder replacement will occur for fast keypress bursts.
pub disable_paste_burst: Option<bool>,
/// OTEL configuration.
pub otel: Option<crate::config_types::OtelConfigToml>,
}
impl From<ConfigToml> for UserSavedConfig {
@@ -1068,6 +1078,19 @@ impl Config {
.as_ref()
.map(|t| t.notifications.clone())
.unwrap_or_default(),
otel: {
let t: OtelConfigToml = cfg.otel.unwrap_or_default();
let log_user_prompt = t.log_user_prompt.unwrap_or(false);
let environment = t
.environment
.unwrap_or(DEFAULT_OTEL_ENVIRONMENT.to_string());
let exporter = t.exporter.unwrap_or(OtelExporterKind::None);
OtelConfig {
log_user_prompt,
environment,
exporter,
}
},
};
Ok(config)
}
@@ -1809,6 +1832,7 @@ model_verbosity = "high"
active_profile: Some("o3".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
},
o3_profile_config
);
@@ -1868,6 +1892,7 @@ model_verbosity = "high"
active_profile: Some("gpt3".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
};
assert_eq!(expected_gpt3_profile_config, gpt3_profile_config);
@@ -1942,6 +1967,7 @@ model_verbosity = "high"
active_profile: Some("zdr".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
};
assert_eq!(expected_zdr_profile_config, zdr_profile_config);
@@ -2002,6 +2028,7 @@ model_verbosity = "high"
active_profile: Some("gpt5".to_string()),
disable_paste_burst: false,
tui_notifications: Default::default(),
otel: OtelConfig::default(),
};
assert_eq!(expected_gpt5_profile_config, gpt5_profile_config);
codex-rs/core/src/config_types.rs
+60
View File
@@ -13,6 +13,8 @@ use serde::Deserialize;
use serde::Serialize;
use serde::de::Error as SerdeError;
pub const DEFAULT_OTEL_ENVIRONMENT: &str = "dev";
#[derive(Serialize, Debug, Clone, PartialEq)]
pub struct McpServerConfig {
#[serde(flatten)]
@@ -219,6 +221,64 @@ pub enum HistoryPersistence {
None,
}
// ===== OTEL configuration =====
#[derive(Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum OtelHttpProtocol {
/// Binary payload
Binary,
/// JSON payload
Json,
}
/// Which OTEL exporter to use.
#[derive(Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum OtelExporterKind {
None,
OtlpHttp {
endpoint: String,
headers: HashMap<String, String>,
protocol: OtelHttpProtocol,
},
OtlpGrpc {
endpoint: String,
headers: HashMap<String, String>,
},
}
/// OTEL settings loaded from config.toml. Fields are optional so we can apply defaults.
#[derive(Deserialize, Debug, Clone, PartialEq, Default)]
pub struct OtelConfigToml {
/// Log user prompt in traces
pub log_user_prompt: Option<bool>,
/// Mark traces with environment (dev, staging, prod, test). Defaults to dev.
pub environment: Option<String>,
/// Exporter to use. Defaults to `otlp-file`.
pub exporter: Option<OtelExporterKind>,
}
/// Effective OTEL settings after defaults are applied.
#[derive(Debug, Clone, PartialEq)]
pub struct OtelConfig {
pub log_user_prompt: bool,
pub environment: String,
pub exporter: OtelExporterKind,
}
impl Default for OtelConfig {
fn default() -> Self {
OtelConfig {
log_user_prompt: false,
environment: DEFAULT_OTEL_ENVIRONMENT.to_owned(),
exporter: OtelExporterKind::None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(untagged)]
pub enum Notifications {
codex-rs/core/src/custom_prompts.rs
+100 -1
View File
@@ -63,16 +63,88 @@ pub async fn discover_prompts_in_excluding(
Ok(s) => s,
Err(_) => continue,
};
let (description, argument_hint, body) = parse_frontmatter(&content);
out.push(CustomPrompt {
name,
path,
content,
content: body,
description,
argument_hint,
});
}
out.sort_by(|a, b| a.name.cmp(&b.name));
out
}
/// Parse optional YAML-like frontmatter at the beginning of `content`.
/// Supported keys:
/// - `description`: short description shown in the slash popup
/// - `argument-hint` or `argument_hint`: brief hint string shown after the description
/// Returns (description, argument_hint, body_without_frontmatter).
fn parse_frontmatter(content: &str) -> (Option<String>, Option<String>, String) {
let mut segments = content.split_inclusive('\n');
let Some(first_segment) = segments.next() else {
return (None, None, String::new());
};
let first_line = first_segment.trim_end_matches(['\r', '\n']);
if first_line.trim() != "---" {
return (None, None, content.to_string());
}
let mut desc: Option<String> = None;
let mut hint: Option<String> = None;
let mut frontmatter_closed = false;
let mut consumed = first_segment.len();
for segment in segments {
let line = segment.trim_end_matches(['\r', '\n']);
let trimmed = line.trim();
if trimmed == "---" {
frontmatter_closed = true;
consumed += segment.len();
break;
}
if trimmed.is_empty() || trimmed.starts_with('#') {
consumed += segment.len();
continue;
}
if let Some((k, v)) = trimmed.split_once(':') {
let key = k.trim().to_ascii_lowercase();
let mut val = v.trim().to_string();
if val.len() >= 2 {
let bytes = val.as_bytes();
let first = bytes[0];
let last = bytes[bytes.len() - 1];
if (first == b'\"' && last == b'\"') || (first == b'\'' && last == b'\'') {
val = val[1..val.len().saturating_sub(1)].to_string();
}
}
match key.as_str() {
"description" => desc = Some(val),
"argument-hint" | "argument_hint" => hint = Some(val),
_ => {}
}
}
consumed += segment.len();
}
if !frontmatter_closed {
// Unterminated frontmatter: treat input as-is.
return (None, None, content.to_string());
}
let body = if consumed >= content.len() {
String::new()
} else {
content[consumed..].to_string()
};
(desc, hint, body)
}
#[cfg(test)]
mod tests {
use super::*;
@@ -124,4 +196,31 @@ mod tests {
let names: Vec<String> = found.into_iter().map(|e| e.name).collect();
assert_eq!(names, vec!["good"]);
}
#[tokio::test]
async fn parses_frontmatter_and_strips_from_body() {
let tmp = tempdir().expect("create TempDir");
let dir = tmp.path();
let file = dir.join("withmeta.md");
let text = "---\nname: ignored\ndescription: \"Quick review command\"\nargument-hint: \"[file] [priority]\"\n---\nActual body with $1 and $ARGUMENTS";
fs::write(&file, text).unwrap();
let found = discover_prompts_in(dir).await;
assert_eq!(found.len(), 1);
let p = &found[0];
assert_eq!(p.name, "withmeta");
assert_eq!(p.description.as_deref(), Some("Quick review command"));
assert_eq!(p.argument_hint.as_deref(), Some("[file] [priority]"));
// Body should not include the frontmatter delimiters.
assert_eq!(p.content, "Actual body with $1 and $ARGUMENTS");
}
#[test]
fn parse_frontmatter_preserves_body_newlines() {
let content = "---\r\ndescription: \"Line endings\"\r\nargument_hint: \"[arg]\"\r\n---\r\nFirst line\r\nSecond line\r\n";
let (desc, hint, body) = parse_frontmatter(content);
assert_eq!(desc.as_deref(), Some("Line endings"));
assert_eq!(hint.as_deref(), Some("[arg]"));
assert_eq!(body, "First line\r\nSecond line\r\n");
}
}
codex-rs/core/src/default_client.rs
+33 -5
View File
@@ -2,6 +2,7 @@ use crate::spawn::CODEX_SANDBOX_ENV_VAR;
use reqwest::header::HeaderValue;
use std::sync::LazyLock;
use std::sync::Mutex;
use std::sync::OnceLock;
/// Set this to add a suffix to the User-Agent string.
///
@@ -26,8 +27,15 @@ pub struct Originator {
pub value: String,
pub header_value: HeaderValue,
}
static ORIGINATOR: OnceLock<Originator> = OnceLock::new();
pub static ORIGINATOR: LazyLock<Originator> = LazyLock::new(|| {
#[derive(Debug)]
pub enum SetOriginatorError {
InvalidHeaderValue,
AlreadyInitialized,
}
fn init_originator_from_env() -> Originator {
let default = "codex_cli_rs";
let value = std::env::var(CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR)
.unwrap_or_else(|_| default.to_string());
@@ -45,14 +53,34 @@ pub static ORIGINATOR: LazyLock<Originator> = LazyLock::new(|| {
}
}
}
});
}
fn build_originator(value: String) -> Result<Originator, SetOriginatorError> {
let header_value =
HeaderValue::from_str(&value).map_err(|_| SetOriginatorError::InvalidHeaderValue)?;
Ok(Originator {
value,
header_value,
})
}
pub fn set_default_originator(value: &str) -> Result<(), SetOriginatorError> {
let originator = build_originator(value.to_string())?;
ORIGINATOR
.set(originator)
.map_err(|_| SetOriginatorError::AlreadyInitialized)
}
pub fn originator() -> &'static Originator {
ORIGINATOR.get_or_init(init_originator_from_env)
}
pub fn get_codex_user_agent() -> String {
let build_version = env!("CARGO_PKG_VERSION");
let os_info = os_info::get();
let prefix = format!(
"{}/{build_version} ({} {}; {}) {}",
ORIGINATOR.value.as_str(),
originator().value.as_str(),
os_info.os_type(),
os_info.version(),
os_info.architecture().unwrap_or("unknown"),
@@ -100,7 +128,7 @@ fn sanitize_user_agent(candidate: String, fallback: &str) -> String {
tracing::warn!(
"Falling back to default Codex originator because base user agent string is invalid"
);
ORIGINATOR.value.clone()
originator().value.clone()
}
}
@@ -109,7 +137,7 @@ pub fn create_client() -> reqwest::Client {
use reqwest::header::HeaderMap;
let mut headers = HeaderMap::new();
headers.insert("originator", ORIGINATOR.header_value.clone());
headers.insert("originator", originator().header_value.clone());
let ua = get_codex_user_agent();
let mut builder = reqwest::Client::builder()
codex-rs/core/src/lib.rs
+2
View File
@@ -104,3 +104,5 @@ pub use codex_protocol::models::LocalShellExecAction;
pub use codex_protocol::models::LocalShellStatus;
pub use codex_protocol::models::ReasoningItemContent;
pub use codex_protocol::models::ResponseItem;
pub mod otel_init;
codex-rs/core/src/mcp_connection_manager.rs
+1 -2
View File
@@ -196,8 +196,7 @@ impl McpConnectionManager {
// Validate server name before spawning
if !is_valid_mcp_server_name(&server_name) {
let error = anyhow::anyhow!(
"invalid server name '{}': must match pattern ^[a-zA-Z0-9_-]+$",
server_name
"invalid server name '{server_name}': must match pattern ^[a-zA-Z0-9_-]+$"
);
errors.insert(server_name, error);
continue;
codex-rs/core/src/otel_init.rs
+61
View File
@@ -0,0 +1,61 @@
use crate::config::Config;
use crate::config_types::OtelExporterKind as Kind;
use crate::config_types::OtelHttpProtocol as Protocol;
use crate::default_client::originator;
use codex_otel::config::OtelExporter;
use codex_otel::config::OtelHttpProtocol;
use codex_otel::config::OtelSettings;
use codex_otel::otel_provider::OtelProvider;
use std::error::Error;
/// Build an OpenTelemetry provider from the app Config.
///
/// Returns `None` when OTEL export is disabled.
pub fn build_provider(
config: &Config,
service_version: &str,
) -> Result<Option<OtelProvider>, Box<dyn Error>> {
let exporter = match &config.otel.exporter {
Kind::None => OtelExporter::None,
Kind::OtlpHttp {
endpoint,
headers,
protocol,
} => {
let protocol = match protocol {
Protocol::Json => OtelHttpProtocol::Json,
Protocol::Binary => OtelHttpProtocol::Binary,
};
OtelExporter::OtlpHttp {
endpoint: endpoint.clone(),
headers: headers
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
protocol,
}
}
Kind::OtlpGrpc { endpoint, headers } => OtelExporter::OtlpGrpc {
endpoint: endpoint.clone(),
headers: headers
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
},
};
OtelProvider::from(&OtelSettings {
service_name: originator().value.to_owned(),
service_version: service_version.to_string(),
codex_home: config.codex_home.clone(),
environment: config.otel.environment.to_string(),
exporter,
})
}
/// Filter predicate for exporting only Codex-owned events via OTEL.
/// Keeps events that originated from codex_otel module
pub fn codex_export_filter(meta: &tracing::Metadata<'_>) -> bool {
meta.target().starts_with("codex_otel")
}
codex-rs/core/src/rollout/list.rs
+91 -10
View File
@@ -36,13 +36,16 @@ pub struct ConversationsPage {
pub struct ConversationItem {
/// Absolute path to the rollout file.
pub path: PathBuf,
/// First up to 5 JSONL records parsed as JSON (includes meta line).
/// First up to `HEAD_RECORD_LIMIT` JSONL records parsed as JSON (includes meta line).
pub head: Vec<serde_json::Value>,
/// Last up to `TAIL_RECORD_LIMIT` JSONL response records parsed as JSON.
pub tail: Vec<serde_json::Value>,
}
/// Hard cap to bound worst‑case work per request.
const MAX_SCAN_FILES: usize = 100;
const HEAD_RECORD_LIMIT: usize = 10;
const TAIL_RECORD_LIMIT: usize = 10;
/// Pagination cursor identifying a file by timestamp and UUID.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -176,13 +179,13 @@ async fn traverse_directories_for_paths(
}
// Read head and simultaneously detect message events within the same
// first N JSONL records to avoid a second file read.
let (head, saw_session_meta, saw_user_event) =
read_head_and_flags(&path, HEAD_RECORD_LIMIT)
let (head, tail, saw_session_meta, saw_user_event) =
read_head_and_tail(&path, HEAD_RECORD_LIMIT, TAIL_RECORD_LIMIT)
.await
.unwrap_or((Vec::new(), false, false));
.unwrap_or((Vec::new(), Vec::new(), false, false));
// Apply filters: must have session meta and at least one user message event
if saw_session_meta && saw_user_event {
items.push(ConversationItem { path, head });
items.push(ConversationItem { path, head, tail });
}
}
}
@@ -286,10 +289,11 @@ fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDateTime, Uui
Some((ts, uuid))
}
async fn read_head_and_flags(
async fn read_head_and_tail(
path: &Path,
max_records: usize,
) -> io::Result<(Vec<serde_json::Value>, bool, bool)> {
head_limit: usize,
tail_limit: usize,
) -> io::Result<(Vec<serde_json::Value>, Vec<serde_json::Value>, bool, bool)> {
use tokio::io::AsyncBufReadExt;
let file = tokio::fs::File::open(path).await?;
@@ -299,7 +303,7 @@ async fn read_head_and_flags(
let mut saw_session_meta = false;
let mut saw_user_event = false;
while head.len() < max_records {
while head.len() < head_limit {
let line_opt = lines.next_line().await?;
let Some(line) = line_opt else { break };
let trimmed = line.trim();
@@ -336,7 +340,84 @@ async fn read_head_and_flags(
}
}
Ok((head, saw_session_meta, saw_user_event))
let tail = if tail_limit == 0 {
Vec::new()
} else {
read_tail_records(path, tail_limit).await?
};
Ok((head, tail, saw_session_meta, saw_user_event))
}
async fn read_tail_records(path: &Path, max_records: usize) -> io::Result<Vec<serde_json::Value>> {
use std::io::SeekFrom;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncSeekExt;
if max_records == 0 {
return Ok(Vec::new());
}
const CHUNK_SIZE: usize = 8192;
let mut file = tokio::fs::File::open(path).await?;
let mut pos = file.seek(SeekFrom::End(0)).await?;
if pos == 0 {
return Ok(Vec::new());
}
let mut buffer: Vec<u8> = Vec::new();
loop {
let slice_start = match (pos > 0, buffer.iter().position(|&b| b == b'\n')) {
(true, Some(idx)) => idx + 1,
_ => 0,
};
let tail = collect_last_response_values(&buffer[slice_start..], max_records);
if tail.len() >= max_records || pos == 0 {
return Ok(tail);
}
let read_size = CHUNK_SIZE.min(pos as usize);
if read_size == 0 {
return Ok(tail);
}
pos -= read_size as u64;
file.seek(SeekFrom::Start(pos)).await?;
let mut chunk = vec![0; read_size];
file.read_exact(&mut chunk).await?;
chunk.extend_from_slice(&buffer);
buffer = chunk;
}
}
fn collect_last_response_values(buffer: &[u8], max_records: usize) -> Vec<serde_json::Value> {
use std::borrow::Cow;
if buffer.is_empty() || max_records == 0 {
return Vec::new();
}
let text: Cow<'_, str> = String::from_utf8_lossy(buffer);
let mut collected_rev: Vec<serde_json::Value> = Vec::new();
for line in text.lines().rev() {
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let parsed: serde_json::Result<RolloutLine> = serde_json::from_str(trimmed);
let Ok(rollout_line) = parsed else { continue };
if let RolloutItem::ResponseItem(item) = rollout_line.item
&& let Ok(val) = serde_json::to_value(item)
{
collected_rev.push(val);
if collected_rev.len() == max_records {
break;
}
}
}
collected_rev.reverse();
collected_rev
}
/// Locate a recorded conversation rollout file by its UUID string using the existing
codex-rs/core/src/rollout/recorder.rs
+2 -2
View File
@@ -24,7 +24,7 @@ use super::list::Cursor;
use super::list::get_conversations;
use super::policy::is_persisted_response_item;
use crate::config::Config;
use crate::default_client::ORIGINATOR;
use crate::default_client::originator;
use crate::git_info::collect_git_info;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::ResumedHistory;
@@ -124,7 +124,7 @@ impl RolloutRecorder {
id: session_id,
timestamp,
cwd: config.cwd.clone(),
originator: ORIGINATOR.value.clone(),
originator: originator().value.clone(),
cli_version: env!("CARGO_PKG_VERSION").to_string(),
instructions,
}),
codex-rs/core/src/rollout/tests.rs
+268
View File
@@ -17,6 +17,18 @@ use crate::rollout::list::ConversationsPage;
use crate::rollout::list::Cursor;
use crate::rollout::list::get_conversation;
use crate::rollout::list::get_conversations;
use anyhow::Result;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InputMessageKind;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::UserMessageEvent;
fn write_session_file(
root: &Path,
@@ -146,14 +158,17 @@ async fn test_list_conversations_latest_first() {
ConversationItem {
path: p1,
head: head_3,
tail: Vec::new(),
},
ConversationItem {
path: p2,
head: head_2,
tail: Vec::new(),
},
ConversationItem {
path: p3,
head: head_1,
tail: Vec::new(),
},
],
next_cursor: Some(expected_cursor),
@@ -219,10 +234,12 @@ async fn test_pagination_cursor() {
ConversationItem {
path: p5,
head: head_5,
tail: Vec::new(),
},
ConversationItem {
path: p4,
head: head_4,
tail: Vec::new(),
},
],
next_cursor: Some(expected_cursor1.clone()),
@@ -269,10 +286,12 @@ async fn test_pagination_cursor() {
ConversationItem {
path: p3,
head: head_3,
tail: Vec::new(),
},
ConversationItem {
path: p2,
head: head_2,
tail: Vec::new(),
},
],
next_cursor: Some(expected_cursor2.clone()),
@@ -304,6 +323,7 @@ async fn test_pagination_cursor() {
items: vec![ConversationItem {
path: p1,
head: head_1,
tail: Vec::new(),
}],
next_cursor: Some(expected_cursor3),
num_scanned_files: 5, // scanned 05, 04 (anchor), 03, 02 (anchor), 01
@@ -346,6 +366,7 @@ async fn test_get_conversation_contents() {
items: vec![ConversationItem {
path: expected_path,
head: expected_head,
tail: Vec::new(),
}],
next_cursor: Some(expected_cursor),
num_scanned_files: 1,
@@ -366,6 +387,250 @@ async fn test_get_conversation_contents() {
assert_eq!(content, expected_content);
}
#[tokio::test]
async fn test_tail_includes_last_response_items() -> Result<()> {
let temp = TempDir::new().unwrap();
let home = temp.path();
let ts = "2025-06-01T08-00-00";
let uuid = Uuid::from_u128(42);
let day_dir = home.join("sessions").join("2025").join("06").join("01");
fs::create_dir_all(&day_dir)?;
let file_path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let mut file = File::create(&file_path)?;
let conversation_id = ConversationId::from_string(&uuid.to_string())?;
let meta_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
id: conversation_id,
timestamp: ts.to_string(),
instructions: None,
cwd: ".".into(),
originator: "test_originator".into(),
cli_version: "test_version".into(),
},
git: None,
}),
};
writeln!(file, "{}", serde_json::to_string(&meta_line)?)?;
let user_event_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "hello".into(),
kind: Some(InputMessageKind::Plain),
images: None,
})),
};
writeln!(file, "{}", serde_json::to_string(&user_event_line)?)?;
let total_messages = 12usize;
for idx in 0..total_messages {
let response_line = RolloutLine {
timestamp: format!("{ts}-{idx:02}"),
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("reply-{idx}"),
}],
}),
};
writeln!(file, "{}", serde_json::to_string(&response_line)?)?;
}
drop(file);
let page = get_conversations(home, 1, None).await?;
let item = page.items.first().expect("conversation item");
let tail_len = item.tail.len();
assert_eq!(tail_len, 10usize.min(total_messages));
let expected: Vec<serde_json::Value> = (total_messages - tail_len..total_messages)
.map(|idx| {
serde_json::to_value(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("reply-{idx}"),
}],
})
.expect("serialize response item")
})
.collect();
assert_eq!(item.tail, expected);
Ok(())
}
#[tokio::test]
async fn test_tail_handles_short_sessions() -> Result<()> {
let temp = TempDir::new().unwrap();
let home = temp.path();
let ts = "2025-06-02T08-30-00";
let uuid = Uuid::from_u128(7);
let day_dir = home.join("sessions").join("2025").join("06").join("02");
fs::create_dir_all(&day_dir)?;
let file_path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let mut file = File::create(&file_path)?;
let conversation_id = ConversationId::from_string(&uuid.to_string())?;
let meta_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
id: conversation_id,
timestamp: ts.to_string(),
instructions: None,
cwd: ".".into(),
originator: "test_originator".into(),
cli_version: "test_version".into(),
},
git: None,
}),
};
writeln!(file, "{}", serde_json::to_string(&meta_line)?)?;
let user_event_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "hi".into(),
kind: Some(InputMessageKind::Plain),
images: None,
})),
};
writeln!(file, "{}", serde_json::to_string(&user_event_line)?)?;
for idx in 0..3 {
let response_line = RolloutLine {
timestamp: format!("{ts}-{idx:02}"),
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("short-{idx}"),
}],
}),
};
writeln!(file, "{}", serde_json::to_string(&response_line)?)?;
}
drop(file);
let page = get_conversations(home, 1, None).await?;
let tail = &page.items.first().expect("conversation item").tail;
assert_eq!(tail.len(), 3);
let expected: Vec<serde_json::Value> = (0..3)
.map(|idx| {
serde_json::to_value(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("short-{idx}"),
}],
})
.expect("serialize response item")
})
.collect();
assert_eq!(tail, &expected);
Ok(())
}
#[tokio::test]
async fn test_tail_skips_trailing_non_responses() -> Result<()> {
let temp = TempDir::new().unwrap();
let home = temp.path();
let ts = "2025-06-03T10-00-00";
let uuid = Uuid::from_u128(11);
let day_dir = home.join("sessions").join("2025").join("06").join("03");
fs::create_dir_all(&day_dir)?;
let file_path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let mut file = File::create(&file_path)?;
let conversation_id = ConversationId::from_string(&uuid.to_string())?;
let meta_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
id: conversation_id,
timestamp: ts.to_string(),
instructions: None,
cwd: ".".into(),
originator: "test_originator".into(),
cli_version: "test_version".into(),
},
git: None,
}),
};
writeln!(file, "{}", serde_json::to_string(&meta_line)?)?;
let user_event_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "hello".into(),
kind: Some(InputMessageKind::Plain),
images: None,
})),
};
writeln!(file, "{}", serde_json::to_string(&user_event_line)?)?;
for idx in 0..4 {
let response_line = RolloutLine {
timestamp: format!("{ts}-{idx:02}"),
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("response-{idx}"),
}],
}),
};
writeln!(file, "{}", serde_json::to_string(&response_line)?)?;
}
let compacted_line = RolloutLine {
timestamp: format!("{ts}-compacted"),
item: RolloutItem::Compacted(CompactedItem {
message: "compacted".into(),
}),
};
writeln!(file, "{}", serde_json::to_string(&compacted_line)?)?;
let shutdown_event = RolloutLine {
timestamp: format!("{ts}-shutdown"),
item: RolloutItem::EventMsg(EventMsg::ShutdownComplete),
};
writeln!(file, "{}", serde_json::to_string(&shutdown_event)?)?;
drop(file);
let page = get_conversations(home, 1, None).await?;
let tail = &page.items.first().expect("conversation item").tail;
let expected: Vec<serde_json::Value> = (0..4)
.map(|idx| {
serde_json::to_value(ResponseItem::Message {
id: None,
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: format!("response-{idx}"),
}],
})
.expect("serialize response item")
})
.collect();
assert_eq!(tail, &expected);
Ok(())
}
#[tokio::test]
async fn test_stable_ordering_same_second_pagination() {
let temp = TempDir::new().unwrap();
@@ -410,10 +675,12 @@ async fn test_stable_ordering_same_second_pagination() {
ConversationItem {
path: p3,
head: head(u3),
tail: Vec::new(),
},
ConversationItem {
path: p2,
head: head(u2),
tail: Vec::new(),
},
],
next_cursor: Some(expected_cursor1.clone()),
@@ -436,6 +703,7 @@ async fn test_stable_ordering_same_second_pagination() {
items: vec![ConversationItem {
path: p1,
head: head(u1),
tail: Vec::new(),
}],
next_cursor: Some(expected_cursor2),
num_scanned_files: 3, // scanned u3, u2 (anchor), u1
codex-rs/core/src/safety.rs
+28 -7
View File
@@ -15,9 +15,14 @@ use crate::protocol::SandboxPolicy;
#[derive(Debug, PartialEq)]
pub enum SafetyCheck {
AutoApprove { sandbox_type: SandboxType },
AutoApprove {
sandbox_type: SandboxType,
user_explicitly_approved: bool,
},
AskUser,
Reject { reason: String },
Reject {
reason: String,
},
}
pub fn assess_patch_safety(
@@ -54,12 +59,16 @@ pub fn assess_patch_safety(
// fall back to asking the user because the patch may touch arbitrary
// paths outside the project.
match get_platform_sandbox() {
Some(sandbox_type) => SafetyCheck::AutoApprove { sandbox_type },
Some(sandbox_type) => SafetyCheck::AutoApprove {
sandbox_type,
user_explicitly_approved: false,
},
None if sandbox_policy == &SandboxPolicy::DangerFullAccess => {
// If the user has explicitly requested DangerFullAccess, then
// we can auto-approve even without a sandbox.
SafetyCheck::AutoApprove {
sandbox_type: SandboxType::None,
user_explicitly_approved: false,
}
}
None => SafetyCheck::AskUser,
@@ -118,6 +127,7 @@ pub fn assess_command_safety(
if is_known_safe_command(command) || approved.contains(command) {
return SafetyCheck::AutoApprove {
sandbox_type: SandboxType::None,
user_explicitly_approved: false,
};
}
@@ -143,13 +153,17 @@ pub(crate) fn assess_safety_for_untrusted_command(
| (Never, DangerFullAccess)
| (OnRequest, DangerFullAccess) => SafetyCheck::AutoApprove {
sandbox_type: SandboxType::None,
user_explicitly_approved: false,
},
(OnRequest, ReadOnly) | (OnRequest, WorkspaceWrite { .. }) => {
if with_escalated_permissions {
SafetyCheck::AskUser
} else {
match get_platform_sandbox() {
Some(sandbox_type) => SafetyCheck::AutoApprove { sandbox_type },
Some(sandbox_type) => SafetyCheck::AutoApprove {
sandbox_type,
user_explicitly_approved: false,
},
// Fall back to asking since the command is untrusted and
// we do not have a sandbox available
None => SafetyCheck::AskUser,
@@ -161,7 +175,10 @@ pub(crate) fn assess_safety_for_untrusted_command(
| (OnFailure, ReadOnly)
| (OnFailure, WorkspaceWrite { .. }) => {
match get_platform_sandbox() {
Some(sandbox_type) => SafetyCheck::AutoApprove { sandbox_type },
Some(sandbox_type) => SafetyCheck::AutoApprove {
sandbox_type,
user_explicitly_approved: false,
},
None => {
if matches!(approval_policy, OnFailure) {
// Since the command is not trusted, even though the
@@ -362,7 +379,8 @@ mod tests {
assert_eq!(
safety_check,
SafetyCheck::AutoApprove {
sandbox_type: SandboxType::None
sandbox_type: SandboxType::None,
user_explicitly_approved: false,
}
);
}
@@ -409,7 +427,10 @@ mod tests {
);
let expected = match get_platform_sandbox() {
Some(sandbox_type) => SafetyCheck::AutoApprove { sandbox_type },
Some(sandbox_type) => SafetyCheck::AutoApprove {
sandbox_type,
user_explicitly_approved: false,
},
None => SafetyCheck::AskUser,
};
assert_eq!(safety_check, expected);