From bbc1d5feb1519190055d53dc4f576757faf19c1c Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:11:39 +0200 Subject: [PATCH 1/6] feat(codex): app-server JSON-RPC transport client First stage of the Codex app-server migration. Adds a self-contained transport module: process lifecycle for a long-lived `codex app-server` child, newline-delimited JSON-RPC framing (background stdout reader -> channel), and pure request/notification/error builders with unit tests. Not wired into the run path yet (that is the next commit); gated behind the CLAI_CODEX_APP_SERVER env flag so `codex exec` stays the default transport. app-server is the protocol every first-party Codex surface runs on and exposes turn/steer for real-time mid-turn input. --- src-tauri/src/assistant/codex_app_server.rs | 459 ++++++++++++++++++++ src-tauri/src/assistant/mod.rs | 1 + 2 files changed, 460 insertions(+) create mode 100644 src-tauri/src/assistant/codex_app_server.rs diff --git a/src-tauri/src/assistant/codex_app_server.rs b/src-tauri/src/assistant/codex_app_server.rs new file mode 100644 index 0000000..1fbf0ae --- /dev/null +++ b/src-tauri/src/assistant/codex_app_server.rs @@ -0,0 +1,459 @@ +//! Codex `app-server` transport — JSON-RPC over a long-lived child process. +//! +//! The default Codex path (`run_codex_turn`) shells out to `codex exec`, whose +//! stdin is consumed once for the initial prompt and then closed. That makes +//! mid-run input impossible without killing and restarting the process. +//! +//! `codex app-server` is the JSON-RPC protocol that powers every first-party +//! Codex surface (VS Code extension, desktop app, web). It exposes `turn/steer`, +//! which injects input into the **currently active turn** — genuine input +//! streaming, no interrupt/restart. This module is the transport layer for that +//! path: process lifecycle, newline-delimited JSON-RPC framing, and pure +//! request/notification builders. The turn *driver* (event → UI mapping, tool +//! persistence, steering policy) lives in `local_agent.rs` so it can reuse the +//! shared Codex stream helpers. +//! +//! Gated behind `CLAI_CODEX_APP_SERVER` (default off) while it bakes; `codex +//! exec` remains the default transport. + +use std::process::Stdio; + +use serde_json::{json, Value}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStderr, ChildStdin, Command}; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver}; + +/// Env var that opts a Codex connection into the app-server transport. +pub(crate) const APP_SERVER_ENABLED_ENV: &str = "CLAI_CODEX_APP_SERVER"; + +/// Whether the app-server transport is enabled for this process. Off unless the +/// env var is explicitly truthy, so `codex exec` stays the default. +pub(crate) fn app_server_enabled() -> bool { + matches!( + std::env::var(APP_SERVER_ENABLED_ENV) + .ok() + .as_deref() + .map(str::trim) + .map(str::to_ascii_lowercase) + .as_deref(), + Some("1") | Some("true") | Some("yes") | Some("on") + ) +} + +/// A running `codex app-server` process. Reads are pumped off the stdout pipe +/// on a background task into an unbounded channel so the driver can `select!` +/// over messages, a steering poll timer, and cancellation without deadlocking +/// on the pipe. +pub(crate) struct AppServerTransport { + child: Child, + stdin: ChildStdin, + stderr: Option, + rx: UnboundedReceiver, +} + +impl AppServerTransport { + /// Spawn ` app-server` with piped stdio and start the reader task. + /// `command` is expected to already carry the env (MCP token, timeouts) and + /// working directory, built via `providers::build_host_cli_command` so it + /// survives the Flatpak host hop. + pub(crate) fn spawn(mut command: Command) -> Result { + command + .arg("app-server") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = command.spawn().map_err(|e| { + format!("Failed to launch `codex app-server`: {e}. Is Codex CLI installed and on PATH?") + })?; + let stdin = child + .stdin + .take() + .ok_or_else(|| "codex app-server stdin was not captured".to_string())?; + let stdout = child + .stdout + .take() + .ok_or_else(|| "codex app-server stdout was not captured".to_string())?; + let stderr = child.stderr.take(); + + let (tx, rx) = unbounded_channel(); + tokio::spawn(async move { + let mut lines = BufReader::new(stdout).lines(); + while let Ok(Some(line)) = lines.next_line().await { + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line) { + Ok(value) => { + if tx.send(value).is_err() { + break; // driver dropped the receiver + } + } + Err(error) => { + tracing::warn!(%error, line = %line, "codex app-server: unparseable JSON line"); + } + } + } + }); + + Ok(Self { + child, + stdin, + stderr, + rx, + }) + } + + /// Take the stderr pipe so the caller can attach the shared stderr logger + /// (used to enrich failure messages). Returns `None` after the first call. + pub(crate) fn take_stderr(&mut self) -> Option { + self.stderr.take() + } + + /// Write one JSON-RPC message as a single newline-delimited line. + pub(crate) async fn send(&mut self, message: &Value) -> Result<(), String> { + let mut line = serde_json::to_string(message).map_err(|e| e.to_string())?; + line.push('\n'); + self.stdin + .write_all(line.as_bytes()) + .await + .map_err(|e| format!("codex app-server stdin write failed: {e}"))?; + self.stdin + .flush() + .await + .map_err(|e| format!("codex app-server stdin flush failed: {e}")) + } + + /// Next server message, or `None` once the process closes stdout. + pub(crate) async fn recv(&mut self) -> Option { + self.rx.recv().await + } + + /// Force-kill the child (used on cancel / teardown). + pub(crate) async fn kill(&mut self) { + let _ = self.child.kill().await; + } +} + +// --------------------------------------------------------------------------- +// JSON-RPC message classification (pure) +// --------------------------------------------------------------------------- + +/// A response carries an `id` and a `result`/`error`, with no `method`. +pub(crate) fn is_response(value: &Value) -> bool { + value.get("method").is_none() + && value.get("id").is_some() + && (value.get("result").is_some() || value.get("error").is_some()) +} + +/// A server→client request carries both a `method` and an `id`. +pub(crate) fn is_server_request(value: &Value) -> bool { + value.get("method").is_some() && value.get("id").is_some() +} + +/// A notification carries a `method` and no `id`. +pub(crate) fn notification_method(value: &Value) -> Option<&str> { + if value.get("id").is_some() { + return None; + } + value.get("method").and_then(Value::as_str) +} + +/// The response id (matched against outgoing request ids during the handshake). +pub(crate) fn response_id(value: &Value) -> Option { + value.get("id").and_then(Value::as_i64) +} + +// --------------------------------------------------------------------------- +// Request builders (pure) +// --------------------------------------------------------------------------- + +fn request(id: i64, method: &str, params: Value) -> Value { + json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }) +} + +pub(crate) fn initialize_request(id: i64, client_version: &str) -> Value { + request( + id, + "initialize", + json!({ + "clientInfo": { "name": "clai", "title": "CLAI", "version": client_version } + }), + ) +} + +/// A single `text` [`UserInput`] element. +pub(crate) fn text_user_input(text: &str) -> Value { + json!({ "type": "text", "text": text }) +} + +/// A single `localImage` [`UserInput`] element (Codex reads the file itself). +pub(crate) fn local_image_user_input(path: &str) -> Value { + json!({ "type": "localImage", "path": path }) +} + +/// The `config` object mirroring the `-c mcp_servers.clai.*` flags the `exec` +/// path passes, so the app-server turn can reach the same local MCP server. +pub(crate) fn mcp_servers_config(mcp_url: &str, token_env: &str, tool_timeout_secs: u64) -> Value { + json!({ + "mcp_servers": { + "clai": { + "url": mcp_url, + "bearer_token_env_var": token_env, + "enabled": true, + "required": true, + "tool_timeout_sec": tool_timeout_secs, + } + } + }) +} + +/// Build `thread/start` params. `approvalPolicy: never` + `sandbox: +/// danger-full-access` is the app-server parallel of `exec`'s +/// `--dangerously-bypass-approvals-and-sandbox`: CLAI provides the external +/// sandbox (bwrap) and permission system through its MCP tools, so Codex must +/// not gate or sandbox anything itself. +#[allow(clippy::too_many_arguments)] +pub(crate) fn thread_start_request( + id: i64, + cwd: Option<&str>, + model: Option<&str>, + mcp_url: &str, + token_env: &str, + tool_timeout_secs: u64, +) -> Value { + let mut params = json!({ + "approvalPolicy": "never", + "sandbox": "danger-full-access", + "config": mcp_servers_config(mcp_url, token_env, tool_timeout_secs), + }); + if let Some(cwd) = cwd { + params["cwd"] = json!(cwd); + } + if let Some(model) = model { + params["model"] = json!(model); + } + request(id, "thread/start", params) +} + +/// Build `thread/resume` params for an existing Codex thread id. +pub(crate) fn thread_resume_request(id: i64, thread_id: &str, model: Option<&str>) -> Value { + let mut params = json!({ + "threadId": thread_id, + "approvalPolicy": "never", + "sandbox": "danger-full-access", + }); + if let Some(model) = model { + params["model"] = json!(model); + } + request(id, "thread/resume", params) +} + +/// Build `turn/start` for a thread with the given input elements. +pub(crate) fn turn_start_request(id: i64, thread_id: &str, input: Vec) -> Value { + request( + id, + "turn/start", + json!({ "threadId": thread_id, "input": input }), + ) +} + +/// Build `turn/steer` — inject input into the active turn. Guarded by +/// `expectedTurnId`: the server rejects it if that turn is no longer active. +pub(crate) fn turn_steer_request( + id: i64, + thread_id: &str, + expected_turn_id: &str, + input: Vec, +) -> Value { + request( + id, + "turn/steer", + json!({ + "threadId": thread_id, + "expectedTurnId": expected_turn_id, + "input": input, + }), + ) +} + +/// Build `turn/interrupt` (used on cancellation). +pub(crate) fn turn_interrupt_request(id: i64, thread_id: &str) -> Value { + request(id, "turn/interrupt", json!({ "threadId": thread_id })) +} + +// --------------------------------------------------------------------------- +// Server→client request responses (pure) +// --------------------------------------------------------------------------- + +/// Best-effort response to a server→client request. With `approvalPolicy: +/// never` the server should not ask for approvals, but if it does we answer +/// permissively (CLAI already owns the real sandbox/permission gate) rather +/// than let the turn hang. Requests we can't answer get a JSON-RPC error so the +/// server can proceed instead of blocking on us. +pub(crate) fn server_request_response(id: &Value, method: &str) -> Value { + match method { + // Legacy (v1) approval requests use ReviewDecision. + "execCommandApproval" | "applyPatchApproval" => json!({ + "jsonrpc": "2.0", + "id": id, + "result": { "decision": "approved" } + }), + // v2 approval requests use a per-kind decision enum whose "allow" arm + // is `accept`. + "item/commandExecution/requestApproval" | "item/fileChange/requestApproval" => json!({ + "jsonrpc": "2.0", + "id": id, + "result": { "decision": "accept" } + }), + // Anything else (permission profiles, elicitations, token refresh, + // attestation, client-side tool calls) we don't implement — decline + // cleanly so the server doesn't wait on us. + _ => json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32601, "message": format!("clai does not handle server request `{method}`") } + }), + } +} + +// --------------------------------------------------------------------------- +// Error classification (pure) +// --------------------------------------------------------------------------- + +/// Turn a Codex app-server error (`error.message` + `error.codexErrorInfo`) +/// into a message that CLAI's existing CLI error classifiers recognize. +/// +/// The `exec` path only sees free text, so the recovery classifiers +/// (`is_context_limit_error`, `is_usage_limit_error`) match on phrases. The +/// app-server gives us a *structured* code (`contextWindowExceeded`, +/// `usageLimitExceeded`); we fold the matching phrase into the message so the +/// same recovery/notice paths fire. +pub(crate) fn classify_error_message(codex_error_info: Option<&str>, message: &str) -> String { + match codex_error_info { + Some("contextWindowExceeded") + if !message.to_ascii_lowercase().contains("context window") => + { + format!("{message} (context window exceeded)") + } + Some("usageLimitExceeded") if !message.to_ascii_lowercase().contains("usage limit") => { + format!("{message} (usage limit exceeded)") + } + _ => message.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn enabled_only_for_truthy_values() { + for (val, want) in [ + ("1", true), + ("true", true), + ("TRUE", true), + ("on", true), + ("yes", true), + ("0", false), + ("false", false), + ("", false), + ("maybe", false), + ] { + std::env::set_var(APP_SERVER_ENABLED_ENV, val); + assert_eq!(app_server_enabled(), want, "value {val:?}"); + } + std::env::remove_var(APP_SERVER_ENABLED_ENV); + assert!(!app_server_enabled()); + } + + #[test] + fn message_classification_matches_jsonrpc_shapes() { + let resp = json!({ "jsonrpc": "2.0", "id": 3, "result": { "ok": true } }); + assert!(is_response(&resp)); + assert!(!is_server_request(&resp)); + assert_eq!(notification_method(&resp), None); + assert_eq!(response_id(&resp), Some(3)); + + let note = json!({ "jsonrpc": "2.0", "method": "turn/started", "params": {} }); + assert!(!is_response(¬e)); + assert!(!is_server_request(¬e)); + assert_eq!(notification_method(¬e), Some("turn/started")); + + let sreq = + json!({ "jsonrpc": "2.0", "id": 9, "method": "execCommandApproval", "params": {} }); + assert!(!is_response(&sreq)); + assert!(is_server_request(&sreq)); + assert_eq!(notification_method(&sreq), None); + } + + #[test] + fn thread_start_bypasses_approvals_and_carries_mcp_config() { + let req = thread_start_request( + 2, + Some("/ws"), + Some("gpt-5.5"), + "http://127.0.0.1:9/mcp", + "CLAI_MCP_TOKEN", + 3600, + ); + let params = &req["params"]; + assert_eq!(params["approvalPolicy"], "never"); + assert_eq!(params["sandbox"], "danger-full-access"); + assert_eq!(params["cwd"], "/ws"); + assert_eq!(params["model"], "gpt-5.5"); + let clai = ¶ms["config"]["mcp_servers"]["clai"]; + assert_eq!(clai["url"], "http://127.0.0.1:9/mcp"); + assert_eq!(clai["bearer_token_env_var"], "CLAI_MCP_TOKEN"); + assert_eq!(clai["enabled"], true); + assert_eq!(clai["tool_timeout_sec"], 3600); + } + + #[test] + fn thread_start_omits_default_model() { + let req = thread_start_request(2, None, None, "u", "T", 60); + assert!(req["params"].get("model").is_none()); + assert!(req["params"].get("cwd").is_none()); + } + + #[test] + fn steer_carries_expected_turn_precondition() { + let req = turn_steer_request(7, "thread-1", "turn-9", vec![text_user_input("hi")]); + assert_eq!(req["method"], "turn/steer"); + assert_eq!(req["params"]["threadId"], "thread-1"); + assert_eq!(req["params"]["expectedTurnId"], "turn-9"); + assert_eq!(req["params"]["input"][0]["type"], "text"); + assert_eq!(req["params"]["input"][0]["text"], "hi"); + } + + #[test] + fn approval_requests_answered_permissively() { + let id = json!(5); + assert_eq!( + server_request_response(&id, "execCommandApproval")["result"]["decision"], + "approved" + ); + assert_eq!( + server_request_response(&id, "item/fileChange/requestApproval")["result"]["decision"], + "accept" + ); + // Unknown requests get a clean JSON-RPC error, never a hang. + assert_eq!( + server_request_response(&id, "attestation/generate")["error"]["code"], + -32601 + ); + } + + #[test] + fn error_classification_folds_structured_codes_into_text() { + let ctx = classify_error_message(Some("contextWindowExceeded"), "boom"); + assert!(ctx.to_ascii_lowercase().contains("context window")); + let usage = classify_error_message(Some("usageLimitExceeded"), "boom"); + assert!(usage.to_ascii_lowercase().contains("usage limit")); + // Already-descriptive messages are left alone. + assert_eq!( + classify_error_message(Some("contextWindowExceeded"), "the context window is full"), + "the context window is full" + ); + assert_eq!(classify_error_message(None, "plain"), "plain"); + } +} diff --git a/src-tauri/src/assistant/mod.rs b/src-tauri/src/assistant/mod.rs index 2f7aca0..3916781 100644 --- a/src-tauri/src/assistant/mod.rs +++ b/src-tauri/src/assistant/mod.rs @@ -4,6 +4,7 @@ //! abstraction, event protocol, and persistence foundations. pub mod auth; +pub mod codex_app_server; pub mod compaction; pub mod engine; pub mod events; From 48dc6fd4a1b454cd50245c8669a81459e7ea5b5e Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:21:38 +0200 Subject: [PATCH 2/6] feat(codex): run turns through app-server when enabled Second stage: wire the app-server transport into the Codex run path, behaviour-parity with codex exec (no steering yet). When CLAI_CODEX_APP_SERVER is set, dispatch routes to run_codex_turn_app_server, which does the initialize -> thread/start|resume -> turn/start handshake, then maps the notification stream (item/started, item/completed, thread/tokenUsage/updated, error, turn/completed) onto the existing CodexStreamState + handle_codex_item helpers by normalizing app-server items into the exec item shape. approvalPolicy=never + danger-full-access mirror the exec bypass; structured error codes are folded into the CLI error classifiers so context/usage recovery still fires. codex exec remains the default. turn_steer_request carries a temporary allow(dead_code) until the next commit wires steering. --- src-tauri/src/assistant/codex_app_server.rs | 1 + src-tauri/src/assistant/local_agent.rs | 491 +++++++++++++++++++- 2 files changed, 480 insertions(+), 12 deletions(-) diff --git a/src-tauri/src/assistant/codex_app_server.rs b/src-tauri/src/assistant/codex_app_server.rs index 1fbf0ae..012f4a0 100644 --- a/src-tauri/src/assistant/codex_app_server.rs +++ b/src-tauri/src/assistant/codex_app_server.rs @@ -259,6 +259,7 @@ pub(crate) fn turn_start_request(id: i64, thread_id: &str, input: Vec) -> /// Build `turn/steer` — inject input into the active turn. Guarded by /// `expectedTurnId`: the server rejects it if that turn is no longer active. +#[allow(dead_code)] // wired by the steering stage pub(crate) fn turn_steer_request( id: i64, thread_id: &str, diff --git a/src-tauri/src/assistant/local_agent.rs b/src-tauri/src/assistant/local_agent.rs index 6fa5f2b..cc685ee 100644 --- a/src-tauri/src/assistant/local_agent.rs +++ b/src-tauri/src/assistant/local_agent.rs @@ -12,6 +12,7 @@ use tokio::process::Command; use tokio_util::sync::CancellationToken; use uuid::Uuid; +use crate::assistant::codex_app_server; use crate::assistant::compaction; use crate::assistant::engine::{ build_system_prompt, build_trigger_message, AssistantDeps, AssistantEngineError, RunTurnInput, @@ -396,18 +397,33 @@ pub async fn run_session_turn( result } CliProviderRuntime::Codex => { - run_codex_turn( - deps, - &mut session, - &connection, - &run_id, - mcp_runtime.url(), - binding_guard.token(), - &input.cancel_token, - &input.trigger, - &mut assistant_slot, - ) - .await + if codex_app_server::app_server_enabled() { + run_codex_turn_app_server( + deps, + &mut session, + &connection, + &run_id, + mcp_runtime.url(), + binding_guard.token(), + &input.cancel_token, + &input.trigger, + &mut assistant_slot, + ) + .await + } else { + run_codex_turn( + deps, + &mut session, + &connection, + &run_id, + mcp_runtime.url(), + binding_guard.token(), + &input.cancel_token, + &input.trigger, + &mut assistant_slot, + ) + .await + } } CliProviderRuntime::OpenCode => { run_opencode_turn( @@ -1470,6 +1486,457 @@ async fn run_codex_turn( Ok(usage) } +// --------------------------------------------------------------------------- +// Codex app-server transport driver (CLAI_CODEX_APP_SERVER) +// --------------------------------------------------------------------------- + +/// Run a Codex turn over the `codex app-server` JSON-RPC transport instead of +/// `codex exec`. Behaviour-parity with [`run_codex_turn`] for now (streaming, +/// tool calls, usage, errors); `turn/steer` mid-run delivery is layered on in a +/// follow-up commit. Reuses the shared [`CodexStreamState`] + tool-call helpers +/// by normalizing app-server items into the exec item shape. +#[allow(clippy::too_many_arguments)] +async fn run_codex_turn_app_server( + deps: &AssistantDeps, + session: &mut AssistantSession, + connection: &ProviderConnection, + run_id: &str, + mcp_url: &str, + mcp_token: &str, + cancel_token: &CancellationToken, + trigger: &crate::assistant::types::RunTrigger, + assistant_slot: &mut Option, +) -> Result, LocalAgentRunError> { + use crate::assistant::codex_app_server as aps; + + let existing_thread_id = session.context.cli_session_id.clone(); + let prompt = prepare_prompt( + deps, + session, + run_id, + trigger, + CliProviderRuntime::Codex.metadata_source(), + CliProviderRuntime::Codex.display_name(), + existing_thread_id.is_none(), + ) + .await?; + let system_prompt = system_prompt_text(&deps.app, session, trigger).await; + let prompt = codex_turn_prompt(&system_prompt, &prompt); + let prompt_chars = prompt.chars().count(); + if prompt_chars > CODEX_TURN_INPUT_MAX_CHARS { + return Err(LocalAgentRunError::failed(codex_input_too_large_message( + prompt_chars, + ))); + } + + let assistant_message = ensure_assistant_message_slot( + deps, + session, + run_id, + CliProviderRuntime::Codex.metadata_source(), + assistant_slot, + ) + .await?; + + let configured_binary = connection + .base_url + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("codex"); + let binary = crate::providers::resolve_command_path(configured_binary) + .unwrap_or_else(|| configured_binary.to_string()); + let workspace_root = workspace_root_for_session(deps, session); + + let command = crate::providers::build_host_cli_command( + &binary, + &[ + (CODEX_MCP_TOKEN_ENV, mcp_token), + ("MCP_TIMEOUT", MCP_STARTUP_TIMEOUT_MS), + ], + workspace_root.as_deref(), + ); + let mut transport = + aps::AppServerTransport::spawn(command).map_err(LocalAgentRunError::failed)?; + + let stderr_tail: Arc>> = + Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_TAIL_LINES))); + if let Some(stderr) = transport.take_stderr() { + spawn_stderr_logger( + run_id.to_string(), + CliProviderRuntime::Codex.display_name(), + stderr, + stderr_tail.clone(), + ); + } + + let client_version = env!("CARGO_PKG_VERSION"); + let model = { + let m = connection.model_id.trim(); + (!m.is_empty() && m != "default").then_some(m) + }; + let cwd = workspace_root + .as_ref() + .map(|p| p.to_string_lossy().into_owned()); + let tool_timeout = crate::assistant::tools::CLI_MCP_CLIENT_TIMEOUT.as_secs(); + + // Handshake: initialize -> thread/start|resume. Startup notifications are + // ignored by the request/await helper. + aps_request_await( + &mut transport, + cancel_token, + aps::initialize_request(1, client_version), + ) + .await?; + let thread_request = match existing_thread_id.as_deref() { + Some(thread_id) => aps::thread_resume_request(2, thread_id, model), + None => aps::thread_start_request( + 2, + cwd.as_deref(), + model, + mcp_url, + CODEX_MCP_TOKEN_ENV, + tool_timeout, + ), + }; + let thread_result = aps_request_await(&mut transport, cancel_token, thread_request).await?; + if let Some(thread_id) = thread_result + .get("thread") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + { + set_cli_session_id(deps, session, thread_id.to_string(), CODEX_PROVIDER_ID).await?; + } + let thread_id = + session.context.cli_session_id.clone().ok_or_else(|| { + LocalAgentRunError::failed("codex app-server did not return a thread id") + })?; + + // Fire the turn. We do not block on its response; the turn is tracked via + // notifications in the loop below. + let mut input = vec![aps::text_user_input(&prompt)]; + for image_path in resolve_codex_image_paths(deps, session, run_id).await { + input.push(aps::local_image_user_input(&image_path.to_string_lossy())); + } + transport + .send(&aps::turn_start_request(3, &thread_id, input)) + .await + .map_err(LocalAgentRunError::failed)?; + + let mut state = CodexStreamState::new(); + let mut usage: Option = None; + let mut result_error: Option = None; + + loop { + let message = tokio::select! { + _ = cancel_token.cancelled() => { + let _ = transport.send(&aps::turn_interrupt_request(99, &thread_id)).await; + transport.kill().await; + finalize_assistant_message_from_parts( + deps, session, run_id, &assistant_message, &state.parts, + ) + .await?; + return Err(LocalAgentRunError::Cancelled { usage }); + } + msg = transport.recv() => msg, + }; + let Some(message) = message else { + break; // app-server closed stdout + }; + if aps::is_server_request(&message) { + if let Some(id) = message.get("id") { + let method = message + .get("method") + .and_then(Value::as_str) + .unwrap_or_default(); + let _ = transport + .send(&aps::server_request_response(id, method)) + .await; + } + continue; + } + let Some(method) = aps::notification_method(&message) else { + continue; // responses are not needed after the handshake + }; + let turn_done = handle_app_server_notification( + deps, + session, + run_id, + &assistant_message, + method, + &message, + &mut state, + &mut usage, + &mut result_error, + ) + .await?; + if turn_done { + break; + } + } + + transport.kill().await; + finalize_assistant_message_from_parts(deps, session, run_id, &assistant_message, &state.parts) + .await?; + + if let Some(message) = result_error { + let enriched = append_stderr_tail(&message, &stderr_tail); + return Err(LocalAgentRunError::Failed { + message: enriched, + usage, + }); + } + Ok(usage) +} + +/// Send a JSON-RPC request and read messages until its matching response, +/// answering any interleaved server->client requests and ignoring startup +/// notifications. Returns the response `result` (or a failure on error/close). +async fn aps_request_await( + transport: &mut codex_app_server::AppServerTransport, + cancel_token: &CancellationToken, + request: Value, +) -> Result { + let want_id = request.get("id").and_then(Value::as_i64); + transport + .send(&request) + .await + .map_err(LocalAgentRunError::failed)?; + loop { + let message = tokio::select! { + _ = cancel_token.cancelled() => return Err(LocalAgentRunError::Cancelled { usage: None }), + msg = transport.recv() => msg, + }; + let Some(message) = message else { + return Err(LocalAgentRunError::failed( + "codex app-server closed the connection during handshake", + )); + }; + if codex_app_server::is_server_request(&message) { + if let Some(id) = message.get("id") { + let method = message + .get("method") + .and_then(Value::as_str) + .unwrap_or_default(); + let _ = transport + .send(&codex_app_server::server_request_response(id, method)) + .await; + } + continue; + } + if codex_app_server::is_response(&message) + && codex_app_server::response_id(&message) == want_id + { + if let Some(error) = message.get("error") { + let msg = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("codex app-server request failed"); + return Err(LocalAgentRunError::failed(msg.to_string())); + } + return Ok(message.get("result").cloned().unwrap_or(Value::Null)); + } + // Ignore notifications that arrive during the handshake. + } +} + +/// Handle one app-server notification. Returns `true` when the turn is finished +/// (`turn/completed`). Reuses [`handle_codex_item`] by normalizing app-server +/// items into the exec item shape. +#[allow(clippy::too_many_arguments)] +async fn handle_app_server_notification( + deps: &AssistantDeps, + session: &mut AssistantSession, + run_id: &str, + assistant_message: &AssistantMessage, + method: &str, + message: &Value, + state: &mut CodexStreamState, + usage: &mut Option, + result_error: &mut Option, +) -> Result { + let null = Value::Null; + let params = message.get("params").unwrap_or(&null); + match method { + "thread/started" => { + if let Some(thread_id) = params + .get("thread") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + { + set_cli_session_id(deps, session, thread_id.to_string(), CODEX_PROVIDER_ID).await?; + } + } + "item/started" => { + // Persist the running state of an MCP tool call early so the UI can + // show it in-flight, mirroring the exec `item.started` handling. + if let Some(item) = params.get("item") { + if item.get("type").and_then(Value::as_str) == Some("mcpToolCall") { + if let Some(normalized) = normalize_app_server_item(item) { + handle_codex_item( + deps, + session, + run_id, + assistant_message, + state, + &normalized, + false, + ) + .await?; + } + } + } + } + "item/completed" => { + if let Some(item) = params.get("item") { + if let Some(normalized) = normalize_app_server_item(item) { + handle_codex_item( + deps, + session, + run_id, + assistant_message, + state, + &normalized, + true, + ) + .await?; + } + } + } + "thread/tokenUsage/updated" => { + if let Some(parsed) = run_usage_from_app_server(params.get("tokenUsage")) { + *usage = Some(parsed); + } + } + "error" => { + if result_error.is_none() { + let error = params.get("error").unwrap_or(&null); + let msg = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("Codex run failed"); + let code = error.get("codexErrorInfo").and_then(Value::as_str); + *result_error = Some(codex_app_server::classify_error_message(code, msg)); + } + } + "turn/completed" => { + let turn = params.get("turn"); + if turn.and_then(|t| t.get("status")).and_then(Value::as_str) == Some("failed") + && result_error.is_none() + { + if let Some(error) = turn.and_then(|t| t.get("error")) { + let msg = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("Codex turn failed"); + let code = error.get("codexErrorInfo").and_then(Value::as_str); + *result_error = Some(codex_app_server::classify_error_message(code, msg)); + } + } + return Ok(true); + } + _ => {} + } + Ok(false) +} + +/// Normalize an app-server `ThreadItem` (camelCase) into the exec item shape so +/// the shared [`handle_codex_item`] can persist/emit it unchanged. Returns +/// `None` for item kinds CLAI does not surface (userMessage, plan, ...). +fn normalize_app_server_item(item: &Value) -> Option { + let kind = item.get("type").and_then(Value::as_str)?; + match kind { + "agentMessage" => Some(serde_json::json!({ + "type": "agent_message", + "text": item.get("text").cloned().unwrap_or(Value::Null), + })), + "reasoning" => { + let text = app_server_reasoning_text(item); + if text.is_empty() { + None + } else { + Some(serde_json::json!({ "type": "reasoning", "text": text })) + } + } + // These already share the exec field names (id/server/tool/arguments/ + // result/error/status, command, query); only the discriminant differs. + "mcpToolCall" | "commandExecution" | "fileChange" | "webSearch" => { + let mut normalized = item.clone(); + let exec_type = match kind { + "mcpToolCall" => "mcp_tool_call", + "commandExecution" => "command_execution", + "fileChange" => "file_change", + _ => "web_search", + }; + normalized["type"] = Value::String(exec_type.to_string()); + Some(normalized) + } + _ => None, + } +} + +/// Flatten an app-server `reasoning` item's `content`/`summary` arrays into a +/// single string. Elements are either `{ "text": "..." }` objects or bare +/// strings; anything else is skipped. +fn app_server_reasoning_text(item: &Value) -> String { + let mut out: Vec = Vec::new(); + for key in ["content", "summary"] { + if let Some(arr) = item.get(key).and_then(Value::as_array) { + for element in arr { + if let Some(text) = element.get("text").and_then(Value::as_str) { + if !text.is_empty() { + out.push(text.to_string()); + } + } else if let Some(text) = element.as_str() { + if !text.is_empty() { + out.push(text.to_string()); + } + } + } + } + if !out.is_empty() { + break; // prefer content; fall back to summary only if content empty + } + } + out.join("\n") +} + +/// Map an app-server `ThreadTokenUsage` (`total` breakdown) to [`RunUsage`]. +fn run_usage_from_app_server(token_usage: Option<&Value>) -> Option { + let total = token_usage?.get("total")?; + let get = |key: &str| { + total + .get(key) + .and_then(Value::as_i64) + .and_then(|v| u64::try_from(v).ok()) + }; + let input_tokens = get("inputTokens"); + let output_tokens = get("outputTokens"); + let reasoning_tokens = get("reasoningOutputTokens"); + let total_tokens = + get("totalTokens").or(match (input_tokens, output_tokens, reasoning_tokens) { + (None, None, None) => None, + _ => Some( + input_tokens.unwrap_or(0) + + output_tokens.unwrap_or(0) + + reasoning_tokens.unwrap_or(0), + ), + }); + if input_tokens.is_none() + && output_tokens.is_none() + && reasoning_tokens.is_none() + && total_tokens.is_none() + { + return None; + } + Some(RunUsage { + input_tokens, + output_tokens, + reasoning_tokens, + total_tokens, + }) +} + fn codex_turn_prompt(system_prompt: &str, prompt: &str) -> String { format!( "System instructions for this CLAI run:\n{}\n\nUse the connected `clai` MCP tools for workspace work, shell execution, file access, and user interaction.\n\nUser/task prompt:\n{}", From 191e6bcafa36cf325acb0741cce2002334f56754 Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:26:33 +0200 Subject: [PATCH 3/6] feat(codex): steer queued messages into the live turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third stage: real-time mid-turn input via turn/steer — the payoff over the exec stop-and-restart model. A 400ms poll in the app-server driver reads pending queued messages for this connection while a turn is active and injects their text into the running turn with turn/steer, guarded by the active expectedTurnId. Delivery is confirmed on the steer response: accepted -> mark delivered + emit QueuedMessagesDelivered; rejected (turn ended / id mismatch) -> leave queued so the existing followup-run path delivers them. Image-bearing messages are deferred to the followup run (which resolves image paths). In-flight tracking prevents double-sending before the response lands. --- src-tauri/src/assistant/codex_app_server.rs | 1 - src-tauri/src/assistant/local_agent.rs | 184 +++++++++++++++++++- 2 files changed, 181 insertions(+), 4 deletions(-) diff --git a/src-tauri/src/assistant/codex_app_server.rs b/src-tauri/src/assistant/codex_app_server.rs index 012f4a0..1fbf0ae 100644 --- a/src-tauri/src/assistant/codex_app_server.rs +++ b/src-tauri/src/assistant/codex_app_server.rs @@ -259,7 +259,6 @@ pub(crate) fn turn_start_request(id: i64, thread_id: &str, input: Vec) -> /// Build `turn/steer` — inject input into the active turn. Guarded by /// `expectedTurnId`: the server rejects it if that turn is no longer active. -#[allow(dead_code)] // wired by the steering stage pub(crate) fn turn_steer_request( id: i64, thread_id: &str, diff --git a/src-tauri/src/assistant/local_agent.rs b/src-tauri/src/assistant/local_agent.rs index cc685ee..39128a0 100644 --- a/src-tauri/src/assistant/local_agent.rs +++ b/src-tauri/src/assistant/local_agent.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::path::PathBuf; use std::process::Stdio; use std::sync::{Arc, Mutex}; @@ -36,6 +36,9 @@ const CLAUDE_DISABLED_TOOLS: &str = "Bash,Read,Edit,Write,Glob,Grep,WebFetch,WebSearch,Task,TodoWrite,NotebookEdit,LSP"; const CODEX_MCP_TOKEN_ENV: &str = "CLAI_MCP_TOKEN"; const CODEX_TURN_INPUT_MAX_CHARS: usize = 1_048_576; +/// How often the app-server driver polls the queue to steer new user +/// messages into the active turn. +const STEER_POLL_INTERVAL_MS: u64 = 400; /// MCP server *startup* timeout (ms) for CLI providers that read `MCP_TIMEOUT` /// (Claude Code, Codex). Distinct from the per-tool backstop @@ -1626,11 +1629,28 @@ async fn run_codex_turn_app_server( let mut state = CodexStreamState::new(); let mut usage: Option = None; let mut result_error: Option = None; + // Set on turn/started, cleared on turn/completed. Steering is only valid + // while a turn is active, and `turn/steer` must carry this exact id. + let mut active_turn_id: Option = None; + // Monotonic id for client-initiated requests after the handshake + // (steer/interrupt), kept distinct from the handshake ids 1/2/3. + let mut next_request_id: i64 = 10; + // steer request id -> queued message ids carried by that steer, awaiting + // the server's accept/reject response. + let mut pending_steer: HashMap> = HashMap::new(); + // Message ids with a steer in flight so a later poll tick does not + // re-send them before the response lands. + let mut inflight_steer: HashSet = HashSet::new(); + let mut steer_poll = + tokio::time::interval(std::time::Duration::from_millis(STEER_POLL_INTERVAL_MS)); + steer_poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { let message = tokio::select! { _ = cancel_token.cancelled() => { - let _ = transport.send(&aps::turn_interrupt_request(99, &thread_id)).await; + let _ = transport + .send(&aps::turn_interrupt_request(next_request_id, &thread_id)) + .await; transport.kill().await; finalize_assistant_message_from_parts( deps, session, run_id, &assistant_message, &state.parts, @@ -1638,6 +1658,33 @@ async fn run_codex_turn_app_server( .await?; return Err(LocalAgentRunError::Cancelled { usage }); } + _ = steer_poll.tick() => { + // Steer queued messages into the *live* turn (the win over the + // exec stop-and-restart model). Only while a turn is active. + if let Some(turn_id) = active_turn_id.clone() { + if let Some((request, message_ids)) = build_codex_steer( + deps, + session, + connection.id.as_str(), + &thread_id, + &turn_id, + next_request_id, + &mut inflight_steer, + ) + .await + { + if transport.send(&request).await.is_ok() { + pending_steer.insert(next_request_id, message_ids); + } else { + for id in message_ids { + inflight_steer.remove(&id); + } + } + next_request_id += 1; + } + } + continue; + } msg = transport.recv() => msg, }; let Some(message) = message else { @@ -1655,8 +1702,26 @@ async fn run_codex_turn_app_server( } continue; } + if aps::is_response(&message) { + // The only responses we act on are steer accept/reject; everything + // else (handshake echoes, turn/start ack) is informational. + if let Some(id) = aps::response_id(&message) { + if let Some(message_ids) = pending_steer.remove(&id) { + resolve_codex_steer_response( + deps, + session, + run_id, + &message, + message_ids, + &mut inflight_steer, + ) + .await; + } + } + continue; + } let Some(method) = aps::notification_method(&message) else { - continue; // responses are not needed after the handshake + continue; }; let turn_done = handle_app_server_notification( deps, @@ -1668,6 +1733,7 @@ async fn run_codex_turn_app_server( &mut state, &mut usage, &mut result_error, + &mut active_turn_id, ) .await?; if turn_done { @@ -1754,6 +1820,7 @@ async fn handle_app_server_notification( state: &mut CodexStreamState, usage: &mut Option, result_error: &mut Option, + active_turn_id: &mut Option, ) -> Result { let null = Value::Null; let params = message.get("params").unwrap_or(&null); @@ -1767,6 +1834,13 @@ async fn handle_app_server_notification( set_cli_session_id(deps, session, thread_id.to_string(), CODEX_PROVIDER_ID).await?; } } + "turn/started" => { + *active_turn_id = params + .get("turn") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + .map(str::to_string); + } "item/started" => { // Persist the running state of an MCP tool call early so the UI can // show it in-flight, mirroring the exec `item.started` handling. @@ -1833,6 +1907,7 @@ async fn handle_app_server_notification( *result_error = Some(codex_app_server::classify_error_message(code, msg)); } } + *active_turn_id = None; return Ok(true); } _ => {} @@ -1840,6 +1915,109 @@ async fn handle_app_server_notification( Ok(false) } +/// Read pending queued user messages for this connection and build a +/// `turn/steer` request carrying their text. Image-bearing messages are left +/// queued for the followup run (which resolves image paths); steering carries +/// text only. Marks the chosen ids in-flight so a later tick will not re-send +/// them before the accept/reject response lands. Returns `None` when there is +/// nothing steerable right now. +async fn build_codex_steer( + deps: &AssistantDeps, + session: &AssistantSession, + connection_id: &str, + thread_id: &str, + turn_id: &str, + request_id: i64, + inflight: &mut HashSet, +) -> Option<(Value, Vec)> { + // Interrupting while the user is being asked a question would tear the + // question down; leave messages queued until it resolves. + if crate::assistant::tools::ask_user::session_has_pending_ask(&session.id) { + return None; + } + let pending = repository::list_pending_queued_messages(&deps.pool, &session.id) + .await + .ok()?; + let mut input = Vec::new(); + let mut ids = Vec::new(); + for queued in pending { + if queued.connection_id != connection_id || inflight.contains(&queued.message.id) { + continue; + } + if queued + .message + .content + .iter() + .any(|part| matches!(part, ContentPart::Image { .. })) + { + continue; // defer image messages to the followup run + } + let text = message_text(&queued.message); + if text.trim().is_empty() { + continue; + } + input.push(codex_app_server::text_user_input(&text)); + ids.push(queued.message.id.clone()); + } + if input.is_empty() { + return None; + } + for id in &ids { + inflight.insert(id.clone()); + } + Some(( + codex_app_server::turn_steer_request(request_id, thread_id, turn_id, input), + ids, + )) +} + +/// Resolve a `turn/steer` response: on accept, mark the carried messages +/// delivered and notify the UI; on reject (turn ended / id mismatch) leave them +/// queued so the followup run delivers them. Always clears their in-flight mark. +async fn resolve_codex_steer_response( + deps: &AssistantDeps, + session: &AssistantSession, + run_id: &str, + response: &Value, + message_ids: Vec, + inflight: &mut HashSet, +) { + for id in &message_ids { + inflight.remove(id); + } + if response.get("error").is_some() { + tracing::info!( + run_id, + count = message_ids.len(), + "codex turn/steer rejected; leaving messages queued for the followup run" + ); + return; + } + match repository::mark_queued_messages_delivered(&deps.pool, &session.id, run_id, &message_ids) + .await + { + Ok(()) => { + let _ = emit_event( + &deps.app, + session, + Some(run_id), + AssistantUiEvent::QueuedMessagesDelivered { + message_ids: message_ids.clone(), + }, + ); + tracing::info!( + run_id, + count = message_ids.len(), + "Steered queued user message(s) into the live Codex turn" + ); + } + Err(error) => tracing::warn!( + run_id, %error, + "codex steer accepted but marking delivered failed; the followup run may re-deliver" + ), + } +} + /// Normalize an app-server `ThreadItem` (camelCase) into the exec item shape so /// the shared [`handle_codex_item`] can persist/emit it unchanged. Returns /// `None` for item kinds CLAI does not surface (userMessage, plan, ...). From 8c2ad9b5d8e5fdab442acd1720cb6b17c586f33d Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:29:02 +0200 Subject: [PATCH 4/6] test(codex): cover app-server item/usage normalization + document flag Fourth stage: unit tests for the pure mapping layer (normalize_app_server_item discriminant mapping, reasoning content/summary flattening, token-usage breakdown -> RunUsage) and an 'Enabling / testing' note in the module rustdoc (repo convention documents env flags in code, not markdown). Default stays codex exec; flipping the default is deferred until the app-server path is validated live. --- src-tauri/src/assistant/codex_app_server.rs | 9 +++ src-tauri/src/assistant/local_agent.rs | 74 +++++++++++++++++++++ 2 files changed, 83 insertions(+) diff --git a/src-tauri/src/assistant/codex_app_server.rs b/src-tauri/src/assistant/codex_app_server.rs index 1fbf0ae..8d6c49d 100644 --- a/src-tauri/src/assistant/codex_app_server.rs +++ b/src-tauri/src/assistant/codex_app_server.rs @@ -15,6 +15,15 @@ //! //! Gated behind `CLAI_CODEX_APP_SERVER` (default off) while it bakes; `codex //! exec` remains the default transport. +//! +//! # Enabling / testing +//! +//! Set `CLAI_CODEX_APP_SERVER=1` in the environment CLAI launches under, then +//! use a Codex connection. Turns run over `codex app-server`; a message sent +//! while a turn is in flight is injected via `turn/steer` (watch for the +//! "Steered queued user message(s) into the live Codex turn" log line) instead +//! of interrupting/restarting the process. Unset the var to fall back to +//! `codex exec`. use std::process::Stdio; diff --git a/src-tauri/src/assistant/local_agent.rs b/src-tauri/src/assistant/local_agent.rs index 39128a0..51e80b5 100644 --- a/src-tauri/src/assistant/local_agent.rs +++ b/src-tauri/src/assistant/local_agent.rs @@ -5331,4 +5331,78 @@ mod tests { "Bearer token" ); } + + #[test] + fn normalize_app_server_item_maps_discriminants_to_exec_shape() { + // agentMessage -> agent_message with text preserved. + let agent = serde_json::json!({ "type": "agentMessage", "id": "a1", "text": "hello" }); + let norm = normalize_app_server_item(&agent).unwrap(); + assert_eq!(norm["type"], "agent_message"); + assert_eq!(norm["text"], "hello"); + + // mcpToolCall keeps its exec-aligned fields, only the discriminant changes. + let tool = serde_json::json!({ + "type": "mcpToolCall", "id": "t1", "server": "clai", "tool": "bash_exec", + "arguments": { "command": "ls" }, "status": "completed" + }); + let norm = normalize_app_server_item(&tool).unwrap(); + assert_eq!(norm["type"], "mcp_tool_call"); + assert_eq!(norm["server"], "clai"); + assert_eq!(norm["tool"], "bash_exec"); + assert_eq!(norm["arguments"]["command"], "ls"); + + for (kind, exec) in [ + ("commandExecution", "command_execution"), + ("fileChange", "file_change"), + ("webSearch", "web_search"), + ] { + let item = serde_json::json!({ "type": kind, "id": "x" }); + assert_eq!(normalize_app_server_item(&item).unwrap()["type"], exec); + } + + // userMessage and unknown kinds are not surfaced. + assert!(normalize_app_server_item(&serde_json::json!({ "type": "userMessage" })).is_none()); + assert!(normalize_app_server_item(&serde_json::json!({ "type": "plan" })).is_none()); + } + + #[test] + fn app_server_reasoning_text_flattens_content_then_summary() { + let item = serde_json::json!({ + "type": "reasoning", + "content": [{ "text": "step one" }, { "text": "step two" }], + "summary": [{ "text": "ignored while content is present" }] + }); + assert_eq!(app_server_reasoning_text(&item), "step one\nstep two"); + + // Falls back to summary only when content is empty. + let summary_only = serde_json::json!({ + "type": "reasoning", "content": [], "summary": [{ "text": "s" }] + }); + assert_eq!(app_server_reasoning_text(&summary_only), "s"); + + // Empty reasoning yields no text (item is then dropped by the caller). + let empty = serde_json::json!({ "type": "reasoning" }); + assert!(app_server_reasoning_text(&empty).is_empty()); + assert!(normalize_app_server_item(&empty).is_none()); + } + + #[test] + fn run_usage_from_app_server_maps_total_breakdown() { + let token_usage = serde_json::json!({ + "total": { + "inputTokens": 100, "outputTokens": 40, + "reasoningOutputTokens": 10, "cachedInputTokens": 5, "totalTokens": 150 + }, + "last": {} + }); + let usage = run_usage_from_app_server(Some(&token_usage)).unwrap(); + assert_eq!(usage.input_tokens, Some(100)); + assert_eq!(usage.output_tokens, Some(40)); + assert_eq!(usage.reasoning_tokens, Some(10)); + assert_eq!(usage.total_tokens, Some(150)); + + // No usage at all -> None (don't clobber a prior value with zeros). + assert!(run_usage_from_app_server(None).is_none()); + assert!(run_usage_from_app_server(Some(&serde_json::json!({}))).is_none()); + } } From 1cc3967253f8c0324ac27d4555456c37fd36106f Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sun, 5 Jul 2026 11:22:09 +0200 Subject: [PATCH 5/6] fix(codex): bound the app-server handshake + fail on mid-turn close Addresses the validated findings from the external review of this PR: - aps_request_await now bounds each handshake request (initialize, thread/start|resume) with a 30s deadline, so a wedged `codex app-server` fails fast with a clear message instead of hanging the run until the user cancels. - Stdout closing without `turn/completed` is now surfaced as a run failure instead of finalizing a silently truncated turn as success (found while verifying the review; partial parts are still persisted). - Regression test: agentMessage without a `text` field normalizes to an explicit null in the exec shape. - Stale docstring fixed (steering is in this PR, not a follow-up) and the per-turn/best-effort lifetime of the steer bookkeeping documented. Rejected from the review (verified against the code): the two "major" findings assume pending_steer outlives the turn; it is a local of run_codex_turn_app_server and turn/completed breaks the loop, so it lives exactly one turn and ids cannot collide across turns. --- src-tauri/src/assistant/local_agent.rs | 47 ++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 6 deletions(-) diff --git a/src-tauri/src/assistant/local_agent.rs b/src-tauri/src/assistant/local_agent.rs index 51e80b5..09d1e4e 100644 --- a/src-tauri/src/assistant/local_agent.rs +++ b/src-tauri/src/assistant/local_agent.rs @@ -40,6 +40,11 @@ const CODEX_TURN_INPUT_MAX_CHARS: usize = 1_048_576; /// messages into the active turn. const STEER_POLL_INTERVAL_MS: u64 = 400; +/// Bound on each app-server handshake request (`initialize`, +/// `thread/start|resume`). Generous: covers a cold `codex` start on a slow +/// disk, but converts a wedged server into a clear fast-fail. +const APP_SERVER_HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// MCP server *startup* timeout (ms) for CLI providers that read `MCP_TIMEOUT` /// (Claude Code, Codex). Distinct from the per-tool backstop /// ([`CLI_MCP_CLIENT_TIMEOUT`]): a slow first handshake under load shouldn't @@ -1494,10 +1499,10 @@ async fn run_codex_turn( // --------------------------------------------------------------------------- /// Run a Codex turn over the `codex app-server` JSON-RPC transport instead of -/// `codex exec`. Behaviour-parity with [`run_codex_turn`] for now (streaming, -/// tool calls, usage, errors); `turn/steer` mid-run delivery is layered on in a -/// follow-up commit. Reuses the shared [`CodexStreamState`] + tool-call helpers -/// by normalizing app-server items into the exec item shape. +/// `codex exec`: streaming, tool calls, usage, errors, plus real-time mid-run +/// input via `turn/steer` (queued messages are injected into the live turn). +/// Reuses the shared [`CodexStreamState`] + tool-call helpers by normalizing +/// app-server items into the exec item shape. #[allow(clippy::too_many_arguments)] async fn run_codex_turn_app_server( deps: &AssistantDeps, @@ -1636,7 +1641,10 @@ async fn run_codex_turn_app_server( // (steer/interrupt), kept distinct from the handshake ids 1/2/3. let mut next_request_id: i64 = 10; // steer request id -> queued message ids carried by that steer, awaiting - // the server's accept/reject response. + // the server's accept/reject response. Best-effort and per-turn: the map + // lives only until `turn/completed` breaks the loop, and an entry whose + // response never arrives just leaves its messages pending in the DB for + // the follow-up run (no delivery is lost, nothing outlives the turn). let mut pending_steer: HashMap> = HashMap::new(); // Message ids with a steer in flight so a later poll tick does not // re-send them before the response lands. @@ -1688,7 +1696,13 @@ async fn run_codex_turn_app_server( msg = transport.recv() => msg, }; let Some(message) = message else { - break; // app-server closed stdout + // Stdout closed without `turn/completed`: the server died mid-turn. + // Surface it as a failure so the run isn't finalized as a silently + // truncated success (partial parts are still persisted below). + if result_error.is_none() { + result_error = Some("codex app-server closed the connection mid-turn".to_string()); + } + break; }; if aps::is_server_request(&message) { if let Some(id) = message.get("id") { @@ -1764,6 +1778,10 @@ async fn aps_request_await( request: Value, ) -> Result { let want_id = request.get("id").and_then(Value::as_i64); + // Bound the handshake so a stalled `codex app-server` (broken install, + // auth stall) fails fast with a clear error instead of hanging the run + // until the user cancels. Mirrors the MCP_STARTUP_TIMEOUT_MS posture. + let deadline = tokio::time::Instant::now() + APP_SERVER_HANDSHAKE_TIMEOUT; transport .send(&request) .await @@ -1771,6 +1789,12 @@ async fn aps_request_await( loop { let message = tokio::select! { _ = cancel_token.cancelled() => return Err(LocalAgentRunError::Cancelled { usage: None }), + _ = tokio::time::sleep_until(deadline) => { + return Err(LocalAgentRunError::failed(format!( + "codex app-server did not answer the handshake within {}s", + APP_SERVER_HANDSHAKE_TIMEOUT.as_secs() + ))); + } msg = transport.recv() => msg, }; let Some(message) = message else { @@ -5365,6 +5389,17 @@ mod tests { assert!(normalize_app_server_item(&serde_json::json!({ "type": "plan" })).is_none()); } + #[test] + fn normalize_app_server_item_maps_missing_agent_text_to_null() { + // Some servers may omit `text` on agentMessage; the normalized exec + // shape must carry an explicit null (=> handled as "no text"), not + // panic or invent content. + let item = serde_json::json!({"id": "item_1", "type": "agentMessage"}); + let normalized = normalize_app_server_item(&item).expect("normalizes"); + assert_eq!(normalized["type"], "agent_message"); + assert!(normalized["text"].is_null()); + } + #[test] fn app_server_reasoning_text_flattens_content_then_summary() { let item = serde_json::json!({ From 6fe049c49a15e3c09e6c611c0883f7dda4714b45 Mon Sep 17 00:00:00 2001 From: juan <2930882+juacker@users.noreply.github.com> Date: Sun, 5 Jul 2026 12:07:00 +0200 Subject: [PATCH 6/6] fix(codex): gate the app-server native shell + order steered messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two issues found running the app-server branch (workspace test session): 1. SECURITY (regression vs exec): the app-server thread/start+resume set approvalPolicy=never + sandbox=danger-full-access but left Codex's built-in shell_tool ENABLED, so Codex ran shell commands through its own native path (observed: pwd/date/ls/python), bypassing CLAI's permission checks and bwrap sandbox entirely. The exec path passes `--disable shell_tool` for exactly this reason. Fix: the thread config now sets `features.shell_tool=false` (the documented equivalent of `--disable shell_tool`) on both start and resume, so Codex has no unmediated execution path — every command routes through the gated MCP `bash_exec`. thread/resume now also re-applies the full config (each turn spawns a fresh app-server process, so it must be re-set). 2. ORDERING: the whole turn accreted into one assistant message whose created_at is frozen at turn start, so a mid-run (steered) user message — created when the user hit send — sorted AFTER the entire reply, showing the handling of a comment before the comment itself. Fix: on steer accept, close the current assistant message and open a fresh one (split_codex_assistant_message), yielding the natural order pre-steer output -> user message -> post-steer output. An empty leading placeholder is dropped instead of left as an empty bubble. Tests: thread_start/thread_resume now assert features.shell_tool=false; full lib suite 742/742, clippy -D warnings clean. --- src-tauri/src/assistant/codex_app_server.rs | 52 +++++++++++-- src-tauri/src/assistant/local_agent.rs | 81 +++++++++++++++++++-- 2 files changed, 122 insertions(+), 11 deletions(-) diff --git a/src-tauri/src/assistant/codex_app_server.rs b/src-tauri/src/assistant/codex_app_server.rs index 8d6c49d..3cb59b1 100644 --- a/src-tauri/src/assistant/codex_app_server.rs +++ b/src-tauri/src/assistant/codex_app_server.rs @@ -202,8 +202,15 @@ pub(crate) fn local_image_user_input(path: &str) -> Value { /// The `config` object mirroring the `-c mcp_servers.clai.*` flags the `exec` /// path passes, so the app-server turn can reach the same local MCP server. -pub(crate) fn mcp_servers_config(mcp_url: &str, token_env: &str, tool_timeout_secs: u64) -> Value { +pub(crate) fn thread_config(mcp_url: &str, token_env: &str, tool_timeout_secs: u64) -> Value { json!({ + // Disable Codex's built-in `shell_tool` (and thus its unmediated native + // command execution). `--disable shell_tool` on the exec path is + // documented as `-c features.shell_tool=false`; this is its app-server + // equivalent. Without it, Codex runs shell commands itself, bypassing + // CLAI's permission checks and bwrap sandbox entirely — every command + // must instead go through the gated MCP `bash_exec` tool below. + "features": { "shell_tool": false }, "mcp_servers": { "clai": { "url": mcp_url, @@ -219,8 +226,10 @@ pub(crate) fn mcp_servers_config(mcp_url: &str, token_env: &str, tool_timeout_se /// Build `thread/start` params. `approvalPolicy: never` + `sandbox: /// danger-full-access` is the app-server parallel of `exec`'s /// `--dangerously-bypass-approvals-and-sandbox`: CLAI provides the external -/// sandbox (bwrap) and permission system through its MCP tools, so Codex must -/// not gate or sandbox anything itself. +/// sandbox (bwrap) and permission system through its MCP tools. Crucially, the +/// `config` (see [`thread_config`]) also disables Codex's native `shell_tool`, +/// so Codex has *no* unmediated execution path — every command routes through +/// the permission-gated MCP `bash_exec`, matching the exec path exactly. #[allow(clippy::too_many_arguments)] pub(crate) fn thread_start_request( id: i64, @@ -233,7 +242,7 @@ pub(crate) fn thread_start_request( let mut params = json!({ "approvalPolicy": "never", "sandbox": "danger-full-access", - "config": mcp_servers_config(mcp_url, token_env, tool_timeout_secs), + "config": thread_config(mcp_url, token_env, tool_timeout_secs), }); if let Some(cwd) = cwd { params["cwd"] = json!(cwd); @@ -244,12 +253,24 @@ pub(crate) fn thread_start_request( request(id, "thread/start", params) } -/// Build `thread/resume` params for an existing Codex thread id. -pub(crate) fn thread_resume_request(id: i64, thread_id: &str, model: Option<&str>) -> Value { +/// Build `thread/resume` params for an existing Codex thread id. Re-applies the +/// same bypass + [`thread_config`] as `thread/start` (each turn spawns a fresh +/// `codex app-server` process, so the shell_tool disable and MCP registration +/// must be set again on resume, mirroring how the exec path re-passes its flags +/// on `exec resume`). +pub(crate) fn thread_resume_request( + id: i64, + thread_id: &str, + model: Option<&str>, + mcp_url: &str, + token_env: &str, + tool_timeout_secs: u64, +) -> Value { let mut params = json!({ "threadId": thread_id, "approvalPolicy": "never", "sandbox": "danger-full-access", + "config": thread_config(mcp_url, token_env, tool_timeout_secs), }); if let Some(model) = model { params["model"] = json!(model); @@ -415,6 +436,25 @@ mod tests { assert_eq!(clai["bearer_token_env_var"], "CLAI_MCP_TOKEN"); assert_eq!(clai["enabled"], true); assert_eq!(clai["tool_timeout_sec"], 3600); + // Codex's native shell must be disabled so it has no ungated + // execution path (parity with exec's `--disable shell_tool`). + assert_eq!(params["config"]["features"]["shell_tool"], false); + } + + #[test] + fn thread_resume_disables_shell_tool_and_carries_mcp_config() { + let req = thread_resume_request(2, "thread-1", Some("gpt-5.5"), "u", "T", 42); + let params = &req["params"]; + assert_eq!(req["method"], "thread/resume"); + assert_eq!(params["threadId"], "thread-1"); + assert_eq!(params["approvalPolicy"], "never"); + assert_eq!(params["sandbox"], "danger-full-access"); + assert_eq!(params["config"]["features"]["shell_tool"], false); + assert_eq!(params["config"]["mcp_servers"]["clai"]["url"], "u"); + assert_eq!( + params["config"]["mcp_servers"]["clai"]["tool_timeout_sec"], + 42 + ); } #[test] diff --git a/src-tauri/src/assistant/local_agent.rs b/src-tauri/src/assistant/local_agent.rs index 09d1e4e..bea0645 100644 --- a/src-tauri/src/assistant/local_agent.rs +++ b/src-tauri/src/assistant/local_agent.rs @@ -1537,7 +1537,7 @@ async fn run_codex_turn_app_server( ))); } - let assistant_message = ensure_assistant_message_slot( + let mut assistant_message = ensure_assistant_message_slot( deps, session, run_id, @@ -1597,7 +1597,14 @@ async fn run_codex_turn_app_server( ) .await?; let thread_request = match existing_thread_id.as_deref() { - Some(thread_id) => aps::thread_resume_request(2, thread_id, model), + Some(thread_id) => aps::thread_resume_request( + 2, + thread_id, + model, + mcp_url, + CODEX_MCP_TOKEN_ENV, + tool_timeout, + ), None => aps::thread_start_request( 2, cwd.as_deref(), @@ -1721,7 +1728,7 @@ async fn run_codex_turn_app_server( // else (handshake echoes, turn/start ack) is informational. if let Some(id) = aps::response_id(&message) { if let Some(message_ids) = pending_steer.remove(&id) { - resolve_codex_steer_response( + let accepted = resolve_codex_steer_response( deps, session, run_id, @@ -1730,6 +1737,23 @@ async fn run_codex_turn_app_server( &mut inflight_steer, ) .await; + // The steered user message was written to the conversation + // when the user hit send (an earlier `created_at`). Close + // the current assistant message and open a fresh one so the + // post-steer output sorts *after* that user message instead + // of accreting into one bubble that renders before it. + if accepted { + assistant_message = split_codex_assistant_message( + deps, + session, + run_id, + CliProviderRuntime::Codex.metadata_source(), + &assistant_message, + &mut state, + assistant_slot, + ) + .await?; + } } } continue; @@ -1995,6 +2019,49 @@ async fn build_codex_steer( )) } +/// After a `turn/steer` is accepted, close the current assistant message and +/// open a fresh one. The steered user message already sits in the conversation +/// with the `created_at` it got when the user hit send, so without this split +/// the whole turn accretes into a single assistant message whose `created_at` +/// is frozen at turn start — making the reply (including the handling of the +/// steered comment) render *before* the comment itself. Splitting yields the +/// natural order: pre-steer output, the user's message, then post-steer output. +async fn split_codex_assistant_message( + deps: &AssistantDeps, + session: &AssistantSession, + run_id: &str, + metadata_source: &str, + current: &AssistantMessage, + state: &mut CodexStreamState, + slot: &mut Option, +) -> Result { + let has_content = !non_empty_content_parts(&state.parts).is_empty() + || !state.persisted_tool_item_ids.is_empty(); + if has_content { + // Persist the pre-steer segment as its own completed message. + finalize_assistant_message_from_parts(deps, session, run_id, current, &state.parts).await?; + } else { + // Nothing emitted yet: drop the empty placeholder rather than leave an + // empty bubble before the user's steered message. + let _ = repository::delete_message(&deps.pool, ¤t.id).await; + let _ = emit_event( + &deps.app, + session, + Some(run_id), + AssistantUiEvent::MessageDeleted { + message_id: current.id.clone(), + }, + ); + } + // Start the post-steer segment fresh. Keep the tool-item dedup set so an + // item echoed across the split is not persisted twice; reset the delta + // throttle so the new message's first delta emits promptly. + state.parts.clear(); + state.last_update_emit_at = None; + *slot = None; + ensure_assistant_message_slot(deps, session, run_id, metadata_source, slot).await +} + /// Resolve a `turn/steer` response: on accept, mark the carried messages /// delivered and notify the UI; on reject (turn ended / id mismatch) leave them /// queued so the followup run delivers them. Always clears their in-flight mark. @@ -2005,7 +2072,7 @@ async fn resolve_codex_steer_response( response: &Value, message_ids: Vec, inflight: &mut HashSet, -) { +) -> bool { for id in &message_ids { inflight.remove(id); } @@ -2015,7 +2082,7 @@ async fn resolve_codex_steer_response( count = message_ids.len(), "codex turn/steer rejected; leaving messages queued for the followup run" ); - return; + return false; } match repository::mark_queued_messages_delivered(&deps.pool, &session.id, run_id, &message_ids) .await @@ -2040,6 +2107,10 @@ async fn resolve_codex_steer_response( "codex steer accepted but marking delivered failed; the followup run may re-deliver" ), } + // Accepted by Codex (the input entered the live turn) regardless of the DB + // mark result — the caller splits the assistant message either way so the + // post-steer output is ordered after the user's message. + true } /// Normalize an app-server `ThreadItem` (camelCase) into the exec item shape so