Compare commits

...

12 Commits

Author SHA1 Message Date
starr-openai
cdd468a909 exec-server: use jsonrpsee for RPC dispatch
Rewrite the transport request/response path to register exec-server methods on a jsonrpsee RpcModule instead of the custom JSON-RPC router/processor, while keeping the in-process client on direct typed handler calls.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 21:35:34 +00:00
starr-openai
3d40df7939 exec-server: make in-process client call handler directly
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 21:14:17 +00:00
starr-openai
9c53cce1d1 exec-server: add in-process client mode
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 20:10:22 +00:00
starr-openai
adfc5aa0c5 codex: address PR review feedback (#14862)
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 18:05:24 +00:00
starr-openai
008d096dc2 Expand exec-server unit test coverage
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 04:12:02 +00:00
starr-openai
4319be2d7c Document exec-server design flow and add lifecycle tests
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 03:21:04 +00:00
starr-openai
63706bc34e refactor(exec-server): split routing from handler
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 02:49:37 +00:00
starr-openai
695b6ab90e refactor(exec-server): tighten client lifecycle and output model
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 02:34:49 +00:00
starr-openai
dc5f035527 test(exec-server): add unit coverage for transport and handshake
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 01:52:06 +00:00
starr-openai
632cd242bc refactor(exec-server): split transports from client launch
Separate the transport-neutral JSON-RPC connection and server processor from
local process spawning, add websocket support, and document the new API
shape.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 01:33:51 +00:00
starr-openai
5ce96df361 docs(exec-server): add protocol README
Document the standalone exec-server crate, its stdio JSON-RPC
transport, and the current request/response and notification
payloads.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 00:37:34 +00:00
starr-openai
8b06c0b555 Add codex-exec-server crate
This adds the standalone exec-server stdio JSON-RPC crate and its
smoke tests without wiring it into the CLI or unified-exec yet.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 00:29:58 +00:00
19 changed files with 4317 additions and 0 deletions

6
MODULE.bazel.lock generated
View File

@@ -955,6 +955,10 @@
"jni_0.21.1": "{\"dependencies\":[{\"name\":\"cesu8\",\"req\":\"^1.1.0\"},{\"name\":\"cfg-if\",\"req\":\"^1.0.0\"},{\"name\":\"combine\",\"req\":\"^4.1.0\"},{\"name\":\"java-locator\",\"optional\":true,\"req\":\"^0.1\"},{\"name\":\"jni-sys\",\"req\":\"^0.3.0\"},{\"name\":\"libloading\",\"optional\":true,\"req\":\"^0.7\"},{\"name\":\"log\",\"req\":\"^0.4.4\"},{\"name\":\"thiserror\",\"req\":\"^1.0.20\"},{\"kind\":\"dev\",\"name\":\"assert_matches\",\"req\":\"^1.5.0\"},{\"kind\":\"dev\",\"name\":\"lazy_static\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"rusty-fork\",\"req\":\"^0.3.0\"},{\"kind\":\"build\",\"name\":\"walkdir\",\"req\":\"^2\"},{\"features\":[\"Win32_Globalization\"],\"name\":\"windows-sys\",\"req\":\"^0.45.0\",\"target\":\"cfg(windows)\"},{\"kind\":\"dev\",\"name\":\"bytemuck\",\"req\":\"^1.13.0\",\"target\":\"cfg(windows)\"}],\"features\":{\"default\":[],\"invocation\":[\"java-locator\",\"libloading\"]}}",
"jobserver_0.1.34": "{\"dependencies\":[{\"features\":[\"std\"],\"name\":\"getrandom\",\"req\":\"^0.3.2\",\"target\":\"cfg(windows)\"},{\"name\":\"libc\",\"req\":\"^0.2.171\",\"target\":\"cfg(unix)\"},{\"features\":[\"fs\"],\"kind\":\"dev\",\"name\":\"nix\",\"req\":\"^0.28.0\",\"target\":\"cfg(unix)\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3.10.1\"}],\"features\":{}}",
"js-sys_0.3.85": "{\"dependencies\":[{\"default_features\":false,\"name\":\"once_cell\",\"req\":\"^1.12\"},{\"default_features\":false,\"name\":\"wasm-bindgen\",\"req\":\"=0.2.108\"}],\"features\":{\"default\":[\"std\"],\"std\":[\"wasm-bindgen/std\"]}}",
"jsonrpsee-core_0.26.0": "{\"dependencies\":[{\"name\":\"async-trait\",\"req\":\"^0.1\"},{\"name\":\"bytes\",\"optional\":true,\"req\":\"^1.6\"},{\"name\":\"futures-timer\",\"optional\":true,\"req\":\"^3\"},{\"default_features\":false,\"features\":[\"alloc\"],\"name\":\"futures-util\",\"optional\":true,\"req\":\"^0.3.14\"},{\"name\":\"http\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"http-body\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"http-body-util\",\"optional\":true,\"req\":\"^0.1.0\"},{\"kind\":\"dev\",\"name\":\"http-body-util\",\"req\":\"^0.1.0\"},{\"name\":\"jsonrpsee-types\",\"req\":\"^0.26.0\"},{\"name\":\"parking_lot\",\"optional\":true,\"req\":\"^0.12\"},{\"name\":\"pin-project\",\"optional\":true,\"req\":\"^1.1.3\"},{\"name\":\"rand\",\"optional\":true,\"req\":\"^0.9\"},{\"name\":\"rustc-hash\",\"optional\":true,\"req\":\"^2\"},{\"default_features\":false,\"features\":[\"derive\"],\"name\":\"serde\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"alloc\",\"raw_value\",\"std\",\"raw_value\"],\"name\":\"serde_json\",\"req\":\"^1.0.142\"},{\"default_features\":false,\"features\":[\"alloc\",\"raw_value\"],\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.142\"},{\"name\":\"thiserror\",\"req\":\"^2\"},{\"name\":\"tokio\",\"optional\":true,\"req\":\"^1.42\"},{\"features\":[\"macros\",\"rt\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1.42\"},{\"name\":\"tokio-stream\",\"optional\":true,\"req\":\"^0.1.7\"},{\"name\":\"tower\",\"optional\":true,\"req\":\"^0.5\"},{\"name\":\"tracing\",\"req\":\"^0.1.34\"},{\"name\":\"wasm-bindgen-futures\",\"optional\":true,\"req\":\"^0.4.19\",\"target\":\"cfg(target_arch = \\\"wasm32\\\")\"}],\"features\":{\"async-client\":[\"client\",\"futures-util\",\"rustc-hash\",\"tokio/macros\",\"tokio/rt\",\"tokio/time\",\"futures-timer\",\"tokio-stream\",\"pin-project\"],\"async-wasm-client\":[\"client\",\"futures-util\",\"wasm-bindgen-futures\",\"rustc-hash/std\",\"futures-timer/wasm-bindgen\",\"tokio/macros\",\"tokio/time\",\"pin-project\"],\"client\":[\"futures-util/sink\",\"tokio/sync\",\"tower\",\"pin-project\",\"http\"],\"default\":[],\"http-helpers\":[\"bytes\",\"futures-util\",\"http-body\",\"http-body-util\",\"http\"],\"server\":[\"futures-util\",\"rustc-hash/std\",\"parking_lot\",\"rand\",\"tokio/rt\",\"tokio/sync\",\"tokio/macros\",\"tokio/time\",\"tower\",\"http\",\"pin-project\"]}}",
"jsonrpsee-server_0.26.0": "{\"dependencies\":[{\"default_features\":false,\"features\":[\"io\",\"async-await-macro\"],\"name\":\"futures-util\",\"req\":\"^0.3.14\"},{\"name\":\"http\",\"req\":\"^1\"},{\"name\":\"http-body\",\"req\":\"^1\"},{\"name\":\"http-body-util\",\"req\":\"^0.1.0\"},{\"features\":[\"server\",\"http1\",\"http2\"],\"name\":\"hyper\",\"req\":\"^1.5\"},{\"features\":[\"tokio\",\"service\",\"tokio\",\"server-auto\"],\"name\":\"hyper-util\",\"req\":\"^0.1\"},{\"features\":[\"server\",\"http-helpers\"],\"name\":\"jsonrpsee-core\",\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-types\",\"req\":\"^0.26.0\"},{\"name\":\"pin-project\",\"req\":\"^1.1.3\"},{\"name\":\"route-recognizer\",\"req\":\"^0.3.1\"},{\"name\":\"serde\",\"req\":\"^1\"},{\"features\":[\"raw_value\"],\"name\":\"serde_json\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"socket2\",\"req\":\"^0.6.0\"},{\"features\":[\"http\"],\"name\":\"soketto\",\"req\":\"^0.8.1\"},{\"name\":\"thiserror\",\"req\":\"^2\"},{\"features\":[\"net\",\"rt-multi-thread\",\"macros\",\"time\"],\"name\":\"tokio\",\"req\":\"^1.23.1\"},{\"features\":[\"sync\"],\"name\":\"tokio-stream\",\"req\":\"^0.1.7\"},{\"features\":[\"compat\"],\"name\":\"tokio-util\",\"req\":\"^0.7\"},{\"features\":[\"util\"],\"name\":\"tower\",\"req\":\"^0.5\"},{\"features\":[\"timeout\"],\"kind\":\"dev\",\"name\":\"tower\",\"req\":\"^0.5\"},{\"name\":\"tracing\",\"req\":\"^0.1.34\"},{\"features\":[\"env-filter\"],\"kind\":\"dev\",\"name\":\"tracing-subscriber\",\"req\":\"^0.3\"}],\"features\":{}}",
"jsonrpsee-types_0.26.0": "{\"dependencies\":[{\"name\":\"http\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"derive\"],\"name\":\"serde\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"alloc\",\"raw_value\"],\"name\":\"serde_json\",\"req\":\"^1.0.142\"},{\"name\":\"thiserror\",\"req\":\"^2\"}],\"features\":{}}",
"jsonrpsee_0.26.0": "{\"dependencies\":[{\"name\":\"jsonrpsee-client-transport\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-core\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-http-client\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-proc-macros\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-server\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-types\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-wasm-client\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"jsonrpsee-ws-client\",\"optional\":true,\"req\":\"^0.26.0\"},{\"name\":\"tokio\",\"optional\":true,\"req\":\"^1.42\"},{\"name\":\"tracing\",\"optional\":true,\"req\":\"^0.1.34\"}],\"features\":{\"async-client\":[\"jsonrpsee-core/async-client\"],\"async-wasm-client\":[\"jsonrpsee-core/async-wasm-client\"],\"client\":[\"http-client\",\"ws-client\",\"wasm-client\",\"client-ws-transport-tls\",\"client-web-transport\",\"async-client\",\"async-wasm-client\",\"client-core\"],\"client-core\":[\"jsonrpsee-core/client\"],\"client-web-transport\":[\"jsonrpsee-client-transport/web\"],\"client-ws-transport-no-tls\":[\"jsonrpsee-client-transport/ws\"],\"client-ws-transport-tls\":[\"jsonrpsee-client-transport/ws\",\"jsonrpsee-client-transport/tls-rustls-platform-verifier\"],\"full\":[\"client\",\"server\",\"macros\"],\"http-client\":[\"jsonrpsee-http-client\",\"jsonrpsee-types\",\"jsonrpsee-core/client\"],\"macros\":[\"jsonrpsee-proc-macros\",\"jsonrpsee-types\",\"tracing\"],\"server\":[\"jsonrpsee-server\",\"server-core\",\"jsonrpsee-types\",\"tokio\"],\"server-core\":[\"jsonrpsee-core/server\"],\"wasm-client\":[\"jsonrpsee-wasm-client\",\"jsonrpsee-types\",\"jsonrpsee-core/client\"],\"ws-client\":[\"jsonrpsee-ws-client\",\"jsonrpsee-types\",\"jsonrpsee-core/client\"]}}",
"keyring_3.6.3": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"base64\",\"req\":\"^0.22\"},{\"name\":\"byteorder\",\"optional\":true,\"req\":\"^1.2\",\"target\":\"cfg(target_os = \\\"windows\\\")\"},{\"features\":[\"derive\",\"wrap_help\"],\"kind\":\"dev\",\"name\":\"clap\",\"req\":\"^4\"},{\"name\":\"dbus-secret-service\",\"optional\":true,\"req\":\"^4.0.0-rc.1\",\"target\":\"cfg(target_os = \\\"openbsd\\\")\"},{\"name\":\"dbus-secret-service\",\"optional\":true,\"req\":\"^4.0.0-rc.2\",\"target\":\"cfg(target_os = \\\"linux\\\")\"},{\"name\":\"dbus-secret-service\",\"optional\":true,\"req\":\"^4.0.1\",\"target\":\"cfg(target_os = \\\"freebsd\\\")\"},{\"kind\":\"dev\",\"name\":\"doc-comment\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\"^0.11.5\"},{\"kind\":\"dev\",\"name\":\"fastrand\",\"req\":\"^2\"},{\"features\":[\"std\"],\"name\":\"linux-keyutils\",\"optional\":true,\"req\":\"^0.2\",\"target\":\"cfg(target_os = \\\"linux\\\")\"},{\"name\":\"log\",\"req\":\"^0.4.22\"},{\"name\":\"openssl\",\"optional\":true,\"req\":\"^0.10.66\"},{\"kind\":\"dev\",\"name\":\"rpassword\",\"req\":\"^7\"},{\"kind\":\"dev\",\"name\":\"rprompt\",\"req\":\"^2\"},{\"name\":\"secret-service\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"freebsd\\\")\"},{\"name\":\"secret-service\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"linux\\\")\"},{\"name\":\"secret-service\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"openbsd\\\")\"},{\"name\":\"security-framework\",\"optional\":true,\"req\":\"^2\",\"target\":\"cfg(target_os = \\\"ios\\\")\"},{\"name\":\"security-framework\",\"optional\":true,\"req\":\"^3\",\"target\":\"cfg(target_os = \\\"macos\\\")\"},{\"kind\":\"dev\",\"name\":\"whoami\",\"req\":\"^1.5\"},{\"features\":[\"Win32_Foundation\",\"Win32_Security_Credentials\"],\"name\":\"windows-sys\",\"optional\":true,\"req\":\"^0.60\",\"target\":\"cfg(target_os = \\\"windows\\\")\"},{\"name\":\"zbus\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"freebsd\\\")\"},{\"name\":\"zbus\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"linux\\\")\"},{\"name\":\"zbus\",\"optional\":true,\"req\":\"^4\",\"target\":\"cfg(target_os = \\\"openbsd\\\")\"},{\"name\":\"zeroize\",\"req\":\"^1.8.1\",\"target\":\"cfg(target_os = \\\"windows\\\")\"}],\"features\":{\"apple-native\":[\"dep:security-framework\"],\"async-io\":[\"zbus?/async-io\"],\"async-secret-service\":[\"dep:secret-service\",\"dep:zbus\"],\"crypto-openssl\":[\"dbus-secret-service?/crypto-openssl\",\"secret-service?/crypto-openssl\"],\"crypto-rust\":[\"dbus-secret-service?/crypto-rust\",\"secret-service?/crypto-rust\"],\"linux-native\":[\"dep:linux-keyutils\"],\"linux-native-async-persistent\":[\"linux-native\",\"async-secret-service\"],\"linux-native-sync-persistent\":[\"linux-native\",\"sync-secret-service\"],\"sync-secret-service\":[\"dep:dbus-secret-service\"],\"tokio\":[\"zbus?/tokio\"],\"vendored\":[\"dbus-secret-service?/vendored\",\"openssl?/vendored\"],\"windows-native\":[\"dep:windows-sys\",\"dep:byteorder\"]}}",
"kqueue-sys_1.0.4": "{\"dependencies\":[{\"name\":\"bitflags\",\"req\":\"^1.2.1\"},{\"name\":\"libc\",\"req\":\"^0.2.74\"}],\"features\":{}}",
"kqueue_1.1.1": "{\"dependencies\":[{\"features\":[\"html_reports\"],\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.5\"},{\"kind\":\"dev\",\"name\":\"dhat\",\"req\":\"^0.3.2\"},{\"name\":\"kqueue-sys\",\"req\":\"^1.0.4\"},{\"name\":\"libc\",\"req\":\"^0.2.17\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3.1.0\"}],\"features\":{}}",
@@ -1180,6 +1184,7 @@
"ring_0.17.14": "{\"dependencies\":[{\"default_features\":false,\"kind\":\"build\",\"name\":\"cc\",\"req\":\"^1.2.8\"},{\"default_features\":false,\"name\":\"cfg-if\",\"req\":\"^1.0.0\"},{\"name\":\"getrandom\",\"req\":\"^0.2.10\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2.148\",\"target\":\"cfg(all(any(all(target_arch = \\\"aarch64\\\", target_endian = \\\"little\\\"), all(target_arch = \\\"arm\\\", target_endian = \\\"little\\\")), any(target_os = \\\"android\\\", target_os = \\\"linux\\\")))\"},{\"default_features\":false,\"name\":\"libc\",\"req\":\"^0.2.155\",\"target\":\"cfg(all(all(target_arch = \\\"aarch64\\\", target_endian = \\\"little\\\"), target_vendor = \\\"apple\\\", any(target_os = \\\"ios\\\", target_os = \\\"macos\\\", target_os = \\\"tvos\\\", target_os = \\\"visionos\\\", target_os = \\\"watchos\\\")))\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"libc\",\"req\":\"^0.2.148\",\"target\":\"cfg(any(unix, windows, target_os = \\\"wasi\\\"))\"},{\"name\":\"untrusted\",\"req\":\"^0.9\"},{\"default_features\":false,\"features\":[\"std\"],\"kind\":\"dev\",\"name\":\"wasm-bindgen-test\",\"req\":\"^0.3.37\",\"target\":\"cfg(all(target_arch = \\\"wasm32\\\", target_os = \\\"unknown\\\"))\"},{\"features\":[\"Win32_Foundation\",\"Win32_System_Threading\"],\"name\":\"windows-sys\",\"req\":\"^0.52\",\"target\":\"cfg(all(all(target_arch = \\\"aarch64\\\", target_endian = \\\"little\\\"), target_os = \\\"windows\\\"))\"}],\"features\":{\"alloc\":[],\"default\":[\"alloc\",\"dev_urandom_fallback\"],\"dev_urandom_fallback\":[],\"less-safe-getrandom-custom-or-rdrand\":[],\"less-safe-getrandom-espidf\":[],\"slow_tests\":[],\"std\":[\"alloc\"],\"test_logging\":[],\"unstable-testing-arm-no-hw\":[],\"unstable-testing-arm-no-neon\":[],\"wasm32_unknown_unknown_js\":[\"getrandom/js\"]}}",
"rmcp-macros_0.15.0": "{\"dependencies\":[{\"name\":\"darling\",\"req\":\"^0.23\"},{\"name\":\"proc-macro2\",\"req\":\"^1\"},{\"name\":\"quote\",\"req\":\"^1\"},{\"name\":\"serde_json\",\"req\":\"^1.0\"},{\"features\":[\"full\"],\"name\":\"syn\",\"req\":\"^2\"}],\"features\":{}}",
"rmcp_0.15.0": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"anyhow\",\"req\":\"^1.0\"},{\"name\":\"async-trait\",\"req\":\"^0.1.89\"},{\"kind\":\"dev\",\"name\":\"async-trait\",\"req\":\"^0.1\"},{\"name\":\"axum\",\"optional\":true,\"req\":\"^0.8\"},{\"name\":\"base64\",\"optional\":true,\"req\":\"^0.22\"},{\"name\":\"bytes\",\"optional\":true,\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"serde\",\"clock\",\"std\",\"oldtime\"],\"name\":\"chrono\",\"req\":\"^0.4.38\",\"target\":\"cfg(all(target_family = \\\"wasm\\\", target_os = \\\"unknown\\\"))\"},{\"features\":[\"serde\"],\"name\":\"chrono\",\"req\":\"^0.4.38\",\"target\":\"cfg(not(all(target_family = \\\"wasm\\\", target_os = \\\"unknown\\\")))\"},{\"name\":\"futures\",\"req\":\"^0.3\"},{\"name\":\"http\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"http-body\",\"optional\":true,\"req\":\"^1\"},{\"name\":\"http-body-util\",\"optional\":true,\"req\":\"^0.1\"},{\"default_features\":false,\"features\":[\"reqwest\"],\"name\":\"oauth2\",\"optional\":true,\"req\":\"^5.0\"},{\"name\":\"pastey\",\"optional\":true,\"req\":\"^0.2.0\"},{\"name\":\"pin-project-lite\",\"req\":\"^0.2\"},{\"features\":[\"tokio1\"],\"name\":\"process-wrap\",\"optional\":true,\"req\":\"^9.0\"},{\"name\":\"rand\",\"optional\":true,\"req\":\"^0.9\"},{\"default_features\":false,\"features\":[\"json\",\"stream\"],\"name\":\"reqwest\",\"optional\":true,\"req\":\"^0.12\"},{\"name\":\"rmcp-macros\",\"optional\":true,\"req\":\"^0.15.0\"},{\"features\":[\"chrono04\"],\"name\":\"schemars\",\"optional\":true,\"req\":\"^1.0\"},{\"features\":[\"chrono04\"],\"kind\":\"dev\",\"name\":\"schemars\",\"req\":\"^1.1.0\"},{\"features\":[\"derive\",\"rc\"],\"name\":\"serde\",\"req\":\"^1.0\"},{\"name\":\"serde_json\",\"req\":\"^1.0\"},{\"name\":\"sse-stream\",\"optional\":true,\"req\":\"^0.2\"},{\"name\":\"thiserror\",\"req\":\"^2\"},{\"features\":[\"sync\",\"macros\",\"rt\",\"time\"],\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"full\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"name\":\"tokio-stream\",\"optional\":true,\"req\":\"^0.1\"},{\"name\":\"tokio-util\",\"req\":\"^0.7\"},{\"name\":\"tower-service\",\"optional\":true,\"req\":\"^0.3\"},{\"name\":\"tracing\",\"req\":\"^0.1\"},{\"features\":[\"env-filter\",\"std\",\"fmt\"],\"kind\":\"dev\",\"name\":\"tracing-subscriber\",\"req\":\"^0.3\"},{\"name\":\"url\",\"optional\":true,\"req\":\"^2.4\"},{\"features\":[\"v4\"],\"name\":\"uuid\",\"optional\":true,\"req\":\"^1\"}],\"features\":{\"__reqwest\":[\"dep:reqwest\"],\"auth\":[\"dep:oauth2\",\"__reqwest\",\"dep:url\"],\"client\":[\"dep:tokio-stream\"],\"client-side-sse\":[\"dep:sse-stream\",\"dep:http\"],\"default\":[\"base64\",\"macros\",\"server\"],\"elicitation\":[\"dep:url\"],\"macros\":[\"dep:rmcp-macros\",\"dep:pastey\"],\"reqwest\":[\"__reqwest\",\"reqwest?/rustls-tls\"],\"reqwest-native-tls\":[\"__reqwest\",\"reqwest?/native-tls\"],\"reqwest-tls-no-provider\":[\"__reqwest\",\"reqwest?/rustls-tls-no-provider\"],\"schemars\":[\"dep:schemars\"],\"server\":[\"transport-async-rw\",\"dep:schemars\",\"dep:pastey\"],\"server-side-http\":[\"uuid\",\"dep:rand\",\"dep:tokio-stream\",\"dep:http\",\"dep:http-body\",\"dep:http-body-util\",\"dep:bytes\",\"dep:sse-stream\",\"dep:axum\",\"tower\"],\"tower\":[\"dep:tower-service\"],\"transport-async-rw\":[\"tokio/io-util\",\"tokio-util/codec\"],\"transport-child-process\":[\"transport-async-rw\",\"tokio/process\",\"dep:process-wrap\"],\"transport-io\":[\"transport-async-rw\",\"tokio/io-std\"],\"transport-streamable-http-client\":[\"client-side-sse\",\"transport-worker\"],\"transport-streamable-http-client-reqwest\":[\"transport-streamable-http-client\",\"__reqwest\"],\"transport-streamable-http-server\":[\"transport-streamable-http-server-session\",\"server-side-http\",\"transport-worker\"],\"transport-streamable-http-server-session\":[\"transport-async-rw\",\"dep:tokio-stream\"],\"transport-worker\":[\"dep:tokio-stream\"]}}",
"route-recognizer_0.3.1": "{\"dependencies\":[],\"features\":{}}",
"rsa_0.9.10": "{\"dependencies\":[{\"features\":[\"alloc\"],\"kind\":\"dev\",\"name\":\"base64ct\",\"req\":\"^1\"},{\"default_features\":false,\"name\":\"const-oid\",\"req\":\"^0.9\"},{\"default_features\":false,\"features\":[\"alloc\",\"oid\"],\"name\":\"digest\",\"req\":\"^0.10.5\"},{\"kind\":\"dev\",\"name\":\"hex-literal\",\"req\":\"^0.4.1\"},{\"default_features\":false,\"features\":[\"i128\",\"prime\",\"zeroize\"],\"name\":\"num-bigint\",\"package\":\"num-bigint-dig\",\"req\":\"^0.8.6\"},{\"default_features\":false,\"name\":\"num-integer\",\"req\":\"^0.1.39\"},{\"default_features\":false,\"features\":[\"libm\"],\"name\":\"num-traits\",\"req\":\"^0.2.9\"},{\"default_features\":false,\"features\":[\"alloc\",\"pkcs8\"],\"name\":\"pkcs1\",\"req\":\"^0.7.5\"},{\"default_features\":false,\"features\":[\"alloc\"],\"name\":\"pkcs8\",\"req\":\"^0.10.2\"},{\"kind\":\"dev\",\"name\":\"proptest\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8\"},{\"kind\":\"dev\",\"name\":\"rand_chacha\",\"req\":\"^0.3\"},{\"default_features\":false,\"name\":\"rand_core\",\"req\":\"^0.6.4\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"rand_core\",\"req\":\"^0.6\"},{\"kind\":\"dev\",\"name\":\"rand_xorshift\",\"req\":\"^0.3\"},{\"default_features\":false,\"features\":[\"derive\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0.184\"},{\"kind\":\"dev\",\"name\":\"serde_test\",\"req\":\"^1.0.89\"},{\"default_features\":false,\"features\":[\"oid\"],\"name\":\"sha1\",\"optional\":true,\"req\":\"^0.10.5\"},{\"default_features\":false,\"features\":[\"oid\"],\"kind\":\"dev\",\"name\":\"sha1\",\"req\":\"^0.10.5\"},{\"default_features\":false,\"features\":[\"oid\"],\"name\":\"sha2\",\"optional\":true,\"req\":\"^0.10.6\"},{\"default_features\":false,\"features\":[\"oid\"],\"kind\":\"dev\",\"name\":\"sha2\",\"req\":\"^0.10.6\"},{\"default_features\":false,\"features\":[\"oid\"],\"kind\":\"dev\",\"name\":\"sha3\",\"req\":\"^0.10.7\"},{\"default_features\":false,\"features\":[\"alloc\",\"digest\",\"rand_core\"],\"name\":\"signature\",\"req\":\">2.0, <2.3\"},{\"default_features\":false,\"features\":[\"alloc\"],\"name\":\"spki\",\"req\":\"^0.7.3\"},{\"default_features\":false,\"name\":\"subtle\",\"req\":\"^2.1.1\"},{\"features\":[\"alloc\"],\"name\":\"zeroize\",\"req\":\"^1.5\"}],\"features\":{\"default\":[\"std\",\"pem\",\"u64_digit\"],\"getrandom\":[\"rand_core/getrandom\"],\"hazmat\":[],\"nightly\":[\"num-bigint/nightly\"],\"pem\":[\"pkcs1/pem\",\"pkcs8/pem\"],\"pkcs5\":[\"pkcs8/encryption\"],\"serde\":[\"dep:serde\",\"num-bigint/serde\"],\"std\":[\"digest/std\",\"pkcs1/std\",\"pkcs8/std\",\"rand_core/std\",\"signature/std\"],\"u64_digit\":[\"num-bigint/u64_digit\"]}}",
"rust-embed-impl_8.11.0": "{\"dependencies\":[{\"name\":\"proc-macro2\",\"req\":\"^1\"},{\"name\":\"quote\",\"req\":\"^1\"},{\"name\":\"rust-embed-utils\",\"req\":\"^8.11.0\"},{\"name\":\"shellexpand\",\"optional\":true,\"req\":\"^3\"},{\"default_features\":false,\"features\":[\"derive\",\"parsing\",\"proc-macro\",\"printing\"],\"name\":\"syn\",\"req\":\"^2\"},{\"name\":\"walkdir\",\"req\":\"^2.3.1\"}],\"features\":{\"compression\":[],\"debug-embed\":[],\"deterministic-timestamps\":[],\"include-exclude\":[\"rust-embed-utils/include-exclude\"],\"interpolate-folder-path\":[\"shellexpand\"],\"mime-guess\":[\"rust-embed-utils/mime-guess\"]}}",
"rust-embed-utils_8.11.0": "{\"dependencies\":[{\"name\":\"globset\",\"optional\":true,\"req\":\"^0.4.8\"},{\"name\":\"mime_guess\",\"optional\":true,\"req\":\"^2.0.4\"},{\"name\":\"sha2\",\"req\":\"^0.10.5\"},{\"name\":\"walkdir\",\"req\":\"^2.3.1\"}],\"features\":{\"debug-embed\":[],\"include-exclude\":[\"globset\"],\"mime-guess\":[\"mime_guess\"]}}",
@@ -1270,6 +1275,7 @@
"smol_str_0.3.5": "{\"dependencies\":[{\"name\":\"arbitrary\",\"optional\":true,\"req\":\"^1.3\"},{\"default_features\":false,\"name\":\"borsh\",\"optional\":true,\"req\":\"^1.4.0\"},{\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.7\"},{\"kind\":\"dev\",\"name\":\"proptest\",\"req\":\"^1.5\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.9.2\"},{\"features\":[\"derive\"],\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0\"},{\"default_features\":false,\"name\":\"serde_core\",\"optional\":true,\"req\":\"^1.0.220\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0\"}],\"features\":{\"default\":[\"std\"],\"serde\":[\"dep:serde_core\"],\"std\":[\"serde_core?/std\",\"borsh?/std\"]}}",
"socket2_0.5.10": "{\"dependencies\":[{\"name\":\"libc\",\"req\":\"^0.2.171\",\"target\":\"cfg(unix)\"},{\"features\":[\"Win32_Foundation\",\"Win32_Networking_WinSock\",\"Win32_System_IO\",\"Win32_System_Threading\",\"Win32_System_WindowsProgramming\"],\"name\":\"windows-sys\",\"req\":\"^0.52\",\"target\":\"cfg(windows)\"}],\"features\":{\"all\":[]}}",
"socket2_0.6.2": "{\"dependencies\":[{\"name\":\"libc\",\"req\":\"^0.2.172\",\"target\":\"cfg(unix)\"},{\"features\":[\"Win32_Foundation\",\"Win32_Networking_WinSock\",\"Win32_System_IO\",\"Win32_System_Threading\",\"Win32_System_WindowsProgramming\"],\"name\":\"windows-sys\",\"req\":\"^0.60\",\"target\":\"cfg(windows)\"}],\"features\":{\"all\":[]}}",
"soketto_0.8.1": "{\"dependencies\":[{\"default_features\":false,\"features\":[\"alloc\"],\"name\":\"base64\",\"req\":\"^0.22\"},{\"default_features\":false,\"name\":\"bytes\",\"req\":\"^1.0\"},{\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\"^0.11.1\"},{\"default_features\":false,\"features\":[\"zlib\"],\"name\":\"flate2\",\"optional\":true,\"req\":\"^1.0.13\"},{\"default_features\":false,\"features\":[\"bilock\",\"std\",\"unstable\"],\"name\":\"futures\",\"req\":\"^0.3.1\"},{\"name\":\"http\",\"optional\":true,\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"http-body-util\",\"req\":\"^0.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"httparse\",\"req\":\"^1.3.4\"},{\"features\":[\"full\"],\"kind\":\"dev\",\"name\":\"hyper\",\"req\":\"^1.2\"},{\"features\":[\"tokio\"],\"kind\":\"dev\",\"name\":\"hyper-util\",\"req\":\"^0.1\"},{\"default_features\":false,\"name\":\"log\",\"req\":\"^0.4.8\"},{\"kind\":\"dev\",\"name\":\"quickcheck\",\"req\":\"^1\"},{\"default_features\":false,\"features\":[\"std\",\"std_rng\"],\"name\":\"rand\",\"req\":\"^0.8\"},{\"default_features\":false,\"name\":\"sha1\",\"req\":\"^0.10\"},{\"features\":[\"full\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"net\"],\"kind\":\"dev\",\"name\":\"tokio-stream\",\"req\":\"^0.1\"},{\"features\":[\"compat\"],\"kind\":\"dev\",\"name\":\"tokio-util\",\"req\":\"^0.7\"}],\"features\":{\"default\":[],\"deflate\":[\"flate2\"]}}",
"spin_0.9.8": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.4\"},{\"name\":\"lock_api_crate\",\"optional\":true,\"package\":\"lock_api\",\"req\":\"^0.4\"},{\"default_features\":false,\"name\":\"portable-atomic\",\"optional\":true,\"req\":\"^1\"}],\"features\":{\"barrier\":[\"mutex\"],\"default\":[\"lock_api\",\"mutex\",\"spin_mutex\",\"rwlock\",\"once\",\"lazy\",\"barrier\"],\"fair_mutex\":[\"mutex\"],\"lazy\":[\"once\"],\"lock_api\":[\"lock_api_crate\"],\"mutex\":[],\"once\":[],\"portable_atomic\":[\"portable-atomic\"],\"rwlock\":[],\"spin_mutex\":[\"mutex\"],\"std\":[],\"ticket_mutex\":[\"mutex\"],\"use_ticket_mutex\":[\"mutex\",\"ticket_mutex\"]}}",
"spki_0.7.3": "{\"dependencies\":[{\"features\":[\"derive\"],\"name\":\"arbitrary\",\"optional\":true,\"req\":\"^1.2\"},{\"default_features\":false,\"name\":\"base64ct\",\"optional\":true,\"req\":\"^1\"},{\"features\":[\"oid\"],\"name\":\"der\",\"req\":\"^0.7.2\"},{\"kind\":\"dev\",\"name\":\"hex-literal\",\"req\":\"^0.4\"},{\"default_features\":false,\"name\":\"sha2\",\"optional\":true,\"req\":\"^0.10\"},{\"kind\":\"dev\",\"name\":\"tempfile\",\"req\":\"^3\"}],\"features\":{\"alloc\":[\"base64ct?/alloc\",\"der/alloc\"],\"arbitrary\":[\"std\",\"dep:arbitrary\",\"der/arbitrary\"],\"base64\":[\"dep:base64ct\"],\"fingerprint\":[\"sha2\"],\"pem\":[\"alloc\",\"der/pem\"],\"std\":[\"der/std\",\"alloc\"]}}",
"sqlx-core_0.8.6": "{\"dependencies\":[{\"name\":\"async-io\",\"optional\":true,\"req\":\"^1.9.0\"},{\"name\":\"async-std\",\"optional\":true,\"req\":\"^1.12\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"base64\",\"req\":\"^0.22.0\"},{\"name\":\"bigdecimal\",\"optional\":true,\"req\":\"^0.4.0\"},{\"name\":\"bit-vec\",\"optional\":true,\"req\":\"^0.6.3\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"bstr\",\"optional\":true,\"req\":\"^1.0\"},{\"name\":\"bytes\",\"req\":\"^1.1.0\"},{\"default_features\":false,\"features\":[\"clock\"],\"name\":\"chrono\",\"optional\":true,\"req\":\"^0.4.34\"},{\"name\":\"crc\",\"optional\":true,\"req\":\"^3\"},{\"name\":\"crossbeam-queue\",\"req\":\"^0.3.2\"},{\"name\":\"either\",\"req\":\"^1.6.1\"},{\"name\":\"event-listener\",\"req\":\"^5.2.0\"},{\"default_features\":false,\"name\":\"futures-core\",\"req\":\"^0.3.19\"},{\"name\":\"futures-intrusive\",\"req\":\"^0.5.0\"},{\"name\":\"futures-io\",\"req\":\"^0.3.24\"},{\"default_features\":false,\"features\":[\"alloc\",\"sink\",\"io\"],\"name\":\"futures-util\",\"req\":\"^0.3.19\"},{\"name\":\"hashbrown\",\"req\":\"^0.15.0\"},{\"name\":\"hashlink\",\"req\":\"^0.10.0\"},{\"name\":\"indexmap\",\"req\":\"^2.0\"},{\"name\":\"ipnet\",\"optional\":true,\"req\":\"^2.3.0\"},{\"name\":\"ipnetwork\",\"optional\":true,\"req\":\"^0.20.0\"},{\"default_features\":false,\"name\":\"log\",\"req\":\"^0.4.18\"},{\"name\":\"mac_address\",\"optional\":true,\"req\":\"^1.1.5\"},{\"default_features\":false,\"name\":\"memchr\",\"req\":\"^2.4.1\"},{\"name\":\"native-tls\",\"optional\":true,\"req\":\"^0.2.10\"},{\"name\":\"once_cell\",\"req\":\"^1.9.0\"},{\"name\":\"percent-encoding\",\"req\":\"^2.1.0\"},{\"name\":\"regex\",\"optional\":true,\"req\":\"^1.5.5\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"rust_decimal\",\"optional\":true,\"req\":\"^1.26.1\"},{\"default_features\":false,\"features\":[\"std\",\"tls12\"],\"name\":\"rustls\",\"optional\":true,\"req\":\"^0.23.15\"},{\"name\":\"rustls-native-certs\",\"optional\":true,\"req\":\"^0.8.0\"},{\"features\":[\"derive\",\"rc\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0.132\"},{\"features\":[\"raw_value\"],\"name\":\"serde_json\",\"optional\":true,\"req\":\"^1.0.73\"},{\"default_features\":false,\"name\":\"sha2\",\"optional\":true,\"req\":\"^0.10.0\"},{\"name\":\"smallvec\",\"req\":\"^1.7.0\"},{\"default_features\":false,\"features\":[\"postgres\",\"sqlite\",\"mysql\",\"migrate\",\"macros\",\"time\",\"uuid\"],\"kind\":\"dev\",\"name\":\"sqlx\",\"req\":\"=0.8.6\"},{\"name\":\"thiserror\",\"req\":\"^2.0.0\"},{\"features\":[\"formatting\",\"parsing\",\"macros\"],\"name\":\"time\",\"optional\":true,\"req\":\"^0.3.36\"},{\"default_features\":false,\"features\":[\"time\",\"net\",\"sync\",\"fs\",\"io-util\",\"rt\"],\"name\":\"tokio\",\"optional\":true,\"req\":\"^1\"},{\"features\":[\"rt\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"fs\"],\"name\":\"tokio-stream\",\"optional\":true,\"req\":\"^0.1.8\"},{\"features\":[\"log\"],\"name\":\"tracing\",\"req\":\"^0.1.37\"},{\"name\":\"url\",\"req\":\"^2.2.2\"},{\"name\":\"uuid\",\"optional\":true,\"req\":\"^1.1.2\"},{\"name\":\"webpki-roots\",\"optional\":true,\"req\":\"^0.26\"}],\"features\":{\"_rt-async-std\":[\"async-std\",\"async-io\"],\"_rt-tokio\":[\"tokio\",\"tokio-stream\"],\"_tls-native-tls\":[\"native-tls\"],\"_tls-none\":[],\"_tls-rustls\":[\"rustls\"],\"_tls-rustls-aws-lc-rs\":[\"_tls-rustls\",\"rustls/aws-lc-rs\",\"webpki-roots\"],\"_tls-rustls-ring-native-roots\":[\"_tls-rustls\",\"rustls/ring\",\"rustls-native-certs\"],\"_tls-rustls-ring-webpki\":[\"_tls-rustls\",\"rustls/ring\",\"webpki-roots\"],\"any\":[],\"default\":[],\"json\":[\"serde\",\"serde_json\"],\"migrate\":[\"sha2\",\"crc\"],\"offline\":[\"serde\",\"either/serde\"]}}",

120
codex-rs/Cargo.lock generated
View File

@@ -1989,6 +1989,27 @@ dependencies = [
"wiremock",
]
[[package]]
name = "codex-exec-server"
version = "0.0.0"
dependencies = [
"anyhow",
"base64 0.22.1",
"clap",
"codex-app-server-protocol",
"codex-utils-cargo-bin",
"codex-utils-pty",
"futures",
"jsonrpsee",
"pretty_assertions",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tokio-tungstenite",
"tracing",
]
[[package]]
name = "codex-execpolicy"
version = "0.0.0"
@@ -5430,6 +5451,82 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "jsonrpsee"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f3f48dc3e6b8bd21e15436c1ddd0bc22a6a54e8ec46fedd6adf3425f396ec6a"
dependencies = [
"jsonrpsee-core",
"jsonrpsee-server",
"jsonrpsee-types",
"tokio",
]
[[package]]
name = "jsonrpsee-core"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "316c96719901f05d1137f19ba598b5fe9c9bc39f4335f67f6be8613921946480"
dependencies = [
"async-trait",
"bytes",
"futures-util",
"http 1.4.0",
"http-body",
"http-body-util",
"jsonrpsee-types",
"parking_lot",
"pin-project",
"rand 0.9.2",
"rustc-hash 2.1.1",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tower",
"tracing",
]
[[package]]
name = "jsonrpsee-server"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4c51b7c290bb68ce3af2d029648148403863b982f138484a73f02a9dd52dbd7f"
dependencies = [
"futures-util",
"http 1.4.0",
"http-body",
"http-body-util",
"hyper",
"hyper-util",
"jsonrpsee-core",
"jsonrpsee-types",
"pin-project",
"route-recognizer",
"serde",
"serde_json",
"soketto",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
"tokio-util",
"tower",
"tracing",
]
[[package]]
name = "jsonrpsee-types"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bc88ff4688e43cc3fa9883a8a95c6fa27aa2e76c96e610b737b6554d650d7fd5"
dependencies = [
"http 1.4.0",
"serde",
"serde_json",
"thiserror 2.0.18",
]
[[package]]
name = "keyring"
version = "3.6.3"
@@ -7989,6 +8086,12 @@ dependencies = [
"syn 2.0.114",
]
[[package]]
name = "route-recognizer"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "afab94fb28594581f62d981211a9a4d53cc8130bbcbbb89a0440d9b8e81a7746"
[[package]]
name = "rsa"
version = "0.9.10"
@@ -8955,6 +9058,22 @@ dependencies = [
"windows-sys 0.60.2",
]
[[package]]
name = "soketto"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e859df029d160cb88608f5d7df7fb4753fd20fdfb4de5644f3d8b8440841721"
dependencies = [
"base64 0.22.1",
"bytes",
"futures",
"http 1.4.0",
"httparse",
"log",
"rand 0.8.5",
"sha1",
]
[[package]]
name = "spin"
version = "0.9.8"
@@ -9890,6 +10009,7 @@ checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
dependencies = [
"bytes",
"futures-core",
"futures-io",
"futures-sink",
"futures-util",
"pin-project-lite",

View File

@@ -25,6 +25,7 @@ members = [
"hooks",
"secrets",
"exec",
"exec-server",
"execpolicy",
"execpolicy-legacy",
"keyring-store",
@@ -273,6 +274,7 @@ starlark = "0.13.0"
strum = "0.27.2"
strum_macros = "0.28.0"
supports-color = "3.0.2"
jsonrpsee = { version = "0.26.0", features = ["server"] }
syntect = "5"
sys-locale = "0.3.2"
tempfile = "3.23.0"

View File

@@ -0,0 +1,40 @@
[package]
name = "codex-exec-server"
version.workspace = true
edition.workspace = true
license.workspace = true
[[bin]]
name = "codex-exec-server"
path = "src/bin/codex-exec-server.rs"
[lints]
workspace = true
[dependencies]
base64 = { workspace = true }
clap = { workspace = true, features = ["derive"] }
codex-app-server-protocol = { workspace = true }
codex-utils-pty = { workspace = true }
futures = { workspace = true }
jsonrpsee = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = [
"io-std",
"io-util",
"macros",
"net",
"process",
"rt-multi-thread",
"sync",
"time",
] }
tokio-tungstenite = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
anyhow = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
pretty_assertions = { workspace = true }

View File

@@ -0,0 +1,242 @@
# exec-server design notes
This document sketches a likely direction for integrating `codex-exec-server`
with unified exec without baking the full tool-call policy stack into the
server.
The goal is:
- keep exec-server generic and reusable
- keep approval, sandbox, and retry policy in `core`
- preserve the unified-exec event flow the model already depends on
- support retained output caps so polling and snapshot-style APIs do not grow
memory without bound
## Unified exec today
Today the flow for LLM-visible interactive execution is:
1. The model sees the `exec_command` and `write_stdin` tools.
2. `UnifiedExecHandler` parses the tool arguments and allocates a process id.
3. `UnifiedExecProcessManager::exec_command(...)` calls
`open_session_with_sandbox(...)`.
4. `ToolOrchestrator` drives approval, sandbox selection, managed network
approval, and sandbox-denial retry behavior.
5. `UnifiedExecRuntime` builds a `CommandSpec`, asks the current
`SandboxAttempt` to transform it into an `ExecRequest`, and passes that
resolved request back to the process manager.
6. `open_session_with_exec_env(...)` spawns the process from that resolved
`ExecRequest`.
7. Unified exec emits an `ExecCommandBegin` event.
8. Unified exec starts a background output watcher that emits
`ExecCommandOutputDelta` events.
9. The initial tool call collects output until the requested yield deadline and
returns an `ExecCommandToolOutput` snapshot to the model.
10. If the process is still running, unified exec stores it and later emits
`ExecCommandEnd` when the exit watcher fires.
11. A later `write_stdin` tool call writes to the stored process, emits a
`TerminalInteraction` event, collects another bounded snapshot, and returns
that tool response to the model.
Important observation: the 250ms / 10s yield-window behavior is not really a
process-server concern. It is a client-side convenience layer for the LLM tool
API. The server should focus on raw process lifecycle and streaming events.
## Proposed boundary
The clean split is:
- exec-server server: process lifecycle, output streaming, retained output caps
- exec-server client: `wait`, `communicate`, yield-window helpers, session
bookkeeping
- unified exec in `core`: tool parsing, event emission, approvals, sandboxing,
managed networking, retry semantics
If exec-server is used by unified exec later, the boundary should sit between
step 5 and step 6 above: after policy has produced a resolved spawn request, but
before the actual PTY or pipe spawn.
## Suggested process API
Start simple and explicit:
- `process/start`
- `process/write`
- `process/closeStdin`
- `process/resize`
- `process/terminate`
- `process/wait`
- `process/snapshot`
Server notifications:
- `process/output`
- `process/exited`
- optionally `process/started`
- optionally `process/failed`
Suggested request shapes:
```rust
enum ProcessStartRequest {
Direct(DirectExecSpec),
Prepared(PreparedExecSpec),
}
struct DirectExecSpec {
process_id: String,
argv: Vec<String>,
cwd: PathBuf,
env: HashMap<String, String>,
arg0: Option<String>,
io: ProcessIo,
}
struct PreparedExecSpec {
process_id: String,
request: PreparedExecRequest,
io: ProcessIo,
}
enum ProcessIo {
Pty { rows: u16, cols: u16 },
Pipe { stdin: StdinMode },
}
enum StdinMode {
Open,
Closed,
}
enum TerminateMode {
Graceful { timeout_ms: u64 },
Force,
}
```
Notes:
- `processId` remains a protocol handle, not an OS pid.
- `wait` is a good generic API because many callers want process completion
without manually wiring notifications.
- `communicate` is also a reasonable API, but it should probably start as a
client helper built on top of `write + closeStdin + wait + snapshot`.
- If an RPC form of `communicate` is added later, it should be a convenience
wrapper rather than the primitive execution model.
## Output capping
Even with event streaming, the server should retain a bounded amount of output
per process so callers can poll, wait, or reconnect without unbounded memory
growth.
Suggested behavior:
- stream every output chunk live via `process/output`
- retain capped output per process in memory
- keep stdout and stderr separately for pipe-backed processes
- for PTY-backed processes, treat retained output as a single terminal stream
- expose truncation metadata on snapshots
Suggested snapshot response:
```rust
struct ProcessSnapshot {
stdout: Vec<u8>,
stderr: Vec<u8>,
terminal: Vec<u8>,
truncated: bool,
exit_code: Option<i32>,
running: bool,
}
```
Implementation-wise, the current `HeadTailBuffer` pattern used by unified exec
is a good fit. The cap should be server config, not request config, so memory
use stays predictable.
## Sandboxing and networking
### How unified exec does it today
Unified exec does not hand raw command args directly to the PTY layer for tool
calls. Instead, it:
1. computes approval requirements
2. chooses a sandbox attempt
3. applies managed-network policy if needed
4. transforms `CommandSpec` into `ExecRequest`
5. spawns from that resolved `ExecRequest`
That split is already valuable and should be preserved.
### Recommended exec-server design
Do not put approval policy into exec-server.
Instead, support two execution modes:
- `Direct`: raw command, intended for orchestrator-side or already-trusted use
- `Prepared`: already-resolved spawn request, intended for tool-call execution
For tool calls from the LLM side:
1. `core` runs the existing approval + sandbox + managed-network flow
2. `core` produces a resolved `ExecRequest`
3. the exec-server client sends `PreparedExecSpec`
4. exec-server spawns exactly that request and streams process events
For orchestrator-side execution:
1. caller sends `DirectExecSpec`
2. exec-server spawns directly without running approval or sandbox policy
This gives one generic process API while keeping the policy-sensitive logic in
the place that already owns it.
### Why not make exec-server own sandbox selection?
That would force exec-server to understand:
- approval policy
- exec policy / prefix rules
- managed-network approval flow
- sandbox retry semantics
- guardian routing
- feature-flag-driven sandbox selection
- platform-specific sandbox helper configuration
That is too opinionated for a reusable process service.
## Optional future server config
If exec-server grows beyond the current prototype, a config object like this
would be enough:
```rust
struct ExecServerConfig {
shutdown_grace_period_ms: u64,
max_processes_per_connection: usize,
retained_output_bytes_per_process: usize,
allow_direct_exec: bool,
allow_prepared_exec: bool,
}
```
That keeps policy surface small:
- lifecycle limits live in the server
- trust and sandbox policy stay with the caller
## Mapping back to LLM-visible events
If unified exec is later backed by exec-server, the `core` client wrapper should
keep owning the translation into the existing event model:
- `process/start` success -> `ExecCommandBegin`
- `process/output` -> `ExecCommandOutputDelta`
- local `process/write` call -> `TerminalInteraction`
- `process/exited` plus retained transcript -> `ExecCommandEnd`
That preserves the current LLM-facing contract while making the process backend
swappable.

View File

@@ -0,0 +1,390 @@
# codex-exec-server
`codex-exec-server` is a small standalone JSON-RPC server for spawning and
controlling subprocesses through `codex-utils-pty`.
It currently provides:
- a standalone binary: `codex-exec-server`
- a transport-agnostic server runtime with stdio and websocket entrypoints
- a Rust client: `ExecServerClient`
- a direct in-process client mode: `ExecServerClient::connect_in_process`
- a separate local launch helper: `spawn_local_exec_server`
- a small protocol module with shared request/response types
This crate is intentionally narrow. It is not wired into the main Codex CLI or
unified-exec in this PR; it is only the standalone transport layer.
The internal shape is intentionally closer to `app-server` than the first cut:
- transport adapters are separate from the per-connection request processor
- JSON-RPC request dispatch is backed by `jsonrpsee` and kept separate from the
stateful exec handler
- the client only speaks the protocol; it does not spawn a server subprocess
- the client can also bypass the JSON-RPC transport/routing layer in local
in-process mode and call the typed handler directly
- local child-process launch is handled by a separate helper/factory layer
That split is meant to leave reusable seams if exec-server and app-server later
share transport or JSON-RPC connection utilities. It also keeps the core
handler testable without the RPC server implementation itself.
Design notes for a likely future integration with unified exec, including
rough call flow, buffering, and sandboxing boundaries, live in
[DESIGN.md](./DESIGN.md).
## Transport
The server speaks the same JSON-RPC message shapes over multiple transports.
The standalone binary supports:
- `stdio://` (default)
- `ws://IP:PORT`
Wire framing:
- stdio: one newline-delimited JSON-RPC message per line on stdin/stdout
- websocket: one JSON-RPC message per websocket text frame
Like the app-server transport, messages on the wire omit the `"jsonrpc":"2.0"`
field and use the shared `codex-app-server-protocol` envelope types.
The current protocol version is:
```text
exec-server.v0
```
## Lifecycle
Each connection follows this sequence:
1. Send `initialize`.
2. Wait for the `initialize` response.
3. Send `initialized`.
4. Start and manage processes with `process/start`, `process/read`,
`process/write`, and `process/terminate`.
5. Read streaming notifications from `process/output` and
`process/exited`.
If the client sends exec methods before completing the `initialize` /
`initialized` handshake, the server rejects them.
If a connection closes, the server terminates any remaining managed processes
for that connection.
## API
### `initialize`
Initial handshake request.
Request params:
```json
{
"clientName": "my-client"
}
```
Response:
```json
{
"protocolVersion": "exec-server.v0"
}
```
### `initialized`
Handshake acknowledgement notification sent by the client after a successful
`initialize` response. Exec methods are rejected until this arrives.
Params are currently ignored. Sending any other client notification method is a
protocol error.
### `process/start`
Starts a new managed process.
Request params:
```json
{
"processId": "proc-1",
"argv": ["bash", "-lc", "printf 'hello\\n'"],
"cwd": "/absolute/working/directory",
"env": {
"PATH": "/usr/bin:/bin"
},
"tty": true,
"arg0": null
}
```
Field definitions:
- `argv`: command vector. It must be non-empty.
- `cwd`: absolute working directory used for the child process.
- `env`: environment variables passed to the child process.
- `tty`: when `true`, spawn a PTY-backed interactive process; when `false`,
spawn a pipe-backed process with closed stdin.
- `arg0`: optional argv0 override forwarded to `codex-utils-pty`.
Response:
```json
{
"processId": "proc-1"
}
```
Behavior notes:
- `processId` is chosen by the client and must be unique for the connection.
- PTY-backed processes accept later writes through `process/write`.
- Pipe-backed processes are launched with stdin closed and reject writes.
- Output is streamed asynchronously via `process/output`.
- Exit is reported asynchronously via `process/exited`.
### `process/write`
Writes raw bytes to a running PTY-backed process stdin.
Request params:
```json
{
"processId": "proc-1",
"chunk": "aGVsbG8K"
}
```
`chunk` is base64-encoded raw bytes. In the example above it is `hello\n`.
Response:
```json
{
"accepted": true
}
```
Behavior notes:
- Writes to an unknown `processId` are rejected.
- Writes to a non-PTY process are rejected because stdin is already closed.
### `process/read`
Reads retained output from a managed process by sequence number.
Request params:
```json
{
"processId": "proc-1",
"afterSeq": 0,
"maxBytes": 65536,
"waitMs": 250
}
```
Response:
```json
{
"chunks": [
{
"seq": 1,
"stream": "pty",
"chunk": "aGVsbG8K"
}
],
"nextSeq": 2,
"exited": false,
"exitCode": null
}
```
Behavior notes:
- Output is retained in bounded server memory so callers can poll without
relying only on notifications.
- `afterSeq` is exclusive: `0` reads from the beginning of the retained buffer.
- `waitMs` waits briefly for new output or exit if nothing is currently
available.
- Once retained output exceeds the per-process cap, oldest chunks are dropped.
### `process/terminate`
Terminates a running managed process.
Request params:
```json
{
"processId": "proc-1"
}
```
Response:
```json
{
"running": true
}
```
If the process is already unknown or already removed, the server responds with:
```json
{
"running": false
}
```
## Notifications
### `process/output`
Streaming output chunk from a running process.
Params:
```json
{
"processId": "proc-1",
"stream": "stdout",
"chunk": "aGVsbG8K"
}
```
Fields:
- `processId`: process identifier
- `stream`: `"stdout"`, `"stderr"`, or `"pty"` for PTY-backed processes
- `chunk`: base64-encoded output bytes
### `process/exited`
Final process exit notification.
Params:
```json
{
"processId": "proc-1",
"exitCode": 0
}
```
## Errors
The server returns JSON-RPC errors with these codes:
- `-32600`: invalid request
- `-32602`: invalid params
- `-32603`: internal error
Typical error cases:
- unknown method
- malformed params
- empty `argv`
- duplicate `processId`
- writes to unknown processes
- writes to non-PTY processes
## Rust surface
The crate exports:
- `ExecServerClient`
- `ExecServerClientConnectOptions`
- `RemoteExecServerConnectArgs`
- `ExecServerLaunchCommand`
- `ExecServerEvent`
- `SpawnedExecServer`
- `ExecServerError`
- `ExecServerTransport`
- `spawn_local_exec_server(...)`
- protocol structs such as `ExecParams`, `ExecResponse`,
`WriteParams`, `TerminateParams`, `ExecOutputDeltaNotification`, and
`ExecExitedNotification`
- `run_main()` and `run_main_with_transport(...)`
### Binary
Run over stdio:
```text
codex-exec-server
```
Run as a websocket server:
```text
codex-exec-server --listen ws://127.0.0.1:8080
```
### Client
Connect the client to an existing server transport:
- `ExecServerClient::connect_stdio(...)`
- `ExecServerClient::connect_websocket(...)`
- `ExecServerClient::connect_in_process(...)` for a local no-transport mode
backed directly by the typed handler
Timeout behavior:
- stdio and websocket clients both enforce an initialize-handshake timeout
- websocket clients also enforce a connect timeout before the handshake begins
Events:
- `ExecServerClient::event_receiver()` yields `ExecServerEvent`
- output events include both `stream` (`stdout`, `stderr`, or `pty`) and raw
bytes
- process lifetime is tracked by server notifications such as
`process/exited`, not by a client-side process registry
Spawning a local child process is deliberately separate:
- `spawn_local_exec_server(...)`
## Example session
Initialize:
```json
{"id":1,"method":"initialize","params":{"clientName":"example-client"}}
{"id":1,"result":{"protocolVersion":"exec-server.v0"}}
{"method":"initialized","params":{}}
```
Start a process:
```json
{"id":2,"method":"process/start","params":{"processId":"proc-1","argv":["bash","-lc","printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"],"cwd":"/tmp","env":{"PATH":"/usr/bin:/bin"},"tty":true,"arg0":null}}
{"id":2,"result":{"processId":"proc-1"}}
{"method":"process/output","params":{"processId":"proc-1","stream":"pty","chunk":"cmVhZHkK"}}
```
Write to the process:
```json
{"id":3,"method":"process/write","params":{"processId":"proc-1","chunk":"aGVsbG8K"}}
{"id":3,"result":{"accepted":true}}
{"method":"process/output","params":{"processId":"proc-1","stream":"pty","chunk":"ZWNobzpoZWxsbwo="}}
```
Terminate it:
```json
{"id":4,"method":"process/terminate","params":{"processId":"proc-1"}}
{"id":4,"result":{"running":true}}
{"method":"process/exited","params":{"processId":"proc-1","exitCode":0}}
```

View File

@@ -0,0 +1,23 @@
use clap::Parser;
use codex_exec_server::ExecServerTransport;
#[derive(Debug, Parser)]
struct ExecServerArgs {
/// Transport endpoint URL. Supported values: `stdio://` (default),
/// `ws://IP:PORT`.
#[arg(
long = "listen",
value_name = "URL",
default_value = ExecServerTransport::DEFAULT_LISTEN_URL
)]
listen: ExecServerTransport,
}
#[tokio::main]
async fn main() {
let args = ExecServerArgs::parse();
if let Err(err) = codex_exec_server::run_main_with_transport(args.listen).await {
eprintln!("{err}");
std::process::exit(1);
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,27 @@
use std::time::Duration;
use crate::protocol::ExecExitedNotification;
use crate::protocol::ExecOutputDeltaNotification;
/// Connection options for any exec-server client transport.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecServerClientConnectOptions {
pub client_name: String,
pub initialize_timeout: Duration,
}
/// WebSocket connection arguments for a remote exec-server.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemoteExecServerConnectArgs {
pub websocket_url: String,
pub client_name: String,
pub connect_timeout: Duration,
pub initialize_timeout: Duration,
}
/// Connection-level server events.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ExecServerEvent {
OutputDelta(ExecOutputDeltaNotification),
Exited(ExecExitedNotification),
}

View File

@@ -0,0 +1,417 @@
use codex_app_server_protocol::JSONRPCMessage;
use futures::SinkExt;
use futures::StreamExt;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::io::BufWriter;
use tokio::sync::mpsc;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::tungstenite::Message;
pub(crate) const CHANNEL_CAPACITY: usize = 128;
#[derive(Debug)]
pub(crate) enum JsonRpcConnectionEvent {
Message(JSONRPCMessage),
Disconnected { reason: Option<String> },
}
pub(crate) struct JsonRpcConnection {
outgoing_tx: mpsc::Sender<JSONRPCMessage>,
incoming_rx: mpsc::Receiver<JsonRpcConnectionEvent>,
}
impl JsonRpcConnection {
pub(crate) fn from_stdio<R, W>(reader: R, writer: W, connection_label: String) -> Self
where
R: AsyncRead + Unpin + Send + 'static,
W: AsyncWrite + Unpin + Send + 'static,
{
let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
let reader_label = connection_label.clone();
let incoming_tx_for_reader = incoming_tx.clone();
tokio::spawn(async move {
let mut lines = BufReader::new(reader).lines();
loop {
match lines.next_line().await {
Ok(Some(line)) => {
if line.trim().is_empty() {
continue;
}
match serde_json::from_str::<JSONRPCMessage>(&line) {
Ok(message) => {
if incoming_tx_for_reader
.send(JsonRpcConnectionEvent::Message(message))
.await
.is_err()
{
break;
}
}
Err(err) => {
send_disconnected(
&incoming_tx_for_reader,
Some(format!(
"failed to parse JSON-RPC message from {reader_label}: {err}"
)),
)
.await;
break;
}
}
}
Ok(None) => {
send_disconnected(&incoming_tx_for_reader, None).await;
break;
}
Err(err) => {
send_disconnected(
&incoming_tx_for_reader,
Some(format!(
"failed to read JSON-RPC message from {reader_label}: {err}"
)),
)
.await;
break;
}
}
}
});
tokio::spawn(async move {
let mut writer = BufWriter::new(writer);
while let Some(message) = outgoing_rx.recv().await {
if let Err(err) = write_jsonrpc_line_message(&mut writer, &message).await {
send_disconnected(
&incoming_tx,
Some(format!(
"failed to write JSON-RPC message to {connection_label}: {err}"
)),
)
.await;
break;
}
}
});
Self {
outgoing_tx,
incoming_rx,
}
}
pub(crate) fn from_websocket<S>(stream: WebSocketStream<S>, connection_label: String) -> Self
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY);
let (mut websocket_writer, mut websocket_reader) = stream.split();
let reader_label = connection_label.clone();
let incoming_tx_for_reader = incoming_tx.clone();
tokio::spawn(async move {
loop {
match websocket_reader.next().await {
Some(Ok(Message::Text(text))) => {
match serde_json::from_str::<JSONRPCMessage>(text.as_ref()) {
Ok(message) => {
if incoming_tx_for_reader
.send(JsonRpcConnectionEvent::Message(message))
.await
.is_err()
{
break;
}
}
Err(err) => {
send_disconnected(
&incoming_tx_for_reader,
Some(format!(
"failed to parse websocket JSON-RPC message from {reader_label}: {err}"
)),
)
.await;
break;
}
}
}
Some(Ok(Message::Binary(bytes))) => {
match serde_json::from_slice::<JSONRPCMessage>(bytes.as_ref()) {
Ok(message) => {
if incoming_tx_for_reader
.send(JsonRpcConnectionEvent::Message(message))
.await
.is_err()
{
break;
}
}
Err(err) => {
send_disconnected(
&incoming_tx_for_reader,
Some(format!(
"failed to parse websocket JSON-RPC message from {reader_label}: {err}"
)),
)
.await;
break;
}
}
}
Some(Ok(Message::Close(_))) => {
send_disconnected(&incoming_tx_for_reader, None).await;
break;
}
Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
Some(Ok(_)) => {}
Some(Err(err)) => {
send_disconnected(
&incoming_tx_for_reader,
Some(format!(
"failed to read websocket JSON-RPC message from {reader_label}: {err}"
)),
)
.await;
break;
}
None => {
send_disconnected(&incoming_tx_for_reader, None).await;
break;
}
}
}
});
tokio::spawn(async move {
while let Some(message) = outgoing_rx.recv().await {
match serialize_jsonrpc_message(&message) {
Ok(encoded) => {
if let Err(err) = websocket_writer.send(Message::Text(encoded.into())).await
{
send_disconnected(
&incoming_tx,
Some(format!(
"failed to write websocket JSON-RPC message to {connection_label}: {err}"
)),
)
.await;
break;
}
}
Err(err) => {
send_disconnected(
&incoming_tx,
Some(format!(
"failed to serialize JSON-RPC message for {connection_label}: {err}"
)),
)
.await;
break;
}
}
}
});
Self {
outgoing_tx,
incoming_rx,
}
}
pub(crate) fn into_parts(
self,
) -> (
mpsc::Sender<JSONRPCMessage>,
mpsc::Receiver<JsonRpcConnectionEvent>,
) {
(self.outgoing_tx, self.incoming_rx)
}
}
async fn send_disconnected(
incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
reason: Option<String>,
) {
let _ = incoming_tx
.send(JsonRpcConnectionEvent::Disconnected { reason })
.await;
}
async fn write_jsonrpc_line_message<W>(
writer: &mut BufWriter<W>,
message: &JSONRPCMessage,
) -> std::io::Result<()>
where
W: AsyncWrite + Unpin,
{
let encoded =
serialize_jsonrpc_message(message).map_err(|err| std::io::Error::other(err.to_string()))?;
writer.write_all(encoded.as_bytes()).await?;
writer.write_all(b"\n").await?;
writer.flush().await
}
fn serialize_jsonrpc_message(message: &JSONRPCMessage) -> Result<String, serde_json::Error> {
serde_json::to_string(message)
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use pretty_assertions::assert_eq;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::sync::mpsc;
use tokio::time::timeout;
use super::JsonRpcConnection;
use super::JsonRpcConnectionEvent;
use super::serialize_jsonrpc_message;
async fn recv_event(
incoming_rx: &mut mpsc::Receiver<JsonRpcConnectionEvent>,
) -> JsonRpcConnectionEvent {
let recv_result = timeout(Duration::from_secs(1), incoming_rx.recv()).await;
let maybe_event = match recv_result {
Ok(maybe_event) => maybe_event,
Err(err) => panic!("timed out waiting for connection event: {err}"),
};
match maybe_event {
Some(event) => event,
None => panic!("connection event stream ended unexpectedly"),
}
}
async fn read_jsonrpc_line<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> JSONRPCMessage
where
R: tokio::io::AsyncRead + Unpin,
{
let next_line = timeout(Duration::from_secs(1), lines.next_line()).await;
let line_result = match next_line {
Ok(line_result) => line_result,
Err(err) => panic!("timed out waiting for JSON-RPC line: {err}"),
};
let maybe_line = match line_result {
Ok(maybe_line) => maybe_line,
Err(err) => panic!("failed to read JSON-RPC line: {err}"),
};
let line = match maybe_line {
Some(line) => line,
None => panic!("connection closed before JSON-RPC line arrived"),
};
match serde_json::from_str::<JSONRPCMessage>(&line) {
Ok(message) => message,
Err(err) => panic!("failed to parse JSON-RPC line: {err}"),
}
}
#[tokio::test]
async fn stdio_connection_reads_and_writes_jsonrpc_messages() {
let (mut writer_to_connection, connection_reader) = tokio::io::duplex(1024);
let (connection_writer, reader_from_connection) = tokio::io::duplex(1024);
let connection =
JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
let (outgoing_tx, mut incoming_rx) = connection.into_parts();
let incoming_message = JSONRPCMessage::Request(JSONRPCRequest {
id: RequestId::Integer(7),
method: "initialize".to_string(),
params: Some(serde_json::json!({ "clientName": "test-client" })),
trace: None,
});
let encoded = match serialize_jsonrpc_message(&incoming_message) {
Ok(encoded) => encoded,
Err(err) => panic!("failed to serialize incoming message: {err}"),
};
if let Err(err) = writer_to_connection
.write_all(format!("{encoded}\n").as_bytes())
.await
{
panic!("failed to write to connection: {err}");
}
let event = recv_event(&mut incoming_rx).await;
match event {
JsonRpcConnectionEvent::Message(message) => {
assert_eq!(message, incoming_message);
}
JsonRpcConnectionEvent::Disconnected { reason } => {
panic!("unexpected disconnect event: {reason:?}");
}
}
let outgoing_message = JSONRPCMessage::Response(JSONRPCResponse {
id: RequestId::Integer(7),
result: serde_json::json!({ "protocolVersion": "exec-server.v0" }),
});
if let Err(err) = outgoing_tx.send(outgoing_message.clone()).await {
panic!("failed to queue outgoing message: {err}");
}
let mut lines = BufReader::new(reader_from_connection).lines();
let message = read_jsonrpc_line(&mut lines).await;
assert_eq!(message, outgoing_message);
}
#[tokio::test]
async fn stdio_connection_reports_parse_errors() {
let (mut writer_to_connection, connection_reader) = tokio::io::duplex(1024);
let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
let connection =
JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
if let Err(err) = writer_to_connection.write_all(b"not-json\n").await {
panic!("failed to write invalid JSON: {err}");
}
let event = recv_event(&mut incoming_rx).await;
match event {
JsonRpcConnectionEvent::Disconnected { reason } => {
let reason = match reason {
Some(reason) => reason,
None => panic!("expected a parse error reason"),
};
assert!(
reason.contains("failed to parse JSON-RPC message from test"),
"unexpected disconnect reason: {reason}"
);
}
JsonRpcConnectionEvent::Message(message) => {
panic!("unexpected JSON-RPC message: {message:?}");
}
}
}
#[tokio::test]
async fn stdio_connection_reports_clean_disconnect() {
let (writer_to_connection, connection_reader) = tokio::io::duplex(1024);
let (connection_writer, _reader_from_connection) = tokio::io::duplex(1024);
let connection =
JsonRpcConnection::from_stdio(connection_reader, connection_writer, "test".to_string());
let (_outgoing_tx, mut incoming_rx) = connection.into_parts();
drop(writer_to_connection);
let event = recv_event(&mut incoming_rx).await;
match event {
JsonRpcConnectionEvent::Disconnected { reason } => {
assert_eq!(reason, None);
}
JsonRpcConnectionEvent::Message(message) => {
panic!("unexpected JSON-RPC message: {message:?}");
}
}
}
}

View File

@@ -0,0 +1,30 @@
mod client;
mod client_api;
mod connection;
mod local;
mod protocol;
mod server;
pub use client::ExecServerClient;
pub use client::ExecServerError;
pub use client_api::ExecServerClientConnectOptions;
pub use client_api::ExecServerEvent;
pub use client_api::RemoteExecServerConnectArgs;
pub use local::ExecServerLaunchCommand;
pub use local::SpawnedExecServer;
pub use local::spawn_local_exec_server;
pub use protocol::ExecExitedNotification;
pub use protocol::ExecOutputDeltaNotification;
pub use protocol::ExecOutputStream;
pub use protocol::ExecParams;
pub use protocol::ExecResponse;
pub use protocol::InitializeParams;
pub use protocol::InitializeResponse;
pub use protocol::TerminateParams;
pub use protocol::TerminateResponse;
pub use protocol::WriteParams;
pub use protocol::WriteResponse;
pub use server::ExecServerTransport;
pub use server::ExecServerTransportParseError;
pub use server::run_main;
pub use server::run_main_with_transport;

View File

@@ -0,0 +1,70 @@
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Mutex as StdMutex;
use tokio::process::Child;
use tokio::process::Command;
use crate::client::ExecServerClient;
use crate::client::ExecServerError;
use crate::client_api::ExecServerClientConnectOptions;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExecServerLaunchCommand {
pub program: PathBuf,
pub args: Vec<String>,
}
pub struct SpawnedExecServer {
client: ExecServerClient,
child: StdMutex<Option<Child>>,
}
impl SpawnedExecServer {
pub fn client(&self) -> &ExecServerClient {
&self.client
}
}
impl Drop for SpawnedExecServer {
fn drop(&mut self) {
if let Ok(mut child_guard) = self.child.lock()
&& let Some(child) = child_guard.as_mut()
{
let _ = child.start_kill();
}
}
}
pub async fn spawn_local_exec_server(
command: ExecServerLaunchCommand,
options: ExecServerClientConnectOptions,
) -> Result<SpawnedExecServer, ExecServerError> {
let mut child = Command::new(&command.program);
child.args(&command.args);
child.stdin(Stdio::piped());
child.stdout(Stdio::piped());
child.stderr(Stdio::inherit());
child.kill_on_drop(true);
let mut child = child.spawn().map_err(ExecServerError::Spawn)?;
let stdin = child.stdin.take().ok_or_else(|| {
ExecServerError::Protocol("exec-server stdin was not captured".to_string())
})?;
let stdout = child.stdout.take().ok_or_else(|| {
ExecServerError::Protocol("exec-server stdout was not captured".to_string())
})?;
let client = match ExecServerClient::connect_stdio(stdin, stdout, options).await {
Ok(client) => client,
Err(err) => {
let _ = child.start_kill();
return Err(err);
}
};
Ok(SpawnedExecServer {
client,
child: StdMutex::new(Some(child)),
})
}

View File

@@ -0,0 +1,162 @@
use std::collections::HashMap;
use std::path::PathBuf;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use serde::Deserialize;
use serde::Serialize;
pub const INITIALIZE_METHOD: &str = "initialize";
pub const INITIALIZED_METHOD: &str = "initialized";
pub const EXEC_METHOD: &str = "process/start";
pub const EXEC_READ_METHOD: &str = "process/read";
pub const EXEC_WRITE_METHOD: &str = "process/write";
pub const EXEC_TERMINATE_METHOD: &str = "process/terminate";
pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
pub const EXEC_EXITED_METHOD: &str = "process/exited";
pub const PROTOCOL_VERSION: &str = "exec-server.v0";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct ByteChunk(#[serde(with = "base64_bytes")] pub Vec<u8>);
impl ByteChunk {
pub fn into_inner(self) -> Vec<u8> {
self.0
}
}
impl From<Vec<u8>> for ByteChunk {
fn from(value: Vec<u8>) -> Self {
Self(value)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct InitializeParams {
pub client_name: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct InitializeResponse {
pub protocol_version: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecParams {
/// Client-chosen logical process handle scoped to this connection/session.
/// This is a protocol key, not an OS pid.
pub process_id: String,
pub argv: Vec<String>,
pub cwd: PathBuf,
pub env: HashMap<String, String>,
pub tty: bool,
pub arg0: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecResponse {
pub process_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ReadParams {
pub process_id: String,
pub after_seq: Option<u64>,
pub max_bytes: Option<usize>,
pub wait_ms: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProcessOutputChunk {
pub seq: u64,
pub stream: ExecOutputStream,
pub chunk: ByteChunk,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ReadResponse {
pub chunks: Vec<ProcessOutputChunk>,
pub next_seq: u64,
pub exited: bool,
pub exit_code: Option<i32>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct WriteParams {
pub process_id: String,
pub chunk: ByteChunk,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct WriteResponse {
pub accepted: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TerminateParams {
pub process_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TerminateResponse {
pub running: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ExecOutputStream {
Stdout,
Stderr,
Pty,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecOutputDeltaNotification {
pub process_id: String,
pub stream: ExecOutputStream,
pub chunk: ByteChunk,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExecExitedNotification {
pub process_id: String,
pub exit_code: i32,
}
mod base64_bytes {
use super::BASE64_STANDARD;
use base64::Engine as _;
use serde::Deserialize;
use serde::Deserializer;
use serde::Serializer;
pub fn serialize<S>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&BASE64_STANDARD.encode(bytes))
}
pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
where
D: Deserializer<'de>,
{
let encoded = String::deserialize(deserializer)?;
BASE64_STANDARD
.decode(encoded)
.map_err(serde::de::Error::custom)
}
}

View File

@@ -0,0 +1,20 @@
mod handler;
mod processor;
mod routing;
mod transport;
pub(crate) use handler::ExecServerHandler;
pub(crate) use routing::ExecServerOutboundMessage;
pub(crate) use routing::ExecServerServerNotification;
pub use transport::ExecServerTransport;
pub use transport::ExecServerTransportParseError;
pub async fn run_main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
run_main_with_transport(ExecServerTransport::Stdio).await
}
pub async fn run_main_with_transport(
transport: ExecServerTransport,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
transport::run_transport(transport).await
}

View File

@@ -0,0 +1,372 @@
use std::collections::HashMap;
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::Duration;
use codex_utils_pty::ExecCommandSession;
use codex_utils_pty::TerminalSize;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use crate::protocol::ExecExitedNotification;
use crate::protocol::ExecOutputDeltaNotification;
use crate::protocol::ExecOutputStream;
use crate::protocol::ExecResponse;
use crate::protocol::InitializeResponse;
use crate::protocol::PROTOCOL_VERSION;
use crate::protocol::ProcessOutputChunk;
use crate::protocol::ReadResponse;
use crate::protocol::TerminateResponse;
use crate::protocol::WriteResponse;
use crate::server::routing::ExecServerOutboundMessage;
use crate::server::routing::ExecServerServerNotification;
use crate::server::routing::internal_error;
use crate::server::routing::invalid_params;
use crate::server::routing::invalid_request;
const RETAINED_OUTPUT_BYTES_PER_PROCESS: usize = 1024 * 1024;
#[derive(Clone)]
struct RetainedOutputChunk {
seq: u64,
stream: ExecOutputStream,
chunk: Vec<u8>,
}
struct RunningProcess {
session: ExecCommandSession,
tty: bool,
output: VecDeque<RetainedOutputChunk>,
retained_bytes: usize,
next_seq: u64,
exit_code: Option<i32>,
}
pub(crate) struct ExecServerHandler {
outbound_tx: mpsc::Sender<ExecServerOutboundMessage>,
// Keyed by client-chosen logical `processId` scoped to this connection.
// This is a protocol handle, not an OS pid.
processes: Arc<Mutex<HashMap<String, RunningProcess>>>,
initialize_requested: bool,
initialized: bool,
}
impl ExecServerHandler {
pub(crate) fn new(outbound_tx: mpsc::Sender<ExecServerOutboundMessage>) -> Self {
Self {
outbound_tx,
processes: Arc::new(Mutex::new(HashMap::new())),
initialize_requested: false,
initialized: false,
}
}
pub(crate) async fn shutdown(&self) {
let remaining = {
let mut processes = self.processes.lock().await;
processes
.drain()
.map(|(_, process)| process)
.collect::<Vec<_>>()
};
for process in remaining {
process.session.terminate();
}
}
pub(crate) fn initialized(&mut self) -> Result<(), String> {
if !self.initialize_requested {
return Err("received `initialized` notification before `initialize`".into());
}
self.initialized = true;
Ok(())
}
pub(crate) fn initialize(
&mut self,
) -> Result<InitializeResponse, codex_app_server_protocol::JSONRPCErrorError> {
if self.initialize_requested {
return Err(invalid_request(
"initialize may only be sent once per connection".to_string(),
));
}
self.initialize_requested = true;
Ok(InitializeResponse {
protocol_version: PROTOCOL_VERSION.to_string(),
})
}
fn require_initialized(&self) -> Result<(), codex_app_server_protocol::JSONRPCErrorError> {
if !self.initialize_requested {
return Err(invalid_request(
"client must call initialize before using exec methods".to_string(),
));
}
if !self.initialized {
return Err(invalid_request(
"client must send initialized before using exec methods".to_string(),
));
}
Ok(())
}
pub(crate) async fn exec(
&self,
params: crate::protocol::ExecParams,
) -> Result<ExecResponse, codex_app_server_protocol::JSONRPCErrorError> {
self.require_initialized()?;
let process_id = params.process_id.clone();
{
let process_map = self.processes.lock().await;
if process_map.contains_key(&process_id) {
return Err(invalid_request(format!(
"process {process_id} already exists"
)));
}
}
let (program, args) = params
.argv
.split_first()
.ok_or_else(|| invalid_params("argv must not be empty".to_string()))?;
let spawned = if params.tty {
codex_utils_pty::spawn_pty_process(
program,
args,
params.cwd.as_path(),
&params.env,
&params.arg0,
TerminalSize::default(),
)
.await
} else {
codex_utils_pty::spawn_pipe_process_no_stdin(
program,
args,
params.cwd.as_path(),
&params.env,
&params.arg0,
)
.await
}
.map_err(|err| internal_error(err.to_string()))?;
{
let mut process_map = self.processes.lock().await;
process_map.insert(
process_id.clone(),
RunningProcess {
session: spawned.session,
tty: params.tty,
output: std::collections::VecDeque::new(),
retained_bytes: 0,
next_seq: 1,
exit_code: None,
},
);
}
tokio::spawn(stream_output(
process_id.clone(),
if params.tty {
ExecOutputStream::Pty
} else {
ExecOutputStream::Stdout
},
spawned.stdout_rx,
self.outbound_tx.clone(),
Arc::clone(&self.processes),
));
tokio::spawn(stream_output(
process_id.clone(),
if params.tty {
ExecOutputStream::Pty
} else {
ExecOutputStream::Stderr
},
spawned.stderr_rx,
self.outbound_tx.clone(),
Arc::clone(&self.processes),
));
tokio::spawn(watch_exit(
process_id.clone(),
spawned.exit_rx,
self.outbound_tx.clone(),
Arc::clone(&self.processes),
));
Ok(ExecResponse { process_id })
}
pub(crate) async fn read(
&self,
params: crate::protocol::ReadParams,
) -> Result<ReadResponse, codex_app_server_protocol::JSONRPCErrorError> {
self.require_initialized()?;
let after_seq = params.after_seq.unwrap_or(0);
let max_bytes = params.max_bytes.unwrap_or(usize::MAX);
let wait = Duration::from_millis(params.wait_ms.unwrap_or(0));
let deadline = tokio::time::Instant::now() + wait;
loop {
let response = {
let process_map = self.processes.lock().await;
let process = process_map.get(&params.process_id).ok_or_else(|| {
invalid_request(format!("unknown process id {}", params.process_id))
})?;
let mut chunks = Vec::new();
let mut total_bytes = 0;
for retained in process.output.iter().filter(|chunk| chunk.seq > after_seq) {
let chunk_len = retained.chunk.len();
if !chunks.is_empty() && total_bytes + chunk_len > max_bytes {
break;
}
total_bytes += chunk_len;
chunks.push(ProcessOutputChunk {
seq: retained.seq,
stream: retained.stream,
chunk: retained.chunk.clone().into(),
});
if total_bytes >= max_bytes {
break;
}
}
ReadResponse {
chunks,
next_seq: process.next_seq,
exited: process.exit_code.is_some(),
exit_code: process.exit_code,
}
};
if !response.chunks.is_empty()
|| response.exited
|| tokio::time::Instant::now() >= deadline
{
return Ok(response);
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
}
pub(crate) async fn write(
&self,
params: crate::protocol::WriteParams,
) -> Result<WriteResponse, codex_app_server_protocol::JSONRPCErrorError> {
self.require_initialized()?;
let writer_tx = {
let process_map = self.processes.lock().await;
let process = process_map.get(&params.process_id).ok_or_else(|| {
invalid_request(format!("unknown process id {}", params.process_id))
})?;
if !process.tty {
return Err(invalid_request(format!(
"stdin is closed for process {}",
params.process_id
)));
}
process.session.writer_sender()
};
writer_tx
.send(params.chunk.into_inner())
.await
.map_err(|_| internal_error("failed to write to process stdin".to_string()))?;
Ok(WriteResponse { accepted: true })
}
pub(crate) async fn terminate(
&self,
params: crate::protocol::TerminateParams,
) -> Result<TerminateResponse, codex_app_server_protocol::JSONRPCErrorError> {
self.require_initialized()?;
let running = {
let process_map = self.processes.lock().await;
if let Some(process) = process_map.get(&params.process_id) {
process.session.terminate();
true
} else {
false
}
};
Ok(TerminateResponse { running })
}
}
async fn stream_output(
process_id: String,
stream: ExecOutputStream,
mut receiver: tokio::sync::mpsc::Receiver<Vec<u8>>,
outbound_tx: mpsc::Sender<ExecServerOutboundMessage>,
processes: Arc<Mutex<HashMap<String, RunningProcess>>>,
) {
while let Some(chunk) = receiver.recv().await {
let notification = {
let mut processes = processes.lock().await;
let Some(process) = processes.get_mut(&process_id) else {
break;
};
let seq = process.next_seq;
process.next_seq += 1;
process.retained_bytes += chunk.len();
process.output.push_back(RetainedOutputChunk {
seq,
stream,
chunk: chunk.clone(),
});
while process.retained_bytes > RETAINED_OUTPUT_BYTES_PER_PROCESS {
let Some(evicted) = process.output.pop_front() else {
break;
};
process.retained_bytes = process.retained_bytes.saturating_sub(evicted.chunk.len());
}
ExecOutputDeltaNotification {
process_id: process_id.clone(),
stream,
chunk: chunk.into(),
}
};
if outbound_tx
.send(ExecServerOutboundMessage::Notification(
ExecServerServerNotification::OutputDelta(notification),
))
.await
.is_err()
{
break;
}
}
}
async fn watch_exit(
process_id: String,
exit_rx: tokio::sync::oneshot::Receiver<i32>,
outbound_tx: mpsc::Sender<ExecServerOutboundMessage>,
processes: Arc<Mutex<HashMap<String, RunningProcess>>>,
) {
let exit_code = exit_rx.await.unwrap_or(-1);
{
let mut processes = processes.lock().await;
if let Some(process) = processes.get_mut(&process_id) {
process.exit_code = Some(exit_code);
}
}
let _ = outbound_tx
.send(ExecServerOutboundMessage::Notification(
ExecServerServerNotification::Exited(ExecExitedNotification {
process_id,
exit_code,
}),
))
.await;
}
#[cfg(test)]
mod tests {}

View File

@@ -0,0 +1,224 @@
use std::sync::Arc;
use codex_app_server_protocol::JSONRPCMessage;
use jsonrpsee::RpcModule;
use jsonrpsee::core::RpcResult;
use jsonrpsee::types::ErrorObjectOwned;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tracing::debug;
use tracing::warn;
use crate::connection::CHANNEL_CAPACITY;
use crate::connection::JsonRpcConnection;
use crate::connection::JsonRpcConnectionEvent;
use crate::protocol::EXEC_METHOD;
use crate::protocol::EXEC_READ_METHOD;
use crate::protocol::EXEC_TERMINATE_METHOD;
use crate::protocol::EXEC_WRITE_METHOD;
use crate::protocol::ExecParams;
use crate::protocol::INITIALIZE_METHOD;
use crate::protocol::INITIALIZED_METHOD;
use crate::protocol::ReadParams;
use crate::protocol::TerminateParams;
use crate::protocol::WriteParams;
use crate::server::handler::ExecServerHandler;
use crate::server::routing::ExecServerOutboundMessage;
use crate::server::routing::encode_outbound_message;
const MAX_JSON_RESPONSE_BYTES: usize = 16 * 1024 * 1024;
pub(crate) async fn run_connection(connection: JsonRpcConnection) {
let (json_outgoing_tx, mut incoming_rx) = connection.into_parts();
let (outgoing_tx, mut outgoing_rx) =
mpsc::channel::<ExecServerOutboundMessage>(CHANNEL_CAPACITY);
let handler = Arc::new(Mutex::new(ExecServerHandler::new(outgoing_tx.clone())));
let rpc_module = match build_rpc_module(Arc::clone(&handler)) {
Ok(rpc_module) => rpc_module,
Err(err) => {
warn!("failed to build exec-server RPC module: {err}");
return;
}
};
let notification_outgoing_tx = json_outgoing_tx.clone();
let outbound_task = tokio::spawn(async move {
while let Some(message) = outgoing_rx.recv().await {
let json_message = match encode_outbound_message(message) {
Ok(json_message) => json_message,
Err(err) => {
warn!("failed to serialize exec-server outbound notification: {err}");
break;
}
};
if notification_outgoing_tx.send(json_message).await.is_err() {
break;
}
}
});
while let Some(event) = incoming_rx.recv().await {
match event {
JsonRpcConnectionEvent::Message(message) => {
if let Err(err) =
handle_incoming_message(&handler, &rpc_module, message, &json_outgoing_tx).await
{
warn!("closing exec-server connection after protocol error: {err}");
break;
}
}
JsonRpcConnectionEvent::Disconnected { reason } => {
if let Some(reason) = reason {
debug!("exec-server connection disconnected: {reason}");
}
break;
}
}
}
handler.lock().await.shutdown().await;
drop(outgoing_tx);
let _ = outbound_task.await;
}
fn build_rpc_module(
handler: Arc<Mutex<ExecServerHandler>>,
) -> Result<RpcModule<()>, Box<dyn std::error::Error + Send + Sync>> {
let mut rpc = RpcModule::new(());
let initialize_handler = Arc::clone(&handler);
rpc.register_async_method(INITIALIZE_METHOD, move |_, _, _| {
let initialize_handler = Arc::clone(&initialize_handler);
async move { handler_result_to_rpc(initialize_handler.lock().await.initialize()) }
})?;
let initialized_handler = Arc::clone(&handler);
rpc.register_async_method(INITIALIZED_METHOD, move |_, _, _| {
let initialized_handler = Arc::clone(&initialized_handler);
async move {
initialized_handler
.lock()
.await
.initialized()
.map_err(|message| ErrorObjectOwned::owned(-32600, message, None::<()>))?;
Ok::<(), ErrorObjectOwned>(())
}
})?;
let exec_handler = Arc::clone(&handler);
rpc.register_async_method(EXEC_METHOD, move |params, _, _| {
let exec_handler = Arc::clone(&exec_handler);
async move {
let params = params
.parse::<ExecParams>()
.map_err(|err| ErrorObjectOwned::owned(-32602, err.to_string(), None::<()>))?;
handler_result_to_rpc(exec_handler.lock().await.exec(params).await)
}
})?;
let read_handler = Arc::clone(&handler);
rpc.register_async_method(EXEC_READ_METHOD, move |params, _, _| {
let read_handler = Arc::clone(&read_handler);
async move {
let params = params
.parse::<ReadParams>()
.map_err(|err| ErrorObjectOwned::owned(-32602, err.to_string(), None::<()>))?;
handler_result_to_rpc(read_handler.lock().await.read(params).await)
}
})?;
let write_handler = Arc::clone(&handler);
rpc.register_async_method(EXEC_WRITE_METHOD, move |params, _, _| {
let write_handler = Arc::clone(&write_handler);
async move {
let params = params
.parse::<WriteParams>()
.map_err(|err| ErrorObjectOwned::owned(-32602, err.to_string(), None::<()>))?;
handler_result_to_rpc(write_handler.lock().await.write(params).await)
}
})?;
let terminate_handler = Arc::clone(&handler);
rpc.register_async_method(EXEC_TERMINATE_METHOD, move |params, _, _| {
let terminate_handler = Arc::clone(&terminate_handler);
async move {
let params = params
.parse::<TerminateParams>()
.map_err(|err| ErrorObjectOwned::owned(-32602, err.to_string(), None::<()>))?;
handler_result_to_rpc(terminate_handler.lock().await.terminate(params).await)
}
})?;
Ok(rpc)
}
fn handler_result_to_rpc<T>(
result: Result<T, codex_app_server_protocol::JSONRPCErrorError>,
) -> RpcResult<T> {
result.map_err(|error| {
let code = i32::try_from(error.code).unwrap_or(-32603);
ErrorObjectOwned::owned(code, error.message, error.data)
})
}
async fn handle_incoming_message(
handler: &Arc<Mutex<ExecServerHandler>>,
rpc_module: &RpcModule<()>,
message: JSONRPCMessage,
json_outgoing_tx: &mpsc::Sender<JSONRPCMessage>,
) -> Result<(), String> {
if let JSONRPCMessage::Notification(notification) = &message
&& notification.method == INITIALIZED_METHOD
{
return handler.lock().await.initialized();
}
let mut raw_request = serde_json::to_string(&message)
.map_err(|err| format!("failed to encode request: {err}"))?;
raw_request = inject_jsonrpc_version(&raw_request)?;
let response = rpc_module
.raw_json_request(&raw_request, MAX_JSON_RESPONSE_BYTES)
.await
.map_err(|err| err.to_string())?;
let (response, _subscription_rx) = response;
if response.get() == "null" {
return Ok(());
}
let mut response: JSONRPCMessage =
serde_json::from_str(response.get()).map_err(|err| err.to_string())?;
strip_jsonrpc_version(&mut response);
json_outgoing_tx
.send(response)
.await
.map_err(|_| "JSON-RPC connection closed".to_string())
}
fn inject_jsonrpc_version(raw_request: &str) -> Result<String, String> {
let mut value: serde_json::Value =
serde_json::from_str(raw_request).map_err(|err| err.to_string())?;
let Some(object) = value.as_object_mut() else {
return Err("expected JSON-RPC object".to_string());
};
object
.entry("jsonrpc".to_string())
.or_insert_with(|| serde_json::Value::String("2.0".to_string()));
serde_json::to_string(&value).map_err(|err| err.to_string())
}
fn strip_jsonrpc_version(message: &mut JSONRPCMessage) {
let encoded = match serde_json::to_value(&*message) {
Ok(encoded) => encoded,
Err(_) => return,
};
let mut object = match encoded {
serde_json::Value::Object(object) => object,
_ => return,
};
object.remove("jsonrpc");
if let Ok(decoded) = serde_json::from_value::<JSONRPCMessage>(serde_json::Value::Object(object))
{
*message = decoded;
}
}

View File

@@ -0,0 +1,68 @@
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use crate::protocol::EXEC_EXITED_METHOD;
use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
use crate::protocol::ExecExitedNotification;
use crate::protocol::ExecOutputDeltaNotification;
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum ExecServerOutboundMessage {
Notification(ExecServerServerNotification),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ExecServerServerNotification {
OutputDelta(ExecOutputDeltaNotification),
Exited(ExecExitedNotification),
}
pub(crate) fn encode_outbound_message(
message: ExecServerOutboundMessage,
) -> Result<JSONRPCMessage, serde_json::Error> {
match message {
ExecServerOutboundMessage::Notification(notification) => Ok(JSONRPCMessage::Notification(
serialize_notification(notification)?,
)),
}
}
pub(crate) fn invalid_request(message: String) -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32600,
data: None,
message,
}
}
pub(crate) fn invalid_params(message: String) -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32602,
data: None,
message,
}
}
pub(crate) fn internal_error(message: String) -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32603,
data: None,
message,
}
}
fn serialize_notification(
notification: ExecServerServerNotification,
) -> Result<JSONRPCNotification, serde_json::Error> {
match notification {
ExecServerServerNotification::OutputDelta(params) => Ok(JSONRPCNotification {
method: EXEC_OUTPUT_DELTA_METHOD.to_string(),
params: Some(serde_json::to_value(params)?),
}),
ExecServerServerNotification::Exited(params) => Ok(JSONRPCNotification {
method: EXEC_EXITED_METHOD.to_string(),
params: Some(serde_json::to_value(params)?),
}),
}
}

View File

@@ -0,0 +1,166 @@
use std::net::SocketAddr;
use std::str::FromStr;
use tokio::net::TcpListener;
use tokio_tungstenite::accept_async;
use tracing::warn;
use crate::connection::JsonRpcConnection;
use crate::server::processor::run_connection;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ExecServerTransport {
Stdio,
WebSocket { bind_address: SocketAddr },
}
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum ExecServerTransportParseError {
UnsupportedListenUrl(String),
InvalidWebSocketListenUrl(String),
}
impl std::fmt::Display for ExecServerTransportParseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ExecServerTransportParseError::UnsupportedListenUrl(listen_url) => write!(
f,
"unsupported --listen URL `{listen_url}`; expected `stdio://` or `ws://IP:PORT`"
),
ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url) => write!(
f,
"invalid websocket --listen URL `{listen_url}`; expected `ws://IP:PORT`"
),
}
}
}
impl std::error::Error for ExecServerTransportParseError {}
impl ExecServerTransport {
pub const DEFAULT_LISTEN_URL: &str = "stdio://";
pub fn from_listen_url(listen_url: &str) -> Result<Self, ExecServerTransportParseError> {
if listen_url == Self::DEFAULT_LISTEN_URL {
return Ok(Self::Stdio);
}
if let Some(socket_addr) = listen_url.strip_prefix("ws://") {
let bind_address = socket_addr.parse::<SocketAddr>().map_err(|_| {
ExecServerTransportParseError::InvalidWebSocketListenUrl(listen_url.to_string())
})?;
return Ok(Self::WebSocket { bind_address });
}
Err(ExecServerTransportParseError::UnsupportedListenUrl(
listen_url.to_string(),
))
}
}
impl FromStr for ExecServerTransport {
type Err = ExecServerTransportParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::from_listen_url(s)
}
}
pub(crate) async fn run_transport(
transport: ExecServerTransport,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
match transport {
ExecServerTransport::Stdio => {
run_connection(JsonRpcConnection::from_stdio(
tokio::io::stdin(),
tokio::io::stdout(),
"exec-server stdio".to_string(),
))
.await;
Ok(())
}
ExecServerTransport::WebSocket { bind_address } => {
run_websocket_listener(bind_address).await
}
}
}
async fn run_websocket_listener(
bind_address: SocketAddr,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let listener = TcpListener::bind(bind_address).await?;
let local_addr = listener.local_addr()?;
print_websocket_startup_banner(local_addr);
loop {
let (stream, peer_addr) = listener.accept().await?;
tokio::spawn(async move {
match accept_async(stream).await {
Ok(websocket) => {
run_connection(JsonRpcConnection::from_websocket(
websocket,
format!("exec-server websocket {peer_addr}"),
))
.await;
}
Err(err) => {
warn!(
"failed to accept exec-server websocket connection from {peer_addr}: {err}"
);
}
}
});
}
}
#[allow(clippy::print_stderr)]
fn print_websocket_startup_banner(addr: SocketAddr) {
eprintln!("codex-exec-server listening on ws://{addr}");
}
#[cfg(test)]
mod tests {
use pretty_assertions::assert_eq;
use super::ExecServerTransport;
#[test]
fn exec_server_transport_parses_stdio_listen_url() {
let transport =
ExecServerTransport::from_listen_url(ExecServerTransport::DEFAULT_LISTEN_URL)
.expect("stdio listen URL should parse");
assert_eq!(transport, ExecServerTransport::Stdio);
}
#[test]
fn exec_server_transport_parses_websocket_listen_url() {
let transport = ExecServerTransport::from_listen_url("ws://127.0.0.1:1234")
.expect("websocket listen URL should parse");
assert_eq!(
transport,
ExecServerTransport::WebSocket {
bind_address: "127.0.0.1:1234".parse().expect("valid socket address"),
}
);
}
#[test]
fn exec_server_transport_rejects_invalid_websocket_listen_url() {
let err = ExecServerTransport::from_listen_url("ws://localhost:1234")
.expect_err("hostname bind address should be rejected");
assert_eq!(
err.to_string(),
"invalid websocket --listen URL `ws://localhost:1234`; expected `ws://IP:PORT`"
);
}
#[test]
fn exec_server_transport_rejects_unsupported_listen_url() {
let err = ExecServerTransport::from_listen_url("http://127.0.0.1:1234")
.expect_err("unsupported scheme should fail");
assert_eq!(
err.to_string(),
"unsupported --listen URL `http://127.0.0.1:1234`; expected `stdio://` or `ws://IP:PORT`"
);
}
}

View File

@@ -0,0 +1,298 @@
#![cfg(unix)]
use std::process::Stdio;
use std::time::Duration;
use anyhow::Context;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_exec_server::ExecOutputStream;
use codex_exec_server::ExecParams;
use codex_exec_server::ExecServerClient;
use codex_exec_server::ExecServerClientConnectOptions;
use codex_exec_server::ExecServerEvent;
use codex_exec_server::ExecServerLaunchCommand;
use codex_exec_server::InitializeParams;
use codex_exec_server::InitializeResponse;
use codex_exec_server::RemoteExecServerConnectArgs;
use codex_exec_server::spawn_local_exec_server;
use codex_utils_cargo_bin::cargo_bin;
use pretty_assertions::assert_eq;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::process::Command;
use tokio::sync::broadcast;
use tokio::time::timeout;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_accepts_initialize_over_stdio() -> anyhow::Result<()> {
let binary = cargo_bin("codex-exec-server")?;
let mut child = Command::new(binary);
child.stdin(Stdio::piped());
child.stdout(Stdio::piped());
child.stderr(Stdio::inherit());
let mut child = child.spawn()?;
let mut stdin = child.stdin.take().expect("stdin");
let stdout = child.stdout.take().expect("stdout");
let mut stdout = BufReader::new(stdout).lines();
let initialize = JSONRPCMessage::Request(JSONRPCRequest {
id: RequestId::Integer(1),
method: "initialize".to_string(),
params: Some(serde_json::to_value(InitializeParams {
client_name: "exec-server-test".to_string(),
})?),
trace: None,
});
stdin
.write_all(format!("{}\n", serde_json::to_string(&initialize)?).as_bytes())
.await?;
let response_line = timeout(Duration::from_secs(5), stdout.next_line()).await??;
let response_line = response_line.expect("response line");
let response: JSONRPCMessage = serde_json::from_str(&response_line)?;
let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = response else {
panic!("expected initialize response");
};
assert_eq!(id, RequestId::Integer(1));
let initialize_response: InitializeResponse = serde_json::from_value(result)?;
assert_eq!(initialize_response.protocol_version, "exec-server.v0");
let initialized = JSONRPCMessage::Notification(JSONRPCNotification {
method: "initialized".to_string(),
params: Some(serde_json::json!({})),
});
stdin
.write_all(format!("{}\n", serde_json::to_string(&initialized)?).as_bytes())
.await?;
child.start_kill()?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_client_streams_output_and_accepts_writes() -> anyhow::Result<()> {
let mut env = std::collections::HashMap::new();
if let Some(path) = std::env::var_os("PATH") {
env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
}
let server = spawn_local_exec_server(
ExecServerLaunchCommand {
program: cargo_bin("codex-exec-server")?,
args: Vec::new(),
},
ExecServerClientConnectOptions {
client_name: "exec-server-test".to_string(),
initialize_timeout: Duration::from_secs(5),
},
)
.await?;
let client = server.client();
let mut events = client.event_receiver();
let response = client
.exec(ExecParams {
process_id: "proc-1".to_string(),
argv: vec![
"bash".to_string(),
"-lc".to_string(),
"printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
.to_string(),
],
cwd: std::env::current_dir()?,
env,
tty: true,
arg0: None,
})
.await?;
let process_id = response.process_id;
let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
assert_eq!(stream, ExecOutputStream::Pty);
assert!(
ready_output.contains("ready"),
"expected initial ready output"
);
client.write(&process_id, b"hello\n".to_vec()).await?;
let (stream, echoed_output) =
recv_until_contains(&mut events, &process_id, "echo:hello").await?;
assert_eq!(stream, ExecOutputStream::Pty);
assert!(
echoed_output.contains("echo:hello"),
"expected echoed output"
);
client.terminate(&process_id).await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_client_connects_over_websocket() -> anyhow::Result<()> {
let mut env = std::collections::HashMap::new();
if let Some(path) = std::env::var_os("PATH") {
env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
}
let binary = cargo_bin("codex-exec-server")?;
let mut child = Command::new(binary);
child.args(["--listen", "ws://127.0.0.1:0"]);
child.stdin(Stdio::null());
child.stdout(Stdio::null());
child.stderr(Stdio::piped());
let mut child = child.spawn()?;
let stderr = child.stderr.take().expect("stderr");
let mut stderr_lines = BufReader::new(stderr).lines();
let websocket_url = read_websocket_url(&mut stderr_lines).await?;
let client = ExecServerClient::connect_websocket(RemoteExecServerConnectArgs {
websocket_url,
client_name: "exec-server-test".to_string(),
connect_timeout: Duration::from_secs(5),
initialize_timeout: Duration::from_secs(5),
})
.await?;
let mut events = client.event_receiver();
let response = client
.exec(ExecParams {
process_id: "proc-1".to_string(),
argv: vec![
"bash".to_string(),
"-lc".to_string(),
"printf 'ready\\n'; while IFS= read -r line; do printf 'echo:%s\\n' \"$line\"; done"
.to_string(),
],
cwd: std::env::current_dir()?,
env,
tty: true,
arg0: None,
})
.await?;
let process_id = response.process_id;
let (stream, ready_output) = recv_until_contains(&mut events, &process_id, "ready").await?;
assert_eq!(stream, ExecOutputStream::Pty);
assert!(
ready_output.contains("ready"),
"expected initial ready output"
);
client.write(&process_id, b"hello\n".to_vec()).await?;
let (stream, echoed_output) =
recv_until_contains(&mut events, &process_id, "echo:hello").await?;
assert_eq!(stream, ExecOutputStream::Pty);
assert!(
echoed_output.contains("echo:hello"),
"expected echoed output"
);
client.terminate(&process_id).await?;
child.start_kill()?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn websocket_disconnect_terminates_processes_for_that_connection() -> anyhow::Result<()> {
let mut env = std::collections::HashMap::new();
if let Some(path) = std::env::var_os("PATH") {
env.insert("PATH".to_string(), path.to_string_lossy().into_owned());
}
let marker_path = std::env::temp_dir().join(format!(
"codex-exec-server-disconnect-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_nanos()
));
let _ = std::fs::remove_file(&marker_path);
let binary = cargo_bin("codex-exec-server")?;
let mut child = Command::new(binary);
child.args(["--listen", "ws://127.0.0.1:0"]);
child.stdin(Stdio::null());
child.stdout(Stdio::null());
child.stderr(Stdio::piped());
let mut child = child.spawn()?;
let stderr = child.stderr.take().expect("stderr");
let mut stderr_lines = BufReader::new(stderr).lines();
let websocket_url = read_websocket_url(&mut stderr_lines).await?;
{
let client = ExecServerClient::connect_websocket(RemoteExecServerConnectArgs {
websocket_url,
client_name: "exec-server-test".to_string(),
connect_timeout: Duration::from_secs(5),
initialize_timeout: Duration::from_secs(5),
})
.await?;
let _response = client
.exec(ExecParams {
process_id: "proc-1".to_string(),
argv: vec![
"bash".to_string(),
"-lc".to_string(),
format!("sleep 2; printf disconnected > {}", marker_path.display()),
],
cwd: std::env::current_dir()?,
env,
tty: false,
arg0: None,
})
.await?;
}
tokio::time::sleep(Duration::from_secs(3)).await;
assert!(
!marker_path.exists(),
"managed process should be terminated when the websocket client disconnects"
);
child.start_kill()?;
let _ = std::fs::remove_file(&marker_path);
Ok(())
}
async fn read_websocket_url<R>(lines: &mut tokio::io::Lines<BufReader<R>>) -> anyhow::Result<String>
where
R: tokio::io::AsyncRead + Unpin,
{
let line = timeout(Duration::from_secs(5), lines.next_line()).await??;
let line = line.context("missing websocket startup banner")?;
let websocket_url = line
.split_whitespace()
.find(|part| part.starts_with("ws://"))
.context("missing websocket URL in startup banner")?;
Ok(websocket_url.to_string())
}
async fn recv_until_contains(
events: &mut broadcast::Receiver<ExecServerEvent>,
process_id: &str,
needle: &str,
) -> anyhow::Result<(ExecOutputStream, String)> {
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
let mut collected = String::new();
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
let event = timeout(remaining, events.recv()).await??;
if let ExecServerEvent::OutputDelta(output_event) = event
&& output_event.process_id == process_id
{
collected.push_str(&String::from_utf8_lossy(&output_event.chunk.into_inner()));
if collected.contains(needle) {
return Ok((output_event.stream, collected));
}
}
}
}