From 2b4cc221df31d43e63a6bc818b7529a7477313d5 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 6 Apr 2026 11:56:54 -0700 Subject: [PATCH 1/9] fix(bazel): fix simdutf (#16925) ## Summary This changes our V8 Bazel wiring so `simdutf` no longer comes from a live `git_repository` fetch against Chromium's Googlesource host. Instead, we pull `simdutf` from a pinned GitHub release archive and keep the V8 `simdutf` target wired through the external repo. The archive-backed target is set up to match the way V8 consumes `simdutf` today, including the amalgamated `src/simdutf.cpp` entrypoint and the internal files it includes. ## Why CI was intermittently failing while Bazel tried to fetch: `https://chromium.googlesource.com/chromium/src/third_party/simdutf/` That fetch was returning HTTP 429s, which then fan out into failures in the Bazel jobs, the SDK job, and the argument-comment lint jobs since they all go through the same dependency resolution path. ## What changed - replaced the `simdutf` `git_repository` in the patched V8 module deps with a pinned `http_archive` - pointed that archive at `simdutf` `v7.7.0` on GitHub - added the archive hash so the fetch is deterministic - kept the V8 BUILD patch pointing `:simdutf` at the external `@simdutf//:simdutf` target - configured the Bazel `cc_library` for the archive to use the amalgamated `src/simdutf.cpp` source plus the internal headers / textual includes it depends on ## Validation - ran `bazel build @v8//:simdutf` - confirmed the target builds successfully with the new archive-backed wiring --- patches/v8_bazel_rules.patch | 1 - patches/v8_module_deps.patch | 13 +++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/patches/v8_bazel_rules.patch b/patches/v8_bazel_rules.patch index 5457e3b854..10e1a57679 100644 --- a/patches/v8_bazel_rules.patch +++ b/patches/v8_bazel_rules.patch @@ -191,7 +191,6 @@ index 85f31b7..7314584 100644 + actual = "@simdutf//:simdutf", ) - v8_library( @@ -4593,7 +4598,7 @@ v8_library( copts = ["-Wno-implicit-fallthrough"], icu_deps = [ diff --git a/patches/v8_module_deps.patch b/patches/v8_module_deps.patch index 6c35dbacd4..8f0fed9cf4 100644 --- a/patches/v8_module_deps.patch +++ b/patches/v8_module_deps.patch @@ -6,7 +6,7 @@ diff --git a/orig/v8-14.6.202.11/MODULE.bazel b/mod/v8-14.6.202.11/MODULE.bazel --- a/orig/v8-14.6.202.11/MODULE.bazel +++ b/mod/v8-14.6.202.11/MODULE.bazel -@@ -8,7 +8,57 @@ +@@ -8,7 +8,65 @@ bazel_dep(name = "rules_python", version = "1.0.0") bazel_dep(name = "platforms", version = "1.0.0") bazel_dep(name = "abseil-cpp", version = "20250814.0") @@ -42,11 +42,12 @@ diff --git a/orig/v8-14.6.202.11/MODULE.bazel b/mod/v8-14.6.202.11/MODULE.bazel + urls = ["https://github.com/fastfloat/fast_float/archive/refs/tags/v8.0.2.tar.gz"], +) + -+git_repository( ++http_archive( + name = "simdutf", -+ build_file_content = 'load("@rules_cc//cc:defs.bzl", "cc_library")\n\ncc_library(\n name = "simdutf",\n srcs = ["simdutf.cpp"],\n hdrs = ["simdutf.h"],\n copts = ["-std=c++20"],\n include_prefix = "third_party/simdutf",\n visibility = ["//visibility:public"],\n)\n', -+ commit = "93b35aec29256f705c97f675fe4623578bd7a395", -+ remote = "https://chromium.googlesource.com/chromium/src/third_party/simdutf", ++ build_file_content = 'load("@rules_cc//cc:defs.bzl", "cc_library")\n\ncc_library(\n name = "simdutf",\n srcs = ["src/simdutf.cpp"],\n hdrs = glob(["include/**/*.h"]),\n textual_hdrs = glob(["src/**/*.h", "src/**/*.cpp"], exclude = ["src/simdutf.cpp"]),\n copts = ["-std=c++20"],\n include_prefix = "third_party/simdutf",\n includes = ["include", "src"],\n strip_include_prefix = "include",\n visibility = ["//visibility:public"],\n)\n', ++ sha256 = "0180de81a1dd48a87b8c0442ffa81734f3db91a7350914107a449935124e3c6f", ++ strip_prefix = "simdutf-7.7.0", ++ urls = ["https://github.com/simdutf/simdutf/archive/refs/tags/v7.7.0.tar.gz"], +) + +git_repository( @@ -65,7 +66,7 @@ diff --git a/orig/v8-14.6.202.11/MODULE.bazel b/mod/v8-14.6.202.11/MODULE.bazel pip = use_extension("@rules_python//python/extensions:pip.bzl", "pip") pip.parse( -@@ -22,171 +72,3 @@ +@@ -22,171 +80,3 @@ ) use_repo(pip, "v8_python_deps") From 0225479f0d213e8363656fb4bd6bba12e0d8532f Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Mon, 6 Apr 2026 12:06:58 -0700 Subject: [PATCH 2/9] bazel: Always save bazel repository cache (#16926) This should improve the cache hit ratio for external deps and such --- .github/workflows/bazel.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml index eeefcdadac..2e14184800 100644 --- a/.github/workflows/bazel.yml +++ b/.github/workflows/bazel.yml @@ -223,7 +223,7 @@ jobs: # Save bazel repository cache explicitly; make non-fatal so cache uploading # never fails the overall job. Only save when key wasn't hit. - name: Save bazel repository cache - if: always() && !cancelled() && steps.cache_bazel_repository_restore.outputs.cache-hit != 'true' + if: always() && !cancelled() continue-on-error: true uses: actions/cache/save@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5 with: From 4eabc3dcb1b1e902c9f97b97008499acb27bd5d0 Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Mon, 6 Apr 2026 12:07:19 -0700 Subject: [PATCH 3/9] bazel: Enable `--experimental_remote_downloader` (#16928) This should allow bazel to properly cache external deps. --- .bazelrc | 1 + 1 file changed, 1 insertion(+) diff --git a/.bazelrc b/.bazelrc index 603ad46e4b..3a49a22c20 100644 --- a/.bazelrc +++ b/.bazelrc @@ -44,6 +44,7 @@ common --remote_timeout=3600 common --noexperimental_throttle_remote_action_building common --experimental_remote_execution_keepalive common --grpc_keepalive_time=30s +common --experimental_remote_downloader=grpcs://remote.buildbuddy.io # This limits both in-flight executions and concurrent downloads. Even with high number # of jobs execution will still be limited by CPU cores, so this just pays a bit of From bd30bad96f92de4d6b7f2429ddfc322937bc670a Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 6 Apr 2026 12:14:27 -0700 Subject: [PATCH 4/9] fix(guardian): fix ordering of guardian events (#16462) Guardian events were emitted a bit out of order for CommandExecution items. This would make it hard for the frontend to render a guardian auto-review, which has this payload: ``` pub struct ItemGuardianApprovalReviewStartedNotification { pub thread_id: String, pub turn_id: String, pub target_item_id: String, pub review: GuardianApprovalReview, // FYI this is no longer a json blob pub action: Option, } ``` There is a `target_item_id` the auto-approval review is referring to, but the actual item had not been emitted yet. Before this PR: - `item/autoApprovalReview/started` - `item/autoApprovalReview/completed`, and if approved... - `item/started` - `item/completed` After this PR: - `item/started` - `item/autoApprovalReview/started` - `item/autoApprovalReview/completed` - `item/completed` This lines up much better with existing patterns (i.e. human review in `Default mode`, where app-server would send a server request to prompt for user approval after `item/started`), and makes it easier for clients to render what guardian is actually reviewing. We do this following a similar pattern as `FileChange` (aka apply patch) items, where we create a FileChange item and emit `item/started` if we see the apply patch approval request, before the actual apply patch call runs. --- codex-rs/Cargo.lock | 2 +- codex-rs/app-server-protocol/Cargo.toml | 2 +- codex-rs/app-server-protocol/src/lib.rs | 1 + .../src/protocol/item_builders.rs | 299 +++++++ .../app-server-protocol/src/protocol/mod.rs | 1 + .../src/protocol/thread_history.rs | 279 ++++--- .../app-server/src/bespoke_event_handling.rs | 788 +++++++++++++----- codex-rs/app-server/src/thread_state.rs | 33 + 8 files changed, 1084 insertions(+), 321 deletions(-) create mode 100644 codex-rs/app-server-protocol/src/protocol/item_builders.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index bc2d73fdcf..18e9aa434c 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1492,6 +1492,7 @@ dependencies = [ "codex-experimental-api-macros", "codex-git-utils", "codex-protocol", + "codex-shell-command", "codex-utils-absolute-path", "codex-utils-cargo-bin", "inventory", @@ -1501,7 +1502,6 @@ dependencies = [ "serde", "serde_json", "serde_with", - "shlex", "similar", "strum_macros 0.28.0", "tempfile", diff --git a/codex-rs/app-server-protocol/Cargo.toml b/codex-rs/app-server-protocol/Cargo.toml index a9d90831ef..d9ed5e8730 100644 --- a/codex-rs/app-server-protocol/Cargo.toml +++ b/codex-rs/app-server-protocol/Cargo.toml @@ -17,12 +17,12 @@ clap = { workspace = true, features = ["derive"] } codex-experimental-api-macros = { workspace = true } codex-git-utils = { workspace = true } codex-protocol = { workspace = true } +codex-shell-command = { workspace = true } codex-utils-absolute-path = { workspace = true } schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } serde_with = { workspace = true } -shlex = { workspace = true } strum_macros = { workspace = true } thiserror = { workspace = true } rmcp = { workspace = true, default-features = false, features = [ diff --git a/codex-rs/app-server-protocol/src/lib.rs b/codex-rs/app-server-protocol/src/lib.rs index b1a8f474f3..d5c2f4b243 100644 --- a/codex-rs/app-server-protocol/src/lib.rs +++ b/codex-rs/app-server-protocol/src/lib.rs @@ -15,6 +15,7 @@ pub use export::generate_ts_with_options; pub use export::generate_types; pub use jsonrpc_lite::*; pub use protocol::common::*; +pub use protocol::item_builders::*; pub use protocol::thread_history::*; pub use protocol::v1::ApplyPatchApprovalParams; pub use protocol::v1::ApplyPatchApprovalResponse; diff --git a/codex-rs/app-server-protocol/src/protocol/item_builders.rs b/codex-rs/app-server-protocol/src/protocol/item_builders.rs new file mode 100644 index 0000000000..804169db6f --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/item_builders.rs @@ -0,0 +1,299 @@ +//! Shared builders for synthetic [`ThreadItem`] values emitted by the app-server layer. +//! +//! These items do not come from first-class core `ItemStarted` / `ItemCompleted` events. +//! Instead, the app-server synthesizes them so clients can render a coherent lifecycle for +//! approvals and other pre-execution flows before the underlying tool has started or when the +//! tool never starts at all. +//! +//! Keeping these builders in one place is useful for two reasons: +//! - Live notifications and rebuilt `thread/read` history both need to construct the same +//! synthetic items, so sharing the logic avoids drift between those paths. +//! - The projection is presentation-specific. Core protocol events stay generic, while the +//! app-server protocol decides how to surface those events as `ThreadItem`s for clients. +use crate::protocol::common::ServerNotification; +use crate::protocol::v2::CommandAction; +use crate::protocol::v2::CommandExecutionSource; +use crate::protocol::v2::CommandExecutionStatus; +use crate::protocol::v2::FileUpdateChange; +use crate::protocol::v2::GuardianApprovalReview; +use crate::protocol::v2::GuardianApprovalReviewStatus; +use crate::protocol::v2::ItemGuardianApprovalReviewCompletedNotification; +use crate::protocol::v2::ItemGuardianApprovalReviewStartedNotification; +use crate::protocol::v2::PatchApplyStatus; +use crate::protocol::v2::PatchChangeKind; +use crate::protocol::v2::ThreadItem; +use codex_protocol::ThreadId; +use codex_protocol::protocol::ApplyPatchApprovalRequestEvent; +use codex_protocol::protocol::ExecApprovalRequestEvent; +use codex_protocol::protocol::ExecCommandBeginEvent; +use codex_protocol::protocol::ExecCommandEndEvent; +use codex_protocol::protocol::FileChange; +use codex_protocol::protocol::GuardianAssessmentAction; +use codex_protocol::protocol::GuardianAssessmentEvent; +use codex_protocol::protocol::PatchApplyBeginEvent; +use codex_protocol::protocol::PatchApplyEndEvent; +use codex_shell_command::parse_command::parse_command; +use codex_shell_command::parse_command::shlex_join; +use std::collections::HashMap; +use std::path::PathBuf; + +pub fn build_file_change_approval_request_item( + payload: &ApplyPatchApprovalRequestEvent, +) -> ThreadItem { + ThreadItem::FileChange { + id: payload.call_id.clone(), + changes: convert_patch_changes(&payload.changes), + status: PatchApplyStatus::InProgress, + } +} + +pub fn build_file_change_begin_item(payload: &PatchApplyBeginEvent) -> ThreadItem { + ThreadItem::FileChange { + id: payload.call_id.clone(), + changes: convert_patch_changes(&payload.changes), + status: PatchApplyStatus::InProgress, + } +} + +pub fn build_file_change_end_item(payload: &PatchApplyEndEvent) -> ThreadItem { + ThreadItem::FileChange { + id: payload.call_id.clone(), + changes: convert_patch_changes(&payload.changes), + status: (&payload.status).into(), + } +} + +pub fn build_command_execution_approval_request_item( + payload: &ExecApprovalRequestEvent, +) -> ThreadItem { + ThreadItem::CommandExecution { + id: payload.call_id.clone(), + command: shlex_join(&payload.command), + cwd: payload.cwd.clone(), + process_id: None, + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::InProgress, + command_actions: payload + .parsed_cmd + .iter() + .cloned() + .map(CommandAction::from) + .collect(), + aggregated_output: None, + exit_code: None, + duration_ms: None, + } +} + +pub fn build_command_execution_begin_item(payload: &ExecCommandBeginEvent) -> ThreadItem { + ThreadItem::CommandExecution { + id: payload.call_id.clone(), + command: shlex_join(&payload.command), + cwd: payload.cwd.clone(), + process_id: payload.process_id.clone(), + source: payload.source.into(), + status: CommandExecutionStatus::InProgress, + command_actions: payload + .parsed_cmd + .iter() + .cloned() + .map(CommandAction::from) + .collect(), + aggregated_output: None, + exit_code: None, + duration_ms: None, + } +} + +pub fn build_command_execution_end_item(payload: &ExecCommandEndEvent) -> ThreadItem { + let aggregated_output = if payload.aggregated_output.is_empty() { + None + } else { + Some(payload.aggregated_output.clone()) + }; + let duration_ms = i64::try_from(payload.duration.as_millis()).unwrap_or(i64::MAX); + + ThreadItem::CommandExecution { + id: payload.call_id.clone(), + command: shlex_join(&payload.command), + cwd: payload.cwd.clone(), + process_id: payload.process_id.clone(), + source: payload.source.into(), + status: (&payload.status).into(), + command_actions: payload + .parsed_cmd + .iter() + .cloned() + .map(CommandAction::from) + .collect(), + aggregated_output, + exit_code: Some(payload.exit_code), + duration_ms: Some(duration_ms), + } +} + +/// Build a guardian-derived [`ThreadItem`]. +/// +/// Currently this only synthesizes [`ThreadItem::CommandExecution`] for +/// [`GuardianAssessmentAction::Command`] and [`GuardianAssessmentAction::Execve`]. +pub fn build_item_from_guardian_event( + assessment: &GuardianAssessmentEvent, + status: CommandExecutionStatus, +) -> Option { + match &assessment.action { + GuardianAssessmentAction::Command { command, cwd, .. } => { + let command = command.clone(); + let command_actions = vec![CommandAction::Unknown { + command: command.clone(), + }]; + Some(ThreadItem::CommandExecution { + id: assessment.id.clone(), + command, + cwd: cwd.clone(), + process_id: None, + source: CommandExecutionSource::Agent, + status, + command_actions, + aggregated_output: None, + exit_code: None, + duration_ms: None, + }) + } + GuardianAssessmentAction::Execve { + program, argv, cwd, .. + } => { + let argv = if argv.is_empty() { + vec![program.clone()] + } else { + std::iter::once(program.clone()) + .chain(argv.iter().skip(1).cloned()) + .collect::>() + }; + let command = shlex_join(&argv); + let parsed_cmd = parse_command(&argv); + let command_actions = if parsed_cmd.is_empty() { + vec![CommandAction::Unknown { + command: command.clone(), + }] + } else { + parsed_cmd.into_iter().map(CommandAction::from).collect() + }; + Some(ThreadItem::CommandExecution { + id: assessment.id.clone(), + command, + cwd: cwd.clone(), + process_id: None, + source: CommandExecutionSource::Agent, + status, + command_actions, + aggregated_output: None, + exit_code: None, + duration_ms: None, + }) + } + GuardianAssessmentAction::ApplyPatch { .. } + | GuardianAssessmentAction::NetworkAccess { .. } + | GuardianAssessmentAction::McpToolCall { .. } => None, + } +} + +pub fn guardian_auto_approval_review_notification( + conversation_id: &ThreadId, + event_turn_id: &str, + assessment: &GuardianAssessmentEvent, +) -> ServerNotification { + // TODO(ccunningham): Attach guardian review state to the reviewed tool + // item's lifecycle instead of sending standalone review notifications so + // the app-server API can persist and replay review state via `thread/read`. + let turn_id = if assessment.turn_id.is_empty() { + event_turn_id.to_string() + } else { + assessment.turn_id.clone() + }; + let review = GuardianApprovalReview { + status: match assessment.status { + codex_protocol::protocol::GuardianAssessmentStatus::InProgress => { + GuardianApprovalReviewStatus::InProgress + } + codex_protocol::protocol::GuardianAssessmentStatus::Approved => { + GuardianApprovalReviewStatus::Approved + } + codex_protocol::protocol::GuardianAssessmentStatus::Denied => { + GuardianApprovalReviewStatus::Denied + } + codex_protocol::protocol::GuardianAssessmentStatus::Aborted => { + GuardianApprovalReviewStatus::Aborted + } + }, + risk_score: assessment.risk_score, + risk_level: assessment.risk_level.map(Into::into), + rationale: assessment.rationale.clone(), + }; + let action = assessment.action.clone().into(); + match assessment.status { + codex_protocol::protocol::GuardianAssessmentStatus::InProgress => { + ServerNotification::ItemGuardianApprovalReviewStarted( + ItemGuardianApprovalReviewStartedNotification { + thread_id: conversation_id.to_string(), + turn_id, + target_item_id: assessment.id.clone(), + review, + action, + }, + ) + } + codex_protocol::protocol::GuardianAssessmentStatus::Approved + | codex_protocol::protocol::GuardianAssessmentStatus::Denied + | codex_protocol::protocol::GuardianAssessmentStatus::Aborted => { + ServerNotification::ItemGuardianApprovalReviewCompleted( + ItemGuardianApprovalReviewCompletedNotification { + thread_id: conversation_id.to_string(), + turn_id, + target_item_id: assessment.id.clone(), + review, + action, + }, + ) + } + } +} + +pub fn convert_patch_changes(changes: &HashMap) -> Vec { + let mut converted: Vec = changes + .iter() + .map(|(path, change)| FileUpdateChange { + path: path.to_string_lossy().into_owned(), + kind: map_patch_change_kind(change), + diff: format_file_change_diff(change), + }) + .collect(); + converted.sort_by(|a, b| a.path.cmp(&b.path)); + converted +} + +fn map_patch_change_kind(change: &FileChange) -> PatchChangeKind { + match change { + FileChange::Add { .. } => PatchChangeKind::Add, + FileChange::Delete { .. } => PatchChangeKind::Delete, + FileChange::Update { move_path, .. } => PatchChangeKind::Update { + move_path: move_path.clone(), + }, + } +} + +fn format_file_change_diff(change: &FileChange) -> String { + match change { + FileChange::Add { content } => content.clone(), + FileChange::Delete { content } => content.clone(), + FileChange::Update { + unified_diff, + move_path, + } => { + if let Some(path) = move_path { + format!("{unified_diff}\n\nMoved to: {}", path.display()) + } else { + unified_diff.clone() + } + } + } +} diff --git a/codex-rs/app-server-protocol/src/protocol/mod.rs b/codex-rs/app-server-protocol/src/protocol/mod.rs index 1e0410f4b9..4179d361c7 100644 --- a/codex-rs/app-server-protocol/src/protocol/mod.rs +++ b/codex-rs/app-server-protocol/src/protocol/mod.rs @@ -2,6 +2,7 @@ // Exposes protocol pieces used by `lib.rs` via `pub use protocol::common::*;`. pub mod common; +pub mod item_builders; mod mappers; mod serde_helpers; pub mod thread_history; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 48fa56d687..99a8f6e626 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -1,16 +1,18 @@ +use crate::protocol::item_builders::build_command_execution_begin_item; +use crate::protocol::item_builders::build_command_execution_end_item; +use crate::protocol::item_builders::build_file_change_approval_request_item; +use crate::protocol::item_builders::build_file_change_begin_item; +use crate::protocol::item_builders::build_file_change_end_item; +use crate::protocol::item_builders::build_item_from_guardian_event; use crate::protocol::v2::CollabAgentState; use crate::protocol::v2::CollabAgentTool; use crate::protocol::v2::CollabAgentToolCallStatus; -use crate::protocol::v2::CommandAction; use crate::protocol::v2::CommandExecutionStatus; use crate::protocol::v2::DynamicToolCallOutputContentItem; use crate::protocol::v2::DynamicToolCallStatus; -use crate::protocol::v2::FileUpdateChange; use crate::protocol::v2::McpToolCallError; use crate::protocol::v2::McpToolCallResult; use crate::protocol::v2::McpToolCallStatus; -use crate::protocol::v2::PatchApplyStatus; -use crate::protocol::v2::PatchChangeKind; use crate::protocol::v2::ThreadItem; use crate::protocol::v2::Turn; use crate::protocol::v2::TurnError as V2TurnError; @@ -31,6 +33,8 @@ use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandBeginEvent; use codex_protocol::protocol::ExecCommandEndEvent; +use codex_protocol::protocol::GuardianAssessmentEvent; +use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::ImageGenerationBeginEvent; use codex_protocol::protocol::ImageGenerationEndEvent; use codex_protocol::protocol::ItemCompletedEvent; @@ -53,6 +57,14 @@ use std::collections::HashMap; use tracing::warn; use uuid::Uuid; +#[cfg(test)] +use crate::protocol::v2::CommandAction; +#[cfg(test)] +use crate::protocol::v2::FileUpdateChange; +#[cfg(test)] +use crate::protocol::v2::PatchApplyStatus; +#[cfg(test)] +use crate::protocol::v2::PatchChangeKind; #[cfg(test)] use codex_protocol::protocol::ExecCommandStatus as CoreExecCommandStatus; #[cfg(test)] @@ -149,6 +161,7 @@ impl ThreadHistoryBuilder { EventMsg::WebSearchEnd(payload) => self.handle_web_search_end(payload), EventMsg::ExecCommandBegin(payload) => self.handle_exec_command_begin(payload), EventMsg::ExecCommandEnd(payload) => self.handle_exec_command_end(payload), + EventMsg::GuardianAssessment(payload) => self.handle_guardian_assessment(payload), EventMsg::ApplyPatchApprovalRequest(payload) => { self.handle_apply_patch_approval_request(payload) } @@ -375,57 +388,12 @@ impl ThreadHistoryBuilder { } fn handle_exec_command_begin(&mut self, payload: &ExecCommandBeginEvent) { - let command = shlex::try_join(payload.command.iter().map(String::as_str)) - .unwrap_or_else(|_| payload.command.join(" ")); - let command_actions = payload - .parsed_cmd - .iter() - .cloned() - .map(CommandAction::from) - .collect(); - let item = ThreadItem::CommandExecution { - id: payload.call_id.clone(), - command, - cwd: payload.cwd.clone(), - process_id: payload.process_id.clone(), - source: payload.source.into(), - status: CommandExecutionStatus::InProgress, - command_actions, - aggregated_output: None, - exit_code: None, - duration_ms: None, - }; + let item = build_command_execution_begin_item(payload); self.upsert_item_in_turn_id(&payload.turn_id, item); } fn handle_exec_command_end(&mut self, payload: &ExecCommandEndEvent) { - let status: CommandExecutionStatus = (&payload.status).into(); - let duration_ms = i64::try_from(payload.duration.as_millis()).unwrap_or(i64::MAX); - let aggregated_output = if payload.aggregated_output.is_empty() { - None - } else { - Some(payload.aggregated_output.clone()) - }; - let command = shlex::try_join(payload.command.iter().map(String::as_str)) - .unwrap_or_else(|_| payload.command.join(" ")); - let command_actions = payload - .parsed_cmd - .iter() - .cloned() - .map(CommandAction::from) - .collect(); - let item = ThreadItem::CommandExecution { - id: payload.call_id.clone(), - command, - cwd: payload.cwd.clone(), - process_id: payload.process_id.clone(), - source: payload.source.into(), - status, - command_actions, - aggregated_output, - exit_code: Some(payload.exit_code), - duration_ms: Some(duration_ms), - }; + let item = build_command_execution_end_item(payload); // Command completions can arrive out of order. Unified exec may return // while a PTY is still running, then emit ExecCommandEnd later from a // background exit watcher when that process finally exits. By then, a @@ -434,12 +402,26 @@ impl ThreadHistoryBuilder { self.upsert_item_in_turn_id(&payload.turn_id, item); } - fn handle_apply_patch_approval_request(&mut self, payload: &ApplyPatchApprovalRequestEvent) { - let item = ThreadItem::FileChange { - id: payload.call_id.clone(), - changes: convert_patch_changes(&payload.changes), - status: PatchApplyStatus::InProgress, + fn handle_guardian_assessment(&mut self, payload: &GuardianAssessmentEvent) { + let status = match payload.status { + GuardianAssessmentStatus::InProgress => CommandExecutionStatus::InProgress, + GuardianAssessmentStatus::Denied | GuardianAssessmentStatus::Aborted => { + CommandExecutionStatus::Declined + } + GuardianAssessmentStatus::Approved => return, }; + let Some(item) = build_item_from_guardian_event(payload, status) else { + return; + }; + if payload.turn_id.is_empty() { + self.upsert_item_in_current_turn(item); + } else { + self.upsert_item_in_turn_id(&payload.turn_id, item); + } + } + + fn handle_apply_patch_approval_request(&mut self, payload: &ApplyPatchApprovalRequestEvent) { + let item = build_file_change_approval_request_item(payload); if payload.turn_id.is_empty() { self.upsert_item_in_current_turn(item); } else { @@ -448,11 +430,7 @@ impl ThreadHistoryBuilder { } fn handle_patch_apply_begin(&mut self, payload: &PatchApplyBeginEvent) { - let item = ThreadItem::FileChange { - id: payload.call_id.clone(), - changes: convert_patch_changes(&payload.changes), - status: PatchApplyStatus::InProgress, - }; + let item = build_file_change_begin_item(payload); if payload.turn_id.is_empty() { self.upsert_item_in_current_turn(item); } else { @@ -461,12 +439,7 @@ impl ThreadHistoryBuilder { } fn handle_patch_apply_end(&mut self, payload: &PatchApplyEndEvent) { - let status: PatchApplyStatus = (&payload.status).into(); - let item = ThreadItem::FileChange { - id: payload.call_id.clone(), - changes: convert_patch_changes(&payload.changes), - status, - }; + let item = build_file_change_end_item(payload); if payload.turn_id.is_empty() { self.upsert_item_in_current_turn(item); } else { @@ -1076,21 +1049,6 @@ fn render_review_output_text(output: &ReviewOutputEvent) -> String { } } -pub fn convert_patch_changes( - changes: &HashMap, -) -> Vec { - let mut converted: Vec = changes - .iter() - .map(|(path, change)| FileUpdateChange { - path: path.to_string_lossy().into_owned(), - kind: map_patch_change_kind(change), - diff: format_file_change_diff(change), - }) - .collect(); - converted.sort_by(|a, b| a.path.cmp(&b.path)); - converted -} - fn convert_dynamic_tool_content_items( items: &[codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem], ) -> Vec { @@ -1108,33 +1066,6 @@ fn convert_dynamic_tool_content_items( .collect() } -fn map_patch_change_kind(change: &codex_protocol::protocol::FileChange) -> PatchChangeKind { - match change { - codex_protocol::protocol::FileChange::Add { .. } => PatchChangeKind::Add, - codex_protocol::protocol::FileChange::Delete { .. } => PatchChangeKind::Delete, - codex_protocol::protocol::FileChange::Update { move_path, .. } => PatchChangeKind::Update { - move_path: move_path.clone(), - }, - } -} - -fn format_file_change_diff(change: &codex_protocol::protocol::FileChange) -> String { - match change { - codex_protocol::protocol::FileChange::Add { content } => content.clone(), - codex_protocol::protocol::FileChange::Delete { content } => content.clone(), - codex_protocol::protocol::FileChange::Update { - unified_diff, - move_path, - } => { - if let Some(path) = move_path { - format!("{unified_diff}\n\nMoved to: {}", path.display()) - } else { - unified_diff.clone() - } - } - } -} - fn upsert_turn_item(items: &mut Vec, item: ThreadItem) { if let Some(existing_item) = items .iter_mut() @@ -2030,6 +1961,136 @@ mod tests { ); } + #[test] + fn reconstructs_declined_guardian_command_item() { + let events = vec![ + EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-1".into(), + model_context_window: None, + collaboration_mode_kind: Default::default(), + }), + EventMsg::UserMessage(UserMessageEvent { + message: "review this command".into(), + images: None, + text_elements: Vec::new(), + local_images: Vec::new(), + }), + EventMsg::GuardianAssessment(GuardianAssessmentEvent { + id: "guardian-exec".into(), + turn_id: "turn-1".into(), + status: GuardianAssessmentStatus::InProgress, + risk_score: None, + risk_level: None, + rationale: None, + action: serde_json::from_value(serde_json::json!({ + "type": "command", + "source": "shell", + "command": "rm -rf /tmp/guardian", + "cwd": "/tmp", + })) + .expect("guardian action"), + }), + EventMsg::GuardianAssessment(GuardianAssessmentEvent { + id: "guardian-exec".into(), + turn_id: "turn-1".into(), + status: GuardianAssessmentStatus::Denied, + risk_score: Some(97), + risk_level: Some(codex_protocol::protocol::GuardianRiskLevel::High), + rationale: Some("Would delete user data.".into()), + action: serde_json::from_value(serde_json::json!({ + "type": "command", + "source": "shell", + "command": "rm -rf /tmp/guardian", + "cwd": "/tmp", + })) + .expect("guardian action"), + }), + ]; + + let items = events + .into_iter() + .map(RolloutItem::EventMsg) + .collect::>(); + let turns = build_turns_from_rollout_items(&items); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].items.len(), 2); + assert_eq!( + turns[0].items[1], + ThreadItem::CommandExecution { + id: "guardian-exec".into(), + command: "rm -rf /tmp/guardian".into(), + cwd: PathBuf::from("/tmp"), + process_id: None, + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::Declined, + command_actions: vec![CommandAction::Unknown { + command: "rm -rf /tmp/guardian".into(), + }], + aggregated_output: None, + exit_code: None, + duration_ms: None, + } + ); + } + + #[test] + fn reconstructs_in_progress_guardian_execve_item() { + let events = vec![ + EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-1".into(), + model_context_window: None, + collaboration_mode_kind: Default::default(), + }), + EventMsg::UserMessage(UserMessageEvent { + message: "run a subcommand".into(), + images: None, + text_elements: Vec::new(), + local_images: Vec::new(), + }), + EventMsg::GuardianAssessment(GuardianAssessmentEvent { + id: "guardian-execve".into(), + turn_id: "turn-1".into(), + status: GuardianAssessmentStatus::InProgress, + risk_score: None, + risk_level: None, + rationale: None, + action: serde_json::from_value(serde_json::json!({ + "type": "execve", + "source": "shell", + "program": "/bin/rm", + "argv": ["/usr/bin/rm", "-f", "/tmp/file.sqlite"], + "cwd": "/tmp", + })) + .expect("guardian action"), + }), + ]; + + let items = events + .into_iter() + .map(RolloutItem::EventMsg) + .collect::>(); + let turns = build_turns_from_rollout_items(&items); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].items.len(), 2); + assert_eq!( + turns[0].items[1], + ThreadItem::CommandExecution { + id: "guardian-execve".into(), + command: "/bin/rm -f /tmp/file.sqlite".into(), + cwd: PathBuf::from("/tmp"), + process_id: None, + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::InProgress, + command_actions: vec![CommandAction::Unknown { + command: "/bin/rm -f /tmp/file.sqlite".into(), + }], + aggregated_output: None, + exit_code: None, + duration_ms: None, + } + ); + } + #[test] fn assigns_late_exec_completion_to_original_turn() { let events = vec![ diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 78c60b0aee..01eb4bd578 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -7,9 +7,9 @@ use crate::error_code::INVALID_REQUEST_ERROR_CODE; use crate::outgoing_message::ClientRequestResult; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::server_request_error::is_turn_transition_server_request_error; -use crate::thread_state::ThreadListenerCommand; use crate::thread_state::ThreadState; use crate::thread_state::TurnSummary; +use crate::thread_state::resolve_server_request_on_thread_listener; use crate::thread_status::ThreadWatchActiveGuard; use crate::thread_status::ThreadWatchManager; use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; @@ -43,14 +43,10 @@ use codex_app_server_protocol::FileChangeRequestApprovalParams; use codex_app_server_protocol::FileChangeRequestApprovalResponse; use codex_app_server_protocol::FileUpdateChange; use codex_app_server_protocol::GrantedPermissionProfile as V2GrantedPermissionProfile; -use codex_app_server_protocol::GuardianApprovalReview; -use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::HookCompletedNotification; use codex_app_server_protocol::HookStartedNotification; use codex_app_server_protocol::InterruptConversationResponse; use codex_app_server_protocol::ItemCompletedNotification; -use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification; -use codex_app_server_protocol::ItemGuardianApprovalReviewStartedNotification; use codex_app_server_protocol::ItemStartedNotification; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::McpServerElicitationAction; @@ -102,8 +98,14 @@ use codex_app_server_protocol::TurnPlanStep; use codex_app_server_protocol::TurnPlanUpdatedNotification; use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::build_command_execution_end_item; +use codex_app_server_protocol::build_file_change_approval_request_item; +use codex_app_server_protocol::build_file_change_begin_item; +use codex_app_server_protocol::build_file_change_end_item; +use codex_app_server_protocol::build_item_from_guardian_event; use codex_app_server_protocol::build_turns_from_rollout_items; use codex_app_server_protocol::convert_patch_changes; +use codex_app_server_protocol::guardian_auto_approval_review_notification; use codex_core::CodexThread; use codex_core::ThreadManager; use codex_core::find_thread_name_by_id; @@ -114,13 +116,10 @@ use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynam use codex_protocol::dynamic_tools::DynamicToolResponse as CoreDynamicToolResponse; use codex_protocol::items::parse_hook_prompt_message; use codex_protocol::plan_tool::UpdatePlanArgs; -use codex_protocol::protocol::ApplyPatchApprovalRequestEvent; use codex_protocol::protocol::CodexErrorInfo as CoreCodexErrorInfo; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecApprovalRequestEvent; -use codex_protocol::protocol::ExecCommandEndEvent; -use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::McpToolCallBeginEvent; use codex_protocol::protocol::McpToolCallEndEvent; use codex_protocol::protocol::Op; @@ -137,7 +136,6 @@ use codex_protocol::request_user_input::RequestUserInputResponse as CoreRequestU use codex_sandboxing::policy_transforms::intersect_permission_profiles; use codex_shell_command::parse_command::shlex_join; use std::collections::HashMap; -use std::convert::TryFrom; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -153,105 +151,13 @@ enum CommandExecutionApprovalPresentation { Command(CommandExecutionCompletionItem), } +#[derive(Debug, PartialEq)] struct CommandExecutionCompletionItem { command: String, cwd: PathBuf, command_actions: Vec, } -async fn resolve_server_request_on_thread_listener( - thread_state: &Arc>, - request_id: RequestId, -) { - let (completion_tx, completion_rx) = oneshot::channel(); - let listener_command_tx = { - let state = thread_state.lock().await; - state.listener_command_tx() - }; - let Some(listener_command_tx) = listener_command_tx else { - error!("failed to remove pending client request: thread listener is not running"); - return; - }; - - if listener_command_tx - .send(ThreadListenerCommand::ResolveServerRequest { - request_id, - completion_tx, - }) - .is_err() - { - error!( - "failed to remove pending client request: thread listener command channel is closed" - ); - return; - } - - if let Err(err) = completion_rx.await { - error!("failed to remove pending client request: {err}"); - } -} - -fn guardian_auto_approval_review_notification( - conversation_id: &ThreadId, - event_turn_id: &str, - assessment: &GuardianAssessmentEvent, -) -> ServerNotification { - // TODO(ccunningham): Attach guardian review state to the reviewed tool - // item's lifecycle instead of sending standalone review notifications so - // the app-server API can persist and replay review state via `thread/read`. - let turn_id = if assessment.turn_id.is_empty() { - event_turn_id.to_string() - } else { - assessment.turn_id.clone() - }; - let review = GuardianApprovalReview { - status: match assessment.status { - codex_protocol::protocol::GuardianAssessmentStatus::InProgress => { - GuardianApprovalReviewStatus::InProgress - } - codex_protocol::protocol::GuardianAssessmentStatus::Approved => { - GuardianApprovalReviewStatus::Approved - } - codex_protocol::protocol::GuardianAssessmentStatus::Denied => { - GuardianApprovalReviewStatus::Denied - } - codex_protocol::protocol::GuardianAssessmentStatus::Aborted => { - GuardianApprovalReviewStatus::Aborted - } - }, - risk_score: assessment.risk_score, - risk_level: assessment.risk_level.map(Into::into), - rationale: assessment.rationale.clone(), - }; - let action = assessment.action.clone().into(); - match assessment.status { - codex_protocol::protocol::GuardianAssessmentStatus::InProgress => { - ServerNotification::ItemGuardianApprovalReviewStarted( - ItemGuardianApprovalReviewStartedNotification { - thread_id: conversation_id.to_string(), - turn_id, - target_item_id: assessment.id.clone(), - review, - action, - }, - ) - } - codex_protocol::protocol::GuardianAssessmentStatus::Approved - | codex_protocol::protocol::GuardianAssessmentStatus::Denied - | codex_protocol::protocol::GuardianAssessmentStatus::Aborted => { - ServerNotification::ItemGuardianApprovalReviewCompleted( - ItemGuardianApprovalReviewCompletedNotification { - thread_id: conversation_id.to_string(), - turn_id, - target_item_id: assessment.id.clone(), - review, - action, - }, - ) - } - } -} - #[allow(clippy::too_many_arguments)] pub(crate) async fn apply_bespoke_event_handling( event: Event, @@ -344,12 +250,71 @@ pub(crate) async fn apply_bespoke_event_handling( EventMsg::Warning(_warning_event) => {} EventMsg::GuardianAssessment(assessment) => { if let ApiVersion::V2 = api_version { + let pending_command_execution = match build_item_from_guardian_event( + &assessment, + CommandExecutionStatus::InProgress, + ) { + Some(ThreadItem::CommandExecution { + command, + cwd, + command_actions, + .. + }) => Some(CommandExecutionCompletionItem { + command, + cwd, + command_actions, + }), + Some(_) | None => None, + }; + let assessment_turn_id = if assessment.turn_id.is_empty() { + event_turn_id.clone() + } else { + assessment.turn_id.clone() + }; + if assessment.status + == codex_protocol::protocol::GuardianAssessmentStatus::InProgress + && let Some(completion_item) = pending_command_execution.as_ref() + { + start_command_execution_item( + &conversation_id, + assessment_turn_id.clone(), + assessment.id.clone(), + completion_item.command.clone(), + completion_item.cwd.clone(), + completion_item.command_actions.clone(), + CommandExecutionSource::Agent, + &outgoing, + &thread_state, + ) + .await; + } let notification = guardian_auto_approval_review_notification( &conversation_id, &event_turn_id, &assessment, ); outgoing.send_server_notification(notification).await; + if matches!( + assessment.status, + codex_protocol::protocol::GuardianAssessmentStatus::Denied + | codex_protocol::protocol::GuardianAssessmentStatus::Aborted + ) && let Some(completion_item) = pending_command_execution + { + complete_command_execution_item( + &conversation_id, + assessment_turn_id, + assessment.id.clone(), + completion_item.command, + completion_item.cwd, + /*process_id*/ None, + CommandExecutionSource::Agent, + completion_item.command_actions, + CommandExecutionStatus::Declined, + &outgoing, + &thread_state, + ) + .await; + } } } EventMsg::ModelReroute(event) => { @@ -503,13 +468,7 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } } - EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { - call_id, - turn_id, - changes, - reason, - grant_root, - }) => { + EventMsg::ApplyPatchApprovalRequest(event) => { let permission_guard = thread_watch_manager .note_permission_requested(&conversation_id.to_string()) .await; @@ -517,14 +476,15 @@ pub(crate) async fn apply_bespoke_event_handling( ApiVersion::V1 => { let params = ApplyPatchApprovalParams { conversation_id, - call_id: call_id.clone(), - file_changes: changes.clone(), - reason, - grant_root, + call_id: event.call_id.clone(), + file_changes: event.changes.clone(), + reason: event.reason.clone(), + grant_root: event.grant_root.clone(), }; let (_pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::ApplyPatchApproval(params)) .await; + let call_id = event.call_id.clone(); tokio::spawn(async move { let _permission_guard = permission_guard; on_patch_approval_response(call_id, rx, conversation).await; @@ -533,9 +493,8 @@ pub(crate) async fn apply_bespoke_event_handling( ApiVersion::V2 => { // Until we migrate the core to be aware of a first class FileChangeItem // and emit the corresponding EventMsg, we repurpose the call_id as the item_id. - let item_id = call_id.clone(); - let patch_changes = convert_patch_changes(&changes); - + let item_id = event.call_id.clone(); + let patch_changes = convert_patch_changes(&event.changes); let first_start = { let mut state = thread_state.lock().await; state @@ -544,11 +503,7 @@ pub(crate) async fn apply_bespoke_event_handling( .insert(item_id.clone()) }; if first_start { - let item = ThreadItem::FileChange { - id: item_id.clone(), - changes: patch_changes.clone(), - status: PatchApplyStatus::InProgress, - }; + let item = build_file_change_approval_request_item(&event); let notification = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), @@ -561,10 +516,10 @@ pub(crate) async fn apply_bespoke_event_handling( let params = FileChangeRequestApprovalParams { thread_id: conversation_id.to_string(), - turn_id: turn_id.clone(), + turn_id: event.turn_id.clone(), item_id: item_id.clone(), - reason, - grant_root, + reason: event.reason.clone(), + grant_root: event.grant_root.clone(), }; let (pending_request_id, rx) = outgoing .send_request(ServerRequestPayload::FileChangeRequestApproval(params)) @@ -668,6 +623,22 @@ pub(crate) async fn apply_bespoke_event_handling( Some(completion_item), ), }; + if approval_id.is_none() + && let Some(completion_item) = completion_item.as_ref() + { + start_command_execution_item( + &conversation_id, + event_turn_id.clone(), + call_id.clone(), + completion_item.command.clone(), + completion_item.cwd.clone(), + completion_item.command_actions.clone(), + CommandExecutionSource::Agent, + &outgoing, + &thread_state, + ) + .await; + } let proposed_execpolicy_amendment_v2 = proposed_execpolicy_amendment.map(V2ExecPolicyAmendment::from); let proposed_network_policy_amendments_v2 = proposed_network_policy_amendments @@ -1555,7 +1526,6 @@ pub(crate) async fn apply_bespoke_event_handling( // Until we migrate the core to be aware of a first class FileChangeItem // and emit the corresponding EventMsg, we repurpose the call_id as the item_id. let item_id = patch_begin_event.call_id.clone(); - let changes = convert_patch_changes(&patch_begin_event.changes); let first_start = { let mut state = thread_state.lock().await; @@ -1565,11 +1535,7 @@ pub(crate) async fn apply_bespoke_event_handling( .insert(item_id.clone()) }; if first_start { - let item = ThreadItem::FileChange { - id: item_id.clone(), - changes, - status: PatchApplyStatus::InProgress, - }; + let item = build_file_change_begin_item(&patch_begin_event); let notification = ItemStartedNotification { thread_id: conversation_id.to_string(), turn_id: event_turn_id.clone(), @@ -1584,14 +1550,10 @@ pub(crate) async fn apply_bespoke_event_handling( // Until we migrate the core to be aware of a first class FileChangeItem // and emit the corresponding EventMsg, we repurpose the call_id as the item_id. let item_id = patch_end_event.call_id.clone(); - - let status: PatchApplyStatus = (&patch_end_event.status).into(); - let changes = convert_patch_changes(&patch_end_event.changes); complete_file_change_item( conversation_id, item_id, - changes, - status, + build_file_change_end_item(&patch_end_event), event_turn_id.clone(), &outgoing, &thread_state, @@ -1608,35 +1570,35 @@ pub(crate) async fn apply_bespoke_event_handling( let command = shlex_join(&exec_command_begin_event.command); let cwd = exec_command_begin_event.cwd; let process_id = exec_command_begin_event.process_id; - - { + let first_start = { let mut state = thread_state.lock().await; state .turn_summary .command_execution_started - .insert(item_id.clone()); + .insert(item_id.clone()) + }; + if first_start { + let item = ThreadItem::CommandExecution { + id: item_id, + command, + cwd, + process_id, + source: exec_command_begin_event.source.into(), + status: CommandExecutionStatus::InProgress, + command_actions, + aggregated_output: None, + exit_code: None, + duration_ms: None, + }; + let notification = ItemStartedNotification { + thread_id: conversation_id.to_string(), + turn_id: event_turn_id.clone(), + item, + }; + outgoing + .send_server_notification(ServerNotification::ItemStarted(notification)) + .await; } - - let item = ThreadItem::CommandExecution { - id: item_id, - command, - cwd, - process_id, - source: exec_command_begin_event.source.into(), - status: CommandExecutionStatus::InProgress, - command_actions, - aggregated_output: None, - exit_code: None, - duration_ms: None, - }; - let notification = ItemStartedNotification { - thread_id: conversation_id.to_string(), - turn_id: event_turn_id.clone(), - item, - }; - outgoing - .send_server_notification(ServerNotification::ItemStarted(notification)) - .await; } EventMsg::ExecCommandOutputDelta(exec_command_output_delta_event) => { let item_id = exec_command_output_delta_event.call_id.clone(); @@ -1694,20 +1656,7 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } EventMsg::ExecCommandEnd(exec_command_end_event) => { - let ExecCommandEndEvent { - call_id, - command, - cwd, - parsed_cmd, - process_id, - aggregated_output, - exit_code, - duration, - source, - status, - .. - } = exec_command_end_event; - + let call_id = exec_command_end_event.call_id.clone(); { let mut state = thread_state.lock().await; state @@ -1716,32 +1665,7 @@ pub(crate) async fn apply_bespoke_event_handling( .remove(&call_id); } - let status: CommandExecutionStatus = (&status).into(); - let command_actions = parsed_cmd - .into_iter() - .map(V2ParsedCommand::from) - .collect::>(); - - let aggregated_output = if aggregated_output.is_empty() { - None - } else { - Some(aggregated_output) - }; - - let duration_ms = i64::try_from(duration.as_millis()).unwrap_or(i64::MAX); - - let item = ThreadItem::CommandExecution { - id: call_id, - command: shlex_join(&command), - cwd, - process_id, - source: source.into(), - status, - command_actions, - aggregated_output, - exit_code: Some(exit_code), - duration_ms: Some(duration_ms), - }; + let item = build_command_execution_end_item(&exec_command_end_event); let notification = ItemCompletedNotification { thread_id: conversation_id.to_string(), @@ -1966,8 +1890,7 @@ async fn emit_turn_completed_with_status( async fn complete_file_change_item( conversation_id: ThreadId, item_id: String, - changes: Vec, - status: PatchApplyStatus, + item: ThreadItem, turn_id: String, outgoing: &ThreadScopedOutgoingMessageSender, thread_state: &Arc>, @@ -1976,11 +1899,6 @@ async fn complete_file_change_item( state.turn_summary.file_change_started.remove(&item_id); drop(state); - let item = ThreadItem::FileChange { - id: item_id, - changes, - status, - }; let notification = ItemCompletedNotification { thread_id: conversation_id.to_string(), turn_id, @@ -1991,9 +1909,52 @@ async fn complete_file_change_item( .await; } +#[allow(clippy::too_many_arguments)] +async fn start_command_execution_item( + conversation_id: &ThreadId, + turn_id: String, + item_id: String, + command: String, + cwd: PathBuf, + command_actions: Vec, + source: CommandExecutionSource, + outgoing: &ThreadScopedOutgoingMessageSender, + thread_state: &Arc>, +) -> bool { + let first_start = { + let mut state = thread_state.lock().await; + state + .turn_summary + .command_execution_started + .insert(item_id.clone()) + }; + if first_start { + let notification = ItemStartedNotification { + thread_id: conversation_id.to_string(), + turn_id, + item: ThreadItem::CommandExecution { + id: item_id, + command, + cwd, + process_id: None, + source, + status: CommandExecutionStatus::InProgress, + command_actions, + aggregated_output: None, + exit_code: None, + duration_ms: None, + }, + }; + outgoing + .send_server_notification(ServerNotification::ItemStarted(notification)) + .await; + } + first_start +} + #[allow(clippy::too_many_arguments)] async fn complete_command_execution_item( - conversation_id: ThreadId, + conversation_id: &ThreadId, turn_id: String, item_id: String, command: String, @@ -2003,7 +1964,18 @@ async fn complete_command_execution_item( command_actions: Vec, status: CommandExecutionStatus, outgoing: &ThreadScopedOutgoingMessageSender, + thread_state: &Arc>, ) { + let mut state = thread_state.lock().await; + let should_emit = state + .turn_summary + .command_execution_started + .remove(&item_id); + drop(state); + if !should_emit { + return; + } + let item = ThreadItem::CommandExecution { id: item_id, command, @@ -2592,8 +2564,11 @@ async fn on_file_change_request_approval_response( complete_file_change_item( conversation_id, item_id.clone(), - changes, - status, + ThreadItem::FileChange { + id: item_id.clone(), + changes, + status, + }, event_turn_id.clone(), &outgoing, &thread_state, @@ -2710,7 +2685,7 @@ async fn on_command_execution_request_approval_response( && let Some(completion_item) = completion_item { complete_command_execution_item( - conversation_id, + &conversation_id, event_turn_id.clone(), item_id.clone(), completion_item.command, @@ -2720,6 +2695,7 @@ async fn on_command_execution_request_approval_response( completion_item.command_actions, status, &outgoing, + &thread_state, ) .await; } @@ -2861,6 +2837,7 @@ mod tests { use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::TurnPlanStepStatus; + use codex_login::CodexAuth; use codex_protocol::items::HookPromptFragment; use codex_protocol::items::build_hook_prompt_message; use codex_protocol::mcp::CallToolResult; @@ -2871,16 +2848,22 @@ mod tests { use codex_protocol::protocol::CollabResumeBeginEvent; use codex_protocol::protocol::CollabResumeEndEvent; use codex_protocol::protocol::CreditsSnapshot; + use codex_protocol::protocol::GuardianAssessmentEvent; + use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::McpInvocation; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_utils_absolute_path::AbsolutePathBuf; + use core_test_support::load_default_config_for_test; use pretty_assertions::assert_eq; use rmcp::model::Content; use serde_json::Value as JsonValue; + use serde_json::json; + use std::path::PathBuf; use std::time::Duration; + use tempfile::TempDir; use tokio::sync::Mutex; use tokio::sync::mpsc; @@ -2901,6 +2884,84 @@ mod tests { } } + fn command_execution_completion_item(command: &str) -> CommandExecutionCompletionItem { + CommandExecutionCompletionItem { + command: command.to_string(), + cwd: PathBuf::from("/tmp"), + command_actions: vec![V2ParsedCommand::Unknown { + command: command.to_string(), + }], + } + } + + fn guardian_command_assessment( + id: &str, + turn_id: &str, + status: GuardianAssessmentStatus, + ) -> GuardianAssessmentEvent { + let (risk_score, risk_level, rationale) = match status { + GuardianAssessmentStatus::InProgress => (None, None, None), + GuardianAssessmentStatus::Approved => ( + Some(12), + Some(codex_protocol::protocol::GuardianRiskLevel::Low), + Some("looks safe".to_string()), + ), + GuardianAssessmentStatus::Denied => ( + Some(88), + Some(codex_protocol::protocol::GuardianRiskLevel::High), + Some("too risky".to_string()), + ), + GuardianAssessmentStatus::Aborted => (None, None, None), + }; + GuardianAssessmentEvent { + id: id.to_string(), + turn_id: turn_id.to_string(), + status, + risk_score, + risk_level, + rationale, + action: serde_json::from_value(json!({ + "type": "command", + "source": "shell", + "command": format!("rm -f /tmp/{id}.sqlite"), + "cwd": "/tmp", + })) + .expect("guardian action"), + } + } + + struct GuardianAssessmentTestContext { + conversation_id: ThreadId, + conversation: Arc, + thread_manager: Arc, + outgoing: ThreadScopedOutgoingMessageSender, + thread_state: Arc>, + thread_watch_manager: ThreadWatchManager, + codex_home: PathBuf, + } + + impl GuardianAssessmentTestContext { + async fn apply_guardian_assessment_event(&self, assessment: GuardianAssessmentEvent) { + let event_turn_id = assessment.turn_id.clone(); + apply_bespoke_event_handling( + Event { + id: event_turn_id, + msg: EventMsg::GuardianAssessment(assessment), + }, + self.conversation_id.clone(), + self.conversation.clone(), + self.thread_manager.clone(), + self.outgoing.clone(), + self.thread_state.clone(), + self.thread_watch_manager.clone(), + ApiVersion::V2, + "test-provider".to_string(), + &self.codex_home, + ) + .await; + } + } + #[test] fn guardian_assessment_started_uses_event_turn_id_fallback() { let conversation_id = ThreadId::new(); @@ -3019,6 +3080,313 @@ mod tests { } } + #[tokio::test] + async fn command_execution_started_helper_emits_once() -> Result<()> { + let conversation_id = ThreadId::new(); + let thread_state = new_thread_state(); + let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); + let outgoing = Arc::new(OutgoingMessageSender::new(tx)); + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(1)], + ThreadId::new(), + ); + let completion_item = command_execution_completion_item("printf hi"); + + let first_start = start_command_execution_item( + &conversation_id, + "turn-1".to_string(), + "cmd-1".to_string(), + completion_item.command.clone(), + completion_item.cwd.clone(), + completion_item.command_actions.clone(), + CommandExecutionSource::Agent, + &outgoing, + &thread_state, + ) + .await; + assert!(first_start); + + let msg = recv_broadcast_message(&mut rx).await?; + match msg { + OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { + assert_eq!(payload.thread_id, conversation_id.to_string()); + assert_eq!(payload.turn_id, "turn-1"); + assert_eq!( + payload.item, + ThreadItem::CommandExecution { + id: "cmd-1".to_string(), + command: completion_item.command.clone(), + cwd: completion_item.cwd.clone(), + process_id: None, + source: CommandExecutionSource::Agent, + status: CommandExecutionStatus::InProgress, + command_actions: completion_item.command_actions.clone(), + aggregated_output: None, + exit_code: None, + duration_ms: None, + } + ); + } + other => bail!("unexpected message: {other:?}"), + } + + let second_start = start_command_execution_item( + &conversation_id, + "turn-1".to_string(), + "cmd-1".to_string(), + completion_item.command.clone(), + completion_item.cwd.clone(), + completion_item.command_actions.clone(), + CommandExecutionSource::Agent, + &outgoing, + &thread_state, + ) + .await; + assert!(!second_start); + assert!(rx.try_recv().is_err(), "duplicate start should not emit"); + Ok(()) + } + + #[tokio::test] + async fn complete_command_execution_item_emits_declined_once_for_pending_command() -> Result<()> + { + let conversation_id = ThreadId::new(); + let thread_state = new_thread_state(); + let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); + let outgoing = Arc::new(OutgoingMessageSender::new(tx)); + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(1)], + ThreadId::new(), + ); + let completion_item = command_execution_completion_item("printf hi"); + + start_command_execution_item( + &conversation_id, + "turn-1".to_string(), + "cmd-1".to_string(), + completion_item.command.clone(), + completion_item.cwd.clone(), + completion_item.command_actions.clone(), + CommandExecutionSource::Agent, + &outgoing, + &thread_state, + ) + .await; + let _started = recv_broadcast_message(&mut rx).await?; + + complete_command_execution_item( + &conversation_id, + "turn-1".to_string(), + "cmd-1".to_string(), + completion_item.command.clone(), + completion_item.cwd.clone(), + /*process_id*/ None, + CommandExecutionSource::Agent, + completion_item.command_actions.clone(), + CommandExecutionStatus::Declined, + &outgoing, + &thread_state, + ) + .await; + + let completed = recv_broadcast_message(&mut rx).await?; + match completed { + OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => { + let ThreadItem::CommandExecution { id, status, .. } = payload.item else { + bail!("expected command execution completion"); + }; + assert_eq!(id, "cmd-1"); + assert_eq!(status, CommandExecutionStatus::Declined); + } + other => bail!("unexpected message: {other:?}"), + } + + complete_command_execution_item( + &conversation_id, + "turn-1".to_string(), + "cmd-1".to_string(), + completion_item.command, + completion_item.cwd, + /*process_id*/ None, + CommandExecutionSource::Agent, + completion_item.command_actions, + CommandExecutionStatus::Declined, + &outgoing, + &thread_state, + ) + .await; + assert!( + rx.try_recv().is_err(), + "completion should not emit after the pending item is cleared" + ); + Ok(()) + } + + #[tokio::test] + async fn guardian_command_execution_notifications_wrap_review_lifecycle() -> Result<()> { + let codex_home = TempDir::new()?; + let config = load_default_config_for_test(&codex_home).await; + let thread_manager = Arc::new( + codex_core::test_support::thread_manager_with_models_provider_and_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + config.model_provider.clone(), + config.codex_home.clone(), + Arc::new(codex_exec_server::EnvironmentManager::new( + /*exec_server_url*/ None, + )), + ), + ); + let codex_core::NewThread { + thread_id: conversation_id, + thread: conversation, + .. + } = thread_manager.start_thread(config).await?; + let thread_state = new_thread_state(); + let thread_watch_manager = ThreadWatchManager::new(); + let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); + let outgoing = Arc::new(OutgoingMessageSender::new(tx)); + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(1)], + conversation_id, + ); + let guardian_context = GuardianAssessmentTestContext { + conversation_id, + conversation: conversation.clone(), + thread_manager: thread_manager.clone(), + outgoing: outgoing.clone(), + thread_state: thread_state.clone(), + thread_watch_manager: thread_watch_manager.clone(), + codex_home: codex_home.path().to_path_buf(), + }; + + guardian_context + .apply_guardian_assessment_event(guardian_command_assessment( + "cmd-guardian-approved", + "turn-guardian-approved", + GuardianAssessmentStatus::InProgress, + )) + .await; + let first = recv_broadcast_message(&mut rx).await?; + match first { + OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { + assert_eq!(payload.turn_id, "turn-guardian-approved"); + let ThreadItem::CommandExecution { id, status, .. } = payload.item else { + bail!("expected command execution item"); + }; + assert_eq!(id, "cmd-guardian-approved"); + assert_eq!(status, CommandExecutionStatus::InProgress); + } + other => bail!("unexpected message: {other:?}"), + } + let second = recv_broadcast_message(&mut rx).await?; + match second { + OutgoingMessage::AppServerNotification( + ServerNotification::ItemGuardianApprovalReviewStarted(payload), + ) => { + assert_eq!(payload.target_item_id, "cmd-guardian-approved"); + assert_eq!( + payload.review.status, + GuardianApprovalReviewStatus::InProgress + ); + } + other => bail!("unexpected message: {other:?}"), + } + + guardian_context + .apply_guardian_assessment_event(guardian_command_assessment( + "cmd-guardian-approved", + "turn-guardian-approved", + GuardianAssessmentStatus::Approved, + )) + .await; + let third = recv_broadcast_message(&mut rx).await?; + match third { + OutgoingMessage::AppServerNotification( + ServerNotification::ItemGuardianApprovalReviewCompleted(payload), + ) => { + assert_eq!(payload.target_item_id, "cmd-guardian-approved"); + assert_eq!( + payload.review.status, + GuardianApprovalReviewStatus::Approved + ); + } + other => bail!("unexpected message: {other:?}"), + } + assert!( + rx.try_recv().is_err(), + "approved review should not complete the command item" + ); + + guardian_context + .apply_guardian_assessment_event(guardian_command_assessment( + "cmd-guardian-denied", + "turn-guardian-denied", + GuardianAssessmentStatus::InProgress, + )) + .await; + let fourth = recv_broadcast_message(&mut rx).await?; + match fourth { + OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) => { + assert_eq!(payload.turn_id, "turn-guardian-denied"); + let ThreadItem::CommandExecution { id, status, .. } = payload.item else { + bail!("expected command execution item"); + }; + assert_eq!(id, "cmd-guardian-denied"); + assert_eq!(status, CommandExecutionStatus::InProgress); + } + other => bail!("unexpected message: {other:?}"), + } + let fifth = recv_broadcast_message(&mut rx).await?; + match fifth { + OutgoingMessage::AppServerNotification( + ServerNotification::ItemGuardianApprovalReviewStarted(payload), + ) => { + assert_eq!(payload.target_item_id, "cmd-guardian-denied"); + assert_eq!( + payload.review.status, + GuardianApprovalReviewStatus::InProgress + ); + } + other => bail!("unexpected message: {other:?}"), + } + + guardian_context + .apply_guardian_assessment_event(guardian_command_assessment( + "cmd-guardian-denied", + "turn-guardian-denied", + GuardianAssessmentStatus::Denied, + )) + .await; + let sixth = recv_broadcast_message(&mut rx).await?; + match sixth { + OutgoingMessage::AppServerNotification( + ServerNotification::ItemGuardianApprovalReviewCompleted(payload), + ) => { + assert_eq!(payload.target_item_id, "cmd-guardian-denied"); + assert_eq!(payload.review.status, GuardianApprovalReviewStatus::Denied); + } + other => bail!("unexpected message: {other:?}"), + } + let seventh = recv_broadcast_message(&mut rx).await?; + match seventh { + OutgoingMessage::AppServerNotification(ServerNotification::ItemCompleted(payload)) => { + let ThreadItem::CommandExecution { id, status, .. } = payload.item else { + bail!("expected command execution completion"); + }; + assert_eq!(id, "cmd-guardian-denied"); + assert_eq!(status, CommandExecutionStatus::Declined); + } + other => bail!("unexpected message: {other:?}"), + } + + assert!(rx.try_recv().is_err(), "no extra messages expected"); + conversation.shutdown_and_wait().await?; + Ok(()) + } + #[test] fn file_change_accept_for_session_maps_to_approved_for_session() { let (decision, completion_status) = diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index be5478dd51..17823fefd9 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -16,6 +16,7 @@ use std::sync::Weak; use tokio::sync::Mutex; use tokio::sync::mpsc; use tokio::sync::oneshot; +use tracing::error; type PendingInterruptQueue = Vec<( ConnectionRequestId, @@ -116,6 +117,38 @@ impl ThreadState { } } +pub(crate) async fn resolve_server_request_on_thread_listener( + thread_state: &Arc>, + request_id: RequestId, +) { + let (completion_tx, completion_rx) = oneshot::channel(); + let listener_command_tx = { + let state = thread_state.lock().await; + state.listener_command_tx() + }; + let Some(listener_command_tx) = listener_command_tx else { + error!("failed to remove pending client request: thread listener is not running"); + return; + }; + + if listener_command_tx + .send(ThreadListenerCommand::ResolveServerRequest { + request_id, + completion_tx, + }) + .is_err() + { + error!( + "failed to remove pending client request: thread listener command channel is closed" + ); + return; + } + + if let Err(err) = completion_rx.await { + error!("failed to remove pending client request: {err}"); + } +} + struct ThreadEntry { state: Arc>, connection_ids: HashSet, From 9bb813353ec73e8116ddea74de0d73fb80106b2d Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 6 Apr 2026 12:16:31 -0700 Subject: [PATCH 5/9] fix(sqlite): don't hard fail migrator if DB is newer (#16924) ## Description This PR makes the SQLite state runtime tolerate databases that have already been migrated by a newer Codex binary. Today, if an older CLI sees migration versions in `_sqlx_migrations` that it doesn't know about, startup fails. This change relaxes that check for the runtime migrators we use in `codex-state` so older binaries can keep opening the DB in that case. ## Why We can end up with mixed-version CLIs running against the same local state DB. In that setup, treating "the database is ahead of me" as a hard error is unnecessarily strict and breaks the older client even when the migration history is otherwise fine. ## Follow-up We still clean up versioned `state_*.sqlite` and `logs_*.sqlite` files during init, so older binaries can treat newer DB files as legacy. That should probably be tightened separately if we want mixed-version local usage to be fully safe. --- codex-rs/state/src/migrations.rs | 25 ++++++++++ codex-rs/state/src/runtime.rs | 85 +++++++++++++++++++++++++++++--- 2 files changed, 104 insertions(+), 6 deletions(-) diff --git a/codex-rs/state/src/migrations.rs b/codex-rs/state/src/migrations.rs index 6e7b9d363c..883129a943 100644 --- a/codex-rs/state/src/migrations.rs +++ b/codex-rs/state/src/migrations.rs @@ -1,4 +1,29 @@ +use std::borrow::Cow; + use sqlx::migrate::Migrator; pub(crate) static STATE_MIGRATOR: Migrator = sqlx::migrate!("./migrations"); pub(crate) static LOGS_MIGRATOR: Migrator = sqlx::migrate!("./logs_migrations"); + +/// Allow an older Codex binary to open a database that has already been +/// migrated by a newer binary running in parallel. +/// +/// We intentionally ignore applied migration versions that are newer than the +/// embedded migration set. Known migration versions are still validated by +/// checksum, so this only relaxes the "database is ahead of me" case. +fn runtime_migrator(base: &'static Migrator) -> Migrator { + Migrator { + migrations: Cow::Borrowed(base.migrations.as_ref()), + ignore_missing: true, + locking: base.locking, + no_tx: base.no_tx, + } +} + +pub(crate) fn runtime_state_migrator() -> Migrator { + runtime_migrator(&STATE_MIGRATOR) +} + +pub(crate) fn runtime_logs_migrator() -> Migrator { + runtime_migrator(&LOGS_MIGRATOR) +} diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 4d3af301b2..d3d81d87d3 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -17,8 +17,8 @@ use crate::ThreadMetadata; use crate::ThreadMetadataBuilder; use crate::ThreadsPage; use crate::apply_rollout_item; -use crate::migrations::LOGS_MIGRATOR; -use crate::migrations::STATE_MIGRATOR; +use crate::migrations::runtime_logs_migrator; +use crate::migrations::runtime_state_migrator; use crate::model::AgentJobRow; use crate::model::ThreadRow; use crate::model::anchor_from_item; @@ -83,6 +83,8 @@ impl StateRuntime { /// rest of the state store. pub async fn init(codex_home: PathBuf, default_provider: String) -> anyhow::Result> { tokio::fs::create_dir_all(&codex_home).await?; + let state_migrator = runtime_state_migrator(); + let logs_migrator = runtime_logs_migrator(); let current_state_name = state_db_filename(); let current_logs_name = logs_db_filename(); remove_legacy_db_files( @@ -101,14 +103,14 @@ impl StateRuntime { .await; let state_path = state_db_path(codex_home.as_path()); let logs_path = logs_db_path(codex_home.as_path()); - let pool = match open_state_sqlite(&state_path, &STATE_MIGRATOR).await { + let pool = match open_state_sqlite(&state_path, &state_migrator).await { Ok(db) => Arc::new(db), Err(err) => { warn!("failed to open state db at {}: {err}", state_path.display()); return Err(err); } }; - let logs_pool = match open_logs_sqlite(&logs_path, &LOGS_MIGRATOR).await { + let logs_pool = match open_logs_sqlite(&logs_path, &logs_migrator).await { Ok(db) => Arc::new(db), Err(err) => { warn!("failed to open logs db at {}: {err}", logs_path.display()); @@ -146,7 +148,7 @@ fn base_sqlite_options(path: &Path) -> SqliteConnectOptions { .log_statements(LevelFilter::Off) } -async fn open_state_sqlite(path: &Path, migrator: &'static Migrator) -> anyhow::Result { +async fn open_state_sqlite(path: &Path, migrator: &Migrator) -> anyhow::Result { let options = base_sqlite_options(path).auto_vacuum(SqliteAutoVacuum::Incremental); let pool = SqlitePoolOptions::new() .max_connections(5) @@ -172,7 +174,7 @@ async fn open_state_sqlite(path: &Path, migrator: &'static Migrator) -> anyhow:: Ok(pool) } -async fn open_logs_sqlite(path: &Path, migrator: &'static Migrator) -> anyhow::Result { +async fn open_logs_sqlite(path: &Path, migrator: &Migrator) -> anyhow::Result { let options = base_sqlite_options(path).auto_vacuum(SqliteAutoVacuum::Incremental); let pool = SqlitePoolOptions::new() .max_connections(5) @@ -268,3 +270,74 @@ fn should_remove_db_file(file_name: &str, current_name: &str, base_name: &str) - }; !version_suffix.is_empty() && version_suffix.chars().all(|ch| ch.is_ascii_digit()) } + +#[cfg(test)] +mod tests { + use super::open_state_sqlite; + use super::runtime_state_migrator; + use super::state_db_path; + use super::test_support::unique_temp_dir; + use crate::migrations::STATE_MIGRATOR; + use sqlx::SqlitePool; + use sqlx::migrate::MigrateError; + use sqlx::sqlite::SqliteConnectOptions; + use std::path::Path; + + async fn open_db_pool(path: &Path) -> SqlitePool { + SqlitePool::connect_with( + SqliteConnectOptions::new() + .filename(path) + .create_if_missing(false), + ) + .await + .expect("open sqlite pool") + } + + #[tokio::test] + async fn open_state_sqlite_tolerates_newer_applied_migrations() { + let codex_home = unique_temp_dir(); + tokio::fs::create_dir_all(&codex_home) + .await + .expect("create codex home"); + let state_path = state_db_path(codex_home.as_path()); + let pool = SqlitePool::connect_with( + SqliteConnectOptions::new() + .filename(&state_path) + .create_if_missing(true), + ) + .await + .expect("open state db"); + STATE_MIGRATOR + .run(&pool) + .await + .expect("apply current state schema"); + sqlx::query( + "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) VALUES (?, ?, ?, ?, ?)", + ) + .bind(9_999_i64) + .bind("future migration") + .bind(true) + .bind(vec![1_u8, 2, 3, 4]) + .bind(1_i64) + .execute(&pool) + .await + .expect("insert future migration record"); + pool.close().await; + + let strict_pool = open_db_pool(state_path.as_path()).await; + let strict_err = STATE_MIGRATOR + .run(&strict_pool) + .await + .expect_err("strict migrator should reject newer applied migrations"); + assert!(matches!(strict_err, MigrateError::VersionMissing(9_999))); + strict_pool.close().await; + + let tolerant_migrator = runtime_state_migrator(); + let tolerant_pool = open_state_sqlite(state_path.as_path(), &tolerant_migrator) + .await + .expect("runtime migrator should tolerate newer applied migrations"); + tolerant_pool.close().await; + + let _ = tokio::fs::remove_dir_all(codex_home).await; + } +} From 46b7e4fb2cfca7e376f167b51dd9c1eac5e9c70e Mon Sep 17 00:00:00 2001 From: starr-openai Date: Mon, 6 Apr 2026 12:21:58 -0700 Subject: [PATCH 6/9] build: restore lzma-sys Bazel wiring for devbox codex run (#16744) ## Summary - restore the `#16634` `lzma-sys` / `xz` Bazel wiring that was reverted from `main` - re-enable direct Bazel linkage to `@xz//:lzma` with the `lzma-sys` build script disabled - restore the matching `MODULE.bazel.lock` entries ## Why `origin/main` currently builds `//codex-rs/cli:cli` on a devbox, but `bazel run //codex-rs/cli:codex -- --version` fails at link time on the same remote path. Restoring `#16634` fixes that repro. ## Validation - on `origin/main`: `bazel build --bes_backend= --bes_results_url= //codex-rs/cli:cli` passed - on `origin/main`: `bazel run --bes_backend= --bes_results_url= //codex-rs/cli:codex -- --version` failed on `dev` - after this patch on the same `dev` mirror: `bazel run --bes_backend= --bes_results_url= //codex-rs/cli:codex -- --version` passed and printed `codex 0.0.0` --------- Co-authored-by: Codex --- MODULE.bazel | 25 ++++++++++++++++++++++--- MODULE.bazel.lock | 2 ++ patches/BUILD.bazel | 2 ++ patches/bzip2_windows_stack_args.patch | 23 +++++++++++++++++++++++ patches/xz_windows_stack_args.patch | 14 ++++++++++++++ 5 files changed, 63 insertions(+), 3 deletions(-) create mode 100644 patches/bzip2_windows_stack_args.patch create mode 100644 patches/xz_windows_stack_args.patch diff --git a/MODULE.bazel b/MODULE.bazel index b42936fde5..bcc4ec076c 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -228,10 +228,18 @@ inject_repo(crate, "zstd") use_repo(crate, "argument_comment_lint_crates") bazel_dep(name = "bzip2", version = "1.0.8.bcr.3") +single_version_override( + module_name = "bzip2", + patch_strip = 1, + patches = [ + "//patches:bzip2_windows_stack_args.patch", + ], +) crate.annotation( crate = "bzip2-sys", - gen_build_script = "on", + gen_build_script = "off", + deps = ["@bzip2//:bz2"], ) inject_repo(crate, "bzip2") @@ -245,14 +253,25 @@ crate.annotation( inject_repo(crate, "zlib") -# TODO(zbarsky): Enable annotation after fixing windows arm64 builds. +bazel_dep(name = "xz", version = "5.4.5.bcr.8") +single_version_override( + module_name = "xz", + patch_strip = 1, + patches = [ + "//patches:xz_windows_stack_args.patch", + ], +) + crate.annotation( crate = "lzma-sys", - gen_build_script = "on", + gen_build_script = "off", + deps = ["@xz//:lzma"], ) bazel_dep(name = "openssl", version = "3.5.4.bcr.0") +inject_repo(crate, "xz") + crate.annotation( build_script_data = [ "@openssl//:gen_dir", diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index 1e2123ec49..746679af39 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -228,6 +228,8 @@ "https://bcr.bazel.build/modules/upb/0.0.0-20220923-a547704/MODULE.bazel": "7298990c00040a0e2f121f6c32544bab27d4452f80d9ce51349b1a28f3005c43", "https://bcr.bazel.build/modules/with_cfg.bzl/0.12.0/MODULE.bazel": "b573395fe63aef4299ba095173e2f62ccfee5ad9bbf7acaa95dba73af9fc2b38", "https://bcr.bazel.build/modules/with_cfg.bzl/0.12.0/source.json": "3f3fbaeafecaf629877ad152a2c9def21f8d330d91aa94c5dc75bbb98c10b8b8", + "https://bcr.bazel.build/modules/xz/5.4.5.bcr.8/MODULE.bazel": "e48a69bd54053c2ec5fffc2a29fb70122afd3e83ab6c07068f63bc6553fa57cc", + "https://bcr.bazel.build/modules/xz/5.4.5.bcr.8/source.json": "bd7e928ccd63505b44f4784f7bbf12cc11f9ff23bf3ca12ff2c91cd74846099e", "https://bcr.bazel.build/modules/zlib/1.2.11/MODULE.bazel": "07b389abc85fdbca459b69e2ec656ae5622873af3f845e1c9d80fe179f3effa0", "https://bcr.bazel.build/modules/zlib/1.3.1.bcr.5/MODULE.bazel": "eec517b5bbe5492629466e11dae908d043364302283de25581e3eb944326c4ca", "https://bcr.bazel.build/modules/zlib/1.3.1.bcr.8/MODULE.bazel": "772c674bb78a0342b8caf32ab5c25085c493ca4ff08398208dcbe4375fe9f776", diff --git a/patches/BUILD.bazel b/patches/BUILD.bazel index 75d6be358e..4db61293c2 100644 --- a/patches/BUILD.bazel +++ b/patches/BUILD.bazel @@ -3,6 +3,7 @@ exports_files([ "aws-lc-sys_memcmp_check.patch", "aws-lc-sys_windows_msvc_prebuilt_nasm.patch", "aws-lc-sys_windows_msvc_memcmp_probe.patch", + "bzip2_windows_stack_args.patch", "llvm_windows_symlink_extract.patch", "rules_rust_windows_bootstrap_process_wrapper_linker.patch", "rules_rust_windows_build_script_runner_paths.patch", @@ -20,5 +21,6 @@ exports_files([ "v8_module_deps.patch", "v8_source_portability.patch", "windows-link.patch", + "xz_windows_stack_args.patch", "zstd-sys_windows_msvc_include_dirs.patch", ]) diff --git a/patches/bzip2_windows_stack_args.patch b/patches/bzip2_windows_stack_args.patch new file mode 100644 index 0000000000..2786e908be --- /dev/null +++ b/patches/bzip2_windows_stack_args.patch @@ -0,0 +1,23 @@ +diff --git a/BUILD.bazel b/BUILD.bazel +--- a/BUILD.bazel ++++ b/BUILD.bazel +@@ -28,4 +28,11 @@ cc_library( + defines = [ + "_FILE_OFFSET_BITS=64", + ], ++ copts = select({ ++ "@platforms//os:windows": [ ++ "-fno-stack-protector", ++ "-mno-stack-arg-probe", ++ ], ++ "//conditions:default": [], ++ }), + includes = ["."], +diff --git a/MODULE.bazel b/MODULE.bazel +--- a/MODULE.bazel ++++ b/MODULE.bazel +@@ -4,3 +4,4 @@ module( + ) + + bazel_dep(name = "rules_cc", version = "0.0.10") ++bazel_dep(name = "platforms", version = "1.0.0") diff --git a/patches/xz_windows_stack_args.patch b/patches/xz_windows_stack_args.patch new file mode 100644 index 0000000000..926fb0a6dc --- /dev/null +++ b/patches/xz_windows_stack_args.patch @@ -0,0 +1,14 @@ +diff --git a/BUILD.bazel b/BUILD.bazel +--- a/BUILD.bazel ++++ b/BUILD.bazel +@@ -154,6 +154,9 @@ cc_library( + ], + copts = select({ +- "@platforms//os:windows": [], ++ "@platforms//os:windows": [ ++ "-fno-stack-protector", ++ "-mno-stack-arg-probe", ++ ], + "//conditions:default": ["-std=c99"], + }), + defines = select({ From 1525bbdb9a77ce7a817b19465a6cce8e2c0800bf Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Mon, 6 Apr 2026 12:46:55 -0700 Subject: [PATCH 7/9] app-server: centralize AuthManager initialization (#16764) Extract a shared helper that builds AuthManager from Config and applies the forced ChatGPT workspace override in one place. Create the shared AuthManager at MessageProcessor call sites so that upcoming new transport's initialization can reuse the same handle, and keep only external auth refresher wiring inside `MessageProcessor`. Remove the now-unused `AuthManager::shared_with_external_auth` helper. --- codex-rs/app-server/src/in_process.rs | 5 ++- codex-rs/app-server/src/lib.rs | 11 +++--- codex-rs/app-server/src/message_processor.rs | 16 +++------ .../src/message_processor/tracing_tests.rs | 5 ++- codex-rs/core/src/config/mod.rs | 15 ++++++++ codex-rs/core/src/connectors.rs | 17 ++++------ codex-rs/core/src/prompt_debug.rs | 8 ++--- codex-rs/login/src/auth/manager.rs | 34 ++++++++++++++----- codex-rs/login/src/lib.rs | 1 + codex-rs/mcp-server/src/message_processor.rs | 5 ++- 10 files changed, 69 insertions(+), 48 deletions(-) diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index eda2c75da8..71beb58dc0 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -80,6 +80,7 @@ use codex_core::config_loader::CloudRequirementsLoader; use codex_core::config_loader::LoaderOverrides; use codex_exec_server::EnvironmentManager; use codex_feedback::CodexFeedback; +use codex_login::AuthManager; use codex_protocol::protocol::SessionSource; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -379,6 +380,8 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { }); let processor_outgoing = Arc::clone(&outgoing_message_sender); + let auth_manager = + AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env); let (processor_tx, mut processor_rx) = mpsc::channel::(channel_capacity); let mut processor_handle = tokio::spawn(async move { let mut processor = MessageProcessor::new(MessageProcessorArgs { @@ -393,7 +396,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { log_db: None, config_warnings: args.config_warnings, session_source: args.session_source, - enable_codex_api_key_env: args.enable_codex_api_key_env, + auth_manager, rpc_transport: AppServerRpcTransport::InProcess, }); let mut thread_created_rx = processor.thread_created_receiver(); diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 8c3742adcc..14250c205e 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -396,11 +396,8 @@ pub async fn run_main_with_transport( } } - let auth_manager = AuthManager::shared( - config.codex_home.clone(), - /*enable_codex_api_key_env*/ false, - config.cli_auth_credentials_store_mode, - ); + let auth_manager = + AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false); cloud_requirements_loader( auth_manager, config.chatgpt_base_url, @@ -611,6 +608,8 @@ pub async fn run_main_with_transport( let processor_handle = tokio::spawn({ let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx)); let outbound_control_tx = outbound_control_tx; + let auth_manager = + AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false); let cli_overrides: Vec<(String, TomlValue)> = cli_kv_overrides.clone(); let loader_overrides = loader_overrides_for_config_api; let mut processor = MessageProcessor::new(MessageProcessorArgs { @@ -625,7 +624,7 @@ pub async fn run_main_with_transport( log_db, config_warnings, session_source, - enable_codex_api_key_env: false, + auth_manager, rpc_transport: analytics_rpc_transport(transport), }); let mut thread_created_rx = processor.thread_created_receiver(); diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 02b948bc1f..8e3c5be81c 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -193,7 +193,7 @@ pub(crate) struct MessageProcessorArgs { pub(crate) log_db: Option, pub(crate) config_warnings: Vec, pub(crate) session_source: SessionSource, - pub(crate) enable_codex_api_key_env: bool, + pub(crate) auth_manager: Arc, pub(crate) rpc_transport: AppServerRpcTransport, } @@ -213,17 +213,12 @@ impl MessageProcessor { log_db, config_warnings, session_source, - enable_codex_api_key_env, + auth_manager, rpc_transport, } = args; - let auth_manager = AuthManager::shared_with_external_auth( - config.codex_home.clone(), - enable_codex_api_key_env, - config.cli_auth_credentials_store_mode, - Arc::new(ExternalAuthRefreshBridge { - outgoing: outgoing.clone(), - }), - ); + auth_manager.set_external_auth(Arc::new(ExternalAuthRefreshBridge { + outgoing: outgoing.clone(), + })); let thread_manager = Arc::new(ThreadManager::new( config.as_ref(), auth_manager.clone(), @@ -235,7 +230,6 @@ impl MessageProcessor { }, environment_manager, )); - auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone()); let analytics_events_client = AnalyticsEventsClient::new( Arc::clone(&auth_manager), config.chatgpt_base_url.trim_end_matches('/').to_string(), diff --git a/codex-rs/app-server/src/message_processor/tracing_tests.rs b/codex-rs/app-server/src/message_processor/tracing_tests.rs index d110f4feea..2e8781606f 100644 --- a/codex-rs/app-server/src/message_processor/tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor/tracing_tests.rs @@ -27,6 +27,7 @@ use codex_core::config_loader::CloudRequirementsLoader; use codex_core::config_loader::LoaderOverrides; use codex_exec_server::EnvironmentManager; use codex_feedback::CodexFeedback; +use codex_login::AuthManager; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::W3cTraceContext; use opentelemetry::global; @@ -234,6 +235,8 @@ fn build_test_processor( ) { let (outgoing_tx, outgoing_rx) = mpsc::channel(16); let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx)); + let auth_manager = + AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false); let processor = MessageProcessor::new(MessageProcessorArgs { outgoing, arg0_paths: Arg0DispatchPaths::default(), @@ -246,7 +249,7 @@ fn build_test_processor( log_db: None, config_warnings: Vec::new(), session_source: SessionSource::VSCode, - enable_codex_api_key_env: false, + auth_manager, rpc_transport: AppServerRpcTransport::Stdio, }); (processor, outgoing_rx) diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index d83fc87371..cb497815f3 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -58,6 +58,7 @@ use codex_features::Features; use codex_features::FeaturesToml; use codex_git_utils::resolve_root_git_project_for_trust; use codex_login::AuthCredentialsStoreMode; +use codex_login::AuthManagerConfig; use codex_mcp::mcp::McpConfig; use codex_model_provider_info::LEGACY_OLLAMA_CHAT_PROVIDER_ID; use codex_model_provider_info::LMSTUDIO_OSS_PROVIDER_ID; @@ -593,6 +594,20 @@ pub struct Config { pub otel: codex_config::types::OtelConfig, } +impl AuthManagerConfig for Config { + fn codex_home(&self) -> PathBuf { + self.codex_home.clone() + } + + fn cli_auth_credentials_store_mode(&self) -> AuthCredentialsStoreMode { + self.cli_auth_credentials_store_mode + } + + fn forced_chatgpt_workspace_id(&self) -> Option { + self.forced_chatgpt_workspace_id.clone() + } +} + #[derive(Debug, Clone, Default)] pub struct ConfigBuilder { codex_home: Option, diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 10e10e99fc..66ba6a406b 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -143,7 +143,8 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( pub async fn list_cached_accessible_connectors_from_mcp_tools( config: &Config, ) -> Option> { - let auth_manager = auth_manager_from_config(config); + let auth_manager = + AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false); let auth = auth_manager.auth().await; if !config.features.apps_enabled_for_auth(auth.as_ref()) { return Some(Vec::new()); @@ -182,7 +183,8 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_options_and_status( config: &Config, force_refetch: bool, ) -> anyhow::Result { - let auth_manager = auth_manager_from_config(config); + let auth_manager = + AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false); let auth = auth_manager.auth().await; if !config.features.apps_enabled_for_auth(auth.as_ref()) { return Ok(AccessibleConnectorsStatus { @@ -417,7 +419,8 @@ async fn list_directory_connectors_for_tool_suggest_with_auth( let token_data = if let Some(auth) = auth { auth.get_token_data().ok() } else { - let auth_manager = auth_manager_from_config(config); + let auth_manager = + AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false); auth_manager .auth() .await @@ -492,14 +495,6 @@ async fn chatgpt_get_request_with_token( } } -fn auth_manager_from_config(config: &Config) -> std::sync::Arc { - AuthManager::shared( - config.codex_home.clone(), - /*enable_codex_api_key_env*/ false, - config.cli_auth_credentials_store_mode, - ) -} - pub fn connector_display_label(connector: &AppInfo) -> String { format_connector_label(&connector.name, &connector.id) } diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index 4a8513bb49..5fe39ee88f 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -26,12 +26,8 @@ pub async fn build_prompt_input( ) -> CodexResult> { config.ephemeral = true; - let auth_manager = AuthManager::shared( - config.codex_home.clone(), - /*enable_codex_api_key_env*/ false, - config.cli_auth_credentials_store_mode, - ); - auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone()); + let auth_manager = + AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false); let thread_manager = ThreadManager::new( &config, diff --git a/codex-rs/login/src/auth/manager.rs b/codex-rs/login/src/auth/manager.rs index 9f58f0f090..f0b31f0f49 100644 --- a/codex-rs/login/src/auth/manager.rs +++ b/codex-rs/login/src/auth/manager.rs @@ -1108,6 +1108,23 @@ pub struct AuthManager { external_auth: RwLock>>, } +/// Configuration view required to construct a shared [`AuthManager`]. +/// +/// Implementations should return the auth-related config values for the +/// already-resolved runtime configuration. The primary implementation is +/// `codex_core::config::Config`, but this trait keeps `codex-login` independent +/// from `codex-core`. +pub trait AuthManagerConfig { + /// Returns the Codex home directory used for auth storage. + fn codex_home(&self) -> PathBuf; + + /// Returns the CLI auth credential storage mode for auth loading. + fn cli_auth_credentials_store_mode(&self) -> AuthCredentialsStoreMode; + + /// Returns the workspace ID that ChatGPT auth should be restricted to, if any. + fn forced_chatgpt_workspace_id(&self) -> Option; +} + impl Debug for AuthManager { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("AuthManager") @@ -1404,19 +1421,18 @@ impl AuthManager { )) } - pub fn shared_with_external_auth( - codex_home: PathBuf, + /// Convenience constructor returning an `Arc` wrapper from resolved config. + pub fn shared_from_config( + config: &impl AuthManagerConfig, enable_codex_api_key_env: bool, - auth_credentials_store_mode: AuthCredentialsStoreMode, - external_auth: Arc, ) -> Arc { - let manager = Self::shared( - codex_home, + let auth_manager = Self::shared( + config.codex_home(), enable_codex_api_key_env, - auth_credentials_store_mode, + config.cli_auth_credentials_store_mode(), ); - manager.set_external_auth(external_auth); - manager + auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id()); + auth_manager } pub fn unauthorized_recovery(self: &Arc) -> UnauthorizedRecovery { diff --git a/codex-rs/login/src/lib.rs b/codex-rs/login/src/lib.rs index ad77d82389..247e5a876a 100644 --- a/codex-rs/login/src/lib.rs +++ b/codex-rs/login/src/lib.rs @@ -23,6 +23,7 @@ pub use auth::AuthConfig; pub use auth::AuthCredentialsStoreMode; pub use auth::AuthDotJson; pub use auth::AuthManager; +pub use auth::AuthManagerConfig; pub use auth::CLIENT_ID; pub use auth::CODEX_API_KEY_ENV_VAR; pub use auth::CodexAuth; diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index a0d4321173..46470f994a 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -56,10 +56,9 @@ impl MessageProcessor { environment_manager: Arc, ) -> Self { let outgoing = Arc::new(outgoing); - let auth_manager = AuthManager::shared( - config.codex_home.clone(), + let auth_manager = AuthManager::shared_from_config( + config.as_ref(), /*enable_codex_api_key_env*/ false, - config.cli_auth_credentials_store_mode, ); let thread_manager = Arc::new(ThreadManager::new( config.as_ref(), From 756ba8baae8b033e02fa8f26b5d16ad5b06adc02 Mon Sep 17 00:00:00 2001 From: Matthew Zeng Date: Mon, 6 Apr 2026 17:08:55 -0400 Subject: [PATCH 8/9] Fix clippy warning (#16939) - [x] Fix clippy warning --- codex-rs/app-server/src/bespoke_event_handling.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 01eb4bd578..f883d078e4 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -2948,7 +2948,7 @@ mod tests { id: event_turn_id, msg: EventMsg::GuardianAssessment(assessment), }, - self.conversation_id.clone(), + self.conversation_id, self.conversation.clone(), self.thread_manager.clone(), self.outgoing.clone(), From 03c07956cfa1e729dde24d767a2ced322455cdb2 Mon Sep 17 00:00:00 2001 From: joeytrasatti-openai Date: Mon, 6 Apr 2026 17:25:05 -0400 Subject: [PATCH 9/9] Revert "[codex-backend] Make thread metadata updates tolerate pending backfill" (#16923) Reverts openai/codex#16877 --- .../app-server/src/codex_message_processor.rs | 18 ------- .../tests/suite/v2/thread_metadata_update.rs | 53 ------------------- 2 files changed, 71 deletions(-) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 86c2218bc8..6475531c07 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -2754,24 +2754,6 @@ impl CodexMessageProcessor { if state_db_ctx.is_none() { state_db_ctx = get_state_db(&self.config).await; } - if state_db_ctx.is_none() { - match StateRuntime::init( - self.config.sqlite_home.clone(), - self.config.model_provider_id.clone(), - ) - .await - { - Ok(ctx) => { - state_db_ctx = Some(ctx); - } - Err(err) => { - warn!( - "failed to initialize state db for thread metadata update at {}: {err}", - self.config.sqlite_home.display() - ); - } - } - } let Some(state_db_ctx) = state_db_ctx else { self.send_internal_error( request_id, diff --git a/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs b/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs index f5c793f744..8bf9a8a9aa 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs @@ -223,59 +223,6 @@ async fn thread_metadata_update_repairs_missing_sqlite_row_for_stored_thread() - Ok(()) } -#[tokio::test] -async fn thread_metadata_update_repairs_stored_thread_before_backfill_completes() -> 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 _state_db = - StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?; - - let preview = "Stored thread preview before backfill"; - let thread_id = create_fake_rollout( - codex_home.path(), - "2025-01-05T12-30-00", - "2025-01-05T12:30:00Z", - preview, - Some("mock_provider"), - /*git_info*/ None, - )?; - - let mut mcp = McpProcess::new(codex_home.path()).await?; - timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; - - let update_id = mcp - .send_thread_metadata_update_request(ThreadMetadataUpdateParams { - thread_id: thread_id.clone(), - git_info: Some(ThreadMetadataGitInfoUpdateParams { - sha: None, - branch: Some(Some("feature/pending-backfill".to_string())), - origin_url: None, - }), - }) - .await?; - let update_resp: JSONRPCResponse = timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(update_id)), - ) - .await??; - let ThreadMetadataUpdateResponse { thread: updated } = - to_response::(update_resp)?; - - assert_eq!(updated.id, thread_id); - assert_eq!(updated.preview, preview); - assert_eq!( - updated.git_info, - Some(GitInfo { - sha: None, - branch: Some("feature/pending-backfill".to_string()), - origin_url: None, - }) - ); - - Ok(()) -} - #[tokio::test] async fn thread_metadata_update_repairs_loaded_thread_without_resetting_summary() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await;