diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 11138fdc8a..ae230e04d4 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1701,7 +1701,6 @@ dependencies = [ "anyhow", "arboard", "assert_matches", - "async-stream", "base64", "chrono", "clap", @@ -6910,6 +6909,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index a300b5a70d..310f7e57bb 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -58,7 +58,7 @@ impl ResponsesClient { self.stream(request.body, request.headers).await } - #[instrument(skip_all, err)] + #[instrument(level = "trace", skip_all, err)] pub async fn stream_prompt( &self, model: &str, diff --git a/codex-rs/codex-client/src/default_client.rs b/codex-rs/codex-client/src/default_client.rs index e79f873cbb..efb4d5aec4 100644 --- a/codex-rs/codex-client/src/default_client.rs +++ b/codex-rs/codex-client/src/default_client.rs @@ -181,7 +181,7 @@ mod tests { use opentelemetry::trace::TracerProvider; use opentelemetry_sdk::propagation::TraceContextPropagator; use opentelemetry_sdk::trace::SdkTracerProvider; - use tracing::info_span; + use tracing::trace_span; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; @@ -195,7 +195,7 @@ mod tests { tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer)); let _guard = subscriber.set_default(); - let span = info_span!("client_request"); + let span = trace_span!("client_request"); let _entered = span.enter(); let span_context = span.context().span().span_context().clone(); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 5f0e322dae..bcfe8e8114 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -66,8 +66,8 @@ use tracing::debug; use tracing::error; use tracing::field; use tracing::info; -use tracing::info_span; use tracing::instrument; +use tracing::trace_span; use tracing::warn; use crate::ModelProviderInfo; @@ -2150,6 +2150,16 @@ pub(crate) async fn run_task( if input.is_empty() { return None; } + + let auto_compact_limit = turn_context + .client + .get_model_family() + .auto_compact_token_limit() + .unwrap_or(i64::MAX); + let total_usage_tokens = sess.get_total_token_usage().await; + if total_usage_tokens >= auto_compact_limit { + run_auto_compact(&sess, &turn_context).await; + } let event = EventMsg::TaskStarted(TaskStartedEvent { model_context_window: turn_context.client.get_model_context_window(), }); @@ -2232,25 +2242,12 @@ pub(crate) async fn run_task( needs_follow_up, last_agent_message: turn_last_agent_message, } = turn_output; - let limit = turn_context - .client - .get_model_family() - .auto_compact_token_limit() - .unwrap_or(i64::MAX); let total_usage_tokens = sess.get_total_token_usage().await; - let token_limit_reached = total_usage_tokens >= limit; + let token_limit_reached = total_usage_tokens >= auto_compact_limit; // as long as compaction works well in getting us way below the token limit, we shouldn't worry about being in an infinite loop. - if token_limit_reached { - if should_use_remote_compact_task( - sess.as_ref(), - &turn_context.client.get_provider(), - ) { - run_inline_remote_auto_compact_task(sess.clone(), turn_context.clone()) - .await; - } else { - run_inline_auto_compact_task(sess.clone(), turn_context.clone()).await; - } + if token_limit_reached && needs_follow_up { + run_auto_compact(&sess, &turn_context).await; continue; } @@ -2292,7 +2289,15 @@ pub(crate) async fn run_task( last_agent_message } -#[instrument( +async fn run_auto_compact(sess: &Arc, turn_context: &Arc) { + if should_use_remote_compact_task(sess.as_ref(), &turn_context.client.get_provider()) { + run_inline_remote_auto_compact_task(Arc::clone(sess), Arc::clone(turn_context)).await; + } else { + run_inline_auto_compact_task(Arc::clone(sess), Arc::clone(turn_context)).await; + } +} + +#[instrument(level = "trace", skip_all, fields( turn_id = %turn_context.sub_id, @@ -2432,7 +2437,7 @@ async fn drain_in_flight( } #[allow(clippy::too_many_arguments)] -#[instrument( +#[instrument(level = "trace", skip_all, fields( turn_id = %turn_context.sub_id, @@ -2461,7 +2466,7 @@ async fn try_run_turn( .client .clone() .stream(prompt) - .instrument(info_span!("stream_request")) + .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) .await??; @@ -2477,9 +2482,9 @@ async fn try_run_turn( let mut last_agent_message: Option = None; let mut active_item: Option = None; let mut should_emit_turn_diff = false; - let receiving_span = info_span!("receiving_stream"); + let receiving_span = trace_span!("receiving_stream"); let outcome: CodexResult = loop { - let handle_responses = info_span!( + let handle_responses = trace_span!( parent: &receiving_span, "handle_responses", otel.name = field::Empty, @@ -2489,7 +2494,7 @@ async fn try_run_turn( let event = match stream .next() - .instrument(info_span!(parent: &handle_responses, "receiving")) + .instrument(trace_span!(parent: &handle_responses, "receiving")) .or_cancel(&cancellation_token) .await { diff --git a/codex-rs/core/src/mcp_connection_manager.rs b/codex-rs/core/src/mcp_connection_manager.rs index 460d598a94..3213b22b71 100644 --- a/codex-rs/core/src/mcp_connection_manager.rs +++ b/codex-rs/core/src/mcp_connection_manager.rs @@ -398,7 +398,7 @@ impl McpConnectionManager { /// Returns a single map that contains all tools. Each key is the /// fully-qualified name for the tool. - #[instrument(skip_all)] + #[instrument(level = "trace", skip_all)] pub async fn list_all_tools(&self) -> HashMap { let mut tools = HashMap::new(); for managed_client in self.clients.values() { diff --git a/codex-rs/core/src/seatbelt.rs b/codex-rs/core/src/seatbelt.rs index 5e01b041d0..6958e52219 100644 --- a/codex-rs/core/src/seatbelt.rs +++ b/codex-rs/core/src/seatbelt.rs @@ -166,30 +166,34 @@ mod tests { use super::create_seatbelt_command_args; use super::macos_dir_params; use crate::protocol::SandboxPolicy; + use crate::seatbelt::MACOS_PATH_TO_SEATBELT_EXECUTABLE; use pretty_assertions::assert_eq; use std::fs; use std::path::Path; use std::path::PathBuf; + use std::process::Command; use tempfile::TempDir; #[test] - fn create_seatbelt_args_with_read_only_git_subpath() { + fn create_seatbelt_args_with_read_only_git_and_codex_subpaths() { // Create a temporary workspace with two writable roots: one containing - // a top-level .git directory and one without it. + // top-level .git and .codex directories and one without them. let tmp = TempDir::new().expect("tempdir"); let PopulatedTmp { - root_with_git, - root_without_git, - root_with_git_canon, - root_with_git_git_canon, - root_without_git_canon, + vulnerable_root, + vulnerable_root_canonical, + dot_git_canonical, + dot_codex_canonical, + empty_root, + empty_root_canonical, } = populate_tmpdir(tmp.path()); let cwd = tmp.path().join("cwd"); + fs::create_dir_all(&cwd).expect("create cwd"); // Build a policy that only includes the two test roots as writable and // does not automatically include defaults TMPDIR or /tmp. let policy = SandboxPolicy::WorkspaceWrite { - writable_roots: vec![root_with_git, root_without_git] + writable_roots: vec![vulnerable_root, empty_root] .into_iter() .map(|p| p.try_into().unwrap()) .collect(), @@ -198,23 +202,34 @@ mod tests { exclude_slash_tmp: true, }; - let args = create_seatbelt_command_args( - vec!["/bin/echo".to_string(), "hello".to_string()], - &policy, - &cwd, - ); + // Create the Seatbelt command to wrap a shell command that tries to + // write to .codex/config.toml in the vulnerable root. + let shell_command: Vec = [ + "bash", + "-c", + "echo 'sandbox_mode = \"danger-full-access\"' > \"$1\"", + "bash", + dot_codex_canonical + .join("config.toml") + .to_string_lossy() + .as_ref(), + ] + .iter() + .map(std::string::ToString::to_string) + .collect(); + let args = create_seatbelt_command_args(shell_command.clone(), &policy, &cwd); // Build the expected policy text using a raw string for readability. // Note that the policy includes: // - the base policy, // - read-only access to the filesystem, - // - write access to WRITABLE_ROOT_0 (but not its .git) and WRITABLE_ROOT_1. + // - write access to WRITABLE_ROOT_0 (but not its .git or .codex), WRITABLE_ROOT_1, and cwd as WRITABLE_ROOT_2. let expected_policy = format!( r#"{MACOS_SEATBELT_BASE_POLICY} ; allow read-only file operations (allow file-read*) (allow file-write* -(require-all (subpath (param "WRITABLE_ROOT_0")) (require-not (subpath (param "WRITABLE_ROOT_0_RO_0"))) ) (subpath (param "WRITABLE_ROOT_1")) (subpath (param "WRITABLE_ROOT_2")) +(require-all (subpath (param "WRITABLE_ROOT_0")) (require-not (subpath (param "WRITABLE_ROOT_0_RO_0"))) (require-not (subpath (param "WRITABLE_ROOT_0_RO_1"))) ) (subpath (param "WRITABLE_ROOT_1")) (subpath (param "WRITABLE_ROOT_2")) ) "#, ); @@ -224,17 +239,26 @@ mod tests { expected_policy, format!( "-DWRITABLE_ROOT_0={}", - root_with_git_canon.to_string_lossy() + vulnerable_root_canonical.to_string_lossy() ), format!( "-DWRITABLE_ROOT_0_RO_0={}", - root_with_git_git_canon.to_string_lossy() + dot_git_canonical.to_string_lossy() + ), + format!( + "-DWRITABLE_ROOT_0_RO_1={}", + dot_codex_canonical.to_string_lossy() ), format!( "-DWRITABLE_ROOT_1={}", - root_without_git_canon.to_string_lossy() + empty_root_canonical.to_string_lossy() + ), + format!( + "-DWRITABLE_ROOT_2={}", + cwd.canonicalize() + .expect("canonicalize cwd") + .to_string_lossy() ), - format!("-DWRITABLE_ROOT_2={}", cwd.to_string_lossy()), ]; expected_args.extend( @@ -243,30 +267,119 @@ mod tests { .map(|(key, value)| format!("-D{key}={value}", value = value.to_string_lossy())), ); - expected_args.extend(vec![ - "--".to_string(), - "/bin/echo".to_string(), - "hello".to_string(), - ]); + expected_args.push("--".to_string()); + expected_args.extend(shell_command); assert_eq!(expected_args, args); + + // Verify that .codex/config.toml cannot be modified under the generated + // Seatbelt policy. + let config_toml = dot_codex_canonical.join("config.toml"); + let output = Command::new(MACOS_PATH_TO_SEATBELT_EXECUTABLE) + .args(&args) + .current_dir(&cwd) + .output() + .expect("execute seatbelt command"); + assert_eq!( + "sandbox_mode = \"read-only\"\n", + String::from_utf8_lossy(&fs::read(&config_toml).expect("read config.toml")), + "config.toml should contain its original contents because it should not have been modified" + ); + assert!( + !output.status.success(), + "command to write {} should fail under seatbelt", + &config_toml.display() + ); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + format!("bash: {}: Operation not permitted\n", config_toml.display()), + ); + + // Create a similar Seatbelt command that tries to write to a file in + // the .git folder, which should also be blocked. + let pre_commit_hook = dot_git_canonical.join("hooks").join("pre-commit"); + let shell_command_git: Vec = [ + "bash", + "-c", + "echo 'pwned!' > \"$1\"", + "bash", + pre_commit_hook.to_string_lossy().as_ref(), + ] + .iter() + .map(std::string::ToString::to_string) + .collect(); + let write_hooks_file_args = create_seatbelt_command_args(shell_command_git, &policy, &cwd); + let output = Command::new(MACOS_PATH_TO_SEATBELT_EXECUTABLE) + .args(&write_hooks_file_args) + .current_dir(&cwd) + .output() + .expect("execute seatbelt command"); + assert!( + !fs::exists(&pre_commit_hook).expect("exists pre-commit hook"), + "{} should not exist because it should not have been created", + pre_commit_hook.display() + ); + assert!( + !output.status.success(), + "command to write {} should fail under seatbelt", + &pre_commit_hook.display() + ); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + format!( + "bash: {}: Operation not permitted\n", + pre_commit_hook.display() + ), + ); + + // Verify that writing a file to the folder containing .git and .codex is allowed. + let allowed_file = vulnerable_root_canonical.join("allowed.txt"); + let shell_command_allowed: Vec = [ + "bash", + "-c", + "echo 'this is allowed' > \"$1\"", + "bash", + allowed_file.to_string_lossy().as_ref(), + ] + .iter() + .map(std::string::ToString::to_string) + .collect(); + let write_allowed_file_args = + create_seatbelt_command_args(shell_command_allowed, &policy, &cwd); + let output = Command::new(MACOS_PATH_TO_SEATBELT_EXECUTABLE) + .args(&write_allowed_file_args) + .current_dir(&cwd) + .output() + .expect("execute seatbelt command"); + assert!( + output.status.success(), + "command to write {} should succeed under seatbelt", + &allowed_file.display() + ); + assert_eq!( + "this is allowed\n", + String::from_utf8_lossy(&fs::read(&allowed_file).expect("read allowed.txt")), + "{} should contain the written text", + allowed_file.display() + ); } #[test] fn create_seatbelt_args_for_cwd_as_git_repo() { // Create a temporary workspace with two writable roots: one containing - // a top-level .git directory and one without it. + // top-level .git and .codex directories and one without them. let tmp = TempDir::new().expect("tempdir"); let PopulatedTmp { - root_with_git, - root_with_git_canon, - root_with_git_git_canon, + vulnerable_root, + vulnerable_root_canonical, + dot_git_canonical, + dot_codex_canonical, .. } = populate_tmpdir(tmp.path()); // Build a policy that does not specify any writable_roots, but does - // use the default ones (cwd and TMPDIR) and verifies the `.git` check - // is done properly for cwd. + // use the default ones (cwd and TMPDIR) and verifies the `.git` and + // `.codex` checks are done properly for cwd. let policy = SandboxPolicy::WorkspaceWrite { writable_roots: vec![], network_access: false, @@ -274,11 +387,21 @@ mod tests { exclude_slash_tmp: false, }; - let args = create_seatbelt_command_args( - vec!["/bin/echo".to_string(), "hello".to_string()], - &policy, - root_with_git.as_path(), - ); + let shell_command: Vec = [ + "bash", + "-c", + "echo 'sandbox_mode = \"danger-full-access\"' > \"$1\"", + "bash", + dot_codex_canonical + .join("config.toml") + .to_string_lossy() + .as_ref(), + ] + .iter() + .map(std::string::ToString::to_string) + .collect(); + let args = + create_seatbelt_command_args(shell_command.clone(), &policy, vulnerable_root.as_path()); let tmpdir_env_var = std::env::var("TMPDIR") .ok() @@ -296,13 +419,13 @@ mod tests { // Note that the policy includes: // - the base policy, // - read-only access to the filesystem, - // - write access to WRITABLE_ROOT_0 (but not its .git) and WRITABLE_ROOT_1. + // - write access to WRITABLE_ROOT_0 (but not its .git or .codex), WRITABLE_ROOT_1, and cwd as WRITABLE_ROOT_2. let expected_policy = format!( r#"{MACOS_SEATBELT_BASE_POLICY} ; allow read-only file operations (allow file-read*) (allow file-write* -(require-all (subpath (param "WRITABLE_ROOT_0")) (require-not (subpath (param "WRITABLE_ROOT_0_RO_0"))) ) (subpath (param "WRITABLE_ROOT_1")){tempdir_policy_entry} +(require-all (subpath (param "WRITABLE_ROOT_0")) (require-not (subpath (param "WRITABLE_ROOT_0_RO_0"))) (require-not (subpath (param "WRITABLE_ROOT_0_RO_1"))) ) (subpath (param "WRITABLE_ROOT_1")){tempdir_policy_entry} ) "#, ); @@ -312,11 +435,15 @@ mod tests { expected_policy, format!( "-DWRITABLE_ROOT_0={}", - root_with_git_canon.to_string_lossy() + vulnerable_root_canonical.to_string_lossy() ), format!( "-DWRITABLE_ROOT_0_RO_0={}", - root_with_git_git_canon.to_string_lossy() + dot_git_canonical.to_string_lossy() + ), + format!( + "-DWRITABLE_ROOT_0_RO_1={}", + dot_codex_canonical.to_string_lossy() ), format!( "-DWRITABLE_ROOT_1={}", @@ -337,42 +464,68 @@ mod tests { .map(|(key, value)| format!("-D{key}={value}", value = value.to_string_lossy())), ); - expected_args.extend(vec![ - "--".to_string(), - "/bin/echo".to_string(), - "hello".to_string(), - ]); + expected_args.push("--".to_string()); + expected_args.extend(shell_command); assert_eq!(expected_args, args); } struct PopulatedTmp { - root_with_git: PathBuf, - root_without_git: PathBuf, - root_with_git_canon: PathBuf, - root_with_git_git_canon: PathBuf, - root_without_git_canon: PathBuf, + /// Path containing a .git and .codex subfolder. + /// For the purposes of this test, we consider this a "vulnerable" root + /// because a bad actor could write to .git/hooks/pre-commit so an + /// unsuspecting user would run code as privileged the next time they + /// ran `git commit` themselves, or modified .codex/config.toml to + /// contain `sandbox_mode = "danger-full-access"` so the agent would + /// have full privileges the next time it ran in that repo. + vulnerable_root: PathBuf, + vulnerable_root_canonical: PathBuf, + dot_git_canonical: PathBuf, + dot_codex_canonical: PathBuf, + + /// Path without .git or .codex subfolders. + empty_root: PathBuf, + /// Canonicalized version of `empty_root`. + empty_root_canonical: PathBuf, } fn populate_tmpdir(tmp: &Path) -> PopulatedTmp { - let root_with_git = tmp.join("with_git"); - let root_without_git = tmp.join("no_git"); - fs::create_dir_all(&root_with_git).expect("create with_git"); - fs::create_dir_all(&root_without_git).expect("create no_git"); - fs::create_dir_all(root_with_git.join(".git")).expect("create .git"); + let vulnerable_root = tmp.join("vulnerable_root"); + fs::create_dir_all(&vulnerable_root).expect("create vulnerable_root"); + + // TODO(mbolin): Should also support the case where `.git` is a file + // with a gitdir: ... line. + Command::new("git") + .arg("init") + .arg(".") + .current_dir(&vulnerable_root) + .output() + .expect("git init ."); + + fs::create_dir_all(vulnerable_root.join(".codex")).expect("create .codex"); + fs::write( + vulnerable_root.join(".codex").join("config.toml"), + "sandbox_mode = \"read-only\"\n", + ) + .expect("write .codex/config.toml"); + + let empty_root = tmp.join("empty_root"); + fs::create_dir_all(&empty_root).expect("create empty_root"); // Ensure we have canonical paths for -D parameter matching. - let root_with_git_canon = root_with_git.canonicalize().expect("canonicalize with_git"); - let root_with_git_git_canon = root_with_git_canon.join(".git"); - let root_without_git_canon = root_without_git + let vulnerable_root_canonical = vulnerable_root .canonicalize() - .expect("canonicalize no_git"); + .expect("canonicalize vulnerable_root"); + let dot_git_canonical = vulnerable_root_canonical.join(".git"); + let dot_codex_canonical = vulnerable_root_canonical.join(".codex"); + let empty_root_canonical = empty_root.canonicalize().expect("canonicalize empty_root"); PopulatedTmp { - root_with_git, - root_without_git, - root_with_git_canon, - root_with_git_git_canon, - root_without_git_canon, + vulnerable_root, + vulnerable_root_canonical, + dot_git_canonical, + dot_codex_canonical, + empty_root, + empty_root_canonical, } } } diff --git a/codex-rs/core/src/stream_events_utils.rs b/codex-rs/core/src/stream_events_utils.rs index 1373fdf248..2e19a3694c 100644 --- a/codex-rs/core/src/stream_events_utils.rs +++ b/codex-rs/core/src/stream_events_utils.rs @@ -39,7 +39,7 @@ pub(crate) struct HandleOutputCtx { pub cancellation_token: CancellationToken, } -#[instrument(skip_all)] +#[instrument(level = "trace", skip_all)] pub(crate) async fn handle_output_item_done( ctx: &mut HandleOutputCtx, item: ResponseItem, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index fa5433ef5e..b5aaf29999 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -159,6 +159,7 @@ impl Session { for task in self.take_all_running_tasks().await { self.handle_task_abort(task, reason.clone()).await; } + self.close_unified_exec_sessions().await; } pub async fn on_task_finished( @@ -167,12 +168,18 @@ impl Session { last_agent_message: Option, ) { let mut active = self.active_turn.lock().await; - if let Some(at) = active.as_mut() + let should_close_sessions = if let Some(at) = active.as_mut() && at.remove_task(&turn_context.sub_id) { *active = None; - } + true + } else { + false + }; drop(active); + if should_close_sessions { + self.close_unified_exec_sessions().await; + } let event = EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message }); self.send_event(turn_context.as_ref(), event).await; } @@ -196,6 +203,13 @@ impl Session { } } + async fn close_unified_exec_sessions(&self) { + self.services + .unified_exec_manager + .terminate_all_sessions() + .await; + } + async fn handle_task_abort(self: &Arc, task: RunningTask, reason: TurnAbortReason) { let sub_id = task.turn_context.sub_id.clone(); if task.cancellation_token.is_cancelled() { diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index 2ee598f21c..56c46ffbc9 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use codex_protocol::user_input::UserInput; use tokio_util::sync::CancellationToken; use tracing::Instrument; -use tracing::info_span; +use tracing::trace_span; use super::SessionTask; use super::SessionTaskContext; @@ -30,7 +30,7 @@ impl SessionTask for RegularTask { ) -> Option { let sess = session.clone_session(); let run_task_span = - info_span!(parent: sess.services.otel_manager.current_span(), "run_task"); + trace_span!(parent: sess.services.otel_manager.current_span(), "run_task"); run_task(sess, ctx, input, cancellation_token) .instrument(run_task_span) .await diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index d146507e5c..dcd3ae40ad 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -6,8 +6,8 @@ use tokio_util::either::Either; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use tracing::Instrument; -use tracing::info_span; use tracing::instrument; +use tracing::trace_span; use crate::codex::Session; use crate::codex::TurnContext; @@ -45,7 +45,7 @@ impl ToolCallRuntime { } } - #[instrument(skip_all, fields(call = ?call))] + #[instrument(level = "trace", skip_all, fields(call = ?call))] pub(crate) fn handle_tool_call( self, call: ToolCall, @@ -60,7 +60,7 @@ impl ToolCallRuntime { let lock = Arc::clone(&self.parallel_execution); let started = Instant::now(); - let dispatch_span = info_span!( + let dispatch_span = trace_span!( "dispatch_tool_call", otel.name = call.tool_name.as_str(), tool_name = call.tool_name.as_str(), diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index 66e6026b3a..9d83b5a637 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -55,7 +55,7 @@ impl ToolRouter { .any(|config| config.spec.name() == tool_name) } - #[instrument(skip_all, err)] + #[instrument(level = "trace", skip_all, err)] pub async fn build_tool_call( session: &Session, item: ResponseItem, @@ -131,7 +131,7 @@ impl ToolRouter { } } - #[instrument(skip_all, err)] + #[instrument(level = "trace", skip_all, err)] pub async fn dispatch_tool_call( &self, session: Arc, diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index dc0cc53f46..280b76dea1 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -13,6 +13,7 @@ use std::path::PathBuf; #[cfg(target_os = "linux")] use assert_cmd::cargo::cargo_bin; +pub mod process; pub mod responses; pub mod streaming_sse; pub mod test_codex; diff --git a/codex-rs/core/tests/common/process.rs b/codex-rs/core/tests/common/process.rs new file mode 100644 index 0000000000..82e42aac95 --- /dev/null +++ b/codex-rs/core/tests/common/process.rs @@ -0,0 +1,48 @@ +use anyhow::Context; +use std::fs; +use std::path::Path; +use std::time::Duration; + +pub async fn wait_for_pid_file(path: &Path) -> anyhow::Result { + let pid = tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Ok(contents) = fs::read_to_string(path) { + let trimmed = contents.trim(); + if !trimmed.is_empty() { + return trimmed.to_string(); + } + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .context("timed out waiting for pid file")?; + + Ok(pid) +} + +pub fn process_is_alive(pid: &str) -> anyhow::Result { + let status = std::process::Command::new("kill") + .args(["-0", pid]) + .status() + .context("failed to probe process liveness with kill -0")?; + Ok(status.success()) +} + +async fn wait_for_process_exit_inner(pid: String) -> anyhow::Result<()> { + loop { + if !process_is_alive(&pid)? { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } +} + +pub async fn wait_for_process_exit(pid: &str) -> anyhow::Result<()> { + let pid = pid.to_string(); + tokio::time::timeout(Duration::from_secs(2), wait_for_process_exit_inner(pid)) + .await + .context("timed out waiting for process to exit")??; + + Ok(()) +} diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index bffb601eb3..f223f4d10d 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -1009,7 +1009,6 @@ async fn auto_compact_runs_after_token_limit_hit() { ev_assistant_message("m3", AUTO_SUMMARY_TEXT), ev_completed_with_tokens("r3", 200), ]); - let sse_resume = sse(vec![ev_completed("r3-resume")]); let sse4 = sse(vec![ ev_assistant_message("m4", FINAL_REPLY), ev_completed_with_tokens("r4", 120), @@ -1038,15 +1037,6 @@ async fn auto_compact_runs_after_token_limit_hit() { }; mount_sse_once_match(&server, third_matcher, sse3).await; - let resume_marker = prefixed_auto_summary; - let resume_matcher = move |req: &wiremock::Request| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - body.contains(resume_marker) - && !body_contains_text(body, SUMMARIZATION_PROMPT) - && !body.contains(POST_AUTO_USER_MSG) - }; - mount_sse_once_match(&server, resume_matcher, sse_resume).await; - let fourth_matcher = |req: &wiremock::Request| { let body = std::str::from_utf8(&req.body).unwrap_or(""); body.contains(POST_AUTO_USER_MSG) && !body_contains_text(body, SUMMARIZATION_PROMPT) @@ -1106,8 +1096,8 @@ async fn auto_compact_runs_after_token_limit_hit() { let requests = get_responses_requests(&server).await; assert_eq!( requests.len(), - 5, - "expected user turns, a compaction request, a resumed turn, and the follow-up turn; got {}", + 4, + "expected user turns, a compaction request, and the follow-up turn; got {}", requests.len() ); let is_auto_compact = |req: &wiremock::Request| { @@ -1131,19 +1121,6 @@ async fn auto_compact_runs_after_token_limit_hit() { "auto compact should add a third request" ); - let resume_summary_marker = prefixed_auto_summary; - let resume_index = requests - .iter() - .enumerate() - .find_map(|(idx, req)| { - let body = std::str::from_utf8(&req.body).unwrap_or(""); - (body.contains(resume_summary_marker) - && !body_contains_text(body, SUMMARIZATION_PROMPT) - && !body.contains(POST_AUTO_USER_MSG)) - .then_some(idx) - }) - .expect("resume request missing after compaction"); - let follow_up_index = requests .iter() .enumerate() @@ -1154,15 +1131,12 @@ async fn auto_compact_runs_after_token_limit_hit() { .then_some(idx) }) .expect("follow-up request missing"); - assert_eq!(follow_up_index, 4, "follow-up request should be last"); + assert_eq!(follow_up_index, 3, "follow-up request should be last"); let body_first = requests[0].body_json::().unwrap(); let body_auto = requests[auto_compact_index] .body_json::() .unwrap(); - let body_resume = requests[resume_index] - .body_json::() - .unwrap(); let body_follow_up = requests[follow_up_index] .body_json::() .unwrap(); @@ -1201,23 +1175,6 @@ async fn auto_compact_runs_after_token_limit_hit() { "auto compact should send the summarization prompt as a user message", ); - let input_resume = body_resume.get("input").and_then(|v| v.as_array()).unwrap(); - assert!( - input_resume.iter().any(|item| { - item.get("type").and_then(|v| v.as_str()) == Some("message") - && item.get("role").and_then(|v| v.as_str()) == Some("user") - && item - .get("content") - .and_then(|v| v.as_array()) - .and_then(|arr| arr.first()) - .and_then(|entry| entry.get("text")) - .and_then(|v| v.as_str()) - .map(|text| text.contains(prefixed_auto_summary)) - .unwrap_or(false) - }), - "resume request should include compacted history" - ); - let input_follow_up = body_follow_up .get("input") .and_then(|v| v.as_array()) @@ -1276,6 +1233,10 @@ async fn auto_compact_persists_rollout_entries() { ev_assistant_message("m3", &auto_summary_payload), ev_completed_with_tokens("r3", 200), ]); + let sse4 = sse(vec![ + ev_assistant_message("m4", FINAL_REPLY), + ev_completed_with_tokens("r4", 120), + ]); let first_matcher = |req: &wiremock::Request| { let body = std::str::from_utf8(&req.body).unwrap_or(""); @@ -1299,12 +1260,19 @@ async fn auto_compact_persists_rollout_entries() { }; mount_sse_once_match(&server, third_matcher, sse3).await; + let fourth_matcher = |req: &wiremock::Request| { + let body = std::str::from_utf8(&req.body).unwrap_or(""); + body.contains(POST_AUTO_USER_MSG) && !body_contains_text(body, SUMMARIZATION_PROMPT) + }; + mount_sse_once_match(&server, fourth_matcher, sse4).await; + let model_provider = non_openai_model_provider(&server); let home = TempDir::new().unwrap(); let mut config = load_default_config_for_test(&home); config.model_provider = model_provider; set_test_compact_prompt(&mut config); + config.model_auto_compact_token_limit = Some(200_000); let conversation_manager = ConversationManager::with_models_provider( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), @@ -1335,6 +1303,16 @@ async fn auto_compact_persists_rollout_entries() { .unwrap(); wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: POST_AUTO_USER_MSG.into(), + }], + }) + .await + .unwrap(); + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + codex.submit(Op::Shutdown).await.unwrap(); wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await; @@ -1731,6 +1709,8 @@ async fn auto_compact_allows_multiple_attempts_when_interleaved_with_other_turn_ ev_assistant_message("m6", FINAL_REPLY), ev_completed_with_tokens("r6", 120), ]); + let follow_up_user = "FOLLOW_UP_AUTO_COMPACT"; + let final_user = "FINAL_AUTO_COMPACT"; mount_sse_sequence(&server, vec![sse1, sse2, sse3, sse4, sse5, sse6]).await; @@ -1751,31 +1731,31 @@ async fn auto_compact_allows_multiple_attempts_when_interleaved_with_other_turn_ .unwrap() .conversation; - codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: MULTI_AUTO_MSG.into(), - }], - }) - .await - .unwrap(); - let mut auto_compact_lifecycle_events = Vec::new(); - loop { - let event = codex.next_event().await.unwrap(); - if event.id.starts_with("auto-compact-") - && matches!( - event.msg, - EventMsg::TaskStarted(_) | EventMsg::TaskComplete(_) - ) - { - auto_compact_lifecycle_events.push(event); - continue; - } - if let EventMsg::TaskComplete(_) = &event.msg - && !event.id.starts_with("auto-compact-") - { - break; + for user in [MULTI_AUTO_MSG, follow_up_user, final_user] { + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { text: user.into() }], + }) + .await + .unwrap(); + + loop { + let event = codex.next_event().await.unwrap(); + if event.id.starts_with("auto-compact-") + && matches!( + event.msg, + EventMsg::TaskStarted(_) | EventMsg::TaskComplete(_) + ) + { + auto_compact_lifecycle_events.push(event); + continue; + } + if let EventMsg::TaskComplete(_) = &event.msg + && !event.id.starts_with("auto-compact-") + { + break; + } } } @@ -1821,6 +1801,7 @@ async fn auto_compact_triggers_after_function_call_over_95_percent_usage() { let context_window = 100; let limit = context_window * 90 / 100; let over_limit_tokens = context_window * 95 / 100 + 1; + let follow_up_user = "FOLLOW_UP_AFTER_LIMIT"; let first_turn = sse(vec![ ev_function_call(DUMMY_CALL_ID, DUMMY_FUNCTION_NAME, "{}"), @@ -1873,6 +1854,17 @@ async fn auto_compact_triggers_after_function_call_over_95_percent_usage() { wait_for_event(&codex, |msg| matches!(msg, EventMsg::TaskComplete(_))).await; + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: follow_up_user.into(), + }], + }) + .await + .unwrap(); + + wait_for_event(&codex, |msg| matches!(msg, EventMsg::TaskComplete(_))).await; + // Assert first request captured expected user message that triggers function call. let first_request = first_turn_mock.single_request().input(); assert!( @@ -1916,6 +1908,7 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { let first_user = "COUNT_PRE_LAST_REASONING"; let second_user = "TRIGGER_COMPACT_AT_LIMIT"; + let third_user = "AFTER_REMOTE_COMPACT"; let pre_last_reasoning_content = "a".repeat(2_400); let post_last_reasoning_content = "b".repeat(4_000); @@ -1928,7 +1921,7 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { ev_reasoning_item("post-reasoning", &["post"], &[&post_last_reasoning_content]), ev_completed_with_tokens("r2", 80), ]); - let resume_turn = sse(vec![ + let third_turn = sse(vec![ ev_assistant_message("m4", FINAL_REPLY), ev_completed_with_tokens("r4", 1), ]); @@ -1940,8 +1933,8 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { first_turn, // Turn 2: reasoning after last user (should be ignored for compaction). second_turn, - // Turn 3: resume after remote compaction. - resume_turn, + // Turn 3: next user turn after remote compaction. + third_turn, ], ) .await; @@ -1973,7 +1966,10 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { .expect("build codex") .codex; - for (idx, user) in [first_user, second_user].into_iter().enumerate() { + for (idx, user) in [first_user, second_user, third_user] + .into_iter() + .enumerate() + { codex .submit(Op::UserInput { items: vec![UserInput::Text { text: user.into() }], @@ -1982,10 +1978,10 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { .unwrap(); wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; - if idx == 0 { + if idx < 2 { assert!( compact_mock.requests().is_empty(), - "remote compaction should not run after the first turn" + "remote compaction should not run before the next user turn" ); } } @@ -2006,20 +2002,21 @@ async fn auto_compact_counts_encrypted_reasoning_before_last_user() { assert_eq!( requests.len(), 3, - "conversation should include two user turns and a post-compaction resume" + "conversation should include three user turns" ); let second_request_body = requests[1].body_json().to_string(); assert!( !second_request_body.contains("REMOTE_COMPACT_SUMMARY"), "second turn should not include compacted history" ); - let resume_body = requests[2].body_json().to_string(); + let third_request_body = requests[2].body_json().to_string(); assert!( - resume_body.contains("REMOTE_COMPACT_SUMMARY") || resume_body.contains(FINAL_REPLY), - "resume request should follow remote compact and use compacted history" + third_request_body.contains("REMOTE_COMPACT_SUMMARY") + || third_request_body.contains(FINAL_REPLY), + "third turn should include compacted history" ); assert!( - resume_body.contains("ENCRYPTED_COMPACTION_SUMMARY"), - "resume request should include compaction summary item" + third_request_body.contains("ENCRYPTED_COMPACTION_SUMMARY"), + "third turn should include compaction summary item" ); } diff --git a/codex-rs/core/tests/suite/otel.rs b/codex-rs/core/tests/suite/otel.rs index 65e96a4fec..922c7b9cfc 100644 --- a/codex-rs/core/tests/suite/otel.rs +++ b/codex-rs/core/tests/suite/otel.rs @@ -25,6 +25,7 @@ use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use std::sync::Mutex; +use tracing::Level; use tracing_test::traced_test; use tracing_subscriber::fmt::format::FmtSpan; @@ -454,6 +455,7 @@ async fn handle_responses_span_records_response_kind_and_tool_name() { let subscriber = tracing_subscriber::fmt() .with_level(true) .with_ansi(false) + .with_max_level(Level::TRACE) .with_span_events(FmtSpan::FULL) .with_writer(MockWriter::new(buffer)) .finish(); @@ -517,6 +519,7 @@ async fn record_responses_sets_span_fields_for_response_events() { let subscriber = tracing_subscriber::fmt() .with_level(true) .with_ansi(false) + .with_max_level(Level::TRACE) .with_span_events(FmtSpan::FULL) .with_writer(MockWriter::new(buffer)) .finish(); diff --git a/codex-rs/core/tests/suite/unified_exec.rs b/codex-rs/core/tests/suite/unified_exec.rs index af5e4a1497..c9e5f0e051 100644 --- a/codex-rs/core/tests/suite/unified_exec.rs +++ b/codex-rs/core/tests/suite/unified_exec.rs @@ -14,6 +14,8 @@ use codex_core::protocol::SandboxPolicy; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::user_input::UserInput; use core_test_support::assert_regex_match; +use core_test_support::process::wait_for_pid_file; +use core_test_support::process::wait_for_process_exit; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; @@ -31,6 +33,7 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_event_match; use core_test_support::wait_for_event_with_timeout; +use pretty_assertions::assert_eq; use regex_lite::Regex; use serde_json::Value; use serde_json::json; @@ -1640,6 +1643,111 @@ async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<() Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unified_exec_closes_long_running_session_at_turn_end() -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_sandbox!(Ok(())); + skip_if_windows!(Ok(())); + + let server = start_mock_server().await; + + let mut builder = test_codex().with_config(|config| { + config.use_experimental_unified_exec_tool = true; + config.features.enable(Feature::UnifiedExec); + }); + let TestCodex { + codex, + cwd, + session_configured, + .. + } = builder.build(&server).await?; + + let temp_dir = tempfile::tempdir()?; + let pid_path = temp_dir.path().join("uexec_pid"); + let pid_path_str = pid_path.to_string_lossy(); + + let call_id = "uexec-long-running"; + let command = format!("printf '%s' $$ > '{pid_path_str}' && exec sleep 3000"); + let args = json!({ + "cmd": command, + "yield_time_ms": 250, + }); + + let responses = vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_assistant_message("msg-1", "done"), + ev_completed("resp-2"), + ]), + ]; + mount_sse_sequence(&server, responses).await; + + let session_model = session_configured.model.clone(); + + codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "close unified exec sessions on turn end".into(), + }], + final_output_json_schema: None, + cwd: cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::DangerFullAccess, + model: session_model, + effort: None, + summary: ReasoningSummary::Auto, + }) + .await?; + + let begin_event = wait_for_event_match(&codex, |msg| match msg { + EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => Some(ev.clone()), + _ => None, + }) + .await; + + let begin_process_id = begin_event + .process_id + .clone() + .expect("expected process_id for long-running unified exec session"); + + let pid = wait_for_pid_file(&pid_path).await?; + assert!( + pid.chars().all(|ch| ch.is_ascii_digit()), + "expected numeric pid, got {pid:?}" + ); + + let mut end_event = None; + let mut task_complete = false; + loop { + let msg = wait_for_event(&codex, |_| true).await; + match msg { + EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => end_event = Some(ev), + EventMsg::TaskComplete(_) => task_complete = true, + _ => {} + } + if task_complete && end_event.is_some() { + break; + } + } + + let end_event = end_event.expect("expected ExecCommandEnd event for unified exec session"); + assert_eq!(end_event.call_id, call_id); + let end_process_id = end_event + .process_id + .clone() + .expect("expected process_id in unified exec end event"); + assert_eq!(end_process_id, begin_process_id); + + wait_for_process_exit(&pid).await?; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn unified_exec_reuses_session_via_stdin() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/otel/src/otel_manager.rs b/codex-rs/otel/src/otel_manager.rs index 268897f8ae..33750d83c5 100644 --- a/codex-rs/otel/src/otel_manager.rs +++ b/codex-rs/otel/src/otel_manager.rs @@ -25,7 +25,7 @@ use std::time::Instant; use strum_macros::Display; use tokio::time::error::Elapsed; use tracing::Span; -use tracing::info_span; +use tracing::trace_span; use tracing_opentelemetry::OpenTelemetrySpanExt; #[derive(Debug, Clone, Serialize, Display)] @@ -67,7 +67,7 @@ impl OtelManager { terminal_type: String, session_source: SessionSource, ) -> OtelManager { - let session_span = info_span!("new_session", conversation_id = %conversation_id, session_source = %session_source); + let session_span = trace_span!("new_session", conversation_id = %conversation_id, session_source = %session_source); if let Some(context) = traceparent_context_from_env() { session_span.set_parent(context); diff --git a/codex-rs/otel/src/otel_provider.rs b/codex-rs/otel/src/otel_provider.rs index 8e2826f834..b9d9559325 100644 --- a/codex-rs/otel/src/otel_provider.rs +++ b/codex-rs/otel/src/otel_provider.rs @@ -134,7 +134,7 @@ impl OtelProvider { self.tracer.as_ref().map(|tracer| { tracing_opentelemetry::layer() .with_tracer(tracer.clone()) - .with_filter(LevelFilter::INFO) + .with_filter(LevelFilter::TRACE) }) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index c333d431ca..201aba07ef 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -306,12 +306,14 @@ pub enum SandboxPolicy { /// A writable root path accompanied by a list of subpaths that should remain /// read‑only even when the root is writable. This is primarily used to ensure -/// top‑level VCS metadata directories (e.g. `.git`) under a writable root are -/// not modified by the agent. +/// that folders containing files that could be modified to escalate the +/// privileges of the agent (e.g. `.codex`, `.git`, notably `.git/hooks`) under +/// a writable root are not modified by the agent. #[derive(Debug, Clone, PartialEq, Eq, JsonSchema)] pub struct WritableRoot { pub root: AbsolutePathBuf, + /// By construction, these subpaths are all under `root`. pub read_only_subpaths: Vec, } @@ -458,6 +460,13 @@ impl SandboxPolicy { if top_level_git.as_path().is_dir() { subpaths.push(top_level_git); } + #[allow(clippy::expect_used)] + let top_level_codex = writable_root + .join(".codex") + .expect(".codex is a valid relative path"); + if top_level_codex.as_path().is_dir() { + subpaths.push(top_level_codex); + } WritableRoot { root: writable_root, read_only_subpaths: subpaths, diff --git a/codex-rs/tui/Cargo.toml b/codex-rs/tui/Cargo.toml index 1d9f311591..06167c973a 100644 --- a/codex-rs/tui/Cargo.toml +++ b/codex-rs/tui/Cargo.toml @@ -23,7 +23,6 @@ workspace = true [dependencies] anyhow = { workspace = true } -async-stream = { workspace = true } base64 = { workspace = true } chrono = { workspace = true, features = ["serde"] } clap = { workspace = true, features = ["derive"] } @@ -81,7 +80,7 @@ tokio = { workspace = true, features = [ "test-util", "time", ] } -tokio-stream = { workspace = true } +tokio-stream = { workspace = true, features = ["sync"] } toml = { workspace = true } tracing = { workspace = true, features = ["log"] } tracing-appender = { workspace = true } diff --git a/codex-rs/tui/src/tui.rs b/codex-rs/tui/src/tui.rs index f9566da908..0770b76648 100644 --- a/codex-rs/tui/src/tui.rs +++ b/codex-rs/tui/src/tui.rs @@ -16,7 +16,6 @@ use crossterm::event::DisableBracketedPaste; use crossterm::event::DisableFocusChange; use crossterm::event::EnableBracketedPaste; use crossterm::event::EnableFocusChange; -use crossterm::event::Event; use crossterm::event::KeyEvent; use crossterm::event::KeyboardEnhancementFlags; use crossterm::event::PopKeyboardEnhancementFlags; @@ -32,7 +31,6 @@ use ratatui::crossterm::terminal::enable_raw_mode; use ratatui::layout::Offset; use ratatui::layout::Rect; use ratatui::text::Line; -use tokio::select; use tokio::sync::broadcast; use tokio_stream::Stream; @@ -42,11 +40,12 @@ use crate::custom_terminal::Terminal as CustomTerminal; use crate::notifications::DesktopNotificationBackend; use crate::notifications::NotificationBackendKind; use crate::notifications::detect_backend; -#[cfg(unix)] -use crate::tui::job_control::SUSPEND_KEY; +use crate::tui::event_stream::EventBroker; +use crate::tui::event_stream::TuiEventStream; #[cfg(unix)] use crate::tui::job_control::SuspendContext; +mod event_stream; mod frame_requester; #[cfg(unix)] mod job_control; @@ -156,7 +155,7 @@ fn set_panic_hook() { })); } -#[derive(Debug)] +#[derive(Clone, Debug)] pub enum TuiEvent { Key(KeyEvent), Paste(String), @@ -166,6 +165,7 @@ pub enum TuiEvent { pub struct Tui { frame_requester: FrameRequester, draw_tx: broadcast::Sender<()>, + event_broker: Arc, pub(crate) terminal: Terminal, pending_history_lines: Vec>, alt_saved_viewport: Option, @@ -194,6 +194,7 @@ impl Tui { Self { frame_requester, draw_tx, + event_broker: Arc::new(EventBroker::new()), terminal, pending_history_lines: vec![], alt_saved_viewport: None, @@ -214,6 +215,18 @@ impl Tui { self.enhanced_keys_supported } + // todo(sayan) unused for now; intend to use to enable opening external editors + #[allow(unused)] + pub fn pause_events(&mut self) { + self.event_broker.pause_events(); + } + + // todo(sayan) unused for now; intend to use to enable opening external editors + #[allow(unused)] + pub fn resume_events(&mut self) { + self.event_broker.resume_events(); + } + /// Emit a desktop notification now if the terminal is unfocused. /// Returns true if a notification was posted. pub fn notify(&mut self, message: impl AsRef) -> bool { @@ -262,79 +275,21 @@ impl Tui { } pub fn event_stream(&self) -> Pin + Send + 'static>> { - use tokio_stream::StreamExt; - - let mut crossterm_events = crossterm::event::EventStream::new(); - let mut draw_rx = self.draw_tx.subscribe(); - - // State for tracking how we should resume from ^Z suspend. #[cfg(unix)] - let suspend_context = self.suspend_context.clone(); - #[cfg(unix)] - let alt_screen_active = self.alt_screen_active.clone(); - - let terminal_focused = self.terminal_focused.clone(); - let event_stream = async_stream::stream! { - loop { - select! { - event_result = crossterm_events.next() => { - match event_result { - Some(Ok(event)) => { - match event { - Event::Key(key_event) => { - #[cfg(unix)] - if SUSPEND_KEY.is_press(key_event) { - let _ = suspend_context.suspend(&alt_screen_active); - // We continue here after resume. - yield TuiEvent::Draw; - continue; - } - yield TuiEvent::Key(key_event); - } - Event::Resize(_, _) => { - yield TuiEvent::Draw; - } - Event::Paste(pasted) => { - yield TuiEvent::Paste(pasted); - } - Event::FocusGained => { - terminal_focused.store(true, Ordering::Relaxed); - crate::terminal_palette::requery_default_colors(); - yield TuiEvent::Draw; - } - Event::FocusLost => { - terminal_focused.store(false, Ordering::Relaxed); - } - _ => {} - } - } - Some(Err(_)) | None => { - // Exit the loop in case of broken pipe as we will never - // recover from it - break; - } - } - } - result = draw_rx.recv() => { - match result { - Ok(_) => { - yield TuiEvent::Draw; - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - // We dropped one or more draw notifications; coalesce to a single draw. - yield TuiEvent::Draw; - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - // Sender dropped. This stream likely outlived its owning `Tui`; - // exit to avoid spinning on a permanently-closed receiver. - break; - } - } - } - } - } - }; - Box::pin(event_stream) + let stream = TuiEventStream::new( + self.event_broker.clone(), + self.draw_tx.subscribe(), + self.terminal_focused.clone(), + self.suspend_context.clone(), + self.alt_screen_active.clone(), + ); + #[cfg(not(unix))] + let stream = TuiEventStream::new( + self.event_broker.clone(), + self.draw_tx.subscribe(), + self.terminal_focused.clone(), + ); + Box::pin(stream) } /// Enter alternate screen and expand the viewport to full terminal size, saving the current diff --git a/codex-rs/tui/src/tui/event_stream.rs b/codex-rs/tui/src/tui/event_stream.rs new file mode 100644 index 0000000000..2ce0aa7d2c --- /dev/null +++ b/codex-rs/tui/src/tui/event_stream.rs @@ -0,0 +1,511 @@ +//! Event stream plumbing for the TUI. +//! +//! - [`EventBroker`] holds the shared crossterm stream so multiple callers reuse the same +//! input source and can drop/recreate it on pause/resume without rebuilding consumers. +//! - [`TuiEventStream`] wraps a draw event subscription plus the shared [`EventBroker`] and maps crossterm +//! events into [`TuiEvent`]. +//! - [`EventSource`] abstracts the underlying event producer; the real implementation is +//! [`CrosstermEventSource`] and tests can swap in [`FakeEventSource`]. +//! +//! The motivation for dropping/recreating the crossterm event stream is to enable the TUI to fully relinquish stdin. +//! If the stream is not dropped, it will continue to read from stdin even if it is not actively being polled +//! (due to how crossterm's EventStream is implemented), potentially stealing input from other processes reading stdin, +//! like terminal text editors. This race can cause missed input or capturing terminal query responses (for example, OSC palette/size queries) +//! that the other process expects to read. Stopping polling, instead of dropping the stream, is only sufficient when the +//! pause happens before the stream enters a pending state; otherwise the crossterm reader thread may keep reading +//! from stdin, so the safer approach is to drop and recreate the event stream when we need to hand off the terminal. +//! +//! See https://ratatui.rs/recipes/apps/spawn-vim/ and https://www.reddit.com/r/rust/comments/1f3o33u/myterious_crossterm_input_after_running_vim for more details. + +use std::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; + +use crossterm::event::Event; +use tokio::sync::broadcast; +use tokio::sync::watch; +use tokio_stream::Stream; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::WatchStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; + +use super::TuiEvent; + +/// Result type produced by an event source. +pub type EventResult = std::io::Result; + +/// Abstraction over a source of terminal events. Allows swapping in a fake for tests. +/// Value in production is [`CrosstermEventSource`]. +pub trait EventSource: Send + 'static { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll>; +} + +/// Shared crossterm input state for all [`TuiEventStream`] instances. A single crossterm EventStream +/// is reused so all streams still see the same input source. +/// +/// This intermediate layer enables dropping/recreating the underlying EventStream (pause/resume) without rebuilding consumers. +pub struct EventBroker { + state: Mutex>, + resume_events_tx: watch::Sender<()>, +} + +/// Tracks state of underlying [`EventSource`]. +enum EventBrokerState { + Paused, // Underlying event source (i.e., crossterm EventStream) dropped + Start, // A new event source will be created on next poll + Running(S), // Event source is currently running +} + +impl EventBrokerState { + /// Return the running event source, starting it if needed; None when paused. + fn active_event_source_mut(&mut self) -> Option<&mut S> { + match self { + EventBrokerState::Paused => None, + EventBrokerState::Start => { + *self = EventBrokerState::Running(S::default()); + match self { + EventBrokerState::Running(events) => Some(events), + EventBrokerState::Paused | EventBrokerState::Start => unreachable!(), + } + } + EventBrokerState::Running(events) => Some(events), + } + } +} + +impl EventBroker { + pub fn new() -> Self { + let (resume_events_tx, _resume_events_rx) = watch::channel(()); + Self { + state: Mutex::new(EventBrokerState::Start), + resume_events_tx, + } + } + + /// Drop the underlying event source + pub fn pause_events(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *state = EventBrokerState::Paused; + } + + /// Create a new instance of the underlying event source + pub fn resume_events(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *state = EventBrokerState::Start; + let _ = self.resume_events_tx.send(()); + } + + /// Subscribe to a notification that fires whenever [`Self::resume_events`] is called. + /// + /// This is used to wake `poll_crossterm_event` when it is paused and waiting for the + /// underlying crossterm stream to be recreated. + pub fn resume_events_rx(&self) -> watch::Receiver<()> { + self.resume_events_tx.subscribe() + } +} + +/// Real crossterm-backed event source. +pub struct CrosstermEventSource(pub crossterm::event::EventStream); + +impl Default for CrosstermEventSource { + fn default() -> Self { + Self(crossterm::event::EventStream::new()) + } +} + +impl EventSource for CrosstermEventSource { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_next(cx) + } +} + +/// TuiEventStream is a struct for reading TUI events (draws and user input). +/// Each instance has its own draw subscription (the draw channel is broadcast, so +/// multiple receivers are fine), while crossterm input is funneled through a +/// single shared [`EventBroker`] because crossterm uses a global stdin reader and +/// does not support fan-out. Multiple TuiEventStream instances can exist during the app lifetime +/// (for nested or sequential screens), but only one should be polled at a time, +/// otherwise one instance can consume ("steal") input events and the other will miss them. +pub struct TuiEventStream { + broker: Arc>, + draw_stream: BroadcastStream<()>, + resume_stream: WatchStream<()>, + terminal_focused: Arc, + poll_draw_first: bool, + #[cfg(unix)] + suspend_context: crate::tui::job_control::SuspendContext, + #[cfg(unix)] + alt_screen_active: Arc, +} + +impl TuiEventStream { + pub fn new( + broker: Arc>, + draw_rx: broadcast::Receiver<()>, + terminal_focused: Arc, + #[cfg(unix)] suspend_context: crate::tui::job_control::SuspendContext, + #[cfg(unix)] alt_screen_active: Arc, + ) -> Self { + let resume_stream = WatchStream::from_changes(broker.resume_events_rx()); + Self { + broker, + draw_stream: BroadcastStream::new(draw_rx), + resume_stream, + terminal_focused, + poll_draw_first: false, + #[cfg(unix)] + suspend_context, + #[cfg(unix)] + alt_screen_active, + } + } + + /// Poll the shared crossterm stream for the next mapped `TuiEvent`. + /// + /// This skips events we don't use (mouse events, etc.) and keeps polling until it yields + /// a mapped event, hits `Pending`, or sees EOF/error. When the broker is paused, it drops + /// the underlying stream and returns `Pending` to fully release stdin. + pub fn poll_crossterm_event(&mut self, cx: &mut Context<'_>) -> Poll> { + // Some crossterm events map to None (e.g. FocusLost, mouse); loop so we keep polling + // until we return a mapped event, hit Pending, or see EOF/error. + loop { + let poll_result = { + let mut state = self + .broker + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let events = match state.active_event_source_mut() { + Some(events) => events, + None => { + drop(state); + // Poll resume_stream so resume_events wakes a stream paused here + match Pin::new(&mut self.resume_stream).poll_next(cx) { + Poll::Ready(Some(())) => continue, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Pending => return Poll::Pending, + } + } + }; + match Pin::new(events).poll_next(cx) { + Poll::Ready(Some(Ok(event))) => Some(event), + Poll::Ready(Some(Err(_))) | Poll::Ready(None) => { + *state = EventBrokerState::Start; + return Poll::Ready(None); + } + Poll::Pending => { + drop(state); + // Poll resume_stream so resume_events can wake us even while waiting on stdin + match Pin::new(&mut self.resume_stream).poll_next(cx) { + Poll::Ready(Some(())) => continue, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Pending => return Poll::Pending, + } + } + } + }; + + if let Some(mapped) = poll_result.and_then(|event| self.map_crossterm_event(event)) { + return Poll::Ready(Some(mapped)); + } + } + } + + /// Poll the draw broadcast stream for the next draw event. Draw events are used to trigger a redraw of the TUI. + pub fn poll_draw_event(&mut self, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.draw_stream).poll_next(cx) { + Poll::Ready(Some(Ok(()))) => Poll::Ready(Some(TuiEvent::Draw)), + Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(_)))) => { + Poll::Ready(Some(TuiEvent::Draw)) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } + + /// Map a crossterm event to a [`TuiEvent`], skipping events we don't use (mouse events, etc.). + fn map_crossterm_event(&mut self, event: Event) -> Option { + match event { + Event::Key(key_event) => { + #[cfg(unix)] + if crate::tui::job_control::SUSPEND_KEY.is_press(key_event) { + let _ = self.suspend_context.suspend(&self.alt_screen_active); + return Some(TuiEvent::Draw); + } + Some(TuiEvent::Key(key_event)) + } + Event::Resize(_, _) => Some(TuiEvent::Draw), + Event::Paste(pasted) => Some(TuiEvent::Paste(pasted)), + Event::FocusGained => { + self.terminal_focused.store(true, Ordering::Relaxed); + crate::terminal_palette::requery_default_colors(); + Some(TuiEvent::Draw) + } + Event::FocusLost => { + self.terminal_focused.store(false, Ordering::Relaxed); + None + } + _ => None, + } + } +} + +impl Unpin for TuiEventStream {} + +impl Stream for TuiEventStream { + type Item = TuiEvent; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + // approximate fairness + no starvation via round-robin. + let draw_first = self.poll_draw_first; + self.poll_draw_first = !self.poll_draw_first; + + if draw_first { + if let Poll::Ready(event) = self.poll_draw_event(cx) { + return Poll::Ready(event); + } + if let Poll::Ready(event) = self.poll_crossterm_event(cx) { + return Poll::Ready(event); + } + } else { + if let Poll::Ready(event) = self.poll_crossterm_event(cx) { + return Poll::Ready(event); + } + if let Poll::Ready(event) = self.poll_draw_event(cx) { + return Poll::Ready(event); + } + } + + Poll::Pending + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crossterm::event::Event; + use crossterm::event::KeyCode; + use crossterm::event::KeyEvent; + use crossterm::event::KeyModifiers; + use pretty_assertions::assert_eq; + use std::task::Context; + use std::task::Poll; + use std::time::Duration; + use tokio::sync::broadcast; + use tokio::sync::mpsc; + use tokio::time::timeout; + use tokio_stream::StreamExt; + + /// Simple fake event source for tests; feed events via the handle. + struct FakeEventSource { + rx: mpsc::UnboundedReceiver, + tx: mpsc::UnboundedSender, + } + + struct FakeEventSourceHandle { + broker: Arc>, + } + + impl FakeEventSource { + fn new() -> Self { + let (tx, rx) = mpsc::unbounded_channel(); + Self { rx, tx } + } + } + + impl Default for FakeEventSource { + fn default() -> Self { + Self::new() + } + } + + impl FakeEventSourceHandle { + fn new(broker: Arc>) -> Self { + Self { broker } + } + + fn send(&self, event: EventResult) { + let mut state = self + .broker + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(source) = state.active_event_source_mut() else { + return; + }; + let _ = source.tx.send(event); + } + } + + impl EventSource for FakeEventSource { + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().rx).poll_recv(cx) + } + } + + fn make_stream( + broker: Arc>, + draw_rx: broadcast::Receiver<()>, + terminal_focused: Arc, + ) -> TuiEventStream { + TuiEventStream::new( + broker, + draw_rx, + terminal_focused, + #[cfg(unix)] + crate::tui::job_control::SuspendContext::new(), + #[cfg(unix)] + Arc::new(AtomicBool::new(false)), + ) + } + + type SetupState = ( + Arc>, + FakeEventSourceHandle, + broadcast::Sender<()>, + broadcast::Receiver<()>, + Arc, + ); + + fn setup() -> SetupState { + let source = FakeEventSource::new(); + let broker = Arc::new(EventBroker::new()); + *broker.state.lock().unwrap() = EventBrokerState::Running(source); + let handle = FakeEventSourceHandle::new(broker.clone()); + + let (draw_tx, draw_rx) = broadcast::channel(1); + let terminal_focused = Arc::new(AtomicBool::new(true)); + (broker, handle, draw_tx, draw_rx, terminal_focused) + } + + #[tokio::test(flavor = "current_thread")] + async fn key_event_skips_unmapped() { + let (broker, handle, _draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker, draw_rx, terminal_focused); + + handle.send(Ok(Event::FocusLost)); + handle.send(Ok(Event::Key(KeyEvent::new( + KeyCode::Char('a'), + KeyModifiers::NONE, + )))); + + let next = stream.next().await.unwrap(); + match next { + TuiEvent::Key(key) => { + assert_eq!(key, KeyEvent::new(KeyCode::Char('a'), KeyModifiers::NONE)); + } + other => panic!("expected key event, got {other:?}"), + } + } + + #[tokio::test(flavor = "current_thread")] + async fn draw_and_key_events_yield_both() { + let (broker, handle, draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker, draw_rx, terminal_focused); + + let expected_key = KeyEvent::new(KeyCode::Char('a'), KeyModifiers::NONE); + let _ = draw_tx.send(()); + handle.send(Ok(Event::Key(expected_key))); + + let first = stream.next().await.unwrap(); + let second = stream.next().await.unwrap(); + + let mut saw_draw = false; + let mut saw_key = false; + for event in [first, second] { + match event { + TuiEvent::Draw => { + saw_draw = true; + } + TuiEvent::Key(key) => { + assert_eq!(key, expected_key); + saw_key = true; + } + other => panic!("expected draw or key event, got {other:?}"), + } + } + + assert!(saw_draw && saw_key, "expected both draw and key events"); + } + + #[tokio::test(flavor = "current_thread")] + async fn lagged_draw_maps_to_draw() { + let (broker, _handle, draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker, draw_rx.resubscribe(), terminal_focused); + + // Fill channel to force Lagged on the receiver. + let _ = draw_tx.send(()); + let _ = draw_tx.send(()); + + let first = stream.next().await; + assert!(matches!(first, Some(TuiEvent::Draw))); + } + + #[tokio::test(flavor = "current_thread")] + async fn error_or_eof_ends_stream() { + let (broker, handle, _draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker, draw_rx, terminal_focused); + + handle.send(Err(std::io::Error::other("boom"))); + + let next = stream.next().await; + assert!(next.is_none()); + } + + #[tokio::test(flavor = "current_thread")] + async fn resume_wakes_paused_stream() { + let (broker, handle, _draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker.clone(), draw_rx, terminal_focused); + + broker.pause_events(); + + let task = tokio::spawn(async move { stream.next().await }); + tokio::task::yield_now().await; + + broker.resume_events(); + let expected_key = KeyEvent::new(KeyCode::Char('r'), KeyModifiers::NONE); + handle.send(Ok(Event::Key(expected_key))); + + let event = timeout(Duration::from_millis(100), task) + .await + .expect("timed out waiting for resumed event") + .expect("join failed"); + match event { + Some(TuiEvent::Key(key)) => assert_eq!(key, expected_key), + other => panic!("expected key event, got {other:?}"), + } + } + + #[tokio::test(flavor = "current_thread")] + async fn resume_wakes_pending_stream() { + let (broker, handle, _draw_tx, draw_rx, terminal_focused) = setup(); + let mut stream = make_stream(broker.clone(), draw_rx, terminal_focused); + + let task = tokio::spawn(async move { stream.next().await }); + tokio::task::yield_now().await; + + broker.pause_events(); + broker.resume_events(); + let expected_key = KeyEvent::new(KeyCode::Char('p'), KeyModifiers::NONE); + handle.send(Ok(Event::Key(expected_key))); + + let event = timeout(Duration::from_millis(100), task) + .await + .expect("timed out waiting for resumed event") + .expect("join failed"); + match event { + Some(TuiEvent::Key(key)) => assert_eq!(key, expected_key), + other => panic!("expected key event, got {other:?}"), + } + } +} diff --git a/codex-rs/tui2/src/app.rs b/codex-rs/tui2/src/app.rs index 032a228ec4..f89938f27f 100644 --- a/codex-rs/tui2/src/app.rs +++ b/codex-rs/tui2/src/app.rs @@ -21,6 +21,8 @@ use crate::skill_error_prompt::SkillErrorPromptOutcome; use crate::skill_error_prompt::run_skill_error_prompt; use crate::tui; use crate::tui::TuiEvent; +use crate::tui::scrolling::TranscriptLineMeta; +use crate::tui::scrolling::TranscriptScroll; use crate::update_action::UpdateAction; use crate::wrapping::RtOptions; use crate::wrapping::word_wrap_line; @@ -339,21 +341,6 @@ pub(crate) struct App { skip_world_writable_scan_once: bool, } -/// Scroll state for the inline transcript viewport. -/// -/// This tracks whether the transcript is pinned to the latest line or anchored -/// at a specific cell/line pair so later viewport changes can implement -/// scrollback without losing the notion of "bottom". -#[allow(dead_code)] -#[derive(Debug, Clone, Copy, Default)] -enum TranscriptScroll { - #[default] - ToBottom, - Scrolled { - cell_index: usize, - line_in_cell: usize, - }, -} /// Content-relative selection within the inline transcript viewport. /// /// Selection endpoints are expressed in terms of flattened, wrapped transcript @@ -494,7 +481,7 @@ impl App { file_search, enhanced_keys_supported, transcript_cells: Vec::new(), - transcript_scroll: TranscriptScroll::ToBottom, + transcript_scroll: TranscriptScroll::default(), transcript_selection: TranscriptSelection::default(), transcript_view_top: 0, transcript_total_lines: 0, @@ -562,13 +549,13 @@ impl App { let session_lines = if width == 0 { Vec::new() } else { - let (lines, meta) = Self::build_transcript_lines(&app.transcript_cells, width); + let (lines, line_meta) = Self::build_transcript_lines(&app.transcript_cells, width); let is_user_cell: Vec = app .transcript_cells .iter() .map(|cell| cell.as_any().is::()) .collect(); - Self::render_lines_to_ansi(&lines, &meta, &is_user_cell, width) + Self::render_lines_to_ansi(&lines, &line_meta, &is_user_cell, width) }; tui.terminal.clear()?; @@ -676,7 +663,7 @@ impl App { ) -> u16 { let area = frame.area(); if area.width == 0 || area.height == 0 { - self.transcript_scroll = TranscriptScroll::ToBottom; + self.transcript_scroll = TranscriptScroll::default(); self.transcript_view_top = 0; self.transcript_total_lines = 0; return area.bottom().saturating_sub(chat_height); @@ -685,7 +672,7 @@ impl App { let chat_height = chat_height.min(area.height); let max_transcript_height = area.height.saturating_sub(chat_height); if max_transcript_height == 0 { - self.transcript_scroll = TranscriptScroll::ToBottom; + self.transcript_scroll = TranscriptScroll::default(); self.transcript_view_top = 0; self.transcript_total_lines = 0; return area.y; @@ -698,10 +685,10 @@ impl App { height: max_transcript_height, }; - let (lines, meta) = Self::build_transcript_lines(cells, transcript_area.width); + let (lines, line_meta) = Self::build_transcript_lines(cells, transcript_area.width); if lines.is_empty() { Clear.render_ref(transcript_area, frame.buffer); - self.transcript_scroll = TranscriptScroll::ToBottom; + self.transcript_scroll = TranscriptScroll::default(); self.transcript_view_top = 0; self.transcript_total_lines = 0; return area.y; @@ -709,7 +696,7 @@ impl App { let wrapped = word_wrap_lines_borrowed(&lines, transcript_area.width.max(1) as usize); if wrapped.is_empty() { - self.transcript_scroll = TranscriptScroll::ToBottom; + self.transcript_scroll = TranscriptScroll::default(); self.transcript_view_top = 0; self.transcript_total_lines = 0; return area.y; @@ -731,10 +718,10 @@ impl App { .initial_indent(base_opts.subsequent_indent.clone()) }; let seg_count = word_wrap_line(line, opts).len(); - let is_user_row = meta + let is_user_row = line_meta .get(idx) - .and_then(Option::as_ref) - .map(|(cell_index, _)| is_user_cell.get(*cell_index).copied().unwrap_or(false)) + .and_then(TranscriptLineMeta::cell_index) + .map(|cell_index| is_user_cell.get(cell_index).copied().unwrap_or(false)) .unwrap_or(false); wrapped_is_user_row.extend(std::iter::repeat_n(is_user_row, seg_count)); first = false; @@ -745,30 +732,8 @@ impl App { let max_visible = std::cmp::min(max_transcript_height as usize, total_lines); let max_start = total_lines.saturating_sub(max_visible); - let top_offset = match self.transcript_scroll { - TranscriptScroll::ToBottom => max_start, - TranscriptScroll::Scrolled { - cell_index, - line_in_cell, - } => { - let mut anchor = None; - for (idx, entry) in meta.iter().enumerate() { - if let Some((ci, li)) = entry - && *ci == cell_index - && *li == line_in_cell - { - anchor = Some(idx); - break; - } - } - if let Some(idx) = anchor { - idx.min(max_start) - } else { - self.transcript_scroll = TranscriptScroll::ToBottom; - max_start - } - } - }; + let (scroll_state, top_offset) = self.transcript_scroll.resolve_top(&line_meta, max_start); + self.transcript_scroll = scroll_state; self.transcript_view_top = top_offset; let transcript_visible_height = max_visible as u16; @@ -974,69 +939,10 @@ impl App { return; } - let (lines, meta) = Self::build_transcript_lines(&self.transcript_cells, width); - let total_lines = lines.len(); - if total_lines <= visible_lines { - self.transcript_scroll = TranscriptScroll::ToBottom; - return; - } - - let max_start = total_lines.saturating_sub(visible_lines); - - let current_top = match self.transcript_scroll { - TranscriptScroll::ToBottom => max_start, - TranscriptScroll::Scrolled { - cell_index, - line_in_cell, - } => { - let mut anchor = None; - for (idx, entry) in meta.iter().enumerate() { - if let Some((ci, li)) = entry - && *ci == cell_index - && *li == line_in_cell - { - anchor = Some(idx); - break; - } - } - anchor.unwrap_or(max_start).min(max_start) - } - }; - - if delta_lines == 0 { - return; - } - - let new_top = if delta_lines < 0 { - current_top.saturating_sub(delta_lines.unsigned_abs() as usize) - } else { - current_top - .saturating_add(delta_lines as usize) - .min(max_start) - }; - - if new_top == max_start { - self.transcript_scroll = TranscriptScroll::ToBottom; - } else { - let anchor = meta.iter().skip(new_top).find_map(|entry| *entry); - if let Some((cell_index, line_in_cell)) = anchor { - self.transcript_scroll = TranscriptScroll::Scrolled { - cell_index, - line_in_cell, - }; - } else if let Some(prev_idx) = (0..=new_top).rfind(|&idx| meta[idx].is_some()) { - if let Some((cell_index, line_in_cell)) = meta[prev_idx] { - self.transcript_scroll = TranscriptScroll::Scrolled { - cell_index, - line_in_cell, - }; - } else { - self.transcript_scroll = TranscriptScroll::ToBottom; - } - } else { - self.transcript_scroll = TranscriptScroll::ToBottom; - } - } + let (_, line_meta) = Self::build_transcript_lines(&self.transcript_cells, width); + self.transcript_scroll = + self.transcript_scroll + .scrolled_by(delta_lines, &line_meta, visible_lines); tui.frame_requester().schedule_frame(); } @@ -1053,8 +959,8 @@ impl App { return; } - let (lines, meta) = Self::build_transcript_lines(&self.transcript_cells, width); - if lines.is_empty() || meta.is_empty() { + let (lines, line_meta) = Self::build_transcript_lines(&self.transcript_cells, width); + if lines.is_empty() || line_meta.is_empty() { return; } @@ -1073,22 +979,8 @@ impl App { } }; - let mut anchor = None; - if let Some((cell_index, line_in_cell)) = meta.iter().skip(top_offset).flatten().next() { - anchor = Some((*cell_index, *line_in_cell)); - } - if anchor.is_none() - && let Some((cell_index, line_in_cell)) = - meta[..top_offset].iter().rev().flatten().next() - { - anchor = Some((*cell_index, *line_in_cell)); - } - - if let Some((cell_index, line_in_cell)) = anchor { - self.transcript_scroll = TranscriptScroll::Scrolled { - cell_index, - line_in_cell, - }; + if let Some(scroll_state) = TranscriptScroll::anchor_for(&line_meta, top_offset) { + self.transcript_scroll = scroll_state; } } @@ -1096,16 +988,17 @@ impl App { /// /// Returns both the visible `Line` buffer and a parallel metadata vector /// that maps each line back to its originating `(cell_index, line_in_cell)` - /// pair, or `None` for spacer lines. This allows the scroll state to anchor - /// to a specific history cell even as new content arrives or the viewport - /// size changes, and gives exit transcript renderers enough structure to - /// style user rows differently from agent rows. + /// pair (see `TranscriptLineMeta::CellLine`), or `TranscriptLineMeta::Spacer` for + /// synthetic spacer rows inserted between cells. This allows the scroll state + /// to anchor to a specific history cell even as new content arrives or the + /// viewport size changes, and gives exit transcript renderers enough structure + /// to style user rows differently from agent rows. fn build_transcript_lines( cells: &[Arc], width: u16, - ) -> (Vec>, Vec>) { + ) -> (Vec>, Vec) { let mut lines: Vec> = Vec::new(); - let mut meta: Vec> = Vec::new(); + let mut line_meta: Vec = Vec::new(); let mut has_emitted_lines = false; for (cell_index, cell) in cells.iter().enumerate() { @@ -1117,19 +1010,22 @@ impl App { if !cell.is_stream_continuation() { if has_emitted_lines { lines.push(Line::from("")); - meta.push(None); + line_meta.push(TranscriptLineMeta::Spacer); } else { has_emitted_lines = true; } } for (line_in_cell, line) in cell_lines.into_iter().enumerate() { - meta.push(Some((cell_index, line_in_cell))); + line_meta.push(TranscriptLineMeta::CellLine { + cell_index, + line_in_cell, + }); lines.push(line); } } - (lines, meta) + (lines, line_meta) } /// Render flattened transcript lines into ANSI strings suitable for @@ -1144,7 +1040,7 @@ impl App { /// and tools see consistent escape sequences. fn render_lines_to_ansi( lines: &[Line<'static>], - meta: &[Option<(usize, usize)>], + line_meta: &[TranscriptLineMeta], is_user_cell: &[bool], width: u16, ) -> Vec { @@ -1152,10 +1048,10 @@ impl App { .iter() .enumerate() .map(|(idx, line)| { - let is_user_row = meta + let is_user_row = line_meta .get(idx) - .and_then(|entry| entry.as_ref()) - .map(|(cell_index, _)| is_user_cell.get(*cell_index).copied().unwrap_or(false)) + .and_then(TranscriptLineMeta::cell_index) + .map(|cell_index| is_user_cell.get(cell_index).copied().unwrap_or(false)) .unwrap_or(false); let mut merged_spans: Vec> = line @@ -2262,7 +2158,7 @@ mod tests { active_profile: None, file_search, transcript_cells: Vec::new(), - transcript_scroll: TranscriptScroll::ToBottom, + transcript_scroll: TranscriptScroll::default(), transcript_selection: TranscriptSelection::default(), transcript_view_top: 0, transcript_total_lines: 0, @@ -2306,7 +2202,7 @@ mod tests { active_profile: None, file_search, transcript_cells: Vec::new(), - transcript_scroll: TranscriptScroll::ToBottom, + transcript_scroll: TranscriptScroll::default(), transcript_selection: TranscriptSelection::default(), transcript_view_top: 0, transcript_total_lines: 0, @@ -2576,11 +2472,14 @@ mod tests { fn render_lines_to_ansi_pads_user_rows_to_full_width() { let line: Line<'static> = Line::from("hi"); let lines = vec![line]; - let meta = vec![Some((0usize, 0usize))]; + let line_meta = vec![TranscriptLineMeta::CellLine { + cell_index: 0, + line_in_cell: 0, + }]; let is_user_cell = vec![true]; let width: u16 = 10; - let rendered = App::render_lines_to_ansi(&lines, &meta, &is_user_cell, width); + let rendered = App::render_lines_to_ansi(&lines, &line_meta, &is_user_cell, width); assert_eq!(rendered.len(), 1); assert!(rendered[0].contains("hi")); } diff --git a/codex-rs/tui2/src/tui.rs b/codex-rs/tui2/src/tui.rs index 807c807c87..712c5cf553 100644 --- a/codex-rs/tui2/src/tui.rs +++ b/codex-rs/tui2/src/tui.rs @@ -49,6 +49,7 @@ use crate::tui::job_control::SuspendContext; mod frame_requester; #[cfg(unix)] mod job_control; +pub(crate) mod scrolling; /// A type alias for the terminal type used in this application pub type Terminal = CustomTerminal>; diff --git a/codex-rs/tui2/src/tui/scrolling.rs b/codex-rs/tui2/src/tui/scrolling.rs new file mode 100644 index 0000000000..95346793dc --- /dev/null +++ b/codex-rs/tui2/src/tui/scrolling.rs @@ -0,0 +1,366 @@ +//! Inline transcript scrolling primitives. +//! +//! The TUI renders the transcript as a list of logical *cells* (user prompts, agent responses, +//! banners, etc.). Each frame flattens those cells into a sequence of visual lines (after wrapping) +//! plus a parallel `line_meta` vector that maps each visual line back to its origin +//! (`TranscriptLineMeta`) (see `App::build_transcript_lines` and the design notes in +//! `codex-rs/tui2/docs/tui_viewport_and_history.md`). +//! +//! This module defines the scroll state for the inline transcript viewport and helpers to: +//! - Resolve that state into a concrete top-row offset for the current frame. +//! - Apply a scroll delta (mouse wheel / PgUp / PgDn) in terms of *visual lines*. +//! - Convert a concrete top-row offset back into a stable anchor. +//! +//! Why anchors instead of a raw "top row" index? +//! - When the transcript grows, a raw index drifts relative to the user's chosen content. +//! - By anchoring to a particular `(cell_index, line_in_cell)`, we can re-find the same content in +//! the newly flattened line list on the next frame. +//! +//! Spacer rows between non-continuation cells are represented as `TranscriptLineMeta::Spacer`. +//! They are not valid anchors; `anchor_for` will pick the nearest non-spacer line when needed. + +/// Per-flattened-line metadata for the transcript view. +/// +/// Each rendered line in the flattened transcript has a corresponding `TranscriptLineMeta` entry +/// describing where that visual line came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TranscriptLineMeta { + /// A visual line that belongs to a transcript cell. + CellLine { + cell_index: usize, + line_in_cell: usize, + }, + /// A synthetic spacer row inserted between non-continuation cells. + Spacer, +} + +impl TranscriptLineMeta { + pub(crate) fn cell_line(&self) -> Option<(usize, usize)> { + match *self { + Self::CellLine { + cell_index, + line_in_cell, + } => Some((cell_index, line_in_cell)), + Self::Spacer => None, + } + } + + pub(crate) fn cell_index(&self) -> Option { + match *self { + Self::CellLine { cell_index, .. } => Some(cell_index), + Self::Spacer => None, + } + } +} + +/// Scroll state for the inline transcript viewport. +/// +/// This tracks whether the transcript is pinned to the latest line or anchored +/// at a specific cell/line pair so later viewport changes can implement +/// scrollback without losing the notion of "bottom". +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) enum TranscriptScroll { + #[default] + /// Follow the most recent line in the transcript. + ToBottom, + /// Anchor the viewport to a specific transcript cell and line. + /// + /// `cell_index` indexes into the logical transcript cell list. `line_in_cell` is the 0-based + /// visual line index within that cell as produced by the current wrapping/layout. + Scrolled { + cell_index: usize, + line_in_cell: usize, + }, +} + +impl TranscriptScroll { + /// Resolve the top row for the current scroll state. + /// + /// `line_meta` is a line-parallel mapping of flattened transcript lines. + /// + /// `max_start` is the maximum valid top-row offset for the current viewport height (i.e. the + /// last scroll position that still yields a full viewport of content). + /// + /// Returns the (possibly updated) scroll state plus the resolved top-row offset. If the current + /// anchor can no longer be found in `line_meta` (for example because the transcript was + /// truncated), this falls back to `ToBottom` so the UI stays usable. + pub(crate) fn resolve_top( + self, + line_meta: &[TranscriptLineMeta], + max_start: usize, + ) -> (Self, usize) { + match self { + Self::ToBottom => (Self::ToBottom, max_start), + Self::Scrolled { + cell_index, + line_in_cell, + } => { + let anchor = anchor_index(line_meta, cell_index, line_in_cell); + match anchor { + Some(idx) => (self, idx.min(max_start)), + None => (Self::ToBottom, max_start), + } + } + } + } + + /// Apply a scroll delta and return the updated scroll state. + /// + /// `delta_lines` is in *visual lines* (after wrapping): negative deltas scroll upward into + /// scrollback, positive deltas scroll downward toward the latest content. + /// + /// See `resolve_top` for `line_meta` semantics. `visible_lines` is the viewport height in rows. + /// If all flattened lines fit in the viewport, this always returns `ToBottom`. + pub(crate) fn scrolled_by( + self, + delta_lines: i32, + line_meta: &[TranscriptLineMeta], + visible_lines: usize, + ) -> Self { + if delta_lines == 0 { + return self; + } + + let total_lines = line_meta.len(); + if total_lines <= visible_lines { + return Self::ToBottom; + } + + let max_start = total_lines.saturating_sub(visible_lines); + let current_top = match self { + Self::ToBottom => max_start, + Self::Scrolled { + cell_index, + line_in_cell, + } => anchor_index(line_meta, cell_index, line_in_cell) + .unwrap_or(max_start) + .min(max_start), + }; + + let new_top = if delta_lines < 0 { + current_top.saturating_sub(delta_lines.unsigned_abs() as usize) + } else { + current_top + .saturating_add(delta_lines as usize) + .min(max_start) + }; + + if new_top == max_start { + return Self::ToBottom; + } + + Self::anchor_for(line_meta, new_top).unwrap_or(Self::ToBottom) + } + + /// Anchor to the first available line at or near the given start offset. + /// + /// This is the inverse of "resolving a scroll state to a top-row offset": + /// given a concrete flattened line index, pick a stable `(cell_index, line_in_cell)` anchor. + /// + /// See `resolve_top` for `line_meta` semantics. This prefers the nearest line at or after `start` + /// (skipping spacer rows), falling back to the nearest line before it when needed. + pub(crate) fn anchor_for(line_meta: &[TranscriptLineMeta], start: usize) -> Option { + let anchor = + anchor_at_or_after(line_meta, start).or_else(|| anchor_at_or_before(line_meta, start)); + anchor.map(|(cell_index, line_in_cell)| Self::Scrolled { + cell_index, + line_in_cell, + }) + } +} + +/// Locate the flattened line index for a specific transcript cell and line. +/// +/// This scans `meta` for the exact `(cell_index, line_in_cell)` anchor. It returns `None` when the +/// anchor is not present in the current frame's flattened line list (for example if a cell was +/// removed or its displayed line count changed). +fn anchor_index( + line_meta: &[TranscriptLineMeta], + cell_index: usize, + line_in_cell: usize, +) -> Option { + line_meta + .iter() + .enumerate() + .find_map(|(idx, entry)| match *entry { + TranscriptLineMeta::CellLine { + cell_index: ci, + line_in_cell: li, + } if ci == cell_index && li == line_in_cell => Some(idx), + _ => None, + }) +} + +/// Find the first transcript line at or after the given flattened index. +fn anchor_at_or_after(line_meta: &[TranscriptLineMeta], start: usize) -> Option<(usize, usize)> { + if line_meta.is_empty() { + return None; + } + let start = start.min(line_meta.len().saturating_sub(1)); + line_meta + .iter() + .skip(start) + .find_map(TranscriptLineMeta::cell_line) +} + +/// Find the nearest transcript line at or before the given flattened index. +fn anchor_at_or_before(line_meta: &[TranscriptLineMeta], start: usize) -> Option<(usize, usize)> { + if line_meta.is_empty() { + return None; + } + let start = start.min(line_meta.len().saturating_sub(1)); + line_meta[..=start] + .iter() + .rev() + .find_map(TranscriptLineMeta::cell_line) +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + + fn meta(entries: &[TranscriptLineMeta]) -> Vec { + entries.to_vec() + } + + fn cell_line(cell_index: usize, line_in_cell: usize) -> TranscriptLineMeta { + TranscriptLineMeta::CellLine { + cell_index, + line_in_cell, + } + } + + #[test] + fn resolve_top_to_bottom_clamps_to_max_start() { + let meta = meta(&[ + cell_line(0, 0), + cell_line(0, 1), + TranscriptLineMeta::Spacer, + cell_line(1, 0), + ]); + + let (state, top) = TranscriptScroll::ToBottom.resolve_top(&meta, 3); + + assert_eq!(state, TranscriptScroll::ToBottom); + assert_eq!(top, 3); + } + + #[test] + fn resolve_top_scrolled_keeps_anchor_when_present() { + let meta = meta(&[ + cell_line(0, 0), + TranscriptLineMeta::Spacer, + cell_line(1, 0), + cell_line(1, 1), + ]); + let scroll = TranscriptScroll::Scrolled { + cell_index: 1, + line_in_cell: 0, + }; + + let (state, top) = scroll.resolve_top(&meta, 2); + + assert_eq!(state, scroll); + assert_eq!(top, 2); + } + + #[test] + fn resolve_top_scrolled_falls_back_when_anchor_missing() { + let meta = meta(&[cell_line(0, 0), TranscriptLineMeta::Spacer, cell_line(1, 0)]); + let scroll = TranscriptScroll::Scrolled { + cell_index: 2, + line_in_cell: 0, + }; + + let (state, top) = scroll.resolve_top(&meta, 1); + + assert_eq!(state, TranscriptScroll::ToBottom); + assert_eq!(top, 1); + } + + #[test] + fn scrolled_by_moves_upward_and_anchors() { + let meta = meta(&[ + cell_line(0, 0), + cell_line(0, 1), + cell_line(1, 0), + TranscriptLineMeta::Spacer, + cell_line(2, 0), + cell_line(2, 1), + ]); + + let state = TranscriptScroll::ToBottom.scrolled_by(-1, &meta, 3); + + assert_eq!( + state, + TranscriptScroll::Scrolled { + cell_index: 1, + line_in_cell: 0 + } + ); + } + + #[test] + fn scrolled_by_returns_to_bottom_when_scrolling_down() { + let meta = meta(&[ + cell_line(0, 0), + cell_line(0, 1), + cell_line(1, 0), + cell_line(2, 0), + ]); + let scroll = TranscriptScroll::Scrolled { + cell_index: 0, + line_in_cell: 0, + }; + + let state = scroll.scrolled_by(5, &meta, 2); + + assert_eq!(state, TranscriptScroll::ToBottom); + } + + #[test] + fn scrolled_by_to_bottom_when_all_lines_fit() { + let meta = meta(&[cell_line(0, 0), cell_line(0, 1)]); + + let state = TranscriptScroll::Scrolled { + cell_index: 0, + line_in_cell: 0, + } + .scrolled_by(-1, &meta, 5); + + assert_eq!(state, TranscriptScroll::ToBottom); + } + + #[test] + fn anchor_for_prefers_after_then_before() { + let meta = meta(&[ + TranscriptLineMeta::Spacer, + cell_line(0, 0), + TranscriptLineMeta::Spacer, + cell_line(1, 0), + ]); + + assert_eq!( + TranscriptScroll::anchor_for(&meta, 0), + Some(TranscriptScroll::Scrolled { + cell_index: 0, + line_in_cell: 0 + }) + ); + assert_eq!( + TranscriptScroll::anchor_for(&meta, 2), + Some(TranscriptScroll::Scrolled { + cell_index: 1, + line_in_cell: 0 + }) + ); + assert_eq!( + TranscriptScroll::anchor_for(&meta, 3), + Some(TranscriptScroll::Scrolled { + cell_index: 1, + line_in_cell: 0 + }) + ); + } +} diff --git a/codex-rs/windows-sandbox-rs/src/audit.rs b/codex-rs/windows-sandbox-rs/src/audit.rs index 4d63cf7a72..4385a33502 100644 --- a/codex-rs/windows-sandbox-rs/src/audit.rs +++ b/codex-rs/windows-sandbox-rs/src/audit.rs @@ -251,7 +251,7 @@ pub fn apply_capability_denies_for_world_writable( } std::fs::create_dir_all(codex_home)?; let cap_path = cap_sid_file(codex_home); - let caps = load_or_create_cap_sids(codex_home); + let caps = load_or_create_cap_sids(codex_home)?; std::fs::write(&cap_path, serde_json::to_string(&caps)?)?; let (active_sid, workspace_roots): (*mut c_void, Vec) = match sandbox_policy { SandboxPolicy::WorkspaceWrite { writable_roots, .. } => { diff --git a/codex-rs/windows-sandbox-rs/src/cap.rs b/codex-rs/windows-sandbox-rs/src/cap.rs index 2f4c96237a..5100684965 100644 --- a/codex-rs/windows-sandbox-rs/src/cap.rs +++ b/codex-rs/windows-sandbox-rs/src/cap.rs @@ -1,3 +1,5 @@ +use anyhow::Context; +use anyhow::Result; use rand::rngs::SmallRng; use rand::RngCore; use rand::SeedableRng; @@ -26,25 +28,39 @@ fn make_random_cap_sid_string() -> String { format!("S-1-5-21-{}-{}-{}-{}", a, b, c, d) } -pub fn load_or_create_cap_sids(codex_home: &Path) -> CapSids { +fn persist_caps(path: &Path, caps: &CapSids) -> Result<()> { + if let Some(dir) = path.parent() { + fs::create_dir_all(dir) + .with_context(|| format!("create cap sid dir {}", dir.display()))?; + } + let json = serde_json::to_string(caps)?; + fs::write(path, json).with_context(|| format!("write cap sid file {}", path.display()))?; + Ok(()) +} + +pub fn load_or_create_cap_sids(codex_home: &Path) -> Result { let path = cap_sid_file(codex_home); if path.exists() { - if let Ok(txt) = fs::read_to_string(&path) { - let t = txt.trim(); - if t.starts_with('{') && t.ends_with('}') { - if let Ok(obj) = serde_json::from_str::(t) { - return obj; - } - } else if !t.is_empty() { - return CapSids { - workspace: t.to_string(), - readonly: make_random_cap_sid_string(), - }; + let txt = fs::read_to_string(&path) + .with_context(|| format!("read cap sid file {}", path.display()))?; + let t = txt.trim(); + if t.starts_with('{') && t.ends_with('}') { + if let Ok(obj) = serde_json::from_str::(t) { + return Ok(obj); } + } else if !t.is_empty() { + let caps = CapSids { + workspace: t.to_string(), + readonly: make_random_cap_sid_string(), + }; + persist_caps(&path, &caps)?; + return Ok(caps); } } - CapSids { + let caps = CapSids { workspace: make_random_cap_sid_string(), readonly: make_random_cap_sid_string(), - } + }; + persist_caps(&path, &caps)?; + Ok(caps) } diff --git a/codex-rs/windows-sandbox-rs/src/elevated_impl.rs b/codex-rs/windows-sandbox-rs/src/elevated_impl.rs index b72ee75974..bf3d50147d 100644 --- a/codex-rs/windows-sandbox-rs/src/elevated_impl.rs +++ b/codex-rs/windows-sandbox-rs/src/elevated_impl.rs @@ -2,7 +2,6 @@ mod windows_impl { use crate::acl::allow_null_device; use crate::allow::compute_allow_paths; use crate::allow::AllowDenyPaths; - use crate::cap::cap_sid_file; use crate::cap::load_or_create_cap_sids; use crate::env::ensure_non_interactive_pager; use crate::env::inherit_path_env; @@ -53,13 +52,6 @@ mod windows_impl { use windows_sys::Win32::System::Threading::STARTUPINFOW; /// Ensures the parent directory of a path exists before writing to it. - fn ensure_dir(p: &Path) -> Result<()> { - if let Some(d) = p.parent() { - std::fs::create_dir_all(d)?; - } - Ok(()) - } - /// Walks upward from `start` to locate the git worktree root, following gitfile redirects. fn find_git_root(start: &Path) -> Option { let mut cur = dunce::canonicalize(start).ok()?; @@ -246,44 +238,26 @@ mod windows_impl { let sandbox_creds = require_logon_sandbox_creds(&policy, sandbox_policy_cwd, cwd, &env_map, codex_home)?; log_note("cli creds ready", logs_base_dir); - let cap_sid_path = cap_sid_file(codex_home); - // Build capability SID for ACL grants. + if matches!(&policy, SandboxPolicy::DangerFullAccess) { + anyhow::bail!("DangerFullAccess is not supported for sandboxing") + } + let caps = load_or_create_cap_sids(codex_home)?; let (psid_to_use, cap_sid_str) = match &policy { - SandboxPolicy::ReadOnly => { - let caps = load_or_create_cap_sids(codex_home); - ensure_dir(&cap_sid_path)?; - fs::write(&cap_sid_path, serde_json::to_string(&caps)?)?; - ( - unsafe { convert_string_sid_to_sid(&caps.readonly).unwrap() }, - caps.readonly.clone(), - ) - } - SandboxPolicy::WorkspaceWrite { .. } => { - let caps = load_or_create_cap_sids(codex_home); - ensure_dir(&cap_sid_path)?; - fs::write(&cap_sid_path, serde_json::to_string(&caps)?)?; - ( - unsafe { convert_string_sid_to_sid(&caps.workspace).unwrap() }, - caps.workspace.clone(), - ) - } - SandboxPolicy::DangerFullAccess => { - anyhow::bail!("DangerFullAccess is not supported for sandboxing") - } + SandboxPolicy::ReadOnly => ( + unsafe { convert_string_sid_to_sid(&caps.readonly).unwrap() }, + caps.readonly.clone(), + ), + SandboxPolicy::WorkspaceWrite { .. } => ( + unsafe { convert_string_sid_to_sid(&caps.workspace).unwrap() }, + caps.workspace.clone(), + ), + SandboxPolicy::DangerFullAccess => unreachable!("DangerFullAccess handled above"), }; - let AllowDenyPaths { allow, deny } = + let AllowDenyPaths { allow: _, deny: _ } = compute_allow_paths(&policy, sandbox_policy_cwd, ¤t_dir, &env_map); // Deny/allow ACEs are now applied during setup; avoid per-command churn. - log_note( - &format!( - "cli skipping per-command ACL grants (allow_count={} deny_count={})", - allow.len(), - deny.len() - ), - logs_base_dir, - ); unsafe { allow_null_device(psid_to_use); } diff --git a/codex-rs/windows-sandbox-rs/src/lib.rs b/codex-rs/windows-sandbox-rs/src/lib.rs index 1bcfa12074..700075d7aa 100644 --- a/codex-rs/windows-sandbox-rs/src/lib.rs +++ b/codex-rs/windows-sandbox-rs/src/lib.rs @@ -85,7 +85,6 @@ mod windows_impl { use super::acl::revoke_ace; use super::allow::compute_allow_paths; use super::allow::AllowDenyPaths; - use super::cap::cap_sid_file; use super::cap::load_or_create_cap_sids; use super::env::apply_no_network_to_env; use super::env::ensure_non_interactive_pager; @@ -104,7 +103,6 @@ mod windows_impl { use anyhow::Result; use std::collections::HashMap; use std::ffi::c_void; - use std::fs; use std::io; use std::path::Path; use std::path::PathBuf; @@ -130,13 +128,6 @@ mod windows_impl { !policy.has_full_network_access() } - fn ensure_dir(p: &Path) -> Result<()> { - if let Some(d) = p.parent() { - std::fs::create_dir_all(d)?; - } - Ok(()) - } - fn ensure_codex_home_exists(p: &Path) -> Result<()> { std::fs::create_dir_all(p)?; Ok(()) @@ -194,32 +185,28 @@ mod windows_impl { apply_no_network_to_env(&mut env_map)?; } ensure_codex_home_exists(codex_home)?; - let current_dir = cwd.to_path_buf(); - let logs_base_dir = Some(codex_home); + let sandbox_base = codex_home.join(".sandbox"); + std::fs::create_dir_all(&sandbox_base)?; + let logs_base_dir = Some(sandbox_base.as_path()); log_start(&command, logs_base_dir); - let cap_sid_path = cap_sid_file(codex_home); let is_workspace_write = matches!(&policy, SandboxPolicy::WorkspaceWrite { .. }); + if matches!(&policy, SandboxPolicy::DangerFullAccess) { + anyhow::bail!("DangerFullAccess is not supported for sandboxing") + } + let caps = load_or_create_cap_sids(codex_home)?; let (h_token, psid_to_use): (HANDLE, *mut c_void) = unsafe { match &policy { SandboxPolicy::ReadOnly => { - let caps = load_or_create_cap_sids(codex_home); - ensure_dir(&cap_sid_path)?; - fs::write(&cap_sid_path, serde_json::to_string(&caps)?)?; let psid = convert_string_sid_to_sid(&caps.readonly).unwrap(); super::token::create_readonly_token_with_cap(psid)? } SandboxPolicy::WorkspaceWrite { .. } => { - let caps = load_or_create_cap_sids(codex_home); - ensure_dir(&cap_sid_path)?; - fs::write(&cap_sid_path, serde_json::to_string(&caps)?)?; let psid = convert_string_sid_to_sid(&caps.workspace).unwrap(); super::token::create_workspace_write_token_with_cap(psid)? } - SandboxPolicy::DangerFullAccess => { - anyhow::bail!("DangerFullAccess is not supported for sandboxing") - } + SandboxPolicy::DangerFullAccess => unreachable!("DangerFullAccess handled above"), } }; diff --git a/codex-rs/windows-sandbox-rs/src/setup_main_win.rs b/codex-rs/windows-sandbox-rs/src/setup_main_win.rs index 80433ebd55..dccbd97693 100644 --- a/codex-rs/windows-sandbox-rs/src/setup_main_win.rs +++ b/codex-rs/windows-sandbox-rs/src/setup_main_win.rs @@ -20,6 +20,7 @@ use rand::RngCore; use rand::SeedableRng; use serde::Deserialize; use serde::Serialize; +use std::collections::HashSet; use std::ffi::c_void; use std::ffi::OsStr; use std::fs::File; @@ -392,8 +393,8 @@ fn run_netsh_firewall(sid: &str, log: &mut File) -> Result<()> { log_line( log, &format!( - "firewall rule configured via COM with LocalUserAuthorizedList={local_user_spec}" - ), + "firewall rule configured via COM with LocalUserAuthorizedList={local_user_spec}" + ), )?; Ok(()) })() @@ -647,7 +648,7 @@ fn run_setup(payload: &Payload, log: &mut File, sbx_dir: &Path) -> Result<()> { string_from_sid_bytes(&online_sid).map_err(anyhow::Error::msg)? ), )?; - let caps = load_or_create_cap_sids(&payload.codex_home); + let caps = load_or_create_cap_sids(&payload.codex_home)?; let cap_psid = unsafe { convert_string_sid_to_sid(&caps.workspace) .ok_or_else(|| anyhow::anyhow!("convert capability SID failed"))? @@ -758,7 +759,19 @@ fn run_setup(payload: &Payload, log: &mut File, sbx_dir: &Path) -> Result<()> { } } + let cap_sid_str = caps.workspace.clone(); + let online_sid_str = string_from_sid_bytes(&online_sid).map_err(anyhow::Error::msg)?; + let sid_strings = vec![offline_sid_str.clone(), online_sid_str, cap_sid_str]; + let write_mask = + FILE_GENERIC_READ | FILE_GENERIC_WRITE | FILE_GENERIC_EXECUTE | DELETE | FILE_DELETE_CHILD; + let mut grant_tasks: Vec = Vec::new(); + + let mut seen_write_roots: HashSet = HashSet::new(); + for root in &payload.write_roots { + if !seen_write_roots.insert(root.clone()) { + continue; + } if !root.exists() { log_line( log, @@ -766,12 +779,6 @@ fn run_setup(payload: &Payload, log: &mut File, sbx_dir: &Path) -> Result<()> { )?; continue; } - let sids = vec![offline_psid, online_psid, cap_psid]; - let write_mask = FILE_GENERIC_READ - | FILE_GENERIC_WRITE - | FILE_GENERIC_EXECUTE - | DELETE - | FILE_DELETE_CHILD; let mut need_grant = false; for (label, psid) in [ ("offline", offline_psid), @@ -817,25 +824,7 @@ fn run_setup(payload: &Payload, log: &mut File, sbx_dir: &Path) -> Result<()> { root.display() ), )?; - match unsafe { ensure_allow_write_aces(root, &sids) } { - Ok(res) => { - log_line( - log, - &format!( - "write ACE {} on {}", - if res { "added" } else { "already present" }, - root.display() - ), - )?; - } - Err(e) => { - refresh_errors.push(format!("write ACE failed on {}: {}", root.display(), e)); - log_line( - log, - &format!("write ACE grant failed on {}: {}", root.display(), e), - )?; - } - } + grant_tasks.push(root.clone()); } else { log_line( log, @@ -847,6 +836,65 @@ fn run_setup(payload: &Payload, log: &mut File, sbx_dir: &Path) -> Result<()> { } } + let (tx, rx) = mpsc::channel::<(PathBuf, Result)>(); + std::thread::scope(|scope| { + for root in grant_tasks { + let sid_strings = sid_strings.clone(); + let tx = tx.clone(); + scope.spawn(move || { + // Convert SID strings to psids locally in this thread. + let mut psids: Vec<*mut c_void> = Vec::new(); + for sid_str in &sid_strings { + if let Some(psid) = unsafe { convert_string_sid_to_sid(sid_str) } { + psids.push(psid); + } else { + let _ = tx.send((root.clone(), Err(anyhow::anyhow!("convert SID failed")))); + return; + } + } + + let res = unsafe { ensure_allow_write_aces(&root, &psids) }; + + for psid in psids { + unsafe { + LocalFree(psid as HLOCAL); + } + } + let _ = tx.send((root, res)); + }); + } + drop(tx); + for (root, res) in rx { + match res { + Ok(added) => { + if log_line( + log, + &format!( + "write ACE {} on {}", + if added { "added" } else { "already present" }, + root.display() + ), + ) + .is_err() + { + // ignore log errors inside scoped thread + } + } + Err(e) => { + refresh_errors.push(format!("write ACE failed on {}: {}", root.display(), e)); + if log_line( + log, + &format!("write ACE grant failed on {}: {}", root.display(), e), + ) + .is_err() + { + // ignore log errors inside scoped thread + } + } + } + } + }); + if refresh_only { log_line( log, diff --git a/codex-rs/windows-sandbox-rs/src/setup_orchestrator.rs b/codex-rs/windows-sandbox-rs/src/setup_orchestrator.rs index 1bd34ebb0f..8ac0157a6d 100644 --- a/codex-rs/windows-sandbox-rs/src/setup_orchestrator.rs +++ b/codex-rs/windows-sandbox-rs/src/setup_orchestrator.rs @@ -1,6 +1,7 @@ use serde::Deserialize; use serde::Serialize; use std::collections::HashMap; +use std::collections::HashSet; use std::ffi::c_void; use std::os::windows::process::CommandExt; use std::path::Path; @@ -54,13 +55,22 @@ pub fn run_setup_refresh( if matches!(policy, SandboxPolicy::DangerFullAccess) { return Ok(()); } + let (read_roots, write_roots) = build_payload_roots( + policy, + policy_cwd, + command_cwd, + env_map, + codex_home, + None, + None, + ); let payload = ElevationPayload { version: SETUP_VERSION, offline_username: OFFLINE_USERNAME.to_string(), online_username: ONLINE_USERNAME.to_string(), codex_home: codex_home.to_path_buf(), - read_roots: gather_read_roots(command_cwd, policy), - write_roots: gather_write_roots(policy, policy_cwd, command_cwd, env_map), + read_roots, + write_roots, real_user: std::env::var("USERNAME").unwrap_or_else(|_| "Administrators".to_string()), refresh_only: true, }; @@ -219,7 +229,14 @@ pub(crate) fn gather_write_roots( let AllowDenyPaths { allow, .. } = compute_allow_paths(policy, policy_cwd, command_cwd, env_map); roots.extend(allow); - canonical_existing(&roots) + let mut dedup: HashSet = HashSet::new(); + let mut out: Vec = Vec::new(); + for r in canonical_existing(&roots) { + if dedup.insert(r.clone()) { + out.push(r); + } + } + out } #[derive(Serialize)] @@ -355,19 +372,15 @@ pub fn run_elevated_setup( // Ensure the shared sandbox directory exists before we send it to the elevated helper. let sbx_dir = sandbox_dir(codex_home); std::fs::create_dir_all(&sbx_dir)?; - let mut write_roots = if let Some(roots) = write_roots_override { - roots - } else { - gather_write_roots(policy, policy_cwd, command_cwd, env_map) - }; - if !write_roots.contains(&sbx_dir) { - write_roots.push(sbx_dir.clone()); - } - let read_roots = if let Some(roots) = read_roots_override { - roots - } else { - gather_read_roots(command_cwd, policy) - }; + let (read_roots, write_roots) = build_payload_roots( + policy, + policy_cwd, + command_cwd, + env_map, + codex_home, + read_roots_override, + write_roots_override, + ); let payload = ElevationPayload { version: SETUP_VERSION, offline_username: OFFLINE_USERNAME.to_string(), @@ -381,3 +394,31 @@ pub fn run_elevated_setup( let needs_elevation = !is_elevated()?; run_setup_exe(&payload, needs_elevation) } + +fn build_payload_roots( + policy: &SandboxPolicy, + policy_cwd: &Path, + command_cwd: &Path, + env_map: &HashMap, + codex_home: &Path, + read_roots_override: Option>, + write_roots_override: Option>, +) -> (Vec, Vec) { + let sbx_dir = sandbox_dir(codex_home); + let mut write_roots = if let Some(roots) = write_roots_override { + canonical_existing(&roots) + } else { + gather_write_roots(policy, policy_cwd, command_cwd, env_map) + }; + if !write_roots.contains(&sbx_dir) { + write_roots.push(sbx_dir.clone()); + } + let mut read_roots = if let Some(roots) = read_roots_override { + canonical_existing(&roots) + } else { + gather_read_roots(command_cwd, policy) + }; + let write_root_set: HashSet = write_roots.iter().cloned().collect(); + read_roots.retain(|root| !write_root_set.contains(root)); + (read_roots, write_roots) +} diff --git a/docs/config.md b/docs/config.md index 364298c2b5..511e15f03c 100644 --- a/docs/config.md +++ b/docs/config.md @@ -316,7 +316,7 @@ disk, but attempts to write a file or access the network will be blocked. A more relaxed policy is `workspace-write`. When specified, the current working directory for the Codex task will be writable (as well as `$TMPDIR` on macOS). Note that the CLI defaults to using the directory where it was spawned as `cwd`, though this can be overridden using `--cwd/-C`. -On macOS (and soon Linux), all writable roots (including `cwd`) that contain a `.git/` folder _as an immediate child_ will configure the `.git/` folder to be read-only while the rest of the Git repository will be writable. This means that commands like `git commit` will fail, by default (as it entails writing to `.git/`), and will require Codex to ask for permission. +On macOS (and soon Linux), all writable roots (including `cwd`) that contain a `.git/` or `.codex/` folder _as an immediate child_ will configure those folders to be read-only while the rest of the root stays writable. This means that commands like `git commit` will fail, by default (as it entails writing to `.git/`), and will require Codex to ask for permission. ```toml # same as `--sandbox workspace-write`