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

Follow-up to https://github.com/openai/codex/pull/18178. This handles the straightforward await-holding cleanup first, without enabling the Clippy lint yet; the long-term goal is to enable the lint in a follow-up once the remaining cases are resolved or explicitly justified.

Browse Source
This commit is contained in:
Michael Bolin
2026-04-16 14:36:00 -07:00
parent 8720b7bdce
commit 9b65a0cd60
14 changed files with 357 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+11 -9
View File
@@ -2016,9 +2016,10 @@ async fn complete_file_change_item(
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
let mut state = thread_state.lock().await;
state.turn_summary.file_change_started.remove(&item_id);
drop(state);
{
let mut state = thread_state.lock().await;
state.turn_summary.file_change_started.remove(&item_id);
}
let notification = ItemCompletedNotification {
thread_id: conversation_id.to_string(),
@@ -2087,12 +2088,13 @@ async fn complete_command_execution_item(
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
let mut state = thread_state.lock().await;
let should_emit = state
.turn_summary
.command_execution_started
.remove(&item_id);
drop(state);
let should_emit = {
let mut state = thread_state.lock().await;
state
.turn_summary
.command_execution_started
.remove(&item_id)
};
if !should_emit {
return;
}
codex-rs/app-server/src/codex_message_processor.rs
+10 -11
View File
@@ -3497,9 +3497,10 @@ impl CodexMessageProcessor {
// No ThreadRollback event will arrive if an error occurs.
// Clean up and reply immediately.
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
let mut thread_state = thread_state.lock().await;
thread_state.pending_rollbacks = None;
drop(thread_state);
{
let mut thread_state = thread_state.lock().await;
thread_state.pending_rollbacks = None;
}
self.send_internal_error(request, format!("failed to start rollback: {err}"))
.await;
@@ -7969,8 +7970,10 @@ impl CodexMessageProcessor {
start_fuzzy_file_search_session(session_id.clone(), roots, self.outgoing.clone());
match session {
Ok(session) => {
let mut sessions = self.fuzzy_search_sessions.lock().await;
sessions.insert(session_id, session);
{
let mut sessions = self.fuzzy_search_sessions.lock().await;
sessions.insert(session_id, session);
}
self.outgoing
.send_response(request_id, FuzzyFileSearchSessionStartResponse {})
.await;
@@ -10297,13 +10300,9 @@ mod tests {
assert_eq!(cancel_rx.await, Ok(()));
let state = manager.thread_state(thread_id).await;
let subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await;
let state = state.lock().await;
assert!(
manager
.subscribed_connection_ids(thread_id)
.await
.is_empty()
);
assert!(subscribed_connection_ids.is_empty());
assert!(state.cancel_tx.is_none());
assert!(state.active_turn_snapshot().is_none());
Ok(())
codex-rs/app-server/src/message_processor/tracing_tests.rs
+11 -5
View File
@@ -101,9 +101,9 @@ fn request_from_client_request(request: ClientRequest) -> JSONRPCRequest {
.expect("client request should convert to JSON-RPC")
}
fn tracing_test_guard() -> &'static tokio::sync::Mutex<()> {
static GUARD: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
GUARD.get_or_init(|| tokio::sync::Mutex::new(()))
fn tracing_test_guard() -> &'static tokio::sync::Semaphore {
static GUARD: OnceLock<tokio::sync::Semaphore> = OnceLock::new();
GUARD.get_or_init(|| tokio::sync::Semaphore::new(1))
}
struct TracingHarness {
@@ -505,7 +505,10 @@ where
#[tokio::test(flavor = "current_thread")]
async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() -> Result<()> {
let _guard = tracing_test_guard().lock().await;
let _guard = tracing_test_guard()
.acquire()
.await
.expect("tracing test guard should stay open");
let mut harness = TracingHarness::new().await?;
let RemoteTrace {
@@ -584,7 +587,10 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() ->
#[tokio::test(flavor = "current_thread")]
async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
let _guard = tracing_test_guard().lock().await;
let _guard = tracing_test_guard()
.acquire()
.await
.expect("tracing test guard should stay open");
let mut harness = TracingHarness::new().await?;
let thread_start_response = harness.start_thread(/*request_id*/ 2, /*trace*/ None).await;
let thread_id = thread_start_response.thread.id.clone();
codex-rs/app-server/src/transport/remote_control/websocket.rs
+23 -14
View File
@@ -392,7 +392,15 @@ impl RemoteControlWebsocket {
ping_interval: std::time::Duration,
shutdown_token: CancellationToken,
) -> io::Result<()> {
for server_envelope in state.lock().await.outbound_buffer.server_envelopes() {
let server_envelopes = {
let state = state.lock().await;
state
.outbound_buffer
.server_envelopes()
.cloned()
.collect::<Vec<_>>()
};
for server_envelope in server_envelopes {
let payload = match serde_json::to_string(&server_envelope) {
Ok(payload) => payload,
Err(err) => {
@@ -594,21 +602,22 @@ impl RemoteControlWebsocket {
}
};
let mut websocket_state = state.lock().await;
if let Some(cursor) = client_envelope.cursor.as_deref() {
websocket_state.subscribe_cursor = Some(cursor.to_string());
}
if let ClientEvent::Ack = &client_envelope.event
&& let Some(acked_seq_id) = client_envelope.seq_id
&& let Some(stream_id) = client_envelope.stream_id.as_ref()
{
websocket_state.outbound_buffer.ack(
&client_envelope.client_id,
stream_id,
acked_seq_id,
);
let mut websocket_state = state.lock().await;
if let Some(cursor) = client_envelope.cursor.as_deref() {
websocket_state.subscribe_cursor = Some(cursor.to_string());
}
if let ClientEvent::Ack = &client_envelope.event
&& let Some(acked_seq_id) = client_envelope.seq_id
&& let Some(stream_id) = client_envelope.stream_id.as_ref()
{
websocket_state.outbound_buffer.ack(
&client_envelope.client_id,
stream_id,
acked_seq_id,
);
}
}
drop(websocket_state);
if client_tracker
.handle_message(client_envelope)
codex-rs/core/src/codex.rs
+166 -177
View File
@@ -2618,37 +2618,44 @@ impl Session {
&self,
updates: SessionSettingsUpdate,
) -> ConstraintResult<()> {
let mut state = self.state.lock().await;
match state.session_configuration.apply(&updates) {
Ok(updated) => {
let previous_cwd = state.session_configuration.cwd.clone();
let sandbox_policy_changed =
state.session_configuration.sandbox_policy != updated.sandbox_policy;
let next_cwd = updated.cwd.clone();
let codex_home = updated.codex_home.clone();
let session_source = updated.session_source.clone();
state.session_configuration = updated;
drop(state);
self.maybe_refresh_shell_snapshot_for_cwd(
&previous_cwd,
&next_cwd,
&codex_home,
&session_source,
);
if sandbox_policy_changed {
self.refresh_managed_network_proxy_for_current_sandbox_policy()
.await;
let (previous_cwd, sandbox_policy_changed, next_cwd, codex_home, session_source) = {
let mut state = self.state.lock().await;
let updated = match state.session_configuration.apply(&updates) {
Ok(updated) => updated,
Err(err) => {
warn!("rejected session settings update: {err}");
return Err(err);
}
};
Ok(())
}
Err(err) => {
warn!("rejected session settings update: {err}");
Err(err)
}
let previous_cwd = state.session_configuration.cwd.clone();
let sandbox_policy_changed =
state.session_configuration.sandbox_policy != updated.sandbox_policy;
let next_cwd = updated.cwd.clone();
let codex_home = updated.codex_home.clone();
let session_source = updated.session_source.clone();
state.session_configuration = updated;
(
previous_cwd,
sandbox_policy_changed,
next_cwd,
codex_home,
session_source,
)
};
self.maybe_refresh_shell_snapshot_for_cwd(
&previous_cwd,
&next_cwd,
&codex_home,
&session_source,
);
if sandbox_policy_changed {
self.refresh_managed_network_proxy_for_current_sandbox_policy()
.await;
}
Ok(())
}
pub(crate) async fn new_turn_with_sub_id(
@@ -2656,13 +2663,7 @@ impl Session {
sub_id: String,
updates: SessionSettingsUpdate,
) -> ConstraintResult<Arc<TurnContext>> {
let (
session_configuration,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
) = {
let update_result = {
let mut state = self.state.lock().await;
match state.session_configuration.clone().apply(&updates) {
Ok(next) => {
@@ -2672,26 +2673,35 @@ impl Session {
let codex_home = next.codex_home.clone();
let session_source = next.session_source.clone();
state.session_configuration = next.clone();
(
Ok((
next,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
)
}
Err(err) => {
drop(state);
self.send_event_raw(Event {
id: sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: err.to_string(),
codex_error_info: Some(CodexErrorInfo::BadRequest),
}),
})
.await;
return Err(err);
))
}
Err(err) => Err(err),
}
};
let (
session_configuration,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
) = match update_result {
Ok(update) => update,
Err(err) => {
self.send_event_raw(Event {
id: sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: err.to_string(),
codex_error_info: Some(CodexErrorInfo::BadRequest),
}),
})
.await;
return Err(err);
}
};
@@ -3273,10 +3283,9 @@ impl Session {
// Add the tx_approve callback to the map before sending the request.
let (tx_approve, rx_approve) = oneshot::channel();
let prev_entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_approval(effective_approval_id.clone(), tx_approve)
}
None => None,
@@ -3337,10 +3346,9 @@ impl Session {
let (tx_approve, rx_approve) = oneshot::channel();
let approval_id = call_id.clone();
let prev_entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_approval(approval_id.clone(), tx_approve)
}
None => None,
@@ -3390,10 +3398,9 @@ impl Session {
let (tx_response, rx_response) = oneshot::channel();
let prev_entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_request_permissions(call_id.clone(), tx_response)
}
None => None,
@@ -3426,10 +3433,9 @@ impl Session {
let (tx_response, rx_response) = oneshot::channel();
let event_id = sub_id.clone();
let prev_entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_user_input(sub_id, tx_response)
}
None => None,
@@ -3491,10 +3497,9 @@ impl Session {
let (tx_response, rx_response) = oneshot::channel();
let prev_entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_elicitation(
server_name.clone(),
request_id.clone(),
@@ -3532,15 +3537,12 @@ impl Session {
sub_id: &str,
response: RequestUserInputResponse,
) {
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
ts.remove_pending_user_input(sub_id)
}
None => None,
let entry = match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.remove_pending_user_input(sub_id)
}
None => None,
};
match entry {
Some(tx_response) => {
@@ -3558,26 +3560,23 @@ impl Session {
response: RequestPermissionsResponse,
) {
let mut granted_for_session = None;
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
let entry = ts.remove_pending_request_permissions(call_id);
if entry.is_some() && !response.permissions.is_empty() {
match response.scope {
PermissionGrantScope::Turn => {
ts.record_granted_permissions(response.permissions.clone().into());
}
PermissionGrantScope::Session => {
granted_for_session = Some(response.permissions.clone());
}
let entry = match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
let entry = ts.remove_pending_request_permissions(call_id);
if entry.is_some() && !response.permissions.is_empty() {
match response.scope {
PermissionGrantScope::Turn => {
ts.record_granted_permissions(response.permissions.clone().into());
}
PermissionGrantScope::Session => {
granted_for_session = Some(response.permissions.clone());
}
}
entry
}
None => None,
entry
}
None => None,
};
if let Some(permissions) = granted_for_session {
let mut state = self.state.lock().await;
@@ -3594,9 +3593,8 @@ impl Session {
}
pub(crate) async fn granted_turn_permissions(&self) -> Option<PermissionProfile> {
let active = self.active_turn.lock().await;
let active = active.as_ref()?;
let ts = active.turn_state.lock().await;
let turn_state = self.active_turn_state().await?;
let ts = turn_state.lock().await;
ts.granted_permissions()
}
@@ -3606,15 +3604,12 @@ impl Session {
}
pub async fn notify_dynamic_tool_response(&self, call_id: &str, response: DynamicToolResponse) {
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
ts.remove_pending_dynamic_tool(call_id)
}
None => None,
let entry = match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.remove_pending_dynamic_tool(call_id)
}
None => None,
};
match entry {
Some(tx_response) => {
@@ -3627,15 +3622,12 @@ impl Session {
}
pub async fn notify_approval(&self, approval_id: &str, decision: ReviewDecision) {
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
ts.remove_pending_approval(approval_id)
}
None => None,
let entry = match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.remove_pending_approval(approval_id)
}
None => None,
};
match entry {
Some(tx_approve) => {
@@ -3653,15 +3645,12 @@ impl Session {
id: RequestId,
response: ElicitationResponse,
) -> anyhow::Result<()> {
let entry = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
ts.remove_pending_elicitation(&server_name, &id)
}
None => None,
let entry = match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.remove_pending_elicitation(&server_name, &id)
}
None => None,
};
if let Some(tx_response) = entry {
tx_response
@@ -4271,52 +4260,47 @@ impl Session {
return Err(SteerInputError::EmptyInput);
}
let mut active = self.active_turn.lock().await;
let Some(active_turn) = active.as_mut() else {
return Err(SteerInputError::NoActiveTurn(input));
};
let Some((active_turn_id, _)) = active_turn.tasks.first() else {
return Err(SteerInputError::NoActiveTurn(input));
};
if let Some(expected_turn_id) = expected_turn_id
&& expected_turn_id != active_turn_id
{
return Err(SteerInputError::ExpectedTurnMismatch {
expected: expected_turn_id.to_string(),
actual: active_turn_id.clone(),
});
}
match active_turn.tasks.first().map(|(_, task)| task.kind) {
Some(crate::state::TaskKind::Regular) => {}
Some(crate::state::TaskKind::Review) => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Review,
let (active_turn_id, turn_state) = {
let active = self.active_turn.lock().await;
let Some(active_turn) = active.as_ref() else {
return Err(SteerInputError::NoActiveTurn(input));
};
let Some((active_turn_id, active_task)) = active_turn.tasks.first() else {
return Err(SteerInputError::NoActiveTurn(input));
};
if let Some(expected_turn_id) = expected_turn_id
&& expected_turn_id != active_turn_id
{
return Err(SteerInputError::ExpectedTurnMismatch {
expected: expected_turn_id.to_string(),
actual: active_turn_id.clone(),
});
}
Some(crate::state::TaskKind::Compact) => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Compact,
});
match active_task.kind {
crate::state::TaskKind::Regular => {}
crate::state::TaskKind::Review => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Review,
});
}
crate::state::TaskKind::Compact => {
return Err(SteerInputError::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Compact,
});
}
}
None => return Err(SteerInputError::NoActiveTurn(input)),
}
if let Some(responsesapi_client_metadata) = responsesapi_client_metadata
&& let Some((_, active_task)) = active_turn.tasks.first()
{
active_task
.turn_context
.turn_metadata_state
.set_responsesapi_client_metadata(responsesapi_client_metadata);
}
let mut turn_state = active_turn.turn_state.lock().await;
if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {
active_task
.turn_context
.turn_metadata_state
.set_responsesapi_client_metadata(responsesapi_client_metadata);
}
(active_turn_id.clone(), Arc::clone(&active_turn.turn_state))
};
let mut turn_state = turn_state.lock().await;
turn_state.push_pending_input(input.into());
turn_state.accept_mailbox_delivery_for_current_turn();
Ok(active_turn_id.clone())
Ok(active_turn_id)
}
/// Returns the input if there was no task running to inject into.
@@ -4324,10 +4308,9 @@ impl Session {
&self,
input: Vec<ResponseInputItem>,
) -> Result<(), Vec<ResponseInputItem>> {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
for item in input {
ts.push_pending_input(item);
}
@@ -4373,6 +4356,15 @@ impl Session {
})
}
pub(crate) async fn active_turn_state(
&self,
) -> Option<Arc<tokio::sync::Mutex<crate::state::TurnState>>> {
let active = self.active_turn.lock().await;
active
.as_ref()
.map(|active_turn| Arc::clone(&active_turn.turn_state))
}
pub(crate) fn subscribe_mailbox_seq(&self) -> watch::Receiver<u64> {
self.mailbox.subscribe()
}
@@ -4386,10 +4378,9 @@ impl Session {
}
pub async fn prepend_pending_input(&self, input: Vec<ResponseInputItem>) -> Result<(), ()> {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.prepend_pending_input(input);
Ok(())
}
@@ -4399,10 +4390,9 @@ impl Session {
pub async fn get_pending_input(&self) -> Vec<ResponseInputItem> {
let (pending_input, accepts_mailbox_delivery) = {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
(
ts.take_pending_input(),
ts.accepts_mailbox_delivery_for_current_turn(),
@@ -4454,10 +4444,9 @@ impl Session {
pub async fn has_pending_input(&self) -> bool {
let (has_turn_pending_input, accepts_mailbox_delivery) = {
let active = self.active_turn.lock().await;
match active.as_ref() {
Some(at) => {
let ts = at.turn_state.lock().await;
match self.active_turn_state().await {
Some(turn_state) => {
let ts = turn_state.lock().await;
(
ts.has_pending_input(),
ts.accepts_mailbox_delivery_for_current_turn(),
codex-rs/core/src/codex_tests_guardian.rs
+5 -3
View File
@@ -302,9 +302,11 @@ async fn shell_handler_allows_sticky_turn_permissions_without_inline_request_per
.expect("test setup should allow enabling request permissions tool");
*session.active_turn.lock().await = Some(ActiveTurn::default());
{
let mut active_turn = session.active_turn.lock().await;
let active_turn = active_turn.as_mut().expect("active turn");
let mut turn_state = active_turn.turn_state.lock().await;
let turn_state = {
let active_turn = session.active_turn.lock().await;
Arc::clone(&active_turn.as_ref().expect("active turn").turn_state)
};
let mut turn_state = turn_state.lock().await;
turn_state.record_granted_permissions(PermissionProfile {
network: Some(NetworkPermissions {
enabled: Some(true),
codex-rs/core/src/tools/handlers/dynamic.rs
+3 -4
View File
@@ -82,10 +82,9 @@ async fn request_dynamic_tool(
let (tx_response, rx_response) = oneshot::channel();
let event_id = call_id.clone();
let prev_entry = {
let mut active = session.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
let mut ts = at.turn_state.lock().await;
match session.active_turn_state().await {
Some(turn_state) => {
let mut ts = turn_state.lock().await;
ts.insert_pending_dynamic_tool(call_id.clone(), tx_response)
}
None => None,
codex-rs/core/src/tools/js_repl/mod.rs
+59 -51
View File
@@ -884,17 +884,22 @@ impl JsReplManager {
let (req_id, rx) = {
let req_id = Uuid::new_v4().to_string();
let mut pending = pending_execs.lock().await;
let (tx, rx) = tokio::sync::oneshot::channel();
pending.insert(req_id.clone(), tx);
exec_contexts.lock().await.insert(
req_id.clone(),
ExecContext {
session: Arc::clone(&session),
turn: Arc::clone(&turn),
tracker,
},
);
{
let mut pending = pending_execs.lock().await;
pending.insert(req_id.clone(), tx);
}
{
let mut exec_contexts = exec_contexts.lock().await;
exec_contexts.insert(
req_id.clone(),
ExecContext {
session: Arc::clone(&session),
turn: Arc::clone(&turn),
tracker,
},
);
}
(req_id, rx)
};
self.register_top_level_exec(req_id.clone(), turn.sub_id.clone())
@@ -956,9 +961,10 @@ impl JsReplManager {
let response = match tokio::time::timeout(Duration::from_millis(timeout_ms), rx).await {
Ok(Ok(msg)) => msg,
Ok(Err(_)) => {
let mut pending = pending_execs.lock().await;
let removed = pending.remove(&req_id).is_some();
drop(pending);
let removed = {
let mut pending = pending_execs.lock().await;
pending.remove(&req_id).is_some()
};
if removed {
self.clear_top_level_exec_if_matches(&req_id).await;
}
@@ -1340,40 +1346,40 @@ impl JsReplManager {
KernelToHost::EmitImage(req) => {
let exec_id = req.exec_id.clone();
let emit_id = req.id.clone();
let response =
if let Some(ctx) = exec_contexts.lock().await.get(&exec_id).cloned() {
match validate_emitted_image_url(&req.image_url) {
Ok(()) => {
let content_item = emitted_image_content_item(
ctx.turn.as_ref(),
req.image_url,
req.detail,
);
JsReplManager::record_exec_content_item(
&exec_tool_calls,
&exec_id,
content_item,
)
.await;
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: true,
error: None,
})
}
Err(error) => HostToKernel::EmitImageResult(EmitImageResult {
let context = { exec_contexts.lock().await.get(&exec_id).cloned() };
let response = if let Some(ctx) = context {
match validate_emitted_image_url(&req.image_url) {
Ok(()) => {
let content_item = emitted_image_content_item(
ctx.turn.as_ref(),
req.image_url,
req.detail,
);
JsReplManager::record_exec_content_item(
&exec_tool_calls,
&exec_id,
content_item,
)
.await;
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some(error),
}),
ok: true,
error: None,
})
}
} else {
HostToKernel::EmitImageResult(EmitImageResult {
Err(error) => HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some("js_repl exec context not found".to_string()),
})
};
error: Some(error),
}),
}
} else {
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some("js_repl exec context not found".to_string()),
})
};
if let Err(err) = JsReplManager::write_message(&stdin, &response).await {
let snapshot =
@@ -1502,14 +1508,16 @@ impl JsReplManager {
}
}
let mut pending = pending_execs.lock().await;
let pending_exec_ids = pending.keys().cloned().collect::<Vec<_>>();
for (_id, tx) in pending.drain() {
let _ = tx.send(ExecResultMessage::Err {
message: kernel_exit_message.clone(),
});
}
drop(pending);
let pending_exec_ids = {
let mut pending = pending_execs.lock().await;
let pending_exec_ids = pending.keys().cloned().collect::<Vec<_>>();
for (_id, tx) in pending.drain() {
let _ = tx.send(ExecResultMessage::Err {
message: kernel_exit_message.clone(),
});
}
pending_exec_ids
};
if !pending_exec_ids.is_empty() {
Self::clear_top_level_exec_if_matches_any_map(&manager_kernel, &pending_exec_ids).await;
}
codex-rs/core/src/tools/network_approval.rs
+16 -8
View File
@@ -217,8 +217,10 @@ impl NetworkApprovalService {
}
pub(crate) async fn unregister_call(&self, registration_id: &str) {
let mut active_calls = self.active_calls.lock().await;
active_calls.shift_remove(registration_id);
{
let mut active_calls = self.active_calls.lock().await;
active_calls.shift_remove(registration_id);
}
let mut call_outcomes = self.call_outcomes.lock().await;
call_outcomes.remove(registration_id);
}
@@ -336,8 +338,10 @@ impl NetworkApprovalService {
let Some(turn_context) = Self::active_turn_context(session.as_ref()).await else {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
{
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
}
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
@@ -346,8 +350,10 @@ impl NetworkApprovalService {
};
if !sandbox_policy_allows_network_approval_flow(turn_context.sandbox_policy.get()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
{
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
}
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
@@ -356,8 +362,10 @@ impl NetworkApprovalService {
}
if !allows_network_approval_flow(turn_context.approval_policy.value()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
{
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
}
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
codex-rs/core/src/tools/registry.rs
+2 -3
View File
@@ -246,9 +246,8 @@ impl ToolRegistry {
let mcp_server_origin_ref = mcp_server_origin.as_deref();
{
let mut active = invocation.session.active_turn.lock().await;
if let Some(active_turn) = active.as_mut() {
let mut turn_state = active_turn.turn_state.lock().await;
if let Some(active_turn) = invocation.session.active_turn_state().await {
let mut turn_state = active_turn.lock().await;
turn_state.tool_calls = turn_state.tool_calls.saturating_add(1);
}
}
codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs
+14 -13
View File
@@ -418,22 +418,23 @@ async fn fail_session_post_when_armed(
return next.run(request).await;
}
let mut armed_failure = state.armed_failure.lock().await;
if let Some(failure) = armed_failure.as_mut()
&& failure.remaining > 0
{
failure.remaining -= 1;
let status = failure.status;
if failure.remaining == 0 {
*armed_failure = None;
let mut armed_failure = state.armed_failure.lock().await;
if let Some(failure) = armed_failure.as_mut()
&& failure.remaining > 0
{
failure.remaining -= 1;
let status = failure.status;
if failure.remaining == 0 {
*armed_failure = None;
}
let mut response = Response::new(Body::from(format!(
"forced session failure with status {status}"
)));
*response.status_mut() = status;
return response;
}
let mut response = Response::new(Body::from(format!(
"forced session failure with status {status}"
)));
*response.status_mut() = status;
return response;
}
drop(armed_failure);
next.run(request).await
}
codex-rs/shell-escalation/src/unix/escalate_server.rs
+10 -10
View File
@@ -391,11 +391,11 @@ mod tests {
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use tempfile::TempDir;
use tokio::sync::Semaphore;
use tokio::time::Instant;
use tokio::time::sleep;
static ESCALATE_SERVER_TEST_LOCK: LazyLock<tokio::sync::Mutex<()>> =
LazyLock::new(|| tokio::sync::Mutex::new(()));
static ESCALATE_SERVER_TEST_LOCK: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(1));
struct DeterministicEscalationPolicy {
decision: EscalationDecision,
@@ -596,7 +596,7 @@ mod tests {
/// until `close_client_socket()` is called.
#[tokio::test]
async fn start_session_exposes_wrapper_env_overlay() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let execve_wrapper = PathBuf::from("/tmp/codex-execve-wrapper");
let execve_wrapper_str = execve_wrapper.to_string_lossy().to_string();
let server = EscalateServer::new(
@@ -638,7 +638,7 @@ mod tests {
#[tokio::test]
async fn exec_closes_parent_socket_after_shell_spawn() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let after_spawn_invoked = Arc::new(AtomicBool::new(false));
let server = EscalateServer::new(
PathBuf::from("/bin/bash"),
@@ -672,7 +672,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_respects_run_in_sandbox_decision() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -712,7 +712,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_resolves_relative_file_against_request_workdir()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let tmp = tempfile::TempDir::new()?;
let workdir = tmp.path().join("workspace");
@@ -751,7 +751,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_executes_escalated_command() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -844,7 +844,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_accepts_received_fds_that_overlap_destinations()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let mut pipe_fds = [0; 2];
if unsafe { libc::pipe(pipe_fds.as_mut_ptr()) } == -1 {
return Err(std::io::Error::last_os_error().into());
@@ -916,7 +916,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_passes_permissions_to_executor() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -972,7 +972,7 @@ mod tests {
#[tokio::test]
async fn dropping_session_aborts_intercept_workers_and_kills_spawned_child()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let tmp = TempDir::new()?;
let pid_file = tmp.path().join("escalated-child.pid");
let pid_file_display = pid_file.display().to_string();
codex-rs/utils/pty/src/pipe.rs
+3 -4
View File
@@ -162,12 +162,11 @@ async fn spawn_process_with_stdin_mode(
let (stdout_tx, stdout_rx) = mpsc::channel::<Vec<u8>>(128);
let (stderr_tx, stderr_rx) = mpsc::channel::<Vec<u8>>(128);
let writer_handle = if let Some(stdin) = stdin {
let writer = Arc::new(tokio::sync::Mutex::new(stdin));
tokio::spawn(async move {
let mut writer = stdin;
while let Some(bytes) = writer_rx.recv().await {
let mut guard = writer.lock().await;
let _ = guard.write_all(&bytes).await;
let _ = guard.flush().await;
let _ = writer.write_all(&bytes).await;
let _ = writer.flush().await;
}
})
} else {
codex-rs/utils/readiness/src/lib.rs
+24 -5
View File
@@ -277,17 +277,36 @@ mod tests {
#[tokio::test]
async fn subscribe_returns_error_when_lock_is_held() {
let flag = ReadinessFlag::new();
let _guard = flag
.tokens
.try_lock()
.expect("initial lock acquisition should succeed");
let flag = Arc::new(ReadinessFlag::new());
let (locked_tx, locked_rx) = std::sync::mpsc::channel();
let (release_tx, release_rx) = std::sync::mpsc::channel();
let lock_thread = {
let flag = Arc::clone(&flag);
std::thread::spawn(move || {
let _guard = flag.tokens.blocking_lock();
locked_tx
.send(())
.expect("test should receive lock acquisition notification");
release_rx
.recv()
.expect("test should release held readiness lock");
})
};
locked_rx
.recv()
.expect("test should observe held readiness lock");
let err = flag
.subscribe()
.await
.expect_err("contended subscribe should report a lock failure");
assert_matches!(err, ReadinessError::TokenLockFailed);
release_tx
.send(())
.expect("test should release readiness lock thread");
lock_thread
.join()
.expect("readiness lock thread should not panic");
}
#[tokio::test]