openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge d12b99159b into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2025-08-14 21:43:39 -07:00
committed by GitHub
parent 92fdf1a15d d12b99159b
commit f76b7cfa00
1 changed files with 22 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
codex-rs/core/src/codex.rs
View File

@@ -7,7 +7,7 @@ use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::RwLock;
use std::sync::atomic::AtomicU64;
use std::time::Duration;
@@ -241,8 +241,8 @@ pub(crate) struct Session {
/// Optional rollout recorder for persisting the conversation transcript so
/// sessions can be replayed or inspected later.
rollout: Mutex<Option<RolloutRecorder>>,
state: Mutex<State>,
rollout: RwLock<Option<RolloutRecorder>>,
state: RwLock<State>,
codex_linux_sandbox_exe: Option<PathBuf>,
user_shell: shell::Shell,
show_raw_agent_reasoning: bool,
@@ -461,8 +461,8 @@ impl Session {
cwd,
mcp_connection_manager,
notify,
state: Mutex::new(state),
rollout: Mutex::new(rollout_recorder),
state: RwLock::new(state),
rollout: RwLock::new(rollout_recorder),
codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(),
disable_response_storage,
user_shell: default_shell,
@@ -521,7 +521,7 @@ impl Session {
}
pub fn set_task(&self, task: AgentTask) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
if let Some(current_task) = state.current_task.take() {
current_task.abort();
}
@@ -529,7 +529,7 @@ impl Session {
}
pub fn remove_task(&self, sub_id: &str) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
if let Some(task) = &state.current_task {
if task.sub_id == sub_id {
state.current_task.take();
@@ -565,7 +565,7 @@ impl Session {
};
let _ = self.tx_event.send(event).await;
{
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
state.pending_approvals.insert(sub_id, tx_approve);
}
rx_approve
@@ -591,21 +591,21 @@ impl Session {
};
let _ = self.tx_event.send(event).await;
{
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
state.pending_approvals.insert(sub_id, tx_approve);
}
rx_approve
}
pub fn notify_approval(&self, sub_id: &str, decision: ReviewDecision) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
if let Some(tx_approve) = state.pending_approvals.remove(sub_id) {
tx_approve.send(decision).ok();
}
}
pub fn add_approved_command(&self, cmd: Vec<String>) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
state.approved_commands.insert(cmd);
}
@@ -615,14 +615,14 @@ impl Session {
debug!("Recording items for conversation: {items:?}");
self.record_state_snapshot(items).await;
self.state.lock().unwrap().history.record_items(items);
self.state.write().unwrap().history.record_items(items);
}
async fn record_state_snapshot(&self, items: &[ResponseItem]) {
let snapshot = { crate::rollout::SessionStateSnapshot {} };
let recorder = {
let guard = self.rollout.lock().unwrap();
let guard = self.rollout.read().unwrap();
guard.as_ref().cloned()
};
@@ -800,12 +800,12 @@ impl Session {
/// Build the full turn input by concatenating the current conversation
/// history with additional items for this turn.
pub fn turn_input_with_history(&self, extra: Vec<ResponseItem>) -> Vec<ResponseItem> {
[self.state.lock().unwrap().history.contents(), extra].concat()
[self.state.read().unwrap().history.contents(), extra].concat()
}
/// Returns the input if there was no task running to inject into
pub fn inject_input(&self, input: Vec<InputItem>) -> Result<(), Vec<InputItem>> {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
if state.current_task.is_some() {
state.pending_input.push(input.into());
Ok(())
@@ -815,7 +815,7 @@ impl Session {
}
pub fn get_pending_input(&self) -> Vec<ResponseInputItem> {
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
if state.pending_input.is_empty() {
Vec::with_capacity(0)
} else {
@@ -839,7 +839,7 @@ impl Session {
fn abort(&self) {
info!("Aborting existing session");
let mut state = self.state.lock().unwrap();
let mut state = self.state.write().unwrap();
state.pending_approvals.clear();
state.pending_input.clear();
if let Some(task) = state.current_task.take() {
@@ -1043,7 +1043,7 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
// Gracefully flush and shutdown rollout recorder on session end so tests
// that inspect the rollout file do not race with the background writer.
let recorder_opt = sess.rollout.lock().unwrap().take();
let recorder_opt = sess.rollout.write().unwrap().take();
if let Some(rec) = recorder_opt {
if let Err(e) = rec.shutdown().await {
warn!("failed to shutdown rollout recorder: {e}");
@@ -1459,7 +1459,7 @@ async fn try_run_turn(
}
ResponseEvent::OutputTextDelta(delta) => {
{
let mut st = sess.state.lock().unwrap();
let mut st = sess.state.write().unwrap();
st.history.append_assistant_text(&delta);
}
@@ -1575,7 +1575,7 @@ async fn run_compact_task(
};
sess.send_event(event).await;
let mut state = sess.state.lock().unwrap();
let mut state = sess.state.write().unwrap();
state.history.keep_last_messages(1);
}
@@ -1886,7 +1886,7 @@ async fn handle_container_exec_with_params(
}
None => {
let safety = {
let state = sess.state.lock().unwrap();
let state = sess.state.read().unwrap();
assess_command_safety(
&params.command,
sess.approval_policy,
@@ -2226,7 +2226,7 @@ async fn drain_to_completed(sess: &Session, sub_id: &str, prompt: &Prompt) -> Co
match event {
Ok(ResponseEvent::OutputItemDone(item)) => {
// Record only to in-memory conversation history; avoid state snapshot.
let mut state = sess.state.lock().unwrap();
let mut state = sess.state.write().unwrap();
state.history.record_items(std::slice::from_ref(&item));
}
Ok(ResponseEvent::Completed {