From bd2a7329233b5bfc1e211a0d18419ca5c697c1e1 Mon Sep 17 00:00:00 2001 From: "Adam Perry @ OpenAI" Date: Thu, 28 May 2026 08:46:30 -0700 Subject: [PATCH 1/6] Add app-server startup benchmark crate (#24651) ## Summary - Add a new `app-server-start-bench` crate to measure app-server startup performance - Wire the benchmark into the workspace and Bazel build so it can be run consistently - Update lockfiles and repo automation to account for the new package --- .github/workflows/rust-ci-full.yml | 7 + codex-rs/Cargo.lock | 11 ++ codex-rs/Cargo.toml | 1 + .../benchmarks/app-server-start/BUILD.bazel | 20 ++ .../benchmarks/app-server-start/Cargo.toml | 16 ++ .../benchmarks/app-server-start/src/main.rs | 174 ++++++++++++++++++ justfile | 22 ++- 7 files changed, 247 insertions(+), 4 deletions(-) create mode 100644 codex-rs/benchmarks/app-server-start/BUILD.bazel create mode 100644 codex-rs/benchmarks/app-server-start/Cargo.toml create mode 100644 codex-rs/benchmarks/app-server-start/src/main.rs diff --git a/.github/workflows/rust-ci-full.yml b/.github/workflows/rust-ci-full.yml index 2111ed7b0a..702ad12dbb 100644 --- a/.github/workflows/rust-ci-full.yml +++ b/.github/workflows/rust-ci-full.yml @@ -466,6 +466,13 @@ jobs: - name: cargo clippy run: cargo clippy --target ${{ matrix.target }} --tests --profile ${{ matrix.profile }} --timings -- -D warnings + - uses: taiki-e/install-action@44c6d64aa62cd779e873306675c7a58e86d6d532 # v2.62.49 + with: + tool: just + + - name: End-to-end benchmark smoke test + run: just bench-e2e-smoke + - name: Upload Cargo timings (clippy) if: always() uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0 diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index aac0d6f90c..21f8af0d6e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2058,6 +2058,17 @@ dependencies = [ "uuid", ] +[[package]] +name = "codex-app-server-start-bench" +version = "0.0.0" +dependencies = [ + "anyhow", + "codex-app-server-protocol", + "divan", + "serde_json", + "tempfile", +] + [[package]] name = "codex-app-server-test-client" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index b1f9e4c31a..5ef6a45fc3 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -14,6 +14,7 @@ members = [ "app-server-client", "app-server-protocol", "app-server-test-client", + "benchmarks/app-server-start", "debug-client", "apply-patch", "arg0", diff --git a/codex-rs/benchmarks/app-server-start/BUILD.bazel b/codex-rs/benchmarks/app-server-start/BUILD.bazel new file mode 100644 index 0000000000..4b8f09413e --- /dev/null +++ b/codex-rs/benchmarks/app-server-start/BUILD.bazel @@ -0,0 +1,20 @@ +load("//:defs.bzl", "codex_rust_crate", "workspace_root_test") + +codex_rust_crate( + name = "app-server-start-bench", + crate_name = "codex_app_server_start_bench", +) + +workspace_root_test( + name = "startup-bench", + # Include the Bazel-built CLI in this target's runfiles and expose its + # resolved executable path through the environment consumed by the runner. + runfile_env = { + "//codex-rs/cli:codex": "CODEX_BIN", + }, + size = "large", + tags = ["manual"], + test_bin = ":codex-app-server-start-bench", + timeout = "long", + workspace_root_marker = "//codex-rs/utils/cargo-bin:repo_root.marker", +) diff --git a/codex-rs/benchmarks/app-server-start/Cargo.toml b/codex-rs/benchmarks/app-server-start/Cargo.toml new file mode 100644 index 0000000000..edf680eeea --- /dev/null +++ b/codex-rs/benchmarks/app-server-start/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "codex-app-server-start-bench" +version.workspace = true +edition.workspace = true +license.workspace = true +publish = false + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } +codex-app-server-protocol = { workspace = true } +serde_json = { workspace = true } +divan = { workspace = true } +tempfile = { workspace = true } diff --git a/codex-rs/benchmarks/app-server-start/src/main.rs b/codex-rs/benchmarks/app-server-start/src/main.rs new file mode 100644 index 0000000000..5363b49ffc --- /dev/null +++ b/codex-rs/benchmarks/app-server-start/src/main.rs @@ -0,0 +1,174 @@ +use std::io::BufRead; +use std::io::BufReader; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::process::Child; +use std::process::ChildStdin; +use std::process::ChildStdout; +use std::process::Command; +use std::process::Stdio; + +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::InitializeResponse; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use divan::Bencher; + +fn main() { + divan::main(); +} + +// Process startup is slow enough that 30 samples keep runs practical, and +// sample size 1 lets each measured server be reaped before the next starts. +// The benchmark warms one CODEX_HOME before timing to measure normal restarts. +// This e2e runner receives its separately built Codex binary from just or Bazel. +#[divan::bench(sample_count = 30, sample_size = 1, skip_ext_time)] +#[allow(clippy::expect_used)] +fn initialize_response(bencher: Bencher) { + let codex_bin = std::env::var_os("CODEX_BIN") + .map(PathBuf::from) + .expect("CODEX_BIN must point to the codex binary; run via `just bench-e2e` or Bazel"); + let codex_home = tempfile::tempdir().expect("benchmark CODEX_HOME should be created"); + drop( + start_until_initialize_response(&codex_bin, codex_home.path()) + .expect("benchmark CODEX_HOME should be initialized"), + ); + + bencher.bench_local(|| { + start_until_initialize_response(&codex_bin, codex_home.path()) + .expect("codex app-server should return an initialize response") + }); +} + +/// A running app-server that has returned a valid `initialize` response. +/// +/// Divan drops benchmark outputs after it stops each timer interval. Returning +/// this value from the measured closure keeps the `initialized` notification +/// and forced process reaping outside startup latency. +struct InitializedAppServer { + child: Child, + stdin: Option, + stdout: Option>, + acknowledge_on_drop: bool, +} + +/// Spawn a stdio app-server and return once it responds successfully to `initialize`. +fn start_until_initialize_response( + codex_bin: &Path, + codex_home: &Path, +) -> Result { + let request_id = RequestId::Integer(0); + let child = Command::new(codex_bin) + .arg("app-server") + .arg("--listen") + .arg("stdio://") + .env("CODEX_HOME", codex_home) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .with_context(|| format!("failed to spawn `{}` app-server", codex_bin.display()))?; + let mut server = InitializedAppServer { + child, + stdin: None, + stdout: None, + acknowledge_on_drop: false, + }; + server.stdin = Some( + server + .child + .stdin + .take() + .context("app-server stdin unavailable")?, + ); + server.stdout = Some(BufReader::new( + server + .child + .stdout + .take() + .context("app-server stdout unavailable")?, + )); + + let request = ClientRequest::Initialize { + request_id: request_id.clone(), + params: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-start-bench".to_string(), + title: Some("Codex App Server Start Benchmark".to_string()), + version: env!("CARGO_PKG_VERSION").to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: false, + request_attestation: false, + opt_out_notification_methods: None, + }), + }, + }; + let stdin = server + .stdin + .as_mut() + .context("app-server stdin unavailable")?; + writeln!(stdin, "{}", serde_json::to_string(&request)?)?; + stdin + .flush() + .context("failed to flush initialize request")?; + + let mut line = String::new(); + loop { + line.clear(); + if server + .stdout + .as_mut() + .context("app-server stdout unavailable")? + .read_line(&mut line)? + == 0 + { + bail!("app-server closed stdout before returning initialize response"); + } + + match serde_json::from_str::(line.trim())? { + JSONRPCMessage::Response(JSONRPCResponse { id, result }) if id == request_id => { + let _: InitializeResponse = serde_json::from_value(result) + .context("initialize response missing expected payload")?; + server.acknowledge_on_drop = true; + return Ok(server); + } + JSONRPCMessage::Error(error) if error.id == request_id => { + bail!("initialize failed: {error:?}"); + } + JSONRPCMessage::Request(_) + | JSONRPCMessage::Response(_) + | JSONRPCMessage::Notification(_) + | JSONRPCMessage::Error(_) => {} + } + } +} + +impl Drop for InitializedAppServer { + fn drop(&mut self) { + if self.acknowledge_on_drop + && let Some(stdin) = self.stdin.as_mut() + { + let initialized = JSONRPCMessage::Notification(JSONRPCNotification { + method: "initialized".to_string(), + params: None, + }); + if let Ok(payload) = serde_json::to_string(&initialized) { + let _ = writeln!(stdin, "{payload}"); + let _ = stdin.flush(); + } + } + let _ = self.stdin.take(); + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} diff --git a/justfile b/justfile index d5e9fc3a36..d061b90fda 100644 --- a/justfile +++ b/justfile @@ -2,6 +2,9 @@ set working-directory := "codex-rs" set positional-arguments rust_min_stack := "8388608" # 8 MiB +e2e_benchmark_packages := "codex-app-server-start-bench" +e2e_codex_bin := if os() == "windows" { "./target/release/codex.exe" } else { "./target/release/codex" } +e2e_smoke_codex_bin := if os() == "windows" { "./target/debug/codex.exe" } else { "./target/debug/codex" } # Display help help: @@ -30,6 +33,11 @@ app-server-test-client *args: cargo build -p codex-cli cargo run -p codex-app-server-test-client -- --codex-bin ./target/debug/codex "$@" +# Run end-to-end performance benchmarks that require a built Codex binary. +bench-e2e *args: + cargo build --release -p codex-cli --bin codex + for package in {{ e2e_benchmark_packages }}; do CODEX_BIN="{{ e2e_codex_bin }}" cargo run --release -p "$package" -- --bench "$@"; done + # Format Rust and Python SDK code. fmt: cargo fmt -- --config imports_granularity=Item 2>/dev/null @@ -55,13 +63,19 @@ test *args: RUST_MIN_STACK={{ rust_min_stack }} cargo nextest run --no-fail-fast "$@" just bench-smoke -# Run explicit workspace benchmark targets. -bench *args: +# Run unit-test-style benchmark targets managed entirely by Cargo. +bench-unit *args: cargo bench --workspace --bench '*' "$@" -# Run benchmark targets once to ensure they start successfully. +# Smoke Cargo-managed benchmarks and compile e2e runners without measured binaries. bench-smoke: - just bench -- --test + just bench-unit -- --test + for package in {{ e2e_benchmark_packages }}; do cargo build -p "$package" --bin "$package"; done + +# Run end-to-end performance benchmark targets once. +bench-e2e-smoke: + cargo build -p codex-cli --bin codex + for package in {{ e2e_benchmark_packages }}; do CODEX_BIN="{{ e2e_smoke_codex_bin }}" cargo run -p "$package" -- --test; done # Build and run Codex from source using Bazel. # Note we have to use the combination of `[no-cd]` and `--run_under="cd $PWD &&"` From e426d48f6de446ba4e262ec1933e1e1d7dcddd03 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 28 May 2026 17:47:17 +0200 Subject: [PATCH 2/6] Gate goal tools by thread eligibility (#24925) ## Why Goal tools create and update goal state for a persistent thread. The extension was only checking whether goals were enabled before advertising those tools, which meant they could be surfaced in contexts that should not receive thread goal controls: ephemeral threads without persistent thread state and review subagents. Those sessions can still run the goal extension lifecycle, but the thread tools should only be visible when the current thread can safely use them. ## What changed - Adds a `GoalRuntimeConfig` that separates goal enablement from whether goal tools are available for the current thread. - Computes tool eligibility on thread start from `persistent_thread_state_available` and `SessionSource`, hiding tools for review subagents. - Uses `GoalRuntimeHandle::tools_visible()` when contributing thread tools so enabled runtime state does not automatically imply tool exposure. - Adds backend coverage for hiding goal tools on ephemeral threads and review subagents. ## Testing - Added `goal_tools_hidden_for_ephemeral_threads`. - Added `goal_tools_hidden_for_review_subagents`. --- codex-rs/ext/goal/src/extension.rs | 15 ++++- codex-rs/ext/goal/src/runtime.rs | 15 ++++- .../ext/goal/tests/goal_extension_backend.rs | 55 ++++++++++++++++++- 3 files changed, 79 insertions(+), 6 deletions(-) diff --git a/codex-rs/ext/goal/src/extension.rs b/codex-rs/ext/goal/src/extension.rs index 68068acbb4..5421d17290 100644 --- a/codex-rs/ext/goal/src/extension.rs +++ b/codex-rs/ext/goal/src/extension.rs @@ -22,6 +22,8 @@ use codex_extension_api::TurnStartInput; use codex_extension_api::TurnStopInput; use codex_otel::MetricsClient; use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::TokenUsageInfo; @@ -29,6 +31,7 @@ use crate::accounting::BudgetLimitedGoalDisposition; use crate::accounting::GoalAccountingState; use crate::events::GoalEventEmitter; use crate::metrics::GoalMetrics; +use crate::runtime::GoalRuntimeConfig; use crate::runtime::GoalRuntimeHandle; use crate::spec::UPDATE_GOAL_TOOL_NAME; use crate::steering::budget_limit_steering_item; @@ -85,6 +88,11 @@ where { async fn on_thread_start(&self, input: ThreadStartInput<'_, C>) { let enabled = (self.goals_enabled)(input.config); + let tools_available_for_thread = input.persistent_thread_state_available + && !matches!( + input.session_source, + SessionSource::SubAgent(SubAgentSource::Review) + ); input .thread_store .insert(GoalExtensionConfig::from_enabled(enabled)); @@ -102,7 +110,10 @@ where self.metrics.clone(), self.thread_manager.clone(), accounting_state, - enabled, + GoalRuntimeConfig { + enabled, + tools_available_for_thread, + }, ) }); runtime.set_enabled(enabled); @@ -330,7 +341,7 @@ where let Some(runtime) = goal_runtime_handle(thread_store) else { return Vec::new(); }; - if !runtime.is_enabled() { + if !runtime.tools_visible() { return Vec::new(); } diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index 16a4ff6a2f..d78eda3de5 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/codex-rs/ext/goal/src/runtime.rs @@ -20,6 +20,11 @@ pub struct GoalRuntimeHandle { inner: Arc, } +pub(crate) struct GoalRuntimeConfig { + pub(crate) enabled: bool, + pub(crate) tools_available_for_thread: bool, +} + struct GoalRuntimeInner { thread_id: ThreadId, state_dbs: Arc, @@ -28,6 +33,7 @@ struct GoalRuntimeInner { thread_manager: Weak, accounting_state: Arc, enabled: AtomicBool, + tools_available_for_thread: bool, } pub(crate) struct AccountedGoalProgress { @@ -66,7 +72,7 @@ impl GoalRuntimeHandle { metrics: GoalMetrics, thread_manager: Weak, accounting_state: Arc, - enabled: bool, + config: GoalRuntimeConfig, ) -> Self { Self { inner: Arc::new(GoalRuntimeInner { @@ -76,7 +82,8 @@ impl GoalRuntimeHandle { metrics, thread_manager, accounting_state, - enabled: AtomicBool::new(enabled), + enabled: AtomicBool::new(config.enabled), + tools_available_for_thread: config.tools_available_for_thread, }), } } @@ -89,6 +96,10 @@ impl GoalRuntimeHandle { self.inner.enabled.load(Ordering::Relaxed) } + pub(crate) fn tools_visible(&self) -> bool { + self.is_enabled() && self.inner.tools_available_for_thread + } + pub(crate) fn thread_id(&self) -> ThreadId { self.inner.thread_id } diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index 7dbdd2e246..ed0f59357b 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -28,6 +28,7 @@ use codex_protocol::config_types::Settings; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; @@ -83,6 +84,38 @@ async fn installed_goal_tools_create_goal_and_fill_empty_preview() -> anyhow::Re Ok(()) } +#[tokio::test] +async fn goal_tools_hidden_for_ephemeral_threads() -> anyhow::Result<()> { + let runtime = test_runtime().await?; + let thread_id = test_thread_id()?; + let tools = installed_tools_with_start( + runtime, + thread_id, + SessionSource::Cli, + /*persistent_thread_state_available*/ false, + ) + .await; + + assert_eq!(Vec::::new(), tool_names(&tools)); + Ok(()) +} + +#[tokio::test] +async fn goal_tools_hidden_for_review_subagents() -> anyhow::Result<()> { + let runtime = test_runtime().await?; + let thread_id = test_thread_id()?; + let tools = installed_tools_with_start( + runtime, + thread_id, + SessionSource::SubAgent(SubAgentSource::Review), + /*persistent_thread_state_available*/ true, + ) + .await; + + assert_eq!(Vec::::new(), tool_names(&tools)); + Ok(()) +} + #[tokio::test] async fn installed_goal_tools_reject_duplicate_goal_creation() -> anyhow::Result<()> { let runtime = test_runtime().await?; @@ -877,6 +910,21 @@ async fn thread_resume_rehydrates_active_goal_idle_accounting() -> anyhow::Resul async fn installed_tools( runtime: Arc, thread_id: ThreadId, +) -> Vec>> { + installed_tools_with_start( + runtime, + thread_id, + SessionSource::Cli, + /*persistent_thread_state_available*/ true, + ) + .await +} + +async fn installed_tools_with_start( + runtime: Arc, + thread_id: ThreadId, + session_source: SessionSource, + persistent_thread_state_available: bool, ) -> Vec>> { let mut builder = ExtensionRegistryBuilder::<()>::new(); install_with_backend( @@ -889,13 +937,12 @@ async fn installed_tools( let registry = builder.build(); let session_store = ExtensionData::new("session-1"); let thread_store = ExtensionData::new(thread_id.to_string()); - let session_source = SessionSource::Cli; for contributor in registry.thread_lifecycle_contributors() { contributor .on_thread_start(ThreadStartInput { config: &(), session_source: &session_source, - persistent_thread_state_available: true, + persistent_thread_state_available, session_store: &session_store, thread_store: &thread_store, }) @@ -909,6 +956,10 @@ async fn installed_tools( .collect() } +fn tool_names(tools: &[Arc>]) -> Vec { + tools.iter().map(|tool| tool.tool_name().name).collect() +} + struct GoalExtensionHarness { registry: codex_extension_api::ExtensionRegistry<()>, session_store: ExtensionData, From a061befb469e9cb620767c1230aa1429008178b3 Mon Sep 17 00:00:00 2001 From: "Adam Perry @ OpenAI" Date: Thu, 28 May 2026 08:49:01 -0700 Subject: [PATCH 3/6] Remove libubsan CI workaround (#24782) It seems that this was added to allow rustc to load proc macros that had been compiled with UBSan enabled, which zig does for debug and `ReleaseSafe` builds. When zig drives the link of the final binary it knows to include the ubsan runtime, but our zig-built artifacts are being linked into a binary whose linking rustc drives. This removes the libubsan workaround we have and replaces it with `-fno-sanitize=undefined` passed to zig. The new argument is passed at the end of zig's args so should take precedence over any earlier arguments from the script's caller. --- .github/scripts/install-musl-build-tools.sh | 13 +++- .github/workflows/rust-ci-full.yml | 66 +-------------------- .github/workflows/rust-release.yml | 58 +----------------- 3 files changed, 13 insertions(+), 124 deletions(-) diff --git a/.github/scripts/install-musl-build-tools.sh b/.github/scripts/install-musl-build-tools.sh index e4c6683d0e..49035f5391 100644 --- a/.github/scripts/install-musl-build-tools.sh +++ b/.github/scripts/install-musl-build-tools.sh @@ -150,7 +150,9 @@ for arg in "\$@"; do args+=("\${arg}") done -exec "${zig_bin}" cc -target "${zig_target}" "\${args[@]}" +# Zig enables UBSan for debug C builds by default. Rust links these objects +# without Zig's sanitizer runtime, so keep native dependencies uninstrumented. +exec "${zig_bin}" cc -target "${zig_target}" "\${args[@]}" -fno-sanitize=undefined EOF cat >"${cxx}" <> "$GITHUB_ENV" pkg_config_path_var="PKG_CONFIG_PATH_${TARGET}" pkg_config_path_var="${pkg_config_path_var//-/_}" echo "${pkg_config_path_var}=${libcap_pkgconfig_dir}" >> "$GITHUB_ENV" +pkg_config_libdir_var="PKG_CONFIG_LIBDIR_${TARGET}" +pkg_config_libdir_var="${pkg_config_libdir_var//-/_}" +# Do not let musl cross-builds resolve native libraries from the host glibc +# pkg-config directories. libcap is the only target package provided here. +echo "${pkg_config_libdir_var}=${libcap_pkgconfig_dir}" >> "$GITHUB_ENV" if [[ -n "${sysroot}" && "${sysroot}" != "/" ]]; then echo "PKG_CONFIG_SYSROOT_DIR=${sysroot}" >> "$GITHUB_ENV" diff --git a/.github/workflows/rust-ci-full.yml b/.github/workflows/rust-ci-full.yml index 702ad12dbb..3031cb6853 100644 --- a/.github/workflows/rust-ci-full.yml +++ b/.github/workflows/rust-ci-full.yml @@ -260,11 +260,7 @@ jobs: set -euo pipefail if command -v apt-get >/dev/null 2>&1; then sudo apt-get update -y - packages=(pkg-config libcap-dev) - if [[ "${{ matrix.target }}" == 'x86_64-unknown-linux-musl' || "${{ matrix.target }}" == 'aarch64-unknown-linux-musl' ]]; then - packages+=(libubsan1) - fi - sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends "${packages[@]}" + sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev fi - uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0 with: @@ -348,14 +344,6 @@ jobs: sccache-${{ matrix.runner }}-${{ matrix.target }}-${{ matrix.profile }}-${{ steps.lockhash.outputs.hash }}- sccache-${{ matrix.runner }}-${{ matrix.target }}-${{ matrix.profile }}- - - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Disable sccache wrapper (musl) - shell: bash - run: | - set -euo pipefail - echo "RUSTC_WRAPPER=" >> "$GITHUB_ENV" - echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV" - - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} name: Prepare APT cache directories (musl) shell: bash @@ -389,58 +377,6 @@ jobs: shell: bash run: bash "${GITHUB_WORKSPACE}/.github/scripts/install-musl-build-tools.sh" - - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Configure rustc UBSan wrapper (musl host) - shell: bash - run: | - set -euo pipefail - ubsan="" - if command -v ldconfig >/dev/null 2>&1; then - ubsan="$(ldconfig -p | grep -m1 'libubsan\.so\.1' | sed -E 's/.*=> (.*)$/\1/')" - fi - wrapper_root="${RUNNER_TEMP:-/tmp}" - wrapper="${wrapper_root}/rustc-ubsan-wrapper" - cat > "${wrapper}" <> "$GITHUB_ENV" - echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV" - - - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Clear sanitizer flags (musl) - shell: bash - run: | - set -euo pipefail - # Clear global Rust flags so host/proc-macro builds don't pull in UBSan. - echo "RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_ENCODED_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "RUSTDOCFLAGS=" >> "$GITHUB_ENV" - # Override any runner-level Cargo config rustflags as well. - echo "CARGO_BUILD_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV" - - sanitize_flags() { - local input="$1" - input="${input//-fsanitize=undefined/}" - input="${input//-fno-sanitize-recover=undefined/}" - input="${input//-fno-sanitize-trap=undefined/}" - echo "$input" - } - - cflags="$(sanitize_flags "${CFLAGS-}")" - cxxflags="$(sanitize_flags "${CXXFLAGS-}")" - echo "CFLAGS=${cflags}" >> "$GITHUB_ENV" - echo "CXXFLAGS=${cxxflags}" >> "$GITHUB_ENV" - - if: ${{ !contains(matrix.target, 'windows') }} name: Configure rusty_v8 artifact overrides and verify checksums uses: ./.github/actions/setup-rusty-v8 diff --git a/.github/workflows/rust-release.yml b/.github/workflows/rust-release.yml index eaf886cd01..07a2e87ef2 100644 --- a/.github/workflows/rust-release.yml +++ b/.github/workflows/rust-release.yml @@ -245,15 +245,6 @@ jobs: set -euo pipefail sudo apt-get update -y sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev - - name: Install UBSan runtime (musl) - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl' }} - shell: bash - run: | - set -euo pipefail - if command -v apt-get >/dev/null 2>&1; then - sudo apt-get update -y - sudo DEBIAN_FRONTEND=noninteractive apt-get install -y libubsan1 - fi - uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0 with: targets: ${{ matrix.target }} @@ -283,30 +274,7 @@ jobs: run: bash "${GITHUB_WORKSPACE}/.github/scripts/install-musl-build-tools.sh" - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Configure rustc UBSan wrapper (musl host) - shell: bash - run: | - set -euo pipefail - ubsan="" - if command -v ldconfig >/dev/null 2>&1; then - ubsan="$(ldconfig -p | grep -m1 'libubsan\.so\.1' | sed -E 's/.*=> (.*)$/\1/')" - fi - wrapper_root="${RUNNER_TEMP:-/tmp}" - wrapper="${wrapper_root}/rustc-ubsan-wrapper" - cat > "${wrapper}" <> "$GITHUB_ENV" - echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV" - - - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Clear sanitizer flags (musl) + name: Disable aws-lc jitter entropy (musl) shell: bash run: | set -euo pipefail @@ -316,30 +284,6 @@ jobs: target_no_jitter="${target_no_jitter//-/_}" echo "${target_no_jitter}=1" >> "$GITHUB_ENV" - # Clear global Rust flags so host/proc-macro builds don't pull in UBSan. - echo "RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_ENCODED_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "RUSTDOCFLAGS=" >> "$GITHUB_ENV" - # Override any runner-level Cargo config rustflags as well. - echo "CARGO_BUILD_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV" - echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV" - - sanitize_flags() { - local input="$1" - input="${input//-fsanitize=undefined/}" - input="${input//-fno-sanitize-recover=undefined/}" - input="${input//-fno-sanitize-trap=undefined/}" - echo "$input" - } - - cflags="$(sanitize_flags "${CFLAGS-}")" - cxxflags="$(sanitize_flags "${CXXFLAGS-}")" - echo "CFLAGS=${cflags}" >> "$GITHUB_ENV" - echo "CXXFLAGS=${cxxflags}" >> "$GITHUB_ENV" - - name: Configure rusty_v8 artifact overrides and verify checksums uses: ./.github/actions/setup-rusty-v8 with: From 2066874415e1c3dcd1ec28cbf26506dd74e35d2e Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Thu, 28 May 2026 09:13:43 -0700 Subject: [PATCH 4/6] extension-api: add TurnItemEmitter to tool calls (#24813) ## Why Extension-contributed tools need to emit visible turn items through Codex's normal event and persistence pipeline. ## What - Add `TurnItemEmitter` to extension `ToolCall`s and route the core implementation through `Session::emit_turn_item_*`. - Hold weak session and turn references so retained tool calls cannot keep host state alive. - Provide a no-op emitter for extension test callers. ## Test Plan - `just test -p codex-core -E 'test(passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call)'` --------- Co-authored-by: jif-oai --- .../src/tools/handlers/extension_tools.rs | 109 +++++++++++++++++- codex-rs/ext/extension-api/src/lib.rs | 4 + .../ext/goal/tests/goal_extension_backend.rs | 2 + codex-rs/ext/memories/src/tests.rs | 7 ++ codex-rs/tools/src/lib.rs | 4 + codex-rs/tools/src/tool_call.rs | 58 +++++++++- 6 files changed, 179 insertions(+), 5 deletions(-) diff --git a/codex-rs/core/src/tools/handlers/extension_tools.rs b/codex-rs/core/src/tools/handlers/extension_tools.rs index 6e6ef019c8..e9b16e97d5 100644 --- a/codex-rs/core/src/tools/handlers/extension_tools.rs +++ b/codex-rs/core/src/tools/handlers/extension_tools.rs @@ -1,11 +1,18 @@ use std::sync::Arc; +use std::sync::Weak; +use codex_protocol::items::TurnItem; use codex_tools::ConversationHistory; +use codex_tools::ExtensionTurnItem; use codex_tools::ToolCall as ExtensionToolCall; use codex_tools::ToolName; use codex_tools::ToolSpec; +use codex_tools::TurnItemEmissionFuture; +use codex_tools::TurnItemEmitter; use crate::function_tool::FunctionCallError; +use crate::session::session::Session; +use crate::session::turn_context::TurnContext; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; @@ -52,6 +59,39 @@ impl CoreToolRuntime for ExtensionToolAdapter { } } +struct CoreTurnItemEmitter { + session: Weak, + turn: Weak, +} + +fn extension_turn_item(item: ExtensionTurnItem) -> TurnItem { + match item { + ExtensionTurnItem::WebSearch(item) => TurnItem::WebSearch(item), + } +} + +impl TurnItemEmitter for CoreTurnItemEmitter { + fn emit_started<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> { + Box::pin(async move { + let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else { + return; + }; + let item = extension_turn_item(item); + session.emit_turn_item_started(turn.as_ref(), &item).await; + }) + } + + fn emit_completed<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> { + Box::pin(async move { + let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else { + return; + }; + let item = extension_turn_item(item); + session.emit_turn_item_completed(turn.as_ref(), item).await; + }) + } +} + async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall { let conversation_history = ConversationHistory::new(invocation.session.clone_history().await.into_raw_items()); @@ -61,6 +101,10 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall { tool_name: invocation.tool_name.clone(), truncation_policy: invocation.turn.truncation_policy, conversation_history, + turn_item_emitter: Arc::new(CoreTurnItemEmitter { + session: Arc::downgrade(&invocation.session), + turn: Arc::downgrade(&invocation.turn), + }), payload: invocation.payload.clone(), } } @@ -69,8 +113,13 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall { mod tests { use std::sync::Arc; + use codex_protocol::items::TurnItem; + use codex_protocol::items::WebSearchItem; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; + use codex_protocol::models::WebSearchAction; + use codex_protocol::protocol::EventMsg; + use codex_tools::ExtensionTurnItem; use pretty_assertions::assert_eq; use serde_json::json; use tokio::sync::Mutex; @@ -147,6 +196,16 @@ mod tests { &self, call: codex_tools::ToolCall, ) -> Result, codex_tools::FunctionCallError> { + let item = ExtensionTurnItem::WebSearch(WebSearchItem { + id: call.call_id.clone(), + query: "rust trait object".to_string(), + action: WebSearchAction::Search { + query: Some("rust trait object".to_string()), + queries: None, + }, + }); + call.turn_item_emitter.emit_started(item.clone()).await; + call.turn_item_emitter.emit_completed(item).await; *self.captured_call.lock().await = Some(call); Ok(Box::new(codex_tools::JsonToolOutput::new( json!({ "ok": true }), @@ -191,12 +250,14 @@ mod tests { } #[tokio::test] - async fn passes_turn_fields_to_extension_call() { + async fn passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call() { let captured_call = Arc::new(Mutex::new(None)); let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor { captured_call: Arc::clone(&captured_call), })); - let (session, turn) = crate::session::tests::make_session_and_context().await; + let (session, turn, rx) = crate::session::tests::make_session_and_context_with_rx().await; + let weak_session = Arc::downgrade(&session); + let weak_turn = Arc::downgrade(&turn); let turn_id = turn.sub_id.clone(); let truncation_policy = turn.truncation_policy; let history_item = ResponseItem::Message { @@ -211,8 +272,8 @@ mod tests { .record_conversation_items(&turn, std::slice::from_ref(&history_item)) .await; let invocation = ToolInvocation { - session: session.into(), - turn: turn.into(), + session, + turn, cancellation_token: tokio_util::sync::CancellationToken::new(), tracker: Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())), call_id: "call-extension".to_string(), @@ -228,6 +289,8 @@ mod tests { .expect("extension call should succeed"); let captured_call = captured_call.lock().await.clone().expect("captured call"); + assert!(weak_session.upgrade().is_none()); + assert!(weak_turn.upgrade().is_none()); assert_eq!(captured_call.turn_id, turn_id); assert_eq!(captured_call.call_id, "call-extension"); assert_eq!( @@ -245,5 +308,43 @@ mod tests { } payload => panic!("expected function payload, got {payload:?}"), } + + let started = rx.recv().await.expect("item started event"); + let EventMsg::ItemStarted(started) = started.msg else { + panic!("expected item started event"); + }; + let TurnItem::WebSearch(started_item) = started.item else { + panic!("expected web search item"); + }; + let begin = rx.recv().await.expect("legacy web search begin event"); + let EventMsg::WebSearchBegin(begin) = begin.msg else { + panic!("expected legacy web search begin event"); + }; + let completed = rx.recv().await.expect("item completed event"); + let EventMsg::ItemCompleted(completed) = completed.msg else { + panic!("expected item completed event"); + }; + let TurnItem::WebSearch(completed_item) = completed.item else { + panic!("expected web search item"); + }; + let end = rx.recv().await.expect("legacy web search end event"); + let EventMsg::WebSearchEnd(end) = end.msg else { + panic!("expected legacy web search end event"); + }; + + let expected = WebSearchItem { + id: "call-extension".to_string(), + query: "rust trait object".to_string(), + action: WebSearchAction::Search { + query: Some("rust trait object".to_string()), + queries: None, + }, + }; + assert_eq!(started_item, expected); + assert_eq!(completed_item, expected); + assert_eq!(begin.call_id, expected.id); + assert_eq!(end.call_id, expected.id); + assert_eq!(end.query, expected.query); + assert_eq!(end.action, expected.action); } } diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index a6618bdfa5..4705581c8a 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -11,8 +11,10 @@ pub use capabilities::NoopResponseItemInjector; pub use capabilities::ResponseItemInjectionFuture; pub use capabilities::ResponseItemInjector; pub use codex_tools::ConversationHistory; +pub use codex_tools::ExtensionTurnItem; pub use codex_tools::FunctionCallError; pub use codex_tools::JsonToolOutput; +pub use codex_tools::NoopTurnItemEmitter; pub use codex_tools::ResponsesApiTool; pub use codex_tools::ToolCall; pub use codex_tools::ToolExecutor; @@ -20,6 +22,8 @@ pub use codex_tools::ToolName; pub use codex_tools::ToolOutput; pub use codex_tools::ToolPayload; pub use codex_tools::ToolSpec; +pub use codex_tools::TurnItemEmissionFuture; +pub use codex_tools::TurnItemEmitter; pub use codex_tools::parse_tool_input_schema; pub use codex_tools::parse_tool_input_schema_without_compaction; pub use contributors::ApprovalReviewContributor; diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index ed0f59357b..7524d40fe1 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -8,6 +8,7 @@ use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::FunctionCallError; +use codex_extension_api::NoopTurnItemEmitter; use codex_extension_api::ThreadResumeInput; use codex_extension_api::ThreadStartInput; use codex_extension_api::ToolCall; @@ -1121,6 +1122,7 @@ fn tool_call(tool_name: &str, call_id: &str, arguments: serde_json::Value) -> To tool_name: codex_extension_api::ToolName::plain(tool_name), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload: ToolPayload::Function { arguments: arguments.to_string(), }, diff --git a/codex-rs/ext/memories/src/tests.rs b/codex-rs/ext/memories/src/tests.rs index b70c5f802d..f2d79c850c 100644 --- a/codex-rs/ext/memories/src/tests.rs +++ b/codex-rs/ext/memories/src/tests.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use codex_extension_api::ContextContributor; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::NoopTurnItemEmitter; use codex_extension_api::PromptSlot; use codex_extension_api::ToolCall; use codex_extension_api::ToolContributor; @@ -212,6 +213,7 @@ async fn add_ad_hoc_note_tool_creates_note_file() { tool_name: memory_tool_name(crate::ADD_AD_HOC_NOTE_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload: payload.clone(), }) .await @@ -253,6 +255,7 @@ async fn add_ad_hoc_note_tool_rejects_paths_as_filenames() { tool_name: memory_tool_name(crate::ADD_AD_HOC_NOTE_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload, }) .await; @@ -295,6 +298,7 @@ async fn read_tool_reads_memory_file() { tool_name: memory_tool_name(crate::READ_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload: payload.clone(), }) .await @@ -340,6 +344,7 @@ async fn search_tool_accepts_multiple_queries() { tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload: payload.clone(), }) .await @@ -411,6 +416,7 @@ async fn search_tool_accepts_windowed_all_match_mode() { tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload: payload.clone(), }) .await @@ -462,6 +468,7 @@ async fn search_tool_rejects_legacy_single_query() { tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME), truncation_policy: TruncationPolicy::Bytes(1024), conversation_history: codex_extension_api::ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), payload, }) .await; diff --git a/codex-rs/tools/src/lib.rs b/codex-rs/tools/src/lib.rs index 4c0fb44a7c..91fa5830a3 100644 --- a/codex-rs/tools/src/lib.rs +++ b/codex-rs/tools/src/lib.rs @@ -62,7 +62,11 @@ pub use responses_api::mcp_tool_to_deferred_responses_api_tool; pub use responses_api::mcp_tool_to_responses_api_tool; pub use responses_api::tool_definition_to_responses_api_tool; pub use tool_call::ConversationHistory; +pub use tool_call::ExtensionTurnItem; +pub use tool_call::NoopTurnItemEmitter; pub use tool_call::ToolCall; +pub use tool_call::TurnItemEmissionFuture; +pub use tool_call::TurnItemEmitter; pub use tool_config::ShellCommandBackendConfig; pub use tool_config::ToolEnvironmentMode; pub use tool_config::ToolUserShellType; diff --git a/codex-rs/tools/src/tool_call.rs b/codex-rs/tools/src/tool_call.rs index 32d428648f..3c04e216a6 100644 --- a/codex-rs/tools/src/tool_call.rs +++ b/codex-rs/tools/src/tool_call.rs @@ -1,8 +1,11 @@ use crate::FunctionCallError; use crate::ToolName; use crate::ToolPayload; +use codex_protocol::items::WebSearchItem; use codex_protocol::models::ResponseItem; use codex_utils_output_truncation::TruncationPolicy; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; /// Raw response history snapshot available when an extension tool is invoked. @@ -23,17 +26,70 @@ impl ConversationHistory { } } +/// Future returned when an extension tool emits a visible turn-item lifecycle event. +pub type TurnItemEmissionFuture<'a> = Pin + Send + 'a>>; + +/// Visible turn items that an extension fully owns and may emit as-is. +/// +/// Add only item kinds that require no additional host finalization before +/// persistence or client delivery. Richer items need a host-owned publish path. +#[derive(Clone, Debug, PartialEq)] +pub enum ExtensionTurnItem { + WebSearch(WebSearchItem), +} + +/// Host-provided capability for extension tools to emit finalized visible turn items. +/// +/// Implementations route lifecycle events through the host's normal item event +/// pipeline, including any persistence and client delivery owned by the host. +pub trait TurnItemEmitter: Send + Sync { + /// Emits the beginning of one visible turn item. + fn emit_started<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a>; + + /// Emits the completion of one visible turn item. + fn emit_completed<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a>; +} + +/// Turn-item emitter used when a caller does not expose visible item emission. +#[derive(Debug, Default, Clone, Copy)] +pub struct NoopTurnItemEmitter; + +impl TurnItemEmitter for NoopTurnItemEmitter { + fn emit_started<'a>(&'a self, _item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> { + Box::pin(std::future::ready(())) + } + + fn emit_completed<'a>(&'a self, _item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> { + Box::pin(std::future::ready(())) + } +} + // TODO: this is temporary and will disappear in the next PR (as we make codex-extension-api generic on Invocation. -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct ToolCall { pub turn_id: String, pub call_id: String, pub tool_name: ToolName, pub truncation_policy: TruncationPolicy, pub conversation_history: ConversationHistory, + pub turn_item_emitter: Arc, pub payload: ToolPayload, } +impl std::fmt::Debug for ToolCall { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ToolCall") + .field("turn_id", &self.turn_id) + .field("call_id", &self.call_id) + .field("tool_name", &self.tool_name) + .field("truncation_policy", &self.truncation_policy) + .field("conversation_history", &self.conversation_history) + .field("turn_item_emitter", &"") + .field("payload", &self.payload) + .finish() + } +} + impl ToolCall { pub fn function_arguments(&self) -> Result<&str, FunctionCallError> { match &self.payload { From 2a1158b8e2f941afed79db95731b16c8a8db5774 Mon Sep 17 00:00:00 2001 From: Brent Traut Date: Thu, 28 May 2026 09:18:13 -0700 Subject: [PATCH 5/6] feat(app-server): include turns page on thread resume (#23534) ## Summary The client currently calls `thread/resume` to establish live updates and immediately follows it with `thread/turns/list` to hydrate recent turns. This lets `thread/resume` return that page directly, eliminating a round trip and the ordering/deduplication gap between the two calls. Experimental clients opt in with `initialTurnsPage: { limit, sortDirection, itemsView }`. The response returns `initialTurnsPage` as a `TurnsPage`, including cursors for paging further back in history. Keeping the controls in a nested opt-in object provides the useful `thread/turns/list` knobs without spreading page-specific parameters across `thread/resume`. ## Verification - `just fmt` - `just write-app-server-schema --experimental` - `just write-app-server-schema` - `cargo test -p codex-app-server-protocol` - `cargo test -p codex-app-server thread_resume_initial_turns_page_matches_requested_turns_list_page --tests` - `cargo test -p codex-app-server thread_resume_rejoins_running_thread_even_with_override_mismatch --tests` - `just fix -p codex-app-server-protocol -p codex-app-server` --- .../analytics/src/analytics_client_tests.rs | 1 + codex-rs/analytics/src/client_tests.rs | 1 + .../schema/json/ClientRequest.json | 36 ++++ .../codex_app_server_protocol.schemas.json | 62 +++++++ .../codex_app_server_protocol.v2.schemas.json | 62 +++++++ .../schema/json/v2/ThreadResumeParams.json | 68 +++++++ .../schema/json/v2/ThreadResumeResponse.json | 26 +++ .../v2/ThreadResumeInitialTurnsPageParams.ts | 19 ++ .../schema/typescript/v2/TurnsPage.ts | 6 + .../schema/typescript/v2/index.ts | 2 + .../src/protocol/v2/tests.rs | 80 ++++++++ .../src/protocol/v2/thread.rs | 43 +++++ codex-rs/app-server/README.md | 20 ++ codex-rs/app-server/src/request_processors.rs | 1 + .../request_processors/thread_lifecycle.rs | 23 ++- .../request_processors/thread_processor.rs | 172 ++++++++++++------ .../thread_processor_tests.rs | 1 + .../thread_resume_redaction.rs | 12 +- codex-rs/app-server/src/thread_state.rs | 2 + .../app-server/tests/suite/v2/thread_read.rs | 72 ++++++++ .../tests/suite/v2/thread_resume.rs | 121 +++++++----- codex-rs/tui/src/app_server_session.rs | 1 + 22 files changed, 730 insertions(+), 101 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index fa64b46d7a..48379cc39b 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -263,6 +263,7 @@ fn sample_thread_resume_response_with_source( sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }) } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index c6a928afa3..d0f8fc40e2 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -172,6 +172,7 @@ fn sample_thread_resume_response() -> ClientResponsePayload { sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }) } diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 63e8fbde1a..a09d793bfe 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -3543,6 +3543,42 @@ } ] }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 28dc006b5e..1d1a64a2fc 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -17150,6 +17150,42 @@ "title": "ThreadRealtimeTranscriptDoneNotification", "type": "object" }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/v2/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/v2/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", @@ -18488,6 +18524,32 @@ "title": "TurnSteerResponse", "type": "object" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/v2/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index d1c59e9855..1b49b38641 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -14974,6 +14974,42 @@ "title": "ThreadRealtimeTranscriptDoneNotification", "type": "object" }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, "ThreadResumeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", @@ -16312,6 +16348,32 @@ "title": "TurnSteerResponse", "type": "object" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json index 58b85d41a4..364cb36617 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json @@ -983,6 +983,74 @@ "danger-full-access" ], "type": "string" + }, + "SortDirection": { + "enum": [ + "asc", + "desc" + ], + "type": "string" + }, + "ThreadResumeInitialTurnsPageParams": { + "properties": { + "itemsView": { + "anyOf": [ + { + "$ref": "#/definitions/TurnItemsView" + }, + { + "type": "null" + } + ], + "description": "How much item detail to include for each returned turn; defaults to summary." + }, + "limit": { + "description": "Optional turn page size.", + "format": "uint32", + "minimum": 0.0, + "type": [ + "integer", + "null" + ] + }, + "sortDirection": { + "anyOf": [ + { + "$ref": "#/definitions/SortDirection" + }, + { + "type": "null" + } + ], + "description": "Optional turn pagination direction; defaults to descending." + } + }, + "type": "object" + }, + "TurnItemsView": { + "oneOf": [ + { + "description": "`items` was not loaded for this turn. The field is intentionally empty.", + "enum": [ + "notLoaded" + ], + "type": "string" + }, + { + "description": "`items` contains only a display summary for this turn.", + "enum": [ + "summary" + ], + "type": "string" + }, + { + "description": "`items` contains every ThreadItem available from persisted app-server history for this turn.", + "enum": [ + "full" + ], + "type": "string" + } + ] } }, "description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index b4fa181c88..8f76893af8 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -1995,6 +1995,32 @@ ], "type": "string" }, + "TurnsPage": { + "properties": { + "backwardsCursor": { + "type": [ + "string", + "null" + ] + }, + "data": { + "items": { + "$ref": "#/definitions/Turn" + }, + "type": "array" + }, + "nextCursor": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "data" + ], + "type": "object" + }, "UserInput": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts new file mode 100644 index 0000000000..2dbcd97807 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeInitialTurnsPageParams.ts @@ -0,0 +1,19 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { SortDirection } from "./SortDirection"; +import type { TurnItemsView } from "./TurnItemsView"; + +export type ThreadResumeInitialTurnsPageParams = { +/** + * Optional turn page size. + */ +limit?: number | null, +/** + * Optional turn pagination direction; defaults to descending. + */ +sortDirection?: SortDirection | null, +/** + * How much item detail to include for each returned turn; defaults to summary. + */ +itemsView?: TurnItemsView | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts new file mode 100644 index 0000000000..e91865aeeb --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/TurnsPage.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { Turn } from "./Turn"; + +export type TurnsPage = { data: Array, nextCursor: string | null, backwardsCursor: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 2a4220cb60..0b7316fc4f 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -398,6 +398,7 @@ export type { ThreadRealtimeStartTransport } from "./ThreadRealtimeStartTranspor export type { ThreadRealtimeStartedNotification } from "./ThreadRealtimeStartedNotification"; export type { ThreadRealtimeTranscriptDeltaNotification } from "./ThreadRealtimeTranscriptDeltaNotification"; export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeTranscriptDoneNotification"; +export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams"; export type { ThreadResumeParams } from "./ThreadResumeParams"; export type { ThreadResumeResponse } from "./ThreadResumeResponse"; export type { ThreadRollbackParams } from "./ThreadRollbackParams"; @@ -450,6 +451,7 @@ export type { TurnStartedNotification } from "./TurnStartedNotification"; export type { TurnStatus } from "./TurnStatus"; export type { TurnSteerParams } from "./TurnSteerParams"; export type { TurnSteerResponse } from "./TurnSteerResponse"; +export type { TurnsPage } from "./TurnsPage"; export type { UserInput } from "./UserInput"; export type { WarningNotification } from "./WarningNotification"; export type { WebSearchAction } from "./WebSearchAction"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 8bc01e93d9..0465a4e012 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -111,6 +111,85 @@ fn thread_turns_list_params_accepts_items_view() { assert_eq!(params.items_view, Some(TurnItemsView::NotLoaded)); } +#[test] +fn thread_resume_params_accept_turns_page_bootstrap() { + let params = serde_json::from_value::(json!({ + "threadId": "thr_123", + "initialTurnsPage": { + "limit": 25, + "sortDirection": "asc", + "itemsView": "full", + }, + })) + .expect("thread resume params should deserialize"); + + assert_eq!(params.thread_id, "thr_123"); + assert_eq!( + params.initial_turns_page, + Some(ThreadResumeInitialTurnsPageParams { + limit: Some(25), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::Full), + }) + ); +} + +#[test] +fn thread_resume_response_round_trips_initial_turns_page() { + let response = ThreadResumeResponse { + thread: Thread { + id: "thr_123".to_string(), + session_id: "thr_123".to_string(), + forked_from_id: None, + preview: String::new(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 1, + status: ThreadStatus::Idle, + path: None, + cwd: absolute_path("tmp"), + cli_version: "0.0.0".to_string(), + source: SessionSource::Exec, + thread_source: None, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + }, + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: absolute_path("tmp"), + runtime_workspace_roots: Vec::new(), + instruction_sources: Vec::new(), + approval_policy: AskForApproval::OnFailure, + approvals_reviewer: ApprovalsReviewer::User, + sandbox: SandboxPolicy::DangerFullAccess, + active_permission_profile: None, + reasoning_effort: None, + initial_turns_page: Some(TurnsPage { + data: Vec::new(), + next_cursor: Some("cursor_next".to_string()), + backwards_cursor: Some("cursor_back".to_string()), + }), + }; + + let value = serde_json::to_value(&response).expect("serialize thread resume response"); + assert_eq!( + value.get("initialTurnsPage"), + Some(&json!({ + "data": [], + "nextCursor": "cursor_next", + "backwardsCursor": "cursor_back", + })) + ); + let decoded = serde_json::from_value::(value) + .expect("deserialize thread resume response"); + assert_eq!(decoded, response); +} + #[test] fn thread_turns_items_list_round_trips() { let params = ThreadTurnsItemsListParams { @@ -3406,6 +3485,7 @@ fn thread_lifecycle_responses_default_missing_optional_fields() { assert_eq!(fork.instruction_sources, Vec::::new()); assert_eq!(start.active_permission_profile, None); assert_eq!(resume.active_permission_profile, None); + assert_eq!(resume.initial_turns_page, None); assert_eq!(fork.active_permission_profile, None); } diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 4971d5f4b8..8157fce69f 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -397,6 +397,11 @@ pub struct ThreadResumeParams { #[experimental("thread/resume.excludeTurns")] #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub exclude_turns: bool, + /// When present, include a `thread/turns/list` page in the resume response + /// so clients can bootstrap recent turns without a second request. + #[experimental("thread/resume.initialTurnsPage")] + #[ts(optional = nullable)] + pub initial_turns_page: Option, /// Deprecated and ignored by app-server. Kept only so older clients can /// continue sending the field while rollout persistence always uses the /// limited history policy. @@ -435,6 +440,44 @@ pub struct ThreadResumeResponse { #[serde(default)] pub active_permission_profile: Option, pub reasoning_effort: Option, + /// `thread/turns/list` page returned when requested by `initialTurnsPage`. + #[experimental("thread/resume.initialTurnsPage")] + #[serde(default)] + pub initial_turns_page: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadResumeInitialTurnsPageParams { + /// Optional turn page size. + #[ts(optional = nullable)] + pub limit: Option, + /// Optional turn pagination direction; defaults to descending. + #[ts(optional = nullable)] + pub sort_direction: Option, + /// How much item detail to include for each returned turn; defaults to summary. + #[ts(optional = nullable)] + pub items_view: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnsPage { + pub data: Vec, + pub next_cursor: Option, + pub backwards_cursor: Option, +} + +impl From for TurnsPage { + fn from(response: ThreadTurnsListResponse) -> Self { + Self { + data: response.data, + next_cursor: response.next_cursor, + backwards_cursor: response.backwards_cursor, + } + } } #[derive( diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 88edc22f9e..d7dfb00fc1 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -281,6 +281,8 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage. +Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. + By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead. Example: @@ -297,6 +299,24 @@ Example: "excludeTurns": true } } { "id": 12, "result": { "thread": { "id": "thr_123", "turns": [], … } } } + +{ "method": "thread/resume", "id": 13, "params": { + "threadId": "thr_123", + "excludeTurns": true, + "initialTurnsPage": { + "limit": 20, + "sortDirection": "desc", + "itemsView": "summary" + } +} } +{ "id": 13, "result": { + "thread": { "id": "thr_123", "turns": [], … }, + "initialTurnsPage": { + "data": [ ... ], + "nextCursor": "older-turns-cursor-or-null", + "backwardsCursor": "newer-turns-cursor-or-null" + } +} } ``` To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. The returned `thread.sessionId` identifies the current live session tree root. Root threads use their own `thread.id` as `thread.sessionId`; stored threads that are not loaded also report their own `thread.id`, because resuming one makes it the root of a new live session tree. When the source history includes persisted token usage, the server also emits `thread/tokenUsage/updated` for the new thread immediately after the response. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only: diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 66ab8e3a1d..8408a3f67f 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -213,6 +213,7 @@ use codex_app_server_protocol::ThreadRealtimeStartResponse; use codex_app_server_protocol::ThreadRealtimeStartTransport; use codex_app_server_protocol::ThreadRealtimeStopParams; use codex_app_server_protocol::ThreadRealtimeStopResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadRollbackParams; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 985baac91a..f72cfb5a62 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -565,8 +565,28 @@ pub(super) async fn handle_pending_thread_resume_request( has_live_in_progress_turn, ); let token_usage_thread = pending.include_turns.then(|| thread.clone()); + let mut initial_turns_page = if let Some(params) = pending.initial_turns_page.as_ref() { + match super::thread_processor::build_thread_resume_initial_turns_page( + &pending.history_items, + thread.status.clone(), + has_live_in_progress_turn, + active_turn, + params, + ) { + Ok(page) => Some(page), + Err(error) => { + outgoing.send_error(request_id, error).await; + return; + } + } + } else { + None + }; if pending.redact_resume_payloads { - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); + if let Some(initial_turns_page) = initial_turns_page.as_mut() { + redact_thread_resume_payloads(&mut initial_turns_page.data); + } } { @@ -635,6 +655,7 @@ pub(super) async fn handle_pending_thread_resume_request( sandbox, active_permission_profile, reasoning_effort, + initial_turns_page, }; outgoing.send_response(request_id, response).await; // Match cold resume: metadata-only resume should attach the listener without diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 5a2e1a4de8..939c3df519 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2241,8 +2241,6 @@ impl ThreadRequestProcessor { sort_direction, items_view, } = params; - let items_view = items_view.unwrap_or(TurnItemsView::Summary); - let thread_uuid = ThreadId::from_string(&thread_id) .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; @@ -2270,60 +2268,20 @@ impl ThreadRequestProcessor { } else { None }; - let mut turns = reconstruct_thread_turns_for_turns_list( + build_thread_turns_page_response( &items, self.thread_watch_manager .loaded_status_for_thread(&thread_uuid.to_string()) .await, has_live_running_thread, active_turn, - ); - for turn in &mut turns { - match items_view { - TurnItemsView::NotLoaded => { - turn.items.clear(); - turn.items_view = TurnItemsView::NotLoaded; - } - TurnItemsView::Summary => { - let first_user_message = turn - .items - .iter() - .find(|item| matches!(item, ThreadItem::UserMessage { .. })) - .cloned(); - let final_agent_message = turn - .items - .iter() - .rev() - .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) - .cloned(); - turn.items = match (first_user_message, final_agent_message) { - (Some(user_message), Some(agent_message)) - if user_message.id() != agent_message.id() => - { - vec![user_message, agent_message] - } - (Some(user_message), _) => vec![user_message], - (None, Some(agent_message)) => vec![agent_message], - (None, None) => Vec::new(), - }; - turn.items_view = TurnItemsView::Summary; - } - TurnItemsView::Full => { - turn.items_view = TurnItemsView::Full; - } - } - } - let page = paginate_thread_turns( - turns, - cursor.as_deref(), - limit, - sort_direction.unwrap_or(SortDirection::Desc), - )?; - Ok(ThreadTurnsListResponse { - data: page.turns, - next_cursor: page.next_cursor, - backwards_cursor: page.backwards_cursor, - }) + ThreadTurnsPageOptions { + cursor: cursor.as_deref(), + limit, + sort_direction: sort_direction.unwrap_or(SortDirection::Desc), + items_view: items_view.unwrap_or(TurnItemsView::Summary), + }, + ) } async fn load_thread_turns_list_history( @@ -2531,6 +2489,7 @@ impl ThreadRequestProcessor { developer_instructions, personality, exclude_turns, + initial_turns_page, persist_extended_history: _persist_extended_history, } = params; let include_turns = !exclude_turns; @@ -2685,8 +2644,28 @@ impl ThreadRequestProcessor { config_snapshot.active_permission_profile, ); let token_usage_thread = include_turns.then(|| thread.clone()); + let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() { + match build_thread_resume_initial_turns_page( + &response_history.get_rollout_items(), + thread.status.clone(), + /*has_live_running_thread*/ false, + /*active_turn*/ None, + params, + ) { + Ok(page) => Some(page), + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + } + } else { + None + }; if redact_resume_payloads { - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); + if let Some(initial_turns_page) = initial_turns_page.as_mut() { + redact_thread_resume_payloads(&mut initial_turns_page.data); + } } let response = ThreadResumeResponse { @@ -2702,6 +2681,7 @@ impl ThreadRequestProcessor { sandbox, active_permission_profile, reasoning_effort: session_configured.reasoning_effort, + initial_turns_page, }; let connection_id = request_id.connection_id; @@ -2926,6 +2906,7 @@ impl ThreadRequestProcessor { emit_thread_goal_update, thread_goal_state_db, include_turns: !params.exclude_turns, + initial_turns_page: params.initial_turns_page.clone(), redact_resume_payloads, }), ); @@ -3706,6 +3687,95 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result { + cursor: Option<&'a str>, + limit: Option, + sort_direction: SortDirection, + items_view: TurnItemsView, +} + +fn build_thread_turns_page_response( + items: &[RolloutItem], + loaded_status: ThreadStatus, + has_live_running_thread: bool, + active_turn: Option, + options: ThreadTurnsPageOptions<'_>, +) -> Result { + let mut turns = reconstruct_thread_turns_for_turns_list( + items, + loaded_status, + has_live_running_thread, + active_turn, + ); + apply_thread_turns_items_view(&mut turns, options.items_view); + let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?; + Ok(ThreadTurnsListResponse { + data: page.turns, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) +} + +pub(super) fn build_thread_resume_initial_turns_page( + items: &[RolloutItem], + loaded_status: ThreadStatus, + has_live_running_thread: bool, + active_turn: Option, + params: &ThreadResumeInitialTurnsPageParams, +) -> Result { + build_thread_turns_page_response( + items, + loaded_status, + has_live_running_thread, + active_turn, + ThreadTurnsPageOptions { + cursor: None, + limit: params.limit, + sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc), + items_view: params.items_view.unwrap_or(TurnItemsView::Summary), + }, + ) + .map(Into::into) +} + +fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { + for turn in turns { + match items_view { + TurnItemsView::NotLoaded => { + turn.items.clear(); + turn.items_view = TurnItemsView::NotLoaded; + } + TurnItemsView::Summary => { + let first_user_message = turn + .items + .iter() + .find(|item| matches!(item, ThreadItem::UserMessage { .. })) + .cloned(); + let final_agent_message = turn + .items + .iter() + .rev() + .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) + .cloned(); + turn.items = match (first_user_message, final_agent_message) { + (Some(user_message), Some(agent_message)) + if user_message.id() != agent_message.id() => + { + vec![user_message, agent_message] + } + (Some(user_message), _) => vec![user_message], + (None, Some(agent_message)) => vec![agent_message], + (None, None) => Vec::new(), + }; + turn.items_view = TurnItemsView::Summary; + } + TurnItemsView::Full => { + turn.items_view = TurnItemsView::Full; + } + } + } +} + fn reconstruct_thread_turns_for_turns_list( items: &[RolloutItem], loaded_status: ThreadStatus, diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index ab8f52ffc0..2fa95b3d23 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -652,6 +652,7 @@ mod thread_processor_behavior_tests { developer_instructions: None, personality: None, exclude_turns: false, + initial_turns_page: None, persist_extended_history: false, }; let config_snapshot = ThreadConfigSnapshot { diff --git a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs index b1e8608842..b265e3a632 100644 --- a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs +++ b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs @@ -1,6 +1,6 @@ use codex_app_server_protocol::McpToolCallResult; -use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::Turn; use serde_json::Value as JsonValue; // Temporary bandaid for remote clients: thread/resume can include large MCP and @@ -14,8 +14,8 @@ pub(super) fn should_redact_thread_resume_payloads(client_name: Option<&str>) -> client_name.is_some_and(|client_name| CHATGPT_REMOTE_CLIENT_NAMES.contains(&client_name)) } -pub(super) fn redact_thread_resume_payloads(thread: &mut Thread) { - for turn in &mut thread.turns { +pub(super) fn redact_thread_resume_payloads(turns: &mut [Turn]) { + for turn in turns { turn.items.retain_mut(|item| match item { ThreadItem::McpToolCall { arguments, @@ -55,8 +55,8 @@ mod tests { use codex_app_server_protocol::McpToolCallError; use codex_app_server_protocol::McpToolCallStatus; use codex_app_server_protocol::SessionSource; + use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadStatus; - use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStatus; use codex_utils_absolute_path::test_support::PathBufExt; @@ -100,7 +100,7 @@ mod tests { }, ]); - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); assert_eq!(thread.turns[0].items.len(), 2); assert_eq!( @@ -146,7 +146,7 @@ mod tests { duration_ms: Some(8), }]); - redact_thread_resume_payloads(&mut thread); + redact_thread_resume_payloads(&mut thread.turns); assert_eq!( thread.turns[0].items[0], diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 7f864265b2..2d932f0326 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -35,6 +35,8 @@ pub(crate) struct PendingThreadResumeRequest { pub(crate) emit_thread_goal_update: bool, pub(crate) thread_goal_state_db: Option, pub(crate) include_turns: bool, + pub(crate) initial_turns_page: + Option, pub(crate) redact_resume_payloads: bool, } diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index c8e3e179fb..5ba9fa9d2d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -24,6 +24,7 @@ use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadNameUpdatedNotification; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSetNameParams; @@ -631,6 +632,77 @@ async fn thread_read_can_return_archived_threads_by_id() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let filename_ts = "2025-01-05T12-00-00"; + let conversation_id = create_fake_rollout_with_text_elements( + codex_home.path(), + filename_ts, + "2025-01-05T12:00:00Z", + "first", + vec![], + Some("mock_provider"), + /*git_info*/ None, + )?; + let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id); + append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?; + append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let turns_list_id = mcp + .send_thread_turns_list_request(ThreadTurnsListParams { + thread_id: conversation_id.clone(), + cursor: None, + limit: Some(2), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::NotLoaded), + }) + .await?; + let turns_list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turns_list_id)), + ) + .await??; + let expected_page = to_response::(turns_list_resp)?; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id, + exclude_turns: true, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: Some(2), + sort_direction: Some(SortDirection::Asc), + items_view: Some(TurnItemsView::NotLoaded), + }), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { + thread, + initial_turns_page, + .. + } = to_response::(resume_resp)?; + + assert!(thread.turns.is_empty()); + assert_eq!( + initial_turns_page, + Some(codex_app_server_protocol::TurnsPage::from(expected_page)) + ); + + Ok(()) +} + #[tokio::test] async fn thread_turns_list_rejects_cursor_when_anchor_turn_is_rolled_back() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index bb2de2cfbb..2b85427247 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -37,6 +37,7 @@ use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams; use codex_app_server_protocol::ThreadMetadataUpdateParams; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSource; @@ -530,48 +531,59 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { #[tokio::test] async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<()> { for client_name in ["codex_chatgpt_android_remote", "codex_chatgpt_ios_remote"] { - let remote_thread = resume_redaction_fixture(Some(client_name)).await?; - let remote_turn = remote_thread + let remote_resume = resume_redaction_fixture(Some(client_name)).await?; + let remote_turn = remote_resume + .thread .turns .first() .expect("remote resume should include a turn"); - let remote_mcp_item = remote_turn - .items - .iter() - .find(|item| matches!(item, ThreadItem::McpToolCall { .. })) - .expect("remote resume should include redacted MCP item"); - let ThreadItem::McpToolCall { - arguments, - result, - error, - .. - } = remote_mcp_item - else { - unreachable!("matched MCP item"); - }; - assert_eq!(arguments, &json!("[redacted]")); - let result = result.as_ref().expect("redacted MCP result"); - assert_eq!( - result.content, - vec![json!({ - "type": "text", - "text": "[redacted]", - })] - ); - assert_eq!(result.structured_content, None); - assert_eq!(result.meta, None); - assert_eq!(error, &None); - assert!( - !remote_turn + let remote_page_turn = remote_resume + .initial_turns_page + .as_ref() + .expect("remote resume should include the requested initial turns page") + .data + .first() + .expect("remote initial turns page should include a turn"); + for remote_turn in [remote_turn, remote_page_turn] { + let remote_mcp_item = remote_turn .items .iter() - .any(|item| matches!(item, ThreadItem::ImageGeneration { .. })), - "remote resume should drop image generation items for {client_name}" - ); + .find(|item| matches!(item, ThreadItem::McpToolCall { .. })) + .expect("remote resume should include redacted MCP item"); + let ThreadItem::McpToolCall { + arguments, + result, + error, + .. + } = remote_mcp_item + else { + unreachable!("matched MCP item"); + }; + assert_eq!(arguments, &json!("[redacted]")); + let result = result.as_ref().expect("redacted MCP result"); + assert_eq!( + result.content, + vec![json!({ + "type": "text", + "text": "[redacted]", + })] + ); + assert_eq!(result.structured_content, None); + assert_eq!(result.meta, None); + assert_eq!(error, &None); + assert!( + !remote_turn + .items + .iter() + .any(|item| matches!(item, ThreadItem::ImageGeneration { .. })), + "remote resume should drop image generation items for {client_name}" + ); + } } - let normal_thread = resume_redaction_fixture(Some("some_other_client")).await?; - let normal_turn = normal_thread + let normal_resume = resume_redaction_fixture(Some("some_other_client")).await?; + let normal_turn = normal_resume + .thread .turns .first() .expect("normal resume should include a turn"); @@ -616,9 +628,7 @@ async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<( Ok(()) } -async fn resume_redaction_fixture( - client_name: Option<&str>, -) -> Result { +async fn resume_redaction_fixture(client_name: Option<&str>) -> Result { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; @@ -658,6 +668,11 @@ async fn resume_redaction_fixture( let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: conversation_id, + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: None, + sort_direction: None, + items_view: Some(TurnItemsView::Full), + }), ..Default::default() }) .await?; @@ -666,8 +681,7 @@ async fn resume_redaction_fixture( mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; - Ok(thread) + to_response::(resume_resp) } fn append_resume_redaction_history( @@ -2448,11 +2462,13 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R ..Default::default() }) .await?; - timeout( + let running_turn_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_response_message(RequestId::Integer(running_turn_id)), ) .await??; + let TurnStartResponse { turn: running_turn } = + to_response::(running_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, primary.read_stream_until_notification_message("turn/started"), @@ -2464,6 +2480,11 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R thread_id: thread.id.clone(), model: Some("not-the-running-model".to_string()), cwd: Some("/tmp".to_string()), + initial_turns_page: Some(ThreadResumeInitialTurnsPageParams { + limit: None, + sort_direction: None, + items_view: None, + }), ..Default::default() }) .await?; @@ -2472,9 +2493,23 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R primary.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let ThreadResumeResponse { thread, model, .. } = - to_response::(resume_resp)?; + let ThreadResumeResponse { + thread, + model, + initial_turns_page, + .. + } = to_response::(resume_resp)?; assert_eq!(model, "gpt-5.4"); + let initial_turns_page = initial_turns_page.expect("resume should include initial turns page"); + let resumed_running_turn = initial_turns_page + .data + .first() + .expect("resume page should include the running turn"); + assert_eq!(resumed_running_turn.id, running_turn.id); + assert_eq!(resumed_running_turn.items_view, TurnItemsView::Summary); + assert_eq!(resumed_running_turn.status, TurnStatus::InProgress); + assert!(initial_turns_page.backwards_cursor.is_some()); + assert_eq!(initial_turns_page.next_cursor, None); // The running-thread resume response is queued onto the thread listener task. // If the in-flight turn completes before that queued command runs, the response // can legitimately observe the thread as idle. diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index aa2d5a645b..79f8e274b6 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -2300,6 +2300,7 @@ mod tests { .into(), active_permission_profile: None, reasoning_effort: None, + initial_turns_page: None, }; let started = started_thread_from_resume_response( From 1a6aba2f88205950dbf031aa37e2f6ceb7088266 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 28 May 2026 09:19:15 -0700 Subject: [PATCH 6/6] fix: surface denied filesystem reads in prompts Expose denied-read entries from the active permission profile to model-visible context and automatic approval review prompts so escalation decisions can account for policy-restricted paths. --- .../core/src/context/environment_context.rs | 91 ++++++++++++++++++- .../src/context/environment_context_tests.rs | 30 ++++++ .../src/context/permissions_instructions.rs | 78 ++++++++++++---- .../context/permissions_instructions_tests.rs | 45 +++++++++ codex-rs/core/src/guardian/mod.rs | 2 + codex-rs/core/src/guardian/prompt.rs | 49 ++++++++++ codex-rs/core/src/guardian/review_session.rs | 5 +- codex-rs/core/src/guardian/tests.rs | 62 +++++++++++++ 8 files changed, 336 insertions(+), 26 deletions(-) diff --git a/codex-rs/core/src/context/environment_context.rs b/codex-rs/core/src/context/environment_context.rs index a3052b6642..4077944c98 100644 --- a/codex-rs/core/src/context/environment_context.rs +++ b/codex-rs/core/src/context/environment_context.rs @@ -1,9 +1,11 @@ use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnEnvironment; use crate::shell::Shell; +use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::TurnContextNetworkItem; use codex_utils_absolute_path::AbsolutePathBuf; +use std::path::Path; use super::ContextualUserFragment; @@ -13,6 +15,7 @@ pub(crate) struct EnvironmentContext { pub(crate) current_date: Option, pub(crate) timezone: Option, pub(crate) network: Option, + pub(crate) filesystem: Option, pub(crate) subagents: Option, } @@ -83,6 +86,55 @@ impl EnvironmentContextEnvironments { } } +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub(crate) struct FileSystemContext { + unreadable_roots: Vec, + unreadable_globs: Vec, +} + +impl FileSystemContext { + fn new(unreadable_roots: Vec, unreadable_globs: Vec) -> Option { + if unreadable_roots.is_empty() && unreadable_globs.is_empty() { + None + } else { + Some(Self { + unreadable_roots, + unreadable_globs, + }) + } + } + + fn from_permission_profile(permission_profile: &PermissionProfile, cwd: &Path) -> Option { + let file_system_sandbox_policy = permission_profile.file_system_sandbox_policy(); + Self::new( + file_system_sandbox_policy + .get_unreadable_roots_with_cwd(cwd) + .into_iter() + .map(|root| root.to_string_lossy().into_owned()) + .collect(), + file_system_sandbox_policy.get_unreadable_globs_with_cwd(cwd), + ) + } + + fn render(&self) -> String { + let mut rendered = "".to_string(); + Self::push_rendered_element(&mut rendered, "paths", &self.unreadable_roots); + Self::push_rendered_element(&mut rendered, "globs", &self.unreadable_globs); + rendered.push_str(""); + rendered + } + + fn push_rendered_element(rendered_filesystem: &mut String, name: &str, values: &[String]) { + if values.is_empty() { + return; + } + + rendered_filesystem.push_str(&format!("<{name}>")); + rendered_filesystem.push_str(&values.join(",")); + rendered_filesystem.push_str(&format!("")); + } +} + #[derive(Debug, Clone, PartialEq, Eq, Default)] pub(crate) struct NetworkContext { allowed_domains: Vec, @@ -129,6 +181,7 @@ impl EnvironmentContext { current_date, timezone, network, + filesystem: None, subagents, } } @@ -138,6 +191,7 @@ impl EnvironmentContext { current_date: Option, timezone: Option, network: Option, + filesystem: Option, subagents: Option, ) -> Self { Self { @@ -145,6 +199,7 @@ impl EnvironmentContext { current_date, timezone, network, + filesystem, subagents, } } @@ -157,6 +212,7 @@ impl EnvironmentContext { && self.current_date == other.current_date && self.timezone == other.timezone && self.network == other.network + && self.filesystem == other.filesystem && self.subagents == other.subagents } @@ -165,6 +221,7 @@ impl EnvironmentContext { after: &EnvironmentContext, ) -> Self { let before_network = Self::network_from_turn_context_item(before); + let before_filesystem = Self::filesystem_from_turn_context_item(before); let environments = match &after.environments { EnvironmentContextEnvironments::Single(environment) => { if before.cwd.as_path() != environment.cwd.as_path() { @@ -186,17 +243,25 @@ impl EnvironmentContext { } else { before_network }; + let filesystem = if before_filesystem != after.filesystem { + after.filesystem.clone() + } else { + before_filesystem + }; EnvironmentContext::new_with_environments( environments, after.current_date.clone(), after.timezone.clone(), network, + filesystem, /*subagents*/ None, ) } pub(crate) fn from_turn_context(turn_context: &TurnContext, shell: &Shell) -> Self { - Self::new( + #[allow(deprecated)] + let cwd = &turn_context.cwd; + let mut context = Self::new( EnvironmentContextEnvironment::from_turn_environments( &turn_context.environments.turn_environments, shell, @@ -205,7 +270,10 @@ impl EnvironmentContext { turn_context.timezone.clone(), Self::network_from_turn_context(turn_context), /*subagents*/ None, - ) + ); + context.filesystem = + FileSystemContext::from_permission_profile(&turn_context.permission_profile, cwd); + context } pub(crate) fn from_turn_context_item( @@ -216,11 +284,14 @@ impl EnvironmentContext { Ok(cwd) => cwd, Err(_) => AbsolutePathBuf::resolve_path_against_base(&turn_context_item.cwd, "/"), }; - Self::new( - vec![EnvironmentContextEnvironment::legacy(cwd, shell)], + Self::new_with_environments( + EnvironmentContextEnvironments::from_vec(vec![EnvironmentContextEnvironment::legacy( + cwd, shell, + )]), turn_context_item.current_date.clone(), turn_context_item.timezone.clone(), Self::network_from_turn_context_item(turn_context_item), + Self::filesystem_from_turn_context_item(turn_context_item), /*subagents*/ None, ) } @@ -266,6 +337,15 @@ impl EnvironmentContext { denied_domains.clone(), )) } + + fn filesystem_from_turn_context_item( + turn_context_item: &TurnContextItem, + ) -> Option { + FileSystemContext::from_permission_profile( + &turn_context_item.permission_profile(), + &turn_context_item.cwd, + ) + } } impl ContextualUserFragment for EnvironmentContext { @@ -324,6 +404,9 @@ impl ContextualUserFragment for EnvironmentContext { // lines.push(" ".to_string()); } } + if let Some(filesystem) = &self.filesystem { + lines.push(format!(" {}", filesystem.render())); + } if let Some(subagents) = &self.subagents { lines.push(" ".to_string()); lines.extend(subagents.lines().map(|line| format!(" {line}"))); diff --git a/codex-rs/core/src/context/environment_context_tests.rs b/codex-rs/core/src/context/environment_context_tests.rs index 68ff7c9d44..2d0855b326 100644 --- a/codex-rs/core/src/context/environment_context_tests.rs +++ b/codex-rs/core/src/context/environment_context_tests.rs @@ -79,6 +79,36 @@ fn serialize_environment_context_with_network() { assert_eq!(context.render(), expected); } +#[test] +fn serialize_environment_context_with_filesystem_denied_reads() { + let mut context = EnvironmentContext::new( + vec![EnvironmentContextEnvironment { + id: "local".to_string(), + cwd: test_path_buf("/repo").abs(), + shell: fake_shell_name(), + }], + /*current_date*/ None, + /*timezone*/ None, + /*network*/ None, + /*subagents*/ None, + ); + context.filesystem = FileSystemContext::new( + vec!["/repo/private".to_string()], + vec!["/repo/private/**".to_string()], + ); + + let expected = format!( + r#" + {} + bash + /repo/private/repo/private/** +"#, + test_path_buf("/repo").display() + ); + + assert_eq!(context.render(), expected); +} + #[test] fn serialize_read_only_environment_context() { let context = EnvironmentContext::new( diff --git a/codex-rs/core/src/context/permissions_instructions.rs b/codex-rs/core/src/context/permissions_instructions.rs index db8f983c16..e87dceec01 100644 --- a/codex-rs/core/src/context/permissions_instructions.rs +++ b/codex-rs/core/src/context/permissions_instructions.rs @@ -4,6 +4,7 @@ use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::SandboxMode; use codex_protocol::models::PermissionProfile; use codex_protocol::models::format_allow_prefixes; +use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::GranularApprovalConfig; @@ -68,9 +69,11 @@ impl PermissionsInstructions { exec_permission_approvals_enabled: bool, request_permissions_tool_enabled: bool, ) -> Self { - let (sandbox_mode, writable_roots) = sandbox_prompt_from_profile(permission_profile, cwd); + let file_system_sandbox_policy = permission_profile.file_system_sandbox_policy(); + let (sandbox_mode, writable_roots) = + sandbox_prompt_from_policy(&file_system_sandbox_policy, cwd); - Self::from_permissions_with_network( + Self::from_permissions_with_network_and_denied_reads( sandbox_mode, network_access_from_policy(permission_profile.network_sandbox_policy()), PermissionsPromptConfig { @@ -81,14 +84,32 @@ impl PermissionsInstructions { request_permissions_tool_enabled, }, writable_roots, + denied_reads_text(&file_system_sandbox_policy, cwd), ) } + #[cfg(test)] fn from_permissions_with_network( sandbox_mode: SandboxMode, network_access: NetworkAccess, config: PermissionsPromptConfig<'_>, writable_roots: Option>, + ) -> Self { + Self::from_permissions_with_network_and_denied_reads( + sandbox_mode, + network_access, + config, + writable_roots, + /*denied_reads*/ None, + ) + } + + fn from_permissions_with_network_and_denied_reads( + sandbox_mode: SandboxMode, + network_access: NetworkAccess, + config: PermissionsPromptConfig<'_>, + writable_roots: Option>, + denied_reads: Option, ) -> Self { let mut text = String::new(); append_section(&mut text, &sandbox_text(sandbox_mode, network_access)); @@ -105,6 +126,9 @@ impl PermissionsInstructions { if let Some(writable_roots) = writable_roots_text(writable_roots) { append_section(&mut text, &writable_roots); } + if let Some(denied_reads) = denied_reads { + append_section(&mut text, &denied_reads); + } if !text.ends_with('\n') { text.push('\n'); } @@ -112,27 +136,19 @@ impl PermissionsInstructions { } } -fn sandbox_prompt_from_profile( - permission_profile: &PermissionProfile, +fn sandbox_prompt_from_policy( + file_system_policy: &FileSystemSandboxPolicy, cwd: &Path, ) -> (SandboxMode, Option>) { - match permission_profile { - PermissionProfile::Disabled | PermissionProfile::External { .. } => { - (SandboxMode::DangerFullAccess, None) - } - PermissionProfile::Managed { .. } => { - let file_system_policy = permission_profile.file_system_sandbox_policy(); - if file_system_policy.has_full_disk_write_access() { - return (SandboxMode::DangerFullAccess, None); - } + if file_system_policy.has_full_disk_write_access() { + return (SandboxMode::DangerFullAccess, None); + } - let writable_roots = file_system_policy.get_writable_roots_with_cwd(cwd); - if writable_roots.is_empty() { - (SandboxMode::ReadOnly, None) - } else { - (SandboxMode::WorkspaceWrite, Some(writable_roots)) - } - } + let writable_roots = file_system_policy.get_writable_roots_with_cwd(cwd); + if writable_roots.is_empty() { + (SandboxMode::ReadOnly, None) + } else { + (SandboxMode::WorkspaceWrite, Some(writable_roots)) } } @@ -254,6 +270,28 @@ fn writable_roots_text(writable_roots: Option>) -> Option Option { + let mut entries = file_system_policy + .get_unreadable_roots_with_cwd(cwd) + .into_iter() + .map(|root| format!("- path `{}`", root.to_string_lossy())) + .collect::>(); + entries.extend( + file_system_policy + .get_unreadable_globs_with_cwd(cwd) + .into_iter() + .map(|glob| format!("- glob `{glob}`")), + ); + if entries.is_empty() { + return None; + } + + Some(format!( + "## Denied filesystem reads\nThe active permission profile denies reading these paths/globs. Do not request escalation or additional permissions to read them; these denials are policy restrictions.\n{}", + entries.join("\n") + )) +} + fn approved_command_prefixes_text(exec_policy: &Policy) -> Option { format_allow_prefixes(exec_policy.get_allowed_prefixes()) .filter(|prefixes| !prefixes.is_empty()) diff --git a/codex-rs/core/src/context/permissions_instructions_tests.rs b/codex-rs/core/src/context/permissions_instructions_tests.rs index 6d1aa5d886..466b8a35d7 100644 --- a/codex-rs/core/src/context/permissions_instructions_tests.rs +++ b/codex-rs/core/src/context/permissions_instructions_tests.rs @@ -83,6 +83,51 @@ fn builds_permissions_from_profile() { assert!(text.contains(writable_root.to_string_lossy().as_ref())); } +#[test] +fn builds_permissions_from_profile_with_denied_reads() { + let cwd = PathBuf::from("/tmp"); + let denied_root = + AbsolutePathBuf::from_absolute_path(cwd.join("blocked")).expect("absolute path"); + let permission_profile = PermissionProfile::from_runtime_permissions( + &FileSystemSandboxPolicy::restricted(vec![ + FileSystemSandboxEntry { + path: FileSystemPath::Special { + value: codex_protocol::permissions::FileSystemSpecialPath::Root, + }, + access: FileSystemAccessMode::Read, + }, + FileSystemSandboxEntry { + path: FileSystemPath::Path { + path: denied_root.clone(), + }, + access: FileSystemAccessMode::Deny, + }, + FileSystemSandboxEntry { + path: FileSystemPath::GlobPattern { + pattern: "/tmp/blocked/**".to_string(), + }, + access: FileSystemAccessMode::Deny, + }, + ]), + NetworkSandboxPolicy::Restricted, + ); + + let instructions = PermissionsInstructions::from_permission_profile( + &permission_profile, + AskForApproval::OnRequest, + ApprovalsReviewer::AutoReview, + &Policy::empty(), + &cwd, + /*exec_permission_approvals_enabled*/ false, + /*request_permissions_tool_enabled*/ false, + ); + let text = instructions.body(); + assert!(text.contains("## Denied filesystem reads")); + assert!(text.contains("Do not request escalation or additional permissions")); + assert!(text.contains(denied_root.to_string_lossy().as_ref())); + assert!(text.contains("glob `/tmp/blocked/**`")); +} + #[test] fn includes_request_rule_instructions_for_on_request() { let mut exec_policy = Policy::empty(); diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index f5c6fe5231..8cc04e5c42 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -149,6 +149,8 @@ use prompt::GuardianTranscriptEntryKind; #[cfg(test)] use prompt::build_guardian_prompt_items; #[cfg(test)] +use prompt::build_guardian_prompt_items_with_parent_turn; +#[cfg(test)] use prompt::collect_guardian_transcript_entries; #[cfg(test)] use prompt::guardian_output_schema; diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index b1b132a984..062a21deef 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -10,6 +10,7 @@ use serde_json::Value; use crate::compact::content_items_to_text; use crate::event_mapping::is_contextual_user_message_content; use crate::session::session::Session; +use crate::session::turn_context::TurnContext; use codex_utils_output_truncation::approx_bytes_for_tokens; use codex_utils_output_truncation::approx_token_count; use codex_utils_output_truncation::approx_tokens_from_byte_count; @@ -86,11 +87,29 @@ pub(crate) enum GuardianPromptMode { /// Split the variable request into separate user content items so the /// Responses request snapshot shows clear boundaries while preserving exact /// prompt text through trailing newlines. +#[cfg(test)] pub(crate) async fn build_guardian_prompt_items( session: &Session, retry_reason: Option, request: GuardianApprovalRequest, mode: GuardianPromptMode, +) -> serde_json::Result { + build_guardian_prompt_items_with_parent_turn( + session, + /*parent_turn*/ None, + retry_reason, + request, + mode, + ) + .await +} + +pub(crate) async fn build_guardian_prompt_items_with_parent_turn( + session: &Session, + parent_turn: Option<&TurnContext>, + retry_reason: Option, + request: GuardianApprovalRequest, + mode: GuardianPromptMode, ) -> serde_json::Result { let history = session.clone_history().await; let transcript_entries = collect_guardian_transcript_entries(history.raw_items()); @@ -172,6 +191,11 @@ pub(crate) async fn build_guardian_prompt_items( if let Some(note) = omission_note { push_text(format!("\n{note}\n")); } + if let Some(denied_reads_context) = parent_turn.and_then(parent_turn_denied_reads_context) { + push_text("\n>>> PARENT TURN PERMISSION CONTEXT START\n".to_string()); + push_text(denied_reads_context); + push_text(">>> PARENT TURN PERMISSION CONTEXT END\n".to_string()); + } match &request { GuardianApprovalRequest::NetworkAccess { trigger, .. } => { push_text(">>> APPROVAL REQUEST START\n".to_string()); @@ -216,6 +240,31 @@ pub(crate) async fn build_guardian_prompt_items( }) } +fn parent_turn_denied_reads_context(turn: &TurnContext) -> Option { + #[allow(deprecated)] + let cwd = &turn.cwd; + let file_system_policy = turn.permission_profile.file_system_sandbox_policy(); + let mut entries = file_system_policy + .get_unreadable_roots_with_cwd(cwd) + .into_iter() + .map(|root| format!("- path `{}`", root.to_string_lossy())) + .collect::>(); + entries.extend( + file_system_policy + .get_unreadable_globs_with_cwd(cwd) + .into_iter() + .map(|glob| format!("- glob `{glob}`")), + ); + if entries.is_empty() { + return None; + } + + Some(format!( + "The parent turn's active permission profile denies reading these paths/globs. These are policy restrictions; do not approve escalation whose purpose is to read them.\n{}\n", + entries.join("\n") + )) +} + enum GuardianPromptShape { Full, Delta { already_seen_entry_count: usize }, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index f52792cb60..fed87cd75f 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -51,7 +51,7 @@ use super::GUARDIAN_REVIEWER_NAME; use super::GuardianApprovalRequest; use super::prompt::GuardianPromptMode; use super::prompt::GuardianTranscriptCursor; -use super::prompt::build_guardian_prompt_items; +use super::prompt::build_guardian_prompt_items_with_parent_turn; use super::prompt::guardian_policy_prompt; use super::prompt::guardian_policy_prompt_with_config; @@ -686,8 +686,9 @@ async fn run_review_on_session( ) .await; - build_guardian_prompt_items( + build_guardian_prompt_items_with_parent_turn( params.parent_session.as_ref(), + Some(params.parent_turn.as_ref()), params.retry_reason.clone(), params.request.clone(), prompt_mode, diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 6df3b69f61..0bba01e869 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -33,6 +33,11 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::PermissionProfile; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::permissions::FileSystemAccessMode; +use codex_protocol::permissions::FileSystemPath; +use codex_protocol::permissions::FileSystemSandboxEntry; +use codex_protocol::permissions::FileSystemSandboxPolicy; +use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; @@ -357,6 +362,63 @@ async fn build_guardian_prompt_full_mode_preserves_initial_review_format() -> an Ok(()) } +#[tokio::test(flavor = "current_thread")] +async fn build_guardian_prompt_includes_parent_turn_denied_reads() -> anyhow::Result<()> { + let (mut session, mut turn) = crate::session::tests::make_session_and_context().await; + session.conversation_id = fixed_guardian_parent_session_id(); + let denied_root = test_path_buf("/repo/private").abs(); + turn.permission_profile = PermissionProfile::from_runtime_permissions( + &FileSystemSandboxPolicy::restricted(vec![ + FileSystemSandboxEntry { + path: FileSystemPath::Special { + value: codex_protocol::permissions::FileSystemSpecialPath::Root, + }, + access: FileSystemAccessMode::Read, + }, + FileSystemSandboxEntry { + path: FileSystemPath::Path { + path: denied_root.clone(), + }, + access: FileSystemAccessMode::Deny, + }, + FileSystemSandboxEntry { + path: FileSystemPath::GlobPattern { + pattern: "/repo/private/**".to_string(), + }, + access: FileSystemAccessMode::Deny, + }, + ]), + NetworkSandboxPolicy::Restricted, + ); + let session = Arc::new(session); + let turn = Arc::new(turn); + seed_guardian_parent_history(&session, &turn).await; + + let prompt = build_guardian_prompt_items_with_parent_turn( + session.as_ref(), + Some(turn.as_ref()), + Some("Sandbox denied reading /repo/private/secret.txt.".to_string()), + GuardianApprovalRequest::Shell { + id: "shell-1".to_string(), + command: vec!["cat".to_string(), "/repo/private/secret.txt".to_string()], + cwd: test_path_buf("/repo").abs(), + sandbox_permissions: crate::sandboxing::SandboxPermissions::RequireEscalated, + additional_permissions: None, + justification: Some("Need to inspect the secret file.".to_string()), + }, + GuardianPromptMode::Full, + ) + .await?; + + let text = guardian_prompt_text(&prompt.items); + assert!(text.contains("PARENT TURN PERMISSION CONTEXT START")); + assert!(text.contains("do not approve escalation whose purpose is to read them")); + assert!(text.contains(denied_root.to_string_lossy().as_ref())); + assert!(text.contains("glob `/repo/private/**`")); + + Ok(()) +} + #[tokio::test(flavor = "current_thread")] async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyhow::Result<()> { let (session, turn) = guardian_test_session_and_turn_with_base_url("http://localhost").await;