diff --git a/.github/workflows/Dockerfile.bazel b/.github/workflows/Dockerfile.bazel index 0bfba60b83..51c199dcc3 100644 --- a/.github/workflows/Dockerfile.bazel +++ b/.github/workflows/Dockerfile.bazel @@ -4,7 +4,7 @@ FROM ubuntu:24.04 # initial debugging, but we should publish to a more proper location. # # docker buildx create --use -# docker buildx build --platform linux/amd64 -f .github/workflows/Dockerfile.bazel -t mbolin491/codex-bazel:latest --push . +# docker buildx build --platform linux/amd64,linux/arm64 -f .github/workflows/Dockerfile.bazel -t mbolin491/codex-bazel:latest --push . RUN apt-get update && \ apt-get install -y --no-install-recommends \ diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml index 7b30830868..4f97d2de02 100644 --- a/.github/workflows/bazel.yml +++ b/.github/workflows/bazel.yml @@ -108,24 +108,3 @@ jobs: --build_metadata=ROLE=CI \ --build_metadata=VISIBILITY=PUBLIC \ "--remote_header=x-buildbuddy-api-key=$BUILDBUDDY_API_KEY" - - cloud-build: - name: just bazel-remote-test - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v6 - - name: Set up Bazel - uses: bazelbuild/setup-bazelisk@v3 - - name: bazel test //... --config=remote - env: - BUILDBUDDY_API_KEY: ${{ secrets.BUILDBUDDY_API_KEY }} - shell: bash - run: | - set -euo pipefail - bazel test //... \ - --build_metadata=REPO_URL=https://github.com/openai/codex.git \ - --build_metadata=COMMIT_SHA=$(git rev-parse HEAD) \ - --build_metadata=ROLE=CI \ - --build_metadata=VISIBILITY=PUBLIC \ - "--remote_header=x-buildbuddy-api-key=$BUILDBUDDY_API_KEY" \ - --config=remote --platforms=//:rbe --keep_going diff --git a/.github/workflows/ci.bazelrc b/.github/workflows/ci.bazelrc index 11c74388ee..5322d2a8d1 100644 --- a/.github/workflows/ci.bazelrc +++ b/.github/workflows/ci.bazelrc @@ -2,14 +2,19 @@ common --remote_download_minimal common --nobuild_runfile_links common --keep_going -# Prefer to run the build actions entirely remotely so we can dial up the concurrency. -# Currently remote builds only work on Mac hosts, until we untangle the libc constraints mess on linux. +# We prefer to run the build actions entirely remotely so we can dial up the concurrency. +# We have platform-specific tests, so we want to execute the tests on all platforms using the strongest sandboxing available on each platform. + +# On linux, we can do a full remote build/test, by targeting the right (x86/arm) runners, so we have coverage of both. +# Linux crossbuilds don't work until we untangle the libc constraint mess. +common:linux --config=remote +common:linux --strategy=remote +common:linux --platforms=//:rbe + +# On mac, we can run all the build actions remotely but test actions locally. common:macos --config=remote common:macos --strategy=remote - -# We have platform-specific tests, so execute the tests locally using the strongest sandboxing available on each platform. common:macos --strategy=TestRunner=darwin-sandbox,local -# Note: linux-sandbox is stronger, but not available in GHA. -common:linux --strategy=TestRunner=processwrapper-sandbox,local + common:windows --strategy=TestRunner=local diff --git a/BUILD.bazel b/BUILD.bazel index 372a3aee7c..883432655c 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -11,21 +11,9 @@ platform( ], ) -platform( +alias( name = "rbe", - constraint_values = [ - "@platforms//cpu:x86_64", - "@platforms//os:linux", - "@bazel_tools//tools/cpp:clang", - "@toolchains_llvm_bootstrapped//constraints/libc:gnu.2.28", - ], - exec_properties = { - # Ubuntu-based image that includes git, python3, dotslash, and other - # tools that various integration tests need. - # Verify at https://hub.docker.com/layers/mbolin491/codex-bazel/latest/images/sha256:8c9ff94187ea7c08a31e9a81f5fe8046ea3972a6768983c955c4079fa30567fb - "container-image": "docker://docker.io/mbolin491/codex-bazel@sha256:8c9ff94187ea7c08a31e9a81f5fe8046ea3972a6768983c955c4079fa30567fb", - "OSFamily": "Linux", - }, + actual = "@rbe_platform", ) exports_files(["AGENTS.md"]) diff --git a/MODULE.bazel b/MODULE.bazel index d14b61d544..2547325c9d 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -120,3 +120,9 @@ crate.annotation( deps = [":windows_import_lib"], ) use_repo(crate, "crates") + +rbe_platform_repository = use_repo_rule("//:rbe.bzl", "rbe_platform_repository") + +rbe_platform_repository( + name = "rbe_platform", +) diff --git a/announcement_tip.toml b/announcement_tip.toml new file mode 100644 index 0000000000..0070003219 --- /dev/null +++ b/announcement_tip.toml @@ -0,0 +1,17 @@ +# Example announcement tips for Codex TUI. +# Each [[announcements]] entry is evaluated in order; the last matching one is shown. +# Dates are UTC, formatted as YYYY-MM-DD. The from_date is inclusive and the to_date is exclusive. +# version_regex matches against the CLI version (env!("CARGO_PKG_VERSION")); omit to apply to all versions. +# target_app specify which app should display the announcement (cli, vsce, ...). + +[[announcements]] +content = "Welcome to Codex! Check out the new onboarding flow." +from_date = "2024-10-01" +to_date = "2024-10-15" +target_app = "cli" + +# Test announcement only for local build version until 2026-01-10 excluded (past) +[[announcements]] +content = "This is a test announcement" +version_regex = "^0\\.0\\.0$" +to_date = "2026-01-10" diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 22b6ae343b..ce7e892731 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -9,6 +9,7 @@ use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; use std::sync::Arc; use std::sync::Weak; +use tokio::sync::watch; /// Control-plane handle for multi-agent operations. /// `AgentControl` is held by each session (via `SessionServices`). It provides capability to @@ -27,7 +28,6 @@ impl AgentControl { Self { manager } } - #[allow(dead_code)] // Used by upcoming multi-agent tooling. /// Spawn a new agent thread and submit the initial prompt. /// /// If `headless` is true, a background drain task is spawned to prevent unbounded event growth @@ -50,7 +50,6 @@ impl AgentControl { Ok(new_thread.thread_id) } - #[allow(dead_code)] // Used by upcoming multi-agent tooling. /// Send a `user` prompt to an existing agent thread. pub(crate) async fn send_prompt( &self, @@ -69,7 +68,13 @@ impl AgentControl { .await } - #[allow(dead_code)] // Used by upcoming multi-agent tooling. + /// Submit a shutdown request to an existing agent thread. + pub(crate) async fn shutdown_agent(&self, agent_id: ThreadId) -> CodexResult { + let state = self.upgrade()?; + state.send_op(agent_id, Op::Shutdown {}).await + } + + #[allow(dead_code)] // Will be used for collab tools. /// Fetch the last known status for `agent_id`, returning `NotFound` when unavailable. pub(crate) async fn get_status(&self, agent_id: ThreadId) -> AgentStatus { let Ok(state) = self.upgrade() else { @@ -82,6 +87,16 @@ impl AgentControl { thread.agent_status().await } + /// Subscribe to status updates for `agent_id`, yielding the latest value and changes. + pub(crate) async fn subscribe_status( + &self, + agent_id: ThreadId, + ) -> CodexResult> { + let state = self.upgrade()?; + let thread = state.get_thread(agent_id).await?; + Ok(thread.subscribe_status()) + } + fn upgrade(&self) -> CodexResult> { self.manager .upgrade() @@ -114,13 +129,63 @@ fn spawn_headless_drain(thread: Arc) { #[cfg(test)] mod tests { use super::*; + use crate::CodexAuth; + use crate::ThreadManager; use crate::agent::agent_status_from_event; + use crate::config::Config; + use crate::config::ConfigBuilder; + use assert_matches::assert_matches; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnStartedEvent; use pretty_assertions::assert_eq; + use tempfile::TempDir; + + async fn test_config() -> (TempDir, Config) { + let home = TempDir::new().expect("create temp dir"); + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .build() + .await + .expect("load default test config"); + (home, config) + } + + struct AgentControlHarness { + _home: TempDir, + config: Config, + manager: ThreadManager, + control: AgentControl, + } + + impl AgentControlHarness { + async fn new() -> Self { + let (home, config) = test_config().await; + let manager = ThreadManager::with_models_provider_and_home( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.clone(), + ); + let control = manager.agent_control(); + Self { + _home: home, + config, + manager, + control, + } + } + + async fn start_thread(&self) -> (ThreadId, Arc) { + let new_thread = self + .manager + .start_thread(self.config.clone()) + .await + .expect("start thread"); + (new_thread.thread_id, new_thread.thread) + } + } #[tokio::test] async fn send_prompt_errors_when_manager_dropped() { @@ -185,4 +250,135 @@ mod tests { let status = agent_status_from_event(&EventMsg::ShutdownComplete); assert_eq!(status, Some(AgentStatus::Shutdown)); } + + #[tokio::test] + async fn spawn_agent_errors_when_manager_dropped() { + let control = AgentControl::default(); + let (_home, config) = test_config().await; + let err = control + .spawn_agent(config, "hello".to_string(), false) + .await + .expect_err("spawn_agent should fail without a manager"); + assert_eq!( + err.to_string(), + "unsupported operation: thread manager dropped" + ); + } + + #[tokio::test] + async fn send_prompt_errors_when_thread_missing() { + let harness = AgentControlHarness::new().await; + let thread_id = ThreadId::new(); + let err = harness + .control + .send_prompt(thread_id, "hello".to_string()) + .await + .expect_err("send_prompt should fail for missing thread"); + assert_matches!(err, CodexErr::ThreadNotFound(id) if id == thread_id); + } + + #[tokio::test] + async fn get_status_returns_not_found_for_missing_thread() { + let harness = AgentControlHarness::new().await; + let status = harness.control.get_status(ThreadId::new()).await; + assert_eq!(status, AgentStatus::NotFound); + } + + #[tokio::test] + async fn get_status_returns_pending_init_for_new_thread() { + let harness = AgentControlHarness::new().await; + let (thread_id, _) = harness.start_thread().await; + let status = harness.control.get_status(thread_id).await; + assert_eq!(status, AgentStatus::PendingInit); + } + + #[tokio::test] + async fn subscribe_status_errors_for_missing_thread() { + let harness = AgentControlHarness::new().await; + let thread_id = ThreadId::new(); + let err = harness + .control + .subscribe_status(thread_id) + .await + .expect_err("subscribe_status should fail for missing thread"); + assert_matches!(err, CodexErr::ThreadNotFound(id) if id == thread_id); + } + + #[tokio::test] + async fn subscribe_status_updates_on_shutdown() { + let harness = AgentControlHarness::new().await; + let (thread_id, thread) = harness.start_thread().await; + let mut status_rx = harness + .control + .subscribe_status(thread_id) + .await + .expect("subscribe_status should succeed"); + assert_eq!(status_rx.borrow().clone(), AgentStatus::PendingInit); + + let _ = thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); + + let _ = status_rx.changed().await; + assert_eq!(status_rx.borrow().clone(), AgentStatus::Shutdown); + } + + #[tokio::test] + async fn send_prompt_submits_user_message() { + let harness = AgentControlHarness::new().await; + let (thread_id, _thread) = harness.start_thread().await; + + let submission_id = harness + .control + .send_prompt(thread_id, "hello from tests".to_string()) + .await + .expect("send_prompt should succeed"); + assert!(!submission_id.is_empty()); + let expected = ( + thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "hello from tests".to_string(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + assert_eq!(captured, Some(expected)); + } + + #[tokio::test] + async fn spawn_agent_creates_thread_and_sends_prompt() { + let harness = AgentControlHarness::new().await; + let thread_id = harness + .control + .spawn_agent(harness.config.clone(), "spawned".to_string(), false) + .await + .expect("spawn_agent should succeed"); + let _thread = harness + .manager + .get_thread(thread_id) + .await + .expect("thread should be registered"); + let expected = ( + thread_id, + Op::UserInput { + items: vec![UserInput::Text { + text: "spawned".to_string(), + }], + final_output_json_schema: None, + }, + ); + let captured = harness + .manager + .captured_ops() + .into_iter() + .find(|entry| *entry == expected); + assert_eq!(captured, Some(expected)); + } } diff --git a/codex-rs/core/src/agent/status.rs b/codex-rs/core/src/agent/status.rs index 4f99e99aae..74981513fd 100644 --- a/codex-rs/core/src/agent/status.rs +++ b/codex-rs/core/src/agent/status.rs @@ -13,3 +13,7 @@ pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option { _ => None, } } + +pub(crate) fn is_final(status: &AgentStatus) -> bool { + !matches!(status, AgentStatus::PendingInit | AgentStatus::Running) +} diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 00ed4fba0d..7f26acf176 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -164,6 +164,7 @@ use codex_protocol::protocol::InitialHistory; use codex_protocol::user_input::UserInput; use codex_utils_readiness::Readiness; use codex_utils_readiness::ReadinessFlag; +use tokio::sync::watch; /// The high-level interface to the Codex system. /// It operates as a queue pair where you send submissions and receive events. @@ -172,7 +173,7 @@ pub struct Codex { pub(crate) tx_sub: Sender, pub(crate) rx_event: Receiver, // Last known status of the agent. - pub(crate) agent_status: Arc>, + pub(crate) agent_status: watch::Receiver, } /// Wrapper returned by [`Codex::spawn`] containing the spawned [`Codex`], @@ -275,7 +276,7 @@ impl Codex { // Generate a unique ID for the lifetime of this Codex session. let session_source_clone = session_configuration.session_source.clone(); - let agent_status = Arc::new(RwLock::new(AgentStatus::PendingInit)); + let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit); let session = Session::new( session_configuration, @@ -284,7 +285,7 @@ impl Codex { models_manager.clone(), exec_policy, tx_event.clone(), - Arc::clone(&agent_status), + agent_status_tx.clone(), conversation_history, session_source_clone, skills_manager, @@ -303,7 +304,7 @@ impl Codex { next_id: AtomicU64::new(0), tx_sub, rx_event, - agent_status, + agent_status: agent_status_rx, }; #[allow(deprecated)] @@ -345,8 +346,7 @@ impl Codex { } pub(crate) async fn agent_status(&self) -> AgentStatus { - let status = self.agent_status.read().await; - status.clone() + self.agent_status.borrow().clone() } } @@ -356,7 +356,7 @@ impl Codex { pub(crate) struct Session { conversation_id: ThreadId, tx_event: Sender, - agent_status: Arc>, + agent_status: watch::Sender, state: Mutex, /// The set of enabled features should be invariant for the lifetime of the /// session. @@ -557,7 +557,7 @@ impl Session { models_manager: Arc, exec_policy: ExecPolicyManager, tx_event: Sender, - agent_status: Arc>, + agent_status: watch::Sender, initial_history: InitialHistory, session_source: SessionSource, skills_manager: Arc, @@ -703,7 +703,7 @@ impl Session { let sess = Arc::new(Session { conversation_id, tx_event: tx_event.clone(), - agent_status: Arc::clone(&agent_status), + agent_status, state: Mutex::new(state), features: config.features.clone(), active_turn: Mutex::new(None), @@ -1026,8 +1026,7 @@ impl Session { pub(crate) async fn send_event_raw(&self, event: Event) { // Record the last known agent status. if let Some(status) = agent_status_from_event(&event.msg) { - let mut guard = self.agent_status.write().await; - *guard = status; + self.agent_status.send_replace(status); } // Persist the event into rollout (recorder filters as needed) let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; @@ -1045,8 +1044,7 @@ impl Session { pub(crate) async fn send_event_raw_flushed(&self, event: Event) { // Record the last known agent status. if let Some(status) = agent_status_from_event(&event.msg) { - let mut guard = self.agent_status.write().await; - *guard = status; + self.agent_status.send_replace(status); } self.persist_rollout_items(&[RolloutItem::EventMsg(event.msg.clone())]) .await; @@ -3494,7 +3492,7 @@ mod tests { )); let agent_control = AgentControl::default(); let exec_policy = ExecPolicyManager::default(); - let agent_status = Arc::new(RwLock::new(AgentStatus::PendingInit)); + let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = ModelsManager::get_model_offline(config.model.as_deref()); let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), @@ -3557,7 +3555,7 @@ mod tests { let session = Session { conversation_id, tx_event, - agent_status: Arc::clone(&agent_status), + agent_status: agent_status_tx, state: Mutex::new(state), features: config.features.clone(), active_turn: Mutex::new(None), @@ -3588,7 +3586,7 @@ mod tests { )); let agent_control = AgentControl::default(); let exec_policy = ExecPolicyManager::default(); - let agent_status = Arc::new(RwLock::new(AgentStatus::PendingInit)); + let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit); let model = ModelsManager::get_model_offline(config.model.as_deref()); let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), @@ -3651,7 +3649,7 @@ mod tests { let session = Arc::new(Session { conversation_id, tx_event, - agent_status: Arc::clone(&agent_status), + agent_status: agent_status_tx, state: Mutex::new(state), features: config.features.clone(), active_turn: Mutex::new(None), diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 869c18d555..49409f8e84 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -87,7 +87,7 @@ pub(crate) async fn run_codex_thread_interactive( next_id: AtomicU64::new(0), tx_sub: tx_ops, rx_event: rx_sub, - agent_status: Arc::clone(&codex.agent_status), + agent_status: codex.agent_status.clone(), }) } @@ -129,7 +129,7 @@ pub(crate) async fn run_codex_thread_one_shot( // Bridge events so we can observe completion and shut down automatically. let (tx_bridge, rx_bridge) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let ops_tx = io.tx_sub.clone(); - let agent_status = Arc::clone(&io.agent_status); + let agent_status = io.agent_status.clone(); let io_for_bridge = io; tokio::spawn(async move { while let Ok(event) = io_for_bridge.next_event().await { @@ -363,20 +363,23 @@ mod tests { use super::*; use async_channel::bounded; use codex_protocol::models::ResponseItem; + use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::RawResponseItemEvent; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use pretty_assertions::assert_eq; + use tokio::sync::watch; #[tokio::test] async fn forward_events_cancelled_while_send_blocked_shuts_down_delegate() { let (tx_events, rx_events) = bounded(1); let (tx_sub, rx_sub) = bounded(SUBMISSION_CHANNEL_CAPACITY); + let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit); let codex = Arc::new(Codex { next_id: AtomicU64::new(0), tx_sub, rx_event: rx_events, - agent_status: Default::default(), + agent_status, }); let (session, ctx, _rx_evt) = crate::codex::make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index e8a3799303..6d132a9256 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -5,6 +5,7 @@ use crate::protocol::Event; use crate::protocol::Op; use crate::protocol::Submission; use std::path::PathBuf; +use tokio::sync::watch; pub struct CodexThread { codex: Codex, @@ -38,6 +39,10 @@ impl CodexThread { self.codex.agent_status().await } + pub(crate) fn subscribe_status(&self) -> watch::Receiver { + self.codex.agent_status.clone() + } + pub fn rollout_path(&self) -> PathBuf { self.rollout_path.clone() } diff --git a/codex-rs/core/src/command_safety/is_safe_command.rs b/codex-rs/core/src/command_safety/is_safe_command.rs index 01a52026e2..c2d53cdbaf 100644 --- a/codex-rs/core/src/command_safety/is_safe_command.rs +++ b/codex-rs/core/src/command_safety/is_safe_command.rs @@ -1,23 +1,28 @@ use crate::bash::parse_shell_lc_plain_commands; -use crate::command_safety::windows_safe_commands::is_safe_command_windows; +use crate::command_safety::windows_safe_commands::is_powershell_executable; +use crate::command_safety::windows_safe_commands::is_safe_powershell_command; +/// `command` must be the _exact_ proposed list of arguments that will be sent +/// to `execvp(3)`. That is: +/// - If the command is to be run via a shell, then the first argument MUST be +/// the shell executable (e.g. "bash", "zsh", "pwsh", "powershell"). +/// - The first argument will be resolved against the `PATH` environment +/// variable, so it can be either a bare command name (e.g. "ls") or a full +/// path. +/// - If the first argument is NOT a shell executable, then the command MUST be +/// discoverable in the system `PATH` and MUST NOT rely on shell features +/// (e.g. shell built-ins, operators, or syntax). pub fn is_known_safe_command(command: &[String]) -> bool { - let command: Vec = command - .iter() - .map(|s| { - if s == "zsh" { - "bash".to_string() - } else { - s.clone() - } - }) - .collect(); - - if is_safe_command_windows(&command) { - return true; + if command.is_empty() { + return false; } - if is_safe_to_call_with_exec(&command) { + let exe = &command[0]; + if is_powershell_executable(exe) { + return is_safe_powershell_command(command); + } + + if is_safe_to_call_with_exec(command) { return true; } @@ -27,7 +32,7 @@ pub fn is_known_safe_command(command: &[String]) -> bool { // introduce side effects ( "&&", "||", ";", and "|" ). If every // individual command in the script is itself a known‑safe command, then // the composite expression is considered safe. - if let Some(all_commands) = parse_shell_lc_plain_commands(&command) + if let Some(all_commands) = parse_shell_lc_plain_commands(command) && !all_commands.is_empty() && all_commands .iter() @@ -422,4 +427,29 @@ mod tests { "> redirection should be rejected" ); } + + /// This test only works on Windows because it requires access to PowerShell + /// to parse the argument to `pwsh -Command`. + #[cfg(windows)] + #[test] + fn ensure_safe_unix_command_that_is_unsafe_powershell_command_is_rejected() { + assert!( + is_known_safe_command(&vec_str(&["echo", "hi", "@(calc)"])), + "Running echo with execvp is safe because there is no shell interpretation." + ); + assert!( + !is_known_safe_command(&vec_str(&["bash", "-lc", "echo hi @(calc)"])), + "This command should not get marked as safe because the Bash script has a parse error." + ); + assert!( + is_known_safe_command(&vec_str(&["pwsh", "-Command", "echo hi"])), + "Our logic should recognize that `echo hi` is safe to run under PowerShell." + ); + assert!( + !is_known_safe_command(&vec_str(&["pwsh", "-Command", "echo hi @(calc)"])), + r#"Even though `echo hi @(calc)` would not do anything malicious +when run under Bash (because it would be rejected with a parse error), it would run +calc.exec when run under PowerShell and therefore should be rejected."# + ); + } } diff --git a/codex-rs/core/src/command_safety/windows_safe_commands.rs b/codex-rs/core/src/command_safety/windows_safe_commands.rs index ac479a4d2c..82422f8201 100644 --- a/codex-rs/core/src/command_safety/windows_safe_commands.rs +++ b/codex-rs/core/src/command_safety/windows_safe_commands.rs @@ -9,7 +9,11 @@ const POWERSHELL_PARSER_SCRIPT: &str = include_str!("powershell_parser.ps1"); /// On Windows, we conservatively allow only clearly read-only PowerShell invocations /// that match a small safelist. Anything else (including direct CMD commands) is unsafe. -pub fn is_safe_command_windows(command: &[String]) -> bool { +/// +/// `command` must be the _exact_ proposed list of arguments that will be sent +/// to `execvp(3)`, so `command[0]` must satisfy [`is_powershell_executable`] +/// for this function to return true. +pub fn is_safe_powershell_command(command: &[String]) -> bool { if let Some(commands) = try_parse_powershell_command_sequence(command) { commands .iter() @@ -108,7 +112,7 @@ fn parse_powershell_script(executable: &str, script: &str) -> Option bool { +pub fn is_powershell_executable(exe: &str) -> bool { let executable_name = Path::new(exe) .file_name() .and_then(|osstr| osstr.to_str()) diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index a4e8f9c34c..6124f67874 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -56,6 +56,10 @@ pub(crate) struct ThreadManagerState { models_manager: Arc, skills_manager: Arc, session_source: SessionSource, + #[cfg(any(test, feature = "test-support"))] + #[allow(dead_code)] + // Captures submitted ops for testing purpose. + ops_log: Arc>>, } impl ThreadManager { @@ -74,6 +78,8 @@ impl ThreadManager { skills_manager: Arc::new(SkillsManager::new(codex_home)), auth_manager, session_source, + #[cfg(any(test, feature = "test-support"))] + ops_log: Arc::new(std::sync::Mutex::new(Vec::new())), }), #[cfg(any(test, feature = "test-support"))] _test_codex_home_guard: None, @@ -111,6 +117,8 @@ impl ThreadManager { skills_manager: Arc::new(SkillsManager::new(codex_home)), auth_manager, session_source: SessionSource::Exec, + #[cfg(any(test, feature = "test-support"))] + ops_log: Arc::new(std::sync::Mutex::new(Vec::new())), }), _test_codex_home_guard: None, } @@ -202,9 +210,19 @@ impl ThreadManager { .await } - fn agent_control(&self) -> AgentControl { + pub(crate) fn agent_control(&self) -> AgentControl { AgentControl::new(Arc::downgrade(&self.state)) } + + #[cfg(any(test, feature = "test-support"))] + #[allow(dead_code)] + pub(crate) fn captured_ops(&self) -> Vec<(ThreadId, Op)> { + self.state + .ops_log + .lock() + .map(|log| log.clone()) + .unwrap_or_default() + } } impl ThreadManagerState { @@ -217,7 +235,14 @@ impl ThreadManagerState { } pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult { - self.get_thread(thread_id).await?.submit(op).await + let thread = self.get_thread(thread_id).await?; + #[cfg(any(test, feature = "test-support"))] + { + if let Ok(mut log) = self.ops_log.lock() { + log.push((thread_id, op.clone())); + } + } + thread.submit(op).await } #[allow(dead_code)] // Used by upcoming multi-agent tooling. diff --git a/codex-rs/core/src/tools/handlers/collab.rs b/codex-rs/core/src/tools/handlers/collab.rs index e59e15cbc0..0f74ef658a 100644 --- a/codex-rs/core/src/tools/handlers/collab.rs +++ b/codex-rs/core/src/tools/handlers/collab.rs @@ -1,3 +1,4 @@ +use crate::agent::AgentStatus; use crate::codex::TurnContext; use crate::config::Config; use crate::error::CodexErr; @@ -11,29 +12,13 @@ use crate::tools::registry::ToolKind; use async_trait::async_trait; use codex_protocol::ThreadId; use serde::Deserialize; +use serde::Serialize; pub struct CollabHandler; pub(crate) const DEFAULT_WAIT_TIMEOUT_MS: i64 = 30_000; pub(crate) const MAX_WAIT_TIMEOUT_MS: i64 = 300_000; -#[derive(Debug, Deserialize)] -struct SpawnAgentArgs { - message: String, -} - -#[derive(Debug, Deserialize)] -struct SendInputArgs { - id: String, - message: String, -} - -#[derive(Debug, Deserialize)] -struct WaitArgs { - id: String, - timeout_ms: Option, -} - #[derive(Debug, Deserialize)] struct CloseAgentArgs { id: String, @@ -68,10 +53,10 @@ impl ToolHandler for CollabHandler { }; match tool_name.as_str() { - "spawn_agent" => handle_spawn_agent(session, turn, arguments).await, - "send_input" => handle_send_input(session, arguments).await, - "wait" => handle_wait(arguments).await, - "close_agent" => handle_close_agent(arguments).await, + "spawn_agent" => spawn::handle(session, turn, arguments).await, + "send_input" => send_input::handle(session, arguments).await, + "wait" => wait::handle(session, arguments).await, + "close_agent" => close_agent::handle(session, arguments).await, other => Err(FunctionCallError::RespondToModel(format!( "unsupported collab tool {other}" ))), @@ -79,84 +64,233 @@ impl ToolHandler for CollabHandler { } } -async fn handle_spawn_agent( - session: std::sync::Arc, - turn: std::sync::Arc, - arguments: String, -) -> Result { - let args: SpawnAgentArgs = parse_arguments(&arguments)?; - if args.message.trim().is_empty() { - return Err(FunctionCallError::RespondToModel( - "Empty message can't be send to an agent".to_string(), - )); - } - let config = build_agent_spawn_config(turn.as_ref())?; - let result = session - .services - .agent_control - .spawn_agent(config, args.message, true) - .await - .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; +mod spawn { + use super::*; + use crate::codex::Session; + use std::sync::Arc; - Ok(ToolOutput::Function { - content: format!("agent_id: {result}"), - success: Some(true), - content_items: None, - }) + #[derive(Debug, Deserialize)] + struct SpawnAgentArgs { + message: String, + } + + pub async fn handle( + session: Arc, + turn: Arc, + arguments: String, + ) -> Result { + let args: SpawnAgentArgs = parse_arguments(&arguments)?; + if args.message.trim().is_empty() { + return Err(FunctionCallError::RespondToModel( + "Empty message can't be send to an agent".to_string(), + )); + } + let config = build_agent_spawn_config(turn.as_ref())?; + let result = session + .services + .agent_control + .spawn_agent(config, args.message, true) + .await + .map_err(|err| FunctionCallError::Fatal(err.to_string()))?; + + Ok(ToolOutput::Function { + content: format!("agent_id: {result}"), + success: Some(true), + content_items: None, + }) + } } -async fn handle_send_input( - session: std::sync::Arc, - arguments: String, -) -> Result { - let args: SendInputArgs = parse_arguments(&arguments)?; - let agent_id = agent_id(&args.id)?; - if args.message.trim().is_empty() { - return Err(FunctionCallError::RespondToModel( - "Empty message can't be send to an agent".to_string(), - )); +mod send_input { + use super::*; + use crate::codex::Session; + use std::sync::Arc; + + #[derive(Debug, Deserialize)] + struct SendInputArgs { + id: String, + message: String, } - let content = session - .services - .agent_control - .send_prompt(agent_id, args.message) - .await - .map_err(|err| match err { - CodexErr::ThreadNotFound(id) => { - FunctionCallError::RespondToModel(format!("agent with id {id} not found")) + + pub async fn handle( + session: Arc, + arguments: String, + ) -> Result { + let args: SendInputArgs = parse_arguments(&arguments)?; + let agent_id = agent_id(&args.id)?; + if args.message.trim().is_empty() { + return Err(FunctionCallError::RespondToModel( + "Empty message can't be send to an agent".to_string(), + )); + } + let content = session + .services + .agent_control + .send_prompt(agent_id, args.message) + .await + .map_err(|err| match err { + CodexErr::ThreadNotFound(id) => { + FunctionCallError::RespondToModel(format!("agent with id {id} not found")) + } + err => FunctionCallError::Fatal(err.to_string()), + })?; + + Ok(ToolOutput::Function { + content, + success: Some(true), + content_items: None, + }) + } +} + +mod wait { + use super::*; + use crate::agent::status::is_final; + use crate::codex::Session; + use std::sync::Arc; + use std::time::Duration; + use tokio::time::Instant; + use tokio::time::timeout_at; + + #[derive(Debug, Deserialize)] + struct WaitArgs { + id: String, + timeout_ms: Option, + } + + #[derive(Debug, Serialize)] + struct WaitResult { + status: AgentStatus, + timed_out: bool, + } + + pub async fn handle( + session: Arc, + arguments: String, + ) -> Result { + let args: WaitArgs = parse_arguments(&arguments)?; + let agent_id = agent_id(&args.id)?; + + // Validate timeout. + let timeout_ms = args.timeout_ms.unwrap_or(DEFAULT_WAIT_TIMEOUT_MS); + let timeout_ms = match timeout_ms { + ms if ms <= 0 => { + return Err(FunctionCallError::RespondToModel( + "timeout_ms must be greater than zero".to_owned(), + )); } - err => FunctionCallError::Fatal(err.to_string()), + ms => ms.min(MAX_WAIT_TIMEOUT_MS), + }; + + let mut status_rx = session + .services + .agent_control + .subscribe_status(agent_id) + .await + .map_err(|err| match err { + CodexErr::ThreadNotFound(id) => { + FunctionCallError::RespondToModel(format!("agent with id {id} not found")) + } + err => FunctionCallError::Fatal(err.to_string()), + })?; + + // Get last known status. + let mut status = status_rx.borrow_and_update().clone(); + let deadline = Instant::now() + Duration::from_millis(timeout_ms as u64); + + let timed_out = loop { + if is_final(&status) { + break false; + } + + match timeout_at(deadline, status_rx.changed()).await { + Ok(Ok(())) => status = status_rx.borrow().clone(), + Ok(Err(_)) => { + let last_status = session.services.agent_control.get_status(agent_id).await; + if last_status != AgentStatus::NotFound { + // On-purpose we keep the last known status if the agent gets dropped. This + // event is not supposed to happen. + status = last_status; + } + break false; + } + Err(_) => break true, + } + }; + + if matches!(status, AgentStatus::NotFound) { + return Err(FunctionCallError::RespondToModel(format!( + "agent with id {agent_id} not found" + ))); + } + + let result = WaitResult { status, timed_out }; + + let content = serde_json::to_string(&result).map_err(|err| { + FunctionCallError::Fatal(format!("failed to serialize wait result: {err}")) })?; - Ok(ToolOutput::Function { - content, - success: Some(true), - content_items: None, - }) -} - -async fn handle_wait(arguments: String) -> Result { - let args: WaitArgs = parse_arguments(&arguments)?; - let _agent_id = agent_id(&args.id)?; - - let timeout_ms = args.timeout_ms.unwrap_or(DEFAULT_WAIT_TIMEOUT_MS); - if timeout_ms <= 0 { - return Err(FunctionCallError::RespondToModel( - "timeout_ms must be greater than zero".to_string(), - )); + Ok(ToolOutput::Function { + content, + success: Some(!result.timed_out), + content_items: None, + }) } - let _timeout_ms = timeout_ms.min(MAX_WAIT_TIMEOUT_MS); - // TODO(jif): implement agent wait once lifecycle tracking is wired up. - Err(FunctionCallError::Fatal("wait not implemented".to_string())) } -async fn handle_close_agent(arguments: String) -> Result { - let args: CloseAgentArgs = parse_arguments(&arguments)?; - let _agent_id = agent_id(&args.id)?; - // TODO(jif): implement agent shutdown and return the final status. - Err(FunctionCallError::Fatal( - "close_agent not implemented".to_string(), - )) +pub mod close_agent { + use super::*; + use crate::codex::Session; + use std::sync::Arc; + + #[derive(Debug, Deserialize, Serialize)] + pub(super) struct CloseAgentResult { + pub(super) status: AgentStatus, + } + + pub async fn handle( + session: Arc, + arguments: String, + ) -> Result { + let args: CloseAgentArgs = parse_arguments(&arguments)?; + let agent_id = agent_id(&args.id)?; + let mut status_rx = session + .services + .agent_control + .subscribe_status(agent_id) + .await + .map_err(|err| match err { + CodexErr::ThreadNotFound(id) => { + FunctionCallError::RespondToModel(format!("agent with id {id} not found")) + } + err => FunctionCallError::Fatal(err.to_string()), + })?; + let status = status_rx.borrow_and_update().clone(); + + if !matches!(status, AgentStatus::Shutdown) { + let _ = session + .services + .agent_control + .shutdown_agent(agent_id) + .await + .map_err(|err| match err { + CodexErr::ThreadNotFound(id) => { + FunctionCallError::RespondToModel(format!("agent with id {id} not found")) + } + err => FunctionCallError::Fatal(err.to_string()), + })?; + } + + let content = serde_json::to_string(&CloseAgentResult { status }).map_err(|err| { + FunctionCallError::Fatal(format!("failed to serialize close_agent result: {err}")) + })?; + + Ok(ToolOutput::Function { + content, + success: Some(true), + content_items: None, + }) + } } fn agent_id(id: &str) -> Result { @@ -192,3 +326,384 @@ fn build_agent_spawn_config(turn: &TurnContext) -> Result, + turn: Arc, + tool_name: &str, + payload: ToolPayload, + ) -> ToolInvocation { + ToolInvocation { + session, + turn, + tracker: Arc::new(Mutex::new(TurnDiffTracker::default())), + call_id: "call-1".to_string(), + tool_name: tool_name.to_string(), + payload, + } + } + + fn function_payload(args: serde_json::Value) -> ToolPayload { + ToolPayload::Function { + arguments: args.to_string(), + } + } + + fn thread_manager() -> ThreadManager { + ThreadManager::with_models_provider( + CodexAuth::from_api_key("dummy"), + built_in_model_providers()["openai"].clone(), + ) + } + + #[tokio::test] + async fn handler_rejects_non_function_payloads() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "spawn_agent", + ToolPayload::Custom { + input: "hello".to_string(), + }, + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("payload should be rejected"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel( + "collab handler received unsupported payload".to_string() + ) + ); + } + + #[tokio::test] + async fn handler_rejects_unknown_tool() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "unknown_tool", + function_payload(json!({})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("tool should be rejected"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel("unsupported collab tool unknown_tool".to_string()) + ); + } + + #[tokio::test] + async fn spawn_agent_rejects_empty_message() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "spawn_agent", + function_payload(json!({"message": " "})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("empty message should be rejected"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel( + "Empty message can't be send to an agent".to_string() + ) + ); + } + + #[tokio::test] + async fn spawn_agent_errors_when_manager_dropped() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "spawn_agent", + function_payload(json!({"message": "hello"})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("spawn should fail without a manager"); + }; + assert_eq!( + err, + FunctionCallError::Fatal("unsupported operation: thread manager dropped".to_string()) + ); + } + + #[tokio::test] + async fn send_input_rejects_empty_message() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "send_input", + function_payload(json!({"id": ThreadId::new().to_string(), "message": ""})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("empty message should be rejected"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel( + "Empty message can't be send to an agent".to_string() + ) + ); + } + + #[tokio::test] + async fn send_input_rejects_invalid_id() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "send_input", + function_payload(json!({"id": "not-a-uuid", "message": "hi"})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("invalid id should be rejected"); + }; + let FunctionCallError::RespondToModel(msg) = err else { + panic!("expected respond-to-model error"); + }; + assert!(msg.starts_with("invalid agent id not-a-uuid:")); + } + + #[tokio::test] + async fn send_input_reports_missing_agent() { + let (mut session, turn) = make_session_and_context().await; + let manager = thread_manager(); + session.services.agent_control = manager.agent_control(); + let agent_id = ThreadId::new(); + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "send_input", + function_payload(json!({"id": agent_id.to_string(), "message": "hi"})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("missing agent should be reported"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel(format!("agent with id {agent_id} not found")) + ); + } + + #[tokio::test] + async fn wait_rejects_non_positive_timeout() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "wait", + function_payload(json!({"id": ThreadId::new().to_string(), "timeout_ms": 0})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("non-positive timeout should be rejected"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel("timeout_ms must be greater than zero".to_string()) + ); + } + + #[tokio::test] + async fn wait_rejects_invalid_id() { + let (session, turn) = make_session_and_context().await; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "wait", + function_payload(json!({"id": "invalid"})), + ); + let Err(err) = CollabHandler.handle(invocation).await else { + panic!("invalid id should be rejected"); + }; + let FunctionCallError::RespondToModel(msg) = err else { + panic!("expected respond-to-model error"); + }; + assert!(msg.starts_with("invalid agent id invalid:")); + } + + #[tokio::test] + async fn wait_times_out_when_status_is_not_final() { + let (mut session, turn) = make_session_and_context().await; + let manager = thread_manager(); + session.services.agent_control = manager.agent_control(); + let config = turn.client.config().as_ref().clone(); + let thread = manager.start_thread(config).await.expect("start thread"); + let agent_id = thread.thread_id; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "wait", + function_payload(json!({"id": agent_id.to_string(), "timeout_ms": 10})), + ); + let output = CollabHandler + .handle(invocation) + .await + .expect("wait should succeed"); + let ToolOutput::Function { + content, success, .. + } = output + else { + panic!("expected function output"); + }; + assert_eq!(content, r#"{"status":"pending_init","timed_out":true}"#); + assert_eq!(success, Some(false)); + + let _ = thread + .thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); + } + + #[tokio::test] + async fn wait_returns_final_status_without_timeout() { + let (mut session, turn) = make_session_and_context().await; + let manager = thread_manager(); + session.services.agent_control = manager.agent_control(); + let config = turn.client.config().as_ref().clone(); + let thread = manager.start_thread(config).await.expect("start thread"); + let agent_id = thread.thread_id; + let mut status_rx = manager + .agent_control() + .subscribe_status(agent_id) + .await + .expect("subscribe should succeed"); + + let _ = thread + .thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); + let _ = timeout(Duration::from_secs(1), status_rx.changed()) + .await + .expect("shutdown status should arrive"); + + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "wait", + function_payload(json!({"id": agent_id.to_string(), "timeout_ms": 1000})), + ); + let output = CollabHandler + .handle(invocation) + .await + .expect("wait should succeed"); + let ToolOutput::Function { + content, success, .. + } = output + else { + panic!("expected function output"); + }; + assert_eq!(content, r#"{"status":"shutdown","timed_out":false}"#); + assert_eq!(success, Some(true)); + } + + #[tokio::test] + async fn close_agent_submits_shutdown_and_returns_status() { + let (mut session, turn) = make_session_and_context().await; + let manager = thread_manager(); + session.services.agent_control = manager.agent_control(); + let config = turn.client.config().as_ref().clone(); + let thread = manager.start_thread(config).await.expect("start thread"); + let agent_id = thread.thread_id; + let status_before = manager.agent_control().get_status(agent_id).await; + + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "close_agent", + function_payload(json!({"id": agent_id.to_string()})), + ); + let output = CollabHandler + .handle(invocation) + .await + .expect("close_agent should succeed"); + let ToolOutput::Function { + content, success, .. + } = output + else { + panic!("expected function output"); + }; + let result: close_agent::CloseAgentResult = + serde_json::from_str(&content).expect("close_agent result should be json"); + assert_eq!(result.status, status_before); + assert_eq!(success, Some(true)); + + let ops = manager.captured_ops(); + let submitted_shutdown = ops + .iter() + .any(|(id, op)| *id == agent_id && matches!(op, Op::Shutdown)); + assert_eq!(submitted_shutdown, true); + } + + #[tokio::test] + async fn build_agent_spawn_config_uses_turn_context_values() { + let (_session, mut turn) = make_session_and_context().await; + turn.developer_instructions = Some("dev".to_string()); + turn.base_instructions = Some("base".to_string()); + turn.compact_prompt = Some("compact".to_string()); + turn.user_instructions = Some("user".to_string()); + turn.shell_environment_policy = ShellEnvironmentPolicy { + use_profile: true, + ..ShellEnvironmentPolicy::default() + }; + let temp_dir = tempfile::tempdir().expect("temp dir"); + turn.cwd = temp_dir.path().to_path_buf(); + turn.codex_linux_sandbox_exe = Some(PathBuf::from("/bin/echo")); + turn.approval_policy = AskForApproval::Never; + turn.sandbox_policy = SandboxPolicy::DangerFullAccess; + + let config = build_agent_spawn_config(&turn).expect("spawn config"); + let mut expected = (*turn.client.config()).clone(); + expected.model = Some(turn.client.get_model()); + expected.model_provider = turn.client.get_provider(); + expected.model_reasoning_effort = turn.client.get_reasoning_effort(); + expected.model_reasoning_summary = turn.client.get_reasoning_summary(); + expected.developer_instructions = turn.developer_instructions.clone(); + expected.base_instructions = turn.base_instructions.clone(); + expected.compact_prompt = turn.compact_prompt.clone(); + expected.user_instructions = turn.user_instructions.clone(); + expected.shell_environment_policy = turn.shell_environment_policy.clone(); + expected.codex_linux_sandbox_exe = turn.codex_linux_sandbox_exe.clone(); + expected.cwd = turn.cwd.clone(); + expected + .approval_policy + .set(turn.approval_policy) + .expect("approval policy set"); + expected + .sandbox_policy + .set(turn.sandbox_policy) + .expect("sandbox policy set"); + assert_eq!(config, expected); + } +} diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index b2f51caea8..4f6314104d 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -699,7 +699,7 @@ pub enum AgentStatus { Completed(Option), /// Agent encountered an error. Errored(String), - /// Agent has been shutdowned. + /// Agent has been shutdown. Shutdown, /// Agent is not found. NotFound, diff --git a/rbe.bzl b/rbe.bzl new file mode 100644 index 0000000000..c1074c2be5 --- /dev/null +++ b/rbe.bzl @@ -0,0 +1,42 @@ +def _rbe_platform_repo_impl(rctx): + arch = rctx.os.arch + if arch in ["x86_64", "amd64"]: + cpu = "x86_64" + exec_arch = "amd64" + image_sha = "8c9ff94187ea7c08a31e9a81f5fe8046ea3972a6768983c955c4079fa30567fb" + elif arch in ["aarch64", "arm64"]: + cpu = "aarch64" + exec_arch = "arm64" + image_sha = "ad9506086215fccfc66ed8d2be87847324be56790ae6a1964c241c28b77ef141" + else: + fail("Unsupported host arch for rbe platform: {}".format(arch)) + + rctx.file("BUILD.bazel", """\ +platform( + name = "rbe_platform", + constraint_values = [ + "@platforms//cpu:{cpu}", + "@platforms//os:linux", + "@bazel_tools//tools/cpp:clang", + "@toolchains_llvm_bootstrapped//constraints/libc:gnu.2.28", + ], + exec_properties = {{ + # Ubuntu-based image that includes git, python3, dotslash, and other + # tools that various integration tests need. + # Verify at https://hub.docker.com/layers/mbolin491/codex-bazel/latest/images/sha256:{image_sha} + "container-image": "docker://docker.io/mbolin491/codex-bazel@sha256:{image_sha}", + "Arch": "{arch}", + "OSFamily": "Linux", + }}, + visibility = ["//visibility:public"], +) +""".format( + cpu = cpu, + arch = exec_arch, + image_sha = image_sha +)) + +rbe_platform_repository = repository_rule( + implementation = _rbe_platform_repo_impl, + doc = "Sets up a platform for remote builds with an Arch exec_property matching the host.", +)