Compare commits

..

1 Commits

Author SHA1 Message Date
aibrahim-oai
21b12185e3 test: add data-driven SSE tests and fixtures 2025-07-11 13:55:05 -07:00
50 changed files with 685 additions and 2652 deletions

View File

@@ -8,8 +8,8 @@
"@actions/github": "^6.0.1",
},
"devDependencies": {
"@types/bun": "^1.2.18",
"@types/node": "^24.0.13",
"@types/bun": "^1.2.11",
"@types/node": "^22.15.21",
"prettier": "^3.6.2",
"typescript": "^5.8.3",
},
@@ -48,17 +48,13 @@
"@octokit/types": ["@octokit/types@13.10.0", "", { "dependencies": { "@octokit/openapi-types": "^24.2.0" } }, "sha512-ifLaO34EbbPj0Xgro4G5lP5asESjwHracYJvVaPIyXMuiuXLlhic3S47cBdTb+jfODkTE5YtGCLt3Ay3+J97sA=="],
"@types/bun": ["@types/bun@1.2.18", "", { "dependencies": { "bun-types": "1.2.18" } }, "sha512-Xf6RaWVheyemaThV0kUfaAUvCNokFr+bH8Jxp+tTZfx7dAPA8z9ePnP9S9+Vspzuxxx9JRAXhnyccRj3GyCMdQ=="],
"@types/bun": ["@types/bun@1.2.13", "", { "dependencies": { "bun-types": "1.2.13" } }, "sha512-u6vXep/i9VBxoJl3GjZsl/BFIsvML8DfVDO0RYLEwtSZSp981kEO1V5NwRcO1CPJ7AmvpbnDCiMKo3JvbDEjAg=="],
"@types/node": ["@types/node@24.0.13", "", { "dependencies": { "undici-types": "~7.8.0" } }, "sha512-Qm9OYVOFHFYg3wJoTSrz80hoec5Lia/dPp84do3X7dZvLikQvM1YpmvTBEdIr/e+U8HTkFjLHLnl78K/qjf+jQ=="],
"@types/react": ["@types/react@19.1.8", "", { "dependencies": { "csstype": "^3.0.2" } }, "sha512-AwAfQ2Wa5bCx9WP8nZL2uMZWod7J7/JSplxbTmBQ5ms6QpqNYm672H0Vu9ZVKVngQ+ii4R/byguVEUZQyeg44g=="],
"@types/node": ["@types/node@22.15.21", "", { "dependencies": { "undici-types": "~6.21.0" } }, "sha512-EV/37Td6c+MgKAbkcLG6vqZ2zEYHD7bvSrzqqs2RIhbA6w3x+Dqz8MZM3sP6kGTeLrdoOgKZe+Xja7tUB2DNkQ=="],
"before-after-hook": ["before-after-hook@2.2.3", "", {}, "sha512-NzUnlZexiaH/46WDhANlyR2bXRopNg4F/zuSA3OpZnllCUgRaOF2znDioDWrmbNVsuZk6l9pMquQB38cfBZwkQ=="],
"bun-types": ["bun-types@1.2.18", "", { "dependencies": { "@types/node": "*" }, "peerDependencies": { "@types/react": "^19" } }, "sha512-04+Eha5NP7Z0A9YgDAzMk5PHR16ZuLVa83b26kH5+cp1qZW4F6FmAURngE7INf4tKOvCE69vYvDEwoNl1tGiWw=="],
"csstype": ["csstype@3.1.3", "", {}, "sha512-M1uQkMl8rQK/szD0LNhtqxIPLpimGm8sOBwU7lLnCpSbTyY3yeU1Vc7l4KT5zT4s/yOxHH5O7tIuuLOCnLADRw=="],
"bun-types": ["bun-types@1.2.13", "", { "dependencies": { "@types/node": "*" } }, "sha512-rRjA1T6n7wto4gxhAO/ErZEtOXyEZEmnIHQfl0Dt1QQSB4QV0iP6BZ9/YB5fZaHFQ2dwHFrmPaRQ9GGMX01k9Q=="],
"deprecation": ["deprecation@2.3.1", "", {}, "sha512-xmHIy4F3scKVwMsQ4WnVaS8bHOx0DmVwRywosKhaILI0ywMDWPtBSku2HNxRvF7jtwDRsoEwYQSfbxj8b7RlJQ=="],
@@ -72,7 +68,7 @@
"undici": ["undici@5.29.0", "", { "dependencies": { "@fastify/busboy": "^2.0.0" } }, "sha512-raqeBD6NQK4SkWhQzeYKd1KmIG6dllBOTt55Rmkt4HtI9mwdWtJljnrXjAFUBLTSN67HWrOIZ3EPF4kjUw80Bg=="],
"undici-types": ["undici-types@7.8.0", "", {}, "sha512-9UJ2xGDvQ43tYyVMpuHlsgApydB8ZKfVYTsLDhXkFL/6gfkp+U8xTGdh8pMJv1SpZna0zxG1DwsKZsreLbXBxw=="],
"undici-types": ["undici-types@6.21.0", "", {}, "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ=="],
"universal-user-agent": ["universal-user-agent@6.0.1", "", {}, "sha512-yCzhz6FN2wU1NiiQRogkTQszlQSlpWaw8SvVegAc+bDxbzHgh1vX8uIe8OYyMH6DwH+sdTJsgMl36+mSMdRJIQ=="],

View File

@@ -13,8 +13,8 @@
"@actions/github": "^6.0.1"
},
"devDependencies": {
"@types/bun": "^1.2.18",
"@types/node": "^24.0.13",
"@types/bun": "^1.2.11",
"@types/node": "^22.15.21",
"prettier": "^3.6.2",
"typescript": "^5.8.3"
}

18
.vscode/launch.json vendored
View File

@@ -1,18 +0,0 @@
{
"version": "0.2.0",
"configurations": [
{
"type": "lldb",
"request": "launch",
"name": "Cargo launch",
"cargo": {
"cwd": "${workspaceFolder}/codex-rs",
"args": [
"build",
"--bin=codex-tui"
]
},
"args": []
}
]
}

10
.vscode/settings.json vendored
View File

@@ -1,10 +0,0 @@
{
"rust-analyzer.checkOnSave": true,
"rust-analyzer.check.command": "clippy",
"rust-analyzer.check.extraArgs": ["--all-features", "--tests"],
"rust-analyzer.rustfmt.extraArgs": ["--config", "imports_granularity=Item"],
"[rust]": {
"editor.defaultFormatter": "rust-lang.rust-analyzer",
"editor.formatOnSave": true,
}
}

View File

@@ -3,7 +3,3 @@
In the codex-rs folder where the rust code lives:
- Never add or modify any code related to `CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR`. You operate in a sandbox where `CODEX_SANDBOX_NETWORK_DISABLED=1` will be set whenever you use the `shell` tool. Any existing code that uses `CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR` was authored with this fact in mind. It is often used to early exit out of tests that the author knew you would not be able to run given your sandbox limitations.
Before creating a pull request with changes to `codex-rs`, run `just fmt` (in `codex-rs` directory) to format the code and `just fix` (in `codex-rs` directory) to fix any linter issues in the code, ensure the test suite passes by running `cargo test --all-features` in the `codex-rs` directory.
When making individual changes prefer running tests on individual files or projects first.

View File

@@ -15,6 +15,7 @@
* current platform / architecture, an error is thrown.
*/
import { spawnSync } from "child_process";
import fs from "fs";
import path from "path";
import { fileURLToPath, pathToFileURL } from "url";
@@ -34,7 +35,7 @@ const wantsNative = fs.existsSync(path.join(__dirname, "use-native")) ||
: false);
// Try native binary if requested.
if (wantsNative && process.platform !== 'win32') {
if (wantsNative) {
const { platform, arch } = process;
let targetTriple = null;
@@ -73,76 +74,22 @@ if (wantsNative && process.platform !== 'win32') {
}
const binaryPath = path.join(__dirname, "..", "bin", `codex-${targetTriple}`);
// Use an asynchronous spawn instead of spawnSync so that Node is able to
// respond to signals (e.g. Ctrl-C / SIGINT) while the native binary is
// executing. This allows us to forward those signals to the child process
// and guarantees that when either the child terminates or the parent
// receives a fatal signal, both processes exit in a predictable manner.
const { spawn } = await import("child_process");
const child = spawn(binaryPath, process.argv.slice(2), {
const result = spawnSync(binaryPath, process.argv.slice(2), {
stdio: "inherit",
});
child.on("error", (err) => {
// Typically triggered when the binary is missing or not executable.
// Re-throwing here will terminate the parent with a non-zero exit code
// while still printing a helpful stack trace.
// eslint-disable-next-line no-console
console.error(err);
process.exit(1);
});
const exitCode = typeof result.status === "number" ? result.status : 1;
process.exit(exitCode);
}
// Forward common termination signals to the child so that it shuts down
// gracefully. In the handler we temporarily disable the default behavior of
// exiting immediately; once the child has been signaled we simply wait for
// its exit event which will in turn terminate the parent (see below).
const forwardSignal = (signal) => {
if (child.killed) {
return;
}
try {
child.kill(signal);
} catch {
/* ignore */
}
};
// Fallback: execute the original JavaScript CLI.
["SIGINT", "SIGTERM", "SIGHUP"].forEach((sig) => {
process.on(sig, () => forwardSignal(sig));
});
// Resolve the path to the compiled CLI bundle
const cliPath = path.resolve(__dirname, "../dist/cli.js");
const cliUrl = pathToFileURL(cliPath).href;
// When the child exits, mirror its termination reason in the parent so that
// shell scripts and other tooling observe the correct exit status.
// Wrap the lifetime of the child process in a Promise so that we can await
// its termination in a structured way. The Promise resolves with an object
// describing how the child exited: either via exit code or due to a signal.
const childResult = await new Promise((resolve) => {
child.on("exit", (code, signal) => {
if (signal) {
resolve({ type: "signal", signal });
} else {
resolve({ type: "code", exitCode: code ?? 1 });
}
});
});
if (childResult.type === "signal") {
// Re-emit the same signal so that the parent terminates with the expected
// semantics (this also sets the correct exit code of 128 + n).
process.kill(process.pid, childResult.signal);
} else {
process.exit(childResult.exitCode);
}
} else {
// Fallback: execute the original JavaScript CLI.
// Resolve the path to the compiled CLI bundle
const cliPath = path.resolve(__dirname, "../dist/cli.js");
const cliUrl = pathToFileURL(cliPath).href;
// Load and execute the CLI
// Load and execute the CLI
(async () => {
try {
await import(cliUrl);
} catch (err) {
@@ -150,4 +97,4 @@ if (wantsNative && process.platform !== 'win32') {
console.error(err);
process.exit(1);
}
}
})();

View File

@@ -1,9 +0,0 @@
# npm releases
Run the following:
To build the 0.2.x or later version of the npm module, which runs the Rust version of the CLI, build it as follows:
```bash
./codex-cli/scripts/stage_rust_release.py --release-version 0.6.0
```

View File

@@ -4,7 +4,10 @@
# -----------------------------------------------------------------------------
# Stages an npm release for @openai/codex.
#
# Usage:
# The script used to accept a single optional positional argument that indicated
# the temporary directory in which to stage the package. We now support a
# flag-based interface so that we can extend the command with further options
# without breaking the call-site contract.
#
# --tmp <dir> : Use <dir> instead of a freshly created temp directory.
# --native : Bundle the pre-built Rust CLI binaries for Linux alongside
@@ -138,8 +141,7 @@ popd >/dev/null
echo "Staged version $VERSION for release in $TMPDIR"
if [[ "$INCLUDE_NATIVE" -eq 1 ]]; then
echo "Verify the CLI:"
echo " node ${TMPDIR}/bin/codex.js --version"
echo "Test Rust:"
echo " node ${TMPDIR}/bin/codex.js --help"
else
echo "Test Node:"

144
codex-rs/Cargo.lock generated
View File

@@ -250,28 +250,6 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "async-stream"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
dependencies = [
"async-stream-impl",
"futures-core",
"pin-project-lite",
]
[[package]]
name = "async-stream-impl"
version = "0.3.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.104",
]
[[package]]
name = "async-trait"
version = "0.1.88"
@@ -399,15 +377,6 @@ version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6099cdc01846bc367c4e7dd630dc5966dccf36b652fae7a74e17b640411a91b2"
[[package]]
name = "block-buffer"
version = "0.10.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
dependencies = [
"generic-array",
]
[[package]]
name = "bstr"
version = "1.12.0"
@@ -680,20 +649,17 @@ dependencies = [
"seccompiler",
"serde",
"serde_json",
"sha1",
"strum_macros 0.27.1",
"tempfile",
"thiserror 2.0.12",
"time",
"tokio",
"tokio-test",
"tokio-util",
"toml 0.9.1",
"tracing",
"tree-sitter",
"tree-sitter-bash",
"uuid",
"walkdir",
"wildmatch",
"wiremock",
]
@@ -821,7 +787,6 @@ dependencies = [
"color-eyre",
"crossterm",
"image",
"insta",
"lazy_static",
"mcp-types",
"path-clean",
@@ -906,18 +871,6 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "console"
version = "0.15.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "054ccb5b10f9f2cbf51eb355ca1d05c2d279ce1804688d0db74b4733a5aeafd8"
dependencies = [
"encode_unicode",
"libc",
"once_cell",
"windows-sys 0.59.0",
]
[[package]]
name = "convert_case"
version = "0.6.0"
@@ -943,15 +896,6 @@ version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
[[package]]
name = "cpufeatures"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280"
dependencies = [
"libc",
]
[[package]]
name = "crc32fast"
version = "1.4.2"
@@ -1026,16 +970,6 @@ version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5"
[[package]]
name = "crypto-common"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
dependencies = [
"generic-array",
"typenum",
]
[[package]]
name = "ctor"
version = "0.1.26"
@@ -1186,16 +1120,6 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6184e33543162437515c2e2b48714794e37845ec9851711914eec9d308f6ebe8"
[[package]]
name = "digest"
version = "0.10.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
dependencies = [
"block-buffer",
"crypto-common",
]
[[package]]
name = "dirs"
version = "6.0.0"
@@ -1306,12 +1230,6 @@ dependencies = [
"log",
]
[[package]]
name = "encode_unicode"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
[[package]]
name = "encoding_rs"
version = "0.8.35"
@@ -1685,16 +1603,6 @@ dependencies = [
"byteorder",
]
[[package]]
name = "generic-array"
version = "0.14.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
dependencies = [
"typenum",
"version_check",
]
[[package]]
name = "getopts"
version = "0.2.23"
@@ -2202,17 +2110,6 @@ version = "2.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd"
[[package]]
name = "insta"
version = "1.43.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "154934ea70c58054b556dd430b99a98c2a7ff5309ac9891597e339b5c28f4371"
dependencies = [
"console",
"once_cell",
"similar",
]
[[package]]
name = "instability"
version = "0.3.7"
@@ -3994,17 +3891,6 @@ dependencies = [
"syn 2.0.104",
]
[[package]]
name = "sha1"
version = "0.10.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
dependencies = [
"cfg-if",
"cpufeatures",
"digest",
]
[[package]]
name = "sharded-slab"
version = "0.1.7"
@@ -4600,30 +4486,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-stream"
version = "0.1.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047"
dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-test"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2468baabc3311435b55dd935f702f42cd1b8abb7e754fb7dfb16bd36aa88f9f7"
dependencies = [
"async-stream",
"bytes",
"futures-core",
"tokio",
"tokio-stream",
]
[[package]]
name = "tokio-util"
version = "0.7.15"
@@ -4912,12 +4774,6 @@ dependencies = [
"unicode-width 0.2.0",
]
[[package]]
name = "typenum"
version = "1.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f"
[[package]]
name = "unicase"
version = "2.8.1"

View File

@@ -92,32 +92,6 @@ http_headers = { "X-Example-Header" = "example-value" }
env_http_headers = { "X-Example-Features": "EXAMPLE_FEATURES" }
```
### Per-provider network tuning
The following optional settings control retry behaviour and streaming idle timeouts **per model provider**. They must be specified inside the corresponding `[model_providers.<id>]` block in `config.toml`. (Older releases accepted toplevel keys; those are now ignored.)
Example:
```toml
[model_providers.openai]
name = "OpenAI"
base_url = "https://api.openai.com/v1"
env_key = "OPENAI_API_KEY"
# network tuning overrides (all optional; falls back to builtin defaults)
request_max_retries = 4 # retry failed HTTP requests
stream_max_retries = 10 # retry dropped SSE streams
stream_idle_timeout_ms = 300000 # 5m idle timeout
```
#### request_max_retries
How many times Codex will retry a failed HTTP request to the model provider. Defaults to `4`.
#### stream_max_retries
Number of times Codex will attempt to reconnect when a streaming response is interrupted. Defaults to `10`.
#### stream_idle_timeout_ms
How long Codex will wait for activity on a streaming response before treating the connection as lost. Defaults to `300_000` (5 minutes).
## model_provider
Identifies which provider to use from the `model_providers` map. Defaults to `"openai"`. You can override the `base_url` for the built-in `openai` provider via the `OPENAI_BASE_URL` environment variable.
@@ -470,7 +444,7 @@ Currently, `"vscode"` is the default, though Codex does not verify VS Code is in
## hide_agent_reasoning
Codex intermittently emits "reasoning" events that show the model's internal "thinking" before it produces a final answer. Some users may find these events distracting, especially in CI logs or minimal terminal output.
Codex intermittently emits "reasoning" events that show the models internal "thinking" before it produces a final answer. Some users may find these events distracting, especially in CI logs or minimal terminal output.
Setting `hide_agent_reasoning` to `true` suppresses these events in **both** the TUI as well as the headless `exec` sub-command:

View File

@@ -28,7 +28,6 @@ rand = "0.9"
reqwest = { version = "0.12", features = ["json", "stream"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha1 = "0.10.6"
strum_macros = "0.27.1"
thiserror = "2.0.12"
time = { version = "0.3", features = ["formatting", "local-offset", "macros"] }
@@ -65,6 +64,4 @@ maplit = "1.0.2"
predicates = "3"
pretty_assertions = "1.4.1"
tempfile = "3"
tokio-test = "0.4"
walkdir = "2.5.0"
wiremock = "0.6"

View File

@@ -21,6 +21,8 @@ use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::error::CodexErr;
use crate::error::Result;
use crate::flags::OPENAI_REQUEST_MAX_RETRIES;
use crate::flags::OPENAI_STREAM_IDLE_TIMEOUT_MS;
use crate::models::ContentItem;
use crate::models::ResponseItem;
use crate::openai_tools::create_tools_json_for_chat_completions_api;
@@ -119,7 +121,6 @@ pub(crate) async fn stream_chat_completions(
);
let mut attempt = 0;
let max_retries = provider.request_max_retries();
loop {
attempt += 1;
@@ -133,13 +134,9 @@ pub(crate) async fn stream_chat_completions(
match res {
Ok(resp) if resp.status().is_success() => {
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
tokio::spawn(process_chat_sse(
stream,
tx_event,
provider.stream_idle_timeout(),
));
tokio::spawn(process_chat_sse(stream, tx_event));
return Ok(ResponseStream { rx_event });
}
Ok(res) => {
@@ -149,7 +146,7 @@ pub(crate) async fn stream_chat_completions(
return Err(CodexErr::UnexpectedStatus(status, body));
}
if attempt > max_retries {
if attempt > *OPENAI_REQUEST_MAX_RETRIES {
return Err(CodexErr::RetryLimit(status));
}
@@ -165,7 +162,7 @@ pub(crate) async fn stream_chat_completions(
tokio::time::sleep(delay).await;
}
Err(e) => {
if attempt > max_retries {
if attempt > *OPENAI_REQUEST_MAX_RETRIES {
return Err(e.into());
}
let delay = backoff(attempt);
@@ -178,15 +175,14 @@ pub(crate) async fn stream_chat_completions(
/// Lightweight SSE processor for the Chat Completions streaming format. The
/// output is mapped onto Codex's internal [`ResponseEvent`] so that the rest
/// of the pipeline can stay agnostic of the underlying wire format.
async fn process_chat_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
) where
async fn process_chat_sse<S>(stream: S, tx_event: mpsc::Sender<Result<ResponseEvent>>)
where
S: Stream<Item = Result<Bytes>> + Unpin,
{
let mut stream = stream.eventsource();
let idle_timeout = *OPENAI_STREAM_IDLE_TIMEOUT_MS;
// State to accumulate a function call across streaming chunks.
// OpenAI may split the `arguments` string over multiple `delta` events
// until the chunk whose `finish_reason` is `tool_calls` is emitted. We
@@ -430,12 +426,6 @@ where
// will never appear in a Chat Completions stream.
continue;
}
Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(_))))
| Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => {
// Deltas are ignored here since aggregation waits for the
// final OutputItemDone.
continue;
}
}
}
}

View File

@@ -29,6 +29,8 @@ use crate::config_types::ReasoningSummary as ReasoningSummaryConfig;
use crate::error::CodexErr;
use crate::error::Result;
use crate::flags::CODEX_RS_SSE_FIXTURE;
use crate::flags::OPENAI_REQUEST_MAX_RETRIES;
use crate::flags::OPENAI_STREAM_IDLE_TIMEOUT_MS;
use crate::model_provider_info::ModelProviderInfo;
use crate::model_provider_info::WireApi;
use crate::models::ResponseItem;
@@ -107,7 +109,7 @@ impl ModelClient {
if let Some(path) = &*CODEX_RS_SSE_FIXTURE {
// short circuit for tests
warn!(path, "Streaming from fixture");
return stream_from_fixture(path, self.provider.clone()).await;
return stream_from_fixture(path).await;
}
let full_instructions = prompt.get_full_instructions(&self.config.model);
@@ -123,7 +125,6 @@ impl ModelClient {
reasoning,
previous_response_id: prompt.prev_id.clone(),
store: prompt.store,
// TODO: make this configurable
stream: true,
};
@@ -134,7 +135,6 @@ impl ModelClient {
);
let mut attempt = 0;
let max_retries = self.provider.request_max_retries();
loop {
attempt += 1;
@@ -148,15 +148,11 @@ impl ModelClient {
let res = req_builder.send().await;
match res {
Ok(resp) if resp.status().is_success() => {
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
// spawn task to process SSE
let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
tokio::spawn(process_sse(
stream,
tx_event,
self.provider.stream_idle_timeout(),
));
tokio::spawn(process_sse(stream, tx_event));
return Ok(ResponseStream { rx_event });
}
@@ -175,7 +171,7 @@ impl ModelClient {
return Err(CodexErr::UnexpectedStatus(status, body));
}
if attempt > max_retries {
if attempt > *OPENAI_REQUEST_MAX_RETRIES {
return Err(CodexErr::RetryLimit(status));
}
@@ -192,7 +188,7 @@ impl ModelClient {
tokio::time::sleep(delay).await;
}
Err(e) => {
if attempt > max_retries {
if attempt > *OPENAI_REQUEST_MAX_RETRIES {
return Err(e.into());
}
let delay = backoff(attempt);
@@ -201,10 +197,6 @@ impl ModelClient {
}
}
}
pub fn get_provider(&self) -> ModelProviderInfo {
self.provider.clone()
}
}
#[derive(Debug, Deserialize, Serialize)]
@@ -213,7 +205,6 @@ struct SseEvent {
kind: String,
response: Option<Value>,
item: Option<Value>,
delta: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -256,16 +247,14 @@ struct ResponseCompletedOutputTokensDetails {
reasoning_tokens: u64,
}
async fn process_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
) where
async fn process_sse<S>(stream: S, tx_event: mpsc::Sender<Result<ResponseEvent>>)
where
S: Stream<Item = Result<Bytes>> + Unpin,
{
let mut stream = stream.eventsource();
// If the stream stays completely silent for an extended period treat it as disconnected.
let idle_timeout = *OPENAI_STREAM_IDLE_TIMEOUT_MS;
// The response id returned from the "complete" message.
let mut response_completed: Option<ResponseCompleted> = None;
@@ -326,7 +315,7 @@ async fn process_sse<S>(
// duplicated `output` array embedded in the `response.completed`
// payload. That produced two concrete issues:
// 1. No realtime streaming the user only saw output after the
// entire turn had finished, which broke the "typing" UX and
// entire turn had finished, which broke the typing UX and
// made longrunning turns look stalled.
// 2. Duplicate `function_call_output` items both the
// individual *and* the completed array were forwarded, which
@@ -348,22 +337,6 @@ async fn process_sse<S>(
return;
}
}
"response.output_text.delta" => {
if let Some(delta) = event.delta {
let event = ResponseEvent::OutputTextDelta(delta);
if tx_event.send(Ok(event)).await.is_err() {
return;
}
}
}
"response.reasoning_summary_text.delta" => {
if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningSummaryDelta(delta);
if tx_event.send(Ok(event)).await.is_err() {
return;
}
}
}
"response.created" => {
if event.response.is_some() {
let _ = tx_event.send(Ok(ResponseEvent::Created {})).await;
@@ -387,8 +360,10 @@ async fn process_sse<S>(
| "response.function_call_arguments.delta"
| "response.in_progress"
| "response.output_item.added"
| "response.output_text.delta"
| "response.output_text.done"
| "response.reasoning_summary_part.added"
| "response.reasoning_summary_text.delta"
| "response.reasoning_summary_text.done" => {
// Currently, we ignore these events, but we handle them
// separately to skip the logging message in the `other` case.
@@ -399,11 +374,8 @@ async fn process_sse<S>(
}
/// used in tests to stream from a text SSE file
async fn stream_from_fixture(
path: impl AsRef<Path>,
provider: ModelProviderInfo,
) -> Result<ResponseStream> {
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
async fn stream_from_fixture(path: impl AsRef<Path>) -> Result<ResponseStream> {
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
let f = std::fs::File::open(path.as_ref())?;
let lines = std::io::BufReader::new(f).lines();
@@ -416,57 +388,17 @@ async fn stream_from_fixture(
let rdr = std::io::Cursor::new(content);
let stream = ReaderStream::new(rdr).map_err(CodexErr::Io);
tokio::spawn(process_sse(
stream,
tx_event,
provider.stream_idle_timeout(),
));
tokio::spawn(process_sse(stream, tx_event));
Ok(ResponseStream { rx_event })
}
#[cfg(test)]
mod tests {
#![allow(clippy::expect_used, clippy::unwrap_used)]
use super::*;
use serde_json::json;
use tokio::sync::mpsc;
use tokio_test::io::Builder as IoBuilder;
use tokio_util::io::ReaderStream;
// ────────────────────────────
// Helpers
// ────────────────────────────
/// Runs the SSE parser on pre-chunked byte slices and returns every event
/// (including any final `Err` from a stream-closure check).
async fn collect_events(
chunks: &[&[u8]],
provider: ModelProviderInfo,
) -> Vec<Result<ResponseEvent>> {
let mut builder = IoBuilder::new();
for chunk in chunks {
builder.read(chunk);
}
let reader = builder.build();
let stream = ReaderStream::new(reader).map_err(CodexErr::Io);
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(16);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
let mut events = Vec::new();
while let Some(ev) = rx.recv().await {
events.push(ev);
}
events
}
/// Builds an in-memory SSE stream from JSON fixtures and returns only the
/// successfully parsed events (panics on internal channel errors).
async fn run_sse(
events: Vec<serde_json::Value>,
provider: ModelProviderInfo,
) -> Vec<ResponseEvent> {
async fn run_sse(events: Vec<serde_json::Value>) -> Vec<ResponseEvent> {
let mut body = String::new();
for e in events {
let kind = e
@@ -476,14 +408,12 @@ mod tests {
if e.as_object().map(|o| o.len() == 1).unwrap_or(false) {
body.push_str(&format!("event: {kind}\n\n"));
} else {
body.push_str(&format!("event: {kind}\ndata: {e}\n\n"));
body.push_str(&format!("event: {kind}\ndata: {}\n\n", e));
}
}
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(8);
let stream = ReaderStream::new(std::io::Cursor::new(body)).map_err(CodexErr::Io);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
tokio::spawn(process_sse(stream, tx));
let mut out = Vec::new();
while let Some(ev) = rx.recv().await {
out.push(ev.expect("channel closed"));
@@ -491,138 +421,9 @@ mod tests {
out
}
// ────────────────────────────
// Tests from `implement-test-for-responses-api-sse-parser`
// ────────────────────────────
#[tokio::test]
async fn parses_items_and_completed() {
let item1 = json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello"}]
}
})
.to_string();
let item2 = json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "World"}]
}
})
.to_string();
let completed = json!({
"type": "response.completed",
"response": { "id": "resp1" }
})
.to_string();
let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n");
let sse2 = format!("event: response.output_item.done\ndata: {item2}\n\n");
let sse3 = format!("event: response.completed\ndata: {completed}\n\n");
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: "https://test.com".to_string(),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
};
let events = collect_events(
&[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()],
provider,
)
.await;
assert_eq!(events.len(), 3);
matches!(
&events[0],
Ok(ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. }))
if role == "assistant"
);
matches!(
&events[1],
Ok(ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. }))
if role == "assistant"
);
match &events[2] {
Ok(ResponseEvent::Completed {
response_id,
token_usage,
}) => {
assert_eq!(response_id, "resp1");
assert!(token_usage.is_none());
}
other => panic!("unexpected third event: {other:?}"),
}
}
#[tokio::test]
async fn error_when_missing_completed() {
let item1 = json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello"}]
}
})
.to_string();
let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n");
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: "https://test.com".to_string(),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
};
let events = collect_events(&[sse1.as_bytes()], provider).await;
assert_eq!(events.len(), 2);
matches!(events[0], Ok(ResponseEvent::OutputItemDone(_)));
match &events[1] {
Err(CodexErr::Stream(msg)) => {
assert_eq!(msg, "stream closed before response.completed")
}
other => panic!("unexpected second event: {other:?}"),
}
}
// ────────────────────────────
// Table-driven test from `main`
// ────────────────────────────
/// Verifies that the adapter produces the right `ResponseEvent` for a
/// variety of incoming `type` values.
#[tokio::test]
async fn table_driven_event_kinds() {
struct TestCase {
struct Case {
name: &'static str,
event: serde_json::Value,
expect_first: fn(&ResponseEvent) -> bool,
@@ -632,9 +433,11 @@ mod tests {
fn is_created(ev: &ResponseEvent) -> bool {
matches!(ev, ResponseEvent::Created)
}
fn is_output(ev: &ResponseEvent) -> bool {
matches!(ev, ResponseEvent::OutputItemDone(_))
}
fn is_completed(ev: &ResponseEvent) -> bool {
matches!(ev, ResponseEvent::Completed { .. })
}
@@ -655,13 +458,13 @@ mod tests {
});
let cases = vec![
TestCase {
Case {
name: "created",
event: json!({"type": "response.created", "response": {}}),
expect_first: is_created,
expected_len: 2,
},
TestCase {
Case {
name: "output_item.done",
event: json!({
"type": "response.output_item.done",
@@ -676,7 +479,7 @@ mod tests {
expect_first: is_output,
expected_len: 2,
},
TestCase {
Case {
name: "unknown",
event: json!({"type": "response.new_tool_event"}),
expect_first: is_completed,
@@ -687,28 +490,9 @@ mod tests {
for case in cases {
let mut evs = vec![case.event];
evs.push(completed.clone());
let provider = ModelProviderInfo {
name: "test".to_string(),
base_url: "https://test.com".to_string(),
env_key: Some("TEST_API_KEY".to_string()),
env_key_instructions: None,
wire_api: WireApi::Responses,
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(1000),
};
let out = run_sse(evs, provider).await;
let out = run_sse(evs).await;
assert_eq!(out.len(), case.expected_len, "case {}", case.name);
assert!(
(case.expect_first)(&out[0]),
"first event mismatch in case {}",
case.name
);
assert!((case.expect_first)(&out[0]), "case {}", case.name);
}
}
}

View File

@@ -57,8 +57,6 @@ pub enum ResponseEvent {
response_id: String,
token_usage: Option<TokenUsage>,
},
OutputTextDelta(String),
ReasoningSummaryDelta(String),
}
#[derive(Debug, Serialize)]

View File

@@ -49,7 +49,9 @@ use crate::exec::ExecToolCallOutput;
use crate::exec::SandboxType;
use crate::exec::process_exec_tool_call;
use crate::exec_env::create_env;
use crate::flags::OPENAI_STREAM_MAX_RETRIES;
use crate::mcp_connection_manager::McpConnectionManager;
use crate::mcp_connection_manager::try_parse_fully_qualified_tool_name;
use crate::mcp_tool_call::handle_mcp_tool_call;
use crate::models::ContentItem;
use crate::models::FunctionCallOutputPayload;
@@ -59,9 +61,7 @@ use crate::models::ResponseInputItem;
use crate::models::ResponseItem;
use crate::models::ShellToolCallParams;
use crate::project_doc::get_user_instructions;
use crate::protocol::AgentMessageDeltaEvent;
use crate::protocol::AgentMessageEvent;
use crate::protocol::AgentReasoningDeltaEvent;
use crate::protocol::AgentReasoningEvent;
use crate::protocol::ApplyPatchApprovalRequestEvent;
use crate::protocol::AskForApproval;
@@ -103,7 +103,7 @@ impl Codex {
/// submitted to start the session.
pub async fn spawn(config: Config, ctrl_c: Arc<Notify>) -> CodexResult<(Codex, String)> {
let (tx_sub, rx_sub) = async_channel::bounded(64);
let (tx_event, rx_event) = async_channel::bounded(1600);
let (tx_event, rx_event) = async_channel::bounded(64);
let instructions = get_user_instructions(&config).await;
let configure_session = Op::ConfigureSession {
@@ -990,52 +990,6 @@ async fn run_task(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
sess.tx_event.send(event).await.ok();
}
// ---
// Helpers --------------------------------------------------------------------
//
// When a turn is interrupted before Codex can deliver tool output(s) back to
// the model, the next request can fail with a 400 from the OpenAI API:
// {"error": {"message": "No tool output found for function call call_XXXXX", ...}}
// Historically this manifested as a confusing retry loop ("stream error: 400 …")
// because we never learned about the missing `call_id` (the stream was aborted
// before we observed the `ResponseEvent::OutputItemDone` that would have let us
// record it in `pending_call_ids`).
//
// To make interruption robust we parse the error body for the offending call id
// and add it to `pending_call_ids` so the very next retry can inject a synthetic
// `FunctionCallOutput { content: "aborted" }` and satisfy the API contract.
// -----------------------------------------------------------------------------
fn extract_missing_tool_call_id(body: &str) -> Option<String> {
// Try to parse the canonical JSON error shape first.
if let Ok(v) = serde_json::from_str::<serde_json::Value>(body) {
if let Some(msg) = v
.get("error")
.and_then(|e| e.get("message"))
.and_then(|m| m.as_str())
{
if let Some(id) = extract_missing_tool_call_id_from_msg(msg) {
return Some(id);
}
}
}
// Fallback: scan the raw body.
extract_missing_tool_call_id_from_msg(body)
}
fn extract_missing_tool_call_id_from_msg(msg: &str) -> Option<String> {
const NEEDLE: &str = "No tool output found for function call";
let idx = msg.find(NEEDLE)?;
let rest = &msg[idx + NEEDLE.len()..];
// Find the beginning of the call id (typically starts with "call_").
let start = rest.find("call_")?;
let rest = &rest[start..];
// Capture valid id chars [A-Za-z0-9_-/]. Hyphen shows up in some IDs; be permissive.
let end = rest
.find(|c: char| !(c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '/'))
.unwrap_or(rest.len());
Some(rest[..end].to_string())
}
async fn run_turn(
sess: &Session,
sub_id: String,
@@ -1070,58 +1024,13 @@ async fn run_turn(
Ok(output) => return Ok(output),
Err(CodexErr::Interrupted) => return Err(CodexErr::Interrupted),
Err(CodexErr::EnvVar(var)) => return Err(CodexErr::EnvVar(var)),
Err(CodexErr::UnexpectedStatus(status, body)) => {
// Detect the specific 400 "No tool output found for function call ..." error that
// occurs when a user interrupted before Codex could answer a tool call.
if status == reqwest::StatusCode::BAD_REQUEST {
if let Some(call_id) = extract_missing_tool_call_id(&body) {
{
let mut state = sess.state.lock().unwrap();
state.pending_call_ids.insert(call_id.clone());
}
// Surface a friendlier background event so users understand the recovery.
sess
.notify_background_event(
&sub_id,
format!(
"previous turn interrupted before responding to tool {call_id}; sending aborted output and retrying…",
),
)
.await;
// Immediately retry the turn without consuming a provider stream retry budget.
continue;
}
}
// Fall through to generic retry path if we could not autorecover.
let e = CodexErr::UnexpectedStatus(status, body);
// Use the configured provider-specific stream retry budget.
let max_retries = sess.client.get_provider().stream_max_retries();
if retries < max_retries {
retries += 1;
let delay = backoff(retries);
warn!(
"stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...",
);
sess.notify_background_event(
&sub_id,
format!(
"stream error: {e}; retrying {retries}/{max_retries} in {delay:?}",
),
)
.await;
tokio::time::sleep(delay).await;
} else {
return Err(e);
}
}
Err(e) => {
// Use the configured provider-specific stream retry budget.
let max_retries = sess.client.get_provider().stream_max_retries();
if retries < max_retries {
if retries < *OPENAI_STREAM_MAX_RETRIES {
retries += 1;
let delay = backoff(retries);
warn!(
"stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...",
"stream disconnected - retrying turn ({retries}/{} in {delay:?})...",
*OPENAI_STREAM_MAX_RETRIES
);
// Surface retry information to any UI/frontend so the
@@ -1130,7 +1039,8 @@ async fn run_turn(
sess.notify_background_event(
&sub_id,
format!(
"stream error: {e}; retrying {retries}/{max_retries} in {delay:?}",
"stream error: {e}; retrying {retries}/{} in {:?}",
*OPENAI_STREAM_MAX_RETRIES, delay
),
)
.await;
@@ -1211,29 +1121,15 @@ async fn try_run_turn(
let mut stream = sess.client.clone().stream(&prompt).await?;
// Buffer all the incoming messages from the stream first, then execute them.
// If we execute a function call in the middle of handling the stream, it can time out.
let mut input = Vec::new();
while let Some(event) = stream.next().await {
input.push(event?);
}
let mut output = Vec::new();
loop {
// Poll the next item from the model stream. We must inspect *both* Ok and Err
// cases so that transient stream failures (e.g., dropped SSE connection before
// `response.completed`) bubble up and trigger the caller's retry logic.
let event = stream.next().await;
let Some(event) = event else {
// Channel closed without yielding a final Completed event or explicit error.
// Treat as a disconnected stream so the caller can retry.
return Err(CodexErr::Stream(
"stream closed before response.completed".into(),
));
};
let event = match event {
Ok(ev) => ev,
Err(e) => {
// Propagate the underlying stream error to the caller (run_turn), which
// will apply the configured `stream_max_retries` policy.
return Err(e);
}
};
for event in input {
match event {
ResponseEvent::Created => {
let mut state = sess.state.lock().unwrap();
@@ -1274,24 +1170,11 @@ async fn try_run_turn(
let mut state = sess.state.lock().unwrap();
state.previous_response_id = Some(response_id);
return Ok(output);
}
ResponseEvent::OutputTextDelta(delta) => {
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }),
};
sess.tx_event.send(event).await.ok();
}
ResponseEvent::ReasoningSummaryDelta(delta) => {
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }),
};
sess.tx_event.send(event).await.ok();
break;
}
}
}
Ok(output)
}
async fn handle_response_item(
@@ -1400,7 +1283,7 @@ async fn handle_function_call(
handle_container_exec_with_params(params, sess, sub_id, call_id).await
}
_ => {
match sess.mcp_connection_manager.parse_tool_name(&name) {
match try_parse_fully_qualified_tool_name(&name) {
Some((server, tool_name)) => {
// TODO(mbolin): Determine appropriate timeout for tool call.
let timeout = None;

View File

@@ -682,9 +682,6 @@ name = "OpenAI using Chat Completions"
base_url = "https://api.openai.com/v1"
env_key = "OPENAI_API_KEY"
wire_api = "chat"
request_max_retries = 4 # retry failed HTTP requests
stream_max_retries = 10 # retry dropped SSE streams
stream_idle_timeout_ms = 300000 # 5m idle timeout
[profiles.o3]
model = "o3"
@@ -725,9 +722,6 @@ disable_response_storage = true
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: Some(4),
stream_max_retries: Some(10),
stream_idle_timeout_ms: Some(300_000),
};
let model_provider_map = {
let mut model_provider_map = built_in_model_providers();

View File

@@ -11,6 +11,14 @@ env_flags! {
pub OPENAI_TIMEOUT_MS: Duration = Duration::from_millis(300_000), |value| {
value.parse().map(Duration::from_millis)
};
pub OPENAI_REQUEST_MAX_RETRIES: u64 = 4;
pub OPENAI_STREAM_MAX_RETRIES: u64 = 10;
// We generally don't want to disconnect; this updates the timeout to be five minutes
// which matches the upstream typescript codex impl.
pub OPENAI_STREAM_IDLE_TIMEOUT_MS: Duration = Duration::from_millis(300_000), |value| {
value.parse().map(Duration::from_millis)
};
/// Fixture path for offline tests (see client.rs).
pub CODEX_RS_SSE_FIXTURE: Option<&str> = None;

View File

@@ -7,7 +7,6 @@
//! `"<server><MCP_TOOL_NAME_DELIMITER><tool>"` as the key.
use std::collections::HashMap;
use std::collections::HashSet;
use std::time::Duration;
use anyhow::Context;
@@ -17,12 +16,8 @@ use codex_mcp_client::McpClient;
use mcp_types::ClientCapabilities;
use mcp_types::Implementation;
use mcp_types::Tool;
use sha1::Digest;
use sha1::Sha1;
use tokio::task::JoinSet;
use tracing::info;
use tracing::warn;
use crate::config_types::McpServerConfig;
@@ -31,8 +26,7 @@ use crate::config_types::McpServerConfig;
///
/// OpenAI requires tool names to conform to `^[a-zA-Z0-9_-]+$`, so we must
/// choose a delimiter from this character set.
const MCP_TOOL_NAME_DELIMITER: &str = "__";
const MAX_TOOL_NAME_LENGTH: usize = 64;
const MCP_TOOL_NAME_DELIMITER: &str = "__OAI_CODEX_MCP__";
/// Timeout for the `tools/list` request.
const LIST_TOOLS_TIMEOUT: Duration = Duration::from_secs(10);
@@ -41,42 +35,16 @@ const LIST_TOOLS_TIMEOUT: Duration = Duration::from_secs(10);
/// spawned successfully.
pub type ClientStartErrors = HashMap<String, anyhow::Error>;
fn qualify_tools(tools: Vec<ToolInfo>) -> HashMap<String, ToolInfo> {
let mut used_names = HashSet::new();
let mut qualified_tools = HashMap::new();
for tool in tools {
let mut qualified_name = format!(
"{}{}{}",
tool.server_name, MCP_TOOL_NAME_DELIMITER, tool.tool_name
);
if qualified_name.len() > MAX_TOOL_NAME_LENGTH {
let mut hasher = Sha1::new();
hasher.update(qualified_name.as_bytes());
let sha1 = hasher.finalize();
let sha1_str = format!("{sha1:x}");
// Truncate to make room for the hash suffix
let prefix_len = MAX_TOOL_NAME_LENGTH - sha1_str.len();
qualified_name = format!("{}{}", &qualified_name[..prefix_len], sha1_str);
}
if used_names.contains(&qualified_name) {
warn!("skipping duplicated tool {}", qualified_name);
continue;
}
used_names.insert(qualified_name.clone());
qualified_tools.insert(qualified_name, tool);
}
qualified_tools
fn fully_qualified_tool_name(server: &str, tool: &str) -> String {
format!("{server}{MCP_TOOL_NAME_DELIMITER}{tool}")
}
struct ToolInfo {
server_name: String,
tool_name: String,
tool: Tool,
pub(crate) fn try_parse_fully_qualified_tool_name(fq_name: &str) -> Option<(String, String)> {
let (server, tool) = fq_name.split_once(MCP_TOOL_NAME_DELIMITER)?;
if server.is_empty() || tool.is_empty() {
return None;
}
Some((server.to_string(), tool.to_string()))
}
/// A thin wrapper around a set of running [`McpClient`] instances.
@@ -89,7 +57,7 @@ pub(crate) struct McpConnectionManager {
clients: HashMap<String, std::sync::Arc<McpClient>>,
/// Fully qualified tool name -> tool instance.
tools: HashMap<String, ToolInfo>,
tools: HashMap<String, Tool>,
}
impl McpConnectionManager {
@@ -111,19 +79,9 @@ impl McpConnectionManager {
// Launch all configured servers concurrently.
let mut join_set = JoinSet::new();
let mut errors = ClientStartErrors::new();
for (server_name, cfg) in mcp_servers {
// Validate server name before spawning
if !is_valid_mcp_server_name(&server_name) {
let error = anyhow::anyhow!(
"invalid server name '{}': must match pattern ^[a-zA-Z0-9_-]+$",
server_name
);
errors.insert(server_name, error);
continue;
}
// TODO: Verify server name: require `^[a-zA-Z0-9_-]+$`?
join_set.spawn(async move {
let McpServerConfig { command, args, env } = cfg;
let client_res = McpClient::new_stdio_client(command, args, env).await;
@@ -159,6 +117,7 @@ impl McpConnectionManager {
let mut clients: HashMap<String, std::sync::Arc<McpClient>> =
HashMap::with_capacity(join_set.len());
let mut errors = ClientStartErrors::new();
while let Some(res) = join_set.join_next().await {
let (server_name, client_res) = res?; // JoinError propagation
@@ -173,9 +132,7 @@ impl McpConnectionManager {
}
}
let all_tools = list_all_tools(&clients).await?;
let tools = qualify_tools(all_tools);
let tools = list_all_tools(&clients).await?;
Ok((Self { clients, tools }, errors))
}
@@ -183,10 +140,7 @@ impl McpConnectionManager {
/// Returns a single map that contains **all** tools. Each key is the
/// fully-qualified name for the tool.
pub fn list_all_tools(&self) -> HashMap<String, Tool> {
self.tools
.iter()
.map(|(name, tool)| (name.clone(), tool.tool.clone()))
.collect()
self.tools.clone()
}
/// Invoke the tool indicated by the (server, tool) pair.
@@ -208,19 +162,13 @@ impl McpConnectionManager {
.await
.with_context(|| format!("tool call failed for `{server}/{tool}`"))
}
pub fn parse_tool_name(&self, tool_name: &str) -> Option<(String, String)> {
self.tools
.get(tool_name)
.map(|tool| (tool.server_name.clone(), tool.tool_name.clone()))
}
}
/// Query every server for its available tools and return a single map that
/// contains **all** tools. Each key is the fully-qualified name for the tool.
async fn list_all_tools(
pub async fn list_all_tools(
clients: &HashMap<String, std::sync::Arc<McpClient>>,
) -> Result<Vec<ToolInfo>> {
) -> Result<HashMap<String, Tool>> {
let mut join_set = JoinSet::new();
// Spawn one task per server so we can query them concurrently. This
@@ -237,19 +185,18 @@ async fn list_all_tools(
});
}
let mut aggregated: Vec<ToolInfo> = Vec::with_capacity(join_set.len());
let mut aggregated: HashMap<String, Tool> = HashMap::with_capacity(join_set.len());
while let Some(join_res) = join_set.join_next().await {
let (server_name, list_result) = join_res?;
let list_result = list_result?;
for tool in list_result.tools {
let tool_info = ToolInfo {
server_name: server_name.clone(),
tool_name: tool.name.clone(),
tool,
};
aggregated.push(tool_info);
// TODO(mbolin): escape tool names that contain invalid characters.
let fq_name = fully_qualified_tool_name(&server_name, &tool.name);
if aggregated.insert(fq_name.clone(), tool).is_some() {
panic!("tool name collision for '{fq_name}': suspicious");
}
}
}
@@ -261,97 +208,3 @@ async fn list_all_tools(
Ok(aggregated)
}
fn is_valid_mcp_server_name(server_name: &str) -> bool {
!server_name.is_empty()
&& server_name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use super::*;
use mcp_types::ToolInputSchema;
fn create_test_tool(server_name: &str, tool_name: &str) -> ToolInfo {
ToolInfo {
server_name: server_name.to_string(),
tool_name: tool_name.to_string(),
tool: Tool {
annotations: None,
description: Some(format!("Test tool: {tool_name}")),
input_schema: ToolInputSchema {
properties: None,
required: None,
r#type: "object".to_string(),
},
name: tool_name.to_string(),
},
}
}
#[test]
fn test_qualify_tools_short_non_duplicated_names() {
let tools = vec![
create_test_tool("server1", "tool1"),
create_test_tool("server1", "tool2"),
];
let qualified_tools = qualify_tools(tools);
assert_eq!(qualified_tools.len(), 2);
assert!(qualified_tools.contains_key("server1__tool1"));
assert!(qualified_tools.contains_key("server1__tool2"));
}
#[test]
fn test_qualify_tools_duplicated_names_skipped() {
let tools = vec![
create_test_tool("server1", "duplicate_tool"),
create_test_tool("server1", "duplicate_tool"),
];
let qualified_tools = qualify_tools(tools);
// Only the first tool should remain, the second is skipped
assert_eq!(qualified_tools.len(), 1);
assert!(qualified_tools.contains_key("server1__duplicate_tool"));
}
#[test]
fn test_qualify_tools_long_names_same_server() {
let server_name = "my_server";
let tools = vec![
create_test_tool(
server_name,
"extremely_lengthy_function_name_that_absolutely_surpasses_all_reasonable_limits",
),
create_test_tool(
server_name,
"yet_another_extremely_lengthy_function_name_that_absolutely_surpasses_all_reasonable_limits",
),
];
let qualified_tools = qualify_tools(tools);
assert_eq!(qualified_tools.len(), 2);
let mut keys: Vec<_> = qualified_tools.keys().cloned().collect();
keys.sort();
assert_eq!(keys[0].len(), 64);
assert_eq!(
keys[0],
"my_server__extremely_lena02e507efc5a9de88637e436690364fd4219e4ef"
);
assert_eq!(keys[1].len(), 64);
assert_eq!(
keys[1],
"my_server__yet_another_e1c3987bd9c50b826cbe1687966f79f0c602d19ca"
);
}
}

View File

@@ -9,7 +9,6 @@ use serde::Deserialize;
use serde::Serialize;
use std::collections::HashMap;
use std::env::VarError;
use std::time::Duration;
use crate::error::EnvVarError;
use crate::openai_api_key::get_openai_api_key;
@@ -17,9 +16,6 @@ use crate::openai_api_key::get_openai_api_key;
/// Value for the `OpenAI-Originator` header that is sent with requests to
/// OpenAI.
const OPENAI_ORIGINATOR_HEADER: &str = "codex_cli_rs";
const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000;
const DEFAULT_STREAM_MAX_RETRIES: u64 = 10;
const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4;
/// Wire protocol that the provider speaks. Most third-party services only
/// implement the classic OpenAI Chat Completions JSON schema, whereas OpenAI
@@ -30,7 +26,7 @@ const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum WireApi {
/// The experimental "Responses" API exposed by OpenAI at `/v1/responses`.
/// The experimental Responses API exposed by OpenAI at `/v1/responses`.
Responses,
/// Regular Chat Completions compatible with `/v1/chat/completions`.
@@ -68,16 +64,6 @@ pub struct ModelProviderInfo {
/// value should be used. If the environment variable is not set, or the
/// value is empty, the header will not be included in the request.
pub env_http_headers: Option<HashMap<String, String>>,
/// Maximum number of times to retry a failed HTTP request to this provider.
pub request_max_retries: Option<u64>,
/// Number of times to retry reconnecting a dropped streaming response before failing.
pub stream_max_retries: Option<u64>,
/// Idle timeout (in milliseconds) to wait for activity on a streaming response before treating
/// the connection as lost.
pub stream_idle_timeout_ms: Option<u64>,
}
impl ModelProviderInfo {
@@ -175,25 +161,6 @@ impl ModelProviderInfo {
None => Ok(None),
}
}
/// Effective maximum number of request retries for this provider.
pub fn request_max_retries(&self) -> u64 {
self.request_max_retries
.unwrap_or(DEFAULT_REQUEST_MAX_RETRIES)
}
/// Effective maximum number of stream reconnection attempts for this provider.
pub fn stream_max_retries(&self) -> u64 {
self.stream_max_retries
.unwrap_or(DEFAULT_STREAM_MAX_RETRIES)
}
/// Effective idle timeout for streaming responses.
pub fn stream_idle_timeout(&self) -> Duration {
self.stream_idle_timeout_ms
.map(Duration::from_millis)
.unwrap_or(Duration::from_millis(DEFAULT_STREAM_IDLE_TIMEOUT_MS))
}
}
/// Built-in default provider list.
@@ -238,10 +205,6 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
.into_iter()
.collect(),
),
// Use global defaults for retry/timeout unless overridden in config.toml.
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
},
),
]
@@ -271,9 +234,6 @@ base_url = "http://localhost:11434/v1"
query_params: None,
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();
@@ -299,9 +259,6 @@ query_params = { api-version = "2025-04-01-preview" }
}),
http_headers: None,
env_http_headers: None,
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();
@@ -330,9 +287,6 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" }
env_http_headers: Some(maplit::hashmap! {
"X-Example-Env-Header".to_string() => "EXAMPLE_ENV_VAR".to_string(),
}),
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
};
let provider: ModelProviderInfo = toml::from_str(azure_provider_toml).unwrap();

View File

@@ -282,15 +282,9 @@ pub enum EventMsg {
/// Agent text output message
AgentMessage(AgentMessageEvent),
/// Agent text output delta message
AgentMessageDelta(AgentMessageDeltaEvent),
/// Reasoning event from agent.
AgentReasoning(AgentReasoningEvent),
/// Agent reasoning delta event from agent.
AgentReasoningDelta(AgentReasoningDeltaEvent),
/// Ack the client's configure message.
SessionConfigured(SessionConfiguredEvent),
@@ -346,21 +340,11 @@ pub struct AgentMessageEvent {
pub message: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct AgentMessageDeltaEvent {
pub delta: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct AgentReasoningEvent {
pub text: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct AgentReasoningDeltaEvent {
pub delta: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct McpToolCallBeginEvent {
/// Identifier so this can be paired with the McpToolCallEnd event.

View File

@@ -153,16 +153,14 @@ struct LogFileInfo {
}
fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFileInfo> {
// Resolve ~/.codex/sessions/YYYY/MM/DD and create it if missing.
let timestamp = OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
// Resolve ~/.codex/sessions and create it if missing.
let mut dir = config.codex_home.clone();
dir.push(SESSIONS_SUBDIR);
dir.push(timestamp.year().to_string());
dir.push(format!("{:02}", u8::from(timestamp.month())));
dir.push(format!("{:02}", timestamp.day()));
fs::create_dir_all(&dir)?;
let timestamp = OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
// Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for
// compatibility with filesystems that do not allow colons in filenames.
let format: &[FormatItem] =

View File

@@ -1,8 +0,0 @@
event: response.created
data: {"type":"response.created","response":{"id":"resp1"}}
event: response.output_item.done
data: {"type":"response.output_item.done","item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"fixture hello"}]}}
event: response.completed
data: {"type":"response.completed","response":{"id":"resp1","output":[]}}

View File

@@ -1,275 +0,0 @@
#![expect(clippy::unwrap_used)]
use assert_cmd::Command as AssertCommand;
use codex_core::exec::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use serde_json::Value;
use std::time::Duration;
use std::time::Instant;
use tempfile::TempDir;
use uuid::Uuid;
use walkdir::WalkDir;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
/// Tests streaming chat completions through the CLI using a mock server.
/// This test:
/// 1. Sets up a mock server that simulates OpenAI's chat completions API
/// 2. Configures codex to use this mock server via a custom provider
/// 3. Sends a simple "hello?" prompt and verifies the streamed response
/// 4. Ensures the response is received exactly once and contains "hi"
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn chat_mode_stream_cli() {
if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let server = MockServer::start().await;
let sse = concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{}}]}\n\n",
"data: [DONE]\n\n"
);
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse, "text/event-stream"),
)
.expect(1)
.mount(&server)
.await;
let home = TempDir::new().unwrap();
let provider_override = format!(
"model_providers.mock={{ name = \"mock\", base_url = \"{}/v1\", env_key = \"PATH\", wire_api = \"chat\" }}",
server.uri()
);
let mut cmd = AssertCommand::new("cargo");
cmd.arg("run")
.arg("-p")
.arg("codex-cli")
.arg("--quiet")
.arg("--")
.arg("exec")
.arg("--skip-git-repo-check")
.arg("-c")
.arg(&provider_override)
.arg("-c")
.arg("model_provider=\"mock\"")
.arg("-C")
.arg(env!("CARGO_MANIFEST_DIR"))
.arg("hello?");
cmd.env("CODEX_HOME", home.path())
.env("OPENAI_API_KEY", "dummy")
.env("OPENAI_BASE_URL", format!("{}/v1", server.uri()));
let output = cmd.output().unwrap();
println!("Status: {}", output.status);
println!("Stdout:\n{}", String::from_utf8_lossy(&output.stdout));
println!("Stderr:\n{}", String::from_utf8_lossy(&output.stderr));
assert!(output.status.success());
let stdout = String::from_utf8_lossy(&output.stdout);
let hi_lines = stdout.lines().filter(|line| line.trim() == "hi").count();
assert_eq!(hi_lines, 1, "Expected exactly one line with 'hi'");
server.verify().await;
}
/// Tests streaming responses through the CLI using a local SSE fixture file.
/// This test:
/// 1. Uses a pre-recorded SSE response fixture instead of a live server
/// 2. Configures codex to read from this fixture via CODEX_RS_SSE_FIXTURE env var
/// 3. Sends a "hello?" prompt and verifies the response
/// 4. Ensures the fixture content is correctly streamed through the CLI
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responses_api_stream_cli() {
if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let fixture =
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/cli_responses_fixture.sse");
let home = TempDir::new().unwrap();
let mut cmd = AssertCommand::new("cargo");
cmd.arg("run")
.arg("-p")
.arg("codex-cli")
.arg("--quiet")
.arg("--")
.arg("exec")
.arg("--skip-git-repo-check")
.arg("-C")
.arg(env!("CARGO_MANIFEST_DIR"))
.arg("hello?");
cmd.env("CODEX_HOME", home.path())
.env("OPENAI_API_KEY", "dummy")
.env("CODEX_RS_SSE_FIXTURE", fixture)
.env("OPENAI_BASE_URL", "http://unused.local");
let output = cmd.output().unwrap();
assert!(output.status.success());
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(stdout.contains("fixture hello"));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn integration_creates_and_checks_session_file() {
// Honor sandbox network restrictions for CI parity with the other tests.
if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
// 1. Temp home so we read/write isolated session files.
let home = TempDir::new().unwrap();
// 2. Unique marker we'll look for in the session log.
let marker = format!("integration-test-{}", Uuid::new_v4());
let prompt = format!("echo {marker}");
// 3. Use the same offline SSE fixture as responses_api_stream_cli so the test is hermetic.
let fixture =
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/cli_responses_fixture.sse");
// 4. Run the codex CLI through cargo (ensures the right bin is built) and invoke `exec`,
// which is what records a session.
let mut cmd = AssertCommand::new("cargo");
cmd.arg("run")
.arg("-p")
.arg("codex-cli")
.arg("--quiet")
.arg("--")
.arg("exec")
.arg("--skip-git-repo-check")
.arg("-C")
.arg(env!("CARGO_MANIFEST_DIR"))
.arg(&prompt);
cmd.env("CODEX_HOME", home.path())
.env("OPENAI_API_KEY", "dummy")
.env("CODEX_RS_SSE_FIXTURE", &fixture)
// Required for CLI arg parsing even though fixture short-circuits network usage.
.env("OPENAI_BASE_URL", "http://unused.local");
let output = cmd.output().unwrap();
assert!(
output.status.success(),
"codex-cli exec failed: {}",
String::from_utf8_lossy(&output.stderr)
);
// 5. Sessions are written asynchronously; wait briefly for the directory to appear.
let sessions_dir = home.path().join("sessions");
let start = Instant::now();
while !sessions_dir.exists() && start.elapsed() < Duration::from_secs(3) {
std::thread::sleep(Duration::from_millis(50));
}
// 6. Scan all session files and find the one that contains our marker.
let mut matching_files = vec![];
for entry in WalkDir::new(&sessions_dir) {
let entry = entry.unwrap();
if entry.file_type().is_file() && entry.file_name().to_string_lossy().ends_with(".jsonl") {
let path = entry.path();
let content = std::fs::read_to_string(path).unwrap();
let mut lines = content.lines();
// Skip SessionMeta (first line)
let _ = lines.next();
for line in lines {
let item: Value = serde_json::from_str(line).unwrap();
if let Some("message") = item.get("type").and_then(|t| t.as_str()) {
if let Some(content) = item.get("content") {
if content.to_string().contains(&marker) {
matching_files.push(path.to_owned());
break;
}
}
}
}
}
}
assert_eq!(
matching_files.len(),
1,
"Expected exactly one session file containing the marker, found {}",
matching_files.len()
);
let path = &matching_files[0];
// 7. Verify directory structure: sessions/YYYY/MM/DD/filename.jsonl
let rel = match path.strip_prefix(&sessions_dir) {
Ok(r) => r,
Err(_) => panic!("session file should live under sessions/"),
};
let comps: Vec<String> = rel
.components()
.map(|c| c.as_os_str().to_string_lossy().into_owned())
.collect();
assert_eq!(
comps.len(),
4,
"Expected sessions/YYYY/MM/DD/<file>, got {rel:?}"
);
let year = &comps[0];
let month = &comps[1];
let day = &comps[2];
assert!(
year.len() == 4 && year.chars().all(|c| c.is_ascii_digit()),
"Year dir not 4-digit numeric: {year}"
);
assert!(
month.len() == 2 && month.chars().all(|c| c.is_ascii_digit()),
"Month dir not zero-padded 2-digit numeric: {month}"
);
assert!(
day.len() == 2 && day.chars().all(|c| c.is_ascii_digit()),
"Day dir not zero-padded 2-digit numeric: {day}"
);
// Range checks (best-effort; won't fail on leading zeros)
if let Ok(m) = month.parse::<u8>() {
assert!((1..=12).contains(&m), "Month out of range: {m}");
}
if let Ok(d) = day.parse::<u8>() {
assert!((1..=31).contains(&d), "Day out of range: {d}");
}
// 8. Parse SessionMeta line and basic sanity checks.
let content = std::fs::read_to_string(path).unwrap();
let mut lines = content.lines();
let meta: Value = serde_json::from_str(lines.next().unwrap()).unwrap();
assert!(meta.get("id").is_some(), "SessionMeta missing id");
assert!(
meta.get("timestamp").is_some(),
"SessionMeta missing timestamp"
);
// 9. Confirm at least one message contains the marker.
let mut found_message = false;
for line in lines {
let item: Value = serde_json::from_str(line).unwrap();
if item.get("type").map(|t| t == "message").unwrap_or(false) {
if let Some(content) = item.get("content") {
if content.to_string().contains(&marker) {
found_message = true;
break;
}
}
}
}
assert!(
found_message,
"No message found in session file containing the marker"
);
}

View File

@@ -45,10 +45,22 @@ async fn spawn_codex() -> Result<Codex, CodexErr> {
"OPENAI_API_KEY must be set for live tests"
);
// Environment tweaks to keep the tests snappy and inexpensive while still
// exercising retry/robustness logic.
//
// NOTE: Starting with the 2024 edition `std::env::set_var` is `unsafe`
// because changing the process environment races with any other threads
// that might be performing environment look-ups at the same time.
// Restrict the unsafety to this tiny block that happens at the very
// beginning of the test, before we spawn any background tasks that could
// observe the environment.
unsafe {
std::env::set_var("OPENAI_REQUEST_MAX_RETRIES", "2");
std::env::set_var("OPENAI_STREAM_MAX_RETRIES", "2");
}
let codex_home = TempDir::new().unwrap();
let mut config = load_default_config_for_test(&codex_home);
config.model_provider.request_max_retries = Some(2);
config.model_provider.stream_max_retries = Some(2);
let config = load_default_config_for_test(&codex_home);
let (agent, _init_id) = Codex::spawn(config, std::sync::Arc::new(Notify::new())).await?;
Ok(agent)
@@ -67,7 +79,7 @@ async fn live_streaming_and_prev_id_reset() {
let codex = spawn_codex().await.unwrap();
// ---------- Task 1 ----------
// ---------- Task 1 ----------
codex
.submit(Op::UserInput {
items: vec![InputItem::Text {
@@ -101,7 +113,7 @@ async fn live_streaming_and_prev_id_reset() {
"Agent did not stream any AgentMessage before TaskComplete"
);
// ---------- Task 2 (same session) ----------
// ---------- Task 2 (same session) ----------
codex
.submit(Op::UserInput {
items: vec![InputItem::Text {

View File

@@ -88,8 +88,13 @@ async fn keeps_previous_response_id_between_tasks() {
.mount(&server)
.await;
// Configure retry behavior explicitly to avoid mutating process-wide
// environment variables.
// Environment
// Update environment `set_var` is `unsafe` starting with the 2024
// edition so we group the calls into a single `unsafe { … }` block.
unsafe {
std::env::set_var("OPENAI_REQUEST_MAX_RETRIES", "0");
std::env::set_var("OPENAI_STREAM_MAX_RETRIES", "0");
}
let model_provider = ModelProviderInfo {
name: "openai".into(),
base_url: format!("{}/v1", server.uri()),
@@ -102,10 +107,6 @@ async fn keeps_previous_response_id_between_tasks() {
query_params: None,
http_headers: None,
env_http_headers: None,
// disable retries so we don't get duplicate calls in this test
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: None,
};
// Init session

View File

@@ -70,8 +70,19 @@ async fn retries_on_early_close() {
.mount(&server)
.await;
// Configure retry behavior explicitly to avoid mutating process-wide
// environment variables.
// Environment
//
// As of Rust 2024 `std::env::set_var` has been made `unsafe` because
// mutating the process environment is inherently racy when other threads
// are running. We therefore have to wrap every call in an explicit
// `unsafe` block. These are limited to the test-setup section so the
// scope is very small and clearly delineated.
unsafe {
std::env::set_var("OPENAI_REQUEST_MAX_RETRIES", "0");
std::env::set_var("OPENAI_STREAM_MAX_RETRIES", "1");
std::env::set_var("OPENAI_STREAM_IDLE_TIMEOUT_MS", "2000");
}
let model_provider = ModelProviderInfo {
name: "openai".into(),
@@ -85,10 +96,6 @@ async fn retries_on_early_close() {
query_params: None,
http_headers: None,
env_http_headers: None,
// exercise retry path: first attempt yields incomplete stream, so allow 1 retry
request_max_retries: Some(0),
stream_max_retries: Some(1),
stream_idle_timeout_ms: Some(2000),
};
let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());

View File

@@ -30,7 +30,6 @@ pub fn load_default_config_for_test(codex_home: &TempDir) -> Config {
/// with only a `type` field results in an event with no `data:` section. This
/// makes it trivial to extend the fixtures as OpenAI adds new event kinds or
/// fields.
#[allow(dead_code)]
pub fn load_sse_fixture(path: impl AsRef<std::path::Path>) -> String {
let events: Vec<serde_json::Value> =
serde_json::from_reader(std::fs::File::open(path).expect("read fixture"))
@@ -45,17 +44,15 @@ pub fn load_sse_fixture(path: impl AsRef<std::path::Path>) -> String {
if e.as_object().map(|o| o.len() == 1).unwrap_or(false) {
format!("event: {kind}\n\n")
} else {
format!("event: {kind}\ndata: {e}\n\n")
format!("event: {kind}\ndata: {}\n\n", e)
}
})
.collect()
}
/// Same as [`load_sse_fixture`], but replaces the placeholder `__ID__` in the
/// fixture template with the supplied identifier before parsing. This lets a
/// single JSON template be reused by multiple tests that each need a unique
/// `response_id`.
#[allow(dead_code)]
/// Like [`load_sse_fixture`] but substitutes the placeholder `__ID__` with the
/// provided identifier before parsing. Useful when the test needs unique
/// `response_id` values.
pub fn load_sse_fixture_with_id(path: impl AsRef<std::path::Path>, id: &str) -> String {
let raw = std::fs::read_to_string(path).expect("read fixture template");
let replaced = raw.replace("__ID__", id);
@@ -71,7 +68,7 @@ pub fn load_sse_fixture_with_id(path: impl AsRef<std::path::Path>, id: &str) ->
if e.as_object().map(|o| o.len() == 1).unwrap_or(false) {
format!("event: {kind}\n\n")
} else {
format!("event: {kind}\ndata: {e}\n\n")
format!("event: {kind}\ndata: {}\n\n", e)
}
})
.collect()

View File

@@ -51,10 +51,6 @@ pub struct Cli {
#[arg(long = "color", value_enum, default_value_t = Color::Auto)]
pub color: Color,
/// Print events to stdout as JSONL.
#[arg(long = "json", default_value_t = false)]
pub json: bool,
/// Specifies file where the last message from the agent should be written.
#[arg(long = "output-last-message")]
pub last_message_file: Option<PathBuf>,

View File

@@ -1,37 +1,492 @@
use codex_common::elapsed::format_elapsed;
use codex_common::summarize_sandbox_policy;
use codex_core::WireApi;
use codex_core::config::Config;
use codex_core::model_supports_reasoning_summaries;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::BackgroundEventEvent;
use codex_core::protocol::ErrorEvent;
use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecCommandBeginEvent;
use codex_core::protocol::ExecCommandEndEvent;
use codex_core::protocol::FileChange;
use codex_core::protocol::McpToolCallBeginEvent;
use codex_core::protocol::McpToolCallEndEvent;
use codex_core::protocol::PatchApplyBeginEvent;
use codex_core::protocol::PatchApplyEndEvent;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::protocol::TokenUsage;
use owo_colors::OwoColorize;
use owo_colors::Style;
use shlex::try_join;
use std::collections::HashMap;
use std::time::Instant;
pub(crate) trait EventProcessor {
/// Print summary of effective configuration and user prompt.
fn print_config_summary(&mut self, config: &Config, prompt: &str);
/// This should be configurable. When used in CI, users may not want to impose
/// a limit so they can see the full transcript.
const MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL: usize = 20;
/// Handle a single event emitted by the agent.
fn process_event(&mut self, event: Event);
pub(crate) struct EventProcessor {
call_id_to_command: HashMap<String, ExecCommandBegin>,
call_id_to_patch: HashMap<String, PatchApplyBegin>,
/// Tracks in-flight MCP tool calls so we can calculate duration and print
/// a concise summary when the corresponding `McpToolCallEnd` event is
/// received.
call_id_to_tool_call: HashMap<String, McpToolCallBegin>,
// To ensure that --color=never is respected, ANSI escapes _must_ be added
// using .style() with one of these fields. If you need a new style, add a
// new field here.
bold: Style,
italic: Style,
dimmed: Style,
magenta: Style,
red: Style,
green: Style,
cyan: Style,
/// Whether to include `AgentReasoning` events in the output.
show_agent_reasoning: bool,
}
pub(crate) fn create_config_summary_entries(config: &Config) -> Vec<(&'static str, String)> {
let mut entries = vec![
("workdir", config.cwd.display().to_string()),
("model", config.model.clone()),
("provider", config.model_provider_id.clone()),
("approval", format!("{:?}", config.approval_policy)),
("sandbox", summarize_sandbox_policy(&config.sandbox_policy)),
];
if config.model_provider.wire_api == WireApi::Responses
&& model_supports_reasoning_summaries(config)
{
entries.push((
"reasoning effort",
config.model_reasoning_effort.to_string(),
));
entries.push((
"reasoning summaries",
config.model_reasoning_summary.to_string(),
));
impl EventProcessor {
pub(crate) fn create_with_ansi(with_ansi: bool, show_agent_reasoning: bool) -> Self {
let call_id_to_command = HashMap::new();
let call_id_to_patch = HashMap::new();
let call_id_to_tool_call = HashMap::new();
if with_ansi {
Self {
call_id_to_command,
call_id_to_patch,
bold: Style::new().bold(),
italic: Style::new().italic(),
dimmed: Style::new().dimmed(),
magenta: Style::new().magenta(),
red: Style::new().red(),
green: Style::new().green(),
cyan: Style::new().cyan(),
call_id_to_tool_call,
show_agent_reasoning,
}
} else {
Self {
call_id_to_command,
call_id_to_patch,
bold: Style::new(),
italic: Style::new(),
dimmed: Style::new(),
magenta: Style::new(),
red: Style::new(),
green: Style::new(),
cyan: Style::new(),
call_id_to_tool_call,
show_agent_reasoning,
}
}
}
}
struct ExecCommandBegin {
command: Vec<String>,
start_time: Instant,
}
/// Metadata captured when an `McpToolCallBegin` event is received.
struct McpToolCallBegin {
/// Formatted invocation string, e.g. `server.tool({"city":"sf"})`.
invocation: String,
/// Timestamp when the call started so we can compute duration later.
start_time: Instant,
}
struct PatchApplyBegin {
start_time: Instant,
auto_approved: bool,
}
// Timestamped println helper. The timestamp is styled with self.dimmed.
#[macro_export]
macro_rules! ts_println {
($self:ident, $($arg:tt)*) => {{
let now = chrono::Utc::now();
let formatted = now.format("[%Y-%m-%dT%H:%M:%S]");
print!("{} ", formatted.style($self.dimmed));
println!($($arg)*);
}};
}
impl EventProcessor {
/// Print a concise summary of the effective configuration that will be used
/// for the session. This mirrors the information shown in the TUI welcome
/// screen.
pub(crate) fn print_config_summary(&mut self, config: &Config, prompt: &str) {
const VERSION: &str = env!("CARGO_PKG_VERSION");
ts_println!(
self,
"OpenAI Codex v{} (research preview)\n--------",
VERSION
);
let mut entries = vec![
("workdir", config.cwd.display().to_string()),
("model", config.model.clone()),
("provider", config.model_provider_id.clone()),
("approval", format!("{:?}", config.approval_policy)),
("sandbox", summarize_sandbox_policy(&config.sandbox_policy)),
];
if config.model_provider.wire_api == WireApi::Responses
&& model_supports_reasoning_summaries(config)
{
entries.push((
"reasoning effort",
config.model_reasoning_effort.to_string(),
));
entries.push((
"reasoning summaries",
config.model_reasoning_summary.to_string(),
));
}
for (key, value) in entries {
println!("{} {}", format!("{key}:").style(self.bold), value);
}
println!("--------");
// Echo the prompt that will be sent to the agent so it is visible in the
// transcript/logs before any events come in. Note the prompt may have been
// read from stdin, so it may not be visible in the terminal otherwise.
ts_println!(
self,
"{}\n{}",
"User instructions:".style(self.bold).style(self.cyan),
prompt
);
}
entries
pub(crate) fn process_event(&mut self, event: Event) {
let Event { id: _, msg } = event;
match msg {
EventMsg::Error(ErrorEvent { message }) => {
let prefix = "ERROR:".style(self.red);
ts_println!(self, "{prefix} {message}");
}
EventMsg::BackgroundEvent(BackgroundEventEvent { message }) => {
ts_println!(self, "{}", message.style(self.dimmed));
}
EventMsg::TaskStarted | EventMsg::TaskComplete(_) => {
// Ignore.
}
EventMsg::TokenCount(TokenUsage { total_tokens, .. }) => {
ts_println!(self, "tokens used: {total_tokens}");
}
EventMsg::AgentMessage(AgentMessageEvent { message }) => {
ts_println!(
self,
"{}\n{message}",
"codex".style(self.bold).style(self.magenta)
);
}
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
call_id,
command,
cwd,
}) => {
self.call_id_to_command.insert(
call_id.clone(),
ExecCommandBegin {
command: command.clone(),
start_time: Instant::now(),
},
);
ts_println!(
self,
"{} {} in {}",
"exec".style(self.magenta),
escape_command(&command).style(self.bold),
cwd.to_string_lossy(),
);
}
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
call_id,
stdout,
stderr,
exit_code,
}) => {
let exec_command = self.call_id_to_command.remove(&call_id);
let (duration, call) = if let Some(ExecCommandBegin {
command,
start_time,
}) = exec_command
{
(
format!(" in {}", format_elapsed(start_time)),
format!("{}", escape_command(&command).style(self.bold)),
)
} else {
("".to_string(), format!("exec('{call_id}')"))
};
let output = if exit_code == 0 { stdout } else { stderr };
let truncated_output = output
.lines()
.take(MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL)
.collect::<Vec<_>>()
.join("\n");
match exit_code {
0 => {
let title = format!("{call} succeeded{duration}:");
ts_println!(self, "{}", title.style(self.green));
}
_ => {
let title = format!("{call} exited {exit_code}{duration}:");
ts_println!(self, "{}", title.style(self.red));
}
}
println!("{}", truncated_output.style(self.dimmed));
}
EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
call_id,
server,
tool,
arguments,
}) => {
// Build fully-qualified tool name: server.tool
let fq_tool_name = format!("{server}.{tool}");
// Format arguments as compact JSON so they fit on one line.
let args_str = arguments
.as_ref()
.map(|v: &serde_json::Value| {
serde_json::to_string(v).unwrap_or_else(|_| v.to_string())
})
.unwrap_or_default();
let invocation = if args_str.is_empty() {
format!("{fq_tool_name}()")
} else {
format!("{fq_tool_name}({args_str})")
};
self.call_id_to_tool_call.insert(
call_id.clone(),
McpToolCallBegin {
invocation: invocation.clone(),
start_time: Instant::now(),
},
);
ts_println!(
self,
"{} {}",
"tool".style(self.magenta),
invocation.style(self.bold),
);
}
EventMsg::McpToolCallEnd(tool_call_end_event) => {
let is_success = tool_call_end_event.is_success();
let McpToolCallEndEvent { call_id, result } = tool_call_end_event;
// Retrieve start time and invocation for duration calculation and labeling.
let info = self.call_id_to_tool_call.remove(&call_id);
let (duration, invocation) = if let Some(McpToolCallBegin {
invocation,
start_time,
..
}) = info
{
(format!(" in {}", format_elapsed(start_time)), invocation)
} else {
(String::new(), format!("tool('{call_id}')"))
};
let status_str = if is_success { "success" } else { "failed" };
let title_style = if is_success { self.green } else { self.red };
let title = format!("{invocation} {status_str}{duration}:");
ts_println!(self, "{}", title.style(title_style));
if let Ok(res) = result {
let val: serde_json::Value = res.into();
let pretty =
serde_json::to_string_pretty(&val).unwrap_or_else(|_| val.to_string());
for line in pretty.lines().take(MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL) {
println!("{}", line.style(self.dimmed));
}
}
}
EventMsg::PatchApplyBegin(PatchApplyBeginEvent {
call_id,
auto_approved,
changes,
}) => {
// Store metadata so we can calculate duration later when we
// receive the corresponding PatchApplyEnd event.
self.call_id_to_patch.insert(
call_id.clone(),
PatchApplyBegin {
start_time: Instant::now(),
auto_approved,
},
);
ts_println!(
self,
"{} auto_approved={}:",
"apply_patch".style(self.magenta),
auto_approved,
);
// Pretty-print the patch summary with colored diff markers so
// its easy to scan in the terminal output.
for (path, change) in changes.iter() {
match change {
FileChange::Add { content } => {
let header = format!(
"{} {}",
format_file_change(change),
path.to_string_lossy()
);
println!("{}", header.style(self.magenta));
for line in content.lines() {
println!("{}", line.style(self.green));
}
}
FileChange::Delete => {
let header = format!(
"{} {}",
format_file_change(change),
path.to_string_lossy()
);
println!("{}", header.style(self.magenta));
}
FileChange::Update {
unified_diff,
move_path,
} => {
let header = if let Some(dest) = move_path {
format!(
"{} {} -> {}",
format_file_change(change),
path.to_string_lossy(),
dest.to_string_lossy()
)
} else {
format!("{} {}", format_file_change(change), path.to_string_lossy())
};
println!("{}", header.style(self.magenta));
// Colorize diff lines. We keep file header lines
// (--- / +++) without extra coloring so they are
// still readable.
for diff_line in unified_diff.lines() {
if diff_line.starts_with('+') && !diff_line.starts_with("+++") {
println!("{}", diff_line.style(self.green));
} else if diff_line.starts_with('-')
&& !diff_line.starts_with("---")
{
println!("{}", diff_line.style(self.red));
} else {
println!("{diff_line}");
}
}
}
}
}
}
EventMsg::PatchApplyEnd(PatchApplyEndEvent {
call_id,
stdout,
stderr,
success,
}) => {
let patch_begin = self.call_id_to_patch.remove(&call_id);
// Compute duration and summary label similar to exec commands.
let (duration, label) = if let Some(PatchApplyBegin {
start_time,
auto_approved,
}) = patch_begin
{
(
format!(" in {}", format_elapsed(start_time)),
format!("apply_patch(auto_approved={auto_approved})"),
)
} else {
(String::new(), format!("apply_patch('{call_id}')"))
};
let (exit_code, output, title_style) = if success {
(0, stdout, self.green)
} else {
(1, stderr, self.red)
};
let title = format!("{label} exited {exit_code}{duration}:");
ts_println!(self, "{}", title.style(title_style));
for line in output.lines() {
println!("{}", line.style(self.dimmed));
}
}
EventMsg::ExecApprovalRequest(_) => {
// Should we exit?
}
EventMsg::ApplyPatchApprovalRequest(_) => {
// Should we exit?
}
EventMsg::AgentReasoning(agent_reasoning_event) => {
if self.show_agent_reasoning {
ts_println!(
self,
"{}\n{}",
"thinking".style(self.italic).style(self.magenta),
agent_reasoning_event.text
);
}
}
EventMsg::SessionConfigured(session_configured_event) => {
let SessionConfiguredEvent {
session_id,
model,
history_log_id: _,
history_entry_count: _,
} = session_configured_event;
ts_println!(
self,
"{} {}",
"codex session".style(self.magenta).style(self.bold),
session_id.to_string().style(self.dimmed)
);
ts_println!(self, "model: {}", model);
println!();
}
EventMsg::GetHistoryEntryResponse(_) => {
// Currently ignored in exec output.
}
}
}
}
fn escape_command(command: &[String]) -> String {
try_join(command.iter().map(|s| s.as_str())).unwrap_or_else(|_| command.join(" "))
}
fn format_file_change(change: &FileChange) -> &'static str {
match change {
FileChange::Add { .. } => "A",
FileChange::Delete => "D",
FileChange::Update {
move_path: Some(_), ..
} => "R",
FileChange::Update {
move_path: None, ..
} => "M",
}
}

View File

@@ -1,520 +0,0 @@
use codex_common::elapsed::format_elapsed;
use codex_core::config::Config;
use codex_core::protocol::AgentMessageDeltaEvent;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::AgentReasoningDeltaEvent;
use codex_core::protocol::BackgroundEventEvent;
use codex_core::protocol::ErrorEvent;
use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecCommandBeginEvent;
use codex_core::protocol::ExecCommandEndEvent;
use codex_core::protocol::FileChange;
use codex_core::protocol::McpToolCallBeginEvent;
use codex_core::protocol::McpToolCallEndEvent;
use codex_core::protocol::PatchApplyBeginEvent;
use codex_core::protocol::PatchApplyEndEvent;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::protocol::TokenUsage;
use owo_colors::OwoColorize;
use owo_colors::Style;
use shlex::try_join;
use std::collections::HashMap;
use std::io::Write;
use std::time::Instant;
use crate::event_processor::EventProcessor;
use crate::event_processor::create_config_summary_entries;
/// This should be configurable. When used in CI, users may not want to impose
/// a limit so they can see the full transcript.
const MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL: usize = 20;
pub(crate) struct EventProcessorWithHumanOutput {
call_id_to_command: HashMap<String, ExecCommandBegin>,
call_id_to_patch: HashMap<String, PatchApplyBegin>,
/// Tracks in-flight MCP tool calls so we can calculate duration and print
/// a concise summary when the corresponding `McpToolCallEnd` event is
/// received.
call_id_to_tool_call: HashMap<String, McpToolCallBegin>,
// To ensure that --color=never is respected, ANSI escapes _must_ be added
// using .style() with one of these fields. If you need a new style, add a
// new field here.
bold: Style,
italic: Style,
dimmed: Style,
magenta: Style,
red: Style,
green: Style,
cyan: Style,
/// Whether to include `AgentReasoning` events in the output.
show_agent_reasoning: bool,
answer_started: bool,
reasoning_started: bool,
}
impl EventProcessorWithHumanOutput {
pub(crate) fn create_with_ansi(with_ansi: bool, config: &Config) -> Self {
let call_id_to_command = HashMap::new();
let call_id_to_patch = HashMap::new();
let call_id_to_tool_call = HashMap::new();
if with_ansi {
Self {
call_id_to_command,
call_id_to_patch,
bold: Style::new().bold(),
italic: Style::new().italic(),
dimmed: Style::new().dimmed(),
magenta: Style::new().magenta(),
red: Style::new().red(),
green: Style::new().green(),
cyan: Style::new().cyan(),
call_id_to_tool_call,
show_agent_reasoning: !config.hide_agent_reasoning,
answer_started: false,
reasoning_started: false,
}
} else {
Self {
call_id_to_command,
call_id_to_patch,
bold: Style::new(),
italic: Style::new(),
dimmed: Style::new(),
magenta: Style::new(),
red: Style::new(),
green: Style::new(),
cyan: Style::new(),
call_id_to_tool_call,
show_agent_reasoning: !config.hide_agent_reasoning,
answer_started: false,
reasoning_started: false,
}
}
}
}
struct ExecCommandBegin {
command: Vec<String>,
start_time: Instant,
}
/// Metadata captured when an `McpToolCallBegin` event is received.
struct McpToolCallBegin {
/// Formatted invocation string, e.g. `server.tool({"city":"sf"})`.
invocation: String,
/// Timestamp when the call started so we can compute duration later.
start_time: Instant,
}
struct PatchApplyBegin {
start_time: Instant,
auto_approved: bool,
}
// Timestamped println helper. The timestamp is styled with self.dimmed.
#[macro_export]
macro_rules! ts_println {
($self:ident, $($arg:tt)*) => {{
let now = chrono::Utc::now();
let formatted = now.format("[%Y-%m-%dT%H:%M:%S]");
print!("{} ", formatted.style($self.dimmed));
println!($($arg)*);
}};
}
impl EventProcessor for EventProcessorWithHumanOutput {
/// Print a concise summary of the effective configuration that will be used
/// for the session. This mirrors the information shown in the TUI welcome
/// screen.
fn print_config_summary(&mut self, config: &Config, prompt: &str) {
const VERSION: &str = env!("CARGO_PKG_VERSION");
ts_println!(
self,
"OpenAI Codex v{} (research preview)\n--------",
VERSION
);
let entries = create_config_summary_entries(config);
for (key, value) in entries {
println!("{} {}", format!("{key}:").style(self.bold), value);
}
println!("--------");
// Echo the prompt that will be sent to the agent so it is visible in the
// transcript/logs before any events come in. Note the prompt may have been
// read from stdin, so it may not be visible in the terminal otherwise.
ts_println!(
self,
"{}\n{}",
"User instructions:".style(self.bold).style(self.cyan),
prompt
);
}
fn process_event(&mut self, event: Event) {
let Event { id: _, msg } = event;
match msg {
EventMsg::Error(ErrorEvent { message }) => {
let prefix = "ERROR:".style(self.red);
ts_println!(self, "{prefix} {message}");
}
EventMsg::BackgroundEvent(BackgroundEventEvent { message }) => {
ts_println!(self, "{}", message.style(self.dimmed));
}
EventMsg::TaskStarted | EventMsg::TaskComplete(_) => {
// Ignore.
}
EventMsg::TokenCount(TokenUsage { total_tokens, .. }) => {
ts_println!(self, "tokens used: {total_tokens}");
}
EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
if !self.answer_started {
ts_println!(self, "{}\n", "codex".style(self.italic).style(self.magenta));
self.answer_started = true;
}
print!("{delta}");
#[allow(clippy::expect_used)]
std::io::stdout().flush().expect("could not flush stdout");
}
EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }) => {
if !self.show_agent_reasoning {
return;
}
if !self.reasoning_started {
ts_println!(
self,
"{}\n",
"thinking".style(self.italic).style(self.magenta),
);
self.reasoning_started = true;
}
print!("{delta}");
#[allow(clippy::expect_used)]
std::io::stdout().flush().expect("could not flush stdout");
}
EventMsg::AgentMessage(AgentMessageEvent { message }) => {
// if answer_started is false, this means we haven't received any
// delta. Thus, we need to print the message as a new answer.
if !self.answer_started {
ts_println!(
self,
"{}\n{}",
"codex".style(self.italic).style(self.magenta),
message,
);
} else {
println!();
self.answer_started = false;
}
}
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
call_id,
command,
cwd,
}) => {
self.call_id_to_command.insert(
call_id.clone(),
ExecCommandBegin {
command: command.clone(),
start_time: Instant::now(),
},
);
ts_println!(
self,
"{} {} in {}",
"exec".style(self.magenta),
escape_command(&command).style(self.bold),
cwd.to_string_lossy(),
);
}
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
call_id,
stdout,
stderr,
exit_code,
}) => {
let exec_command = self.call_id_to_command.remove(&call_id);
let (duration, call) = if let Some(ExecCommandBegin {
command,
start_time,
}) = exec_command
{
(
format!(" in {}", format_elapsed(start_time)),
format!("{}", escape_command(&command).style(self.bold)),
)
} else {
("".to_string(), format!("exec('{call_id}')"))
};
let output = if exit_code == 0 { stdout } else { stderr };
let truncated_output = output
.lines()
.take(MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL)
.collect::<Vec<_>>()
.join("\n");
match exit_code {
0 => {
let title = format!("{call} succeeded{duration}:");
ts_println!(self, "{}", title.style(self.green));
}
_ => {
let title = format!("{call} exited {exit_code}{duration}:");
ts_println!(self, "{}", title.style(self.red));
}
}
println!("{}", truncated_output.style(self.dimmed));
}
EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
call_id,
server,
tool,
arguments,
}) => {
// Build fully-qualified tool name: server.tool
let fq_tool_name = format!("{server}.{tool}");
// Format arguments as compact JSON so they fit on one line.
let args_str = arguments
.as_ref()
.map(|v: &serde_json::Value| {
serde_json::to_string(v).unwrap_or_else(|_| v.to_string())
})
.unwrap_or_default();
let invocation = if args_str.is_empty() {
format!("{fq_tool_name}()")
} else {
format!("{fq_tool_name}({args_str})")
};
self.call_id_to_tool_call.insert(
call_id.clone(),
McpToolCallBegin {
invocation: invocation.clone(),
start_time: Instant::now(),
},
);
ts_println!(
self,
"{} {}",
"tool".style(self.magenta),
invocation.style(self.bold),
);
}
EventMsg::McpToolCallEnd(tool_call_end_event) => {
let is_success = tool_call_end_event.is_success();
let McpToolCallEndEvent { call_id, result } = tool_call_end_event;
// Retrieve start time and invocation for duration calculation and labeling.
let info = self.call_id_to_tool_call.remove(&call_id);
let (duration, invocation) = if let Some(McpToolCallBegin {
invocation,
start_time,
..
}) = info
{
(format!(" in {}", format_elapsed(start_time)), invocation)
} else {
(String::new(), format!("tool('{call_id}')"))
};
let status_str = if is_success { "success" } else { "failed" };
let title_style = if is_success { self.green } else { self.red };
let title = format!("{invocation} {status_str}{duration}:");
ts_println!(self, "{}", title.style(title_style));
if let Ok(res) = result {
let val: serde_json::Value = res.into();
let pretty =
serde_json::to_string_pretty(&val).unwrap_or_else(|_| val.to_string());
for line in pretty.lines().take(MAX_OUTPUT_LINES_FOR_EXEC_TOOL_CALL) {
println!("{}", line.style(self.dimmed));
}
}
}
EventMsg::PatchApplyBegin(PatchApplyBeginEvent {
call_id,
auto_approved,
changes,
}) => {
// Store metadata so we can calculate duration later when we
// receive the corresponding PatchApplyEnd event.
self.call_id_to_patch.insert(
call_id.clone(),
PatchApplyBegin {
start_time: Instant::now(),
auto_approved,
},
);
ts_println!(
self,
"{} auto_approved={}:",
"apply_patch".style(self.magenta),
auto_approved,
);
// Pretty-print the patch summary with colored diff markers so
// it's easy to scan in the terminal output.
for (path, change) in changes.iter() {
match change {
FileChange::Add { content } => {
let header = format!(
"{} {}",
format_file_change(change),
path.to_string_lossy()
);
println!("{}", header.style(self.magenta));
for line in content.lines() {
println!("{}", line.style(self.green));
}
}
FileChange::Delete => {
let header = format!(
"{} {}",
format_file_change(change),
path.to_string_lossy()
);
println!("{}", header.style(self.magenta));
}
FileChange::Update {
unified_diff,
move_path,
} => {
let header = if let Some(dest) = move_path {
format!(
"{} {} -> {}",
format_file_change(change),
path.to_string_lossy(),
dest.to_string_lossy()
)
} else {
format!("{} {}", format_file_change(change), path.to_string_lossy())
};
println!("{}", header.style(self.magenta));
// Colorize diff lines. We keep file header lines
// (--- / +++) without extra coloring so they are
// still readable.
for diff_line in unified_diff.lines() {
if diff_line.starts_with('+') && !diff_line.starts_with("+++") {
println!("{}", diff_line.style(self.green));
} else if diff_line.starts_with('-')
&& !diff_line.starts_with("---")
{
println!("{}", diff_line.style(self.red));
} else {
println!("{diff_line}");
}
}
}
}
}
}
EventMsg::PatchApplyEnd(PatchApplyEndEvent {
call_id,
stdout,
stderr,
success,
}) => {
let patch_begin = self.call_id_to_patch.remove(&call_id);
// Compute duration and summary label similar to exec commands.
let (duration, label) = if let Some(PatchApplyBegin {
start_time,
auto_approved,
}) = patch_begin
{
(
format!(" in {}", format_elapsed(start_time)),
format!("apply_patch(auto_approved={auto_approved})"),
)
} else {
(String::new(), format!("apply_patch('{call_id}')"))
};
let (exit_code, output, title_style) = if success {
(0, stdout, self.green)
} else {
(1, stderr, self.red)
};
let title = format!("{label} exited {exit_code}{duration}:");
ts_println!(self, "{}", title.style(title_style));
for line in output.lines() {
println!("{}", line.style(self.dimmed));
}
}
EventMsg::ExecApprovalRequest(_) => {
// Should we exit?
}
EventMsg::ApplyPatchApprovalRequest(_) => {
// Should we exit?
}
EventMsg::AgentReasoning(agent_reasoning_event) => {
if self.show_agent_reasoning {
if !self.reasoning_started {
ts_println!(
self,
"{}\n{}",
"codex".style(self.italic).style(self.magenta),
agent_reasoning_event.text,
);
} else {
println!();
self.reasoning_started = false;
}
}
}
EventMsg::SessionConfigured(session_configured_event) => {
let SessionConfiguredEvent {
session_id,
model,
history_log_id: _,
history_entry_count: _,
} = session_configured_event;
ts_println!(
self,
"{} {}",
"codex session".style(self.magenta).style(self.bold),
session_id.to_string().style(self.dimmed)
);
ts_println!(self, "model: {}", model);
println!();
}
EventMsg::GetHistoryEntryResponse(_) => {
// Currently ignored in exec output.
}
}
}
}
fn escape_command(command: &[String]) -> String {
try_join(command.iter().map(|s| s.as_str())).unwrap_or_else(|_| command.join(" "))
}
fn format_file_change(change: &FileChange) -> &'static str {
match change {
FileChange::Add { .. } => "A",
FileChange::Delete => "D",
FileChange::Update {
move_path: Some(_), ..
} => "R",
FileChange::Update {
move_path: None, ..
} => "M",
}
}

View File

@@ -1,48 +0,0 @@
use std::collections::HashMap;
use codex_core::config::Config;
use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use serde_json::json;
use crate::event_processor::EventProcessor;
use crate::event_processor::create_config_summary_entries;
pub(crate) struct EventProcessorWithJsonOutput;
impl EventProcessorWithJsonOutput {
pub fn new() -> Self {
Self {}
}
}
impl EventProcessor for EventProcessorWithJsonOutput {
fn print_config_summary(&mut self, config: &Config, prompt: &str) {
let entries = create_config_summary_entries(config)
.into_iter()
.map(|(key, value)| (key.to_string(), value))
.collect::<HashMap<String, String>>();
#[allow(clippy::expect_used)]
let config_json =
serde_json::to_string(&entries).expect("Failed to serialize config summary to JSON");
println!("{config_json}");
let prompt_json = json!({
"prompt": prompt,
});
println!("{prompt_json}");
}
fn process_event(&mut self, event: Event) {
match event.msg {
EventMsg::AgentMessageDelta(_) | EventMsg::AgentReasoningDelta(_) => {
// Suppress streaming events in JSON mode.
}
_ => {
if let Ok(line) = serde_json::to_string(&event) {
println!("{line}");
}
}
}
}
}

View File

@@ -1,7 +1,5 @@
mod cli;
mod event_processor;
mod event_processor_with_human_output;
mod event_processor_with_json_output;
use std::io::IsTerminal;
use std::io::Read;
@@ -21,15 +19,12 @@ use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::util::is_inside_git_repo;
use event_processor_with_human_output::EventProcessorWithHumanOutput;
use event_processor_with_json_output::EventProcessorWithJsonOutput;
use event_processor::EventProcessor;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing_subscriber::EnvFilter;
use crate::event_processor::EventProcessor;
pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()> {
let Cli {
images,
@@ -41,7 +36,6 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
skip_git_repo_check,
color,
last_message_file,
json: json_mode,
sandbox_mode: sandbox_mode_cli_arg,
prompt,
config_overrides,
@@ -121,15 +115,8 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
};
let config = Config::load_with_cli_overrides(cli_kv_overrides, overrides)?;
let mut event_processor: Box<dyn EventProcessor> = if json_mode {
Box::new(EventProcessorWithJsonOutput::new())
} else {
Box::new(EventProcessorWithHumanOutput::create_with_ansi(
stdout_with_ansi,
&config,
))
};
let mut event_processor =
EventProcessor::create_with_ansi(stdout_with_ansi, !config.hide_agent_reasoning);
// Print the effective configuration and prompt so users can see what Codex
// is using.
event_processor.print_config_summary(&config, &prompt);

View File

@@ -23,10 +23,3 @@ file-search *args:
# format code
fmt:
cargo fmt -- --config imports_granularity=Item
fix:
cargo clippy --fix --all-features --tests --allow-dirty
install:
rustup show active-toolchain
cargo fetch

View File

@@ -9,7 +9,6 @@ use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::Submission;
use codex_core::protocol::TaskCompleteEvent;
use mcp_types::CallToolResult;
use mcp_types::CallToolResultContent;
@@ -67,24 +66,14 @@ pub async fn run_codex_tool_session(
.send(codex_event_to_notification(&first_event))
.await;
// Use the original MCP request ID as the `sub_id` for the Codex submission so that
// any events emitted for this tool-call can be correlated with the
// originating `tools/call` request.
let sub_id = match &id {
RequestId::String(s) => s.clone(),
RequestId::Integer(n) => n.to_string(),
};
let submission = Submission {
id: sub_id,
op: Op::UserInput {
if let Err(e) = codex
.submit(Op::UserInput {
items: vec![InputItem::Text {
text: initial_prompt.clone(),
}],
},
};
if let Err(e) = codex.submit_with_id(submission).await {
})
.await
{
tracing::error!("Failed to submit initial prompt: {e}");
}
@@ -171,12 +160,6 @@ pub async fn run_codex_tool_session(
EventMsg::SessionConfigured(_) => {
tracing::error!("unexpected SessionConfigured event");
}
EventMsg::AgentMessageDelta(_) => {
// TODO: think how we want to support this in the MCP
}
EventMsg::AgentReasoningDelta(_) => {
// TODO: think how we want to support this in the MCP
}
EventMsg::Error(_)
| EventMsg::TaskStarted
| EventMsg::TokenCount(_)

View File

@@ -1,3 +0,0 @@
[toolchain]
channel = "1.88.0"
components = [ "clippy", "rustfmt", "rust-src"]

View File

@@ -61,5 +61,4 @@ unicode-segmentation = "1.12.0"
uuid = "1"
[dev-dependencies]
insta = "1.43.1"
pretty_assertions = "1"

View File

@@ -18,15 +18,8 @@ use crossterm::event::KeyEvent;
use crossterm::event::MouseEvent;
use crossterm::event::MouseEventKind;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::mpsc::Receiver;
use std::sync::mpsc::channel;
use std::thread;
use std::time::Duration;
/// Time window for debouncing redraw requests.
const REDRAW_DEBOUNCE: Duration = Duration::from_millis(10);
/// Top-level application state: which full-screen view is currently active.
#[allow(clippy::large_enum_variant)]
@@ -53,9 +46,6 @@ pub(crate) struct App<'a> {
file_search: FileSearchManager,
/// True when a redraw has been scheduled but not yet executed.
pending_redraw: Arc<Mutex<bool>>,
/// Stored parameters needed to instantiate the ChatWidget later, e.g.,
/// after dismissing the Git-repo warning.
chat_args: Option<ChatWidgetArgs>,
@@ -70,7 +60,7 @@ struct ChatWidgetArgs {
initial_images: Vec<PathBuf>,
}
impl App<'_> {
impl<'a> App<'a> {
pub(crate) fn new(
config: Config,
initial_prompt: Option<String>,
@@ -80,7 +70,6 @@ impl App<'_> {
) -> Self {
let (app_event_tx, app_event_rx) = channel();
let app_event_tx = AppEventSender::new(app_event_tx);
let pending_redraw = Arc::new(Mutex::new(false));
let scroll_event_helper = ScrollEventHelper::new(app_event_tx.clone());
// Spawn a dedicated thread for reading the crossterm event loop and
@@ -94,7 +83,7 @@ impl App<'_> {
app_event_tx.send(AppEvent::KeyEvent(key_event));
}
crossterm::event::Event::Resize(_, _) => {
app_event_tx.send(AppEvent::RequestRedraw);
app_event_tx.send(AppEvent::Redraw);
}
crossterm::event::Event::Mouse(MouseEvent {
kind: MouseEventKind::ScrollUp,
@@ -109,7 +98,21 @@ impl App<'_> {
scroll_event_helper.scroll_down();
}
crossterm::event::Event::Paste(pasted) => {
app_event_tx.send(AppEvent::Paste(pasted));
use crossterm::event::KeyModifiers;
for ch in pasted.chars() {
let key_event = match ch {
'\n' | '\r' => {
// Represent newline as <Shift+Enter> so that the bottom
// pane treats it as a literal newline instead of a submit
// action (submission is only triggered on Enter *without*
// any modifiers).
KeyEvent::new(KeyCode::Enter, KeyModifiers::SHIFT)
}
_ => KeyEvent::new(KeyCode::Char(ch), KeyModifiers::empty()),
};
app_event_tx.send(AppEvent::KeyEvent(key_event));
}
}
_ => {
// Ignore any other events.
@@ -163,7 +166,6 @@ impl App<'_> {
app_state,
config,
file_search,
pending_redraw,
chat_args,
}
}
@@ -174,29 +176,6 @@ impl App<'_> {
self.app_event_tx.clone()
}
/// Schedule a redraw if one is not already pending.
#[allow(clippy::unwrap_used)]
fn schedule_redraw(&self) {
{
#[allow(clippy::unwrap_used)]
let mut flag = self.pending_redraw.lock().unwrap();
if *flag {
return;
}
*flag = true;
}
let tx = self.app_event_tx.clone();
let pending_redraw = self.pending_redraw.clone();
thread::spawn(move || {
thread::sleep(REDRAW_DEBOUNCE);
tx.send(AppEvent::Redraw);
#[allow(clippy::unwrap_used)]
let mut f = pending_redraw.lock().unwrap();
*f = false;
});
}
pub(crate) fn run(
&mut self,
terminal: &mut tui::Tui,
@@ -204,13 +183,10 @@ impl App<'_> {
) -> Result<()> {
// Insert an event to trigger the first render.
let app_event_tx = self.app_event_tx.clone();
app_event_tx.send(AppEvent::RequestRedraw);
app_event_tx.send(AppEvent::Redraw);
while let Ok(event) = self.app_event_rx.recv() {
match event {
AppEvent::RequestRedraw => {
self.schedule_redraw();
}
AppEvent::Redraw => {
self.draw_next_frame(terminal)?;
}
@@ -237,21 +213,7 @@ impl App<'_> {
modifiers: crossterm::event::KeyModifiers::CONTROL,
..
} => {
match &mut self.app_state {
AppState::Chat { widget } => {
if widget.composer_is_empty() {
self.app_event_tx.send(AppEvent::ExitRequest);
} else {
// Treat Ctrl+D as a normal key event when the composer
// is not empty so that it doesn't quit the application
// prematurely.
self.dispatch_key_event(key_event);
}
}
AppState::Login { .. } | AppState::GitWarning { .. } => {
self.app_event_tx.send(AppEvent::ExitRequest);
}
}
self.app_event_tx.send(AppEvent::ExitRequest);
}
_ => {
self.dispatch_key_event(key_event);
@@ -261,9 +223,6 @@ impl App<'_> {
AppEvent::Scroll(scroll_delta) => {
self.dispatch_scroll_event(scroll_delta);
}
AppEvent::Paste(text) => {
self.dispatch_paste_event(text);
}
AppEvent::CodexEvent(event) => {
self.dispatch_codex_event(event);
}
@@ -287,7 +246,7 @@ impl App<'_> {
Vec::new(),
));
self.app_state = AppState::Chat { widget: new_widget };
self.app_event_tx.send(AppEvent::RequestRedraw);
self.app_event_tx.send(AppEvent::Redraw);
}
SlashCommand::ToggleMouseMode => {
if let Err(e) = mouse_capture.toggle() {
@@ -335,8 +294,6 @@ impl App<'_> {
}
fn draw_next_frame(&mut self, terminal: &mut tui::Tui) -> Result<()> {
// TODO: add a throttle to avoid redrawing too often
match &mut self.app_state {
AppState::Chat { widget } => {
terminal.draw(|frame| frame.render_widget_ref(&**widget, frame.area()))?;
@@ -374,7 +331,7 @@ impl App<'_> {
args.initial_images,
));
self.app_state = AppState::Chat { widget };
self.app_event_tx.send(AppEvent::RequestRedraw);
self.app_event_tx.send(AppEvent::Redraw);
}
GitWarningOutcome::Quit => {
self.app_event_tx.send(AppEvent::ExitRequest);
@@ -386,13 +343,6 @@ impl App<'_> {
}
}
fn dispatch_paste_event(&mut self, pasted: String) {
match &mut self.app_state {
AppState::Chat { widget } => widget.handle_paste(pasted),
AppState::Login { .. } | AppState::GitWarning { .. } => {}
}
}
fn dispatch_scroll_event(&mut self, scroll_delta: i32) {
match &mut self.app_state {
AppState::Chat { widget } => widget.handle_scroll_delta(scroll_delta),

View File

@@ -8,17 +8,10 @@ use crate::slash_command::SlashCommand;
pub(crate) enum AppEvent {
CodexEvent(Event),
/// Request a redraw which will be debounced by the [`App`].
RequestRedraw,
/// Actually draw the next frame.
Redraw,
KeyEvent(KeyEvent),
/// Text pasted from the terminal clipboard.
Paste(String),
/// Scroll event with a value representing the "scroll delta" as the net
/// scroll up/down events within a short time window.
Scroll(i32),

View File

@@ -28,9 +28,6 @@ const MIN_TEXTAREA_ROWS: usize = 1;
const BORDER_LINES: u16 = 2;
const BASE_PLACEHOLDER_TEXT: &str = "send a message";
/// If the pasted content exceeds this number of characters, replace it with a
/// placeholder in the UI.
const LARGE_PASTE_CHAR_THRESHOLD: usize = 1000;
/// Result returned when the user interacts with the text area.
pub enum InputResult {
@@ -46,7 +43,6 @@ pub(crate) struct ChatComposer<'a> {
ctrl_c_quit_hint: bool,
dismissed_file_popup_token: Option<String>,
current_file_query: Option<String>,
pending_pastes: Vec<(String, String)>,
}
/// Popup state at most one can be visible at any time.
@@ -70,17 +66,11 @@ impl ChatComposer<'_> {
ctrl_c_quit_hint: false,
dismissed_file_popup_token: None,
current_file_query: None,
pending_pastes: Vec::new(),
};
this.update_border(has_input_focus);
this
}
/// Returns true if the composer currently contains no user input.
pub(crate) fn is_empty(&self) -> bool {
self.textarea.is_empty()
}
/// Update the cached *context-left* percentage and refresh the placeholder
/// text. The UI relies on the placeholder to convey the remaining
/// context when the composer is empty.
@@ -136,20 +126,6 @@ impl ChatComposer<'_> {
self.update_border(has_focus);
}
pub fn handle_paste(&mut self, pasted: String) -> bool {
let char_count = pasted.chars().count();
if char_count > LARGE_PASTE_CHAR_THRESHOLD {
let placeholder = format!("[Pasted Content {char_count} chars]");
self.textarea.insert_str(&placeholder);
self.pending_pastes.push((placeholder, pasted));
} else {
self.textarea.insert_str(&pasted);
}
self.sync_command_popup();
self.sync_file_search_popup();
true
}
/// Integrate results from an asynchronous file search.
pub(crate) fn on_file_search_result(&mut self, query: String, matches: Vec<FileMatch>) {
// Only apply if user is still editing a token starting with `query`.
@@ -438,18 +414,10 @@ impl ChatComposer<'_> {
alt: false,
ctrl: false,
} => {
let mut text = self.textarea.lines().join("\n");
let text = self.textarea.lines().join("\n");
self.textarea.select_all();
self.textarea.cut();
// Replace all pending pastes in the text
for (placeholder, actual) in &self.pending_pastes {
if text.contains(placeholder) {
text = text.replace(placeholder, actual);
}
}
self.pending_pastes.clear();
if text.is_empty() {
(InputResult::None, true)
} else {
@@ -475,71 +443,10 @@ impl ChatComposer<'_> {
/// Handle generic Input events that modify the textarea content.
fn handle_input_basic(&mut self, input: Input) -> (InputResult, bool) {
// Special handling for backspace on placeholders
if let Input {
key: Key::Backspace,
..
} = input
{
if self.try_remove_placeholder_at_cursor() {
return (InputResult::None, true);
}
}
// Normal input handling
self.textarea.input(input);
let text_after = self.textarea.lines().join("\n");
// Check if any placeholders were removed and remove their corresponding pending pastes
self.pending_pastes
.retain(|(placeholder, _)| text_after.contains(placeholder));
(InputResult::None, true)
}
/// Attempts to remove a placeholder if the cursor is at the end of one.
/// Returns true if a placeholder was removed.
fn try_remove_placeholder_at_cursor(&mut self) -> bool {
let (row, col) = self.textarea.cursor();
let line = self
.textarea
.lines()
.get(row)
.map(|s| s.as_str())
.unwrap_or("");
// Find any placeholder that ends at the cursor position
let placeholder_to_remove = self.pending_pastes.iter().find_map(|(ph, _)| {
if col < ph.len() {
return None;
}
let potential_ph_start = col - ph.len();
if line[potential_ph_start..col] == *ph {
Some(ph.clone())
} else {
None
}
});
if let Some(placeholder) = placeholder_to_remove {
// Remove the entire placeholder from the text
let placeholder_len = placeholder.len();
for _ in 0..placeholder_len {
self.textarea.input(Input {
key: Key::Backspace,
ctrl: false,
alt: false,
shift: false,
});
}
// Remove from pending pastes
self.pending_pastes.retain(|(ph, _)| ph != &placeholder);
true
} else {
false
}
}
/// Synchronize `self.command_popup` with the current text in the
/// textarea. This must be called after every modification that can change
/// the text so the popup is shown/updated/hidden as appropriate.
@@ -717,10 +624,7 @@ impl WidgetRef for &ChatComposer<'_> {
#[cfg(test)]
mod tests {
use crate::bottom_pane::AppEventSender;
use crate::bottom_pane::ChatComposer;
use crate::bottom_pane::InputResult;
use crate::bottom_pane::chat_composer::LARGE_PASTE_CHAR_THRESHOLD;
use tui_textarea::TextArea;
#[test]
@@ -866,324 +770,4 @@ mod tests {
);
}
}
#[test]
fn handle_paste_small_inserts_text() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
let needs_redraw = composer.handle_paste("hello".to_string());
assert!(needs_redraw);
assert_eq!(composer.textarea.lines(), ["hello"]);
assert!(composer.pending_pastes.is_empty());
let (result, _) =
composer.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
match result {
InputResult::Submitted(text) => assert_eq!(text, "hello"),
_ => panic!("expected Submitted"),
}
}
#[test]
fn handle_paste_large_uses_placeholder_and_replaces_on_submit() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
let large = "x".repeat(LARGE_PASTE_CHAR_THRESHOLD + 10);
let needs_redraw = composer.handle_paste(large.clone());
assert!(needs_redraw);
let placeholder = format!("[Pasted Content {} chars]", large.chars().count());
assert_eq!(composer.textarea.lines(), [placeholder.as_str()]);
assert_eq!(composer.pending_pastes.len(), 1);
assert_eq!(composer.pending_pastes[0].0, placeholder);
assert_eq!(composer.pending_pastes[0].1, large);
let (result, _) =
composer.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
match result {
InputResult::Submitted(text) => assert_eq!(text, large),
_ => panic!("expected Submitted"),
}
assert!(composer.pending_pastes.is_empty());
}
#[test]
fn edit_clears_pending_paste() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let large = "y".repeat(LARGE_PASTE_CHAR_THRESHOLD + 1);
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
composer.handle_paste(large);
assert_eq!(composer.pending_pastes.len(), 1);
// Any edit that removes the placeholder should clear pending_paste
composer.handle_key_event(KeyEvent::new(KeyCode::Backspace, KeyModifiers::NONE));
assert!(composer.pending_pastes.is_empty());
}
#[test]
fn ui_snapshots() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
use insta::assert_snapshot;
use ratatui::Terminal;
use ratatui::backend::TestBackend;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut terminal = match Terminal::new(TestBackend::new(100, 10)) {
Ok(t) => t,
Err(e) => panic!("Failed to create terminal: {e}"),
};
let test_cases = vec![
("empty", None),
("small", Some("short".to_string())),
("large", Some("z".repeat(LARGE_PASTE_CHAR_THRESHOLD + 5))),
("multiple_pastes", None),
("backspace_after_pastes", None),
];
for (name, input) in test_cases {
// Create a fresh composer for each test case
let mut composer = ChatComposer::new(true, sender.clone());
if let Some(text) = input {
composer.handle_paste(text);
} else if name == "multiple_pastes" {
// First large paste
composer.handle_paste("x".repeat(LARGE_PASTE_CHAR_THRESHOLD + 3));
// Second large paste
composer.handle_paste("y".repeat(LARGE_PASTE_CHAR_THRESHOLD + 7));
// Small paste
composer.handle_paste(" another short paste".to_string());
} else if name == "backspace_after_pastes" {
// Three large pastes
composer.handle_paste("a".repeat(LARGE_PASTE_CHAR_THRESHOLD + 2));
composer.handle_paste("b".repeat(LARGE_PASTE_CHAR_THRESHOLD + 4));
composer.handle_paste("c".repeat(LARGE_PASTE_CHAR_THRESHOLD + 6));
// Move cursor to end and press backspace
composer.textarea.move_cursor(tui_textarea::CursorMove::End);
composer.handle_key_event(KeyEvent::new(KeyCode::Backspace, KeyModifiers::NONE));
}
terminal
.draw(|f| f.render_widget_ref(&composer, f.area()))
.unwrap_or_else(|e| panic!("Failed to draw {name} composer: {e}"));
assert_snapshot!(name, terminal.backend());
}
}
#[test]
fn test_multiple_pastes_submission() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
// Define test cases: (paste content, is_large)
let test_cases = [
("x".repeat(LARGE_PASTE_CHAR_THRESHOLD + 3), true),
(" and ".to_string(), false),
("y".repeat(LARGE_PASTE_CHAR_THRESHOLD + 7), true),
];
// Expected states after each paste
let mut expected_text = String::new();
let mut expected_pending_count = 0;
// Apply all pastes and build expected state
let states: Vec<_> = test_cases
.iter()
.map(|(content, is_large)| {
composer.handle_paste(content.clone());
if *is_large {
let placeholder = format!("[Pasted Content {} chars]", content.chars().count());
expected_text.push_str(&placeholder);
expected_pending_count += 1;
} else {
expected_text.push_str(content);
}
(expected_text.clone(), expected_pending_count)
})
.collect();
// Verify all intermediate states were correct
assert_eq!(
states,
vec![
(
format!("[Pasted Content {} chars]", test_cases[0].0.chars().count()),
1
),
(
format!(
"[Pasted Content {} chars] and ",
test_cases[0].0.chars().count()
),
1
),
(
format!(
"[Pasted Content {} chars] and [Pasted Content {} chars]",
test_cases[0].0.chars().count(),
test_cases[2].0.chars().count()
),
2
),
]
);
// Submit and verify final expansion
let (result, _) =
composer.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
if let InputResult::Submitted(text) = result {
assert_eq!(text, format!("{} and {}", test_cases[0].0, test_cases[2].0));
} else {
panic!("expected Submitted");
}
}
#[test]
fn test_placeholder_deletion() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
// Define test cases: (content, is_large)
let test_cases = [
("a".repeat(LARGE_PASTE_CHAR_THRESHOLD + 5), true),
(" and ".to_string(), false),
("b".repeat(LARGE_PASTE_CHAR_THRESHOLD + 6), true),
];
// Apply all pastes
let mut current_pos = 0;
let states: Vec<_> = test_cases
.iter()
.map(|(content, is_large)| {
composer.handle_paste(content.clone());
if *is_large {
let placeholder = format!("[Pasted Content {} chars]", content.chars().count());
current_pos += placeholder.len();
} else {
current_pos += content.len();
}
(
composer.textarea.lines().join("\n"),
composer.pending_pastes.len(),
current_pos,
)
})
.collect();
// Delete placeholders one by one and collect states
let mut deletion_states = vec![];
// First deletion
composer
.textarea
.move_cursor(tui_textarea::CursorMove::Jump(0, states[0].2 as u16));
composer.handle_key_event(KeyEvent::new(KeyCode::Backspace, KeyModifiers::NONE));
deletion_states.push((
composer.textarea.lines().join("\n"),
composer.pending_pastes.len(),
));
// Second deletion
composer
.textarea
.move_cursor(tui_textarea::CursorMove::Jump(
0,
composer.textarea.lines().join("\n").len() as u16,
));
composer.handle_key_event(KeyEvent::new(KeyCode::Backspace, KeyModifiers::NONE));
deletion_states.push((
composer.textarea.lines().join("\n"),
composer.pending_pastes.len(),
));
// Verify all states
assert_eq!(
deletion_states,
vec![
(" and [Pasted Content 1006 chars]".to_string(), 1),
(" and ".to_string(), 0),
]
);
}
#[test]
fn test_partial_placeholder_deletion() {
use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyModifiers;
let (tx, _rx) = std::sync::mpsc::channel();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(true, sender);
// Define test cases: (cursor_position_from_end, expected_pending_count)
let test_cases = [
5, // Delete from middle - should clear tracking
0, // Delete from end - should clear tracking
];
let paste = "x".repeat(LARGE_PASTE_CHAR_THRESHOLD + 4);
let placeholder = format!("[Pasted Content {} chars]", paste.chars().count());
let states: Vec<_> = test_cases
.into_iter()
.map(|pos_from_end| {
composer.handle_paste(paste.clone());
composer
.textarea
.move_cursor(tui_textarea::CursorMove::Jump(
0,
(placeholder.len() - pos_from_end) as u16,
));
composer.handle_key_event(KeyEvent::new(KeyCode::Backspace, KeyModifiers::NONE));
let result = (
composer.textarea.lines().join("\n").contains(&placeholder),
composer.pending_pastes.len(),
);
composer.textarea.select_all();
composer.textarea.cut();
result
})
.collect();
assert_eq!(
states,
vec![
(false, 0), // After deleting from middle
(false, 0), // After deleting from end
]
);
}
}

View File

@@ -72,7 +72,8 @@ impl ChatComposerHistory {
return false;
}
if textarea.is_empty() {
let lines = textarea.lines();
if lines.len() == 1 && lines[0].is_empty() {
return true;
}
@@ -84,7 +85,6 @@ impl ChatComposerHistory {
return false;
}
let lines = textarea.lines();
matches!(&self.last_history_text, Some(prev) if prev == &lines.join("\n"))
}

View File

@@ -82,15 +82,6 @@ impl BottomPane<'_> {
}
}
pub fn handle_paste(&mut self, pasted: String) {
if self.active_view.is_none() {
let needs_redraw = self.composer.handle_paste(pasted);
if needs_redraw {
self.request_redraw();
}
}
}
/// Update the status indicator text (only when the `StatusIndicatorView` is
/// active).
pub(crate) fn update_status_text(&mut self, text: String) {
@@ -162,10 +153,6 @@ impl BottomPane<'_> {
}
}
pub(crate) fn composer_is_empty(&self) -> bool {
self.composer.is_empty()
}
pub(crate) fn is_task_running(&self) -> bool {
self.is_task_running
}
@@ -212,7 +199,7 @@ impl BottomPane<'_> {
}
pub(crate) fn request_redraw(&self) {
self.app_event_tx.send(AppEvent::RequestRedraw)
self.app_event_tx.send(AppEvent::Redraw)
}
/// Returns true when a popup inside the composer is visible.

View File

@@ -1,14 +0,0 @@
---
source: tui/src/bottom_pane/chat_composer.rs
expression: terminal.backend()
---
"╭──────────────────────────────────────────────────────────────────────────────────────────────────╮"
"│[Pasted Content 1002 chars][Pasted Content 1004 chars] │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"╰───────────────────────────────────────────────Enter to send | Ctrl+D to quit | Ctrl+J for newline╯"

View File

@@ -1,14 +0,0 @@
---
source: tui/src/bottom_pane/chat_composer.rs
expression: terminal.backend()
---
"╭──────────────────────────────────────────────────────────────────────────────────────────────────╮"
"│ send a message │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"╰───────────────────────────────────────────────Enter to send | Ctrl+D to quit | Ctrl+J for newline╯"

View File

@@ -1,14 +0,0 @@
---
source: tui/src/bottom_pane/chat_composer.rs
expression: terminal.backend()
---
"╭──────────────────────────────────────────────────────────────────────────────────────────────────╮"
"│[Pasted Content 1005 chars] │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"╰───────────────────────────────────────────────Enter to send | Ctrl+D to quit | Ctrl+J for newline╯"

View File

@@ -1,14 +0,0 @@
---
source: tui/src/bottom_pane/chat_composer.rs
expression: terminal.backend()
---
"╭──────────────────────────────────────────────────────────────────────────────────────────────────╮"
"│[Pasted Content 1003 chars][Pasted Content 1007 chars] another short paste │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"╰───────────────────────────────────────────────Enter to send | Ctrl+D to quit | Ctrl+J for newline╯"

View File

@@ -1,14 +0,0 @@
---
source: tui/src/bottom_pane/chat_composer.rs
expression: terminal.backend()
---
"╭──────────────────────────────────────────────────────────────────────────────────────────────────╮"
"│short │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"│ │"
"╰───────────────────────────────────────────────Enter to send | Ctrl+D to quit | Ctrl+J for newline╯"

View File

@@ -24,7 +24,7 @@ impl StatusIndicatorView {
}
}
impl BottomPaneView<'_> for StatusIndicatorView {
impl<'a> BottomPaneView<'a> for StatusIndicatorView {
fn update_status_text(&mut self, text: String) -> ConditionalUpdate {
self.update_text(text);
ConditionalUpdate::NeedsRedraw

View File

@@ -3,9 +3,7 @@ use std::sync::Arc;
use codex_core::codex_wrapper::init_codex;
use codex_core::config::Config;
use codex_core::protocol::AgentMessageDeltaEvent;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::AgentReasoningDeltaEvent;
use codex_core::protocol::AgentReasoningEvent;
use codex_core::protocol::ApplyPatchApprovalRequestEvent;
use codex_core::protocol::ErrorEvent;
@@ -51,9 +49,6 @@ pub(crate) struct ChatWidget<'a> {
config: Config,
initial_user_message: Option<UserMessage>,
token_usage: TokenUsage,
reasoning_buffer: String,
answer_buffer: String,
active_task_id: Option<String>,
}
#[derive(Clone, Copy, Eq, PartialEq)]
@@ -140,9 +135,6 @@ impl ChatWidget<'_> {
initial_images,
),
token_usage: TokenUsage::default(),
reasoning_buffer: String::new(),
answer_buffer: String::new(),
active_task_id: None,
}
}
@@ -182,12 +174,6 @@ impl ChatWidget<'_> {
}
}
pub(crate) fn handle_paste(&mut self, text: String) {
if matches!(self.input_focus, InputFocus::BottomPane) {
self.bottom_pane.handle_paste(text);
}
}
fn submit_user_message(&mut self, user_message: UserMessage) {
let UserMessage { text, image_paths } = user_message;
let mut items: Vec<InputItem> = Vec::new();
@@ -224,30 +210,10 @@ impl ChatWidget<'_> {
self.conversation_history.add_user_message(text);
}
self.conversation_history.scroll_to_bottom();
// IMPORTANT: Starting a *new* user turn. Clear any partially streamed
// answer from a previous turn (e.g., one that was interrupted) so that
// the next AgentMessageDelta spawns a fresh agent message cell instead
// of overwriting the last one.
self.answer_buffer.clear();
self.reasoning_buffer.clear();
}
pub(crate) fn handle_codex_event(&mut self, event: Event) {
// Retain the event ID so we can refer to it after destructuring.
let event_id = event.id.clone();
let Event { id: _, msg } = event;
// When we are in the middle of a task (active_task_id is Some) we drop
// streaming text/reasoning events for *other* task IDs. This prevents
// late tokens from an interrupted run from bleeding into the current
// answer.
let should_drop_streaming = self
.active_task_id
.as_ref()
.map(|active| active != &event_id)
.unwrap_or(false);
let Event { id, msg } = event;
match msg {
EventMsg::SessionConfigured(event) => {
// Record session information at the top of the conversation.
@@ -268,69 +234,18 @@ impl ChatWidget<'_> {
self.request_redraw();
}
EventMsg::AgentMessage(AgentMessageEvent { message }) => {
if should_drop_streaming {
return;
}
// if the answer buffer is empty, this means we haven't received any
// delta. Thus, we need to print the message as a new answer.
if self.answer_buffer.is_empty() {
self.conversation_history
.add_agent_message(&self.config, message);
} else {
self.conversation_history
.replace_prev_agent_message(&self.config, message);
}
self.answer_buffer.clear();
self.request_redraw();
}
EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
if should_drop_streaming {
return;
}
if self.answer_buffer.is_empty() {
self.conversation_history
.add_agent_message(&self.config, "".to_string());
}
self.answer_buffer.push_str(&delta.clone());
self.conversation_history
.replace_prev_agent_message(&self.config, self.answer_buffer.clone());
self.request_redraw();
}
EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }) => {
if should_drop_streaming {
return;
}
if self.reasoning_buffer.is_empty() {
self.conversation_history
.add_agent_reasoning(&self.config, "".to_string());
}
self.reasoning_buffer.push_str(&delta.clone());
self.conversation_history
.replace_prev_agent_reasoning(&self.config, self.reasoning_buffer.clone());
.add_agent_message(&self.config, message);
self.request_redraw();
}
EventMsg::AgentReasoning(AgentReasoningEvent { text }) => {
if should_drop_streaming {
return;
}
// if the reasoning buffer is empty, this means we haven't received any
// delta. Thus, we need to print the message as a new reasoning.
if self.reasoning_buffer.is_empty() {
if !self.config.hide_agent_reasoning {
self.conversation_history
.add_agent_reasoning(&self.config, "".to_string());
} else {
// else, we rerender one last time.
self.conversation_history
.replace_prev_agent_reasoning(&self.config, text);
.add_agent_reasoning(&self.config, text);
self.request_redraw();
}
self.reasoning_buffer.clear();
self.request_redraw();
}
EventMsg::TaskStarted => {
// New task has begun update state and clear any stale buffers.
self.active_task_id = Some(event_id);
self.answer_buffer.clear();
self.reasoning_buffer.clear();
self.bottom_pane.clear_ctrl_c_quit_hint();
self.bottom_pane.set_task_running(true);
self.request_redraw();
@@ -338,10 +253,6 @@ impl ChatWidget<'_> {
EventMsg::TaskComplete(TaskCompleteEvent {
last_agent_message: _,
}) => {
// Task finished; clear active_task_id so that subsequent events are processed.
if self.active_task_id.as_ref() == Some(&event_id) {
self.active_task_id = None;
}
self.bottom_pane.set_task_running(false);
self.request_redraw();
}
@@ -351,25 +262,16 @@ impl ChatWidget<'_> {
.set_token_usage(self.token_usage.clone(), self.config.model_context_window);
}
EventMsg::Error(ErrorEvent { message }) => {
// Error events always get surfaced (even for stale task IDs) so that the user sees
// why a run stopped. However, only clear the running indicator if this is the
// active task.
if self.active_task_id.as_ref() == Some(&event_id) {
self.bottom_pane.set_task_running(false);
self.active_task_id = None;
}
self.conversation_history.add_error(message);
self.bottom_pane.set_task_running(false);
}
EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent {
command,
cwd,
reason,
}) => {
if should_drop_streaming {
return;
}
let request = ApprovalRequest::Exec {
id: event_id,
id,
command,
cwd,
reason,
@@ -381,9 +283,6 @@ impl ChatWidget<'_> {
reason,
grant_root,
}) => {
if should_drop_streaming {
return;
}
// ------------------------------------------------------------------
// Before we even prompt the user for approval we surface the patch
// summary in the main conversation so that the dialog appears in a
@@ -402,7 +301,7 @@ impl ChatWidget<'_> {
// Now surface the approval request in the BottomPane as before.
let request = ApprovalRequest::ApplyPatch {
id: event_id,
id,
reason,
grant_root,
};
@@ -414,9 +313,6 @@ impl ChatWidget<'_> {
command,
cwd: _,
}) => {
if should_drop_streaming {
return;
}
self.conversation_history
.add_active_exec_command(call_id, command);
self.request_redraw();
@@ -426,9 +322,6 @@ impl ChatWidget<'_> {
auto_approved,
changes,
}) => {
if should_drop_streaming {
return;
}
// Even when a patch is autoapproved we still display the
// summary so the user can follow along.
self.conversation_history
@@ -444,9 +337,6 @@ impl ChatWidget<'_> {
stdout,
stderr,
}) => {
if should_drop_streaming {
return;
}
self.conversation_history
.record_completed_exec_command(call_id, stdout, stderr, exit_code);
self.request_redraw();
@@ -457,17 +347,11 @@ impl ChatWidget<'_> {
tool,
arguments,
}) => {
if should_drop_streaming {
return;
}
self.conversation_history
.add_active_mcp_tool_call(call_id, server, tool, arguments);
self.request_redraw();
}
EventMsg::McpToolCallEnd(mcp_tool_call_end_event) => {
if should_drop_streaming {
return;
}
let success = mcp_tool_call_end_event.is_success();
let McpToolCallEndEvent { call_id, result } = mcp_tool_call_end_event;
self.conversation_history
@@ -500,7 +384,7 @@ impl ChatWidget<'_> {
}
fn request_redraw(&mut self) {
self.app_event_tx.send(AppEvent::RequestRedraw);
self.app_event_tx.send(AppEvent::Redraw);
}
pub(crate) fn add_diff_output(&mut self, diff_output: String) {
@@ -542,10 +426,6 @@ impl ChatWidget<'_> {
}
}
pub(crate) fn composer_is_empty(&self) -> bool {
self.bottom_pane.composer_is_empty()
}
/// Forward an `Op` directly to codex.
pub(crate) fn submit_op(&self, op: Op) {
if let Err(e) = self.codex_op_tx.send(op) {

View File

@@ -202,14 +202,6 @@ impl ConversationHistoryWidget {
self.add_to_history(HistoryCell::new_agent_reasoning(config, text));
}
pub fn replace_prev_agent_reasoning(&mut self, config: &Config, text: String) {
self.replace_last_agent_reasoning(config, text);
}
pub fn replace_prev_agent_message(&mut self, config: &Config, text: String) {
self.replace_last_agent_message(config, text);
}
pub fn add_background_event(&mut self, message: String) {
self.add_to_history(HistoryCell::new_background_event(message));
}
@@ -257,42 +249,6 @@ impl ConversationHistoryWidget {
});
}
pub fn replace_last_agent_reasoning(&mut self, config: &Config, text: String) {
if let Some(idx) = self
.entries
.iter()
.rposition(|entry| matches!(entry.cell, HistoryCell::AgentReasoning { .. }))
{
let width = self.cached_width.get();
let entry = &mut self.entries[idx];
entry.cell = HistoryCell::new_agent_reasoning(config, text);
let height = if width > 0 {
entry.cell.height(width)
} else {
0
};
entry.line_count.set(height);
}
}
pub fn replace_last_agent_message(&mut self, config: &Config, text: String) {
if let Some(idx) = self
.entries
.iter()
.rposition(|entry| matches!(entry.cell, HistoryCell::AgentMessage { .. }))
{
let width = self.cached_width.get();
let entry = &mut self.entries[idx];
entry.cell = HistoryCell::new_agent_message(config, text);
let height = if width > 0 {
entry.cell.height(width)
} else {
0
};
entry.line_count.set(height);
}
}
pub fn record_completed_exec_command(
&mut self,
call_id: String,
@@ -498,7 +454,7 @@ impl WidgetRef for ConversationHistoryWidget {
{
// Choose a thumb color that stands out only when this pane has focus so that the
// user's attention is naturally drawn to the active viewport. When unfocused we show
// users attention is naturally drawn to the active viewport. When unfocused we show
// a low-contrast thumb so the scrollbar fades into the background without becoming
// invisible.
let thumb_style = if self.has_input_focus {

View File

@@ -65,7 +65,7 @@ impl StatusIndicatorWidget {
std::thread::sleep(Duration::from_millis(200));
counter = counter.wrapping_add(1);
frame_idx_clone.store(counter, Ordering::Relaxed);
app_event_tx_clone.send(AppEvent::RequestRedraw);
app_event_tx_clone.send(AppEvent::Redraw);
}
});
}