Compare commits

..

13 Commits

Author SHA1 Message Date
Ahmed Ibrahim
a9dd1b2e11 proto 2025-10-23 12:12:48 -07:00
Ahmed Ibrahim
54f6a292bb progress 2025-10-23 08:16:49 -07:00
Ahmed Ibrahim
282e696079 feedback 2025-10-23 07:53:03 -07:00
Ahmed Ibrahim
bb26652be1 feedback 2025-10-23 07:52:05 -07:00
Ahmed Ibrahim
d830ec9678 clean 2025-10-22 16:25:52 -07:00
Ahmed Ibrahim
c3ef52f442 clean 2025-10-22 16:25:27 -07:00
Ahmed Ibrahim
d7ad216217 clean 2025-10-22 16:24:19 -07:00
Ahmed Ibrahim
e380b297d0 clean 2025-10-22 16:21:52 -07:00
Ahmed Ibrahim
d63e920ea4 tests 2025-10-22 16:17:05 -07:00
Ahmed Ibrahim
838d82e19c clean 2025-10-22 16:14:20 -07:00
Ahmed Ibrahim
89efcac5a3 clean 2025-10-22 16:12:33 -07:00
Ahmed Ibrahim
6dcb9ff39d clippy 2025-10-22 16:05:32 -07:00
Ahmed Ibrahim
3bf76f79f0 progress 2025-10-22 15:39:34 -07:00
57 changed files with 835 additions and 3448 deletions

10
codex-rs/Cargo.lock generated
View File

@@ -1066,7 +1066,6 @@ dependencies = [
"codex-rmcp-client",
"codex-utils-pty",
"codex-utils-string",
"codex-utils-tokenizer",
"core-foundation 0.9.4",
"core_test_support",
"dirs",
@@ -1634,7 +1633,6 @@ dependencies = [
"anyhow",
"assert_cmd",
"codex-core",
"codex-protocol",
"notify",
"regex-lite",
"serde_json",
@@ -4954,9 +4952,9 @@ dependencies = [
[[package]]
name = "rmcp"
version = "0.8.3"
version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fdad1258f7259fdc0f2dfc266939c82c3b5d1fd72bcde274d600cdc27e60243"
checksum = "4e35d31f89beb59c83bc31363426da25b323ce0c2e5b53c7bf29867d16ee7898"
dependencies = [
"base64",
"bytes",
@@ -4988,9 +4986,9 @@ dependencies = [
[[package]]
name = "rmcp-macros"
version = "0.8.3"
version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ede0589a208cc7ce81d1be68aa7e74b917fcd03c81528408bab0457e187dcd9b"
checksum = "d88518b38110c439a03f0f4eee40e5105d648a530711cb87f98991e3f324a664"
dependencies = [
"darling 0.21.3",
"proc-macro2",

View File

@@ -153,7 +153,7 @@ ratatui = "0.29.0"
ratatui-macros = "0.6.0"
regex-lite = "0.1.7"
reqwest = "0.12"
rmcp = { version = "0.8.3", default-features = false }
rmcp = { version = "0.8.2", default-features = false }
schemars = "0.8.22"
seccompiler = "0.5.0"
sentry = "0.34.0"

View File

@@ -5,7 +5,6 @@ use crate::JSONRPCNotification;
use crate::JSONRPCRequest;
use crate::RequestId;
use codex_protocol::ConversationId;
use codex_protocol::account::Account;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::ReasoningEffort;
use codex_protocol::config_types::ReasoningSummary;
@@ -94,43 +93,6 @@ macro_rules! client_request_definitions {
}
client_request_definitions! {
/// NEW APIs
#[serde(rename = "model/list")]
#[ts(rename = "model/list")]
ListModels {
params: ListModelsParams,
response: ListModelsResponse,
},
#[serde(rename = "account/login")]
#[ts(rename = "account/login")]
LoginAccount {
params: LoginAccountParams,
response: LoginAccountResponse,
},
#[serde(rename = "account/logout")]
#[ts(rename = "account/logout")]
LogoutAccount {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
response: LogoutAccountResponse,
},
#[serde(rename = "account/rateLimits/read")]
#[ts(rename = "account/rateLimits/read")]
GetAccountRateLimits {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
response: GetAccountRateLimitsResponse,
},
#[serde(rename = "account/read")]
#[ts(rename = "account/read")]
GetAccount {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
response: Option<Account>,
},
/// DEPRECATED APIs below
Initialize {
params: InitializeParams,
response: InitializeResponse,
@@ -144,6 +106,13 @@ client_request_definitions! {
params: ListConversationsParams,
response: ListConversationsResponse,
},
#[serde(rename = "model/list")]
#[ts(rename = "model/list")]
/// List available Codex models along with display metadata.
ListModels {
params: ListModelsParams,
response: ListModelsResponse,
},
/// Resume a recorded Codex conversation from a rollout file.
ResumeConversation {
params: ResumeConversationParams,
@@ -222,6 +191,12 @@ client_request_definitions! {
params: ExecOneOffCommandParams,
response: ExecOneOffCommandResponse,
},
#[serde(rename = "account/rateLimits/read")]
#[ts(rename = "account/rateLimits/read")]
GetAccountRateLimits {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
response: GetAccountRateLimitsResponse,
},
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)]
@@ -377,38 +352,6 @@ pub struct ListModelsResponse {
pub next_cursor: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(tag = "type")]
#[ts(tag = "type")]
pub enum LoginAccountParams {
#[serde(rename = "apiKey")]
#[ts(rename = "apiKey")]
ApiKey {
#[serde(rename = "apiKey")]
#[ts(rename = "apiKey")]
api_key: String,
},
#[serde(rename = "chatgpt")]
#[ts(rename = "chatgpt")]
ChatGpt,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct LoginAccountResponse {
/// Only set if the login method is ChatGPT.
#[schemars(with = "String")]
pub login_id: Option<Uuid>,
/// URL the client should open in a browser to initiate the OAuth flow.
/// Only set if the login method is ChatGPT.
pub auth_url: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct LogoutAccountResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
pub struct ResumeConversationParams {
@@ -932,13 +875,6 @@ pub struct AuthStatusChangeNotification {
#[serde(tag = "method", content = "params", rename_all = "camelCase")]
#[strum(serialize_all = "camelCase")]
pub enum ServerNotification {
/// NEW NOTIFICATIONS
#[serde(rename = "account/rateLimits/updated")]
#[ts(rename = "account/rateLimits/updated")]
#[strum(serialize = "account/rateLimits/updated")]
AccountRateLimitsUpdated(RateLimitSnapshot),
/// DEPRECATED NOTIFICATIONS below
/// Authentication status changed
AuthStatusChange(AuthStatusChangeNotification),
@@ -952,7 +888,6 @@ pub enum ServerNotification {
impl ServerNotification {
pub fn to_params(self) -> Result<serde_json::Value, serde_json::Error> {
match self {
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),
@@ -1108,89 +1043,16 @@ mod tests {
Ok(())
}
#[test]
fn serialize_account_login_api_key() -> Result<()> {
let request = ClientRequest::LoginAccount {
request_id: RequestId::Integer(2),
params: LoginAccountParams::ApiKey {
api_key: "secret".to_string(),
},
};
assert_eq!(
json!({
"method": "account/login",
"id": 2,
"params": {
"type": "apiKey",
"apiKey": "secret"
}
}),
serde_json::to_value(&request)?,
);
Ok(())
}
#[test]
fn serialize_account_login_chatgpt() -> Result<()> {
let request = ClientRequest::LoginAccount {
request_id: RequestId::Integer(3),
params: LoginAccountParams::ChatGpt,
};
assert_eq!(
json!({
"method": "account/login",
"id": 3,
"params": {
"type": "chatgpt"
}
}),
serde_json::to_value(&request)?,
);
Ok(())
}
#[test]
fn serialize_account_logout() -> Result<()> {
let request = ClientRequest::LogoutAccount {
request_id: RequestId::Integer(4),
params: None,
};
assert_eq!(
json!({
"method": "account/logout",
"id": 4,
}),
serde_json::to_value(&request)?,
);
Ok(())
}
#[test]
fn serialize_get_account() -> Result<()> {
let request = ClientRequest::GetAccount {
request_id: RequestId::Integer(5),
params: None,
};
assert_eq!(
json!({
"method": "account/read",
"id": 5,
}),
serde_json::to_value(&request)?,
);
Ok(())
}
#[test]
fn serialize_list_models() -> Result<()> {
let request = ClientRequest::ListModels {
request_id: RequestId::Integer(6),
request_id: RequestId::Integer(2),
params: ListModelsParams::default(),
};
assert_eq!(
json!({
"method": "model/list",
"id": 6,
"id": 2,
"params": {}
}),
serde_json::to_value(&request)?,

View File

@@ -176,27 +176,6 @@ impl CodexMessageProcessor {
ClientRequest::ListModels { request_id, params } => {
self.list_models(request_id, params).await;
}
ClientRequest::LoginAccount {
request_id,
params: _,
} => {
self.send_unimplemented_error(request_id, "account/login")
.await;
}
ClientRequest::LogoutAccount {
request_id,
params: _,
} => {
self.send_unimplemented_error(request_id, "account/logout")
.await;
}
ClientRequest::GetAccount {
request_id,
params: _,
} => {
self.send_unimplemented_error(request_id, "account/read")
.await;
}
ClientRequest::ResumeConversation { request_id, params } => {
self.handle_resume_conversation(request_id, params).await;
}
@@ -278,15 +257,6 @@ impl CodexMessageProcessor {
}
}
async fn send_unimplemented_error(&self, request_id: RequestId, method: &str) {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("{method} is not implemented yet"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
async fn login_api_key(&mut self, request_id: RequestId, params: LoginApiKeyParams) {
if matches!(
self.config.forced_login_method,
@@ -1466,15 +1436,6 @@ async fn apply_bespoke_event_handling(
on_exec_approval_response(event_id, rx, conversation).await;
});
}
EventMsg::TokenCount(token_count_event) => {
if let Some(rate_limits) = token_count_event.rate_limits {
outgoing
.send_server_notification(ServerNotification::AccountRateLimitsUpdated(
rate_limits,
))
.await;
}
}
// If this is a TurnAborted, reply to any pending interrupt requests.
EventMsg::TurnAborted(turn_aborted_event) => {
let pending = {

View File

@@ -46,7 +46,6 @@ pub(crate) async fn run_fuzzy_file_search(
threads,
cancel_flag,
COMPUTE_INDICES,
true,
) {
Ok(res) => Ok((root, res)),
Err(err) => Err((root, err)),

View File

@@ -142,8 +142,6 @@ pub(crate) struct OutgoingError {
#[cfg(test)]
mod tests {
use codex_app_server_protocol::LoginChatGptCompleteNotification;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::RateLimitWindow;
use pretty_assertions::assert_eq;
use serde_json::json;
use uuid::Uuid;
@@ -173,34 +171,4 @@ mod tests {
"ensure the strum macros serialize the method field correctly"
);
}
#[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 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,
},
}),
serde_json::to_value(jsonrpc_notification)
.expect("ensure the notification serializes correctly"),
"ensure the notification serializes correctly"
);
}
}

View File

@@ -1,3 +1 @@
mod cli;
#[cfg(not(target_os = "windows"))]
mod tool;

View File

@@ -1,257 +0,0 @@
use assert_cmd::Command;
use pretty_assertions::assert_eq;
use std::fs;
use std::path::Path;
use tempfile::tempdir;
fn run_apply_patch_in_dir(dir: &Path, patch: &str) -> anyhow::Result<assert_cmd::assert::Assert> {
let mut cmd = Command::cargo_bin("apply_patch")?;
cmd.current_dir(dir);
Ok(cmd.arg(patch).assert())
}
fn apply_patch_command(dir: &Path) -> anyhow::Result<Command> {
let mut cmd = Command::cargo_bin("apply_patch")?;
cmd.current_dir(dir);
Ok(cmd)
}
#[test]
fn test_apply_patch_cli_applies_multiple_operations() -> anyhow::Result<()> {
let tmp = tempdir()?;
let modify_path = tmp.path().join("modify.txt");
let delete_path = tmp.path().join("delete.txt");
fs::write(&modify_path, "line1\nline2\n")?;
fs::write(&delete_path, "obsolete\n")?;
let patch = "*** Begin Patch\n*** Add File: nested/new.txt\n+created\n*** Delete File: delete.txt\n*** Update File: modify.txt\n@@\n-line2\n+changed\n*** End Patch";
run_apply_patch_in_dir(tmp.path(), patch)?.success().stdout(
"Success. Updated the following files:\nA nested/new.txt\nM modify.txt\nD delete.txt\n",
);
assert_eq!(
fs::read_to_string(tmp.path().join("nested/new.txt"))?,
"created\n"
);
assert_eq!(fs::read_to_string(&modify_path)?, "line1\nchanged\n");
assert!(!delete_path.exists());
Ok(())
}
#[test]
fn test_apply_patch_cli_applies_multiple_chunks() -> anyhow::Result<()> {
let tmp = tempdir()?;
let target_path = tmp.path().join("multi.txt");
fs::write(&target_path, "line1\nline2\nline3\nline4\n")?;
let patch = "*** Begin Patch\n*** Update File: multi.txt\n@@\n-line2\n+changed2\n@@\n-line4\n+changed4\n*** End Patch";
run_apply_patch_in_dir(tmp.path(), patch)?
.success()
.stdout("Success. Updated the following files:\nM multi.txt\n");
assert_eq!(
fs::read_to_string(&target_path)?,
"line1\nchanged2\nline3\nchanged4\n"
);
Ok(())
}
#[test]
fn test_apply_patch_cli_moves_file_to_new_directory() -> anyhow::Result<()> {
let tmp = tempdir()?;
let original_path = tmp.path().join("old/name.txt");
let new_path = tmp.path().join("renamed/dir/name.txt");
fs::create_dir_all(original_path.parent().expect("parent should exist"))?;
fs::write(&original_path, "old content\n")?;
let patch = "*** Begin Patch\n*** Update File: old/name.txt\n*** Move to: renamed/dir/name.txt\n@@\n-old content\n+new content\n*** End Patch";
run_apply_patch_in_dir(tmp.path(), patch)?
.success()
.stdout("Success. Updated the following files:\nM renamed/dir/name.txt\n");
assert!(!original_path.exists());
assert_eq!(fs::read_to_string(&new_path)?, "new content\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_rejects_empty_patch() -> anyhow::Result<()> {
let tmp = tempdir()?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** End Patch")
.assert()
.failure()
.stderr("No files were modified.\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_reports_missing_context() -> anyhow::Result<()> {
let tmp = tempdir()?;
let target_path = tmp.path().join("modify.txt");
fs::write(&target_path, "line1\nline2\n")?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Update File: modify.txt\n@@\n-missing\n+changed\n*** End Patch")
.assert()
.failure()
.stderr("Failed to find expected lines in modify.txt:\nmissing\n");
assert_eq!(fs::read_to_string(&target_path)?, "line1\nline2\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_rejects_missing_file_delete() -> anyhow::Result<()> {
let tmp = tempdir()?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Delete File: missing.txt\n*** End Patch")
.assert()
.failure()
.stderr("Failed to delete file missing.txt\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_rejects_empty_update_hunk() -> anyhow::Result<()> {
let tmp = tempdir()?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Update File: foo.txt\n*** End Patch")
.assert()
.failure()
.stderr("Invalid patch hunk on line 2: Update file hunk for path 'foo.txt' is empty\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_requires_existing_file_for_update() -> anyhow::Result<()> {
let tmp = tempdir()?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Update File: missing.txt\n@@\n-old\n+new\n*** End Patch")
.assert()
.failure()
.stderr(
"Failed to read file to update missing.txt: No such file or directory (os error 2)\n",
);
Ok(())
}
#[test]
fn test_apply_patch_cli_move_overwrites_existing_destination() -> anyhow::Result<()> {
let tmp = tempdir()?;
let original_path = tmp.path().join("old/name.txt");
let destination = tmp.path().join("renamed/dir/name.txt");
fs::create_dir_all(original_path.parent().expect("parent should exist"))?;
fs::create_dir_all(destination.parent().expect("parent should exist"))?;
fs::write(&original_path, "from\n")?;
fs::write(&destination, "existing\n")?;
run_apply_patch_in_dir(
tmp.path(),
"*** Begin Patch\n*** Update File: old/name.txt\n*** Move to: renamed/dir/name.txt\n@@\n-from\n+new\n*** End Patch",
)?
.success()
.stdout("Success. Updated the following files:\nM renamed/dir/name.txt\n");
assert!(!original_path.exists());
assert_eq!(fs::read_to_string(&destination)?, "new\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_add_overwrites_existing_file() -> anyhow::Result<()> {
let tmp = tempdir()?;
let path = tmp.path().join("duplicate.txt");
fs::write(&path, "old content\n")?;
run_apply_patch_in_dir(
tmp.path(),
"*** Begin Patch\n*** Add File: duplicate.txt\n+new content\n*** End Patch",
)?
.success()
.stdout("Success. Updated the following files:\nA duplicate.txt\n");
assert_eq!(fs::read_to_string(&path)?, "new content\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_delete_directory_fails() -> anyhow::Result<()> {
let tmp = tempdir()?;
fs::create_dir(tmp.path().join("dir"))?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Delete File: dir\n*** End Patch")
.assert()
.failure()
.stderr("Failed to delete file dir\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_rejects_invalid_hunk_header() -> anyhow::Result<()> {
let tmp = tempdir()?;
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Frobnicate File: foo\n*** End Patch")
.assert()
.failure()
.stderr("Invalid patch hunk on line 2: '*** Frobnicate File: foo' is not a valid hunk header. Valid hunk headers: '*** Add File: {path}', '*** Delete File: {path}', '*** Update File: {path}'\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_updates_file_appends_trailing_newline() -> anyhow::Result<()> {
let tmp = tempdir()?;
let target_path = tmp.path().join("no_newline.txt");
fs::write(&target_path, "no newline at end")?;
run_apply_patch_in_dir(
tmp.path(),
"*** Begin Patch\n*** Update File: no_newline.txt\n@@\n-no newline at end\n+first line\n+second line\n*** End Patch",
)?
.success()
.stdout("Success. Updated the following files:\nM no_newline.txt\n");
let contents = fs::read_to_string(&target_path)?;
assert!(contents.ends_with('\n'));
assert_eq!(contents, "first line\nsecond line\n");
Ok(())
}
#[test]
fn test_apply_patch_cli_failure_after_partial_success_leaves_changes() -> anyhow::Result<()> {
let tmp = tempdir()?;
let new_file = tmp.path().join("created.txt");
apply_patch_command(tmp.path())?
.arg("*** Begin Patch\n*** Add File: created.txt\n+hello\n*** Update File: missing.txt\n@@\n-old\n+new\n*** End Patch")
.assert()
.failure()
.stdout("")
.stderr("Failed to read file to update missing.txt: No such file or directory (os error 2)\n");
assert_eq!(fs::read_to_string(&new_file)?, "hello\n");
Ok(())
}

View File

@@ -28,7 +28,6 @@ codex-rmcp-client = { workspace = true }
codex-async-utils = { workspace = true }
codex-utils-string = { workspace = true }
codex-utils-pty = { workspace = true }
codex-utils-tokenizer = { workspace = true }
dirs = { workspace = true }
dunce = { workspace = true }
env-flags = { workspace = true }

View File

@@ -11,7 +11,6 @@ use crate::mcp::auth::McpAuthStatusEntry;
use crate::parse_command::parse_command;
use crate::parse_turn_item;
use crate::response_processing::process_items;
use crate::review_format::format_review_findings_block;
use crate::terminal;
use crate::user_notification::UserNotifier;
use async_channel::Receiver;
@@ -20,7 +19,6 @@ use codex_apply_patch::ApplyPatchAction;
use codex_protocol::ConversationId;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::ConversationPathResponseEvent;
use codex_protocol::protocol::ExitedReviewModeEvent;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::ItemStartedEvent;
use codex_protocol::protocol::ReviewRequest;
@@ -47,7 +45,6 @@ use tokio_util::sync::CancellationToken;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::trace;
use tracing::warn;
use crate::ModelProviderInfo;
@@ -59,7 +56,6 @@ use crate::config::Config;
use crate::config_types::McpServerTransportConfig;
use crate::config_types::ShellEnvironmentPolicy;
use crate::conversation_history::ConversationHistory;
use crate::conversation_history::prefetch_tokenizer_in_background;
use crate::environment_context::EnvironmentContext;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
@@ -87,7 +83,6 @@ use crate::protocol::ListCustomPromptsResponseEvent;
use crate::protocol::Op;
use crate::protocol::RateLimitSnapshot;
use crate::protocol::ReviewDecision;
use crate::protocol::ReviewOutputEvent;
use crate::protocol::SandboxPolicy;
use crate::protocol::SessionConfiguredEvent;
use crate::protocol::StreamErrorEvent;
@@ -160,8 +155,6 @@ impl Codex {
conversation_history: InitialHistory,
session_source: SessionSource,
) -> CodexResult<CodexSpawnOk> {
// Start loading the tokenizer in the background so we don't block later.
prefetch_tokenizer_in_background();
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_event, rx_event) = async_channel::unbounded();
@@ -270,7 +263,6 @@ pub(crate) struct TurnContext {
pub(crate) sandbox_policy: SandboxPolicy,
pub(crate) shell_environment_policy: ShellEnvironmentPolicy,
pub(crate) tools_config: ToolsConfig,
pub(crate) is_review_mode: bool,
pub(crate) final_output_json_schema: Option<Value>,
pub(crate) codex_linux_sandbox_exe: Option<PathBuf>,
}
@@ -405,7 +397,6 @@ impl Session {
sandbox_policy: session_configuration.sandbox_policy.clone(),
shell_environment_policy: config.shell_environment_policy.clone(),
tools_config,
is_review_mode: false,
final_output_json_schema: None,
codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(),
}
@@ -571,9 +562,7 @@ impl Session {
// Dispatch the SessionConfiguredEvent first and then report any errors.
// If resuming, include converted initial messages in the payload so UIs can render them immediately.
let initial_messages = initial_history.get_event_msgs();
sess.record_initial_history(initial_history)
.await
.map_err(anyhow::Error::new)?;
sess.record_initial_history(initial_history).await;
let events = std::iter::once(Event {
id: INITIAL_SUBMIT_ID.to_owned(),
@@ -606,16 +595,13 @@ impl Session {
format!("auto-compact-{id}")
}
async fn record_initial_history(
&self,
conversation_history: InitialHistory,
) -> CodexResult<()> {
async fn record_initial_history(&self, conversation_history: InitialHistory) {
let turn_context = self.new_turn(SessionSettingsUpdate::default()).await;
match conversation_history {
InitialHistory::New => {
// Build and record initial items (user instructions + environment context)
let items = self.build_initial_context(&turn_context);
self.record_conversation_items(&items).await?;
self.record_conversation_items(&items).await;
}
InitialHistory::Resumed(_) | InitialHistory::Forked(_) => {
let rollout_items = conversation_history.get_rollout_items();
@@ -623,9 +609,9 @@ impl Session {
// Always add response items to conversation history
let reconstructed_history =
self.reconstruct_history_from_rollout(&turn_context, &rollout_items)?;
self.reconstruct_history_from_rollout(&turn_context, &rollout_items);
if !reconstructed_history.is_empty() {
self.record_into_history(&reconstructed_history).await?;
self.record_into_history(&reconstructed_history).await;
}
// If persisting, persist all rollout items as-is (recorder filters)
@@ -634,7 +620,6 @@ impl Session {
}
}
}
Ok(())
}
pub(crate) async fn update_settings(&self, updates: SessionSettingsUpdate) {
@@ -643,6 +628,14 @@ impl Session {
state.session_configuration = state.session_configuration.apply(&updates);
}
pub(crate) async fn base_config(&self) -> Arc<Config> {
let state = self.state.lock().await;
state
.session_configuration
.original_config_do_not_use
.clone()
}
pub(crate) async fn new_turn(&self, updates: SessionSettingsUpdate) -> Arc<TurnContext> {
let sub_id = self.next_internal_sub_id();
self.new_turn_with_sub_id(sub_id, updates).await
@@ -865,25 +858,21 @@ impl Session {
/// Records input items: always append to conversation history and
/// persist these response items to rollout.
pub(crate) async fn record_conversation_items(
&self,
items: &[ResponseItem],
) -> CodexResult<()> {
self.record_into_history(items).await?;
pub(crate) async fn record_conversation_items(&self, items: &[ResponseItem]) {
self.record_into_history(items).await;
self.persist_rollout_response_items(items).await;
Ok(())
}
fn reconstruct_history_from_rollout(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
) -> CodexResult<Vec<ResponseItem>> {
) -> Vec<ResponseItem> {
let mut history = ConversationHistory::new();
for item in rollout_items {
match item {
RolloutItem::ResponseItem(response_item) => {
history.record_items(std::iter::once(response_item))?;
history.record_items(std::iter::once(response_item));
}
RolloutItem::Compacted(compacted) => {
let snapshot = history.get_history();
@@ -898,14 +887,13 @@ impl Session {
_ => {}
}
}
Ok(history.get_history())
history.get_history()
}
/// Append ResponseItems to the in-memory conversation history only.
async fn record_into_history(&self, items: &[ResponseItem]) -> CodexResult<()> {
async fn record_into_history(&self, items: &[ResponseItem]) {
let mut state = self.state.lock().await;
state.record_items(items.iter())?;
Ok(())
state.record_items(items.iter());
}
async fn replace_history(&self, items: Vec<ResponseItem>) {
@@ -1014,11 +1002,11 @@ impl Session {
&self,
turn_context: &TurnContext,
response_input: &ResponseInputItem,
) -> CodexResult<()> {
) {
let response_item: ResponseItem = response_input.clone().into();
// Add to conversation history and persist response item to rollout
self.record_conversation_items(std::slice::from_ref(&response_item))
.await?;
.await;
// Derive user message events and persist only UserMessage to rollout
let turn_item = parse_turn_item(&response_item);
@@ -1027,7 +1015,6 @@ impl Session {
self.emit_turn_item_started_completed(turn_context, item, false)
.await;
}
Ok(())
}
/// Helper that emits a BackgroundEvent with the given message. This keeps
@@ -1208,17 +1195,9 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
if let Err(items) = sess.inject_input(items).await {
if let Some(env_item) = sess
.build_environment_update_item(previous_context.as_ref(), &current_context)
&& let Err(err) = sess
.record_conversation_items(std::slice::from_ref(&env_item))
.await
{
sess.send_event(
current_context.as_ref(),
EventMsg::Error(ErrorEvent {
message: err.to_string(),
}),
)
.await;
sess.record_conversation_items(std::slice::from_ref(&env_item))
.await;
}
sess.spawn_task(Arc::clone(&current_context), items, RegularTask)
@@ -1491,7 +1470,6 @@ async fn spawn_review_thread(
sandbox_policy: parent_turn_context.sandbox_policy.clone(),
shell_environment_policy: parent_turn_context.shell_environment_policy.clone(),
cwd: parent_turn_context.cwd.clone(),
is_review_mode: true,
final_output_json_schema: None,
codex_linux_sandbox_exe: parent_turn_context.codex_linux_sandbox_exe.clone(),
};
@@ -1531,9 +1509,9 @@ pub(crate) async fn run_task(
input: Vec<UserInput>,
task_kind: TaskKind,
cancellation_token: CancellationToken,
) -> CodexResult<Option<String>> {
) -> Option<String> {
if input.is_empty() {
return Ok(None);
return None;
}
let event = EventMsg::TaskStarted(TaskStartedEvent {
model_context_window: turn_context.client.get_model_context_window(),
@@ -1541,21 +1519,8 @@ pub(crate) async fn run_task(
sess.send_event(&turn_context, event).await;
let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
// For review threads, keep an isolated in-memory history so the
// model sees a fresh conversation without the parent session's history.
// For normal turns, continue recording to the session history as before.
let is_review_mode = turn_context.is_review_mode;
let mut review_thread_history: ConversationHistory = ConversationHistory::new();
if is_review_mode {
// Seed review threads with environment context so the model knows the working directory.
review_thread_history
.record_items(sess.build_initial_context(turn_context.as_ref()).iter())?;
review_thread_history.record_items(std::iter::once(&initial_input_for_turn.into()))?;
} else {
sess.record_input_and_rollout_usermsg(turn_context.as_ref(), &initial_input_for_turn)
.await?;
}
sess.record_input_and_rollout_usermsg(turn_context.as_ref(), &initial_input_for_turn)
.await;
let mut last_agent_message: Option<String> = None;
// Although from the perspective of codex.rs, TurnDiffTracker has the lifecycle of a Task which contains
@@ -1584,13 +1549,8 @@ pub(crate) async fn run_task(
// conversation history on each turn. The rollout file, however, should
// only record the new items that originated in this turn so that it
// represents an append-only log without duplicates.
let turn_input: Vec<ResponseItem> = if is_review_mode {
if !pending_input.is_empty() {
review_thread_history.record_items(&pending_input)?;
}
review_thread_history.get_history()
} else {
sess.record_conversation_items(&pending_input).await?;
let turn_input: Vec<ResponseItem> = {
sess.record_conversation_items(&pending_input).await;
sess.history_snapshot().await
};
@@ -1632,13 +1592,8 @@ pub(crate) async fn run_task(
let token_limit_reached = total_usage_tokens
.map(|tokens| tokens >= limit)
.unwrap_or(false);
let (responses, items_to_record_in_conversation_history) = process_items(
processed_items,
is_review_mode,
&mut review_thread_history,
&sess,
)
.await?;
let (responses, items_to_record_in_conversation_history) =
process_items(processed_items, &sess).await;
if token_limit_reached {
if auto_compact_recently_attempted {
@@ -1655,8 +1610,7 @@ pub(crate) async fn run_task(
break;
}
auto_compact_recently_attempted = true;
compact::run_inline_auto_compact_task(sess.clone(), turn_context.clone())
.await?;
compact::run_inline_auto_compact_task(sess.clone(), turn_context.clone()).await;
continue;
}
@@ -1681,13 +1635,7 @@ pub(crate) async fn run_task(
Err(CodexErr::TurnAborted {
dangling_artifacts: processed_items,
}) => {
process_items(
processed_items,
is_review_mode,
&mut review_thread_history,
&sess,
)
.await?;
let _ = process_items(processed_items, &sess).await;
// Aborted turn is reported via a different event.
break;
}
@@ -1703,48 +1651,7 @@ pub(crate) async fn run_task(
}
}
// If this was a review thread and we have a final assistant message,
// try to parse it as a ReviewOutput.
//
// If parsing fails, construct a minimal ReviewOutputEvent using the plain
// text as the overall explanation. Else, just exit review mode with None.
//
// Emits an ExitedReviewMode event with the parsed review output.
if turn_context.is_review_mode {
exit_review_mode(
sess.clone(),
Arc::clone(&turn_context),
last_agent_message.as_deref().map(parse_review_output_event),
)
.await?;
}
Ok(last_agent_message)
}
/// Parse the review output; when not valid JSON, build a structured
/// fallback that carries the plain text as the overall explanation.
///
/// Returns: a ReviewOutputEvent parsed from JSON or a fallback populated from text.
fn parse_review_output_event(text: &str) -> ReviewOutputEvent {
// Try direct parse first
if let Ok(ev) = serde_json::from_str::<ReviewOutputEvent>(text) {
return ev;
}
// If wrapped in markdown fences or extra prose, attempt to extract the first JSON object
if let (Some(start), Some(end)) = (text.find('{'), text.rfind('}'))
&& start < end
&& let Some(slice) = text.get(start..=end)
&& let Ok(ev) = serde_json::from_str::<ReviewOutputEvent>(slice)
{
return ev;
}
// Not JSON return a structured ReviewOutputEvent that carries
// the plain text as the overall explanation.
ReviewOutputEvent {
overall_explanation: text.to_string(),
..Default::default()
}
last_agent_message
}
async fn run_turn(
@@ -2033,12 +1940,8 @@ async fn try_run_turn(
ResponseEvent::OutputTextDelta(delta) => {
// In review child threads, suppress assistant text deltas; the
// UI will show a selection popup from the final ReviewOutput.
if !turn_context.is_review_mode {
let event = EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta });
sess.send_event(&turn_context, event).await;
} else {
trace!("suppressing OutputTextDelta in review mode");
}
let event = EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta });
sess.send_event(&turn_context, event).await;
}
ResponseEvent::ReasoningSummaryDelta(delta) => {
let event = EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta });
@@ -2073,13 +1976,7 @@ async fn handle_non_tool_response_item(
ResponseItem::Message { .. }
| ResponseItem::Reasoning { .. }
| ResponseItem::WebSearchCall { .. } => {
let turn_item = match &item {
ResponseItem::Message { .. } if turn_context.is_review_mode => {
trace!("suppressing assistant Message in review mode");
None
}
_ => parse_turn_item(&item),
};
let turn_item = parse_turn_item(&item);
if let Some(turn_item) = turn_item {
sess.emit_turn_item_started_completed(
turn_context.as_ref(),
@@ -2152,59 +2049,6 @@ pub(crate) fn convert_call_tool_result_to_function_call_output_payload(
}
}
/// Emits an ExitedReviewMode Event with optional ReviewOutput,
/// and records a developer message with the review output.
pub(crate) async fn exit_review_mode(
session: Arc<Session>,
turn_context: Arc<TurnContext>,
review_output: Option<ReviewOutputEvent>,
) -> CodexResult<()> {
let event = EventMsg::ExitedReviewMode(ExitedReviewModeEvent {
review_output: review_output.clone(),
});
session.send_event(turn_context.as_ref(), event).await;
let mut user_message = String::new();
if let Some(out) = review_output {
let mut findings_str = String::new();
let text = out.overall_explanation.trim();
if !text.is_empty() {
findings_str.push_str(text);
}
if !out.findings.is_empty() {
let block = format_review_findings_block(&out.findings, None);
findings_str.push_str(&format!("\n{block}"));
}
user_message.push_str(&format!(
r#"<user_action>
<context>User initiated a review task. Here's the full review output from reviewer model. User may select one or more comments to resolve.</context>
<action>review</action>
<results>
{findings_str}
</results>
</user_action>
"#));
} else {
user_message.push_str(r#"<user_action>
<context>User initiated a review task, but was interrupted. If user asks about this, tell them to re-initiate a review with `/review` and wait for it to complete.</context>
<action>review</action>
<results>
None.
</results>
</user_action>
"#);
}
session
.record_conversation_items(&[ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: user_message }],
}])
.await?;
Ok(())
}
fn mcp_init_error_display(
server_name: &str,
entry: Option<&McpAuthStatusEntry>,
@@ -2250,7 +2094,6 @@ mod tests {
use crate::config::ConfigToml;
use crate::config_types::McpServerConfig;
use crate::config_types::McpServerTransportConfig;
use crate::error::Result as CodexResult;
use crate::exec::ExecToolCallOutput;
use crate::mcp::auth::McpAuthStatusEntry;
use crate::tools::format_exec_output_str;
@@ -2266,11 +2109,7 @@ mod tests {
use crate::tools::MODEL_FORMAT_MAX_LINES;
use crate::tools::MODEL_FORMAT_TAIL_LINES;
use crate::tools::ToolRouter;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
use crate::tools::handlers::ShellHandler;
use crate::tools::registry::ToolHandler;
use crate::tools::handle_container_exec_with_params;
use crate::turn_diff_tracker::TurnDiffTracker;
use codex_app_server_protocol::AuthMode;
use codex_protocol::models::ContentItem;
@@ -2291,12 +2130,9 @@ mod tests {
#[test]
fn reconstruct_history_matches_live_compactions() {
let (session, turn_context) = make_session_and_context();
let (rollout_items, expected) =
sample_rollout(&session, &turn_context).expect("sample rollout");
let (rollout_items, expected) = sample_rollout(&session, &turn_context);
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.expect("reconstruct history");
let reconstructed = session.reconstruct_history_from_rollout(&turn_context, &rollout_items);
assert_eq!(expected, reconstructed);
}
@@ -2304,19 +2140,15 @@ mod tests {
#[test]
fn record_initial_history_reconstructs_resumed_transcript() {
let (session, turn_context) = make_session_and_context();
let (rollout_items, expected) =
sample_rollout(&session, &turn_context).expect("sample rollout");
let (rollout_items, expected) = sample_rollout(&session, &turn_context);
tokio_test::block_on(async {
session
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ConversationId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
}))
.await
.expect("record resumed history");
});
tokio_test::block_on(session.record_initial_history(InitialHistory::Resumed(
ResumedHistory {
conversation_id: ConversationId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
},
)));
let actual = tokio_test::block_on(async { session.state.lock().await.history_snapshot() });
assert_eq!(expected, actual);
@@ -2325,15 +2157,9 @@ mod tests {
#[test]
fn record_initial_history_reconstructs_forked_transcript() {
let (session, turn_context) = make_session_and_context();
let (rollout_items, expected) =
sample_rollout(&session, &turn_context).expect("sample rollout");
let (rollout_items, expected) = sample_rollout(&session, &turn_context);
tokio_test::block_on(async {
session
.record_initial_history(InitialHistory::Forked(rollout_items))
.await
.expect("record forked history");
});
tokio_test::block_on(session.record_initial_history(InitialHistory::Forked(rollout_items)));
let actual = tokio_test::block_on(async { session.state.lock().await.history_snapshot() });
assert_eq!(expected, actual);
@@ -2693,21 +2519,15 @@ mod tests {
_ctx: Arc<TurnContext>,
_input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> CodexResult<Option<String>> {
) -> Option<String> {
if self.listen_to_cancellation_token {
cancellation_token.cancelled().await;
return Ok(None);
return None;
}
loop {
sleep(Duration::from_secs(60)).await;
}
}
async fn abort(&self, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>) {
if let TaskKind::Review = self.kind {
let _ = exit_review_mode(session.clone_session(), ctx, None).await;
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -2861,7 +2681,7 @@ mod tests {
fn sample_rollout(
session: &Session,
turn_context: &TurnContext,
) -> CodexResult<(Vec<RolloutItem>, Vec<ResponseItem>)> {
) -> (Vec<RolloutItem>, Vec<ResponseItem>) {
let mut rollout_items = Vec::new();
let mut live_history = ConversationHistory::new();
@@ -2869,7 +2689,7 @@ mod tests {
for item in &initial_context {
rollout_items.push(RolloutItem::ResponseItem(item.clone()));
}
live_history.record_items(initial_context.iter())?;
live_history.record_items(initial_context.iter());
let user1 = ResponseItem::Message {
id: None,
@@ -2878,7 +2698,7 @@ mod tests {
text: "first user".to_string(),
}],
};
live_history.record_items(std::iter::once(&user1))?;
live_history.record_items(std::iter::once(&user1));
rollout_items.push(RolloutItem::ResponseItem(user1.clone()));
let assistant1 = ResponseItem::Message {
@@ -2888,7 +2708,7 @@ mod tests {
text: "assistant reply one".to_string(),
}],
};
live_history.record_items(std::iter::once(&assistant1))?;
live_history.record_items(std::iter::once(&assistant1));
rollout_items.push(RolloutItem::ResponseItem(assistant1.clone()));
let summary1 = "summary one";
@@ -2911,7 +2731,7 @@ mod tests {
text: "second user".to_string(),
}],
};
live_history.record_items(std::iter::once(&user2))?;
live_history.record_items(std::iter::once(&user2));
rollout_items.push(RolloutItem::ResponseItem(user2.clone()));
let assistant2 = ResponseItem::Message {
@@ -2921,7 +2741,7 @@ mod tests {
text: "assistant reply two".to_string(),
}],
};
live_history.record_items(std::iter::once(&assistant2))?;
live_history.record_items(std::iter::once(&assistant2));
rollout_items.push(RolloutItem::ResponseItem(assistant2.clone()));
let summary2 = "summary two";
@@ -2944,7 +2764,7 @@ mod tests {
text: "third user".to_string(),
}],
};
live_history.record_items(std::iter::once(&user3))?;
live_history.record_items(std::iter::once(&user3));
rollout_items.push(RolloutItem::ResponseItem(user3.clone()));
let assistant3 = ResponseItem::Message {
@@ -2954,10 +2774,10 @@ mod tests {
text: "assistant reply three".to_string(),
}],
};
live_history.record_items(std::iter::once(&assistant3))?;
live_history.record_items(std::iter::once(&assistant3));
rollout_items.push(RolloutItem::ResponseItem(assistant3.clone()));
Ok((rollout_items, live_history.get_history()))
(rollout_items, live_history.get_history())
}
#[tokio::test]
@@ -3006,26 +2826,15 @@ mod tests {
let tool_name = "shell";
let call_id = "test-call".to_string();
let handler = ShellHandler;
let resp = handler
.handle(ToolInvocation {
session: Arc::clone(&session),
turn: Arc::clone(&turn_context),
tracker: Arc::clone(&turn_diff_tracker),
call_id,
tool_name: tool_name.to_string(),
payload: ToolPayload::Function {
arguments: serde_json::json!({
"command": params.command.clone(),
"workdir": Some(turn_context.cwd.to_string_lossy().to_string()),
"timeout_ms": params.timeout_ms,
"with_escalated_permissions": params.with_escalated_permissions,
"justification": params.justification.clone(),
})
.to_string(),
},
})
.await;
let resp = handle_container_exec_with_params(
tool_name,
params,
Arc::clone(&session),
Arc::clone(&turn_context),
Arc::clone(&turn_diff_tracker),
call_id,
)
.await;
let Err(FunctionCallError::RespondToModel(output)) = resp else {
panic!("expected error result");
@@ -3044,30 +2853,17 @@ mod tests {
.expect("unique turn context Arc")
.sandbox_policy = SandboxPolicy::DangerFullAccess;
let resp2 = handler
.handle(ToolInvocation {
session: Arc::clone(&session),
turn: Arc::clone(&turn_context),
tracker: Arc::clone(&turn_diff_tracker),
call_id: "test-call-2".to_string(),
tool_name: tool_name.to_string(),
payload: ToolPayload::Function {
arguments: serde_json::json!({
"command": params2.command.clone(),
"workdir": Some(turn_context.cwd.to_string_lossy().to_string()),
"timeout_ms": params2.timeout_ms,
"with_escalated_permissions": params2.with_escalated_permissions,
"justification": params2.justification.clone(),
})
.to_string(),
},
})
.await;
let resp2 = handle_container_exec_with_params(
tool_name,
params2,
Arc::clone(&session),
Arc::clone(&turn_context),
Arc::clone(&turn_diff_tracker),
"test-call-2".to_string(),
)
.await;
let output = match resp2.expect("expected Ok result") {
ToolOutput::Function { content, .. } => content,
_ => panic!("unexpected tool output"),
};
let output = resp2.expect("expected Ok result");
#[derive(Deserialize, PartialEq, Eq, Debug)]
struct ResponseExecMetadata {

View File

@@ -39,35 +39,35 @@ struct HistoryBridgeTemplate<'a> {
pub(crate) async fn run_inline_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
) {
let input = vec![UserInput::Text {
text: SUMMARIZATION_PROMPT.to_string(),
}];
run_compact_task_inner(sess, turn_context, input).await
run_compact_task_inner(sess, turn_context, input).await;
}
pub(crate) async fn run_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
input: Vec<UserInput>,
) -> CodexResult<Option<String>> {
) -> Option<String> {
let start_event = EventMsg::TaskStarted(TaskStartedEvent {
model_context_window: turn_context.client.get_model_context_window(),
});
sess.send_event(&turn_context, start_event).await;
run_compact_task_inner(sess.clone(), turn_context, input).await?;
Ok(None)
run_compact_task_inner(sess.clone(), turn_context, input).await;
None
}
async fn run_compact_task_inner(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
input: Vec<UserInput>,
) -> CodexResult<()> {
) {
let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
let mut history = sess.clone_history().await;
history.record_items(&[initial_input_for_turn.into()])?;
history.record_items(&[initial_input_for_turn.into()]);
let mut truncated_count = 0usize;
@@ -106,7 +106,7 @@ async fn run_compact_task_inner(
break;
}
Err(CodexErr::Interrupted) => {
return Ok(());
return;
}
Err(e @ CodexErr::ContextWindowExceeded) => {
if turn_input.len() > 1 {
@@ -124,7 +124,7 @@ async fn run_compact_task_inner(
message: e.to_string(),
});
sess.send_event(&turn_context, event).await;
return Ok(());
return;
}
Err(e) => {
if retries < max_retries {
@@ -142,7 +142,7 @@ async fn run_compact_task_inner(
message: e.to_string(),
});
sess.send_event(&turn_context, event).await;
return Ok(());
return;
}
}
}
@@ -164,7 +164,6 @@ async fn run_compact_task_inner(
message: "Compact task completed".to_string(),
});
sess.send_event(&turn_context, event).await;
Ok(())
}
pub fn content_items_to_text(content: &[ContentItem]) -> Option<String> {
@@ -253,8 +252,7 @@ async fn drain_to_completed(
};
match event {
Ok(ResponseEvent::OutputItemDone(item)) => {
sess.record_into_history(std::slice::from_ref(&item))
.await?;
sess.record_into_history(std::slice::from_ref(&item)).await;
}
Ok(ResponseEvent::RateLimits(snapshot)) => {
sess.update_rate_limits(turn_context, snapshot).await;

View File

@@ -1,49 +1,21 @@
use std::sync::Arc;
use std::sync::OnceLock;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_utils_tokenizer::Tokenizer;
use tokio::task;
use tracing::error;
static TOKENIZER: OnceLock<Option<Arc<Tokenizer>>> = OnceLock::new();
use crate::error::CodexErr;
/// Transcript of conversation history
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Default)]
pub(crate) struct ConversationHistory {
/// The oldest items are at the beginning of the vector.
items: Vec<ResponseItem>,
token_info: Option<TokenUsageInfo>,
}
impl ConversationHistory {
pub(crate) fn new() -> Self {
Self {
items: Vec::new(),
token_info: None,
}
}
pub(crate) fn token_info(&self) -> Option<TokenUsageInfo> {
self.token_info.clone()
}
pub(crate) fn set_token_usage_full(&mut self, context_window: i64) {
match &mut self.token_info {
Some(info) => info.fill_to_context_window(context_window),
None => {
self.token_info = Some(TokenUsageInfo::full_context_window(context_window));
}
}
Self { items: Vec::new() }
}
/// `items` is ordered from oldest to newest.
pub(crate) fn record_items<I>(&mut self, items: I) -> Result<(), CodexErr>
pub(crate) fn record_items<I>(&mut self, items: I)
where
I: IntoIterator,
I::Item: std::ops::Deref<Target = ResponseItem>,
@@ -52,10 +24,9 @@ impl ConversationHistory {
if !is_api_message(&item) {
continue;
}
self.validate_input(&item)?;
self.items.push(item.clone());
}
Ok(())
}
pub(crate) fn get_history(&mut self) -> Vec<ResponseItem> {
@@ -91,65 +62,6 @@ impl ConversationHistory {
self.items.clone()
}
fn validate_input(&self, item: &ResponseItem) -> Result<(), CodexErr> {
match item {
ResponseItem::Message { content, .. } => {
self.validate_input_content_item(content)?;
Ok(())
}
ResponseItem::FunctionCall { .. }
| ResponseItem::FunctionCallOutput { .. }
| ResponseItem::CustomToolCall { .. }
| ResponseItem::CustomToolCallOutput { .. }
| ResponseItem::LocalShellCall { .. }
| ResponseItem::Reasoning { .. }
| ResponseItem::WebSearchCall { .. } => Ok(()),
ResponseItem::Other => Err(CodexErr::InvalidInput(format!("invalid input: {item:?}"))),
}
}
fn validate_input_content_item(
&self,
content: &[codex_protocol::models::ContentItem],
) -> Result<(), CodexErr> {
let Some(info) = &self.token_info else {
return Ok(());
};
// this will intentionally not check the context for the first turn before getting this information.
// it's acceptable tradeoff.
let Some(context_window) = info.model_context_window else {
return Ok(());
};
let tokenizer = match shared_tokenizer() {
Some(t) => t,
None => return Ok(()),
};
let mut input_tokens: i64 = 0;
for item in content {
match item {
codex_protocol::models::ContentItem::InputText { text } => {
input_tokens += tokenizer.count(text);
}
codex_protocol::models::ContentItem::InputImage { .. } => {
// no validation currently
}
codex_protocol::models::ContentItem::OutputText { .. } => {
// no validation currently
}
}
}
let prior_total = info.last_token_usage.total_tokens;
let combined_tokens = prior_total.saturating_add(input_tokens);
let threshold = context_window * 95 / 100;
if combined_tokens > threshold {
return Err(CodexErr::InvalidInput("input too large".to_string()));
}
Ok(())
}
fn ensure_call_outputs_present(&mut self) {
// Collect synthetic outputs to insert immediately after their calls.
// Store the insertion position (index of call) alongside the item so
@@ -389,18 +301,6 @@ impl ConversationHistory {
self.items.remove(pos);
}
}
pub(crate) fn update_token_info(
&mut self,
usage: &TokenUsage,
model_context_window: Option<i64>,
) {
self.token_info = TokenUsageInfo::new_or_append(
&self.token_info,
&Some(usage.clone()),
model_context_window,
);
}
}
#[inline]
@@ -412,36 +312,6 @@ fn error_or_panic(message: String) {
}
}
fn shared_tokenizer() -> Option<Arc<Tokenizer>> {
TOKENIZER.get().and_then(|opt| opt.as_ref().map(Arc::clone))
}
/// Kick off background initialization of the shared tokenizer without blocking the caller.
pub(crate) fn prefetch_tokenizer_in_background() {
if TOKENIZER.get().is_some() {
return;
}
// Spawn a background task to initialize the tokenizer. Use spawn_blocking in case
// initialization performs CPU-heavy work or file I/O.
tokio::spawn(async {
let result = task::spawn_blocking(Tokenizer::try_default).await;
match result {
Ok(Ok(tokenizer)) => {
let _ = TOKENIZER.set(Some(Arc::new(tokenizer)));
}
Ok(Err(error)) => {
error!("failed to create tokenizer: {error}");
let _ = TOKENIZER.set(None);
}
Err(join_error) => {
error!("failed to join tokenizer init task: {join_error}");
let _ = TOKENIZER.set(None);
}
}
});
}
/// Anything that is not a system message or "reasoning" message is considered
/// an API message.
fn is_api_message(message: &ResponseItem) -> bool {
@@ -480,7 +350,7 @@ mod tests {
fn create_history_with_items(items: Vec<ResponseItem>) -> ConversationHistory {
let mut h = ConversationHistory::new();
h.record_items(items.iter()).unwrap();
h.record_items(items.iter());
h
}
@@ -496,7 +366,7 @@ mod tests {
#[test]
fn filters_non_api_messages() {
let mut h = ConversationHistory::new();
let mut h = ConversationHistory::default();
// System message is not an API message; Other is ignored.
let system = ResponseItem::Message {
id: None,
@@ -505,12 +375,12 @@ mod tests {
text: "ignored".to_string(),
}],
};
h.record_items([&system, &ResponseItem::Other]).unwrap();
h.record_items([&system, &ResponseItem::Other]);
// User and assistant should be retained.
let u = user_msg("hi");
let a = assistant_msg("hello");
h.record_items([&u, &a]).unwrap();
h.record_items([&u, &a]);
let items = h.contents();
assert_eq!(

View File

@@ -158,9 +158,6 @@ pub enum CodexErr {
#[error("{0}")]
EnvVar(EnvVarError),
#[error("invalid input: {0}")]
InvalidInput(String),
}
impl From<CancelErr> for CodexErr {

View File

@@ -14,6 +14,7 @@ mod client_common;
pub mod codex;
mod codex_conversation;
pub use codex_conversation::CodexConversation;
mod codex_delegate;
mod command_safety;
pub mod config;
pub mod config_edit;

View File

@@ -1,6 +1,4 @@
use crate::codex::Session;
use crate::conversation_history::ConversationHistory;
use crate::error::Result as CodexResult;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
@@ -11,10 +9,8 @@ use tracing::warn;
/// - `ResponseInputItem`s to send back to the model on the next turn.
pub(crate) async fn process_items(
processed_items: Vec<crate::codex::ProcessedResponseItem>,
is_review_mode: bool,
review_thread_history: &mut ConversationHistory,
sess: &Session,
) -> CodexResult<(Vec<ResponseInputItem>, Vec<ResponseItem>)> {
) -> (Vec<ResponseInputItem>, Vec<ResponseItem>) {
let mut items_to_record_in_conversation_history = Vec::<ResponseItem>::new();
let mut responses = Vec::<ResponseInputItem>::new();
for processed_response_item in processed_items {
@@ -102,12 +98,8 @@ pub(crate) async fn process_items(
// Only attempt to take the lock if there is something to record.
if !items_to_record_in_conversation_history.is_empty() {
if is_review_mode {
review_thread_history.record_items(items_to_record_in_conversation_history.iter())?;
} else {
sess.record_conversation_items(&items_to_record_in_conversation_history)
.await?;
}
sess.record_conversation_items(&items_to_record_in_conversation_history)
.await;
}
Ok((responses, items_to_record_in_conversation_history))
(responses, items_to_record_in_conversation_history)
}

View File

@@ -1,11 +1,12 @@
use std::cmp::Reverse;
use std::io::{self};
use std::num::NonZero;
use std::path::Path;
use std::path::PathBuf;
use codex_file_search as file_search;
use std::num::NonZero;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use time::OffsetDateTime;
use time::PrimitiveDateTime;
use time::format_description::FormatItem;
@@ -14,7 +15,6 @@ use uuid::Uuid;
use super::SESSIONS_SUBDIR;
use crate::protocol::EventMsg;
use codex_file_search as file_search;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionSource;
@@ -515,7 +515,6 @@ pub async fn find_conversation_path_by_id_str(
threads,
cancel,
compute_indices,
false,
)
.map_err(|e| io::Error::other(format!("file search failed: {e}")))?;

View File

@@ -4,7 +4,6 @@ use codex_protocol::models::ResponseItem;
use crate::codex::SessionConfiguration;
use crate::conversation_history::ConversationHistory;
use crate::error::CodexErr;
use crate::protocol::RateLimitSnapshot;
use crate::protocol::TokenUsage;
use crate::protocol::TokenUsageInfo;
@@ -13,6 +12,7 @@ use crate::protocol::TokenUsageInfo;
pub(crate) struct SessionState {
pub(crate) session_configuration: SessionConfiguration,
pub(crate) history: ConversationHistory,
pub(crate) token_info: Option<TokenUsageInfo>,
pub(crate) latest_rate_limits: Option<RateLimitSnapshot>,
}
@@ -22,18 +22,18 @@ impl SessionState {
Self {
session_configuration,
history: ConversationHistory::new(),
token_info: None,
latest_rate_limits: None,
}
}
// History helpers
pub(crate) fn record_items<I>(&mut self, items: I) -> Result<(), CodexErr>
pub(crate) fn record_items<I>(&mut self, items: I)
where
I: IntoIterator,
I::Item: std::ops::Deref<Target = ResponseItem>,
{
self.history.record_items(items)?;
Ok(())
self.history.record_items(items)
}
pub(crate) fn history_snapshot(&mut self) -> Vec<ResponseItem> {
@@ -54,11 +54,11 @@ impl SessionState {
usage: &TokenUsage,
model_context_window: Option<i64>,
) {
self.history.update_token_info(usage, model_context_window);
}
pub(crate) fn token_info(&self) -> Option<TokenUsageInfo> {
self.history.token_info()
self.token_info = TokenUsageInfo::new_or_append(
&self.token_info,
&Some(usage.clone()),
model_context_window,
);
}
pub(crate) fn set_rate_limits(&mut self, snapshot: RateLimitSnapshot) {
@@ -68,17 +68,17 @@ impl SessionState {
pub(crate) fn token_info_and_rate_limits(
&self,
) -> (Option<TokenUsageInfo>, Option<RateLimitSnapshot>) {
let info = self.token_info().and_then(|info| {
if info.total_token_usage.is_zero() && info.last_token_usage.is_zero() {
None
} else {
Some(info)
}
});
(info, self.latest_rate_limits.clone())
(self.token_info.clone(), self.latest_rate_limits.clone())
}
pub(crate) fn set_token_usage_full(&mut self, context_window: i64) {
self.history.set_token_usage_full(context_window);
match &mut self.token_info {
Some(info) => info.fill_to_context_window(context_window),
None => {
self.token_info = Some(TokenUsageInfo::full_context_window(context_window));
}
}
}
// Pending input/approval moved to TurnState.
}

View File

@@ -5,7 +5,6 @@ use tokio_util::sync::CancellationToken;
use crate::codex::TurnContext;
use crate::codex::compact;
use crate::error::Result as CodexResult;
use crate::state::TaskKind;
use codex_protocol::user_input::UserInput;
@@ -27,7 +26,7 @@ impl SessionTask for CompactTask {
ctx: Arc<TurnContext>,
input: Vec<UserInput>,
_cancellation_token: CancellationToken,
) -> CodexResult<Option<String>> {
) -> Option<String> {
compact::run_compact_task(session.clone_session(), ctx, input).await
}
}

View File

@@ -13,10 +13,10 @@ use tokio_util::task::AbortOnDropHandle;
use tracing::trace;
use tracing::warn;
use crate::AuthManager;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::error::Result as CodexResult;
use crate::protocol::ErrorEvent;
use crate::config::Config;
use crate::protocol::EventMsg;
use crate::protocol::TaskCompleteEvent;
use crate::protocol::TurnAbortReason;
@@ -46,6 +46,14 @@ impl SessionTaskContext {
pub(crate) fn clone_session(&self) -> Arc<Session> {
Arc::clone(&self.session)
}
pub(crate) fn auth_manager(&self) -> Arc<AuthManager> {
Arc::clone(&self.session.services.auth_manager)
}
pub(crate) async fn base_config(&self) -> Arc<Config> {
self.session.base_config().await
}
}
#[async_trait]
@@ -58,7 +66,7 @@ pub(crate) trait SessionTask: Send + Sync + 'static {
ctx: Arc<TurnContext>,
input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> CodexResult<Option<String>>;
) -> Option<String>;
async fn abort(&self, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>) {
let _ = (session, ctx);
@@ -88,7 +96,7 @@ impl Session {
let task_cancellation_token = cancellation_token.child_token();
tokio::spawn(async move {
let ctx_for_finish = Arc::clone(&ctx);
let run_result = task_for_run
let last_agent_message = task_for_run
.run(
Arc::clone(&session_ctx),
ctx,
@@ -100,21 +108,8 @@ impl Session {
if !task_cancellation_token.is_cancelled() {
// Emit completion uniformly from spawn site so all tasks share the same lifecycle.
let sess = session_ctx.clone_session();
match run_result {
Ok(last_agent_message) => {
sess.on_task_finished(ctx_for_finish, last_agent_message)
.await;
}
Err(err) => {
let message = err.to_string();
sess.send_event(
ctx_for_finish.as_ref(),
EventMsg::Error(ErrorEvent { message }),
)
.await;
sess.on_task_finished(ctx_for_finish, None).await;
}
}
sess.on_task_finished(ctx_for_finish, last_agent_message)
.await;
}
done_clone.notify_waiters();
})

View File

@@ -5,7 +5,6 @@ use tokio_util::sync::CancellationToken;
use crate::codex::TurnContext;
use crate::codex::run_task;
use crate::error::Result as CodexResult;
use crate::state::TaskKind;
use codex_protocol::user_input::UserInput;
@@ -27,7 +26,7 @@ impl SessionTask for RegularTask {
ctx: Arc<TurnContext>,
input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> CodexResult<Option<String>> {
) -> Option<String> {
let sess = session.clone_session();
run_task(sess, ctx, input, TaskKind::Regular, cancellation_token).await
}

View File

@@ -1,12 +1,18 @@
use std::sync::Arc;
use async_trait::async_trait;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ReviewOutputEvent;
use codex_protocol::protocol::TaskCompleteEvent;
use tokio_util::sync::CancellationToken;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex::exit_review_mode;
use crate::codex::run_task;
use crate::error::Result as CodexResult;
use crate::codex_delegate::run_codex_conversation;
// use crate::config::Config; // no longer needed directly; use session.base_config()
use crate::review_format::format_review_findings_block;
use crate::state::TaskKind;
use codex_protocol::user_input::UserInput;
@@ -28,12 +34,109 @@ impl SessionTask for ReviewTask {
ctx: Arc<TurnContext>,
input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> CodexResult<Option<String>> {
let sess = session.clone_session();
run_task(sess, ctx, input, TaskKind::Review, cancellation_token).await
) -> Option<String> {
// let sess = session.clone_session();
// run_task(sess, ctx, input, TaskKind::Review, cancellation_token).await
let config = session.base_config().await.as_ref().clone();
let receiver =
match run_codex_conversation(config, session.auth_manager(), input, cancellation_token)
.await
{
Ok(r) => r,
Err(_) => return None,
};
while let Ok(event) = receiver.recv().await {
session
.clone_session()
.send_event(ctx.as_ref(), event.clone())
.await;
if let EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message }) = event {
exit_review_mode(
session.clone_session(),
last_agent_message.as_deref().map(parse_review_output_event),
)
.await;
}
}
Some("".to_string())
}
async fn abort(&self, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>) {
let _ = exit_review_mode(session.clone_session(), ctx, None).await;
async fn abort(&self, session: Arc<SessionTaskContext>, _ctx: Arc<TurnContext>) {
exit_review_mode(session.clone_session(), None).await;
}
}
/// Emits an ExitedReviewMode Event with optional ReviewOutput,
/// and records a developer message with the review output.
pub(crate) async fn exit_review_mode(
session: Arc<Session>,
review_output: Option<ReviewOutputEvent>,
) {
// ExitedReviewMode event can be emitted by the caller if needed.
let mut user_message = String::new();
if let Some(out) = review_output {
let mut findings_str = String::new();
let text = out.overall_explanation.trim();
if !text.is_empty() {
findings_str.push_str(text);
}
if !out.findings.is_empty() {
let block = format_review_findings_block(&out.findings, None);
findings_str.push_str(&format!("\n{block}"));
}
user_message.push_str(&format!(
r#"<user_action>
<context>User initiated a review task. Here's the full review output from reviewer model. User may select one or more comments to resolve.</context>
<action>review</action>
<results>
{findings_str}
</results>
</user_action>
"#));
} else {
user_message.push_str(r#"<user_action>
<context>User initiated a review task, but was interrupted. If user asks about this, tell them to re-initiate a review with `/review` and wait for it to complete.</context>
<action>review</action>
<results>
None.
</results>
</user_action>
"#);
}
session
.record_conversation_items(&[ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: user_message }],
}])
.await;
}
/// Parse the review output; when not valid JSON, build a structured
/// fallback that carries the plain text as the overall explanation.
///
/// Returns: a ReviewOutputEvent parsed from JSON or a fallback populated from text.
fn parse_review_output_event(text: &str) -> ReviewOutputEvent {
// Try direct parse first
if let Ok(ev) = serde_json::from_str::<ReviewOutputEvent>(text) {
return ev;
}
// If wrapped in markdown fences or extra prose, attempt to extract the first JSON object
if let (Some(start), Some(end)) = (text.find('{'), text.rfind('}'))
&& start < end
&& let Some(slice) = text.get(start..=end)
&& let Ok(ev) = serde_json::from_str::<ReviewOutputEvent>(slice)
{
return ev;
}
// Not JSON return a structured ReviewOutputEvent that carries
// the plain text as the overall explanation.
ReviewOutputEvent {
overall_explanation: text.to_string(),
..Default::default()
}
}

View File

@@ -1,9 +1,6 @@
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::error::CodexErr;
use crate::error::SandboxErr;
use crate::exec::ExecToolCallOutput;
use crate::function_tool::FunctionCallError;
use crate::parse_command::parse_command;
use crate::protocol::EventMsg;
use crate::protocol::ExecCommandBeginEvent;
@@ -13,7 +10,6 @@ use crate::protocol::PatchApplyBeginEvent;
use crate::protocol::PatchApplyEndEvent;
use crate::protocol::TurnDiffEvent;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::sandboxing::ToolError;
use std::collections::HashMap;
use std::path::Path;
use std::path::PathBuf;
@@ -200,103 +196,12 @@ impl ToolEmitter {
) => {
emit_patch_end(ctx, String::new(), (*message).to_string(), false).await;
}
(Self::UnifiedExec { command, cwd, .. }, ToolEventStage::Begin) => {
(Self::UnifiedExec { command, cwd, .. }, _) => {
// TODO(jif) add end and failures.
emit_exec_command_begin(ctx, &[command.to_string()], cwd.as_path()).await;
}
(Self::UnifiedExec { .. }, ToolEventStage::Success(output)) => {
emit_exec_end(
ctx,
output.stdout.text.clone(),
output.stderr.text.clone(),
output.aggregated_output.text.clone(),
output.exit_code,
output.duration,
format_exec_output_str(&output),
)
.await;
}
(
Self::UnifiedExec { .. },
ToolEventStage::Failure(ToolEventFailure::Output(output)),
) => {
emit_exec_end(
ctx,
output.stdout.text.clone(),
output.stderr.text.clone(),
output.aggregated_output.text.clone(),
output.exit_code,
output.duration,
format_exec_output_str(&output),
)
.await;
}
(
Self::UnifiedExec { .. },
ToolEventStage::Failure(ToolEventFailure::Message(message)),
) => {
emit_exec_end(
ctx,
String::new(),
(*message).to_string(),
(*message).to_string(),
-1,
Duration::ZERO,
format_exec_output(&message),
)
.await;
}
}
}
pub async fn begin(&self, ctx: ToolEventCtx<'_>) {
self.emit(ctx, ToolEventStage::Begin).await;
}
pub async fn finish(
&self,
ctx: ToolEventCtx<'_>,
out: Result<ExecToolCallOutput, ToolError>,
) -> Result<String, FunctionCallError> {
let event;
let result = match out {
Ok(output) => {
let content = super::format_exec_output_for_model(&output);
let exit_code = output.exit_code;
event = ToolEventStage::Success(output);
if exit_code == 0 {
Ok(content)
} else {
Err(FunctionCallError::RespondToModel(content))
}
}
Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Timeout { output })))
| Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Denied { output }))) => {
let response = super::format_exec_output_for_model(&output);
event = ToolEventStage::Failure(ToolEventFailure::Output(*output));
Err(FunctionCallError::RespondToModel(response))
}
Err(ToolError::Codex(err)) => {
let message = format!("execution error: {err:?}");
let response = super::format_exec_output(&message);
event = ToolEventStage::Failure(ToolEventFailure::Message(message));
Err(FunctionCallError::RespondToModel(response))
}
Err(ToolError::Rejected(msg)) | Err(ToolError::SandboxDenied(msg)) => {
// Normalize common rejection messages for exec tools so tests and
// users see a clear, consistent phrase.
let normalized = if msg == "rejected by user" {
"exec command rejected by user".to_string()
} else {
msg
};
let response = super::format_exec_output(&normalized);
event = ToolEventStage::Failure(ToolEventFailure::Message(normalized));
Err(FunctionCallError::RespondToModel(response))
}
};
self.emit(ctx, event).await;
result
}
}
async fn emit_exec_end(

View File

@@ -1,24 +1,19 @@
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::sync::Arc;
use crate::apply_patch;
use crate::apply_patch::InternalApplyPatchInvocation;
use crate::apply_patch::convert_apply_patch_to_protocol;
use crate::client_common::tools::FreeformTool;
use crate::client_common::tools::FreeformToolFormat;
use crate::client_common::tools::ResponsesApiTool;
use crate::client_common::tools::ToolSpec;
use crate::exec::ExecParams;
use crate::function_tool::FunctionCallError;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
use crate::tools::events::ToolEmitter;
use crate::tools::events::ToolEventCtx;
use crate::tools::orchestrator::ToolOrchestrator;
use crate::tools::handle_container_exec_with_params;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
use crate::tools::runtimes::apply_patch::ApplyPatchRequest;
use crate::tools::runtimes::apply_patch::ApplyPatchRuntime;
use crate::tools::sandboxing::ToolCtx;
use crate::tools::spec::ApplyPatchToolArgs;
use crate::tools::spec::JsonSchema;
use async_trait::async_trait;
@@ -69,85 +64,30 @@ impl ToolHandler for ApplyPatchHandler {
}
};
// Re-parse and verify the patch so we can compute changes and approval.
// Avoid building temporary ExecParams/command vectors; derive directly from inputs.
let cwd = turn.cwd.clone();
let command = vec!["apply_patch".to_string(), patch_input.clone()];
match codex_apply_patch::maybe_parse_apply_patch_verified(&command, &cwd) {
codex_apply_patch::MaybeApplyPatchVerified::Body(changes) => {
match apply_patch::apply_patch(session.as_ref(), turn.as_ref(), &call_id, changes)
.await
{
InternalApplyPatchInvocation::Output(item) => {
let content = item?;
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
InternalApplyPatchInvocation::DelegateToExec(apply) => {
let emitter = ToolEmitter::apply_patch(
convert_apply_patch_to_protocol(&apply.action),
!apply.user_explicitly_approved_this_action,
);
let event_ctx = ToolEventCtx::new(
session.as_ref(),
turn.as_ref(),
&call_id,
Some(&tracker),
);
emitter.begin(event_ctx).await;
let exec_params = ExecParams {
command: vec!["apply_patch".to_string(), patch_input.clone()],
cwd: turn.cwd.clone(),
timeout_ms: None,
env: HashMap::new(),
with_escalated_permissions: None,
justification: None,
arg0: None,
};
let req = ApplyPatchRequest {
patch: apply.action.patch.clone(),
cwd: apply.action.cwd.clone(),
timeout_ms: None,
user_explicitly_approved: apply.user_explicitly_approved_this_action,
codex_exe: turn.codex_linux_sandbox_exe.clone(),
};
let content = handle_container_exec_with_params(
tool_name.as_str(),
exec_params,
Arc::clone(&session),
Arc::clone(&turn),
Arc::clone(&tracker),
call_id.clone(),
)
.await?;
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = ApplyPatchRuntime::new();
let tool_ctx = ToolCtx {
session: session.as_ref(),
turn: turn.as_ref(),
call_id: call_id.clone(),
tool_name: tool_name.to_string(),
};
let out = orchestrator
.run(&mut runtime, &req, &tool_ctx, &turn, turn.approval_policy)
.await;
let event_ctx = ToolEventCtx::new(
session.as_ref(),
turn.as_ref(),
&call_id,
Some(&tracker),
);
let content = emitter.finish(event_ctx, out).await?;
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
}
}
codex_apply_patch::MaybeApplyPatchVerified::CorrectnessError(parse_error) => {
Err(FunctionCallError::RespondToModel(format!(
"apply_patch verification failed: {parse_error}"
)))
}
codex_apply_patch::MaybeApplyPatchVerified::ShellParseError(error) => {
tracing::trace!("Failed to parse apply_patch input, {error:?}");
Err(FunctionCallError::RespondToModel(
"apply_patch handler received invalid patch input".to_string(),
))
}
codex_apply_patch::MaybeApplyPatchVerified::NotApplyPatch => {
Err(FunctionCallError::RespondToModel(
"apply_patch handler received non-apply_patch input".to_string(),
))
}
}
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
}

View File

@@ -2,9 +2,6 @@ use async_trait::async_trait;
use codex_protocol::models::ShellToolCallParams;
use std::sync::Arc;
use crate::apply_patch;
use crate::apply_patch::InternalApplyPatchInvocation;
use crate::apply_patch::convert_apply_patch_to_protocol;
use crate::codex::TurnContext;
use crate::exec::ExecParams;
use crate::exec_env::create_env;
@@ -12,16 +9,9 @@ use crate::function_tool::FunctionCallError;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
use crate::tools::events::ToolEmitter;
use crate::tools::events::ToolEventCtx;
use crate::tools::orchestrator::ToolOrchestrator;
use crate::tools::handle_container_exec_with_params;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
use crate::tools::runtimes::apply_patch::ApplyPatchRequest;
use crate::tools::runtimes::apply_patch::ApplyPatchRuntime;
use crate::tools::runtimes::shell::ShellRequest;
use crate::tools::runtimes::shell::ShellRuntime;
use crate::tools::sandboxing::ToolCtx;
pub struct ShellHandler;
@@ -71,27 +61,35 @@ impl ToolHandler for ShellHandler {
))
})?;
let exec_params = Self::to_exec_params(params, turn.as_ref());
Self::run_exec_like(
let content = handle_container_exec_with_params(
tool_name.as_str(),
exec_params,
session,
turn,
tracker,
call_id,
Arc::clone(&session),
Arc::clone(&turn),
Arc::clone(&tracker),
call_id.clone(),
)
.await
.await?;
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
ToolPayload::LocalShell { params } => {
let exec_params = Self::to_exec_params(params, turn.as_ref());
Self::run_exec_like(
let content = handle_container_exec_with_params(
tool_name.as_str(),
exec_params,
session,
turn,
tracker,
call_id,
Arc::clone(&session),
Arc::clone(&turn),
Arc::clone(&tracker),
call_id.clone(),
)
.await
.await?;
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
_ => Err(FunctionCallError::RespondToModel(format!(
"unsupported payload for shell handler: {tool_name}"
@@ -99,134 +97,3 @@ impl ToolHandler for ShellHandler {
}
}
}
impl ShellHandler {
async fn run_exec_like(
tool_name: &str,
exec_params: ExecParams,
session: Arc<crate::codex::Session>,
turn: Arc<TurnContext>,
tracker: crate::tools::context::SharedTurnDiffTracker,
call_id: String,
) -> Result<ToolOutput, FunctionCallError> {
// Approval policy guard for explicit escalation in non-OnRequest modes.
if exec_params.with_escalated_permissions.unwrap_or(false)
&& !matches!(
turn.approval_policy,
codex_protocol::protocol::AskForApproval::OnRequest
)
{
return Err(FunctionCallError::RespondToModel(format!(
"approval policy is {policy:?}; reject command — you should not ask for escalated permissions if the approval policy is {policy:?}",
policy = turn.approval_policy
)));
}
// Intercept apply_patch if present.
match codex_apply_patch::maybe_parse_apply_patch_verified(
&exec_params.command,
&exec_params.cwd,
) {
codex_apply_patch::MaybeApplyPatchVerified::Body(changes) => {
match apply_patch::apply_patch(session.as_ref(), turn.as_ref(), &call_id, changes)
.await
{
InternalApplyPatchInvocation::Output(item) => {
// Programmatic apply_patch path; return its result.
let content = item?;
return Ok(ToolOutput::Function {
content,
success: Some(true),
});
}
InternalApplyPatchInvocation::DelegateToExec(apply) => {
let emitter = ToolEmitter::apply_patch(
convert_apply_patch_to_protocol(&apply.action),
!apply.user_explicitly_approved_this_action,
);
let event_ctx = ToolEventCtx::new(
session.as_ref(),
turn.as_ref(),
&call_id,
Some(&tracker),
);
emitter.begin(event_ctx).await;
let req = ApplyPatchRequest {
patch: apply.action.patch.clone(),
cwd: apply.action.cwd.clone(),
timeout_ms: exec_params.timeout_ms,
user_explicitly_approved: apply.user_explicitly_approved_this_action,
codex_exe: turn.codex_linux_sandbox_exe.clone(),
};
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = ApplyPatchRuntime::new();
let tool_ctx = ToolCtx {
session: session.as_ref(),
turn: turn.as_ref(),
call_id: call_id.clone(),
tool_name: tool_name.to_string(),
};
let out = orchestrator
.run(&mut runtime, &req, &tool_ctx, &turn, turn.approval_policy)
.await;
let event_ctx = ToolEventCtx::new(
session.as_ref(),
turn.as_ref(),
&call_id,
Some(&tracker),
);
let content = emitter.finish(event_ctx, out).await?;
return Ok(ToolOutput::Function {
content,
success: Some(true),
});
}
}
}
codex_apply_patch::MaybeApplyPatchVerified::CorrectnessError(parse_error) => {
return Err(FunctionCallError::RespondToModel(format!(
"apply_patch verification failed: {parse_error}"
)));
}
codex_apply_patch::MaybeApplyPatchVerified::ShellParseError(error) => {
tracing::trace!("Failed to parse shell command, {error:?}");
// Fall through to regular shell execution.
}
codex_apply_patch::MaybeApplyPatchVerified::NotApplyPatch => {
// Fall through to regular shell execution.
}
}
// Regular shell execution path.
let emitter = ToolEmitter::shell(exec_params.command.clone(), exec_params.cwd.clone());
let event_ctx = ToolEventCtx::new(session.as_ref(), turn.as_ref(), &call_id, None);
emitter.begin(event_ctx).await;
let req = ShellRequest {
command: exec_params.command.clone(),
cwd: exec_params.cwd.clone(),
timeout_ms: exec_params.timeout_ms,
env: exec_params.env.clone(),
with_escalated_permissions: exec_params.with_escalated_permissions,
justification: exec_params.justification.clone(),
};
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = ShellRuntime::new();
let tool_ctx = ToolCtx {
session: session.as_ref(),
turn: turn.as_ref(),
call_id: call_id.clone(),
tool_name: tool_name.to_string(),
};
let out = orchestrator
.run(&mut runtime, &req, &tool_ctx, &turn, turn.approval_policy)
.await;
let event_ctx = ToolEventCtx::new(session.as_ref(), turn.as_ref(), &call_id, None);
let content = emitter.finish(event_ctx, out).await?;
Ok(ToolOutput::Function {
content,
success: Some(true),
})
}
}

View File

@@ -5,9 +5,6 @@ use serde::Deserialize;
use serde::Serialize;
use crate::function_tool::FunctionCallError;
use crate::protocol::EventMsg;
use crate::protocol::ExecCommandOutputDeltaEvent;
use crate::protocol::ExecOutputStream;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
@@ -90,7 +87,11 @@ impl ToolHandler for UnifiedExecHandler {
};
let manager: &UnifiedExecSessionManager = &session.services.unified_exec_manager;
let context = UnifiedExecContext::new(session.clone(), turn.clone(), call_id.clone());
let context = UnifiedExecContext {
session: &session,
turn: turn.as_ref(),
call_id: &call_id,
};
let response = match tool_name.as_str() {
"exec_command" => {
@@ -100,12 +101,8 @@ impl ToolHandler for UnifiedExecHandler {
))
})?;
let event_ctx = ToolEventCtx::new(
context.session.as_ref(),
context.turn.as_ref(),
&context.call_id,
None,
);
let event_ctx =
ToolEventCtx::new(context.session, context.turn, context.call_id, None);
let emitter =
ToolEmitter::unified_exec(args.cmd.clone(), context.turn.cwd.clone(), true);
emitter.emit(event_ctx, ToolEventStage::Begin).await;
@@ -151,18 +148,6 @@ impl ToolHandler for UnifiedExecHandler {
}
};
// Emit a delta event with the chunk of output we just produced, if any.
if !response.output.is_empty() {
let delta = ExecCommandOutputDeltaEvent {
call_id: response.event_call_id.clone(),
stream: ExecOutputStream::Stdout,
chunk: response.output.as_bytes().to_vec(),
};
session
.send_event(turn.as_ref(), EventMsg::ExecCommandOutputDelta(delta))
.await;
}
let content = serialize_response(&response).map_err(|err| {
FunctionCallError::RespondToModel(format!(
"failed to serialize unified exec output: {err:?}"

View File

@@ -9,11 +9,37 @@ pub mod runtimes;
pub mod sandboxing;
pub mod spec;
use crate::apply_patch;
use crate::apply_patch::InternalApplyPatchInvocation;
use crate::apply_patch::convert_apply_patch_to_protocol;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::error::CodexErr;
use crate::error::SandboxErr;
use crate::exec::ExecParams;
use crate::exec::ExecToolCallOutput;
use crate::function_tool::FunctionCallError;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::events::ToolEmitter;
use crate::tools::events::ToolEventCtx;
use crate::tools::events::ToolEventFailure;
use crate::tools::events::ToolEventStage;
use crate::tools::orchestrator::ToolOrchestrator;
use crate::tools::runtimes::apply_patch::ApplyPatchRequest;
use crate::tools::runtimes::apply_patch::ApplyPatchRuntime;
use crate::tools::runtimes::shell::ShellRequest;
use crate::tools::runtimes::shell::ShellRuntime;
use crate::tools::sandboxing::ToolCtx;
use crate::tools::sandboxing::ToolError;
use codex_apply_patch::MaybeApplyPatchVerified;
use codex_apply_patch::maybe_parse_apply_patch_verified;
use codex_protocol::protocol::AskForApproval;
use codex_utils_string::take_bytes_at_char_boundary;
use codex_utils_string::take_last_bytes_at_char_boundary;
pub use router::ToolRouter;
use serde::Serialize;
use std::sync::Arc;
use tracing::trace;
// Model-formatting limits: clients get full streams; only content sent to the model is truncated.
pub(crate) const MODEL_FORMAT_MAX_BYTES: usize = 10 * 1024; // 10 KiB
@@ -28,6 +54,186 @@ pub(crate) const TELEMETRY_PREVIEW_MAX_LINES: usize = 64; // lines
pub(crate) const TELEMETRY_PREVIEW_TRUNCATION_NOTICE: &str =
"[... telemetry preview truncated ...]";
// TODO(jif) break this down
pub(crate) async fn handle_container_exec_with_params(
tool_name: &str,
params: ExecParams,
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
turn_diff_tracker: SharedTurnDiffTracker,
call_id: String,
) -> Result<String, FunctionCallError> {
let _otel_event_manager = turn_context.client.get_otel_event_manager();
if params.with_escalated_permissions.unwrap_or(false)
&& !matches!(turn_context.approval_policy, AskForApproval::OnRequest)
{
return Err(FunctionCallError::RespondToModel(format!(
"approval policy is {policy:?}; reject command — you should not ask for escalated permissions if the approval policy is {policy:?}",
policy = turn_context.approval_policy
)));
}
// check if this was a patch, and apply it if so
let apply_patch_exec = match maybe_parse_apply_patch_verified(&params.command, &params.cwd) {
MaybeApplyPatchVerified::Body(changes) => {
match apply_patch::apply_patch(sess.as_ref(), turn_context.as_ref(), &call_id, changes)
.await
{
InternalApplyPatchInvocation::Output(item) => return item,
InternalApplyPatchInvocation::DelegateToExec(apply_patch_exec) => {
Some(apply_patch_exec)
}
}
}
MaybeApplyPatchVerified::CorrectnessError(parse_error) => {
// It looks like an invocation of `apply_patch`, but we
// could not resolve it into a patch that would apply
// cleanly. Return to model for resample.
return Err(FunctionCallError::RespondToModel(format!(
"apply_patch verification failed: {parse_error}"
)));
}
MaybeApplyPatchVerified::ShellParseError(error) => {
trace!("Failed to parse shell command, {error:?}");
None
}
MaybeApplyPatchVerified::NotApplyPatch => None,
};
let (event_emitter, diff_opt) = match apply_patch_exec.as_ref() {
Some(exec) => (
ToolEmitter::apply_patch(
convert_apply_patch_to_protocol(&exec.action),
!exec.user_explicitly_approved_this_action,
),
Some(&turn_diff_tracker),
),
None => (
ToolEmitter::shell(params.command.clone(), params.cwd.clone()),
None,
),
};
let event_ctx = ToolEventCtx::new(sess.as_ref(), turn_context.as_ref(), &call_id, diff_opt);
event_emitter.emit(event_ctx, ToolEventStage::Begin).await;
// Build runtime contexts only when needed (shell/apply_patch below).
if let Some(exec) = apply_patch_exec {
// Route apply_patch execution through the new orchestrator/runtime.
let req = ApplyPatchRequest {
patch: exec.action.patch.clone(),
cwd: params.cwd.clone(),
timeout_ms: params.timeout_ms,
user_explicitly_approved: exec.user_explicitly_approved_this_action,
codex_exe: turn_context.codex_linux_sandbox_exe.clone(),
};
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = ApplyPatchRuntime::new();
let tool_ctx = ToolCtx {
session: sess.as_ref(),
turn: turn_context.as_ref(),
call_id: call_id.clone(),
tool_name: tool_name.to_string(),
};
let out = orchestrator
.run(
&mut runtime,
&req,
&tool_ctx,
&turn_context,
turn_context.approval_policy,
)
.await;
handle_exec_outcome(&event_emitter, event_ctx, out).await
} else {
// Route shell execution through the new orchestrator/runtime.
let req = ShellRequest {
command: params.command.clone(),
cwd: params.cwd.clone(),
timeout_ms: params.timeout_ms,
env: params.env.clone(),
with_escalated_permissions: params.with_escalated_permissions,
justification: params.justification.clone(),
};
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = ShellRuntime::new();
let tool_ctx = ToolCtx {
session: sess.as_ref(),
turn: turn_context.as_ref(),
call_id: call_id.clone(),
tool_name: tool_name.to_string(),
};
let out = orchestrator
.run(
&mut runtime,
&req,
&tool_ctx,
&turn_context,
turn_context.approval_policy,
)
.await;
handle_exec_outcome(&event_emitter, event_ctx, out).await
}
}
async fn handle_exec_outcome(
event_emitter: &ToolEmitter,
event_ctx: ToolEventCtx<'_>,
out: Result<ExecToolCallOutput, ToolError>,
) -> Result<String, FunctionCallError> {
let event;
let result = match out {
Ok(output) => {
let content = format_exec_output_for_model(&output);
let exit_code = output.exit_code;
event = ToolEventStage::Success(output);
if exit_code == 0 {
Ok(content)
} else {
Err(FunctionCallError::RespondToModel(content))
}
}
Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Timeout { output })))
| Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Denied { output }))) => {
let response = format_exec_output_for_model(&output);
event = ToolEventStage::Failure(ToolEventFailure::Output(*output));
Err(FunctionCallError::RespondToModel(response))
}
Err(ToolError::Codex(err)) => {
let message = format!("execution error: {err:?}");
let response = format_exec_output(&message);
event = ToolEventStage::Failure(ToolEventFailure::Message(message));
Err(FunctionCallError::RespondToModel(format_exec_output(
&response,
)))
}
Err(ToolError::Rejected(msg)) | Err(ToolError::SandboxDenied(msg)) => {
// Normalize common rejection messages for exec tools so tests and
// users see a clear, consistent phrase.
let normalized = if msg == "rejected by user" {
"exec command rejected by user".to_string()
} else {
msg
};
let response = format_exec_output(&normalized);
event = ToolEventStage::Failure(ToolEventFailure::Message(normalized));
Err(FunctionCallError::RespondToModel(format_exec_output(
&response,
)))
}
};
event_emitter.emit(event_ctx, event).await;
result
}
/// Format the combined exec output for sending back to the model.
/// Includes exit code and duration metadata; truncates large bodies safely.
pub fn format_exec_output_for_model(exec_output: &ExecToolCallOutput) -> String {
@@ -157,7 +363,6 @@ fn truncate_formatted_exec_output(content: &str, total_lines: usize) -> String {
#[cfg(test)]
mod tests {
use super::*;
use crate::function_tool::FunctionCallError;
use regex_lite::Regex;
fn truncate_function_error(err: FunctionCallError) -> FunctionCallError {

View File

@@ -98,9 +98,9 @@ impl ToolOrchestrator {
"sandbox denied and no retry".to_string(),
));
}
// Under `Never` or `OnRequest`, do not retry without sandbox; surface a concise message
// Under `Never`, do not retry without sandbox; surface a concise message
// derived from the actual output (platform-agnostic).
if !tool.wants_no_sandbox_approval(approval_policy) {
if matches!(approval_policy, AskForApproval::Never) {
let msg = build_never_denied_message_from_output(output.as_ref());
return Err(ToolError::SandboxDenied(msg));
}

View File

@@ -17,7 +17,6 @@ use crate::tools::sandboxing::ToolCtx;
use crate::tools::sandboxing::ToolError;
use crate::tools::sandboxing::ToolRuntime;
use crate::tools::sandboxing::with_cached_approval;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::ReviewDecision;
use futures::future::BoxFuture;
use std::collections::HashMap;
@@ -128,10 +127,6 @@ impl Approvable<ApplyPatchRequest> for ApplyPatchRuntime {
.await
})
}
fn wants_no_sandbox_approval(&self, policy: AskForApproval) -> bool {
!matches!(policy, AskForApproval::Never)
}
}
impl ToolRuntime<ApplyPatchRequest, ExecToolCallOutput> for ApplyPatchRuntime {

View File

@@ -121,11 +121,6 @@ pub(crate) trait Approvable<Req> {
}
}
/// Decide we can request an approval for no-sandbox execution.
fn wants_no_sandbox_approval(&self, policy: AskForApproval) -> bool {
!matches!(policy, AskForApproval::Never | AskForApproval::OnRequest)
}
fn start_approval_async<'a>(
&'a mut self,
req: &'a Req,

View File

@@ -1,35 +1,18 @@
//! Utilities for truncating large chunks of output while preserving a prefix
//! and suffix on UTF-8 boundaries.
use codex_utils_tokenizer::Tokenizer;
/// Truncate the middle of a UTF-8 string to at most `max_bytes` bytes,
/// preserving the beginning and the end. Returns the possibly truncated
/// string and `Some(original_token_count)` (counted with the local tokenizer;
/// falls back to a 4-bytes-per-token estimate if the tokenizer cannot load)
/// string and `Some(original_token_count)` (estimated at 4 bytes/token)
/// if truncation occurred; otherwise returns the original string and `None`.
pub(crate) fn truncate_middle(s: &str, max_bytes: usize) -> (String, Option<u64>) {
if s.len() <= max_bytes {
return (s.to_string(), None);
}
// Build a tokenizer for counting (default to o200k_base; fall back to cl100k_base).
// If both fail, fall back to a 4-bytes-per-token estimate.
let tok = Tokenizer::try_default().ok();
let token_count = |text: &str| -> u64 {
if let Some(ref t) = tok {
t.count(text) as u64
} else {
(text.len() as u64).div_ceil(4)
}
};
let total_tokens = token_count(s);
let est_tokens = (s.len() as u64).div_ceil(4);
if max_bytes == 0 {
return (
format!("{total_tokens} tokens truncated…"),
Some(total_tokens),
);
return (format!("{est_tokens} tokens truncated…"), Some(est_tokens));
}
fn truncate_on_boundary(input: &str, max_len: usize) -> &str {
@@ -67,17 +50,13 @@ pub(crate) fn truncate_middle(s: &str, max_bytes: usize) -> (String, Option<u64>
idx
}
// Iterate to stabilize marker length → keep budget → boundaries.
let mut guess_tokens: u64 = 1;
let mut guess_tokens = est_tokens;
for _ in 0..4 {
let marker = format!("{guess_tokens} tokens truncated…");
let marker_len = marker.len();
let keep_budget = max_bytes.saturating_sub(marker_len);
if keep_budget == 0 {
return (
format!("{total_tokens} tokens truncated…"),
Some(total_tokens),
);
return (format!("{est_tokens} tokens truncated…"), Some(est_tokens));
}
let left_budget = keep_budget / 2;
@@ -88,72 +67,59 @@ pub(crate) fn truncate_middle(s: &str, max_bytes: usize) -> (String, Option<u64>
suffix_start = prefix_end;
}
// Tokens actually removed (middle slice) using the real tokenizer.
let removed_tokens = token_count(&s[prefix_end..suffix_start]);
let kept_content_bytes = prefix_end + (s.len() - suffix_start);
let truncated_content_bytes = s.len().saturating_sub(kept_content_bytes);
let new_tokens = (truncated_content_bytes as u64).div_ceil(4);
// If the number of digits in the token count does not change the marker length,
// we can finalize output.
let final_marker = format!("{removed_tokens} tokens truncated…");
if final_marker.len() == marker_len {
let kept_content_bytes = prefix_end + (s.len() - suffix_start);
let mut out = String::with_capacity(final_marker.len() + kept_content_bytes + 1);
if new_tokens == guess_tokens {
let mut out = String::with_capacity(marker_len + kept_content_bytes + 1);
out.push_str(&s[..prefix_end]);
out.push_str(&final_marker);
out.push_str(&marker);
out.push('\n');
out.push_str(&s[suffix_start..]);
return (out, Some(total_tokens));
return (out, Some(est_tokens));
}
guess_tokens = removed_tokens;
guess_tokens = new_tokens;
}
// Fallback build after iterations: compute with the last guess.
let marker = format!("{guess_tokens} tokens truncated…");
let marker_len = marker.len();
let keep_budget = max_bytes.saturating_sub(marker_len);
if keep_budget == 0 {
return (
format!("{total_tokens} tokens truncated…"),
Some(total_tokens),
);
return (format!("{est_tokens} tokens truncated…"), Some(est_tokens));
}
let left_budget = keep_budget / 2;
let right_budget = keep_budget - left_budget;
let prefix_end = pick_prefix_end(s, left_budget);
let mut suffix_start = pick_suffix_start(s, right_budget);
if suffix_start < prefix_end {
suffix_start = prefix_end;
}
let suffix_start = pick_suffix_start(s, right_budget);
let mut out = String::with_capacity(marker_len + prefix_end + (s.len() - suffix_start) + 1);
out.push_str(&s[..prefix_end]);
out.push_str(&marker);
out.push('\n');
out.push_str(&s[suffix_start..]);
(out, Some(total_tokens))
(out, Some(est_tokens))
}
#[cfg(test)]
mod tests {
use super::truncate_middle;
use codex_utils_tokenizer::Tokenizer;
#[test]
fn truncate_middle_no_newlines_fallback() {
let tok = Tokenizer::try_default().expect("load tokenizer");
let s = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ*";
let max_bytes = 32;
let (out, original) = truncate_middle(s, max_bytes);
assert!(out.starts_with("abc"));
assert!(out.contains("tokens truncated"));
assert!(out.ends_with("XYZ*"));
assert_eq!(original, Some(tok.count(s) as u64));
assert_eq!(original, Some((s.len() as u64).div_ceil(4)));
}
#[test]
fn truncate_middle_prefers_newline_boundaries() {
let tok = Tokenizer::try_default().expect("load tokenizer");
let mut s = String::new();
for i in 1..=20 {
s.push_str(&format!("{i:03}\n"));
@@ -165,36 +131,50 @@ mod tests {
assert!(out.starts_with("001\n002\n003\n004\n"));
assert!(out.contains("tokens truncated"));
assert!(out.ends_with("017\n018\n019\n020\n"));
assert_eq!(tokens, Some(tok.count(&s) as u64));
assert_eq!(tokens, Some(20));
}
#[test]
fn truncate_middle_handles_utf8_content() {
let tok = Tokenizer::try_default().expect("load tokenizer");
let s = "😀😀😀😀😀😀😀😀😀😀\nsecond line with ascii text\n";
let max_bytes = 32;
let (out, tokens) = truncate_middle(s, max_bytes);
assert!(out.contains("tokens truncated"));
assert!(!out.contains('\u{fffd}'));
assert_eq!(tokens, Some(tok.count(s) as u64));
assert_eq!(tokens, Some((s.len() as u64).div_ceil(4)));
}
#[test]
fn truncate_middle_prefers_newline_boundaries_2() {
let tok = Tokenizer::try_default().expect("load tokenizer");
// Build a multi-line string of 20 numbered lines (each "NNN\n").
let mut s = String::new();
for i in 1..=20 {
s.push_str(&format!("{i:03}\n"));
}
// Total length: 20 lines * 4 bytes per line = 80 bytes.
assert_eq!(s.len(), 80);
// Choose a cap that forces truncation while leaving room for
// a few lines on each side after accounting for the marker.
let max_bytes = 64;
let (out, total) = truncate_middle(&s, max_bytes);
assert!(out.starts_with("001\n002\n003\n004\n"));
assert!(out.contains("tokens truncated"));
assert!(out.ends_with("017\n018\n019\n020\n"));
assert_eq!(total, Some(tok.count(&s) as u64));
// Expect exact output: first 4 lines, marker, last 4 lines, and correct token estimate (80/4 = 20).
assert_eq!(
truncate_middle(&s, max_bytes),
(
r#"001
002
003
004
…12 tokens truncated…
017
018
019
020
"#
.to_string(),
Some(20)
)
);
}
}

View File

@@ -22,8 +22,6 @@
//! - `session_manager.rs`: orchestration (approvals, sandboxing, reuse) and request handling.
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicI32;
use std::time::Duration;
@@ -47,20 +45,10 @@ pub(crate) const MAX_YIELD_TIME_MS: u64 = 30_000;
pub(crate) const DEFAULT_MAX_OUTPUT_TOKENS: usize = 10_000;
pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_BYTES: usize = 1024 * 1024; // 1 MiB
pub(crate) struct UnifiedExecContext {
pub session: Arc<Session>,
pub turn: Arc<TurnContext>,
pub call_id: String,
}
impl UnifiedExecContext {
pub fn new(session: Arc<Session>, turn: Arc<TurnContext>, call_id: String) -> Self {
Self {
session,
turn,
call_id,
}
}
pub(crate) struct UnifiedExecContext<'a> {
pub session: &'a Session,
pub turn: &'a TurnContext,
pub call_id: &'a str,
}
#[derive(Debug)]
@@ -82,7 +70,6 @@ pub(crate) struct WriteStdinRequest<'a> {
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct UnifiedExecResponse {
pub event_call_id: String,
pub chunk_id: String,
pub wall_time: Duration,
pub output: String,
@@ -91,20 +78,10 @@ pub(crate) struct UnifiedExecResponse {
pub original_token_count: Option<usize>,
}
#[derive(Default)]
#[derive(Debug, Default)]
pub(crate) struct UnifiedExecSessionManager {
next_session_id: AtomicI32,
sessions: Mutex<HashMap<i32, SessionEntry>>,
}
struct SessionEntry {
session: session::UnifiedExecSession,
session_ref: Arc<Session>,
turn_ref: Arc<TurnContext>,
call_id: String,
command: String,
cwd: PathBuf,
started_at: tokio::time::Instant,
sessions: Mutex<HashMap<i32, session::UnifiedExecSession>>,
}
pub(crate) fn clamp_yield_time(yield_time_ms: Option<u64>) -> u64 {
@@ -186,8 +163,11 @@ mod tests {
cmd: &str,
yield_time_ms: Option<u64>,
) -> Result<UnifiedExecResponse, UnifiedExecError> {
let context =
UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
let context = UnifiedExecContext {
session,
turn: turn.as_ref(),
call_id: "call",
};
session
.services

View File

@@ -5,13 +5,8 @@ use tokio::sync::mpsc;
use tokio::time::Duration;
use tokio::time::Instant;
use crate::exec::ExecToolCallOutput;
use crate::exec::StreamOutput;
use crate::exec_env::create_env;
use crate::sandboxing::ExecEnv;
use crate::tools::events::ToolEmitter;
use crate::tools::events::ToolEventCtx;
use crate::tools::events::ToolEventStage;
use crate::tools::orchestrator::ToolOrchestrator;
use crate::tools::runtimes::unified_exec::UnifiedExecRequest as UnifiedExecToolRequest;
use crate::tools::runtimes::unified_exec::UnifiedExecRuntime;
@@ -19,7 +14,6 @@ use crate::tools::sandboxing::ToolCtx;
use super::ExecCommandRequest;
use super::MIN_YIELD_TIME_MS;
use super::SessionEntry;
use super::UnifiedExecContext;
use super::UnifiedExecError;
use super::UnifiedExecResponse;
@@ -36,7 +30,7 @@ impl UnifiedExecSessionManager {
pub(crate) async fn exec_command(
&self,
request: ExecCommandRequest<'_>,
context: &UnifiedExecContext,
context: &UnifiedExecContext<'_>,
) -> Result<UnifiedExecResponse, UnifiedExecError> {
let shell_flag = if request.login { "-lc" } else { "-c" };
let command = vec![
@@ -65,36 +59,17 @@ impl UnifiedExecSessionManager {
let session_id = if session.has_exited() {
None
} else {
Some(
self.store_session(session, context, request.command, start)
.await,
)
Some(self.store_session(session).await)
};
let response = UnifiedExecResponse {
event_call_id: context.call_id.clone(),
Ok(UnifiedExecResponse {
chunk_id,
wall_time,
output,
session_id,
exit_code,
original_token_count,
};
// If the command completed during this call, emit an ExecCommandEnd via the emitter.
if response.session_id.is_none() {
let exit = response.exit_code.unwrap_or(-1);
Self::emit_exec_end_from_context(
context,
request.command.to_string(),
response.output.clone(),
exit,
response.wall_time,
)
.await;
}
Ok(response)
})
}
pub(crate) async fn write_stdin(
@@ -123,60 +98,37 @@ impl UnifiedExecSessionManager {
let (output, original_token_count) = truncate_output_to_tokens(&text, max_tokens);
let chunk_id = generate_chunk_id();
let status = self.refresh_session_state(session_id).await;
let (session_id, exit_code, completion_entry, event_call_id) = match status {
SessionStatus::Alive { exit_code, call_id } => {
(Some(session_id), exit_code, None, call_id)
}
SessionStatus::Exited { exit_code, entry } => {
let call_id = entry.call_id.clone();
(None, exit_code, Some(*entry), call_id)
}
SessionStatus::Unknown => {
return Err(UnifiedExecError::UnknownSessionId { session_id });
}
};
let (session_id, exit_code) = self.refresh_session_state(session_id).await;
let response = UnifiedExecResponse {
event_call_id,
Ok(UnifiedExecResponse {
chunk_id,
wall_time,
output,
session_id,
exit_code,
original_token_count,
};
if let (Some(exit), Some(entry)) = (response.exit_code, completion_entry) {
let total_duration = Instant::now().saturating_duration_since(entry.started_at);
Self::emit_exec_end_from_entry(entry, response.output.clone(), exit, total_duration)
.await;
}
Ok(response)
})
}
async fn refresh_session_state(&self, session_id: i32) -> SessionStatus {
async fn refresh_session_state(&self, session_id: i32) -> (Option<i32>, Option<i32>) {
let mut sessions = self.sessions.lock().await;
let Some(entry) = sessions.get(&session_id) else {
return SessionStatus::Unknown;
};
if !sessions.contains_key(&session_id) {
return (None, None);
}
let exit_code = entry.session.exit_code();
let has_exited = sessions
.get(&session_id)
.map(UnifiedExecSession::has_exited)
.unwrap_or(false);
let exit_code = sessions
.get(&session_id)
.and_then(UnifiedExecSession::exit_code);
if entry.session.has_exited() {
let Some(entry) = sessions.remove(&session_id) else {
return SessionStatus::Unknown;
};
SessionStatus::Exited {
exit_code,
entry: Box::new(entry),
}
if has_exited {
sessions.remove(&session_id);
(None, exit_code)
} else {
SessionStatus::Alive {
exit_code,
call_id: entry.call_id.clone(),
}
(Some(session_id), exit_code)
}
}
@@ -186,9 +138,9 @@ impl UnifiedExecSessionManager {
) -> Result<(mpsc::Sender<Vec<u8>>, OutputBuffer, Arc<Notify>), UnifiedExecError> {
let sessions = self.sessions.lock().await;
let (output_buffer, output_notify, writer_tx) =
if let Some(entry) = sessions.get(&session_id) {
let (buffer, notify) = entry.session.output_handles();
(buffer, notify, entry.session.writer_sender())
if let Some(session) = sessions.get(&session_id) {
let (buffer, notify) = session.output_handles();
(buffer, notify, session.writer_sender())
} else {
return Err(UnifiedExecError::UnknownSessionId { session_id });
};
@@ -206,82 +158,14 @@ impl UnifiedExecSessionManager {
.map_err(|_| UnifiedExecError::WriteToStdin)
}
async fn store_session(
&self,
session: UnifiedExecSession,
context: &UnifiedExecContext,
command: &str,
started_at: Instant,
) -> i32 {
async fn store_session(&self, session: UnifiedExecSession) -> i32 {
let session_id = self
.next_session_id
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let entry = SessionEntry {
session,
session_ref: Arc::clone(&context.session),
turn_ref: Arc::clone(&context.turn),
call_id: context.call_id.clone(),
command: command.to_string(),
cwd: context.turn.cwd.clone(),
started_at,
};
self.sessions.lock().await.insert(session_id, entry);
self.sessions.lock().await.insert(session_id, session);
session_id
}
async fn emit_exec_end_from_entry(
entry: SessionEntry,
aggregated_output: String,
exit_code: i32,
duration: Duration,
) {
let output = ExecToolCallOutput {
exit_code,
stdout: StreamOutput::new(aggregated_output.clone()),
stderr: StreamOutput::new(String::new()),
aggregated_output: StreamOutput::new(aggregated_output),
duration,
timed_out: false,
};
let event_ctx = ToolEventCtx::new(
entry.session_ref.as_ref(),
entry.turn_ref.as_ref(),
&entry.call_id,
None,
);
let emitter = ToolEmitter::unified_exec(entry.command, entry.cwd, true);
emitter
.emit(event_ctx, ToolEventStage::Success(output))
.await;
}
async fn emit_exec_end_from_context(
context: &UnifiedExecContext,
command: String,
aggregated_output: String,
exit_code: i32,
duration: Duration,
) {
let output = ExecToolCallOutput {
exit_code,
stdout: StreamOutput::new(aggregated_output.clone()),
stderr: StreamOutput::new(String::new()),
aggregated_output: StreamOutput::new(aggregated_output),
duration,
timed_out: false,
};
let event_ctx = ToolEventCtx::new(
context.session.as_ref(),
context.turn.as_ref(),
&context.call_id,
None,
);
let emitter = ToolEmitter::unified_exec(command, context.turn.cwd.clone(), true);
emitter
.emit(event_ctx, ToolEventStage::Success(output))
.await;
}
pub(crate) async fn open_session_with_exec_env(
&self,
env: &ExecEnv,
@@ -300,7 +184,7 @@ impl UnifiedExecSessionManager {
pub(super) async fn open_session_with_sandbox(
&self,
command: Vec<String>,
context: &UnifiedExecContext,
context: &UnifiedExecContext<'_>,
) -> Result<UnifiedExecSession, UnifiedExecError> {
let mut orchestrator = ToolOrchestrator::new();
let mut runtime = UnifiedExecRuntime::new(self);
@@ -310,9 +194,9 @@ impl UnifiedExecSessionManager {
create_env(&context.turn.shell_environment_policy),
);
let tool_ctx = ToolCtx {
session: context.session.as_ref(),
turn: context.turn.as_ref(),
call_id: context.call_id.clone(),
session: context.session,
turn: context.turn,
call_id: context.call_id.to_string(),
tool_name: "exec_command".to_string(),
};
orchestrator
@@ -320,7 +204,7 @@ impl UnifiedExecSessionManager {
&mut runtime,
&req,
&tool_ctx,
context.turn.as_ref(),
context.turn,
context.turn.approval_policy,
)
.await
@@ -371,15 +255,3 @@ impl UnifiedExecSessionManager {
collected
}
}
enum SessionStatus {
Alive {
exit_code: Option<i32>,
call_id: String,
},
Exited {
exit_code: Option<i32>,
entry: Box<SessionEntry>,
},
Unknown,
}

View File

@@ -10,7 +10,6 @@ path = "lib.rs"
anyhow = { workspace = true }
assert_cmd = { workspace = true }
codex-core = { workspace = true }
codex-protocol = { workspace = true }
notify = { workspace = true }
regex-lite = { workspace = true }
serde_json = { workspace = true }

View File

@@ -1,30 +1,17 @@
use std::mem::swap;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::CodexConversation;
use codex_core::ConversationManager;
use codex_core::ModelProviderInfo;
use codex_core::built_in_model_providers;
use codex_core::config::Config;
use codex_core::features::Feature;
use codex_core::protocol::AskForApproval;
use codex_core::protocol::EventMsg;
use codex_core::protocol::Op;
use codex_core::protocol::SandboxPolicy;
use codex_core::protocol::SessionConfiguredEvent;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::user_input::UserInput;
use serde_json::Value;
use tempfile::TempDir;
use wiremock::MockServer;
use crate::load_default_config_for_test;
use crate::responses::start_mock_server;
use crate::wait_for_event;
type ConfigMutator = dyn FnOnce(&mut Config) + Send;
@@ -109,12 +96,6 @@ impl TestCodexBuilder {
mutator(&mut config);
}
if config.include_apply_patch_tool {
config.features.enable(Feature::ApplyPatchFreeform);
} else {
config.features.disable(Feature::ApplyPatchFreeform);
}
Ok((config, cwd))
}
}
@@ -126,139 +107,6 @@ pub struct TestCodex {
pub session_configured: SessionConfiguredEvent,
}
impl TestCodex {
pub fn cwd_path(&self) -> &Path {
self.cwd.path()
}
pub fn workspace_path(&self, rel: impl AsRef<Path>) -> PathBuf {
self.cwd_path().join(rel)
}
pub async fn submit_turn(&self, prompt: &str) -> Result<()> {
self.submit_turn_with_policy(prompt, SandboxPolicy::DangerFullAccess)
.await
}
pub async fn submit_turn_with_policy(
&self,
prompt: &str,
sandbox_policy: SandboxPolicy,
) -> Result<()> {
let session_model = self.session_configured.model.clone();
self.codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: prompt.into(),
}],
final_output_json_schema: None,
cwd: self.cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
wait_for_event(&self.codex, |event| {
matches!(event, EventMsg::TaskComplete(_))
})
.await;
Ok(())
}
}
pub struct TestCodexHarness {
server: MockServer,
test: TestCodex,
}
impl TestCodexHarness {
pub async fn new() -> Result<Self> {
Self::with_builder(test_codex()).await
}
pub async fn with_config(mutator: impl FnOnce(&mut Config) + Send + 'static) -> Result<Self> {
Self::with_builder(test_codex().with_config(mutator)).await
}
pub async fn with_builder(mut builder: TestCodexBuilder) -> Result<Self> {
let server = start_mock_server().await;
let test = builder.build(&server).await?;
Ok(Self { server, test })
}
pub fn server(&self) -> &MockServer {
&self.server
}
pub fn test(&self) -> &TestCodex {
&self.test
}
pub fn cwd(&self) -> &Path {
self.test.cwd_path()
}
pub fn path(&self, rel: impl AsRef<Path>) -> PathBuf {
self.test.workspace_path(rel)
}
pub async fn submit(&self, prompt: &str) -> Result<()> {
self.test.submit_turn(prompt).await
}
pub async fn submit_with_policy(
&self,
prompt: &str,
sandbox_policy: SandboxPolicy,
) -> Result<()> {
self.test
.submit_turn_with_policy(prompt, sandbox_policy)
.await
}
pub async fn request_bodies(&self) -> Vec<Value> {
self.server
.received_requests()
.await
.expect("requests")
.into_iter()
.map(|req| serde_json::from_slice(&req.body).expect("request body json"))
.collect()
}
pub async fn function_call_output_value(&self, call_id: &str) -> Value {
let bodies = self.request_bodies().await;
function_call_output(&bodies, call_id).clone()
}
pub async fn function_call_stdout(&self, call_id: &str) -> String {
self.function_call_output_value(call_id)
.await
.get("output")
.and_then(Value::as_str)
.expect("output string")
.to_string()
}
}
fn function_call_output<'a>(bodies: &'a [Value], call_id: &str) -> &'a Value {
for body in bodies {
if let Some(items) = body.get("input").and_then(Value::as_array) {
for item in items {
if item.get("type").and_then(Value::as_str) == Some("function_call_output")
&& item.get("call_id").and_then(Value::as_str) == Some(call_id)
{
return item;
}
}
}
}
panic!("function_call_output {call_id} not found");
}
pub fn test_codex() -> TestCodexBuilder {
TestCodexBuilder {
config_mutators: vec![],

File diff suppressed because it is too large Load Diff

View File

@@ -279,6 +279,11 @@ async fn auto_compact_runs_after_token_limit_hit() {
ev_completed_with_tokens("r2", 330_000),
]);
let sse3 = sse(vec![
ev_assistant_message("m3", AUTO_SUMMARY_TEXT),
ev_completed_with_tokens("r3", 200),
]);
let first_matcher = |req: &wiremock::Request| {
let body = std::str::from_utf8(&req.body).unwrap_or("");
body.contains(FIRST_AUTO_MSG)
@@ -295,6 +300,12 @@ async fn auto_compact_runs_after_token_limit_hit() {
};
mount_sse_once_match(&server, second_matcher, sse2).await;
let third_matcher = |req: &wiremock::Request| {
let body = std::str::from_utf8(&req.body).unwrap_or("");
body.contains("You have exceeded the maximum number of tokens")
};
mount_sse_once_match(&server, third_matcher, sse3).await;
let model_provider = ModelProviderInfo {
base_url: Some(format!("{}/v1", server.uri())),
..built_in_model_providers()["openai"].clone()
@@ -331,28 +342,69 @@ async fn auto_compact_runs_after_token_limit_hit() {
.await
.unwrap();
let error_event = wait_for_event(&codex, |ev| matches!(ev, EventMsg::Error(_))).await;
let EventMsg::Error(error_event) = error_event else {
unreachable!("wait_for_event returned unexpected payload");
};
assert_eq!(error_event.message, "invalid input: input too large");
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
// wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
let requests = server.received_requests().await.unwrap();
assert_eq!(
requests.len(),
2,
"auto compact should reject oversize prompts before issuing another request"
assert!(
requests.len() >= 3,
"auto compact should add at least a third request, got {}",
requests.len()
);
let saw_compact_prompt = requests.iter().any(|req| {
let is_auto_compact = |req: &wiremock::Request| {
std::str::from_utf8(&req.body)
.unwrap_or("")
.contains("You have exceeded the maximum number of tokens")
});
assert!(
!saw_compact_prompt,
"no auto compact request should be sent when the summarization prompt exceeds the limit"
};
let auto_compact_count = requests.iter().filter(|req| is_auto_compact(req)).count();
assert_eq!(
auto_compact_count, 1,
"expected exactly one auto compact request"
);
let auto_compact_index = requests
.iter()
.enumerate()
.find_map(|(idx, req)| is_auto_compact(req).then_some(idx))
.expect("auto compact request missing");
assert_eq!(
auto_compact_index, 2,
"auto compact should add a third request"
);
let body_first = requests[0].body_json::<serde_json::Value>().unwrap();
let body3 = requests[auto_compact_index]
.body_json::<serde_json::Value>()
.unwrap();
let instructions = body3
.get("instructions")
.and_then(|v| v.as_str())
.unwrap_or_default();
let baseline_instructions = body_first
.get("instructions")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
assert_eq!(
instructions, baseline_instructions,
"auto compact should keep the standard developer instructions",
);
let input3 = body3.get("input").and_then(|v| v.as_array()).unwrap();
let last3 = input3
.last()
.expect("auto compact request should append a user message");
assert_eq!(last3.get("type").and_then(|v| v.as_str()), Some("message"));
assert_eq!(last3.get("role").and_then(|v| v.as_str()), Some("user"));
let last_text = last3
.get("content")
.and_then(|v| v.as_array())
.and_then(|items| items.first())
.and_then(|item| item.get("text"))
.and_then(|text| text.as_str())
.unwrap_or_default();
assert_eq!(
last_text, SUMMARIZATION_PROMPT,
"auto compact should send the summarization prompt as a user message",
);
}
@@ -817,7 +869,7 @@ async fn auto_compact_triggers_after_function_call_over_95_percent_usage() {
let server = start_mock_server().await;
let context_window = 20_000;
let context_window = 100;
let limit = context_window * 90 / 100;
let over_limit_tokens = context_window * 95 / 100 + 1;

View File

@@ -1,69 +0,0 @@
use codex_core::protocol::EventMsg;
use codex_core::protocol::Op;
use codex_protocol::user_input::UserInput;
use core_test_support::responses;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event_with_timeout;
use std::sync::Arc;
use std::time::Duration;
use wiremock::matchers::any;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn input_validation_should_fail_for_too_large_input() {
let server = start_mock_server().await;
let fixture = test_codex().build(&server).await.unwrap();
let codex = Arc::clone(&fixture.codex);
// First: normal message with a mocked assistant response
let first_response = sse(vec![
ev_response_created("resp-1"),
ev_assistant_message("msg-1", "ok"),
ev_completed("resp-1"),
]);
responses::mount_sse_once_match(&server, any(), first_response).await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "hello world".into(),
}],
})
.await
.unwrap();
// Wait for the normal turn to complete before sending the oversized input
let turn_timeout = Duration::from_secs(1);
wait_for_event_with_timeout(
&codex,
|ev| matches!(ev, EventMsg::TaskComplete(_)),
turn_timeout,
)
.await;
// Then: 300k-token message should trigger validation error
let wait_timeout = Duration::from_millis(100);
let input_300_tokens = "token ".repeat(300_000);
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: input_300_tokens,
}],
})
.await
.unwrap();
let error_event =
wait_for_event_with_timeout(&codex, |ev| matches!(ev, EventMsg::Error(_)), wait_timeout)
.await;
let EventMsg::Error(error_event) = error_event else {
unreachable!("wait_for_event_with_timeout returned unexpected payload");
};
assert_eq!(error_event.message, "invalid input: input too large");
}

View File

@@ -3,8 +3,6 @@
#[cfg(not(target_os = "windows"))]
mod abort_tasks;
#[cfg(not(target_os = "windows"))]
mod apply_patch_cli;
#[cfg(not(target_os = "windows"))]
mod approvals;
mod cli_stream;
mod client;
@@ -13,7 +11,6 @@ mod compact_resume_fork;
mod exec;
mod fork_conversation;
mod grep_files;
mod input_validation;
mod items;
mod json_result;
mod list_dir;

View File

@@ -1,6 +1,5 @@
#![allow(clippy::unwrap_used, clippy::expect_used)]
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use codex_core::find_conversation_path_by_id_str;
@@ -9,8 +8,8 @@ use uuid::Uuid;
/// Create sessions/YYYY/MM/DD and write a minimal rollout file containing the
/// provided conversation id in the SessionMeta line. Returns the absolute path.
fn write_minimal_rollout_with_id(codex_home: &Path, id: Uuid) -> PathBuf {
let sessions = codex_home.join("sessions/2024/01/01");
fn write_minimal_rollout_with_id(codex_home: &TempDir, id: Uuid) -> PathBuf {
let sessions = codex_home.path().join("sessions/2024/01/01");
std::fs::create_dir_all(&sessions).unwrap();
let file = sessions.join(format!("rollout-2024-01-01T00-00-00-{id}.jsonl"));
@@ -41,7 +40,7 @@ fn write_minimal_rollout_with_id(codex_home: &Path, id: Uuid) -> PathBuf {
async fn find_locates_rollout_file_by_id() {
let home = TempDir::new().unwrap();
let id = Uuid::new_v4();
let expected = write_minimal_rollout_with_id(home.path(), id);
let expected = write_minimal_rollout_with_id(&home, id);
let found = find_conversation_path_by_id_str(home.path(), &id.to_string())
.await
@@ -49,33 +48,3 @@ async fn find_locates_rollout_file_by_id() {
assert_eq!(found.unwrap(), expected);
}
#[tokio::test]
async fn find_handles_gitignore_covering_codex_home_directory() {
let repo = TempDir::new().unwrap();
let codex_home = repo.path().join(".codex");
std::fs::create_dir_all(&codex_home).unwrap();
std::fs::write(repo.path().join(".gitignore"), ".codex/**\n").unwrap();
let id = Uuid::new_v4();
let expected = write_minimal_rollout_with_id(&codex_home, id);
let found = find_conversation_path_by_id_str(&codex_home, &id.to_string())
.await
.unwrap();
assert_eq!(found, Some(expected));
}
#[tokio::test]
async fn find_ignores_granular_gitignore_rules() {
let home = TempDir::new().unwrap();
let id = Uuid::new_v4();
let expected = write_minimal_rollout_with_id(home.path(), id);
std::fs::write(home.path().join("sessions/.gitignore"), "*.jsonl\n").unwrap();
let found = find_conversation_path_by_id_str(home.path(), &id.to_string())
.await
.unwrap();
assert_eq!(found, Some(expected));
}

View File

@@ -133,247 +133,6 @@ async fn unified_exec_emits_exec_command_begin_event() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_exec_command_end_event() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.use_experimental_unified_exec_tool = true;
config.features.enable(Feature::UnifiedExec);
});
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let call_id = "uexec-end-event";
let args = json!({
"cmd": "/bin/echo END-EVENT".to_string(),
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "emit end event".into(),
}],
final_output_json_schema: None,
cwd: cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::DangerFullAccess,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
let end_event = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => Some(ev.clone()),
_ => None,
})
.await;
assert_eq!(end_event.exit_code, 0);
assert!(
end_event.aggregated_output.contains("END-EVENT"),
"expected aggregated output to contain marker"
);
wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_output_delta_for_exec_command() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.use_experimental_unified_exec_tool = true;
config.features.enable(Feature::UnifiedExec);
});
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let call_id = "uexec-delta-1";
let args = json!({
"cmd": "printf 'HELLO-UEXEC'",
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "emit delta".into(),
}],
final_output_json_schema: None,
cwd: cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::DangerFullAccess,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
let delta = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandOutputDelta(ev) if ev.call_id == call_id => Some(ev.clone()),
_ => None,
})
.await;
let text = String::from_utf8_lossy(&delta.chunk).to_string();
assert!(
text.contains("HELLO-UEXEC"),
"delta chunk missing expected text: {text:?}"
);
wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_output_delta_for_write_stdin() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.use_experimental_unified_exec_tool = true;
config.features.enable(Feature::UnifiedExec);
});
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let open_call_id = "uexec-open";
let open_args = json!({
"cmd": "/bin/bash -i",
"yield_time_ms": 200,
});
let stdin_call_id = "uexec-stdin-delta";
let stdin_args = json!({
"chars": "echo WSTDIN-MARK\\n",
"session_id": 0,
"yield_time_ms": 800,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
open_call_id,
"exec_command",
&serde_json::to_string(&open_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
stdin_call_id,
"write_stdin",
&serde_json::to_string(&stdin_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "stdin delta".into(),
}],
final_output_json_schema: None,
cwd: cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::DangerFullAccess,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
// Expect a delta event corresponding to the write_stdin call.
let delta = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandOutputDelta(ev) if ev.call_id == open_call_id => {
let text = String::from_utf8_lossy(&ev.chunk);
if text.contains("WSTDIN-MARK") {
Some(ev.clone())
} else {
None
}
}
_ => None,
})
.await;
let text = String::from_utf8_lossy(&delta.chunk).to_string();
assert!(
text.contains("WSTDIN-MARK"),
"stdin delta chunk missing expected text: {text:?}"
);
wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> {
use tokio::time::Duration;
@@ -757,110 +516,6 @@ async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.use_experimental_unified_exec_tool = true;
config.features.enable(Feature::UnifiedExec);
});
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let start_call_id = "uexec-end-on-exit-start";
let start_args = serde_json::json!({
"cmd": "/bin/cat",
"yield_time_ms": 200,
});
let echo_call_id = "uexec-end-on-exit-echo";
let echo_args = serde_json::json!({
"chars": "bye-END\n",
"session_id": 0,
"yield_time_ms": 300,
});
let exit_call_id = "uexec-end-on-exit";
let exit_args = serde_json::json!({
"chars": "\u{0004}",
"session_id": 0,
"yield_time_ms": 500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
echo_call_id,
"write_stdin",
&serde_json::to_string(&echo_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_function_call(
exit_call_id,
"write_stdin",
&serde_json::to_string(&exit_args)?,
),
ev_completed("resp-3"),
]),
sse(vec![
ev_response_created("resp-4"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-4"),
]),
];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "end on exit".into(),
}],
final_output_json_schema: None,
cwd: cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::DangerFullAccess,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
// We expect the ExecCommandEnd event to match the initial exec_command call_id.
let end_event = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandEnd(ev) if ev.call_id == start_call_id => Some(ev.clone()),
_ => None,
})
.await;
assert_eq!(end_event.exit_code, 0);
wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_reuses_session_via_stdin() -> Result<()> {
skip_if_no_network!(Ok(()));

View File

@@ -105,7 +105,6 @@ pub async fn run_main<T: Reporter>(
threads,
cancel_flag,
compute_indices,
true,
)?;
let match_count = matches.len();
let matches_truncated = total_match_count > match_count;
@@ -122,7 +121,6 @@ pub async fn run_main<T: Reporter>(
/// The worker threads will periodically check `cancel_flag` to see if they
/// should stop processing files.
#[allow(clippy::too_many_arguments)]
pub fn run(
pattern_text: &str,
limit: NonZero<usize>,
@@ -131,7 +129,6 @@ pub fn run(
threads: NonZero<usize>,
cancel_flag: Arc<AtomicBool>,
compute_indices: bool,
respect_gitignore: bool,
) -> anyhow::Result<FileSearchResults> {
let pattern = create_pattern(pattern_text);
// Create one BestMatchesList per worker thread so that each worker can
@@ -160,14 +157,6 @@ pub fn run(
.hidden(false)
// Don't require git to be present to apply to apply git-related ignore rules.
.require_git(false);
if !respect_gitignore {
walk_builder
.git_ignore(false)
.git_global(false)
.git_exclude(false)
.ignore(false)
.parents(false);
}
if !exclude.is_empty() {
let mut override_builder = OverrideBuilder::new(search_directory);

View File

@@ -1,35 +0,0 @@
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
use ts_rs::TS;
#[derive(Serialize, Deserialize, Copy, Clone, Debug, PartialEq, Eq, JsonSchema, TS, Default)]
#[serde(rename_all = "lowercase")]
#[ts(rename_all = "lowercase")]
pub enum PlanType {
#[default]
Free,
Plus,
Pro,
Team,
Business,
Enterprise,
Edu,
#[serde(other)]
Unknown,
}
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema, TS)]
#[serde(tag = "type")]
#[ts(tag = "type")]
pub enum Account {
ApiKey {
api_key: String,
},
#[serde(rename = "chatgpt")]
#[ts(rename = "chatgpt")]
ChatGpt {
email: Option<String>,
plan_type: PlanType,
},
}

View File

@@ -1,4 +1,3 @@
pub mod account;
mod conversation_id;
pub use conversation_id::ConversationId;
pub mod config_types;

View File

@@ -575,11 +575,9 @@ pub struct TokenUsage {
pub total_tokens: i64,
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS, Default)]
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct TokenUsageInfo {
/// The total token usage for the session. accumulated from all turns.
pub total_token_usage: TokenUsage,
/// The token usage for the last turn. Received from the API. It's total tokens is the whole window size.
pub last_token_usage: TokenUsage,
#[ts(type = "number | null")]
pub model_context_window: Option<i64>,
@@ -921,6 +919,7 @@ pub enum SessionSource {
VSCode,
Exec,
Mcp,
SubAgent,
#[serde(other)]
Unknown,
}

View File

@@ -2069,35 +2069,6 @@ mod tests {
}
}
#[test]
fn ascii_prefix_survives_non_ascii_followup() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = unbounded_channel::<AppEvent>();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(
true,
sender,
false,
"Ask Codex to do anything".to_string(),
false,
);
let _ = composer.handle_key_event(KeyEvent::new(KeyCode::Char('1'), KeyModifiers::NONE));
assert!(composer.is_in_paste_burst());
let _ = composer.handle_key_event(KeyEvent::new(KeyCode::Char('あ'), KeyModifiers::NONE));
let (result, _) =
composer.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
match result {
InputResult::Submitted(text) => assert_eq!(text, "1あ"),
_ => panic!("expected Submitted"),
}
}
#[test]
fn handle_paste_small_inserts_text() {
use crossterm::event::KeyCode;

View File

@@ -198,15 +198,12 @@ impl PasteBurst {
/// Before applying modified/non-char input: flush buffered burst immediately.
pub fn flush_before_modified_input(&mut self) -> Option<String> {
if !self.is_active() {
return None;
if self.is_active() {
self.active = false;
Some(std::mem::take(&mut self.buffer))
} else {
None
}
self.active = false;
let mut out = std::mem::take(&mut self.buffer);
if let Some((ch, _at)) = self.pending_first_char.take() {
out.push(ch);
}
Some(out)
}
/// Clear only the timing window and any pending first-char.

View File

@@ -745,8 +745,9 @@ impl ChatWidget {
&ev.call_id,
CommandOutput {
exit_code: ev.exit_code,
stdout: ev.stdout.clone(),
stderr: ev.stderr.clone(),
formatted_output: ev.formatted_output.clone(),
aggregated_output: ev.aggregated_output.clone(),
},
ev.duration,
);

View File

@@ -71,26 +71,18 @@ fn upgrade_event_payload_for_tests(mut payload: serde_json::Value) -> serde_json
&& let Some(m) = msg.as_object_mut()
{
let ty = m.get("type").and_then(|v| v.as_str()).unwrap_or("");
if ty == "exec_command_end" {
if ty == "exec_command_end" && !m.contains_key("formatted_output") {
let stdout = m.get("stdout").and_then(|v| v.as_str()).unwrap_or("");
let stderr = m.get("stderr").and_then(|v| v.as_str()).unwrap_or("");
let aggregated = if stderr.is_empty() {
let formatted = if stderr.is_empty() {
stdout.to_string()
} else {
format!("{stdout}{stderr}")
};
if !m.contains_key("formatted_output") {
m.insert(
"formatted_output".to_string(),
serde_json::Value::String(aggregated.clone()),
);
}
if !m.contains_key("aggregated_output") {
m.insert(
"aggregated_output".to_string(),
serde_json::Value::String(aggregated),
);
}
m.insert(
"formatted_output".to_string(),
serde_json::Value::String(formatted),
);
}
}
payload

View File

@@ -3,12 +3,11 @@ use std::time::Instant;
use codex_protocol::parse_command::ParsedCommand;
#[derive(Clone, Debug, Default)]
#[derive(Clone, Debug)]
pub(crate) struct CommandOutput {
pub(crate) exit_code: i32,
/// The aggregated stderr + stdout interleaved.
pub(crate) aggregated_output: String,
/// The formatted output of the command, as seen by the model.
pub(crate) stdout: String,
pub(crate) stderr: String,
pub(crate) formatted_output: String,
}
@@ -83,8 +82,9 @@ impl ExecCell {
call.duration = Some(elapsed);
call.output = Some(CommandOutput {
exit_code: 1,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
aggregated_output: String::new(),
});
}
}

View File

@@ -28,6 +28,7 @@ use unicode_width::UnicodeWidthStr;
pub(crate) const TOOL_CALL_MAX_LINES: usize = 5;
pub(crate) struct OutputLinesParams {
pub(crate) only_err: bool,
pub(crate) include_angle_pipe: bool,
pub(crate) include_prefix: bool,
}
@@ -58,12 +59,22 @@ pub(crate) fn output_lines(
params: OutputLinesParams,
) -> OutputLines {
let OutputLinesParams {
only_err,
include_angle_pipe,
include_prefix,
} = params;
let CommandOutput {
aggregated_output, ..
exit_code,
stdout,
stderr,
..
} = match output {
Some(output) if only_err && output.exit_code == 0 => {
return OutputLines {
lines: Vec::new(),
omitted: None,
};
}
Some(output) => output,
None => {
return OutputLines {
@@ -73,7 +84,7 @@ pub(crate) fn output_lines(
}
};
let src = aggregated_output;
let src = if *exit_code == 0 { stdout } else { stderr };
let lines: Vec<&str> = src.lines().collect();
let total = lines.len();
let limit = TOOL_CALL_MAX_LINES;
@@ -387,6 +398,7 @@ impl ExecCell {
let raw_output = output_lines(
Some(output),
OutputLinesParams {
only_err: false,
include_angle_pipe: false,
include_prefix: false,
},

View File

@@ -172,7 +172,6 @@ impl FileSearchManager {
NUM_FILE_SEARCH_THREADS,
cancellation_token.clone(),
compute_indices,
true,
)
.map(|res| res.matches)
.unwrap_or_default();

View File

@@ -1293,10 +1293,12 @@ pub(crate) fn new_patch_apply_failure(stderr: String) -> PlainHistoryCell {
let output = output_lines(
Some(&CommandOutput {
exit_code: 1,
stdout: String::new(),
stderr,
formatted_output: String::new(),
aggregated_output: stderr,
}),
OutputLinesParams {
only_err: true,
include_angle_pipe: true,
include_prefix: true,
},
@@ -1737,7 +1739,16 @@ mod tests {
duration: None,
});
// Mark call complete so markers are ✓
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(80);
let rendered = render_lines(&lines).join("\n");
@@ -1759,7 +1770,16 @@ mod tests {
duration: None,
});
// Call 1: Search only
cell.complete_call("c1", CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
"c1",
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
// Call 2: Read A
cell = cell
.with_added_call(
@@ -1772,7 +1792,16 @@ mod tests {
}],
)
.unwrap();
cell.complete_call("c2", CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
"c2",
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
// Call 3: Read B
cell = cell
.with_added_call(
@@ -1785,7 +1814,16 @@ mod tests {
}],
)
.unwrap();
cell.complete_call("c3", CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
"c3",
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(80);
let rendered = render_lines(&lines).join("\n");
@@ -1818,7 +1856,16 @@ mod tests {
start_time: Some(Instant::now()),
duration: None,
});
cell.complete_call("c1", CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
"c1",
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(80);
let rendered = render_lines(&lines).join("\n");
insta::assert_snapshot!(rendered);
@@ -1838,7 +1885,16 @@ mod tests {
duration: None,
});
// Mark call complete so it renders as "Ran"
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
// Small width to force wrapping on both lines
let width: u16 = 28;
@@ -1858,7 +1914,16 @@ mod tests {
start_time: Some(Instant::now()),
duration: None,
});
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
// Wide enough that it fits inline
let lines = cell.display_lines(80);
let rendered = render_lines(&lines).join("\n");
@@ -1877,7 +1942,16 @@ mod tests {
start_time: Some(Instant::now()),
duration: None,
});
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(24);
let rendered = render_lines(&lines).join("\n");
insta::assert_snapshot!(rendered);
@@ -1895,7 +1969,16 @@ mod tests {
start_time: Some(Instant::now()),
duration: None,
});
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(80);
let rendered = render_lines(&lines).join("\n");
insta::assert_snapshot!(rendered);
@@ -1914,7 +1997,16 @@ mod tests {
start_time: Some(Instant::now()),
duration: None,
});
cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1));
cell.complete_call(
&call_id,
CommandOutput {
exit_code: 0,
stdout: String::new(),
stderr: String::new(),
formatted_output: String::new(),
},
Duration::from_millis(1),
);
let lines = cell.display_lines(28);
let rendered = render_lines(&lines).join("\n");
insta::assert_snapshot!(rendered);
@@ -1941,8 +2033,9 @@ mod tests {
&call_id,
CommandOutput {
exit_code: 1,
stdout: String::new(),
stderr,
formatted_output: String::new(),
aggregated_output: stderr,
},
Duration::from_millis(1),
);
@@ -1984,8 +2077,9 @@ mod tests {
&call_id,
CommandOutput {
exit_code: 1,
stdout: String::new(),
stderr,
formatted_output: String::new(),
aggregated_output: stderr,
},
Duration::from_millis(5),
);

View File

@@ -6,9 +6,6 @@ use ratatui::style::Style;
use ratatui::style::Stylize;
use ratatui::text::Span;
#[cfg(target_os = "macos")]
const ALT_PREFIX: &str = "⌥ + ";
#[cfg(not(target_os = "macos"))]
const ALT_PREFIX: &str = "alt + ";
const CTRL_PREFIX: &str = "ctrl + ";
const SHIFT_PREFIX: &str = "shift + ";

View File

@@ -724,7 +724,8 @@ mod tests {
"exec-1",
CommandOutput {
exit_code: 0,
aggregated_output: "src\nREADME.md\n".into(),
stdout: "src\nREADME.md\n".into(),
stderr: String::new(),
formatted_output: "src\nREADME.md\n".into(),
},
Duration::from_millis(420),

View File

@@ -1,13 +0,0 @@
---
source: tui/src/status_indicator_widget.rs
assertion_line: 289
expression: terminal.backend()
---
"• Working (0s • esc to interrupt) "
" "
" ↳ first "
" ↳ second "
" ⌥ + ↑ edit "
" "
" "
" "

View File

@@ -284,11 +284,6 @@ mod tests {
terminal
.draw(|f| w.render_ref(f.area(), f.buffer_mut()))
.expect("draw");
#[cfg(target_os = "macos")]
insta::with_settings!({ snapshot_suffix => "macos" }, {
insta::assert_snapshot!(terminal.backend());
});
#[cfg(not(target_os = "macos"))]
insta::assert_snapshot!(terminal.backend());
}

View File

@@ -55,13 +55,8 @@ impl Tokenizer {
Ok(Self { inner })
}
/// Default to `O200kBase`
pub fn try_default() -> Result<Self, TokenizerError> {
Self::new(EncodingKind::O200kBase)
}
/// Build a tokenizer using an `OpenAI` model name (maps to an encoding).
/// Falls back to the `O200kBase` encoding when the model is unknown.
/// Falls back to the `o200k_base` encoding when the model is unknown.
pub fn for_model(model: &str) -> Result<Self, TokenizerError> {
match tiktoken_rs::get_bpe_from_model(model) {
Ok(inner) => Ok(Self { inner }),
@@ -107,12 +102,6 @@ impl Tokenizer {
}
}
impl fmt::Debug for Tokenizer {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Tokenizer").finish()
}
}
#[cfg(test)]
mod tests {
use super::*;