Add TUI tools for managing Codex tasks (#40308)

## Why

The TUI rejected app-server dynamic tool calls as unsupported, so agents could
not inspect or manage other Codex tasks through a TUI session.

## What changed

- Add the `codex_tui` tool namespace for listing, reading, waiting on, creating,
  forking, messaging, renaming, archiving, and restoring tasks.
- Handle tool calls through existing app-server thread operations, with bounded
  inputs and responses and cleanup when calls complete or the TUI disconnects.
- Route delegation tools through an authenticated local MCP server with explicit
  approval prompts, while respecting configured and managed MCP policies.
- Fall back to starting threads without dynamic tools when connected to an older
  app server that does not support them.

## Testing

Add coverage for tool schemas and response bounds, task lifecycle operations,
approval-gated MCP transport, namespace dispatch, cancellation, and legacy
app-server fallback behavior.

GitOrigin-RevId: c782c11f0a3edceb3fa6909a4b92d6ce01a6eed3
This commit is contained in:
Eric Traut
2026-08-24 01:50:53 +00:00
committed by copyberry
parent fb0781b9ee
commit a8468330bb
18 changed files with 3722 additions and 62 deletions
+1
View File
@@ -4377,6 +4377,7 @@ dependencies = [
"app_test_support",
"arboard",
"assert_matches",
"axum",
"base64 0.22.1",
"chrono",
"clap",
+2 -1
View File
@@ -23,6 +23,7 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
axum = { workspace = true, default-features = false, features = ["http1", "tokio"] }
base64 = { workspace = true }
chrono = { workspace = true, features = ["serde"] }
clap = { workspace = true, features = ["derive"] }
@@ -88,7 +89,7 @@ ratatui = { workspace = true, features = [
] }
ratatui-macros = { workspace = true }
regex-lite = { workspace = true }
rmcp = { workspace = true }
rmcp = { workspace = true, features = ["server", "transport-streamable-http-server"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["preserve_order"] }
sha2 = { workspace = true }
+3
View File
@@ -607,6 +607,9 @@ pub(crate) struct App {
primary_session_configured: Option<ThreadSessionState>,
pending_primary_events: VecDeque<ThreadBufferedEvent>,
pending_app_server_requests: PendingAppServerRequests,
dynamic_tool_status_updates:
tokio::sync::broadcast::Sender<codex_app_server_protocol::ThreadStatusChangedNotification>,
dynamic_tool_tasks: HashMap<codex_app_server_protocol::RequestId, (String, JoinHandle<()>)>,
pending_startup_thread_start: bool,
/// Keeps protected screens quarantined until initialized chat receives genuine user input.
startup_protected_input_boundary: bool,
+74
View File
@@ -87,6 +87,10 @@ impl App {
app_server_client: &AppServerSession,
notification: ServerNotification,
) {
if let ServerNotification::ThreadStatusChanged(status) = &notification {
let _ = self.dynamic_tool_status_updates.send(status.clone());
}
if let ServerNotification::ThreadStarted(started) = &notification
&& let SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
@@ -116,6 +120,9 @@ impl App {
}
match &notification {
ServerNotification::ServerRequestResolved(notification) => {
if let Some((_, task)) = self.dynamic_tool_tasks.remove(&notification.request_id) {
task.abort();
}
let notification_thread_id =
codex_protocol::ThreadId::from_string(&notification.thread_id).ok();
self.pending_primary_events.retain(|event| {
@@ -291,6 +298,73 @@ impl App {
app_server_client: &AppServerSession,
request: ServerRequest,
) {
if let ServerRequest::DynamicToolCall { request_id, params } = &request {
if self.dynamic_tool_tasks.contains_key(request_id)
|| (params.namespace.as_deref() != Some(crate::dynamic_tools::NAMESPACE)
&& !app_server_client.uses_embedded_app_server())
{
return;
}
let requires_mcp = crate::dynamic_tools::DELEGATION_TOOLS
.contains(&params.tool.as_str())
|| matches!(
app_server_client.thread_tool_transport(),
crate::dynamic_tools_mcp::ThreadToolTransport::Mcp(_)
);
if app_server_client.uses_embedded_app_server()
|| requires_mcp
|| codex_protocol::ThreadId::from_string(&params.thread_id)
.is_ok_and(|thread_id| self.abandoned_side_threads.contains(&thread_id))
{
let response = crate::dynamic_tools::failure_response(if requires_mcp {
"TUI task tools require the approval-gated MCP server"
} else {
"TUI dynamic tools require an active external task"
});
self.app_event_tx.send(AppEvent::DynamicToolCallCompleted {
request_id: request_id.clone(),
response,
});
return;
}
let request_handle = app_server_client.request_handle();
let app_event_tx = self.app_event_tx.clone();
let status_updates = self.dynamic_tool_status_updates.subscribe();
let request_id = request_id.clone();
let task_request_id = request_id.clone();
let source_thread_id = params.thread_id.clone();
let params = params.clone();
let mut thread_start_params =
crate::app_server_session::thread_start_params_from_config(
&self.config,
app_server_client.thread_params_mode(),
app_server_client.remote_cwd_override(),
/*session_start_source*/ None,
);
app_server_client
.thread_tool_transport()
.configure(&mut thread_start_params);
let task = tokio::spawn(async move {
let response = crate::dynamic_tools::execute(
request_handle,
params,
thread_start_params,
status_updates,
Some(&app_event_tx),
)
.await;
app_event_tx.send(AppEvent::DynamicToolCallCompleted {
request_id,
response,
});
});
self.dynamic_tool_tasks
.insert(task_request_id, (source_thread_id, task));
return;
}
let thread_id = server_request_thread_id(&request);
if thread_id.is_some_and(|thread_id| self.abandoned_side_threads.contains(&thread_id)) {
if let Err(err) = self
+1 -32
View File
@@ -162,12 +162,7 @@ impl PendingAppServerRequests {
);
None
}
ServerRequest::DynamicToolCall { request_id, .. } => {
Some(UnsupportedAppServerRequest {
request_id: request_id.clone(),
message: "Dynamic tool calls are not available in TUI yet.".to_string(),
})
}
ServerRequest::DynamicToolCall { .. } => None,
ServerRequest::ChatgptAuthTokensRefresh { .. } => None,
ServerRequest::AttestationGenerate { request_id, .. } => {
Some(UnsupportedAppServerRequest {
@@ -836,32 +831,6 @@ mod tests {
);
}
#[test]
fn rejects_dynamic_tool_calls_as_unsupported() {
let mut pending = PendingAppServerRequests::default();
let request = ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::Integer(99),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
call_id: "tool-1".to_string(),
namespace: None,
tool: "tool".to_string(),
arguments: json!({}),
},
};
assert!(!pending.contains_server_request(&request));
let unsupported = pending
.note_server_request(&request)
.expect("dynamic tool calls should be rejected");
assert_eq!(unsupported.request_id, AppServerRequestId::Integer(99));
assert_eq!(
unsupported.message,
"Dynamic tool calls are not available in TUI yet."
);
}
#[test]
fn does_not_mark_chatgpt_auth_refresh_as_unsupported() {
let mut pending = PendingAppServerRequests::default();
+48
View File
@@ -100,6 +100,35 @@ impl App {
self.handle_startup_thread_started(app_server, result)
.await?;
}
AppEvent::DynamicToolThreadStarted {
thread_id,
registered,
} => {
self.agents_overview
.dispatched_requests
.entry(thread_id)
.or_default();
let _ = registered.send(());
}
AppEvent::DynamicToolCallCompleted {
request_id,
response,
} => {
self.dynamic_tool_tasks.remove(&request_id);
match serde_json::to_value(response) {
Ok(result) => {
if let Err(error) = app_server
.resolve_server_request(request_id.clone(), result)
.await
{
tracing::warn!(?request_id, %error, "failed to resolve dynamic tool call");
}
}
Err(error) => {
tracing::warn!(?request_id, %error, "failed to serialize dynamic tool response");
}
}
}
AppEvent::RequestOlderScrollbackHistory { thread_id } => {
if self.chat_widget.thread_id() == Some(thread_id)
&& self.overlay.is_none()
@@ -2903,6 +2932,25 @@ impl App {
app_server: &mut AppServerSession,
mode: ExitMode,
) -> AppRunControl {
for (request_id, (_, task)) in self.dynamic_tool_tasks.drain() {
task.abort();
let response = crate::dynamic_tools::failure_response(
"TUI disconnected while handling a dynamic tool call",
);
match serde_json::to_value(response) {
Ok(result) => {
if let Err(error) = app_server
.resolve_server_request(request_id.clone(), result)
.await
{
tracing::warn!(?request_id, %error, "failed to cancel dynamic tool call");
}
}
Err(error) => {
tracing::warn!(?request_id, %error, "failed to serialize dynamic tool response")
}
}
}
match mode {
ExitMode::ShutdownFirst => {
// Mark the thread we are explicitly shutting down for exit so
+16
View File
@@ -497,6 +497,22 @@ impl App {
}
pub(super) async fn discard_thread_local_state(&mut self, thread_id: ThreadId) {
let app_event_tx = self.app_event_tx.clone();
self.dynamic_tool_tasks
.retain(|request_id, (source, task)| {
if source == &thread_id.to_string() {
app_event_tx.send(AppEvent::DynamicToolCallCompleted {
request_id: request_id.clone(),
response: crate::dynamic_tools::failure_response(
"Source task was closed while handling a dynamic tool call",
),
});
task.abort();
false
} else {
true
}
});
self.abort_thread_event_listener(thread_id);
self.thread_event_channels.remove(&thread_id);
self.side_threads.remove(&thread_id);
+20
View File
@@ -28,12 +28,14 @@ fn spawn_startup_thread_start(
let request_handle = app_server.request_handle();
let thread_params_mode = app_server.thread_params_mode();
let remote_cwd_override = app_server.remote_cwd_override().map(Path::to_path_buf);
let thread_tool_transport = app_server.thread_tool_transport();
tokio::spawn(async move {
let result = crate::app_server_session::start_thread_with_request_handle(
request_handle,
config,
thread_params_mode,
remote_cwd_override,
thread_tool_transport,
)
.await
.map_err(|err| format!("{err:#}"));
@@ -177,6 +179,22 @@ impl App {
if let Some(updated_model) = config.model.clone() {
model = updated_model;
}
let dynamic_tool_status_updates = tokio::sync::broadcast::channel(/*capacity*/ 64).0;
if matches!(&app_server_target, AppServerTarget::LocalDaemon { .. })
&& !crate::uses_remote_workspace_or_environment(
&app_server_target,
environment_manager.as_ref(),
)
&& let Err(error) = app_server
.start_dynamic_tool_mcp(
config.clone(),
app_event_tx.clone(),
dynamic_tool_status_updates.clone(),
)
.await
{
tracing::warn!(%error, "TUI task delegation is unavailable without its MCP server");
}
let model_catalog = Arc::new(ModelCatalog::new(available_models.clone()));
let feedback_audience = bootstrap.feedback_audience;
let auth_mode = bootstrap.auth_mode;
@@ -501,6 +519,8 @@ See the Codex keymap documentation for supported actions and examples."
primary_session_configured: None,
pending_primary_events: VecDeque::new(),
pending_app_server_requests: PendingAppServerRequests::default(),
dynamic_tool_status_updates,
dynamic_tool_tasks: HashMap::new(),
pending_startup_thread_start,
startup_protected_input_boundary: true,
startup_pending_protected_request: false,
+2
View File
@@ -72,6 +72,8 @@ pub(super) async fn make_test_app() -> App {
primary_session_configured: None,
pending_primary_events: VecDeque::new(),
pending_app_server_requests: PendingAppServerRequests::default(),
dynamic_tool_status_updates: tokio::sync::broadcast::channel(/*capacity*/ 64).0,
dynamic_tool_tasks: HashMap::new(),
pending_startup_thread_start: false,
startup_protected_input_boundary: false,
startup_pending_protected_request: false,
+17 -1
View File
@@ -5018,7 +5018,7 @@ async fn discard_side_thread_keeps_local_state_when_server_close_fails() -> Resu
#[tokio::test]
async fn background_side_cleanup_removes_local_state_and_ignores_late_events() -> Result<()> {
let mut app = make_test_app().await;
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let mut app_server =
crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?;
let parent_thread_id = ThreadId::new();
@@ -5034,9 +5034,21 @@ async fn background_side_cleanup_removes_local_state_and_ignores_late_events() -
Some("side".to_string()),
/*is_closed*/ false,
);
app.dynamic_tool_tasks.insert(
AppServerRequestId::Integer(123),
(
side_thread_id.to_string(),
tokio::spawn(std::future::pending::<()>()),
),
);
app.discard_side_thread_in_background(&mut app_server, side_thread_id)
.await;
assert_matches!(
events.try_recv(),
Ok(AppEvent::DynamicToolCallCompleted { response, .. }) if !response.success
);
assert!(app.dynamic_tool_tasks.is_empty());
assert_eq!(app.active_thread_id, Some(parent_thread_id));
assert!(!app.side_threads.contains_key(&side_thread_id));
assert!(!app.thread_event_channels.contains_key(&side_thread_id));
@@ -5413,6 +5425,8 @@ async fn make_test_app() -> App {
primary_session_configured: None,
pending_primary_events: VecDeque::new(),
pending_app_server_requests: PendingAppServerRequests::default(),
dynamic_tool_status_updates: tokio::sync::broadcast::channel(/*capacity*/ 64).0,
dynamic_tool_tasks: HashMap::new(),
pending_startup_thread_start: false,
startup_protected_input_boundary: false,
startup_pending_protected_request: false,
@@ -5490,6 +5504,8 @@ async fn make_test_app_with_channels() -> (
primary_session_configured: None,
pending_primary_events: VecDeque::new(),
pending_app_server_requests: PendingAppServerRequests::default(),
dynamic_tool_status_updates: tokio::sync::broadcast::channel(/*capacity*/ 64).0,
dynamic_tool_tasks: HashMap::new(),
pending_startup_thread_start: false,
startup_protected_input_boundary: false,
startup_pending_protected_request: false,
@@ -45,6 +45,7 @@ enum HistoryCapabilities {
Current,
LegacyOnly,
LegacyOnlyUnsupportedVariant,
LegacyDynamicToolsAndHistory,
ForkHydrationFails,
}
@@ -157,10 +158,29 @@ async fn start_recording_app_server_with_history(
.expect("request recorder lock")
.iter()
.any(|recorded| recorded.method == "thread/fork");
let response = if matches!(
let reject_dynamic_tools = history_capabilities
== HistoryCapabilities::LegacyDynamicToolsAndHistory
&& request.method == "thread/start"
&& params
.and_then(|params| params.get("dynamicTools"))
.and_then(serde_json::Value::as_array)
.is_some_and(|tools| {
tools.iter().any(|tool| tool["type"] == "namespace")
});
let response = if reject_dynamic_tools {
JSONRPCMessage::Error(JSONRPCError {
id: request_id,
error: JSONRPCErrorError {
code: -32602,
data: None,
message: "missing field `inputSchema`".to_string(),
},
})
} else if matches!(
history_capabilities,
HistoryCapabilities::LegacyOnly
| HistoryCapabilities::LegacyOnlyUnsupportedVariant
| HistoryCapabilities::LegacyDynamicToolsAndHistory
) && requires_pagination
{
let (code, message) = if history_capabilities
@@ -327,6 +347,864 @@ async fn make_history_test_app() -> Result<(App, tempfile::TempDir)> {
Ok((app, codex_home))
}
fn spawn_approved_task_tool_call(
app: &App,
app_server: &AppServerSession,
request_id: AppServerRequestId,
params: codex_app_server_protocol::DynamicToolCallParams,
) {
let request_handle = app_server.request_handle();
let app_event_tx = app.app_event_tx.clone();
let status_updates = app.dynamic_tool_status_updates.subscribe();
let mut thread_start_params = crate::app_server_session::thread_start_params_from_config(
&app.config,
app_server.thread_params_mode(),
app_server.remote_cwd_override(),
/*session_start_source*/ None,
);
app_server
.thread_tool_transport()
.configure(&mut thread_start_params);
tokio::spawn(async move {
let response = crate::dynamic_tools::execute(
request_handle,
params,
thread_start_params,
status_updates,
Some(&app_event_tx),
)
.await;
app_event_tx.send(AppEvent::DynamicToolCallCompleted {
request_id,
response,
});
});
}
#[tokio::test]
async fn external_transport_excludes_delegation_dynamic_tools_for_both_start_paths() -> Result<()> {
let (app, _codex_home) = make_history_test_app().await?;
let (mut app_server, requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
app_server.start_thread(&app.config).await?;
crate::app_server_session::start_thread_with_request_handle(
app_server.request_handle(),
app.config.clone(),
crate::app_server_session::ThreadParamsMode::Embedded,
/*remote_cwd_override*/ None,
app_server.thread_tool_transport(),
)
.await?;
let starts = recorded_params(&requests, "thread/start");
assert_eq!(starts.len(), 2);
for params in starts {
assert_eq!(params["dynamicTools"][0]["type"], "namespace");
assert_eq!(params["dynamicTools"][0]["name"], "codex_tui");
assert_eq!(
params["dynamicTools"][0]["tools"].as_array().map(Vec::len),
Some(6)
);
assert!(
params["dynamicTools"][0]["tools"]
.as_array()
.is_some_and(|tools| tools.iter().all(|tool| {
tool["deferLoading"] == true
&& !crate::dynamic_tools::DELEGATION_TOOLS
.contains(&tool["name"].as_str().unwrap_or_default())
}))
);
}
app_server.shutdown().await?;
proxy.await??;
Ok(())
}
#[tokio::test]
async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths() -> Result<()> {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
app.config
.web_search_mode
.set(codex_protocol::config_types::WebSearchMode::Live)?;
std::fs::write(
codex_home.path().join("config.toml"),
"web_search = \"disabled\"\n",
)?;
let (mut app_server, requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
app_server
.start_dynamic_tool_mcp(
app.config.clone(),
app.app_event_tx.clone(),
app.dynamic_tool_status_updates.clone(),
)
.await?;
let thread_id = app_server
.start_thread(&app.config)
.await?
.session
.thread_id;
crate::app_server_session::start_thread_with_request_handle(
app_server.request_handle(),
app.config.clone(),
crate::app_server_session::ThreadParamsMode::Embedded,
/*remote_cwd_override*/ None,
app_server.thread_tool_transport(),
)
.await?;
let inventory: codex_app_server_protocol::ListMcpServerStatusResponse = app_server
.request_handle()
.request_typed(ClientRequest::McpServerStatusList {
request_id: AppServerRequestId::String("tui-tool-inventory".to_string()),
params: codex_app_server_protocol::ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(codex_app_server_protocol::McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(thread_id.to_string()),
},
})
.await?;
let tools = &inventory
.data
.iter()
.find(|server| server.name == "codex_tui")
.expect("local daemon must connect to the TUI MCP server")
.tools;
assert_eq!(tools.len(), 9);
for tool in crate::dynamic_tools::DELEGATION_TOOLS {
assert!(tools.contains_key(tool));
}
assert!(
!tools["create_thread"]
.input_schema
.get("properties")
.and_then(serde_json::Value::as_object)
.is_some_and(|properties| properties.contains_key("permissions"))
);
let starts = recorded_params(&requests, "thread/start");
assert_eq!(starts.len(), 2);
for params in &starts {
assert_eq!(params["dynamicTools"], serde_json::Value::Null);
assert_eq!(params["config"]["web_search"], "live");
let server = &params["config"]["mcp_servers.codex_tui"];
assert!(
server["url"]
.as_str()
.is_some_and(|url| url.starts_with("http://127.0.0.1:"))
);
assert!(
server["http_headers"]["Authorization"]
.as_str()
.is_some_and(|header| header.starts_with("Bearer "))
);
assert_eq!(server["default_tools_approval_mode"], "approve");
for tool in crate::dynamic_tools::DELEGATION_TOOLS {
assert_eq!(server["tools"][tool]["approval_mode"], "prompt");
}
}
let mcp_url = starts[0]["config"]["mcp_servers.codex_tui"]["url"]
.as_str()
.expect("MCP server URL");
let unauthorized = codex_http_client::HttpClientBuilder::new()
.build_direct()?
.post(mcp_url)
.send()
.await?;
assert_eq!(unauthorized.status().as_u16(), 401);
app.config
.web_search_mode
.set(codex_protocol::config_types::WebSearchMode::Disabled)?;
let delegation_source = create_history_rollout(
&app.config,
ThreadHistoryMode::Legacy,
"Approved task source",
)?;
app_server
.resume_thread(
app.config.clone(),
delegation_source,
crate::app_server_session::ResumeModelSettings::RestoreFromThread,
)
.await?;
let resumed = recorded_params(&requests, "thread/resume")
.pop()
.expect("resumed task request");
assert_eq!(
resumed["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
app_server
.resume_thread(
app.config.clone(),
delegation_source,
crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
)
.await?;
let reattached = recorded_params(&requests, "thread/resume")
.pop()
.expect("reattached task request");
assert_eq!(
reattached["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
app_server
.fork_thread(app.config.clone(), delegation_source)
.await?;
let forked = recorded_params(&requests, "thread/fork")
.pop()
.expect("forked task request");
assert_eq!(
forked["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
let authorization =
starts[0]["config"]["mcp_servers.codex_tui"]["http_headers"]["Authorization"]
.as_str()
.expect("MCP bearer token");
let client = codex_http_client::HttpClientBuilder::new().build_direct()?;
let call_tool = |id: u32, tool: &'static str, arguments: serde_json::Value| {
client
.post(mcp_url)
.header("Authorization", authorization)
.header("Accept", "application/json, text/event-stream")
.header("MCP-Protocol-Version", "2026-07-28")
.header("MCP-Method", "tools/call")
.header("MCP-Name", tool)
.json(&serde_json::json!({
"jsonrpc": "2.0",
"id": id,
"method": "tools/call",
"params": {
"name": tool,
"arguments": arguments,
"_meta": {
"threadId": delegation_source,
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {}
}
}
}))
};
let response = call_tool(1, "list_threads", serde_json::json!({}))
.send()
.await?;
let status = response.status();
let response = response.text().await?;
assert!(status.is_success(), "{status}: {response}");
assert!(response.contains("threads"), "{response}");
let mut creation = tokio::spawn(
call_tool(
2,
"create_thread",
serde_json::json!({"prompt": "Start an approved task"}),
)
.send(),
);
let registration = tokio::select! {
event = events.recv() => event.expect("approved MCP task must register before starting"),
response = &mut creation => {
let response = response??;
panic!("MCP task creation completed without registration: {}", response.text().await?);
}
_ = tokio::time::sleep(std::time::Duration::from_secs(/*secs*/ 5)) => {
panic!("timed out waiting for MCP task registration");
}
};
let AppEvent::DynamicToolThreadStarted {
thread_id: child_thread_id,
registered,
} = registration
else {
panic!("expected the MCP-created task to register")
};
assert!(registered.send(()).is_ok());
let created = creation.await??;
assert!(created.status().is_success());
assert!(created.text().await?.contains(&child_thread_id.to_string()));
let child = recorded_params(&requests, "thread/start")
.pop()
.expect("MCP child thread/start request");
assert_eq!(child["dynamicTools"], serde_json::Value::Null);
assert!(child["config"]["web_search"].is_null());
assert_eq!(
child["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
let forked = call_tool(3, "fork_thread", serde_json::json!({"threadId": thread_id}))
.send()
.await?;
assert!(forked.status().is_success());
let forked = recorded_params(&requests, "thread/fork")
.pop()
.expect("MCP-created fork request");
assert_eq!(
forked["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
app_server.shutdown().await?;
proxy.await??;
Ok(())
}
#[tokio::test]
async fn local_mcp_respects_configured_servers_and_managed_requirements() -> Result<()> {
for scenario in ["conflicting", "blocked", "mismatched", "allowed"] {
let (mut app, _codex_home) = make_history_test_app().await?;
if scenario == "conflicting" {
let raw = serde_json::from_value::<codex_config::RawMcpServerConfig>(
serde_json::json!({"url": "http://127.0.0.1:1/mcp", "enabled": false}),
)?;
let mut servers = app.config.mcp_servers.get().clone();
servers.insert(
crate::dynamic_tools::NAMESPACE.to_string(),
codex_config::McpServerConfig::try_from(raw)
.map_err(color_eyre::eyre::Report::msg)?,
);
app.config.mcp_servers.set(servers)?;
} else {
let mut allowed_servers = std::collections::BTreeMap::new();
if matches!(scenario, "mismatched" | "allowed") {
let requirement = if scenario == "allowed" {
codex_config::McpServerRequirement::Url(
codex_protocol::mcp_policy::McpServerValueMatcher::Prefix {
value: "http://127.0.0.1:".to_string(),
},
)
} else {
codex_config::McpServerRequirement::Identity {
identity: codex_config::McpServerIdentity::Url {
url: "http://127.0.0.1:1/mcp".to_string(),
},
}
};
allowed_servers.insert(crate::dynamic_tools::NAMESPACE.to_string(), requirement);
}
let requirements = codex_config::ConfigRequirements {
mcp_servers: Some(codex_config::Sourced::new(
allowed_servers,
codex_config::RequirementSource::Unknown,
)),
..Default::default()
};
app.config.config_layer_stack = codex_config::ConfigLayerStack::new(
Vec::new(),
requirements,
codex_config::ConfigRequirementsToml::default(),
)?;
}
let (mut app_server, requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
let result = app_server
.start_dynamic_tool_mcp(
app.config.clone(),
app.app_event_tx.clone(),
app.dynamic_tool_status_updates.clone(),
)
.await;
if scenario == "allowed" {
result?;
} else {
let error = result.expect_err("unavailable internal MCP must fail closed");
assert_eq!(
error.kind(),
if scenario == "conflicting" {
std::io::ErrorKind::AlreadyExists
} else {
std::io::ErrorKind::PermissionDenied
}
);
}
app_server.start_thread(&app.config).await?;
let start = recorded_params(&requests, "thread/start")
.pop()
.expect("fallback task start");
if scenario == "allowed" {
assert!(start["dynamicTools"].is_null());
assert!(start["config"]["mcp_servers.codex_tui"].is_object());
} else {
assert_eq!(
start["dynamicTools"][0]["tools"].as_array().map(Vec::len),
Some(6)
);
assert!(start["config"]["mcp_servers.codex_tui"].is_null());
}
app_server.shutdown().await?;
proxy.await??;
}
Ok(())
}
#[tokio::test]
async fn older_external_server_starts_without_unsupported_dynamic_tools_or_history() -> Result<()> {
let (app, _codex_home) = make_history_test_app().await?;
let (mut app_server, requests, proxy) = start_recording_app_server_with_history(
&app.config,
HistoryCapabilities::LegacyDynamicToolsAndHistory,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
app_server.start_thread(&app.config).await?;
crate::app_server_session::start_thread_with_request_handle(
app_server.request_handle(),
app.config.clone(),
crate::app_server_session::ThreadParamsMode::Embedded,
/*remote_cwd_override*/ None,
app_server.thread_tool_transport(),
)
.await?;
let starts = recorded_params(&requests, "thread/start");
assert_eq!(starts.len(), 6);
for attempts in starts.chunks_exact(3) {
assert_eq!(attempts[0]["dynamicTools"][0]["type"], "namespace");
assert_eq!(attempts[0]["historyMode"], "paginated");
assert_eq!(attempts[1]["dynamicTools"], serde_json::Value::Null);
assert_eq!(attempts[1]["historyMode"], "paginated");
assert_eq!(attempts[2]["dynamicTools"], serde_json::Value::Null);
assert_eq!(attempts[2]["historyMode"], serde_json::Value::Null);
}
app_server.shutdown().await?;
proxy.await??;
Ok(())
}
#[tokio::test]
async fn embedded_server_rejects_unowned_dynamic_tool_calls() -> Result<()> {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
let app_server = crate::start_embedded_app_server_for_picker(&app.config).await?;
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(
ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::Integer(100),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
call_id: "call-1".to_string(),
namespace: Some("codex_app".to_string()),
tool: "list_threads".to_string(),
arguments: serde_json::json!({}),
},
},
)),
)
.await;
let AppEvent::DynamicToolCallCompleted { response, .. } = events
.try_recv()
.expect("embedded dynamic calls must receive a response")
else {
panic!("expected a dynamic tool failure response")
};
assert!(!response.success);
app_server.shutdown().await?;
Ok(())
}
#[tokio::test]
async fn dynamic_tool_requests_ignore_other_namespaces_and_dispatch_tui_namespace() -> Result<()> {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
app.config
.permissions
.set_permission_profile(PermissionProfile::workspace_write_with(
&[app.config.cwd.clone()],
codex_protocol::permissions::NetworkSandboxPolicy::Restricted,
/*exclude_tmpdir_env_var*/ true,
/*exclude_slash_tmp*/ true,
))?;
let (mut app_server, requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ Some("Unavailable name"),
)
.await?;
let thread_id = app_server
.start_thread(&app.config)
.await?
.session
.thread_id
.to_string();
for namespace in [Some("codex_app"), None] {
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(
ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::Integer(100),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: "turn-1".to_string(),
call_id: "call-1".to_string(),
namespace: namespace.map(str::to_string),
tool: "list_threads".to_string(),
arguments: serde_json::json!({}),
},
},
)),
)
.await;
assert!(events.try_recv().is_err());
}
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(
ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::Integer(101),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: "turn-1".to_string(),
call_id: "call-2".to_string(),
namespace: Some("codex_tui".to_string()),
tool: "list_threads".to_string(),
arguments: serde_json::json!({}),
},
},
)),
)
.await;
let event = tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("dynamic tool completion event");
let AppEvent::DynamicToolCallCompleted {
request_id,
response,
} = event
else {
panic!("expected a dynamic tool completion event")
};
assert_eq!(request_id, AppServerRequestId::Integer(101));
assert!(response.success, "{response:?}");
let list_requests = recorded_params(&requests, "thread/list");
assert_eq!(list_requests.len(), 1);
assert_eq!(list_requests[0]["useStateDbOnly"], true);
assert_eq!(list_requests[0]["sourceKinds"], serde_json::Value::Null);
let mut tui = crate::tui::test_support::make_test_tui()?;
app.handle_event(
&mut tui,
&mut app_server,
AppEvent::DynamicToolCallCompleted {
request_id,
response,
},
)
.await?;
let completed = tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), async {
loop {
if let Some(response) = recorded_params(&requests, "server/request/response").pop() {
break response;
}
tokio::task::yield_now().await;
}
})
.await?;
assert_eq!(completed["success"], true);
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(
ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::Integer(102),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: "turn-1".to_string(),
call_id: "call-3".to_string(),
namespace: Some("codex_tui".to_string()),
tool: "set_thread_title".to_string(),
arguments: serde_json::json!({"threadId": thread_id, "title": "Renamed"}),
},
},
)),
)
.await;
let AppEvent::DynamicToolCallCompleted { response, .. } =
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("dynamic mutation completion event")
else {
panic!("expected a dynamic mutation completion event")
};
assert!(response.success, "{response:?}");
assert_eq!(
recorded_params(&requests, "thread/name/set")[0]["name"],
"Renamed"
);
for (index, tool) in crate::dynamic_tools::DELEGATION_TOOLS
.into_iter()
.enumerate()
{
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(
ServerRequest::DynamicToolCall {
request_id: AppServerRequestId::String(format!("rejected-{index}")),
params: codex_app_server_protocol::DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: "turn-1".to_string(),
call_id: format!("rejected-{index}"),
namespace: Some("codex_tui".to_string()),
tool: tool.to_string(),
arguments: serde_json::json!({}),
},
},
)),
)
.await;
let AppEvent::DynamicToolCallCompleted { response, .. } = events
.try_recv()
.expect("legacy delegation call must receive an immediate rejection")
else {
panic!("expected a legacy delegation failure response")
};
assert!(!response.success);
}
let creation_source = create_history_rollout(
&app.config,
ThreadHistoryMode::Legacy,
"Background task source",
)?;
app_server
.resume_thread(
app.config.clone(),
creation_source,
crate::app_server_session::ResumeModelSettings::RestoreFromThread,
)
.await?;
let project: codex_app_server_protocol::ProjectCreateResponse = app_server
.request_handle()
.request_typed(ClientRequest::ProjectCreate {
request_id: AppServerRequestId::String("create-source-project".to_string()),
params: codex_app_server_protocol::ProjectCreateParams {
name: "Source project".to_string(),
roots: vec![codex_app_server_protocol::ProjectRoot {
path: app.config.cwd.clone(),
}],
metadata: None,
idempotency_key: "source-project".to_string(),
},
})
.await?;
let _: codex_app_server_protocol::ThreadMetadataUpdateResponse = app_server
.request_handle()
.request_typed(ClientRequest::ThreadMetadataUpdate {
request_id: AppServerRequestId::String("assign-source-project".to_string()),
params: codex_app_server_protocol::ThreadMetadataUpdateParams {
thread_id: creation_source.to_string(),
project_id: Some(project.project.id.clone()),
git_info: None,
},
})
.await?;
let source_settings: codex_app_server_protocol::ThreadResumeResponse = app_server
.request_handle()
.request_typed(ClientRequest::ThreadResume {
request_id: AppServerRequestId::String("read-source-sandbox".to_string()),
params: codex_app_server_protocol::ThreadResumeParams {
thread_id: creation_source.to_string(),
..codex_app_server_protocol::ThreadResumeParams::default()
},
})
.await?;
assert!(source_settings.active_permission_profile.is_none());
let source_sandbox = serde_json::to_value(source_settings.sandbox)?;
spawn_approved_task_tool_call(
&app,
&app_server,
AppServerRequestId::Integer(103),
codex_app_server_protocol::DynamicToolCallParams {
thread_id: creation_source.to_string(),
turn_id: "turn-1".to_string(),
call_id: "call-4".to_string(),
namespace: Some("codex_tui".to_string()),
tool: "create_thread".to_string(),
arguments: serde_json::json!({
"prompt": "Check <main> & report",
"title": "Unavailable name"
}),
},
);
let registration =
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("background task registration event");
let AppEvent::DynamicToolThreadStarted {
thread_id: created_thread_id,
registered,
} = registration
else {
panic!("expected background task registration before its first turn: {registration:?}")
};
assert!(recorded_params(&requests, "turn/start").is_empty());
app.handle_event(
&mut tui,
&mut app_server,
AppEvent::DynamicToolThreadStarted {
thread_id: created_thread_id,
registered,
},
)
.await?;
assert!(
app.agents_overview
.dispatched_requests
.contains_key(&created_thread_id)
);
let AppEvent::DynamicToolCallCompleted { response, .. } =
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("background task creation completion")
else {
panic!("expected a background task completion event")
};
assert!(response.success, "{response:?}");
assert_eq!(
recorded_params(&requests, "thread/start")
.last()
.expect("background task creation")["projectId"],
project.project.id
);
let turn = recorded_params(&requests, "turn/start")
.pop()
.expect("background task turn request");
assert_eq!(
turn["input"][0]["text"],
format!(
"<codex_delegation>\n <source_thread_id>{creation_source}</source_thread_id>\n <input>Check &lt;main&gt; &amp; report</input>\n</codex_delegation>"
)
);
assert_eq!(turn["sandboxPolicy"], source_sandbox);
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(exec_approval_request(
created_thread_id,
"turn-2",
"item-1",
/*approval_id*/ None,
))),
)
.await;
assert_eq!(
app.agents_overview.dispatched_requests[&created_thread_id].len(),
1
);
spawn_approved_task_tool_call(
&app,
&app_server,
AppServerRequestId::Integer(104),
codex_app_server_protocol::DynamicToolCallParams {
thread_id: thread_id.clone(),
turn_id: "turn-1".to_string(),
call_id: "call-5".to_string(),
namespace: Some("codex_tui".to_string()),
tool: "send_message_to_thread".to_string(),
arguments: serde_json::json!({
"threadId": creation_source,
"prompt": "Follow <up> & report"
}),
},
);
let AppEvent::DynamicToolThreadStarted {
thread_id: continued_thread_id,
registered,
} = tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("follow-up task registration event")
else {
panic!("expected follow-up task registration before its next turn")
};
assert_eq!(continued_thread_id, creation_source);
assert_eq!(recorded_params(&requests, "turn/start").len(), 1);
app.handle_event(
&mut tui,
&mut app_server,
AppEvent::DynamicToolThreadStarted {
thread_id: continued_thread_id,
registered,
},
)
.await?;
let AppEvent::DynamicToolCallCompleted { response, .. } =
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 5), events.recv())
.await?
.expect("follow-up task completion")
else {
panic!("expected a follow-up task completion event")
};
assert!(response.success, "{response:?}");
assert_eq!(
recorded_params(&requests, "turn/start")[1]["input"][0]["text"],
format!(
"<codex_delegation>\n <source_thread_id>{thread_id}</source_thread_id>\n <input>Follow &lt;up&gt; &amp; report</input>\n</codex_delegation>"
)
);
app.dynamic_tool_tasks.insert(
AppServerRequestId::Integer(105),
(thread_id, tokio::spawn(std::future::pending::<()>())),
);
assert_matches!(
app.handle_exit_mode(&mut app_server, ExitMode::ShutdownFirst)
.await,
AppRunControl::Exit(ExitReason::UserRequested)
);
let cancelled = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async {
loop {
if let Some(response) = recorded_params(&requests, "server/request/response")
.into_iter()
.find(|response| response["success"] == false)
{
break response;
}
tokio::task::yield_now().await;
}
})
.await?;
assert_eq!(cancelled["success"], false);
assert!(app.dynamic_tool_tasks.is_empty());
app_server.shutdown().await?;
proxy.await??;
Ok(())
}
#[tokio::test]
async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> Result<()> {
let (mut app, codex_home) = make_history_test_app().await?;
@@ -761,6 +1639,34 @@ async fn remote_legacy_history_start_negotiates_once_for_resume_and_fork() -> Re
);
assert_eq!(recorded_params(&requests, "thread/turns/list").len(), 1);
let (_status_sender, status_updates) = tokio::sync::broadcast::channel(/*capacity*/ 1);
let response = crate::dynamic_tools::execute(
app_server.request_handle(),
codex_app_server_protocol::DynamicToolCallParams {
thread_id: started.session.thread_id.to_string(),
turn_id: "source-turn".to_string(),
call_id: "legacy-wait".to_string(),
namespace: Some(crate::dynamic_tools::NAMESPACE.to_string()),
tool: "wait_threads".to_string(),
arguments: serde_json::json!({
"targets": [{"threadId": legacy_thread_id}],
"timeoutMs": 0
}),
},
codex_app_server_protocol::ThreadStartParams::default(),
status_updates,
/*app_event_tx*/ None,
)
.await;
assert!(response.success, "{response:?}");
assert!(
recorded_params(&requests, "thread/read")
.iter()
.any(|params| {
params["threadId"] == legacy_thread_id.to_string() && params["includeTurns"] == true
})
);
app_server.shutdown().await?;
proxy.await??;
Ok(())
+14
View File
@@ -16,6 +16,7 @@ use crate::inline_visualization::InlineVisualizationContext;
use codex_app_server_protocol::AddCreditsNudgeCreditType;
use codex_app_server_protocol::AddCreditsNudgeEmailStatus;
use codex_app_server_protocol::ConsumeAccountRateLimitResetCreditResponse;
use codex_app_server_protocol::DynamicToolCallResponse;
use codex_app_server_protocol::GetAccountRateLimitsResponse;
use codex_app_server_protocol::GetAccountTokenUsageResponse;
use codex_app_server_protocol::MarketplaceAddResponse;
@@ -29,6 +30,7 @@ use codex_app_server_protocol::PluginMarketplaceEntry;
use codex_app_server_protocol::PluginReadParams;
use codex_app_server_protocol::PluginReadResponse;
use codex_app_server_protocol::PluginUninstallResponse;
use codex_app_server_protocol::RequestId as AppServerRequestId;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadGoalStatus;
@@ -336,6 +338,18 @@ pub(crate) enum AppEvent {
result: Result<AppServerStartedThread, String>,
},
/// Register a dynamically created background thread before its first turn starts.
DynamicToolThreadStarted {
thread_id: ThreadId,
registered: tokio::sync::oneshot::Sender<()>,
},
/// Return a completed client-owned dynamic tool call to app server.
DynamicToolCallCompleted {
request_id: AppServerRequestId,
response: DynamicToolCallResponse,
},
/// Clear the terminal UI (screen + scrollback), start a fresh session, and keep the
/// previous chat resumable.
ClearUi {
+120 -27
View File
@@ -12,7 +12,10 @@ pub(crate) use history::HISTORY_ITEM_SCAN_LIMIT;
pub(crate) use history::HistoryHydrationScope;
pub(crate) use history::thread_items_page_params;
use crate::app_event_sender::AppEventSender;
use crate::bottom_pane::FeedbackAudience;
use crate::dynamic_tools_mcp::DynamicToolMcpServer;
use crate::dynamic_tools_mcp::ThreadToolTransport;
use crate::legacy_core::config::Config;
use crate::service_tier_resolution;
use crate::session_state::MessageHistoryMetadata;
@@ -103,6 +106,7 @@ use codex_app_server_protocol::ThreadSource;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartSource;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnsubscribeParams;
@@ -137,6 +141,7 @@ use color_eyre::eyre::Result;
use color_eyre::eyre::WrapErr;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
@@ -162,7 +167,7 @@ enum ForkPresentation {
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ThreadHistorySupport {
pub(crate) enum ThreadHistorySupport {
Paginated,
LegacyOnly,
}
@@ -200,35 +205,50 @@ pub(crate) fn is_history_pagination_unsupported(source: &JSONRPCErrorError) -> b
.any(|error| message.contains(error)))
}
async fn request_thread_start_with_history_fallback(
pub(crate) async fn request_thread_start_with_history_fallback(
request_handle: &AppServerRequestHandle,
request_id: RequestId,
mut request_id: RequestId,
mut params: ThreadStartParams,
) -> std::result::Result<(ThreadStartResponse, ThreadHistorySupport), TypedRequestError> {
match request_handle
.request_typed(ClientRequest::ThreadStart {
request_id,
params: params.clone(),
})
.await
{
Ok(response) => Ok((response, ThreadHistorySupport::Paginated)),
Err(TypedRequestError::Server { source, .. })
if params.history_mode.is_some() && is_history_pagination_unsupported(&source) =>
let mut history_support = ThreadHistorySupport::Paginated;
loop {
match request_handle
.request_typed(ClientRequest::ThreadStart {
request_id,
params: params.clone(),
})
.await
{
params.history_mode = None;
let response = request_handle
.request_typed(ClientRequest::ThreadStart {
request_id: RequestId::String(format!(
"legacy-thread-start-{}",
Uuid::new_v4()
)),
params,
})
.await?;
Ok((response, ThreadHistorySupport::LegacyOnly))
Ok(response) => return Ok((response, history_support)),
Err(TypedRequestError::Server { source, .. })
if params.history_mode.is_some() && is_history_pagination_unsupported(&source) =>
{
params.history_mode = None;
history_support = ThreadHistorySupport::LegacyOnly;
request_id = RequestId::String(format!("legacy-thread-start-{}", Uuid::new_v4()));
}
Err(TypedRequestError::Server { source, .. })
if params.dynamic_tools.is_some()
&& matches!(
source.code,
JSONRPC_INVALID_REQUEST | JSONRPC_INVALID_PARAMS
)
&& {
let message = source.message.to_ascii_lowercase();
["dynamictools", "dynamic tool", "namespace", "inputschema"]
.into_iter()
.any(|field| message.contains(field))
} =>
{
tracing::warn!(
error = %source.message,
"app server does not support TUI dynamic tools; starting without them"
);
params.dynamic_tools = None;
request_id = RequestId::String(format!("legacy-thread-start-{}", Uuid::new_v4()));
}
Err(err) => return Err(err),
}
Err(err) => Err(err),
}
}
@@ -273,6 +293,7 @@ pub(crate) struct AppServerSession {
available_models: Vec<ModelPreset>,
managed_new_thread_defaults: Option<NewThreadModelDefaults>,
external_agent_config_import_completion_pending: AtomicBool,
dynamic_tool_mcp: Option<Arc<DynamicToolMcpServer>>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -356,6 +377,72 @@ impl AppServerSession {
available_models: Vec::new(),
managed_new_thread_defaults: None,
external_agent_config_import_completion_pending: AtomicBool::new(false),
dynamic_tool_mcp: None,
}
}
pub(crate) async fn start_dynamic_tool_mcp(
&mut self,
config: Config,
app_event_tx: AppEventSender,
status_updates: tokio::sync::broadcast::Sender<ThreadStatusChangedNotification>,
) -> std::io::Result<()> {
if self.uses_embedded_app_server() {
return Ok(());
}
if config
.mcp_servers
.get()
.contains_key(crate::dynamic_tools::NAMESPACE)
{
return Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
"a user-configured MCP server already owns the codex_tui namespace",
));
}
let managed_requirement = config
.config_layer_stack
.requirements()
.mcp_servers
.as_ref()
.map(|requirements| {
requirements
.value
.get(crate::dynamic_tools::NAMESPACE)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"managed MCP requirements do not permit the TUI task-tools server",
)
})
})
.transpose()?;
let thread_start_params = thread_start_params_from_config(
&config,
self.thread_params_mode(),
self.remote_cwd_override(),
/*session_start_source*/ None,
);
self.dynamic_tool_mcp = Some(Arc::new(
DynamicToolMcpServer::start(
self.request_handle(),
thread_start_params,
app_event_tx,
status_updates,
managed_requirement,
)
.await?,
));
Ok(())
}
pub(crate) fn thread_tool_transport(&self) -> ThreadToolTransport {
if self.uses_embedded_app_server() {
ThreadToolTransport::Disabled
} else if let Some(server) = self.dynamic_tool_mcp.as_ref() {
ThreadToolTransport::Mcp(Arc::clone(server))
} else {
ThreadToolTransport::Dynamic
}
}
@@ -631,6 +718,7 @@ impl AppServerSession {
if self.history_support == ThreadHistorySupport::LegacyOnly {
params.history_mode = None;
}
self.thread_tool_transport().configure(&mut params);
let request_handle = self.request_handle();
let (response, history_support) =
request_thread_start_with_history_fallback(&request_handle, request_id, params)
@@ -729,6 +817,8 @@ impl AppServerSession {
self.remote_cwd_override.as_deref(),
)
};
self.thread_tool_transport()
.configure_mcp(&mut params.config);
let response: ThreadForkResponse = match self
.client
.request_typed(ClientRequest::ThreadFork {
@@ -1429,14 +1519,16 @@ pub(crate) async fn start_thread_with_request_handle(
config: Config,
thread_params_mode: ThreadParamsMode,
remote_cwd_override: Option<PathBuf>,
thread_tool_transport: ThreadToolTransport,
) -> Result<AppServerStartedThread> {
let request_id = RequestId::String(format!("startup-thread-start-{}", Uuid::new_v4()));
let params = thread_start_params_from_config(
let mut params = thread_start_params_from_config(
&config,
thread_params_mode,
remote_cwd_override.as_deref(),
/*session_start_source*/ None,
);
thread_tool_transport.configure(&mut params);
let (response, _history_support) =
request_thread_start_with_history_fallback(&request_handle, request_id, params)
.await
@@ -1686,7 +1778,7 @@ fn permissions_selection_from_config(
.map(permission_profile_id_from_active_profile)
}
fn thread_start_params_from_config(
pub(crate) fn thread_start_params_from_config(
config: &Config,
thread_params_mode: ThreadParamsMode,
remote_cwd_override: Option<&std::path::Path>,
@@ -2516,6 +2608,7 @@ mod tests {
);
assert_eq!(params.model_provider, Some(config.model_provider_id));
assert_eq!(params.thread_source, Some(ThreadSource::User));
assert_eq!(params.dynamic_tools, None);
}
#[tokio::test]
@@ -67,6 +67,8 @@ impl AppServerSession {
self.remote_cwd_override.as_deref(),
model_settings,
);
self.thread_tool_transport()
.configure_mcp(&mut params.config);
let mut rollout_maintenance_guard = None;
params.exclude_turns = if self.history_support == ThreadHistorySupport::Paginated {
let known_legacy_history = self
File diff suppressed because it is too large Load Diff
+296
View File
@@ -0,0 +1,296 @@
//! Approval-gated MCP transport for task tools owned by a local-daemon TUI.
use crate::app_event_sender::AppEventSender;
use crate::dynamic_tools;
use axum::Router;
use axum::body::Body;
use axum::extract::State;
use axum::http::Request;
use axum::http::StatusCode;
use axum::http::header::AUTHORIZATION;
use axum::middleware;
use axum::middleware::Next;
use axum::response::Response;
use codex_app_server_client::AppServerRequestHandle;
use codex_app_server_protocol::DynamicToolCallOutputContentItem;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::DynamicToolNamespaceTool;
use codex_app_server_protocol::DynamicToolSpec;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_config::McpServerConfig;
use codex_config::McpServerRequirement;
use codex_config::RawMcpServerConfig;
use rmcp::ErrorData as McpError;
use rmcp::handler::server::ServerHandler;
use rmcp::model::CallToolRequestParams;
use rmcp::model::CallToolResult;
use rmcp::model::ContentBlock;
use rmcp::model::JsonObject;
use rmcp::model::ListToolsResult;
use rmcp::model::PaginatedRequestParams;
use rmcp::model::ServerCapabilities;
use rmcp::model::ServerInfo;
use rmcp::model::Tool;
use rmcp::model::ToolAnnotations;
use rmcp::service::RequestContext;
use rmcp::service::RoleServer;
use rmcp::transport::StreamableHttpServerConfig;
use rmcp::transport::StreamableHttpService;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use serde_json::Value;
use serde_json::json;
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use uuid::Uuid;
#[derive(Clone)]
pub(crate) enum ThreadToolTransport {
Disabled,
Dynamic,
Mcp(Arc<DynamicToolMcpServer>),
}
impl ThreadToolTransport {
pub(crate) fn configure(&self, params: &mut ThreadStartParams) {
match self {
Self::Disabled => params.dynamic_tools = None,
Self::Dynamic => {
params.dynamic_tools = Some(dynamic_tools::non_delegation_tool_specs());
}
Self::Mcp(_) => {
params.dynamic_tools = None;
self.configure_mcp(&mut params.config);
}
}
}
pub(crate) fn configure_mcp(&self, config: &mut Option<HashMap<String, Value>>) {
if let Self::Mcp(server) = self {
config.get_or_insert_default().insert(
format!("mcp_servers.{}", dynamic_tools::NAMESPACE),
server.config.clone(),
);
}
}
}
pub(crate) struct DynamicToolMcpServer {
config: Value,
task: JoinHandle<()>,
}
impl DynamicToolMcpServer {
pub(crate) async fn start(
request_handle: AppServerRequestHandle,
mut thread_start_params: ThreadStartParams,
app_event_tx: AppEventSender,
status_updates: broadcast::Sender<ThreadStatusChangedNotification>,
managed_requirement: Option<&McpServerRequirement>,
) -> std::io::Result<Self> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let address = listener.local_addr()?;
let authorization = Arc::new(format!("Bearer {}", Uuid::new_v4()));
let server_config = json!({
"url": format!("http://{address}/mcp"),
"http_headers": {"Authorization": authorization.as_str()},
"default_tools_approval_mode": "approve",
"tools": {
"create_thread": {"approval_mode": "prompt"},
"send_message_to_thread": {"approval_mode": "prompt"},
"fork_thread": {"approval_mode": "prompt"}
}
});
if let Some(requirement) = managed_requirement {
let raw_config = serde_json::from_value::<RawMcpServerConfig>(server_config.clone())
.map_err(|error| {
std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string())
})?;
let configured_server = McpServerConfig::try_from(raw_config)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if !configured_server.matches_requirement(requirement) {
return Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"managed MCP requirements do not permit the TUI task-tools server",
));
}
}
if let Some(overrides) = thread_start_params.config.as_mut() {
overrides.remove("web_search");
}
let handler = DynamicToolMcpHandler {
request_handle,
thread_start_params,
app_event_tx,
status_updates,
server_config: server_config.clone(),
};
let service = StreamableHttpService::new(
move || Ok(handler.clone()),
Arc::new(LocalSessionManager::default()),
StreamableHttpServerConfig::default(),
);
let router =
Router::new()
.nest_service("/mcp", service)
.layer(middleware::from_fn_with_state(
authorization,
require_authorization,
));
let task = tokio::spawn(async move {
if let Err(error) = axum::serve(listener, router).await {
tracing::warn!(%error, "TUI task-tools MCP server stopped");
}
});
Ok(Self {
config: server_config,
task,
})
}
}
impl Drop for DynamicToolMcpServer {
fn drop(&mut self) {
self.task.abort();
}
}
async fn require_authorization(
State(expected): State<Arc<String>>,
request: Request<Body>,
next: Next,
) -> Result<Response, StatusCode> {
if request
.headers()
.get(AUTHORIZATION)
.is_some_and(|value| value.as_bytes() == expected.as_bytes())
{
Ok(next.run(request).await)
} else {
Err(StatusCode::UNAUTHORIZED)
}
}
#[derive(Clone)]
struct DynamicToolMcpHandler {
request_handle: AppServerRequestHandle,
thread_start_params: ThreadStartParams,
app_event_tx: AppEventSender,
status_updates: broadcast::Sender<ThreadStatusChangedNotification>,
server_config: Value,
}
impl ServerHandler for DynamicToolMcpHandler {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
}
async fn list_tools(
&self,
_request: Option<PaginatedRequestParams>,
_context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, McpError> {
let mut tools = Vec::new();
for spec in dynamic_tools::tool_specs() {
let functions = match spec {
DynamicToolSpec::Function(function) => vec![function],
DynamicToolSpec::Namespace(namespace) => namespace
.tools
.into_iter()
.map(|tool| match tool {
DynamicToolNamespaceTool::Function(function) => function,
})
.collect(),
};
for function in functions {
let schema = serde_json::from_value::<JsonObject>(function.input_schema)
.map_err(|error| McpError::internal_error(error.to_string(), None))?;
let mut tool = Tool::new(
Cow::Owned(function.name),
Cow::Owned(function.description),
Arc::new(schema),
);
tool.annotations = Some(ToolAnnotations::new().read_only(matches!(
tool.name.as_ref(),
"list_threads" | "list_archived_threads" | "read_thread" | "wait_threads"
)));
tools.push(tool);
}
}
Ok(ListToolsResult::with_all_items(tools))
}
async fn call_tool(
&self,
request: CallToolRequestParams,
context: RequestContext<RoleServer>,
) -> Result<rmcp::model::CallToolResponse, McpError> {
let metadata = &context.meta.0.0;
let turn_metadata = metadata
.get("x-codex-turn-metadata")
.and_then(|value| match value {
Value::Object(_) => Some(value.clone()),
Value::String(value) => serde_json::from_str(value).ok(),
_ => None,
});
let thread_id = metadata
.get("threadId")
.and_then(Value::as_str)
.or_else(|| turn_metadata.as_ref()?.get("thread_id")?.as_str())
.filter(|thread_id| !thread_id.is_empty())
.ok_or_else(|| McpError::invalid_params("missing task metadata", None))?;
let turn_id = metadata
.get("turnId")
.and_then(Value::as_str)
.or_else(|| turn_metadata.as_ref()?.get("turn_id")?.as_str())
.map_or_else(|| format!("mcp-turn-{}", Uuid::new_v4()), str::to_string);
let call_id = metadata
.get("callId")
.and_then(Value::as_str)
.map_or_else(|| format!("mcp-call-{}", Uuid::new_v4()), str::to_string);
let params = DynamicToolCallParams {
thread_id: thread_id.to_string(),
turn_id,
call_id,
namespace: Some(dynamic_tools::NAMESPACE.to_string()),
tool: request.name.into_owned(),
arguments: Value::Object(request.arguments.unwrap_or_default()),
};
let mut thread_start_params = self.thread_start_params.clone();
thread_start_params.config.get_or_insert_default().insert(
format!("mcp_servers.{}", dynamic_tools::NAMESPACE),
self.server_config.clone(),
);
let response = dynamic_tools::execute(
self.request_handle.clone(),
params,
thread_start_params,
self.status_updates.subscribe(),
Some(&self.app_event_tx),
)
.await;
let content = response
.content_items
.into_iter()
.map(|item| match item {
DynamicToolCallOutputContentItem::InputText { text } => ContentBlock::text(text),
DynamicToolCallOutputContentItem::InputImage { image_url } => {
ContentBlock::text(image_url)
}
DynamicToolCallOutputContentItem::InputAudio { audio_url } => {
ContentBlock::text(audio_url)
}
})
.collect();
Ok(if response.success {
CallToolResult::success(content)
} else {
CallToolResult::error(content)
}
.into())
}
}
+689
View File
@@ -0,0 +1,689 @@
use super::*;
use crate::app_server_session::AppServerSession;
use crate::app_server_session::ResumeModelSettings;
use crate::legacy_core::config::ConfigBuilder;
use app_test_support::create_fake_paginated_rollout;
use app_test_support::create_fake_rollout;
use app_test_support::rollout_path;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
async fn test_server() -> color_eyre::Result<(TempDir, AppServerSession, String, String)> {
let codex_home = tempfile::tempdir()?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.build()
.await?;
let target = create_fake_paginated_rollout(
codex_home.path(),
"2026-01-02T00-00-00",
"2026-01-02T00:00:00Z",
"Persisted test task",
Some(config.model_provider_id.as_str()),
/*git_info*/ None,
)
.map_err(|error| color_eyre::eyre::eyre!("failed to create test rollout: {error}"))?;
let path = rollout_path(codex_home.path(), "2026-01-02T00-00-00", &target);
let mut records = std::fs::read_to_string(&path)?
.lines()
.map(serde_json::from_str::<Value>)
.collect::<Result<Vec<_>, _>>()?;
for payload in [
json!({
"type": "task_started",
"turn_id": "persisted-turn",
"model_context_window": null
}),
json!({
"type": "item_completed",
"thread_id": target,
"turn_id": "persisted-turn",
"item": {
"type": "AgentMessage",
"id": "persisted-message",
"content": [{"type": "Text", "text": "Persisted assistant output".repeat(40)}]
},
"completed_at_ms": 0
}),
json!({
"type": "item_completed",
"thread_id": target,
"turn_id": "persisted-turn",
"item": {
"type": "DynamicToolCall",
"id": "persisted-tool",
"namespace": "codex_tui",
"tool": "list_threads",
"arguments": {},
"status": "completed",
"success": true
},
"completed_at_ms": 0
}),
json!({
"type": "task_complete",
"turn_id": "persisted-turn",
"last_agent_message": "Persisted assistant output"
}),
] {
serde_json::from_value::<codex_protocol::protocol::EventMsg>(payload.clone())?;
records.push(json!({
"timestamp": "2026-01-02T00:00:00Z",
"ordinal": records.len(),
"type": "event_msg",
"payload": payload
}));
}
let records = records
.into_iter()
.map(|record| record.to_string())
.collect::<Vec<_>>()
.join("\n");
std::fs::write(path, format!("{records}\n"))?;
let mut server = crate::start_embedded_app_server_for_picker(&config).await?;
let source = server
.start_thread(&config)
.await?
.session
.thread_id
.to_string();
server
.resume_thread(
config,
ThreadId::from_string(&target)?,
ResumeModelSettings::RestoreFromThread,
)
.await?;
Ok((codex_home, server, source, target))
}
async fn call_tool(
server: &AppServerSession,
source: &str,
name: &str,
arguments: Value,
) -> DynamicToolCallResponse {
let (_status_sender, status_receiver) = broadcast::channel(/*capacity*/ 8);
execute(
server.request_handle(),
DynamicToolCallParams {
thread_id: source.to_string(),
turn_id: "persisted-turn".to_string(),
call_id: "call-1".to_string(),
namespace: Some(NAMESPACE.to_string()),
tool: name.to_string(),
arguments,
},
ThreadStartParams {
dynamic_tools: Some(tool_specs()),
ephemeral: Some(true),
..ThreadStartParams::default()
},
status_receiver,
/*app_event_tx*/ None,
)
.await
}
fn response_json(response: DynamicToolCallResponse) -> Value {
assert!(response.success, "tool call failed: {response:?}");
let [DynamicToolCallOutputContentItem::InputText { text }] = response.content_items.as_slice()
else {
panic!("expected one JSON text response")
};
serde_json::from_str(text).expect("dynamic tool response should contain JSON")
}
#[test]
fn oversized_responses_are_truncated_without_losing_identifiers() {
let response = success_response(json!({
"thread": {"threadId": "thread-1", "summary": "preview"},
"turns": [{"turnId": "turn-1", "items": [{
"type": "agentMessage", "id": "message-1",
"output": output_summary(&"🦀".repeat(MAX_RESPONSE_BYTES), MAX_RESPONSE_BYTES)
}]}]
}))
.expect("oversized responses should be shortened");
let [DynamicToolCallOutputContentItem::InputText { text }] = response.content_items.as_slice()
else {
panic!("expected one JSON text response")
};
assert!(text.len() <= MAX_RESPONSE_BYTES);
let value: Value = serde_json::from_str(text).expect("response remains valid JSON");
assert_eq!(value["thread"]["threadId"], "thread-1");
assert_eq!(value["turns"][0]["items"][0]["id"], "message-1");
assert_eq!(value["turns"][0]["items"][0]["output"]["truncated"], true);
assert_eq!(
value["turns"][0]["items"][0]["output"]["originalChars"],
MAX_RESPONSE_BYTES
);
assert_eq!(value["truncated"], true);
}
#[test]
fn oversized_task_lists_are_bounded() {
let threads: Vec<_> = (0..10)
.map(|index| {
json!({
"id": format!("00000000-0000-0000-0000-{index:012}"),
"status": "active",
"title": "A task with a descriptive title",
"summary": "A task with a longer preview",
"cwd": "/tmp/project",
"updatedAt": 123
})
})
.collect();
let value = response_json(
success_response(json!({"threads": threads}))
.expect("oversized task lists should be shortened"),
);
let threads = value["threads"].as_array().expect("task summaries");
assert!(!threads.is_empty() && threads.len() < 10);
assert_eq!(value["truncated"], true);
}
#[test]
fn oversized_wait_snapshots_preserve_all_targets() {
let polls: Vec<_> = (0..MAX_WAIT_TARGETS)
.map(|index| {
json!({
"schemaVersion": 1,
"thread": {
"id": format!("00000000-0000-0000-0000-{index:012}"),
"status": "idle"
},
"cursor": format!("opaque-cursor-{index}-{}", "x".repeat(100)),
"revision": 123,
"changed": false,
"latestTurn": null,
"latestAssistantMessageId": null,
"latestAssistantMessage": null,
"latestToolMarkerId": null,
"latestToolMarker": null
})
})
.collect();
let value = response_json(
success_response(json!({"timedOut": true, "wake": null, "polls": polls}))
.expect("all wait targets should fit a compact response"),
);
assert_eq!(
value["polls"].as_array().map(Vec::len),
Some(MAX_WAIT_TARGETS)
);
assert_eq!(value["truncated"], true);
}
#[test]
fn oversized_read_pages_preserve_turns_and_pagination() {
let turns: Vec<_> = (0..10)
.map(|turn_index| {
json!({
"id": format!("turn-{turn_index}"),
"status": "completed",
"items": (0..20)
.map(|item_index| json!({
"id": format!("00000000-0000-0000-{turn_index:04}-{item_index:012}"),
"type": "dynamicToolCall",
"namespace": "codex_tui",
"tool": "read_thread",
"status": "completed"
}))
.collect::<Vec<_>>()
})
})
.collect();
let response = success_response(json!({
"thread": {"id": "thread-1"},
"page": {"nextCursor": "opaque-next-page"},
"turns": turns
}))
.expect("oversized read pages should be shortened");
let [DynamicToolCallOutputContentItem::InputText { text }] = response.content_items.as_slice()
else {
panic!("expected one JSON text response")
};
assert!(text.len() <= MAX_RESPONSE_BYTES);
let value: Value = serde_json::from_str(text).expect("response remains valid JSON");
let turns = value["turns"].as_array().expect("turns remain present");
assert_eq!(turns.len(), 10);
assert_eq!(value["page"]["nextCursor"], "opaque-next-page");
assert!(turns.iter().all(|turn| turn["items"].is_array()));
assert!(turns.iter().any(|turn| {
turn["items"]
.as_array()
.is_some_and(|items| items.len() < 20)
}));
}
#[test]
fn delegated_prompts_match_desktop_xml_contract() {
assert_eq!(
delegated_prompt("thread-1", "Check <main> & report > status"),
"<codex_delegation>\n <source_thread_id>thread-1</source_thread_id>\n <input>Check &lt;main&gt; &amp; report &gt; status</input>\n</codex_delegation>"
);
assert!(
validate_prompt(
&delegated_prompt("thread-1", &"&".repeat(MAX_INPUT_BYTES)),
MAX_DELEGATED_INPUT_BYTES,
)
.is_err()
);
}
#[test]
fn activity_metadata_is_retained_without_including_outputs() -> color_eyre::Result<()> {
let turn: Turn = serde_json::from_value(json!({
"id": "turn-1",
"status": "completed",
"items": [
{"type": "reasoning", "id": "thought-1", "summary": ["Thinking"], "content": ["Private reasoning"]},
{"type": "commandExecution", "id": "command-1", "command": "cargo test", "cwd": "/tmp",
"status": "completed", "commandActions": [], "aggregatedOutput": "Command output", "exitCode": 0},
{"type": "fileChange", "id": "patch-1", "status": "completed",
"changes": [{"path": "src/main.rs", "kind": {"type": "add"}, "diff": "+hello"}]},
{"type": "mcpToolCall", "id": "mcp-1", "server": "docs", "tool": "search",
"status": "completed", "arguments": {}},
{"type": "userMessage", "id": "user-1", "content": [
{"type": "text", "text": delegated_prompt("source-1", "Check <main> & status")},
{"type": "skill", "name": "debug", "path": "/tmp/SKILL.md"},
{"type": "mention", "name": "docs", "path": "app://docs"}
]},
{"type": "agentMessage", "id": "assistant-1", "text": "Working", "phase": "commentary"},
{"type": "webSearch", "id": "web-1", "query": "latest docs", "action": null},
{"type": "sleep", "id": "sleep-1", "durationMs": 1000},
{"type": "imageGeneration", "id": "image-1", "status": "completed",
"revisedPrompt": "a cat", "result": "image bytes"},
{"type": "enteredReviewMode", "id": "review-1", "review": "review changes"}
]
}))?;
let summary = turn_summary(&turn, /*include_outputs*/ false, DEFAULT_OUTPUT_CHARS);
assert_eq!(
summary["items"],
json!([
{"type": "reasoning", "id": "thought-1", "summary": ["Thinking"]},
{"type": "commandExecution", "id": "command-1", "command": "cargo test", "cwd": "/tmp", "exitCode": 0, "status": "completed", "durationMs": null},
{"type": "fileChange", "id": "patch-1", "status": "completed",
"changes": [{"path": "src/main.rs", "kind": {"type": "add"}}]},
{"type": "mcpToolCall", "id": "mcp-1", "server": "docs", "tool": "search", "arguments": {}, "status": "completed", "durationMs": null},
{"type": "userMessage", "id": "user-1", "content": [
{"type": "text", "text": delegated_prompt("source-1", "Check <main> & status"),
"codexDelegation": {"sourceThreadId": "source-1", "input": "Check <main> & status"}},
{"type": "skill", "name": "debug", "path": "/tmp/SKILL.md"},
{"type": "mention", "name": "docs", "path": "app://docs"}
]},
{"type": "agentMessage", "id": "assistant-1", "text": "Working", "phase": "commentary"},
{"type": "webSearch", "id": "web-1", "query": "latest docs", "action": null},
{"type": "sleep", "id": "sleep-1", "durationMs": 1000},
{"type": "imageGeneration", "id": "image-1", "status": "completed",
"revisedPrompt": "a cat", "savedPath": null},
{"type": "enteredReviewMode", "id": "review-1", "review": "review changes"}
])
);
let full = turn_summary(&turn, /*include_outputs*/ true, DEFAULT_OUTPUT_CHARS);
assert_eq!(
full["items"][0]["content"],
json!([{"text": "Private reasoning", "truncated": false}])
);
assert_eq!(
full["items"][1]["output"],
json!({"text": "Command output", "truncated": false})
);
assert_eq!(
full["items"][2]["changes"][0]["diff"],
json!({"text": "+hello", "truncated": false})
);
assert_eq!(
full["items"][8]["result"],
json!({"text": "image bytes", "truncated": false})
);
let no_outputs = turn_summary(
&turn, /*include_outputs*/ true, /*output_chars*/ 0,
);
assert_eq!(no_outputs["items"][5]["text"], "Working");
assert_eq!(
no_outputs["items"][1]["output"],
json!({"text": "", "truncated": true, "originalChars": 14})
);
assert_eq!(
no_outputs["items"][8]["result"],
json!({"text": "", "truncated": true, "originalChars": 11})
);
Ok(())
}
#[tokio::test]
async fn task_management_tools_use_existing_app_server_operations() -> color_eyre::Result<()> {
let (codex_home, server, source, target) = test_server().await?;
let listed = response_json(call_tool(&server, &source, "list_threads", json!({})).await);
assert!(
listed["threads"]
.as_array()
.is_some_and(|threads| { threads.iter().any(|thread| thread["id"] == target) })
);
let legacy = create_fake_rollout(
codex_home.path(),
"2026-01-03T00-00-00",
"2026-01-03T00:00:00Z",
"Legacy test task",
Some("openai"),
/*git_info*/ None,
)
.map_err(|error| color_eyre::eyre::eyre!("failed to create legacy rollout: {error}"))?;
for thread_id in [&target, &legacy] {
let read = response_json(
call_tool(
&server,
&source,
"read_thread",
json!({"threadId": thread_id}),
)
.await,
);
assert_eq!(read["schemaVersion"], 1);
assert_eq!(read["thread"]["id"], *thread_id);
assert_eq!(read["page"]["order"], "newest_first");
assert!(read["turns"].is_array());
}
let renamed = response_json(
call_tool(
&server,
&source,
"set_thread_title",
json!({"threadId": target, "title": "Renamed task"}),
)
.await,
);
assert_eq!(
renamed,
json!({"threadId": target, "title": "Renamed task"})
);
let forked = response_json(
call_tool(&server, &source, "fork_thread", json!({"threadId": target})).await,
);
assert_ne!(forked["threadId"], target);
let self_forked = response_json(call_tool(&server, &target, "fork_thread", json!({})).await);
assert_ne!(self_forked["threadId"], target);
assert_eq!(self_forked["sourceThreadId"], target);
assert_eq!(
self_forked["environment"],
json!({"type": "same-directory"})
);
let self_archive = call_tool(
&server,
&source,
"set_thread_archived",
json!({"threadId": source.to_uppercase(), "archived": true}),
)
.await;
assert!(!self_archive.success);
let archived = response_json(
call_tool(
&server,
&source,
"set_thread_archived",
json!({"threadId": target, "archived": true}),
)
.await,
);
assert_eq!(archived, json!({"threadId": target, "archived": true}));
let mut expected_archived = vec![target.clone()];
for day in 4..12 {
let archived_id = create_fake_paginated_rollout(
codex_home.path(),
&format!("2026-01-{day:02}T00-00-00"),
&format!("2026-01-{day:02}T00:00:00Z"),
"Archived task with a deliberately descriptive pagination title",
Some("openai"),
/*git_info*/ None,
)
.map_err(|error| color_eyre::eyre::eyre!("failed to create archived rollout: {error}"))?;
let archived = call_tool(
&server,
&source,
"set_thread_archived",
json!({"threadId": archived_id, "archived": true}),
)
.await;
assert!(archived.success, "{archived:?}");
expected_archived.push(archived_id);
}
let mut archived_threads =
response_json(call_tool(&server, &source, "list_archived_threads", json!({})).await);
assert!(
archived_threads["threads"]
.as_array()
.is_some_and(|threads| threads.len() < expected_archived.len())
);
let mut listed_archived = Vec::new();
loop {
listed_archived.extend(
archived_threads["threads"]
.as_array()
.into_iter()
.flatten()
.filter_map(|thread| thread["id"].as_str().map(ToString::to_string)),
);
let Some(cursor) = archived_threads["nextCursor"].as_str() else {
break;
};
archived_threads = response_json(
call_tool(
&server,
&source,
"list_archived_threads",
json!({"cursor": cursor}),
)
.await,
);
}
expected_archived.sort();
listed_archived.sort();
assert_eq!(listed_archived, expected_archived);
let restored = response_json(
call_tool(
&server,
&source,
"set_thread_archived",
json!({"threadId": target, "archived": false}),
)
.await,
);
assert_eq!(restored, json!({"threadId": target, "archived": false}));
server.shutdown().await?;
Ok(())
}
#[tokio::test]
async fn wait_threads_returns_bounded_snapshots_and_rejects_self_wait() -> color_eyre::Result<()> {
let (_codex_home, server, source, target) = test_server().await?;
let snapshot = response_json(
call_tool(
&server,
&source,
"wait_threads",
json!({"targets": [{"threadId": target}, {"threadId": "missing-task"}], "timeoutMs": 0}),
)
.await,
);
assert_eq!(snapshot["timedOut"], false);
assert_eq!(snapshot["wake"]["threadId"], target);
assert_eq!(snapshot["wake"]["reason"], "turnCompleted");
assert!(snapshot.get("errors").is_none());
assert_eq!(
snapshot["wake"]["turnId"],
snapshot["polls"][0]["latestTurn"]["id"]
);
let assistant = &snapshot["polls"][0]["latestAssistantMessage"];
assert_eq!(assistant["id"], "persisted-message");
assert_eq!(
assistant["turnId"],
snapshot["polls"][0]["latestTurn"]["id"]
);
assert!(assistant["text"].as_str().is_some_and(|text| {
text.ends_with('…')
&& "Persisted assistant output"
.repeat(40)
.starts_with(text.trim_end_matches('…'))
}));
assert_eq!(
snapshot["polls"][0]["latestToolMarker"],
json!({
"id": "persisted-tool",
"turnId": "persisted-turn",
"type": "dynamicToolCall",
"name": "list_threads",
"status": "completed"
})
);
let cursor = snapshot["polls"][0]["cursor"]
.as_str()
.expect("snapshot cursor")
.to_string();
let cursor_value: Value = serde_json::from_str(&cursor)?;
assert_eq!(
cursor_value["turnId"],
snapshot["polls"][0]["latestTurn"]["id"]
);
assert_eq!(
cursor_value["turnStatus"],
snapshot["polls"][0]["latestTurn"]["status"]
);
assert_eq!(cursor_value["latestItemId"], "persisted-tool");
let unchanged = response_json(
call_tool(
&server,
&source,
"wait_threads",
json!({"targets": [{"threadId": target, "afterCursor": cursor}], "timeoutMs": 0}),
)
.await,
);
assert_eq!(unchanged["timedOut"], true);
assert_eq!(unchanged["wake"], Value::Null);
assert_eq!(unchanged["polls"][0]["changed"], false);
assert_eq!(unchanged["polls"][0]["latestAssistantMessage"], Value::Null);
let self_wait = call_tool(
&server,
&source,
"wait_threads",
json!({"targets": [{"threadId": source.to_uppercase()}], "timeoutMs": 0}),
)
.await;
assert!(!self_wait.success);
let duplicate_wait = call_tool(
&server,
&source,
"wait_threads",
json!({"targets": [{"threadId": target}, {"threadId": target.to_uppercase()}], "timeoutMs": 0}),
)
.await;
assert!(!duplicate_wait.success);
server.shutdown().await?;
Ok(())
}
#[tokio::test]
async fn task_creation_and_followup_start_background_turns() -> color_eyre::Result<()> {
let (_codex_home, server, source, target) = test_server().await?;
for (tool, arguments) in [
(
"create_thread",
json!({"prompt": "&".repeat(MAX_INPUT_BYTES)}),
),
(
"send_message_to_thread",
json!({"threadId": target, "prompt": "&".repeat(MAX_INPUT_BYTES)}),
),
] {
assert!(!call_tool(&server, &source, tool, arguments).await.success);
}
assert!(
!call_tool(
&server,
&source,
"send_message_to_thread",
json!({"threadId": target, "prompt": "Follow up", "model": ""}),
)
.await
.success
);
let ephemeral: codex_app_server_protocol::ThreadStartResponse =
request(&server.request_handle(), |request_id| {
ClientRequest::ThreadStart {
request_id,
params: ThreadStartParams {
ephemeral: Some(true),
..ThreadStartParams::default()
},
}
})
.await
.map_err(color_eyre::eyre::Error::msg)?;
let rejected = call_tool(
&server,
&ephemeral.thread.id,
"create_thread",
json!({"prompt": "Start a background task"}),
)
.await;
assert!(!rejected.success);
let created = response_json(
call_tool(
&server,
&target,
"create_thread",
json!({"prompt": "x".repeat(MAX_INPUT_BYTES), "title": "Background task"}),
)
.await,
);
assert!(created["threadId"].is_string());
let continued = response_json(
call_tool(
&server,
&source,
"send_message_to_thread",
json!({"threadId": target, "prompt": "x".repeat(MAX_INPUT_BYTES)}),
)
.await,
);
assert_eq!(continued["threadId"], target);
let oversized = call_tool(
&server,
&source,
"list_archived_threads",
json!({"cursor": "x".repeat(MAX_RESPONSE_BYTES + 1)}),
)
.await;
assert!(!oversized.success);
assert!(
matches!(&oversized.content_items[..], [DynamicToolCallOutputContentItem::InputText { text }] if text.len() <= MAX_RESPONSE_BYTES)
);
server.shutdown().await?;
Ok(())
}
+11
View File
@@ -120,6 +120,8 @@ mod cwd_prompt;
mod debug_config;
mod diff_model;
mod diff_render;
mod dynamic_tools;
mod dynamic_tools_mcp;
mod exec_cell;
mod exec_command;
mod external_agent_config_migration;
@@ -2998,6 +3000,15 @@ mod tests {
)?;
assert_eq!(config_cwd, None);
let local_daemon = AppServerTarget::LocalDaemon {
endpoint: RemoteAppServerEndpoint::UnixSocket {
socket_path: AbsolutePathBuf::relative_to_current_dir("codex.sock")?,
},
};
assert!(uses_remote_workspace_or_environment(
&local_daemon,
&environment_manager
));
Ok(())
}