From 245153954fd1d8f5fea4ff6e66098ddc25c4b603 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 4 Nov 2025 09:40:22 -0800 Subject: [PATCH] feat: introduce app-server v2 threads and turns --- .github/actions/setup-musl-1_2_5/action.yml | 47 - .github/workflows/rust-ci.yml | 12 +- .github/workflows/rust-release.yml | 8 +- codex-rs/Cargo.lock | 2 + codex-rs/app-server-protocol/Cargo.toml | 1 + codex-rs/app-server-protocol/src/export.rs | 333 +++-- .../src/protocol/common.rs | 234 +++- .../app-server-protocol/src/protocol/v2.rs | 550 ++++++++- .../app-server/src/codex_message_processor.rs | 1083 +++++++++++++---- codex-rs/app-server/src/outgoing_message.rs | 38 +- codex-rs/app-server/tests/common/Cargo.toml | 1 + codex-rs/app-server/tests/common/lib.rs | 3 + .../app-server/tests/common/mcp_process.rs | 68 +- .../tests/common/mock_model_server.rs | 21 + codex-rs/app-server/tests/common/rollout.rs | 77 ++ .../app-server/tests/suite/list_resume.rs | 72 +- codex-rs/app-server/tests/suite/mod.rs | 2 - codex-rs/app-server/tests/suite/v2/mod.rs | 9 +- .../tests/suite/{ => v2}/model_list.rs | 42 +- .../tests/suite/{ => v2}/rate_limits.rs | 18 +- .../tests/suite/v2/thread_archive.rs | 93 ++ .../app-server/tests/suite/v2/thread_list.rs | 208 ++++ .../tests/suite/v2/thread_resume.rs | 79 ++ .../app-server/tests/suite/v2/thread_start.rs | 82 ++ .../tests/suite/v2/turn_interrupt.rs | 132 ++ .../app-server/tests/suite/v2/turn_start.rs | 529 ++++++++ 26 files changed, 3173 insertions(+), 571 deletions(-) delete mode 100644 .github/actions/setup-musl-1_2_5/action.yml create mode 100644 codex-rs/app-server/tests/common/rollout.rs rename codex-rs/app-server/tests/suite/{ => v2}/model_list.rs (84%) rename codex-rs/app-server/tests/suite/{ => v2}/rate_limits.rs (92%) create mode 100644 codex-rs/app-server/tests/suite/v2/thread_archive.rs create mode 100644 codex-rs/app-server/tests/suite/v2/thread_list.rs create mode 100644 codex-rs/app-server/tests/suite/v2/thread_resume.rs create mode 100644 codex-rs/app-server/tests/suite/v2/thread_start.rs create mode 100644 codex-rs/app-server/tests/suite/v2/turn_interrupt.rs create mode 100644 codex-rs/app-server/tests/suite/v2/turn_start.rs diff --git a/.github/actions/setup-musl-1_2_5/action.yml b/.github/actions/setup-musl-1_2_5/action.yml deleted file mode 100644 index 64d4320480..0000000000 --- a/.github/actions/setup-musl-1_2_5/action.yml +++ /dev/null @@ -1,47 +0,0 @@ -name: Setup musl 1.2.5 toolchain -description: Install musl 1.2.5 from source and configure the linker for the requested target. -inputs: - target: - description: Cargo target triple that requires musl (e.g., x86_64-unknown-linux-musl). - required: true -runs: - using: composite - steps: - - name: Install musl 1.2.5 - shell: bash - env: - MUSL_VERSION: 1.2.5 - MUSL_PREFIX: /opt/musl-1.2.5 - DEBIAN_FRONTEND: noninteractive - run: | - set -euo pipefail - sudo apt-get -y update -o Acquire::Retries=3 - sudo apt-get -y install --no-install-recommends build-essential curl pkg-config - - curl -sSfL --retry 3 --retry-delay 1 "https://musl.libc.org/releases/musl-${MUSL_VERSION}.tar.gz" -o /tmp/musl.tar.gz - tar -xf /tmp/musl.tar.gz -C /tmp - - pushd "/tmp/musl-${MUSL_VERSION}" - ./configure --prefix="${MUSL_PREFIX}" - make -j"$(nproc)" - sudo make install - popd - - echo "${MUSL_PREFIX}/bin" >> "$GITHUB_PATH" - musl_gcc="${MUSL_PREFIX}/bin/musl-gcc" - "${musl_gcc}" --version - - case "${{ inputs.target }}" in - x86_64-unknown-linux-musl) - echo "CC_x86_64_unknown_linux_musl=${musl_gcc}" >> "$GITHUB_ENV" - echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_LINKER=${musl_gcc}" >> "$GITHUB_ENV" - ;; - aarch64-unknown-linux-musl) - echo "CC_aarch64_unknown_linux_musl=${musl_gcc}" >> "$GITHUB_ENV" - echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER=${musl_gcc}" >> "$GITHUB_ENV" - ;; - *) - echo "Unsupported musl target '${{ inputs.target }}'" >&2 - exit 1 - ;; - esac diff --git a/.github/workflows/rust-ci.yml b/.github/workflows/rust-ci.yml index 6bc7574dae..103e8cf6bd 100644 --- a/.github/workflows/rust-ci.yml +++ b/.github/workflows/rust-ci.yml @@ -217,10 +217,14 @@ jobs: key: apt-${{ matrix.runner }}-${{ matrix.target }}-v1 - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Setup musl 1.2.5 toolchain - uses: ./.github/actions/setup-musl-1_2_5 - with: - target: ${{ matrix.target }} + name: Install musl build tools + env: + DEBIAN_FRONTEND: noninteractive + shell: bash + run: | + set -euo pipefail + sudo apt-get -y update -o Acquire::Retries=3 + sudo apt-get -y install --no-install-recommends musl-tools pkg-config - name: Install cargo-chef if: ${{ matrix.profile == 'release' }} diff --git a/.github/workflows/rust-release.yml b/.github/workflows/rust-release.yml index f58c84bd9e..26afdaf552 100644 --- a/.github/workflows/rust-release.yml +++ b/.github/workflows/rust-release.yml @@ -92,10 +92,10 @@ jobs: key: cargo-${{ matrix.runner }}-${{ matrix.target }}-release-${{ hashFiles('**/Cargo.lock') }} - if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}} - name: Setup musl 1.2.5 toolchain - uses: ./.github/actions/setup-musl-1_2_5 - with: - target: ${{ matrix.target }} + name: Install musl build tools + run: | + sudo apt-get update + sudo apt-get install -y musl-tools pkg-config - name: Cargo build run: cargo build --target ${{ matrix.target }} --release --bin codex --bin codex-responses-api-proxy diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 4961486e65..42f05a819e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -189,6 +189,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "uuid", "wiremock", ] @@ -871,6 +872,7 @@ dependencies = [ "anyhow", "clap", "codex-protocol", + "mcp-types", "paste", "pretty_assertions", "schemars 0.8.22", diff --git a/codex-rs/app-server-protocol/Cargo.toml b/codex-rs/app-server-protocol/Cargo.toml index 58b0b9d607..5aa1c765e7 100644 --- a/codex-rs/app-server-protocol/Cargo.toml +++ b/codex-rs/app-server-protocol/Cargo.toml @@ -14,6 +14,7 @@ workspace = true anyhow = { workspace = true } clap = { workspace = true, features = ["derive"] } codex-protocol = { workspace = true } +mcp-types = { workspace = true } paste = { workspace = true } schemars = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 3affa7afca..9692fb95c5 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -2,8 +2,12 @@ use crate::ClientNotification; use crate::ClientRequest; use crate::ServerNotification; use crate::ServerRequest; +use crate::export_client_notification_schemas; +use crate::export_client_param_schemas; use crate::export_client_response_schemas; use crate::export_client_responses; +use crate::export_server_notification_schemas; +use crate::export_server_param_schemas; use crate::export_server_response_schemas; use crate::export_server_responses; use anyhow::Context; @@ -28,84 +32,7 @@ use ts_rs::TS; const HEADER: &str = "// GENERATED CODE! DO NOT MODIFY BY HAND!\n\n"; -macro_rules! for_each_schema_type { - ($macro:ident) => { - $macro!(crate::RequestId); - $macro!(crate::JSONRPCMessage); - $macro!(crate::JSONRPCRequest); - $macro!(crate::JSONRPCNotification); - $macro!(crate::JSONRPCResponse); - $macro!(crate::JSONRPCError); - $macro!(crate::JSONRPCErrorError); - $macro!(crate::AddConversationListenerParams); - $macro!(crate::AddConversationSubscriptionResponse); - $macro!(crate::ApplyPatchApprovalParams); - $macro!(crate::ApplyPatchApprovalResponse); - $macro!(crate::ArchiveConversationParams); - $macro!(crate::ArchiveConversationResponse); - $macro!(crate::AuthMode); - $macro!(crate::AccountUpdatedNotification); - $macro!(crate::AuthStatusChangeNotification); - $macro!(crate::CancelLoginChatGptParams); - $macro!(crate::CancelLoginChatGptResponse); - $macro!(crate::ClientInfo); - $macro!(crate::ClientNotification); - $macro!(crate::ClientRequest); - $macro!(crate::ConversationSummary); - $macro!(crate::ExecCommandApprovalParams); - $macro!(crate::ExecCommandApprovalResponse); - $macro!(crate::ExecOneOffCommandParams); - $macro!(crate::ExecOneOffCommandResponse); - $macro!(crate::FuzzyFileSearchParams); - $macro!(crate::FuzzyFileSearchResponse); - $macro!(crate::FuzzyFileSearchResult); - $macro!(crate::GetAuthStatusParams); - $macro!(crate::GetAuthStatusResponse); - $macro!(crate::GetUserAgentResponse); - $macro!(crate::GetUserSavedConfigResponse); - $macro!(crate::GitDiffToRemoteParams); - $macro!(crate::GitDiffToRemoteResponse); - $macro!(crate::GitSha); - $macro!(crate::InitializeParams); - $macro!(crate::InitializeResponse); - $macro!(crate::InputItem); - $macro!(crate::InterruptConversationParams); - $macro!(crate::InterruptConversationResponse); - $macro!(crate::ListConversationsParams); - $macro!(crate::ListConversationsResponse); - $macro!(crate::LoginApiKeyParams); - $macro!(crate::LoginApiKeyResponse); - $macro!(crate::LoginChatGptCompleteNotification); - $macro!(crate::LoginChatGptResponse); - $macro!(crate::LogoutChatGptParams); - $macro!(crate::LogoutChatGptResponse); - $macro!(crate::NewConversationParams); - $macro!(crate::NewConversationResponse); - $macro!(crate::Profile); - $macro!(crate::RemoveConversationListenerParams); - $macro!(crate::RemoveConversationSubscriptionResponse); - $macro!(crate::ResumeConversationParams); - $macro!(crate::ResumeConversationResponse); - $macro!(crate::SandboxSettings); - $macro!(crate::SendUserMessageParams); - $macro!(crate::SendUserMessageResponse); - $macro!(crate::SendUserTurnParams); - $macro!(crate::SendUserTurnResponse); - $macro!(crate::ServerNotification); - $macro!(crate::ServerRequest); - $macro!(crate::SessionConfiguredNotification); - $macro!(crate::SetDefaultModelParams); - $macro!(crate::SetDefaultModelResponse); - $macro!(crate::Tools); - $macro!(crate::UserInfoResponse); - $macro!(crate::UserSavedConfig); - $macro!(codex_protocol::protocol::EventMsg); - $macro!(codex_protocol::protocol::FileChange); - $macro!(codex_protocol::parse_command::ParsedCommand); - $macro!(codex_protocol::protocol::SandboxPolicy); - }; -} - +type JsonSchemaEmitter = fn(&Path) -> Result; pub fn generate_types(out_dir: &Path, prettier: Option<&Path>) -> Result<()> { generate_ts(out_dir, prettier)?; generate_json(out_dir)?; @@ -113,7 +40,9 @@ pub fn generate_types(out_dir: &Path, prettier: Option<&Path>) -> Result<()> { } pub fn generate_ts(out_dir: &Path, prettier: Option<&Path>) -> Result<()> { + let v2_out_dir = out_dir.join("v2"); ensure_dir(out_dir)?; + ensure_dir(&v2_out_dir)?; ClientRequest::export_all_to(out_dir)?; export_client_responses(out_dir)?; @@ -124,12 +53,15 @@ pub fn generate_ts(out_dir: &Path, prettier: Option<&Path>) -> Result<()> { ServerNotification::export_all_to(out_dir)?; generate_index_ts(out_dir)?; + generate_index_ts(&v2_out_dir)?; - let ts_files = ts_files_in(out_dir)?; + // Ensure our header is present on all TS files (root + subdirs like v2/). + let ts_files = ts_files_in_recursive(out_dir)?; for file in &ts_files { prepend_header_if_missing(file)?; } + // Optionally run Prettier on all generated TS files. if let Some(prettier_bin) = prettier && !ts_files.is_empty() { @@ -148,20 +80,57 @@ pub fn generate_ts(out_dir: &Path, prettier: Option<&Path>) -> Result<()> { pub fn generate_json(out_dir: &Path) -> Result<()> { ensure_dir(out_dir)?; + // Emit schemas for core envelope/wire types that may not be referenced + // by params/responses but are useful as standalone: let mut bundle: BTreeMap = BTreeMap::new(); - macro_rules! add_schema { - ($ty:path) => {{ - let name = type_basename(stringify!($ty)); - let schema = write_json_schema_with_return::<$ty>(out_dir, &name)?; - bundle.insert(name, schema); - }}; + let envelope_emitters: &[(&str, JsonSchemaEmitter)] = &[ + ("RequestId", |d| { + write_json_schema_with_return::(d, "RequestId") + }), + ("JSONRPCMessage", |d| { + write_json_schema_with_return::(d, "JSONRPCMessage") + }), + ("JSONRPCRequest", |d| { + write_json_schema_with_return::(d, "JSONRPCRequest") + }), + ("JSONRPCNotification", |d| { + write_json_schema_with_return::(d, "JSONRPCNotification") + }), + ("JSONRPCResponse", |d| { + write_json_schema_with_return::(d, "JSONRPCResponse") + }), + ("JSONRPCError", |d| { + write_json_schema_with_return::(d, "JSONRPCError") + }), + ("JSONRPCErrorError", |d| { + write_json_schema_with_return::(d, "JSONRPCErrorError") + }), + ("ClientRequest", |d| { + write_json_schema_with_return::(d, "ClientRequest") + }), + ("ServerRequest", |d| { + write_json_schema_with_return::(d, "ServerRequest") + }), + ("ClientNotification", |d| { + write_json_schema_with_return::(d, "ClientNotification") + }), + ("ServerNotification", |d| { + write_json_schema_with_return::(d, "ServerNotification") + }), + ]; + for (name, emit) in envelope_emitters { + let schema = emit(out_dir)?; + bundle.insert((*name).to_string(), schema); } - for_each_schema_type!(add_schema); - + // Have the macros generate per-type JSON for params, responses, and notifications. + export_client_param_schemas(out_dir)?; export_client_response_schemas(out_dir)?; + export_server_param_schemas(out_dir)?; export_server_response_schemas(out_dir)?; + export_client_notification_schemas(out_dir)?; + export_server_notification_schemas(out_dir)?; let mut definitions = Map::new(); @@ -177,10 +146,44 @@ pub fn generate_json(out_dir: &Path) -> Result<()> { "ServerRequest", ]; - for (name, schema) in bundle { - let mut schema_value = serde_json::to_value(schema)?; - annotate_schema(&mut schema_value, Some(name.as_str())); + // Merge all generated per-type JSON files (including the envelopes above) + // into a single definitions bundle. Also recognize a v2 namespace so that + // types like RateLimitSnapshot can co-exist between legacy and v2. + let json_files = json_files_in_recursive(out_dir)?; + for path in json_files { + // Skip the bundle we’re about to (re)generate. + if path.file_name().and_then(OsStr::to_str) + == Some("codex_app_server_protocol.schemas.json") + { + continue; + } + let file_stem = path + .file_stem() + .and_then(OsStr::to_str) + .ok_or_else(|| anyhow!(format!("Invalid schema file name {}", path.display())))?; + + // Determine namespace and logical type name from either the stem + // (e.g., "v2::Type") or the directory layout (e.g., ".../v2/Type.json"). + let (ns_opt, logical_name) = detect_namespace(&path, file_stem); + + let mut schema_value: Value = serde_json::from_str( + &fs::read_to_string(&path) + .with_context(|| format!("Failed to read {}", path.display()))?, + ) + .with_context(|| format!("Failed to parse JSON from {}", path.display()))?; + + // Normalize; use the original stem as the base so existing naming rules apply. + annotate_schema(&mut schema_value, Some(file_stem)); + + if let Some(ref ns) = ns_opt { + // Rewrite internal $ref targets to point to the namespaced location + // under the bundle's definitions. + rewrite_refs_to_namespace(&mut schema_value, ns); + } + + // Pull embedded definitions out and insert into our bundle map at the + // appropriate namespace. if let Value::Object(ref mut obj) = schema_value && let Some(defs) = obj.remove("definitions") && let Value::Object(defs_obj) = defs @@ -188,11 +191,37 @@ pub fn generate_json(out_dir: &Path) -> Result<()> { for (def_name, mut def_schema) in defs_obj { if !SPECIAL_DEFINITIONS.contains(&def_name.as_str()) { annotate_schema(&mut def_schema, Some(def_name.as_str())); - definitions.insert(def_name, def_schema); + if let Some(ref ns) = ns_opt { + definitions + .entry(ns.to_string()) + .or_insert_with(|| Value::Object(Map::new())); + if let Value::Object(defs_ns) = definitions + .get_mut(ns) + .ok_or_else(|| anyhow!("just inserted v2 namespace"))? + { + defs_ns.insert(def_name, def_schema); + } + } else { + definitions.insert(def_name, def_schema); + } } } } - definitions.insert(name, schema_value); + + // Insert the schema root itself into the bundle under the namespace. + if let Some(ref ns) = ns_opt { + definitions + .entry(ns.to_string()) + .or_insert_with(|| Value::Object(Map::new())); + if let Value::Object(defs_ns) = definitions + .get_mut(ns) + .ok_or_else(|| anyhow!("just inserted v2 namespace"))? + { + defs_ns.insert(logical_name.to_string(), schema_value); + } + } else { + definitions.insert(logical_name.to_string(), schema_value); + } } let mut root = Map::new(); @@ -223,7 +252,19 @@ where let schema = schema_for!(T); let mut schema_value = serde_json::to_value(schema)?; annotate_schema(&mut schema_value, Some(file_stem)); - write_pretty_json(out_dir.join(format!("{file_stem}.json")), &schema_value) + // If the name looks like a namespaced path (e.g., "v2::Type"), mirror + // the TypeScript layout and write to out_dir/v2/Type.json. Otherwise + // write alongside the legacy files. + let (ns_opt, logical_name) = split_namespace(file_stem); + let out_path = if let Some(ns) = ns_opt { + let dir = out_dir.join(ns); + ensure_dir(&dir)?; + dir.join(format!("{logical_name}.json")) + } else { + out_dir.join(format!("{file_stem}.json")) + }; + + write_pretty_json(out_path, &schema_value) .with_context(|| format!("Failed to write JSON schema for {file_stem}"))?; let annotated_schema = serde_json::from_value(schema_value)?; Ok(annotated_schema) @@ -242,13 +283,76 @@ fn write_pretty_json(path: PathBuf, value: &impl Serialize) -> Result<()> { fs::write(&path, json).with_context(|| format!("Failed to write {}", path.display()))?; Ok(()) } -fn type_basename(type_path: &str) -> String { - type_path - .rsplit_once("::") - .map(|(_, name)| name) - .unwrap_or(type_path) - .trim() - .to_string() + +/// Detect a namespace from a schema file path and stem. +/// Supports both legacy on-disk names like "v2::Type.json" and the new +/// directory layout ".../v2/Type.json". +fn detect_namespace(path: &Path, stem: &str) -> (Option, String) { + // Prefer directory-based detection. + if let Some(parent) = path.parent() + && parent.file_name().and_then(OsStr::to_str) == Some("v2") + { + return (Some("v2".to_string()), stem.to_string()); + } + split_namespace(stem) +} + +/// Split a "ns::Type" name into (Some(ns), Type). Returns (None, name) if no ns. +fn split_namespace(name: &str) -> (Option, String) { + if let Some(idx) = name.find("::") { + let (ns, rest) = name.split_at(idx); + // rest starts with "::" + let typ = &rest[2..]; + return (Some(ns.to_string()), typ.to_string()); + } + (None, name.to_string()) +} + +/// Recursively rewrite $ref values that point at "#/definitions/..." so that +/// they point to a namespaced location under the bundle. +fn rewrite_refs_to_namespace(value: &mut Value, ns: &str) { + match value { + Value::Object(obj) => { + if let Some(Value::String(r)) = obj.get_mut("$ref") + && let Some(suffix) = r.strip_prefix("#/definitions/") + { + *r = format!("#/definitions/{ns}/{suffix}"); + } + for v in obj.values_mut() { + rewrite_refs_to_namespace(v, ns); + } + } + Value::Array(items) => { + for v in items.iter_mut() { + rewrite_refs_to_namespace(v, ns); + } + } + _ => {} + } +} + +fn json_files_in_recursive(dir: &Path) -> Result> { + let mut files = Vec::new(); + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + for entry in + fs::read_dir(&d).with_context(|| format!("Failed to read dir {}", d.display()))? + { + let entry = entry?; + let path = entry.path(); + if path.is_dir() { + // Skip known non-protocol directories. + if path.file_name().and_then(OsStr::to_str) == Some("serde_json") { + continue; + } + stack.push(path); + } else if path.is_file() && path.extension() == Some(OsStr::new("json")) { + files.push(path); + } + } + } + files.sort(); + Ok(files) } fn variant_definition_name(base: &str, variant: &Value) -> Option { @@ -505,6 +609,26 @@ fn ts_files_in(dir: &Path) -> Result> { Ok(files) } +fn ts_files_in_recursive(dir: &Path) -> Result> { + let mut files = Vec::new(); + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + for entry in + fs::read_dir(&d).with_context(|| format!("Failed to read dir {}", d.display()))? + { + let entry = entry?; + let path = entry.path(); + if path.is_dir() { + stack.push(path); + } else if path.is_file() && path.extension() == Some(OsStr::new("ts")) { + files.push(path); + } + } + } + files.sort(); + Ok(files) +} + fn generate_index_ts(out_dir: &Path) -> Result { let mut entries: Vec = Vec::new(); let mut stems: Vec = ts_files_in(out_dir)? @@ -521,6 +645,14 @@ fn generate_index_ts(out_dir: &Path) -> Result { entries.push(format!("export type {{ {name} }} from \"./{name}\";\n")); } + // If this is the root out_dir and a ./v2 folder exists with TS files, + // expose it as a namespace to avoid symbol collisions at the root. + let v2_dir = out_dir.join("v2"); + let has_v2_ts = ts_files_in(&v2_dir).map(|v| !v.is_empty()).unwrap_or(false); + if has_v2_ts { + entries.push("export * as v2 from \"./v2\";\n".to_string()); + } + let mut content = String::with_capacity(HEADER.len() + entries.iter().map(String::len).sum::()); content.push_str(HEADER); @@ -547,6 +679,7 @@ mod tests { #[test] fn generated_ts_has_no_optional_nullable_fields() -> Result<()> { + // Assert that there are no types of the form "?: T | null" in the generated TS files. let output_dir = std::env::temp_dir().join(format!("codex_ts_types_{}", Uuid::now_v7())); fs::create_dir(&output_dir)?; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index efc60830c2..8227f5f41b 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -9,7 +9,6 @@ use crate::protocol::v2; use codex_protocol::ConversationId; use codex_protocol::parse_command::ParsedCommand; use codex_protocol::protocol::FileChange; -use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SandboxCommandAssessment; use paste::paste; @@ -82,16 +81,71 @@ macro_rules! client_request_definitions { )* Ok(()) } + + pub fn export_client_param_schemas( + out_dir: &::std::path::Path, + ) -> ::anyhow::Result<()> { + $( + crate::export::write_json_schema::<$params>(out_dir, stringify!($params))?; + )* + Ok(()) + } }; } client_request_definitions! { /// NEW APIs + // Thread lifecycle + #[serde(rename = "thread/start")] + #[ts(rename = "thread/start")] + ThreadStart { + params: v2::ThreadStartParams, + response: v2::ThreadStartResponse, + }, + #[serde(rename = "thread/resume")] + #[ts(rename = "thread/resume")] + ThreadResume { + params: v2::ThreadResumeParams, + response: v2::ThreadResumeResponse, + }, + #[serde(rename = "thread/archive")] + #[ts(rename = "thread/archive")] + ThreadArchive { + params: v2::ThreadArchiveParams, + response: v2::ThreadArchiveResponse, + }, + #[serde(rename = "thread/list")] + #[ts(rename = "thread/list")] + ThreadList { + params: v2::ThreadListParams, + response: v2::ThreadListResponse, + }, + #[serde(rename = "thread/compact")] + #[ts(rename = "thread/compact")] + ThreadCompact { + params: v2::ThreadCompactParams, + response: v2::ThreadCompactResponse, + }, + + // Turn lifecycle + #[serde(rename = "turn/start")] + #[ts(rename = "turn/start")] + TurnStart { + params: v2::TurnStartParams, + response: v2::TurnStartResponse, + }, + #[serde(rename = "turn/interrupt")] + #[ts(rename = "turn/interrupt")] + TurnInterrupt { + params: v2::TurnInterruptParams, + response: v2::TurnInterruptResponse, + }, + #[serde(rename = "model/list")] #[ts(rename = "model/list")] - ListModels { - params: v2::ListModelsParams, - response: v2::ListModelsResponse, + ModelList { + params: v2::ModelListParams, + response: v2::ModelListResponse, }, #[serde(rename = "account/login")] @@ -117,9 +171,9 @@ client_request_definitions! { #[serde(rename = "feedback/upload")] #[ts(rename = "feedback/upload")] - UploadFeedback { - params: v2::UploadFeedbackParams, - response: v2::UploadFeedbackResponse, + FeedbackUpload { + params: v2::FeedbackUploadParams, + response: v2::FeedbackUploadResponse, }, #[serde(rename = "account/read")] @@ -284,6 +338,92 @@ macro_rules! server_request_definitions { } Ok(()) } + + pub fn export_server_param_schemas( + out_dir: &::std::path::Path, + ) -> ::anyhow::Result<()> { + paste! { + $(crate::export::write_json_schema::<[<$variant Params>]>(out_dir, stringify!([<$variant Params>]))?;)* + } + Ok(()) + } + }; +} + +/// Generates `ServerNotification` enum and helpers, including a JSON Schema +/// exporter for each notification. +macro_rules! server_notification_definitions { + ( + $( + $(#[$variant_meta:meta])* + $variant:ident ( $payload:ty ) + ),* $(,)? + ) => { + /// Notification sent from the server to the client. + #[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS, Display)] + #[serde(tag = "method", content = "params", rename_all = "camelCase")] + #[strum(serialize_all = "camelCase")] + pub enum ServerNotification { + $( + $(#[$variant_meta])* + $variant($payload), + )* + } + + impl ServerNotification { + pub fn to_params(self) -> Result { + match self { + $(Self::$variant(params) => serde_json::to_value(params),)* + } + } + } + + impl TryFrom for ServerNotification { + type Error = serde_json::Error; + + fn try_from(value: JSONRPCNotification) -> Result { + serde_json::from_value(serde_json::to_value(value)?) + } + } + + pub fn export_server_notification_schemas( + _out_dir: &::std::path::Path, + ) -> ::anyhow::Result<()> { + $( + crate::export::write_json_schema::<$payload>(_out_dir, stringify!($payload))?; + )* + Ok(()) + } + }; +} + +/// Notifications sent from the client to the server. +/// Generates `ClientNotification` enum and a JSON Schema exporter. +macro_rules! client_notification_definitions { + ( + $( + $(#[$variant_meta:meta])* + $variant:ident $( ( $payload:ty ) )? + ),* $(,)? + ) => { + #[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS, Display)] + #[serde(tag = "method", content = "params", rename_all = "camelCase")] + #[strum(serialize_all = "camelCase")] + pub enum ClientNotification { + $( + $(#[$variant_meta])* + $variant $( ( $payload ) )? , + )* + } + + pub fn export_client_notification_schemas( + _out_dir: &::std::path::Path, + ) -> ::anyhow::Result<()> { + $( + $(crate::export::write_json_schema::<$payload>(_out_dir, stringify!($payload))?;)? + )* + Ok(()) + } }; } @@ -366,12 +506,48 @@ pub struct FuzzyFileSearchResponse { pub files: Vec, } -/// Notification sent from the server to the client. -#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS, Display)] -#[serde(tag = "method", content = "params", rename_all = "camelCase")] -#[strum(serialize_all = "camelCase")] -pub enum ServerNotification { +server_notification_definitions! { /// NEW NOTIFICATIONS + #[serde(rename = "thread/started")] + #[ts(rename = "thread/started")] + #[strum(serialize = "thread/started")] + ThreadStarted(v2::ThreadStartedNotification), + + #[serde(rename = "turn/started")] + #[ts(rename = "turn/started")] + #[strum(serialize = "turn/started")] + TurnStarted(v2::TurnStartedNotification), + + #[serde(rename = "turn/completed")] + #[ts(rename = "turn/completed")] + #[strum(serialize = "turn/completed")] + TurnCompleted(v2::TurnCompletedNotification), + + #[serde(rename = "item/started")] + #[ts(rename = "item/started")] + #[strum(serialize = "item/started")] + ItemStarted(v2::ItemStartedNotification), + + #[serde(rename = "item/completed")] + #[ts(rename = "item/completed")] + #[strum(serialize = "item/completed")] + ItemCompleted(v2::ItemCompletedNotification), + + #[serde(rename = "item/agentMessage/delta")] + #[ts(rename = "item/agentMessage/delta")] + #[strum(serialize = "item/agentMessage/delta")] + AgentMessageDelta(v2::AgentMessageDeltaNotification), + + #[serde(rename = "item/commandExecution/outputDelta")] + #[ts(rename = "item/commandExecution/outputDelta")] + #[strum(serialize = "item/commandExecution/outputDelta")] + CommandExecutionOutputDelta(v2::CommandExecutionOutputDeltaNotification), + + #[serde(rename = "item/mcpToolCall/progress")] + #[ts(rename = "item/mcpToolCall/progress")] + #[strum(serialize = "item/mcpToolCall/progress")] + McpToolCallProgress(v2::McpToolCallProgressNotification), + #[serde(rename = "account/updated")] #[ts(rename = "account/updated")] #[strum(serialize = "account/updated")] @@ -380,7 +556,7 @@ pub enum ServerNotification { #[serde(rename = "account/rateLimits/updated")] #[ts(rename = "account/rateLimits/updated")] #[strum(serialize = "account/rateLimits/updated")] - AccountRateLimitsUpdated(RateLimitSnapshot), + AccountRateLimitsUpdated(v2::AccountRateLimitsUpdatedNotification), /// DEPRECATED NOTIFICATIONS below /// Authentication status changed @@ -393,31 +569,7 @@ pub enum ServerNotification { SessionConfigured(v1::SessionConfiguredNotification), } -impl ServerNotification { - pub fn to_params(self) -> Result { - match self { - ServerNotification::AccountUpdated(params) => serde_json::to_value(params), - ServerNotification::AccountRateLimitsUpdated(params) => serde_json::to_value(params), - ServerNotification::AuthStatusChange(params) => serde_json::to_value(params), - ServerNotification::LoginChatGptComplete(params) => serde_json::to_value(params), - ServerNotification::SessionConfigured(params) => serde_json::to_value(params), - } - } -} - -impl TryFrom for ServerNotification { - type Error = serde_json::Error; - - fn try_from(value: JSONRPCNotification) -> Result { - serde_json::from_value(serde_json::to_value(value)?) - } -} - -/// Notification sent from the client to the server. -#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS, Display)] -#[serde(tag = "method", content = "params", rename_all = "camelCase")] -#[strum(serialize_all = "camelCase")] -pub enum ClientNotification { +client_notification_definitions! { Initialized, } @@ -671,16 +823,16 @@ mod tests { #[test] fn serialize_list_models() -> Result<()> { - let request = ClientRequest::ListModels { + let request = ClientRequest::ModelList { request_id: RequestId::Integer(6), - params: v2::ListModelsParams::default(), + params: v2::ModelListParams::default(), }; assert_eq!( json!({ "method": "model/list", "id": 6, "params": { - "pageSize": null, + "limit": null, "cursor": null } }), diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 8c39628412..bab28f2b79 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1,17 +1,123 @@ +use std::collections::HashMap; +use std::path::PathBuf; + use crate::protocol::common::AuthMode; use codex_protocol::ConversationId; use codex_protocol::account::PlanType; use codex_protocol::config_types::ReasoningEffort; -use codex_protocol::protocol::RateLimitSnapshot; +use codex_protocol::config_types::ReasoningSummary; +use mcp_types::ContentBlock as McpContentBlock; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; +use serde_json::Value as JsonValue; use ts_rs::TS; use uuid::Uuid; +// Macro to declare a camelCased API v2 enum mirroring a core enum which +// tends to use kebab-case. +macro_rules! v2_enum_from_core { + ( + pub enum $Name:ident from $Src:path { $( $Variant:ident ),+ $(,)? } + ) => { + #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] + #[serde(rename_all = "camelCase")] + #[ts(export_to = "v2/")] + pub enum $Name { $( $Variant ),+ } + + impl $Name { + pub fn to_core(self) -> $Src { + match self { $( $Name::$Variant => <$Src>::$Variant ),+ } + } + } + + impl From<$Src> for $Name { + fn from(value: $Src) -> Self { + match value { $( <$Src>::$Variant => $Name::$Variant ),+ } + } + } + }; +} + +v2_enum_from_core!( + pub enum AskForApproval from codex_protocol::protocol::AskForApproval { + UnlessTrusted, OnFailure, OnRequest, Never + } +); + +v2_enum_from_core!( + pub enum SandboxMode from codex_protocol::config_types::SandboxMode { + ReadOnly, WorkspaceWrite, DangerFullAccess + } +); + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(tag = "mode", rename_all = "camelCase")] +#[ts(tag = "mode")] +#[ts(export_to = "v2/")] +pub enum SandboxPolicy { + DangerFullAccess, + ReadOnly, + WorkspaceWrite { + #[serde(default, skip_serializing_if = "Vec::is_empty")] + writable_roots: Vec, + #[serde(default)] + network_access: bool, + #[serde(default)] + exclude_tmpdir_env_var: bool, + #[serde(default)] + exclude_slash_tmp: bool, + }, +} + +impl SandboxPolicy { + pub fn to_core(&self) -> codex_protocol::protocol::SandboxPolicy { + match self { + SandboxPolicy::DangerFullAccess => { + codex_protocol::protocol::SandboxPolicy::DangerFullAccess + } + SandboxPolicy::ReadOnly => codex_protocol::protocol::SandboxPolicy::ReadOnly, + SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + } => codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { + writable_roots: writable_roots.clone(), + network_access: *network_access, + exclude_tmpdir_env_var: *exclude_tmpdir_env_var, + exclude_slash_tmp: *exclude_slash_tmp, + }, + } + } +} + +impl From for SandboxPolicy { + fn from(value: codex_protocol::protocol::SandboxPolicy) -> Self { + match value { + codex_protocol::protocol::SandboxPolicy::DangerFullAccess => { + SandboxPolicy::DangerFullAccess + } + codex_protocol::protocol::SandboxPolicy::ReadOnly => SandboxPolicy::ReadOnly, + codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + } => SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + }, + } + } +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(tag = "type", rename_all = "camelCase")] #[ts(tag = "type")] +#[ts(export_to = "v2/")] pub enum Account { #[serde(rename = "apiKey", rename_all = "camelCase")] #[ts(rename = "apiKey", rename_all = "camelCase")] @@ -28,6 +134,7 @@ pub enum Account { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(tag = "type")] #[ts(tag = "type")] +#[ts(export_to = "v2/")] pub enum LoginAccountParams { #[serde(rename = "apiKey")] #[ts(rename = "apiKey")] @@ -43,6 +150,7 @@ pub enum LoginAccountParams { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct LoginAccountResponse { /// Only set if the login method is ChatGPT. #[schemars(with = "String")] @@ -55,31 +163,36 @@ pub struct LoginAccountResponse { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct LogoutAccountResponse {} #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct GetAccountRateLimitsResponse { pub rate_limits: RateLimitSnapshot, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct GetAccountResponse { pub account: Account, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)] #[serde(rename_all = "camelCase")] -pub struct ListModelsParams { - /// Optional page size; defaults to a reasonable server-side value. - pub page_size: Option, +#[ts(export_to = "v2/")] +pub struct ModelListParams { /// Opaque pagination cursor returned by a previous call. pub cursor: Option, + /// Optional page size; defaults to a reasonable server-side value. + pub limit: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct Model { pub id: String, pub model: String, @@ -93,6 +206,7 @@ pub struct Model { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] pub struct ReasoningEffortOption { pub reasoning_effort: ReasoningEffort, pub description: String, @@ -100,8 +214,9 @@ pub struct ReasoningEffortOption { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] -pub struct ListModelsResponse { - pub items: Vec, +#[ts(export_to = "v2/")] +pub struct ModelListResponse { + pub data: Vec, /// Opaque cursor to pass to the next call to continue after the last item. /// if None, there are no more items to return. pub next_cursor: Option, @@ -109,7 +224,8 @@ pub struct ListModelsResponse { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] -pub struct UploadFeedbackParams { +#[ts(export_to = "v2/")] +pub struct FeedbackUploadParams { pub classification: String, pub reason: Option, pub conversation_id: Option, @@ -118,10 +234,428 @@ pub struct UploadFeedbackParams { #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] -pub struct UploadFeedbackResponse { +#[ts(export_to = "v2/")] +pub struct FeedbackUploadResponse { + pub thread_id: String, +} + +// === Threads, Turns, and Items === +// Thread APIs +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadStartParams { + pub model: Option, + pub model_provider: Option, + pub cwd: Option, + pub approval_policy: Option, + pub sandbox: Option, + pub config: Option>, + pub base_instructions: Option, + pub developer_instructions: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadStartResponse { + pub thread: Thread, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadResumeParams { pub thread_id: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadResumeResponse { + pub thread: Thread, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadArchiveParams { + pub thread_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadArchiveResponse {} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadListParams { + /// Opaque pagination cursor returned by a previous call. + pub cursor: Option, + /// Optional page size; defaults to a reasonable server-side value. + pub limit: Option, + /// Optional provider filter; when set, only sessions recorded under these + /// providers are returned. When present but empty, includes all providers. + pub model_providers: Option>, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadListResponse { + pub data: Vec, + /// Opaque cursor to pass to the next call to continue after the last item. + /// if None, there are no more items to return. + pub next_cursor: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadCompactParams { + pub thread_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadCompactResponse {} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct Thread { + pub id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct Turn { + pub id: String, + pub items: Vec, + pub status: TurnStatus, + pub error: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnError { + pub message: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum TurnStatus { + Completed, + Interrupted, + Failed, + InProgress, +} + +// Turn APIs + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnStartParams { + pub thread_id: String, + pub input: Vec, + /// Override the working directory for this turn and subsequent turns. + pub cwd: Option, + /// Override the approval policy for this turn and subsequent turns. + pub approval_policy: Option, + /// Override the sandbox policy for this turn and subsequent turns. + pub sandbox_policy: Option, + /// Override the model for this turn and subsequent turns. + pub model: Option, + /// Override the reasoning effort for this turn and subsequent turns. + pub effort: Option, + /// Override the reasoning summary for this turn and subsequent turns. + pub summary: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnStartResponse { + pub turn: Turn, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnInterruptParams { + pub thread_id: String, + pub turn_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnInterruptResponse {} + +// User input types +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type")] +#[ts(export_to = "v2/")] +pub enum UserInput { + Text { text: String }, + Image { url: String }, + LocalImage { path: PathBuf }, +} + +// Thread items + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type")] +#[ts(export_to = "v2/")] +pub enum ThreadItem { + UserMessage { + id: String, + content: Vec, + }, + AgentMessage { + id: String, + text: String, + }, + Reasoning { + id: String, + text: String, + }, + CommandExecution { + id: String, + command: String, + aggregated_output: String, + exit_code: Option, + status: CommandExecutionStatus, + duration_ms: Option, + }, + FileChange { + id: String, + changes: Vec, + status: PatchApplyStatus, + }, + McpToolCall { + id: String, + server: String, + tool: String, + status: McpToolCallStatus, + arguments: JsonValue, + result: Option, + error: Option, + }, + WebSearch { + id: String, + query: String, + }, + TodoList { + id: String, + items: Vec, + }, + ImageView { + id: String, + path: String, + }, + CodeReview { + id: String, + review: String, + }, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum CommandExecutionStatus { + InProgress, + Completed, + Failed, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct FileUpdateChange { + pub path: String, + pub kind: PatchChangeKind, + pub diff: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum PatchChangeKind { + Add, + Delete, + Update, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum PatchApplyStatus { + Completed, + Failed, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum McpToolCallStatus { + InProgress, + Completed, + Failed, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct McpToolCallResult { + pub content: Vec, + pub structured_content: JsonValue, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct McpToolCallError { + pub message: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TodoItem { + pub id: String, + pub text: String, + pub completed: bool, +} + +// === Server Notifications === +// Thread/Turn lifecycle notifications and item progress events +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadStartedNotification { + pub thread: Thread, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnStartedNotification { + pub turn: Turn, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct Usage { + pub input_tokens: i32, + pub cached_input_tokens: i32, + pub output_tokens: i32, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct TurnCompletedNotification { + pub turn: Turn, + // TODO: should usage be stored on the Turn object, and we return that instead? + pub usage: Usage, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ItemStartedNotification { + pub item: ThreadItem, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ItemCompletedNotification { + pub item: ThreadItem, +} + +// Item-specific progress notifications +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct AgentMessageDeltaNotification { + pub item_id: String, + pub delta: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct CommandExecutionOutputDeltaNotification { + pub item_id: String, + pub delta: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct McpToolCallProgressNotification { + pub item_id: String, + pub message: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct AccountRateLimitsUpdatedNotification { + pub rate_limits: RateLimitSnapshot, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RateLimitSnapshot { + pub primary: Option, + pub secondary: Option, +} + +impl From for RateLimitSnapshot { + fn from(value: codex_protocol::protocol::RateLimitSnapshot) -> Self { + Self { + primary: value.primary.map(RateLimitWindow::from), + secondary: value.secondary.map(RateLimitWindow::from), + } + } +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RateLimitWindow { + /// Percentage (0-100) of the window that has been consumed. + pub used_percent: i64, + /// Rolling window duration, in minutes. + #[ts(type = "number | null")] + pub window_duration_mins: Option, + /// Unix timestamp (seconds since epoch) when the window resets. + #[ts(type = "number | null")] + pub resets_at: Option, +} + +impl From for RateLimitWindow { + fn from(value: codex_protocol::protocol::RateLimitWindow) -> Self { + Self { + used_percent: value.used_percent.round() as i64, + window_duration_mins: value.window_minutes, + resets_at: value.resets_at, + } + } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] pub struct AccountUpdatedNotification { diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 8ea4cef8ba..20cf57c0e4 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -4,6 +4,7 @@ use crate::fuzzy_file_search::run_fuzzy_file_search; use crate::models::supported_models; use crate::outgoing_message::OutgoingMessageSender; use crate::outgoing_message::OutgoingNotification; +use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AccountUpdatedNotification; use codex_app_server_protocol::AddConversationListenerParams; use codex_app_server_protocol::AddConversationSubscriptionResponse; @@ -19,6 +20,8 @@ use codex_app_server_protocol::ExecCommandApprovalParams; use codex_app_server_protocol::ExecCommandApprovalResponse; use codex_app_server_protocol::ExecOneOffCommandParams; use codex_app_server_protocol::ExecOneOffCommandResponse; +use codex_app_server_protocol::FeedbackUploadParams; +use codex_app_server_protocol::FeedbackUploadResponse; use codex_app_server_protocol::FuzzyFileSearchParams; use codex_app_server_protocol::FuzzyFileSearchResponse; use codex_app_server_protocol::GetAccountRateLimitsResponse; @@ -33,14 +36,15 @@ use codex_app_server_protocol::InterruptConversationResponse; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::ListConversationsParams; use codex_app_server_protocol::ListConversationsResponse; -use codex_app_server_protocol::ListModelsParams; -use codex_app_server_protocol::ListModelsResponse; use codex_app_server_protocol::LoginApiKeyParams; use codex_app_server_protocol::LoginApiKeyResponse; use codex_app_server_protocol::LoginChatGptCompleteNotification; use codex_app_server_protocol::LoginChatGptResponse; +use codex_app_server_protocol::ModelListParams; +use codex_app_server_protocol::ModelListResponse; use codex_app_server_protocol::NewConversationParams; use codex_app_server_protocol::NewConversationResponse; +use codex_app_server_protocol::RateLimitSnapshot as V2RateLimitSnapshot; use codex_app_server_protocol::RemoveConversationListenerParams; use codex_app_server_protocol::RemoveConversationSubscriptionResponse; use codex_app_server_protocol::RequestId; @@ -55,9 +59,18 @@ use codex_app_server_protocol::ServerRequestPayload; use codex_app_server_protocol::SessionConfiguredNotification; use codex_app_server_protocol::SetDefaultModelParams; use codex_app_server_protocol::SetDefaultModelResponse; -use codex_app_server_protocol::UploadFeedbackParams; -use codex_app_server_protocol::UploadFeedbackResponse; +use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadListResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStartedNotification; use codex_app_server_protocol::UserInfoResponse; +use codex_app_server_protocol::UserInput as V2UserInput; use codex_app_server_protocol::UserSavedConfig; use codex_backend_client::Client as BackendClient; use codex_core::AuthManager; @@ -97,7 +110,7 @@ use codex_protocol::ConversationId; use codex_protocol::config_types::ForcedLoginMethod; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::RateLimitSnapshot; +use codex_protocol::protocol::RateLimitSnapshot as CoreRateLimitSnapshot; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::user_input::UserInput as CoreInputItem; @@ -119,6 +132,9 @@ use tracing::info; use tracing::warn; use uuid::Uuid; +type PendingInterruptQueue = Vec<(RequestId, ApiVersion)>; +type PendingInterrupts = Arc>>; + // Duration before a ChatGPT login attempt is abandoned. const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60); struct ActiveLogin { @@ -142,11 +158,17 @@ pub(crate) struct CodexMessageProcessor { conversation_listeners: HashMap>, active_login: Arc>>, // Queue of pending interrupt requests per conversation. We reply when TurnAborted arrives. - pending_interrupts: Arc>>>, + pending_interrupts: PendingInterrupts, pending_fuzzy_searches: Arc>>>, feedback: CodexFeedback, } +#[derive(Clone, Copy, Debug)] +enum ApiVersion { + V1, + V2, +} + impl CodexMessageProcessor { pub fn new( auth_manager: Arc, @@ -175,6 +197,32 @@ impl CodexMessageProcessor { ClientRequest::Initialize { .. } => { panic!("Initialize should be handled in MessageProcessor"); } + // === v2 Thread/Turn APIs === + ClientRequest::ThreadStart { request_id, params } => { + self.thread_start(request_id, params).await; + } + ClientRequest::ThreadResume { request_id, params } => { + self.thread_resume(request_id, params).await; + } + ClientRequest::ThreadArchive { request_id, params } => { + self.thread_archive(request_id, params).await; + } + ClientRequest::ThreadList { request_id, params } => { + self.thread_list(request_id, params).await; + } + ClientRequest::ThreadCompact { + request_id, + params: _, + } => { + self.send_unimplemented_error(request_id, "thread/compact") + .await; + } + ClientRequest::TurnStart { request_id, params } => { + self.turn_start(request_id, params).await; + } + ClientRequest::TurnInterrupt { request_id, params } => { + self.turn_interrupt(request_id, params).await; + } ClientRequest::NewConversation { request_id, params } => { // Do not tokio::spawn() to process new_conversation() // asynchronously because we need to ensure the conversation is @@ -187,7 +235,7 @@ impl CodexMessageProcessor { ClientRequest::ListConversations { request_id, params } => { self.handle_list_conversations(request_id, params).await; } - ClientRequest::ListModels { request_id, params } => { + ClientRequest::ModelList { request_id, params } => { self.list_models(request_id, params).await; } ClientRequest::LoginAccount { @@ -288,7 +336,7 @@ impl CodexMessageProcessor { } => { self.get_account_rate_limits(request_id).await; } - ClientRequest::UploadFeedback { request_id, params } => { + ClientRequest::FeedbackUpload { request_id, params } => { self.upload_feedback(request_id, params).await; } } @@ -634,7 +682,7 @@ impl CodexMessageProcessor { } } - async fn fetch_account_rate_limits(&self) -> Result { + async fn fetch_account_rate_limits(&self) -> Result { let Some(auth) = self.auth_manager.auth() else { return Err(JSONRPCErrorError { code: INVALID_REQUEST_ERROR_CODE, @@ -662,6 +710,7 @@ impl CodexMessageProcessor { client .get_rate_limits() .await + .map(|rl: CoreRateLimitSnapshot| rl.into()) .map_err(|err| JSONRPCErrorError { code: INTERNAL_ERROR_CODE, message: format!("failed to fetch codex rate limits: {err}"), @@ -813,19 +862,51 @@ impl CodexMessageProcessor { } async fn process_new_conversation(&self, request_id: RequestId, params: NewConversationParams) { - let config = - match derive_config_from_params(params, self.codex_linux_sandbox_exe.clone()).await { - Ok(config) => config, - Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("error deriving config: {err}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - }; + let NewConversationParams { + model, + model_provider, + profile, + cwd, + approval_policy, + sandbox: sandbox_mode, + config: cli_overrides, + base_instructions, + developer_instructions, + compact_prompt, + include_apply_patch_tool, + } = params; + + let overrides = ConfigOverrides { + model, + review_model: None, + config_profile: profile, + cwd: cwd.map(PathBuf::from), + approval_policy, + sandbox_mode, + model_provider, + codex_linux_sandbox_exe: self.codex_linux_sandbox_exe.clone(), + base_instructions, + developer_instructions, + compact_prompt, + include_apply_patch_tool, + show_raw_agent_reasoning: None, + tools_web_search_request: None, + experimental_sandbox_command_assessment: None, + additional_writable_roots: Vec::new(), + }; + + let config = match derive_config_from_params(overrides, cli_overrides).await { + Ok(config) => config, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("error deriving config: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; match self.conversation_manager.new_conversation(config).await { Ok(conversation_id) => { @@ -853,6 +934,528 @@ impl CodexMessageProcessor { } } + async fn thread_start(&mut self, request_id: RequestId, params: ThreadStartParams) { + // Build ConfigOverrides directly from ThreadStartParams for config derivation. + let cli_overrides = params.config; + let overrides = ConfigOverrides { + model: params.model, + review_model: None, + config_profile: None, + cwd: params.cwd.map(PathBuf::from), + approval_policy: params + .approval_policy + .map(codex_app_server_protocol::AskForApproval::to_core), + sandbox_mode: params + .sandbox + .map(codex_app_server_protocol::SandboxMode::to_core), + model_provider: params.model_provider, + codex_linux_sandbox_exe: self.codex_linux_sandbox_exe.clone(), + base_instructions: params.base_instructions, + developer_instructions: params.developer_instructions, + compact_prompt: None, + include_apply_patch_tool: None, + show_raw_agent_reasoning: None, + tools_web_search_request: None, + experimental_sandbox_command_assessment: None, + additional_writable_roots: Vec::new(), + }; + + let config = match derive_config_from_params(overrides, cli_overrides).await { + Ok(config) => config, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("error deriving config: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + match self.conversation_manager.new_conversation(config).await { + Ok(new_conv) => { + let thread = Thread { + id: new_conv.conversation_id.to_string(), + }; + + // Send response first to complete the RPC. + let response = ThreadStartResponse { + thread: thread.clone(), + }; + self.outgoing.send_response(request_id, response).await; + + // Emit thread/started notification for v2 stream consumers. + let notif = ThreadStartedNotification { thread }; + self.outgoing + .send_server_notification(ServerNotification::ThreadStarted(notif)) + .await; + + // Auto-attach a conversation listener when starting a thread. + // Use the same behavior as the v1 API with experimental_raw_events=false. + if let Err(err) = self + .attach_conversation_listener(new_conv.conversation_id, false) + .await + { + tracing::warn!( + "failed to attach listener for conversation {}: {}", + new_conv.conversation_id, + err.message + ); + } + } + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("error creating thread: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + } + } + } + + async fn turn_start( + &self, + request_id: RequestId, + params: codex_app_server_protocol::TurnStartParams, + ) { + // Resolve conversation id from v2 thread id string. + let conversation_id = match ConversationId::from_string(¶ms.thread_id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let Ok(conversation) = self + .conversation_manager + .get_conversation(conversation_id) + .await + else { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("conversation not found: {conversation_id}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + }; + + // Keep a copy of v2 inputs for the notification payload. + let v2_inputs_for_notif = params.input.clone(); + + // Map v2 input items to core input items. + let mapped_items: Vec = params + .input + .into_iter() + .map(|item| match item { + V2UserInput::Text { text } => CoreInputItem::Text { text }, + V2UserInput::Image { url } => CoreInputItem::Image { image_url: url }, + V2UserInput::LocalImage { path } => CoreInputItem::LocalImage { path }, + }) + .collect(); + + let has_any_overrides = params.cwd.is_some() + || params.approval_policy.is_some() + || params.sandbox_policy.is_some() + || params.model.is_some() + || params.effort.is_some() + || params.summary.is_some(); + + // If any overrides are provided, update the session turn context first. + if has_any_overrides { + let _ = conversation + .submit(Op::OverrideTurnContext { + cwd: params.cwd, + approval_policy: params + .approval_policy + .map(codex_app_server_protocol::AskForApproval::to_core), + sandbox_policy: params.sandbox_policy.map(|p| p.to_core()), + model: params.model, + effort: params.effort.map(Some), + summary: params.summary, + }) + .await; + } + + // Start the turn by submitting the user input. Return its submission id as turn_id. + let turn_id = conversation + .submit(Op::UserInput { + items: mapped_items, + }) + .await; + + match turn_id { + Ok(turn_id) => { + let turn = codex_app_server_protocol::Turn { + id: turn_id.clone(), + items: vec![codex_app_server_protocol::ThreadItem::UserMessage { + id: turn_id, + content: v2_inputs_for_notif, + }], + status: codex_app_server_protocol::TurnStatus::InProgress, + error: None, + }; + + let response = codex_app_server_protocol::TurnStartResponse { turn: turn.clone() }; + self.outgoing.send_response(request_id, response).await; + + // Emit v2 turn/started notification. + let notif = codex_app_server_protocol::TurnStartedNotification { turn }; + self.outgoing + .send_server_notification(ServerNotification::TurnStarted(notif)) + .await; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to start turn: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + } + } + } + + async fn thread_archive(&mut self, request_id: RequestId, params: ThreadArchiveParams) { + // Resolve conversation id from v2 thread id string. + let conversation_id = match ConversationId::from_string(¶ms.thread_id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + // Locate rollout path from conversation id. + let rollout_path = match find_conversation_path_by_id_str( + &self.config.codex_home, + &conversation_id.to_string(), + ) + .await + { + Ok(Some(p)) => p, + Ok(None) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("no rollout found for conversation id {conversation_id}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("failed to locate conversation id {conversation_id}: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + match self + .archive_conversation_common(conversation_id, &rollout_path) + .await + { + Ok(()) => { + let response = ThreadArchiveResponse {}; + self.outgoing.send_response(request_id, response).await; + } + Err(err) => { + self.outgoing.send_error(request_id, err).await; + } + } + } + + async fn archive_conversation_common( + &mut self, + conversation_id: ConversationId, + rollout_path: &Path, + ) -> Result<(), JSONRPCErrorError> { + // Verify rollout_path is under sessions dir. + let rollout_folder = self.config.codex_home.join(codex_core::SESSIONS_SUBDIR); + + let canonical_sessions_dir = match tokio::fs::canonicalize(&rollout_folder).await { + Ok(path) => path, + Err(err) => { + return Err(JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!( + "failed to archive conversation: unable to resolve sessions directory: {err}" + ), + data: None, + }); + } + }; + let canonical_rollout_path = tokio::fs::canonicalize(rollout_path).await; + let canonical_rollout_path = if let Ok(path) = canonical_rollout_path + && path.starts_with(&canonical_sessions_dir) + { + path + } else { + return Err(JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` must be in sessions directory", + rollout_path.display() + ), + data: None, + }); + }; + + // Verify file name matches conversation id. + let required_suffix = format!("{conversation_id}.jsonl"); + let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else { + return Err(JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` missing file name", + rollout_path.display() + ), + data: None, + }); + }; + if !file_name + .to_string_lossy() + .ends_with(required_suffix.as_str()) + { + return Err(JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "rollout path `{}` does not match conversation id {conversation_id}", + rollout_path.display() + ), + data: None, + }); + } + + // If the conversation is active, request shutdown and wait briefly. + if let Some(conversation) = self + .conversation_manager + .remove_conversation(&conversation_id) + .await + { + info!("conversation {conversation_id} was active; shutting down"); + let conversation_clone = conversation.clone(); + let notify = Arc::new(tokio::sync::Notify::new()); + let notify_clone = notify.clone(); + let is_shutdown = tokio::spawn(async move { + // Create the notified future outside the loop to avoid losing notifications. + let notified = notify_clone.notified(); + tokio::pin!(notified); + loop { + select! { + _ = &mut notified => { break; } + event = conversation_clone.next_event() => { + match event { + Ok(event) => { + if matches!(event.msg, EventMsg::ShutdownComplete) { break; } + } + // Break on errors to avoid tight loops when the agent loop has exited. + Err(_) => { break; } + } + } + } + } + }); + match conversation.submit(Op::Shutdown).await { + Ok(_) => { + select! { + _ = is_shutdown => {} + _ = tokio::time::sleep(Duration::from_secs(10)) => { + warn!("conversation {conversation_id} shutdown timed out; proceeding with archive"); + // Wake any waiter; use notify_waiters to avoid missing the signal. + notify.notify_waiters(); + } + } + } + Err(err) => { + error!("failed to submit Shutdown to conversation {conversation_id}: {err}"); + notify.notify_waiters(); + } + } + } + + // Move the rollout file to archived. + let result: std::io::Result<()> = async { + let archive_folder = self + .config + .codex_home + .join(codex_core::ARCHIVED_SESSIONS_SUBDIR); + tokio::fs::create_dir_all(&archive_folder).await?; + tokio::fs::rename(&canonical_rollout_path, &archive_folder.join(&file_name)).await?; + Ok(()) + } + .await; + + result.map_err(|err| JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to archive conversation: {err}"), + data: None, + }) + } + + async fn thread_list(&self, request_id: RequestId, params: ThreadListParams) { + let ThreadListParams { + cursor, + limit, + model_providers, + } = params; + + let page_size = limit.unwrap_or(25).max(1) as usize; + + // Reuse canonical legacy listing logic and adapt results to v2 types. + let (summaries, next_cursor) = match self + .list_conversations_common(page_size, cursor, model_providers) + .await + { + Ok(r) => r, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let data = summaries + .into_iter() + .map(|s| Thread { + id: s.conversation_id.to_string(), + }) + .collect(); + + let response = ThreadListResponse { data, next_cursor }; + self.outgoing.send_response(request_id, response).await; + } + + async fn thread_resume(&mut self, request_id: RequestId, params: ThreadResumeParams) { + // Convert thread id string into ConversationId. + let conversation_id = match ConversationId::from_string(¶ms.thread_id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + // Locate the rollout path for the conversation id. + let path = match find_conversation_path_by_id_str( + &self.config.codex_home, + &conversation_id.to_string(), + ) + .await + { + Ok(Some(p)) => p, + Ok(None) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("no rollout found for conversation id {conversation_id}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("failed to locate conversation id {conversation_id}: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + // Read initial history from the rollout and resume the conversation. + let fallback_provider = self.config.model_provider_id.as_str(); + let summary = match read_summary_from_rollout(&path, fallback_provider).await { + Ok(s) => s, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("failed to load rollout `{}`: {err}", path.display()), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let initial_history = match RolloutRecorder::get_rollout_history(&summary.path).await { + Ok(initial_history) => initial_history, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!( + "failed to load rollout `{}` for conversation {conversation_id}: {err}", + summary.path.display() + ), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + match self + .conversation_manager + .resume_conversation_with_history( + self.config.as_ref().clone(), + initial_history, + self.auth_manager.clone(), + ) + .await + { + Ok(_) => { + // Return the same thread id and an empty turn list for now. + let response = ThreadResumeResponse { + thread: Thread { + id: conversation_id.to_string(), + }, + }; + self.outgoing.send_response(request_id, response).await; + + // Auto-attach a conversation listener when resuming a thread. + if let Err(err) = self + .attach_conversation_listener(conversation_id, false) + .await + { + tracing::warn!( + "failed to attach listener for conversation {}: {}", + conversation_id, + err.message + ); + } + } + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("error resuming thread: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + } + } + } + async fn get_conversation_summary( &self, request_id: RequestId, @@ -919,16 +1522,40 @@ impl CodexMessageProcessor { let ListConversationsParams { page_size, cursor, - model_providers: model_provider, + model_providers, } = params; - let page_size = page_size.unwrap_or(25); + + let page_size = page_size.unwrap_or(25).max(1); + + let (items, next_cursor) = match self + .list_conversations_common(page_size, cursor, model_providers) + .await + { + Ok(r) => r, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let response = ListConversationsResponse { items, next_cursor }; + self.outgoing.send_response(request_id, response).await; + } + + async fn list_conversations_common( + &self, + page_size: usize, + cursor: Option, + model_providers: Option>, + ) -> Result<(Vec, Option), JSONRPCErrorError> { // Decode the optional cursor string to a Cursor via serde (Cursor implements Deserialize from string) let cursor_obj: Option = match cursor { Some(s) => serde_json::from_str::(&format!("\"{s}\"")).ok(), None => None, }; let cursor_ref = cursor_obj.as_ref(); - let model_provider_filter = match model_provider { + + let model_provider_filter = match model_providers { Some(providers) => { if providers.is_empty() { None @@ -938,7 +1565,6 @@ impl CodexMessageProcessor { } None => Some(vec![self.config.model_provider_id.clone()]), }; - let model_provider_slice = model_provider_filter.as_deref(); let fallback_provider = self.config.model_provider_id.clone(); let page = match RolloutRecorder::list_conversations( @@ -946,20 +1572,18 @@ impl CodexMessageProcessor { page_size, cursor_ref, INTERACTIVE_SESSION_SOURCES, - model_provider_slice, + model_provider_filter.as_deref(), fallback_provider.as_str(), ) .await { Ok(p) => p, Err(err) => { - let error = JSONRPCErrorError { + return Err(JSONRPCErrorError { code: INTERNAL_ERROR_CODE, message: format!("failed to list conversations: {err}"), data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; + }); } }; @@ -967,7 +1591,7 @@ impl CodexMessageProcessor { .items .into_iter() .filter_map(|it| extract_conversation_summary(it.path, &it.head, &fallback_provider)) - .collect(); + .collect::>(); // Encode next_cursor as a plain string let next_cursor = match page.next_cursor { @@ -978,25 +1602,31 @@ impl CodexMessageProcessor { None => None, }; - let response = ListConversationsResponse { items, next_cursor }; - self.outgoing.send_response(request_id, response).await; + Ok((items, next_cursor)) } - async fn list_models(&self, request_id: RequestId, params: ListModelsParams) { - let ListModelsParams { page_size, cursor } = params; - let models = supported_models(); + async fn list_models(&self, request_id: RequestId, params: ModelListParams) { + let ModelListParams { cursor, limit } = params; + + let mut models = supported_models(); + + // Sort models in descending order by id (default behavior). + models.sort_by(|a, b| b.id.cmp(&a.id)); + let total = models.len(); if total == 0 { - let response = ListModelsResponse { - items: Vec::new(), + let response = ModelListResponse { + data: Vec::new(), next_cursor: None, }; self.outgoing.send_response(request_id, response).await; return; } - let effective_page_size = page_size.unwrap_or(total).max(1).min(total); + // Determine pagination window + let default_limit = total as i32; + let effective_limit = limit.unwrap_or(default_limit).max(1).min(default_limit) as usize; let start = match cursor { Some(cursor) => match cursor.parse::() { Ok(idx) => idx, @@ -1023,14 +1653,15 @@ impl CodexMessageProcessor { return; } - let end = start.saturating_add(effective_page_size).min(total); - let items = models[start..end].to_vec(); + let end = start.saturating_add(effective_limit).min(total); + let data = models[start..end].to_vec(); let next_cursor = if end < total { Some(end.to_string()) } else { None }; - let response = ListModelsResponse { items, next_cursor }; + + let response = ModelListResponse { data, next_cursor }; self.outgoing.send_response(request_id, response).await; } @@ -1048,8 +1679,41 @@ impl CodexMessageProcessor { // Derive a Config using the same logic as new conversation, honoring overrides if provided. let config = match overrides { - Some(overrides) => { - derive_config_from_params(overrides, self.codex_linux_sandbox_exe.clone()).await + Some(params) => { + let NewConversationParams { + model, + model_provider, + profile, + cwd, + approval_policy, + sandbox: sandbox_mode, + config: cli_overrides, + base_instructions, + developer_instructions, + compact_prompt, + include_apply_patch_tool, + } = params; + + let overrides = ConfigOverrides { + model, + review_model: None, + config_profile: profile, + cwd: cwd.map(PathBuf::from), + approval_policy, + sandbox_mode, + model_provider, + codex_linux_sandbox_exe: self.codex_linux_sandbox_exe.clone(), + base_instructions, + developer_instructions, + compact_prompt, + include_apply_patch_tool, + show_raw_agent_reasoning: None, + tools_web_search_request: None, + experimental_sandbox_command_assessment: None, + additional_writable_roots: Vec::new(), + }; + + derive_config_from_params(overrides, cli_overrides).await } None => Ok(self.config.as_ref().clone()), }; @@ -1193,151 +1857,26 @@ impl CodexMessageProcessor { self.outgoing.send_error(request_id, error).await; } - async fn archive_conversation(&self, request_id: RequestId, params: ArchiveConversationParams) { + async fn archive_conversation( + &mut self, + request_id: RequestId, + params: ArchiveConversationParams, + ) { let ArchiveConversationParams { conversation_id, rollout_path, } = params; - // Verify that the rollout path is in the sessions directory or else - // a malicious client could specify an arbitrary path. - let rollout_folder = self.config.codex_home.join(codex_core::SESSIONS_SUBDIR); - let canonical_sessions_dir = match tokio::fs::canonicalize(&rollout_folder).await { - Ok(path) => path, - Err(err) => { - let error = JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to archive conversation: unable to resolve sessions directory: {err}" - ), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - }; - let canonical_rollout_path = tokio::fs::canonicalize(&rollout_path).await; - let canonical_rollout_path = if let Ok(path) = canonical_rollout_path - && path.starts_with(&canonical_sessions_dir) + match self + .archive_conversation_common(conversation_id, &rollout_path) + .await { - path - } else { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` must be in sessions directory", - rollout_path.display() - ), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - }; - - let required_suffix = format!("{conversation_id}.jsonl"); - let Some(file_name) = canonical_rollout_path.file_name().map(OsStr::to_owned) else { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` missing file name", - rollout_path.display() - ), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - }; - - if !file_name - .to_string_lossy() - .ends_with(required_suffix.as_str()) - { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!( - "rollout path `{}` does not match conversation id {conversation_id}", - rollout_path.display() - ), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; - } - - let removed_conversation = self - .conversation_manager - .remove_conversation(&conversation_id) - .await; - if let Some(conversation) = removed_conversation { - info!("conversation {conversation_id} was active; shutting down"); - let conversation_clone = conversation.clone(); - let notify = Arc::new(tokio::sync::Notify::new()); - let notify_clone = notify.clone(); - - // Establish the listener for ShutdownComplete before submitting - // Shutdown so it is not missed. - let is_shutdown = tokio::spawn(async move { - loop { - select! { - _ = notify_clone.notified() => { - break; - } - event = conversation_clone.next_event() => { - if let Ok(event) = event && matches!(event.msg, EventMsg::ShutdownComplete) { - break; - } - } - } - } - }); - - // Request shutdown. - match conversation.submit(Op::Shutdown).await { - Ok(_) => { - // Successfully submitted Shutdown; wait before proceeding. - select! { - _ = is_shutdown => { - // Normal shutdown: proceed with archive. - } - _ = tokio::time::sleep(Duration::from_secs(10)) => { - warn!("conversation {conversation_id} shutdown timed out; proceeding with archive"); - notify.notify_one(); - } - } - } - Err(err) => { - error!("failed to submit Shutdown to conversation {conversation_id}: {err}"); - notify.notify_one(); - // Perhaps we lost a shutdown race, so let's continue to - // clean up the .jsonl file. - } - } - } - - // Move the .jsonl file to the archived sessions subdir. - let result: std::io::Result<()> = async { - let archive_folder = self - .config - .codex_home - .join(codex_core::ARCHIVED_SESSIONS_SUBDIR); - tokio::fs::create_dir_all(&archive_folder).await?; - tokio::fs::rename(&canonical_rollout_path, &archive_folder.join(&file_name)).await?; - Ok(()) - } - .await; - - match result { Ok(()) => { let response = ArchiveConversationResponse {}; self.outgoing.send_response(request_id, response).await; } Err(err) => { - let error = JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!("failed to archive conversation: {err}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; + self.outgoing.send_error(request_id, err).await; } } } @@ -1459,7 +1998,56 @@ impl CodexMessageProcessor { // Record the pending interrupt so we can reply when TurnAborted arrives. { let mut map = self.pending_interrupts.lock().await; - map.entry(conversation_id).or_default().push(request_id); + map.entry(conversation_id) + .or_default() + .push((request_id, ApiVersion::V1)); + } + + // Submit the interrupt; we'll respond upon TurnAborted. + let _ = conversation.submit(Op::Interrupt).await; + } + + async fn turn_interrupt( + &mut self, + request_id: RequestId, + params: codex_app_server_protocol::TurnInterruptParams, + ) { + let codex_app_server_protocol::TurnInterruptParams { thread_id, .. } = params; + + // Resolve conversation id from v2 thread id string. + let conversation_id = match ConversationId::from_string(&thread_id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let Ok(conversation) = self + .conversation_manager + .get_conversation(conversation_id) + .await + else { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("conversation not found: {conversation_id}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + }; + + // Record the pending interrupt so we can reply when TurnAborted arrives. + { + let mut map = self.pending_interrupts.lock().await; + map.entry(conversation_id) + .or_default() + .push((request_id, ApiVersion::V2)); } // Submit the interrupt; we'll respond upon TurnAborted. @@ -1475,26 +2063,49 @@ impl CodexMessageProcessor { conversation_id, experimental_raw_events, } = params; - let Ok(conversation) = self + match self + .attach_conversation_listener(conversation_id, experimental_raw_events) + .await + { + Ok(subscription_id) => { + let response = AddConversationSubscriptionResponse { subscription_id }; + self.outgoing.send_response(request_id, response).await; + } + Err(err) => { + self.outgoing.send_error(request_id, err).await; + } + } + } + + async fn attach_conversation_listener( + &mut self, + conversation_id: ConversationId, + experimental_raw_events: bool, + ) -> Result { + let conversation = match self .conversation_manager .get_conversation(conversation_id) .await - else { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("conversation not found: {conversation_id}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; - return; + { + Ok(conv) => conv, + Err(_) => { + return Err(JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("conversation not found: {conversation_id}"), + data: None, + }); + } }; let subscription_id = Uuid::new_v4(); let (cancel_tx, mut cancel_rx) = oneshot::channel(); self.conversation_listeners .insert(subscription_id, cancel_tx); + let outgoing_for_task = self.outgoing.clone(); let pending_interrupts = self.pending_interrupts.clone(); + let conversation_for_task = conversation; + let conversation_id_for_task = conversation_id; tokio::spawn(async move { loop { tokio::select! { @@ -1502,7 +2113,7 @@ impl CodexMessageProcessor { // User has unsubscribed, so exit this task. break; } - event = conversation.next_event() => { + event = conversation_for_task.next_event() => { let event = match event { Ok(event) => event, Err(err) => { @@ -1532,21 +2143,28 @@ impl CodexMessageProcessor { continue; } }; - params.insert("conversationId".to_string(), conversation_id.to_string().into()); + params.insert("conversationId".to_string(), conversation_id_for_task.to_string().into()); - outgoing_for_task.send_notification(OutgoingNotification { - method, - params: Some(params.into()), - }) - .await; + outgoing_for_task + .send_notification(OutgoingNotification { + method, + params: Some(params.into()), + }) + .await; - apply_bespoke_event_handling(event.clone(), conversation_id, conversation.clone(), outgoing_for_task.clone(), pending_interrupts.clone()).await; + apply_bespoke_event_handling( + event.clone(), + conversation_id_for_task, + conversation_for_task.clone(), + outgoing_for_task.clone(), + pending_interrupts.clone(), + ).await; } } } }); - let response = AddConversationSubscriptionResponse { subscription_id }; - self.outgoing.send_response(request_id, response).await; + + Ok(subscription_id) } async fn remove_conversation_listener( @@ -1634,8 +2252,8 @@ impl CodexMessageProcessor { self.outgoing.send_response(request_id, response).await; } - async fn upload_feedback(&self, request_id: RequestId, params: UploadFeedbackParams) { - let UploadFeedbackParams { + async fn upload_feedback(&self, request_id: RequestId, params: FeedbackUploadParams) { + let FeedbackUploadParams { classification, reason, conversation_id, @@ -1680,7 +2298,7 @@ impl CodexMessageProcessor { match upload_result { Ok(()) => { - let response = UploadFeedbackResponse { thread_id }; + let response = FeedbackUploadResponse { thread_id }; self.outgoing.send_response(request_id, response).await; } Err(err) => { @@ -1711,7 +2329,7 @@ async fn apply_bespoke_event_handling( conversation_id: ConversationId, conversation: Arc, outgoing: Arc, - pending_interrupts: Arc>>>, + pending_interrupts: PendingInterrupts, ) { let Event { id: event_id, msg } = event; match msg { @@ -1764,9 +2382,12 @@ async fn apply_bespoke_event_handling( } EventMsg::TokenCount(token_count_event) => { if let Some(rate_limits) = token_count_event.rate_limits { + let snapshot: V2RateLimitSnapshot = rate_limits.into(); outgoing .send_server_notification(ServerNotification::AccountRateLimitsUpdated( - rate_limits, + AccountRateLimitsUpdatedNotification { + rate_limits: snapshot, + }, )) .await; } @@ -1778,11 +2399,19 @@ async fn apply_bespoke_event_handling( map.remove(&conversation_id).unwrap_or_default() }; if !pending.is_empty() { - let response = InterruptConversationResponse { - abort_reason: turn_aborted_event.reason, - }; - for rid in pending { - outgoing.send_response(rid, response.clone()).await; + for (rid, ver) in pending { + match ver { + ApiVersion::V1 => { + let response = InterruptConversationResponse { + abort_reason: turn_aborted_event.reason.clone(), + }; + outgoing.send_response(rid, response).await; + } + ApiVersion::V2 => { + let response = codex_app_server_protocol::TurnInterruptResponse {}; + outgoing.send_response(rid, response).await; + } + } } } } @@ -1792,41 +2421,9 @@ async fn apply_bespoke_event_handling( } async fn derive_config_from_params( - params: NewConversationParams, - codex_linux_sandbox_exe: Option, + overrides: ConfigOverrides, + cli_overrides: Option>, ) -> std::io::Result { - let NewConversationParams { - model, - model_provider, - profile, - cwd, - approval_policy, - sandbox: sandbox_mode, - config: cli_overrides, - base_instructions, - developer_instructions, - compact_prompt, - include_apply_patch_tool, - } = params; - let overrides = ConfigOverrides { - model, - review_model: None, - config_profile: profile, - cwd: cwd.map(PathBuf::from), - approval_policy, - sandbox_mode, - model_provider, - codex_linux_sandbox_exe, - base_instructions, - developer_instructions, - compact_prompt, - include_apply_patch_tool, - show_raw_agent_reasoning: None, - tools_web_search_request: None, - experimental_sandbox_command_assessment: None, - additional_writable_roots: Vec::new(), - }; - let cli_overrides = cli_overrides .unwrap_or_default() .into_iter() diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 3d56c9bd83..6563c1ff17 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -144,8 +144,8 @@ mod tests { use codex_app_server_protocol::AccountUpdatedNotification; use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::LoginChatGptCompleteNotification; - use codex_protocol::protocol::RateLimitSnapshot; - use codex_protocol::protocol::RateLimitWindow; + use codex_app_server_protocol::RateLimitSnapshot; + use codex_app_server_protocol::RateLimitWindow; use pretty_assertions::assert_eq; use serde_json::json; use uuid::Uuid; @@ -179,26 +179,32 @@ mod tests { #[test] fn verify_account_rate_limits_notification_serialization() { - let notification = ServerNotification::AccountRateLimitsUpdated(RateLimitSnapshot { - primary: Some(RateLimitWindow { - used_percent: 25.0, - window_minutes: Some(15), - resets_at: Some(123), - }), - secondary: None, - }); + let notification = ServerNotification::AccountRateLimitsUpdated( + codex_app_server_protocol::AccountRateLimitsUpdatedNotification { + rate_limits: RateLimitSnapshot { + primary: Some(RateLimitWindow { + used_percent: 25, + window_duration_mins: Some(15), + resets_at: Some(123), + }), + secondary: None, + }, + }, + ); let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification); assert_eq!( json!({ "method": "account/rateLimits/updated", "params": { - "primary": { - "used_percent": 25.0, - "window_minutes": 15, - "resets_at": 123, - }, - "secondary": null, + "rateLimits": { + "primary": { + "usedPercent": 25, + "windowDurationMins": 15, + "resetsAt": 123 + }, + "secondary": null + } }, }), serde_json::to_value(jsonrpc_notification) diff --git a/codex-rs/app-server/tests/common/Cargo.toml b/codex-rs/app-server/tests/common/Cargo.toml index ece8174dd9..176ea956f5 100644 --- a/codex-rs/app-server/tests/common/Cargo.toml +++ b/codex-rs/app-server/tests/common/Cargo.toml @@ -22,3 +22,4 @@ tokio = { workspace = true, features = [ "rt-multi-thread", ] } wiremock = { workspace = true } +uuid = { workspace = true } diff --git a/codex-rs/app-server/tests/common/lib.rs b/codex-rs/app-server/tests/common/lib.rs index 71b27190c2..dc3d24ccae 100644 --- a/codex-rs/app-server/tests/common/lib.rs +++ b/codex-rs/app-server/tests/common/lib.rs @@ -2,6 +2,7 @@ mod auth_fixtures; mod mcp_process; mod mock_model_server; mod responses; +mod rollout; pub use auth_fixtures::ChatGptAuthFixture; pub use auth_fixtures::ChatGptIdTokenClaims; @@ -10,9 +11,11 @@ pub use auth_fixtures::write_chatgpt_auth; use codex_app_server_protocol::JSONRPCResponse; pub use mcp_process::McpProcess; pub use mock_model_server::create_mock_chat_completions_server; +pub use mock_model_server::create_mock_chat_completions_server_unchecked; pub use responses::create_apply_patch_sse_response; pub use responses::create_final_assistant_message_sse_response; pub use responses::create_shell_sse_response; +pub use rollout::create_fake_rollout; use serde::de::DeserializeOwned; pub fn to_response(response: JSONRPCResponse) -> anyhow::Result { diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index 3c1fbf777c..d6226db82f 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -17,12 +17,13 @@ use codex_app_server_protocol::ArchiveConversationParams; use codex_app_server_protocol::CancelLoginChatGptParams; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; +use codex_app_server_protocol::FeedbackUploadParams; use codex_app_server_protocol::GetAuthStatusParams; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::InterruptConversationParams; use codex_app_server_protocol::ListConversationsParams; -use codex_app_server_protocol::ListModelsParams; use codex_app_server_protocol::LoginApiKeyParams; +use codex_app_server_protocol::ModelListParams; use codex_app_server_protocol::NewConversationParams; use codex_app_server_protocol::RemoveConversationListenerParams; use codex_app_server_protocol::ResumeConversationParams; @@ -30,7 +31,12 @@ use codex_app_server_protocol::SendUserMessageParams; use codex_app_server_protocol::SendUserTurnParams; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::SetDefaultModelParams; -use codex_app_server_protocol::UploadFeedbackParams; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::TurnInterruptParams as V2TurnInterruptParams; +use codex_app_server_protocol::TurnStartParams as V2TurnStartParams; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; @@ -246,7 +252,7 @@ impl McpProcess { /// Send a `feedback/upload` JSON-RPC request. pub async fn send_upload_feedback_request( &mut self, - params: UploadFeedbackParams, + params: FeedbackUploadParams, ) -> anyhow::Result { let params = Some(serde_json::to_value(params)?); self.send_request("feedback/upload", params).await @@ -278,12 +284,21 @@ impl McpProcess { /// Send a `model/list` JSON-RPC request. pub async fn send_list_models_request( &mut self, - params: ListModelsParams, + params: ModelListParams, ) -> anyhow::Result { let params = Some(serde_json::to_value(params)?); self.send_request("model/list", params).await } + /// Send a `thread/start` JSON-RPC request (v2). + pub async fn send_thread_start_request( + &mut self, + params: ThreadStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/start", params).await + } + /// Send a `resumeConversation` JSON-RPC request. pub async fn send_resume_conversation_request( &mut self, @@ -307,6 +322,51 @@ impl McpProcess { self.send_request("loginChatGpt", None).await } + /// Send a `thread/archive` JSON-RPC request (v2). + pub async fn send_thread_archive_request( + &mut self, + params: ThreadArchiveParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/archive", params).await + } + + /// Send a `thread/list` JSON-RPC request (v2). + pub async fn send_thread_list_request( + &mut self, + params: ThreadListParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/list", params).await + } + + /// Send a `thread/resume` JSON-RPC request (v2). + pub async fn send_thread_resume_request( + &mut self, + params: ThreadResumeParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/resume", params).await + } + + /// Send a `turn/start` JSON-RPC request (v2). + pub async fn send_turn_start_request( + &mut self, + params: V2TurnStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("turn/start", params).await + } + + /// Send a `turn/interrupt` JSON-RPC request (v2). + pub async fn send_turn_interrupt_request( + &mut self, + params: V2TurnInterruptParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("turn/interrupt", params).await + } + /// Send a `cancelLoginChatGpt` JSON-RPC request. pub async fn send_cancel_login_chat_gpt_request( &mut self, diff --git a/codex-rs/app-server/tests/common/mock_model_server.rs b/codex-rs/app-server/tests/common/mock_model_server.rs index be7f3eb5b3..1092c80a14 100644 --- a/codex-rs/app-server/tests/common/mock_model_server.rs +++ b/codex-rs/app-server/tests/common/mock_model_server.rs @@ -29,6 +29,27 @@ pub async fn create_mock_chat_completions_server(responses: Vec) -> Mock server } +/// Same as `create_mock_chat_completions_server` but does not set an expectation +/// on the number of requests. Useful for flows where the exact number of +/// model calls may vary (e.g., when the pipeline conditionally compacts or +/// when the test only needs a responsive endpoint). +pub async fn create_mock_chat_completions_server_unchecked(responses: Vec) -> MockServer { + let server = MockServer::start().await; + + let seq_responder = SeqResponder { + num_calls: AtomicUsize::new(0), + responses, + }; + + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(seq_responder) + .mount(&server) + .await; + + server +} + struct SeqResponder { num_calls: AtomicUsize, responses: Vec, diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs new file mode 100644 index 0000000000..6537bb59a0 --- /dev/null +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -0,0 +1,77 @@ +use anyhow::Result; +use serde_json::json; +use std::fs; +use std::path::Path; +use uuid::Uuid; + +/// Create a minimal rollout file under `CODEX_HOME/sessions/YYYY/MM/DD/`. +/// +/// - `filename_ts` is the filename timestamp component in `YYYY-MM-DDThh-mm-ss` format. +/// - `meta_rfc3339` is the envelope timestamp used in JSON lines. +/// - `preview` is the user message preview text. +/// - `model_provider` optionally sets the provider in the session meta payload. +/// +/// Returns the generated conversation/session UUID as a string. +pub fn create_fake_rollout( + codex_home: &Path, + filename_ts: &str, + meta_rfc3339: &str, + preview: &str, + model_provider: Option<&str>, +) -> Result { + let uuid = Uuid::new_v4(); + + // sessions/YYYY/MM/DD derived from filename_ts (YYYY-MM-DDThh-mm-ss) + let year = &filename_ts[0..4]; + let month = &filename_ts[5..7]; + let day = &filename_ts[8..10]; + let dir = codex_home.join("sessions").join(year).join(month).join(day); + fs::create_dir_all(&dir)?; + + let file_path = dir.join(format!("rollout-{filename_ts}-{uuid}.jsonl")); + + // Build JSONL lines + let mut payload = json!({ + "id": uuid, + "timestamp": meta_rfc3339, + "cwd": "/", + "originator": "codex", + "cli_version": "0.0.0", + "instructions": null, + }); + if let Some(provider) = model_provider { + payload["model_provider"] = json!(provider); + } + + let lines = [ + json!({ + "timestamp": meta_rfc3339, + "type": "session_meta", + "payload": payload + }) + .to_string(), + json!({ + "timestamp": meta_rfc3339, + "type":"response_item", + "payload": { + "type":"message", + "role":"user", + "content":[{"type":"input_text","text": preview}] + } + }) + .to_string(), + json!({ + "timestamp": meta_rfc3339, + "type":"event_msg", + "payload": { + "type":"user_message", + "message": preview, + "kind": "plain" + } + }) + .to_string(), + ]; + + fs::write(file_path, lines.join("\n") + "\n")?; + Ok(uuid.to_string()) +} diff --git a/codex-rs/app-server/tests/suite/list_resume.rs b/codex-rs/app-server/tests/suite/list_resume.rs index 76b92b3edc..30be93a2e6 100644 --- a/codex-rs/app-server/tests/suite/list_resume.rs +++ b/codex-rs/app-server/tests/suite/list_resume.rs @@ -1,5 +1,6 @@ use anyhow::Result; use app_test_support::McpProcess; +use app_test_support::create_fake_rollout; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; @@ -15,12 +16,8 @@ use codex_core::protocol::EventMsg; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use pretty_assertions::assert_eq; -use serde_json::json; -use std::fs; -use std::path::Path; use tempfile::TempDir; use tokio::time::timeout; -use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); @@ -357,70 +354,3 @@ async fn test_list_and_resume_conversations() -> Result<()> { Ok(()) } - -fn create_fake_rollout( - codex_home: &Path, - filename_ts: &str, - meta_rfc3339: &str, - preview: &str, - model_provider: Option<&str>, -) -> Result<()> { - let uuid = Uuid::new_v4(); - // sessions/YYYY/MM/DD/ derived from filename_ts (YYYY-MM-DDThh-mm-ss) - let year = &filename_ts[0..4]; - let month = &filename_ts[5..7]; - let day = &filename_ts[8..10]; - let dir = codex_home.join("sessions").join(year).join(month).join(day); - fs::create_dir_all(&dir)?; - - let file_path = dir.join(format!("rollout-{filename_ts}-{uuid}.jsonl")); - let mut lines = Vec::new(); - // Meta line with timestamp (flattened meta in payload for new schema) - let mut payload = json!({ - "id": uuid, - "timestamp": meta_rfc3339, - "cwd": "/", - "originator": "codex", - "cli_version": "0.0.0", - "instructions": null, - }); - if let Some(provider) = model_provider { - payload["model_provider"] = json!(provider); - } - lines.push( - json!({ - "timestamp": meta_rfc3339, - "type": "session_meta", - "payload": payload - }) - .to_string(), - ); - // Minimal user message entry as a persisted response item (with envelope timestamp) - lines.push( - json!({ - "timestamp": meta_rfc3339, - "type":"response_item", - "payload": { - "type":"message", - "role":"user", - "content":[{"type":"input_text","text": preview}] - } - }) - .to_string(), - ); - // Add a matching user message event line to satisfy filters - lines.push( - json!({ - "timestamp": meta_rfc3339, - "type":"event_msg", - "payload": { - "type":"user_message", - "message": preview, - "kind": "plain" - } - }) - .to_string(), - ); - fs::write(file_path, lines.join("\n") + "\n")?; - Ok(()) -} diff --git a/codex-rs/app-server/tests/suite/mod.rs b/codex-rs/app-server/tests/suite/mod.rs index ce8b69a030..37f7659f42 100644 --- a/codex-rs/app-server/tests/suite/mod.rs +++ b/codex-rs/app-server/tests/suite/mod.rs @@ -7,8 +7,6 @@ mod fuzzy_file_search; mod interrupt; mod list_resume; mod login; -mod model_list; -mod rate_limits; mod send_message; mod set_default_model; mod user_agent; diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index bd7a1467cd..587afef108 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -1,2 +1,9 @@ -// v2 test suite modules mod account; +mod model_list; +mod rate_limits; +mod thread_archive; +mod thread_list; +mod thread_resume; +mod thread_start; +mod turn_interrupt; +mod turn_start; diff --git a/codex-rs/app-server/tests/suite/model_list.rs b/codex-rs/app-server/tests/suite/v2/model_list.rs similarity index 84% rename from codex-rs/app-server/tests/suite/model_list.rs rename to codex-rs/app-server/tests/suite/v2/model_list.rs index 3472e0d3cb..cb5b4c693f 100644 --- a/codex-rs/app-server/tests/suite/model_list.rs +++ b/codex-rs/app-server/tests/suite/v2/model_list.rs @@ -6,9 +6,9 @@ use app_test_support::McpProcess; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; -use codex_app_server_protocol::ListModelsParams; -use codex_app_server_protocol::ListModelsResponse; use codex_app_server_protocol::Model; +use codex_app_server_protocol::ModelListParams; +use codex_app_server_protocol::ModelListResponse; use codex_app_server_protocol::ReasoningEffortOption; use codex_app_server_protocol::RequestId; use codex_protocol::config_types::ReasoningEffort; @@ -19,7 +19,7 @@ use tokio::time::timeout; const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); const INVALID_REQUEST_ERROR_CODE: i64 = -32600; -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn list_models_returns_all_models_with_large_limit() -> Result<()> { let codex_home = TempDir::new()?; let mut mcp = McpProcess::new(codex_home.path()).await?; @@ -27,9 +27,9 @@ async fn list_models_returns_all_models_with_large_limit() -> Result<()> { timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp - .send_list_models_request(ListModelsParams { - page_size: Some(100), + .send_list_models_request(ModelListParams { cursor: None, + limit: Some(100), }) .await?; @@ -39,7 +39,7 @@ async fn list_models_returns_all_models_with_large_limit() -> Result<()> { ) .await??; - let ListModelsResponse { items, next_cursor } = to_response::(response)?; + let ModelListResponse { data, next_cursor } = to_response::(response)?; let expected_models = vec![ Model { @@ -98,12 +98,12 @@ async fn list_models_returns_all_models_with_large_limit() -> Result<()> { }, ]; - assert_eq!(items, expected_models); + assert_eq!(data, expected_models); assert!(next_cursor.is_none()); Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn list_models_pagination_works() -> Result<()> { let codex_home = TempDir::new()?; let mut mcp = McpProcess::new(codex_home.path()).await?; @@ -111,9 +111,9 @@ async fn list_models_pagination_works() -> Result<()> { timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let first_request = mcp - .send_list_models_request(ListModelsParams { - page_size: Some(1), + .send_list_models_request(ModelListParams { cursor: None, + limit: Some(1), }) .await?; @@ -123,19 +123,19 @@ async fn list_models_pagination_works() -> Result<()> { ) .await??; - let ListModelsResponse { - items: first_items, + let ModelListResponse { + data: first_items, next_cursor: first_cursor, - } = to_response::(first_response)?; + } = to_response::(first_response)?; assert_eq!(first_items.len(), 1); assert_eq!(first_items[0].id, "gpt-5-codex"); let next_cursor = first_cursor.ok_or_else(|| anyhow!("cursor for second page"))?; let second_request = mcp - .send_list_models_request(ListModelsParams { - page_size: Some(1), + .send_list_models_request(ModelListParams { cursor: Some(next_cursor.clone()), + limit: Some(1), }) .await?; @@ -145,10 +145,10 @@ async fn list_models_pagination_works() -> Result<()> { ) .await??; - let ListModelsResponse { - items: second_items, + let ModelListResponse { + data: second_items, next_cursor: second_cursor, - } = to_response::(second_response)?; + } = to_response::(second_response)?; assert_eq!(second_items.len(), 1); assert_eq!(second_items[0].id, "gpt-5"); @@ -156,7 +156,7 @@ async fn list_models_pagination_works() -> Result<()> { Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn list_models_rejects_invalid_cursor() -> Result<()> { let codex_home = TempDir::new()?; let mut mcp = McpProcess::new(codex_home.path()).await?; @@ -164,9 +164,9 @@ async fn list_models_rejects_invalid_cursor() -> Result<()> { timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp - .send_list_models_request(ListModelsParams { - page_size: None, + .send_list_models_request(ModelListParams { cursor: Some("invalid".to_string()), + limit: None, }) .await?; diff --git a/codex-rs/app-server/tests/suite/rate_limits.rs b/codex-rs/app-server/tests/suite/v2/rate_limits.rs similarity index 92% rename from codex-rs/app-server/tests/suite/rate_limits.rs rename to codex-rs/app-server/tests/suite/v2/rate_limits.rs index 16cbf153e9..d0cba83667 100644 --- a/codex-rs/app-server/tests/suite/rate_limits.rs +++ b/codex-rs/app-server/tests/suite/v2/rate_limits.rs @@ -7,10 +7,10 @@ use codex_app_server_protocol::GetAccountRateLimitsResponse; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::LoginApiKeyParams; +use codex_app_server_protocol::RateLimitSnapshot; +use codex_app_server_protocol::RateLimitWindow; use codex_app_server_protocol::RequestId; use codex_core::auth::AuthCredentialsStoreMode; -use codex_protocol::protocol::RateLimitSnapshot; -use codex_protocol::protocol::RateLimitWindow; use pretty_assertions::assert_eq; use serde_json::json; use std::path::Path; @@ -26,7 +26,7 @@ use wiremock::matchers::path; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const INVALID_REQUEST_ERROR_CODE: i64 = -32600; -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn get_account_rate_limits_requires_auth() -> Result<()> { let codex_home = TempDir::new()?; @@ -51,7 +51,7 @@ async fn get_account_rate_limits_requires_auth() -> Result<()> { Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn get_account_rate_limits_requires_chatgpt_auth() -> Result<()> { let codex_home = TempDir::new()?; @@ -78,7 +78,7 @@ async fn get_account_rate_limits_requires_chatgpt_auth() -> Result<()> { Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[tokio::test] async fn get_account_rate_limits_returns_snapshot() -> Result<()> { let codex_home = TempDir::new()?; write_chatgpt_auth( @@ -143,13 +143,13 @@ async fn get_account_rate_limits_returns_snapshot() -> Result<()> { let expected = GetAccountRateLimitsResponse { rate_limits: RateLimitSnapshot { primary: Some(RateLimitWindow { - used_percent: 42.0, - window_minutes: Some(60), + used_percent: 42, + window_duration_mins: Some(60), resets_at: Some(primary_reset_timestamp), }), secondary: Some(RateLimitWindow { - used_percent: 5.0, - window_minutes: Some(1440), + used_percent: 5, + window_duration_mins: Some(1440), resets_at: Some(secondary_reset_timestamp), }), }, diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs new file mode 100644 index 0000000000..083f3da901 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -0,0 +1,93 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_core::ARCHIVED_SESSIONS_SUBDIR; +use codex_core::find_conversation_path_by_id_str; +use std::path::Path; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_archive_moves_rollout_into_archived_directory() -> Result<()> { + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a thread. + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(start_resp)?; + assert!(!thread.id.is_empty()); + + // Locate the rollout path recorded for this thread id. + let rollout_path = find_conversation_path_by_id_str(codex_home.path(), &thread.id) + .await? + .expect("expected rollout path for thread id to exist"); + assert!( + rollout_path.exists(), + "expected {} to exist", + rollout_path.display() + ); + + // Archive the thread. + let archive_id = mcp + .send_thread_archive_request(ThreadArchiveParams { + thread_id: thread.id.clone(), + }) + .await?; + let archive_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(archive_id)), + ) + .await??; + let _: ThreadArchiveResponse = to_response::(archive_resp)?; + + // Verify file moved. + let archived_directory = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); + // The archived file keeps the original filename (rollout-...-.jsonl). + let archived_rollout_path = + archived_directory.join(rollout_path.file_name().expect("rollout file name")); + assert!( + !rollout_path.exists(), + "expected rollout path {} to be moved", + rollout_path.display() + ); + assert!( + archived_rollout_path.exists(), + "expected archived rollout path {} to exist", + archived_rollout_path.display() + ); + + Ok(()) +} + +fn create_config_toml(codex_home: &Path) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write(config_toml, config_contents()) +} + +fn config_contents() -> &'static str { + r#"model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +"# +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs new file mode 100644 index 0000000000..25a15a9385 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -0,0 +1,208 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_fake_rollout; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadListResponse; +use serde_json::json; +use tempfile::TempDir; +use tokio::time::timeout; +use uuid::Uuid; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_list_basic_empty() -> Result<()> { + let codex_home = TempDir::new()?; + create_minimal_config(codex_home.path())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // List threads in an empty CODEX_HOME; should return an empty page with nextCursor: null. + let list_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: Some(10), + model_providers: None, + }) + .await?; + let list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(list_id)), + ) + .await??; + let ThreadListResponse { data, next_cursor } = to_response::(list_resp)?; + assert!(data.is_empty()); + assert!(next_cursor.is_none()); + + Ok(()) +} + +// Minimal config.toml for listing. +fn create_minimal_config(codex_home: &std::path::Path) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + r#" +model = "mock-model" +approval_policy = "never" +"#, + ) +} + +#[tokio::test] +async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> { + let codex_home = TempDir::new()?; + create_minimal_config(codex_home.path())?; + + // Create three rollouts so we can paginate with limit=2. + let _a = create_fake_rollout( + codex_home.path(), + "2025-01-02T12-00-00", + "2025-01-02T12:00:00Z", + "Hello", + Some("mock_provider"), + )?; + let _b = create_fake_rollout( + codex_home.path(), + "2025-01-01T13-00-00", + "2025-01-01T13:00:00Z", + "Hello", + Some("mock_provider"), + )?; + let _c = create_fake_rollout( + codex_home.path(), + "2025-01-01T12-00-00", + "2025-01-01T12:00:00Z", + "Hello", + Some("mock_provider"), + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Page 1: limit 2 → expect next_cursor Some. + let page1_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: Some(2), + model_providers: Some(vec!["mock_provider".to_string()]), + }) + .await?; + let page1_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(page1_id)), + ) + .await??; + let ThreadListResponse { + data: data1, + next_cursor: cursor1, + } = to_response::(page1_resp)?; + assert_eq!(data1.len(), 2); + let cursor1 = cursor1.expect("expected nextCursor on first page"); + + // Page 2: with cursor → expect next_cursor None when no more results. + let page2_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: Some(cursor1), + limit: Some(2), + model_providers: Some(vec!["mock_provider".to_string()]), + }) + .await?; + let page2_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(page2_id)), + ) + .await??; + let ThreadListResponse { + data: data2, + next_cursor: cursor2, + } = to_response::(page2_resp)?; + assert!(data2.len() <= 2); + assert!( + cursor2.is_none(), + "expected nextCursor to be null on last page" + ); + + Ok(()) +} + +#[tokio::test] +async fn thread_list_respects_provider_filter() -> Result<()> { + let codex_home = TempDir::new()?; + create_minimal_config(codex_home.path())?; + + // Create rollouts under two providers. + let _a = create_fake_rollout( + codex_home.path(), + "2025-01-02T10-00-00", + "2025-01-02T10:00:00Z", + "X", + Some("mock_provider"), + )?; // mock_provider + // one with a different provider + let uuid = Uuid::new_v4(); + let dir = codex_home + .path() + .join("sessions") + .join("2025") + .join("01") + .join("02"); + std::fs::create_dir_all(&dir)?; + let file_path = dir.join(format!("rollout-2025-01-02T11-00-00-{uuid}.jsonl")); + let lines = [ + json!({ + "timestamp": "2025-01-02T11:00:00Z", + "type": "session_meta", + "payload": { + "id": uuid, + "timestamp": "2025-01-02T11:00:00Z", + "cwd": "/", + "originator": "codex", + "cli_version": "0.0.0", + "instructions": null, + "source": "vscode", + "model_provider": "other_provider" + } + }) + .to_string(), + json!({ + "timestamp": "2025-01-02T11:00:00Z", + "type":"response_item", + "payload": {"type":"message","role":"user","content":[{"type":"input_text","text":"X"}]} + }) + .to_string(), + json!({ + "timestamp": "2025-01-02T11:00:00Z", + "type":"event_msg", + "payload": {"type":"user_message","message":"X","kind":"plain"} + }) + .to_string(), + ]; + std::fs::write(file_path, lines.join("\n") + "\n")?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Filter to only other_provider; expect 1 item, nextCursor None. + let list_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: Some(10), + model_providers: Some(vec!["other_provider".to_string()]), + }) + .await?; + let resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(list_id)), + ) + .await??; + let ThreadListResponse { data, next_cursor } = to_response::(resp)?; + assert_eq!(data.len(), 1); + assert!(next_cursor.is_none()); + + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs new file mode 100644 index 0000000000..201ef34f5a --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -0,0 +1,79 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_mock_chat_completions_server; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_resume_returns_existing_thread() -> Result<()> { + let server = create_mock_chat_completions_server(vec![]).await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a thread. + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("gpt-5-codex".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(start_resp)?; + + // Resume it via v2 API. + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread: resumed } = + to_response::(resume_resp)?; + assert_eq!(resumed.id, thread.id); + + Ok(()) +} + +// Helper to create a config.toml pointing at the mock model server. +fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "workspace-write" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs new file mode 100644 index 0000000000..c27fcbc16b --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -0,0 +1,82 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_mock_chat_completions_server; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStartedNotification; +use pretty_assertions::assert_eq; +use std::path::Path; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_start_creates_thread_and_emits_started() -> Result<()> { + // Provide a mock server and config so model wiring is valid. + let server = create_mock_chat_completions_server(vec![]).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + // Start server and initialize. + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a v2 thread with an explicit model override. + let req_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("o3".to_string()), + ..Default::default() + }) + .await?; + + // Expect a proper JSON-RPC response with a thread id. + let resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(req_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(resp)?; + assert!(!thread.id.is_empty(), "thread id should not be empty"); + + // A corresponding thread/started notification should arrive. + let notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/started"), + ) + .await??; + let started: ThreadStartedNotification = + serde_json::from_value(notif.params.expect("params must be present"))?; + assert_eq!(started.thread.id, thread.id); + + Ok(()) +} + +// Helper to create a config.toml pointing at the mock model server. +fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "workspace-write" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs new file mode 100644 index 0000000000..6b8d200b06 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs @@ -0,0 +1,132 @@ +#![cfg(unix)] + +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_mock_chat_completions_server; +use app_test_support::create_shell_sse_response; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput as V2UserInput; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn turn_interrupt_aborts_running_turn() -> Result<()> { + // Use a portable sleep command to keep the turn running. + #[cfg(target_os = "windows")] + let shell_command = vec![ + "powershell".to_string(), + "-Command".to_string(), + "Start-Sleep -Seconds 10".to_string(), + ]; + #[cfg(not(target_os = "windows"))] + let shell_command = vec!["sleep".to_string(), "10".to_string()]; + + let tmp = TempDir::new()?; + let codex_home = tmp.path().join("codex_home"); + std::fs::create_dir(&codex_home)?; + let working_directory = tmp.path().join("workdir"); + std::fs::create_dir(&working_directory)?; + + // Mock server: long-running shell command then (after abort) nothing else needed. + let server = create_mock_chat_completions_server(vec![create_shell_sse_response( + shell_command.clone(), + Some(&working_directory), + Some(10_000), + "call_sleep", + )?]) + .await; + create_config_toml(&codex_home, &server.uri())?; + + let mut mcp = McpProcess::new(&codex_home).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a v2 thread and capture its id. + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(thread_resp)?; + + // Start a turn that triggers a long-running command. + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "run sleep".to_string(), + }], + cwd: Some(working_directory.clone()), + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + + // Give the command a brief moment to start. + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + + // Interrupt the in-progress turn by id (v2 API). + let interrupt_id = mcp + .send_turn_interrupt_request(TurnInterruptParams { + thread_id: thread.id, + turn_id: turn.id, + }) + .await?; + let interrupt_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(interrupt_id)), + ) + .await??; + let _resp: TurnInterruptResponse = to_response::(interrupt_resp)?; + + // No fields to assert on; successful deserialization confirms proper response shape. + Ok(()) +} + +// Helper to create a config.toml pointing at the mock model server. +fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "workspace-write" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs new file mode 100644 index 0000000000..7c5bb5fd81 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -0,0 +1,529 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_chat_completions_server; +use app_test_support::create_mock_chat_completions_server_unchecked; +use app_test_support::create_shell_sse_response; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerRequest; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::UserInput as V2UserInput; +use codex_core::protocol_config_types::ReasoningEffort; +use codex_core::protocol_config_types::ReasoningSummary; +use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR; +use codex_protocol::parse_command::ParsedCommand; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use core_test_support::skip_if_no_network; +use pretty_assertions::assert_eq; +use std::env; +use std::path::Path; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<()> { + // Provide a mock server and config so model wiring is valid. + // Three Codex turns hit the mock model (session start + two turn/start calls). + let responses = vec![ + create_final_assistant_message_sse_response("Done")?, + create_final_assistant_message_sse_response("Done")?, + create_final_assistant_message_sse_response("Done")?, + ]; + let server = create_mock_chat_completions_server_unchecked(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a thread (v2) and capture its id. + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(thread_resp)?; + + // Start a turn with only input and thread_id set (no overrides). + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Hello".to_string(), + }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + assert!(!turn.id.is_empty()); + + // Expect a turn/started notification. + let notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await??; + let started: TurnStartedNotification = + serde_json::from_value(notif.params.expect("params must be present"))?; + assert_eq!( + started.turn.status, + codex_app_server_protocol::TurnStatus::InProgress + ); + + // Send a second turn that exercises the overrides path: change the model. + let turn_req2 = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Second".to_string(), + }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: Some("mock-model-override".to_string()), + effort: None, + summary: None, + }) + .await?; + let turn_resp2: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req2)), + ) + .await??; + let TurnStartResponse { turn: turn2 } = to_response::(turn_resp2)?; + assert!(!turn2.id.is_empty()); + // Ensure the second turn has a different id than the first. + assert_ne!(turn.id, turn2.id); + + // Expect a second turn/started notification as well. + let _notif2: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await??; + + // And we should ultimately get a task_complete without having to add a + // legacy conversation listener explicitly (auto-attached by thread/start). + let _task_complete: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + Ok(()) +} + +#[tokio::test] +async fn turn_start_accepts_local_image_input() -> Result<()> { + // Two Codex turns hit the mock model (session start + turn/start). + let responses = vec![ + create_final_assistant_message_sse_response("Done")?, + create_final_assistant_message_sse_response("Done")?, + ]; + // Use the unchecked variant because the request payload includes a LocalImage + // which the strict matcher does not currently cover. + let server = create_mock_chat_completions_server_unchecked(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(thread_resp)?; + + let image_path = codex_home.path().join("image.png"); + // No need to actually write the file; we just exercise the input path. + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::LocalImage { path: image_path }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + assert!(!turn.id.is_empty()); + + // This test only validates that turn/start responds and returns a turn. + Ok(()) +} + +#[tokio::test] +async fn turn_start_exec_approval_toggle_v2() -> Result<()> { + if env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() { + println!("Skipping v2 exec approval toggle test due to sandbox network disabled."); + return Ok(()); + } + + let tmp = TempDir::new()?; + let codex_home = tmp.path().to_path_buf(); + + // Mock server: first turn requests a shell call (elicitation), then completes. + // Second turn same, but we'll set approval_policy=never to avoid elicitation. + let responses = vec![ + create_shell_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + None, + Some(5000), + "call1", + )?, + create_final_assistant_message_sse_response("done 1")?, + create_shell_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + None, + Some(5000), + "call2", + )?, + create_final_assistant_message_sse_response("done 2")?, + ]; + let server = create_mock_chat_completions_server(responses).await; + // Default approval is untrusted to force elicitation on first turn. + create_config_toml_untrusted(codex_home.as_path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.as_path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // thread/start + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(start_resp)?; + + // turn/start — expect ExecCommandApproval request from server + let first_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "run python".to_string(), + }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + // Acknowledge RPC + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(first_turn_id)), + ) + .await??; + + // Receive elicitation + let server_req = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + let ServerRequest::ExecCommandApproval { request_id, params } = server_req else { + panic!("expected ExecCommandApproval request"); + }; + assert_eq!(params.call_id, "call1"); + assert_eq!( + params.parsed_cmd, + vec![ParsedCommand::Unknown { + cmd: "python3 -c 'print(42)'".to_string() + }] + ); + + // Approve and wait for task completion + mcp.send_response( + request_id, + serde_json::json!({ "decision": codex_core::protocol::ReviewDecision::Approved }), + ) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + // Second turn with approval_policy=never should not elicit approval + let second_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "run python again".to_string(), + }], + cwd: None, + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + model: Some("mock-model".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)), + ) + .await??; + + // Ensure we do NOT receive an ExecCommandApproval request before task completes + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + Ok(()) +} + +#[tokio::test] +async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { + // When returning Result from a test, pass an Ok(()) to the skip macro + // so the early return type matches. The no-arg form returns unit. + skip_if_no_network!(Ok(())); + + let tmp = TempDir::new()?; + let codex_home = tmp.path().join("codex_home"); + std::fs::create_dir(&codex_home)?; + let workspace_root = tmp.path().join("workspace"); + std::fs::create_dir(&workspace_root)?; + let first_cwd = workspace_root.join("turn1"); + let second_cwd = workspace_root.join("turn2"); + std::fs::create_dir(&first_cwd)?; + std::fs::create_dir(&second_cwd)?; + + let responses = vec![ + create_shell_sse_response( + vec![ + "bash".to_string(), + "-lc".to_string(), + "echo first turn".to_string(), + ], + None, + Some(5000), + "call-first", + )?, + create_final_assistant_message_sse_response("done first")?, + create_shell_sse_response( + vec![ + "bash".to_string(), + "-lc".to_string(), + "echo second turn".to_string(), + ], + None, + Some(5000), + "call-second", + )?, + create_final_assistant_message_sse_response("done second")?, + ]; + let server = create_mock_chat_completions_server(responses).await; + create_config_toml_untrusted(&codex_home, &server.uri())?; + + let mut mcp = McpProcess::new(&codex_home).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // thread/start + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(start_resp)?; + + // first turn with workspace-write sandbox and first_cwd + let first_turn = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "first turn".to_string(), + }], + cwd: Some(first_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::WorkspaceWrite { + writable_roots: vec![first_cwd.clone()], + network_access: false, + exclude_tmpdir_env_var: false, + exclude_slash_tmp: false, + }), + model: Some("mock-model".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(first_turn)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + // second turn with workspace-write and second_cwd, ensure exec begins in second_cwd + let second_turn = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "second turn".to_string(), + }], + cwd: Some(second_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + model: Some("mock-model".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(second_turn)), + ) + .await??; + + let exec_begin_notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/exec_command_begin"), + ) + .await??; + let params = exec_begin_notification + .params + .clone() + .expect("exec_command_begin params"); + let event: Event = serde_json::from_value(params).expect("deserialize exec begin event"); + let exec_begin = match event.msg { + EventMsg::ExecCommandBegin(exec_begin) => exec_begin, + other => panic!("expected ExecCommandBegin event, got {other:?}"), + }; + assert_eq!(exec_begin.cwd, second_cwd); + assert_eq!( + exec_begin.command, + vec![ + "bash".to_string(), + "-lc".to_string(), + "echo second turn".to_string() + ] + ); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + Ok(()) +} + +// Helper to create a config.toml pointing at the mock model server. +fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "workspace-write" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + +fn create_config_toml_untrusted(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "untrusted" +sandbox_mode = "workspace-write" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "chat" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +}