openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-14 21:40:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge origin/main into rhan/surface-updates and resolve guardian tests conflict

Browse Source
This commit is contained in:
Roy Han
2026-03-16 17:42:39 -07:00
parent 445a6433a3 b77fe8fefe
commit 56b0fe29fd
1516 changed files with 150231 additions and 6961 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+137 -101
View File
@@ -41,6 +41,7 @@ use crate::realtime_conversation::handle_close as handle_realtime_conversation_c
use crate::realtime_conversation::handle_start as handle_realtime_conversation_start;
use crate::realtime_conversation::handle_text as handle_realtime_conversation_text;
use crate::rollout::session_index;
use crate::skills::render_skills_section;
use crate::stream_events_utils::HandleOutputCtx;
use crate::stream_events_utils::handle_non_tool_response_item;
use crate::stream_events_utils::handle_output_item_done;
@@ -203,6 +204,7 @@ use crate::feedback_tags;
use crate::file_watcher::FileWatcher;
use crate::file_watcher::FileWatcherEvent;
use crate::git_info::get_git_repo_root;
use crate::guardian::GuardianReviewSessionManager;
use crate::instructions::UserInstructions;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::mcp::McpManager;
@@ -221,6 +223,7 @@ use crate::mentions::collect_tool_mentions_from_messages;
use crate::network_policy_decision::execpolicy_network_rule_amendment;
use crate::plugins::PluginsManager;
use crate::plugins::build_plugin_injections;
use crate::plugins::render_plugins_section;
use crate::project_doc::get_user_instructions;
use crate::protocol::AgentMessageContentDeltaEvent;
use crate::protocol::AgentReasoningSectionBreakEvent;
@@ -375,6 +378,7 @@ pub(crate) const INITIAL_SUBMIT_ID: &str = "";
pub(crate) const SUBMISSION_CHANNEL_CAPACITY: usize = 512;
const CYBER_VERIFY_URL: &str = "https://chatgpt.com/cyber";
const CYBER_SAFETY_URL: &str = "https://developers.openai.com/codex/concepts/cyber-safety";
const DIRECT_APP_TOOL_EXPOSURE_THRESHOLD: usize = 100;
impl Codex {
/// Spawn a new [`Codex`] and initialize the session.
@@ -423,7 +427,6 @@ impl Codex {
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_event, rx_event) = async_channel::unbounded();
let loaded_plugins = plugins_manager.plugins_for_config(&config);
let loaded_skills = skills_manager.skills_for_config(&config);
for err in &loaded_skills.errors {
@@ -469,16 +472,9 @@ impl Codex {
config.startup_warnings.push(message);
}
let allowed_skills_for_implicit_invocation =
loaded_skills.allowed_skills_for_implicit_invocation();
let user_instructions = get_user_instructions(
&config,
Some(&allowed_skills_for_implicit_invocation),
Some(loaded_plugins.capability_summaries()),
)
.await;
let user_instructions = get_user_instructions(&config).await;
let exec_policy = if crate::guardian::is_guardian_subagent_source(&session_source) {
let exec_policy = if crate::guardian::is_guardian_reviewer_source(&session_source) {
// Guardian review should rely on the built-in shell safety checks,
// not on caller-provided exec-policy rules that could shape the
// reviewer or silently auto-approve commands.
@@ -634,7 +630,7 @@ impl Codex {
/// Submit the `op` wrapped in a `Submission` with a unique ID.
pub async fn submit(&self, op: Op) -> CodexResult<String> {
self.submit_with_trace(op, None).await
self.submit_with_trace(op, /*trace*/ None).await
}
pub async fn submit_with_trace(
@@ -753,6 +749,7 @@ pub(crate) struct Session {
pending_mcp_server_refresh_config: Mutex<Option<McpServerRefreshConfig>>,
pub(crate) conversation: Arc<RealtimeConversationManager>,
pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
pub(crate) guardian_review_session: GuardianReviewSessionManager,
pub(crate) services: SessionServices,
js_repl: Arc<JsReplHandle>,
next_internal_sub_id: AtomicU64,
@@ -858,9 +855,11 @@ impl TurnContext {
};
config.model_reasoning_effort = reasoning_effort;
let collaboration_mode =
self.collaboration_mode
.with_updates(Some(model.clone()), Some(reasoning_effort), None);
let collaboration_mode = self.collaboration_mode.with_updates(
Some(model.clone()),
Some(reasoning_effort),
/*developer_instructions*/ None,
);
let features = self.features.clone();
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
@@ -1163,12 +1162,22 @@ impl Session {
async fn start_managed_network_proxy(
spec: &crate::config::NetworkProxySpec,
exec_policy: &codex_execpolicy::Policy,
sandbox_policy: &SandboxPolicy,
network_policy_decider: Option<Arc<dyn codex_network_proxy::NetworkPolicyDecider>>,
blocked_request_observer: Option<Arc<dyn codex_network_proxy::BlockedRequestObserver>>,
managed_network_requirements_enabled: bool,
audit_metadata: NetworkProxyAuditMetadata,
) -> anyhow::Result<(StartedNetworkProxy, SessionNetworkProxyRuntime)> {
let spec = spec
.with_exec_policy_network_rules(exec_policy)
.map_err(|err| {
tracing::warn!(
"failed to apply execpolicy network rules to managed proxy; continuing with configured network policy: {err}"
);
err
})
.unwrap_or_else(|_| spec.clone());
let network_proxy = spec
.start_proxy(
sandbox_policy,
@@ -1194,6 +1203,7 @@ impl Session {
// todo(aibrahim): store this state somewhere else so we don't need to mut config
let config = session_configuration.original_config_do_not_use.clone();
let mut per_turn_config = (*config).clone();
per_turn_config.cwd = session_configuration.cwd.clone();
per_turn_config.model_reasoning_effort =
session_configuration.collaboration_mode.reasoning_effort();
per_turn_config.model_reasoning_summary = session_configuration.model_reasoning_summary;
@@ -1582,7 +1592,7 @@ impl Session {
config.features.emit_metrics(&session_telemetry);
session_telemetry.counter(
THREAD_STARTED_METRIC,
1,
/*inc*/ 1,
&[(
"is_git",
if get_git_repo_root(&session_configuration.cwd).is_some() {
@@ -1692,8 +1702,10 @@ impl Session {
});
let (network_proxy, session_network_proxy) =
if let Some(spec) = config.permissions.network.as_ref() {
let current_exec_policy = exec_policy.current();
let (network_proxy, session_network_proxy) = Self::start_managed_network_proxy(
spec,
current_exec_policy.as_ref(),
config.permissions.sandbox_policy.get(),
network_policy_decider.as_ref().map(Arc::clone),
blocked_request_observer.as_ref().map(Arc::clone),
@@ -1712,7 +1724,8 @@ impl Session {
(None, None)
};
let mut hook_shell_argv = default_shell.derive_exec_args("", false);
let mut hook_shell_argv =
default_shell.derive_exec_args("", /*use_login_shell*/ false);
let hook_shell_program = hook_shell_argv.remove(0);
let _ = hook_shell_argv.pop();
let hooks = Hooks::new(HooksConfig {
@@ -1803,6 +1816,7 @@ impl Session {
pending_mcp_server_refresh_config: Mutex::new(None),
conversation: Arc::new(RealtimeConversationManager::new()),
active_turn: Mutex::new(None),
guardian_review_session: GuardianReviewSessionManager::default(),
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
@@ -2061,21 +2075,14 @@ impl Session {
InitialHistory::New => {
// Defer initial context insertion until the first real turn starts so
// turn/start overrides can be merged before we write model-visible context.
self.set_previous_turn_settings(None).await;
self.set_previous_turn_settings(/*previous_turn_settings*/ None)
.await;
}
InitialHistory::Resumed(resumed_history) => {
let rollout_items = resumed_history.history;
let reconstructed_rollout = self
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
let previous_turn_settings = self
.apply_rollout_reconstruction(&turn_context, &rollout_items)
.await;
let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone();
self.set_previous_turn_settings(previous_turn_settings.clone())
.await;
{
let mut state = self.state.lock().await;
state.set_reference_context_item(reconstructed_rollout.reference_context_item);
}
// If resuming, warn when the last recorded model differs from the current one.
let curr: &str = turn_context.model_info.slug.as_str();
@@ -2097,13 +2104,6 @@ impl Session {
.await;
}
// Always add response items to conversation history
let reconstructed_history = reconstructed_rollout.history;
if !reconstructed_history.is_empty() {
self.record_into_history(&reconstructed_history, &turn_context)
.await;
}
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
@@ -2118,26 +2118,8 @@ impl Session {
}
}
InitialHistory::Forked(rollout_items) => {
let reconstructed_rollout = self
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
self.apply_rollout_reconstruction(&turn_context, &rollout_items)
.await;
self.set_previous_turn_settings(
reconstructed_rollout.previous_turn_settings.clone(),
)
.await;
{
let mut state = self.state.lock().await;
state.set_reference_context_item(
reconstructed_rollout.reference_context_item.clone(),
);
}
// Always add response items to conversation history
let reconstructed_history = reconstructed_rollout.history;
if !reconstructed_history.is_empty() {
self.record_into_history(&reconstructed_history, &turn_context)
.await;
}
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
@@ -2171,6 +2153,25 @@ impl Session {
}
}
async fn apply_rollout_reconstruction(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
) -> Option<PreviousTurnSettings> {
let reconstructed_rollout = self
.reconstruct_history_from_rollout(turn_context, rollout_items)
.await;
let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone();
self.replace_history(
reconstructed_rollout.history,
reconstructed_rollout.reference_context_item,
)
.await;
self.set_previous_turn_settings(previous_turn_settings.clone())
.await;
previous_turn_settings
}
fn last_token_info_from_rollout(rollout_items: &[RolloutItem]) -> Option<TokenUsageInfo> {
rollout_items.iter().rev().find_map(|item| match item {
RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(),
@@ -2359,8 +2360,7 @@ impl Session {
let skills_outcome = Arc::new(
self.services
.skills_manager
.skills_for_cwd(&session_configuration.cwd, false)
.await,
.skills_for_config(&per_turn_config),
);
let mut turn_context: TurnContext = Self::make_turn_context(
Some(Arc::clone(&self.services.auth_manager)),
@@ -2453,7 +2453,7 @@ impl Session {
startup_turn_context.as_ref(),
&[],
&HashSet::new(),
None,
/*skills_outcome*/ None,
&startup_cancellation_token,
)
.await?;
@@ -2539,8 +2539,13 @@ impl Session {
let state = self.state.lock().await;
state.session_configuration.clone()
};
self.new_turn_from_configuration(sub_id, session_configuration, None, false)
.await
self.new_turn_from_configuration(
sub_id,
session_configuration,
/*final_output_json_schema*/ None,
/*sandbox_policy_changed*/ false,
)
.await
}
async fn build_settings_update_items(
@@ -2613,31 +2618,17 @@ impl Session {
}
pub(crate) async fn send_event_raw(&self, event: Event) {
// Record the last known agent status.
if let Some(status) = agent_status_from_event(&event.msg) {
self.agent_status.send_replace(status);
}
// Persist the event into rollout (recorder filters as needed)
let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())];
self.persist_rollout_items(&rollout_items).await;
if let Err(e) = self.tx_event.send(event).await {
debug!("dropping event because channel is closed: {e}");
}
self.deliver_event_raw(event).await;
}
/// Persist the event to the rollout file, flush it, and only then deliver it to clients.
///
/// Most events can be delivered immediately after queueing the rollout write, but some
/// clients (e.g. app-server thread/rollback) re-read the rollout file synchronously on
/// receipt of the event and depend on the marker already being visible on disk.
pub(crate) async fn send_event_raw_flushed(&self, event: Event) {
async fn deliver_event_raw(&self, event: Event) {
// Record the last known agent status.
if let Some(status) = agent_status_from_event(&event.msg) {
self.agent_status.send_replace(status);
}
self.persist_rollout_items(&[RolloutItem::EventMsg(event.msg.clone())])
.await;
self.flush_rollout().await;
if let Err(e) = self.tx_event.send(event).await {
debug!("dropping event because channel is closed: {e}");
}
@@ -3304,7 +3295,7 @@ impl Session {
pub(crate) async fn record_model_warning(&self, message: impl Into<String>, ctx: &TurnContext) {
self.services
.session_telemetry
.counter("codex.model_warning", 1, &[]);
.counter("codex.model_warning", /*inc*/ 1, &[]);
let item = ResponseItem::Message {
id: None,
role: "user".to_string(),
@@ -3465,7 +3456,7 @@ impl Session {
.into_text(),
);
let separate_guardian_developer_message =
crate::guardian::is_guardian_subagent_source(&session_source);
crate::guardian::is_guardian_reviewer_source(&session_source);
// Keep the guardian policy prompt out of the aggregated developer bundle so it
// stays isolated as its own top-level developer message for guardian subagents.
if !separate_guardian_developer_message
@@ -3516,6 +3507,21 @@ impl Session {
if turn_context.apps_enabled() {
developer_sections.push(render_apps_section());
}
let implicit_skills = turn_context
.turn_skills
.outcome
.allowed_skills_for_implicit_invocation();
if let Some(skills_section) = render_skills_section(&implicit_skills) {
developer_sections.push(skills_section);
}
let loaded_plugins = self
.services
.plugins_manager
.plugins_for_config(&turn_context.config);
if let Some(plugin_section) = render_plugins_section(loaded_plugins.capability_summaries())
{
developer_sections.push(plugin_section);
}
if turn_context.features.enabled(Feature::CodexGitCommit)
&& let Some(commit_message_instruction) = commit_message_trailer_instruction(
turn_context.config.commit_attribution.as_deref(),
@@ -3946,12 +3952,13 @@ impl Session {
server: &str,
tool: &str,
arguments: Option<serde_json::Value>,
meta: Option<serde_json::Value>,
) -> anyhow::Result<CallToolResult> {
self.services
.mcp_connection_manager
.read()
.await
.call_tool(server, tool, arguments)
.call_tool(server, tool, arguments, meta)
.await
}
@@ -4191,7 +4198,7 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
state.session_configuration.collaboration_mode.with_updates(
model.clone(),
effort,
None,
/*developer_instructions*/ None,
)
};
handlers::override_turn_context(
@@ -4357,6 +4364,9 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
break;
}
}
// Also drain cached guardian state if the submission loop exits because
// the channel closed without receiving an explicit shutdown op.
sess.guardian_review_session.shutdown().await;
debug!("Agent loop exited");
}
@@ -4538,7 +4548,9 @@ mod handlers {
current_context.session_telemetry.user_prompt(&items);
// Attempt to inject input into current task.
if let Err(SteerInputError::NoActiveTurn(items)) = sess.steer_input(items, None).await {
if let Err(SteerInputError::NoActiveTurn(items)) =
sess.steer_input(items, /*expected_turn_id*/ None).await
{
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
@@ -5073,29 +5085,22 @@ mod handlers {
};
let rollback_event = ThreadRolledBackEvent { num_turns };
let rollback_msg = EventMsg::ThreadRolledBack(rollback_event.clone());
let replay_items = initial_history
.get_rollout_items()
.into_iter()
.chain(std::iter::once(RolloutItem::EventMsg(
EventMsg::ThreadRolledBack(rollback_event.clone()),
)))
.chain(std::iter::once(RolloutItem::EventMsg(rollback_msg.clone())))
.collect::<Vec<_>>();
let reconstructed = sess
.reconstruct_history_from_rollout(turn_context.as_ref(), replay_items.as_slice())
sess.persist_rollout_items(&[RolloutItem::EventMsg(rollback_msg.clone())])
.await;
sess.replace_history(
reconstructed.history,
reconstructed.reference_context_item.clone(),
)
.await;
sess.set_previous_turn_settings(reconstructed.previous_turn_settings)
sess.flush_rollout().await;
sess.apply_rollout_reconstruction(turn_context.as_ref(), replay_items.as_slice())
.await;
sess.recompute_token_usage(turn_context.as_ref()).await;
sess.send_event_raw_flushed(Event {
sess.deliver_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::ThreadRolledBack(rollback_event),
msg: rollback_msg,
})
.await;
}
@@ -5173,6 +5178,7 @@ mod handlers {
.unified_exec_manager
.terminate_all_processes()
.await;
sess.guardian_review_session.shutdown().await;
info!("Shutting down Codex instance");
let history = sess.clone_history().await;
let turn_count = history
@@ -5289,7 +5295,7 @@ async fn spawn_review_thread(
sess.services.shell_zsh_path.as_ref(),
sess.services.main_execve_wrapper_exe.as_ref(),
)
.with_web_search_config(None)
.with_web_search_config(/*web_search_config*/ None)
.with_allow_login_shell(config.permissions.allow_login_shell)
.with_agent_roles(config.agent_roles.clone());
@@ -5972,7 +5978,7 @@ pub(crate) async fn run_turn(
}
Err(e) => {
info!("Turn error: {e:#}");
let event = EventMsg::Error(e.to_error_event(None));
let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
sess.send_event(&turn_context, event).await;
// let the user continue the conversation
break;
@@ -6236,9 +6242,25 @@ fn build_prompt(
turn_context: &TurnContext,
base_instructions: BaseInstructions,
) -> Prompt {
let deferred_dynamic_tools = turn_context
.dynamic_tools
.iter()
.filter(|tool| tool.defer_loading)
.map(|tool| tool.name.as_str())
.collect::<HashSet<_>>();
let tools = if deferred_dynamic_tools.is_empty() {
router.model_visible_specs()
} else {
router
.model_visible_specs()
.into_iter()
.filter(|spec| !deferred_dynamic_tools.contains(spec.name()))
.collect()
};
Prompt {
input,
tools: router.model_visible_specs(),
tools,
parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
base_instructions,
personality: turn_context.personality,
@@ -6463,8 +6485,6 @@ pub(crate) async fn built_tools(
None
};
// Keep the connector-grouped app view around for the router even though
// app tools only become prompt-visible after explicit selection/discovery.
let app_tools = connectors.as_ref().map(|connectors| {
filter_codex_apps_mcp_tools(&mcp_tools, connectors, &turn_context.config)
});
@@ -6491,6 +6511,21 @@ pub(crate) async fn built_tools(
mcp_tools = selected_mcp_tools;
}
// Expose app tools directly when tool_search is disabled, or when tool_search
// is enabled but the accessible app tool set stays below the direct-exposure threshold.
let expose_app_tools_directly = !turn_context.tools_config.search_tool
|| app_tools
.as_ref()
.is_some_and(|tools| tools.len() < DIRECT_APP_TOOL_EXPOSURE_THRESHOLD);
if expose_app_tools_directly && let Some(app_tools) = app_tools.as_ref() {
mcp_tools.extend(app_tools.clone());
}
let app_tools = if expose_app_tools_directly {
None
} else {
app_tools
};
Ok(Arc::new(ToolRouter::from_config(
&turn_context.tools_config,
ToolRouterParams {
@@ -7010,7 +7045,8 @@ async fn handle_assistant_item_done_in_plan_mode(
{
maybe_complete_plan_item_from_message(sess, turn_context, state, item).await;
if let Some(turn_item) = handle_non_tool_response_item(sess, turn_context, item, true).await
if let Some(turn_item) =
handle_non_tool_response_item(sess, turn_context, item, /*plan_mode*/ true).await
{
emit_turn_item_in_plan_mode(
sess,
@@ -7023,7 +7059,7 @@ async fn handle_assistant_item_done_in_plan_mode(
}
record_completed_response_item(sess, turn_context, item).await;
if let Some(agent_message) = last_assistant_message_from_item(item, true) {
if let Some(agent_message) = last_assistant_message_from_item(item, /*plan_mode*/ true) {
*last_agent_message = Some(agent_message);
}
return true;
@@ -7394,7 +7430,7 @@ async fn try_run_sampling_request(
pub(super) fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -> Option<String> {
for item in responses.iter().rev() {
if let Some(message) = last_assistant_message_from_item(item, false) {
if let Some(message) = last_assistant_message_from_item(item, /*plan_mode*/ false) {
return Some(message);
}
}