fix(tui): forward cwd changes through app-server

Forward standalone turn-context overrides from the TUI to loaded app-server threads via `thread/turnContext/update`, so `/cwd` updates the core thread state instead of only local UI state.

Add protocol schemas and regression coverage showing thread-scoped shell commands use the updated cwd after the context override.
This commit is contained in:
Felipe Coury
2026-04-06 15:29:25 -03:00
parent ae617fc469
commit 93154c3378
17 changed files with 1271 additions and 2 deletions
+1
View File
@@ -143,6 +143,7 @@ Example with notification opt-out:
- `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
- `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success and emits `thread/name/updated` to initialized, opted-in clients. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
- `thread/turnContext/update` — update a loaded thread’s defaults for subsequent turns and thread-scoped operations, such as `cwd`, approval policy, approvals reviewer, sandbox policy, model, reasoning effort, service tier, collaboration mode, or personality; returns `{}` on success.
- `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
- `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
- `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
@@ -158,6 +158,8 @@ use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadTurnContextUpdateParams;
use codex_app_server_protocol::ThreadTurnContextUpdateResponse;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnarchivedNotification;
@@ -739,6 +741,10 @@ impl CodexMessageProcessor {
self.thread_set_name(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadTurnContextUpdate { request_id, params } => {
self.thread_turn_context_update(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadMetadataUpdate { request_id, params } => {
self.thread_metadata_update(to_connection_request_id(request_id), params)
.await;
@@ -2670,6 +2676,75 @@ impl CodexMessageProcessor {
.await;
}
async fn thread_turn_context_update(
&self,
request_id: ConnectionRequestId,
params: ThreadTurnContextUpdateParams,
) {
let ThreadTurnContextUpdateParams {
thread_id,
cwd,
approval_policy,
approvals_reviewer,
sandbox_policy,
model,
effort,
summary,
service_tier,
collaboration_mode,
personality,
} = params;
let (_, thread) = match self.load_thread(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let collaboration_modes_config = CollaborationModesConfig {
default_mode_request_user_input: thread.enabled(Feature::DefaultModeRequestUserInput),
};
let collaboration_mode = collaboration_mode.map(|mode| {
self.normalize_turn_start_collaboration_mode(mode, collaboration_modes_config)
});
match self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::OverrideTurnContext {
cwd,
approval_policy: approval_policy.map(AskForApproval::to_core),
approvals_reviewer: approvals_reviewer
.map(codex_app_server_protocol::ApprovalsReviewer::to_core),
sandbox_policy: sandbox_policy.map(|p| p.to_core()),
windows_sandbox_level: None,
model,
effort,
summary,
service_tier,
collaboration_mode,
personality,
},
)
.await
{
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadTurnContextUpdateResponse {})
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to update thread turn context: {err}"),
)
.await;
}
}
}
async fn thread_metadata_update(
&self,
request_id: ConnectionRequestId,
@@ -73,6 +73,7 @@ use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadSetNameParams;
use codex_app_server_protocol::ThreadShellCommandParams;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadTurnContextUpdateParams;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnsubscribeParams;
use codex_app_server_protocol::TurnCompletedNotification;
@@ -367,6 +368,15 @@ impl McpProcess {
self.send_request("thread/name/set", params).await
}
/// Send a `thread/turnContext/update` JSON-RPC request.
pub async fn send_thread_turn_context_update_request(
&mut self,
params: ThreadTurnContextUpdateParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/turnContext/update", params).await
}
/// Send a `thread/metadata/update` JSON-RPC request.
pub async fn send_thread_metadata_update_request(
&mut self,
@@ -22,6 +22,8 @@ use codex_app_server_protocol::ThreadShellCommandParams;
use codex_app_server_protocol::ThreadShellCommandResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadTurnContextUpdateParams;
use codex_app_server_protocol::ThreadTurnContextUpdateResponse;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
@@ -321,6 +323,89 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_shell_command_uses_updated_turn_context_cwd() -> Result<()> {
let tmp = TempDir::new()?;
let codex_home = tmp.path().join("codex_home");
std::fs::create_dir(&codex_home)?;
let original_workspace = tmp.path().join("original");
std::fs::create_dir(&original_workspace)?;
let updated_workspace = tmp.path().join("updated");
std::fs::create_dir(&updated_workspace)?;
let server = create_mock_responses_server_sequence(vec![]).await;
create_config_toml(
codex_home.as_path(),
&server.uri(),
"never",
&BTreeMap::default(),
)?;
let mut mcp = McpProcess::new(codex_home.as_path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(original_workspace.display().to_string()),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let update_id = mcp
.send_thread_turn_context_update_request(ThreadTurnContextUpdateParams {
thread_id: thread.id.clone(),
cwd: Some(updated_workspace.clone()),
..Default::default()
})
.await?;
let update_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(update_id)),
)
.await??;
let _: ThreadTurnContextUpdateResponse =
to_response::<ThreadTurnContextUpdateResponse>(update_resp)?;
let resolved_workspace = std::fs::canonicalize(updated_workspace.as_path())?;
let (shell_command, expected_output) = current_shell_pwd_command(resolved_workspace.as_path());
let shell_id = mcp
.send_thread_shell_command_request(ThreadShellCommandParams {
thread_id: thread.id,
command: shell_command,
})
.await?;
let shell_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(shell_id)),
)
.await??;
let _: ThreadShellCommandResponse = to_response::<ThreadShellCommandResponse>(shell_resp)?;
let started = wait_for_command_execution_started(&mut mcp, /*expected_id*/ None).await?;
let ThreadItem::CommandExecution { id, source, .. } = &started.item else {
unreachable!("helper returns command execution item");
};
let command_id = id.clone();
assert_eq!(source, &CommandExecutionSource::UserShell);
let completed = wait_for_command_execution_completed(&mut mcp, Some(&command_id)).await?;
let ThreadItem::CommandExecution {
aggregated_output, ..
} = &completed.item
else {
unreachable!("helper returns command execution item");
};
assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str()));
Ok(())
}
fn current_shell_output_command(text: &str) -> Result<(String, String)> {
let command_and_output = match default_user_shell().name() {
"powershell" => {
@@ -339,6 +424,17 @@ fn current_shell_output_command(text: &str) -> Result<(String, String)> {
Ok(command_and_output)
}
fn current_shell_pwd_command(cwd: &Path) -> (String, String) {
match default_user_shell().name() {
"powershell" => (
"(Get-Location).Path".to_string(),
format!("{}\r\n", cwd.display()),
),
"cmd" => ("cd".to_string(), format!("{}\r\n", cwd.display())),
_ => ("pwd".to_string(), format!("{}\n", cwd.display())),
}
}
async fn wait_for_command_execution_started(
mcp: &mut McpProcess,
expected_id: Option<&str>,