openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Part 6

Browse Source
This commit is contained in:
Eric Traut
2026-03-14 12:48:18 -06:00
parent 2cb13efbc8
commit fd9fe119df
7 changed files with 1681 additions and 1032 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+351 -353
View File
File diff suppressed because it is too large Load Diff
codex-rs/tui/src/app/app_server_adapter.rs
+216 -127
View File
@@ -18,7 +18,6 @@ use codex_app_server_protocol::GetAccountParams;
use codex_app_server_protocol::GetAccountRateLimitsResponse;
use codex_app_server_protocol::GetAccountResponse;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::McpServerElicitationAction;
use codex_app_server_protocol::McpServerElicitationRequestResponse;
use codex_app_server_protocol::McpServerRefreshResponse;
@@ -85,6 +84,8 @@ use codex_protocol::request_permissions::PermissionGrantScope;
use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::thread_update::ThreadUpdate;
impl App {
pub(super) async fn list_models_via_app_server(
app_server_client: &InProcessAppServerClient,
@@ -628,12 +629,10 @@ impl App {
.map_err(|err| format!("failed to encode skills/list response: {err}"))?,
)
.map_err(|err| format!("failed to convert skills/list response: {err}"))?;
self.enqueue_primary_event(Event {
self.handle_codex_event_now(Event {
id: String::new(),
msg: EventMsg::ListSkillsResponse(ListSkillsResponseEvent { skills }),
})
.await
.map_err(|err| err.to_string())?;
});
}
Op::RefreshMcpServers { config } => {
let _: McpServerRefreshResponse = send_request_with_response(
@@ -691,6 +690,8 @@ impl App {
"thread/rollback",
)
.await?;
self.route_thread_update(ThreadUpdate::ThreadRolledBack { num_turns })
.await;
}
Op::Review { review_request } => {
let response: ReviewStartResponse = send_request_with_response(
@@ -758,10 +759,10 @@ impl App {
.insert(params.item_id.clone(), request_id.clone());
}
ServerRequest::McpServerElicitationRequest { request_id, params } => {
self.pending_elicitation_request_queues
.entry(params.server_name.clone())
.or_default()
.push_back(request_id.clone());
self.pending_elicitation_request_ids.insert(
(params.server_name.clone(), request_id.to_string()),
request_id.clone(),
);
}
ServerRequest::PermissionsRequestApproval { request_id, params } => {
self.pending_permissions_request_ids
@@ -784,6 +785,24 @@ impl App {
}
}
async fn route_thread_update(&mut self, update: ThreadUpdate) {
let Some(thread_id_text) = update.thread_id() else {
self.handle_thread_update_now(update);
return;
};
let Ok(thread_id) = ThreadId::from_string(&thread_id_text) else {
tracing::warn!("failed to parse thread id for thread update");
return;
};
if self.primary_thread_id.is_none() || self.primary_thread_id == Some(thread_id) {
if let Err(err) = self.enqueue_primary_update(update).await {
tracing::warn!("{err}");
}
} else if let Err(err) = self.handle_routed_thread_update(thread_id, update).await {
tracing::warn!("{err}");
}
}
async fn handle_server_notification(
&mut self,
app_server_client: &InProcessAppServerClient,
@@ -791,9 +810,12 @@ impl App {
) {
match notification {
ServerNotification::TurnStarted(payload) => {
let active_turn_id = payload.turn.id.clone();
if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) {
self.active_turn_ids.insert(thread_id, payload.turn.id);
self.active_turn_ids.insert(thread_id, active_turn_id);
}
self.route_thread_update(ThreadUpdate::TurnStarted(payload))
.await;
}
ServerNotification::TurnCompleted(payload) => {
if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id)
@@ -801,34 +823,146 @@ impl App {
{
self.active_turn_ids.remove(&thread_id);
}
self.route_thread_update(ThreadUpdate::TurnCompleted(payload))
.await;
}
ServerNotification::ThreadStarted(payload) => {
let agent_nickname = payload.thread.agent_nickname.clone();
let agent_role = payload.thread.agent_role.clone();
if let Ok(thread_id) = ThreadId::from_string(&payload.thread.id) {
self.upsert_agent_picker_thread(
thread_id,
payload.thread.agent_nickname,
payload.thread.agent_role,
false,
);
self.upsert_agent_picker_thread(thread_id, agent_nickname, agent_role, false);
}
self.route_thread_update(ThreadUpdate::ThreadStarted(payload))
.await;
}
ServerNotification::ThreadStatusChanged(payload) => {
self.route_thread_update(ThreadUpdate::ThreadStatusChanged(payload))
.await;
}
ServerNotification::ThreadClosed(payload) => {
if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) {
self.mark_agent_picker_thread_closed(thread_id);
self.active_turn_ids.remove(&thread_id);
}
self.route_thread_update(ThreadUpdate::ThreadClosed(payload))
.await;
}
ServerNotification::ThreadNameUpdated(payload) => {
if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id)
&& let Some(channel) = self.thread_event_channels.get(&thread_id)
{
let mut store = channel.store.lock().await;
if let Some(event) = store.session_configured.as_mut()
&& let EventMsg::SessionConfigured(session) = &mut event.msg
{
if let Some(session) = store.session_configured.as_mut() {
session.thread_name = payload.thread_name.clone();
}
}
self.route_thread_update(ThreadUpdate::ThreadNameUpdated(payload))
.await;
}
ServerNotification::ThreadTokenUsageUpdated(payload) => {
self.route_thread_update(ThreadUpdate::ThreadTokenUsageUpdated(payload))
.await;
}
ServerNotification::TurnDiffUpdated(payload) => {
self.route_thread_update(ThreadUpdate::TurnDiffUpdated(payload))
.await;
}
ServerNotification::TurnPlanUpdated(payload) => {
self.route_thread_update(ThreadUpdate::TurnPlanUpdated(payload))
.await;
}
ServerNotification::ItemStarted(payload) => {
self.route_thread_update(ThreadUpdate::ItemStarted(payload))
.await;
}
ServerNotification::ItemGuardianApprovalReviewStarted(payload) => {
self.route_thread_update(ThreadUpdate::ItemGuardianApprovalReviewStarted(payload))
.await;
}
ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => {
self.route_thread_update(ThreadUpdate::ItemGuardianApprovalReviewCompleted(
payload,
))
.await;
}
ServerNotification::ItemCompleted(payload) => {
self.route_thread_update(ThreadUpdate::ItemCompleted(payload))
.await;
}
ServerNotification::AgentMessageDelta(payload) => {
self.route_thread_update(ThreadUpdate::AgentMessageDelta(payload))
.await;
}
ServerNotification::PlanDelta(payload) => {
self.route_thread_update(ThreadUpdate::PlanDelta(payload))
.await;
}
ServerNotification::ReasoningSummaryTextDelta(payload) => {
self.route_thread_update(ThreadUpdate::ReasoningSummaryTextDelta(payload))
.await;
}
ServerNotification::ReasoningSummaryPartAdded(payload) => {
self.route_thread_update(ThreadUpdate::ReasoningSummaryPartAdded(payload))
.await;
}
ServerNotification::ReasoningTextDelta(payload) => {
self.route_thread_update(ThreadUpdate::ReasoningTextDelta(payload))
.await;
}
ServerNotification::TerminalInteraction(payload) => {
self.route_thread_update(ThreadUpdate::TerminalInteraction(payload))
.await;
}
ServerNotification::CommandExecutionOutputDelta(payload) => {
self.route_thread_update(ThreadUpdate::CommandExecutionOutputDelta(payload))
.await;
}
ServerNotification::FileChangeOutputDelta(payload) => {
self.route_thread_update(ThreadUpdate::FileChangeOutputDelta(payload))
.await;
}
ServerNotification::McpToolCallProgress(payload) => {
self.route_thread_update(ThreadUpdate::McpToolCallProgress(payload))
.await;
}
ServerNotification::HookStarted(payload) => {
self.route_thread_update(ThreadUpdate::HookStarted(payload))
.await;
}
ServerNotification::HookCompleted(payload) => {
self.route_thread_update(ThreadUpdate::HookCompleted(payload))
.await;
}
ServerNotification::Error(payload) => {
self.route_thread_update(ThreadUpdate::Error(payload)).await;
}
ServerNotification::ModelRerouted(payload) => {
self.route_thread_update(ThreadUpdate::ModelRerouted(payload))
.await;
}
ServerNotification::DeprecationNotice(payload) => {
self.route_thread_update(ThreadUpdate::DeprecationNotice(payload))
.await;
}
ServerNotification::ThreadRealtimeStarted(payload) => {
self.route_thread_update(ThreadUpdate::ThreadRealtimeStarted(payload))
.await;
}
ServerNotification::ThreadRealtimeItemAdded(payload) => {
self.route_thread_update(ThreadUpdate::ThreadRealtimeItemAdded(payload))
.await;
}
ServerNotification::ThreadRealtimeOutputAudioDelta(payload) => {
self.route_thread_update(ThreadUpdate::ThreadRealtimeOutputAudioDelta(payload))
.await;
}
ServerNotification::ThreadRealtimeError(payload) => {
self.route_thread_update(ThreadUpdate::ThreadRealtimeError(payload))
.await;
}
ServerNotification::ThreadRealtimeClosed(payload) => {
self.route_thread_update(ThreadUpdate::ThreadRealtimeClosed(payload))
.await;
}
ServerNotification::AccountUpdated(_) => {
match Self::read_account_via_app_server(app_server_client).await {
@@ -857,22 +991,24 @@ impl App {
true,
);
}
ServerNotification::SkillsChanged(_) => {
if let Some(thread_id) = self.primary_thread_id {
let _ = self
.submit_app_server_op_inner(
app_server_client,
thread_id,
Op::ListSkills {
cwds: Vec::new(),
force_reload: true,
},
)
.await;
}
}
_ => {}
}
}
fn maybe_correlate_legacy_server_request(&mut self, event: &Event) {
if let EventMsg::ElicitationRequest(ev) = &event.msg
&& let Some(queue) = self
.pending_elicitation_request_queues
.get_mut(&ev.server_name)
&& let Some(request_id) = queue.pop_front()
{
self.pending_elicitation_request_ids
.insert((ev.server_name.clone(), ev.id.to_string()), request_id);
}
}
pub(super) async fn handle_app_server_event(
&mut self,
app_server_client: &InProcessAppServerClient,
@@ -892,51 +1028,58 @@ impl App {
.await;
}
InProcessServerEvent::LegacyNotification(notification) => {
match decode_legacy_notification(notification) {
Ok(decoded) => {
self.maybe_correlate_legacy_server_request(&decoded.event);
match decoded.conversation_id {
Some(conversation_id) => {
match ThreadId::from_string(&conversation_id) {
Ok(thread_id) => {
if self.primary_thread_id.is_none()
|| self.primary_thread_id == Some(thread_id)
{
if let Err(err) =
self.enqueue_primary_event(decoded.event).await
{
tracing::warn!("{err}");
}
} else if let Err(err) = self
.handle_routed_thread_event(thread_id, decoded.event)
.await
{
tracing::warn!("{err}");
}
}
Err(err) => {
tracing::warn!(
conversation_id,
error = %err,
"failed to parse app-server conversation id"
);
}
}
}
None => {
if let Err(err) = self.enqueue_primary_event(decoded.event).await {
tracing::warn!("{err}");
}
}
}
}
Err(err) => {
tracing::warn!("{err}");
}
}
tracing::warn!(
notification.method = %notification.method,
"unexpected legacy notification after TUI app-server typed-event migration"
);
}
InProcessServerEvent::ServerRequest(request) => {
self.note_server_request(&request);
match request.clone() {
ServerRequest::CommandExecutionRequestApproval { request_id, params } => {
self.route_thread_update(ThreadUpdate::CommandExecutionRequestApproval {
request_id,
params,
})
.await;
}
ServerRequest::FileChangeRequestApproval { request_id, params } => {
self.route_thread_update(ThreadUpdate::FileChangeRequestApproval {
request_id,
params,
})
.await;
}
ServerRequest::McpServerElicitationRequest { request_id, params } => {
self.route_thread_update(ThreadUpdate::McpServerElicitationRequest {
request_id,
params,
})
.await;
}
ServerRequest::PermissionsRequestApproval { request_id, params } => {
self.route_thread_update(ThreadUpdate::PermissionsRequestApproval {
request_id,
params,
})
.await;
}
ServerRequest::ToolRequestUserInput { request_id, params } => {
self.route_thread_update(ThreadUpdate::ToolRequestUserInput {
request_id,
params,
})
.await;
}
ServerRequest::DynamicToolCall { request_id, params } => {
self.route_thread_update(ThreadUpdate::DynamicToolCall {
request_id,
params,
})
.await;
}
_ => {}
}
if let ServerRequest::ChatgptAuthTokensRefresh {
request_id,
params: _,
@@ -1216,10 +1359,6 @@ fn elicitation_action_to_api(action: ElicitationAction) -> McpServerElicitationA
}
}
fn normalize_legacy_notification_method(method: &str) -> &str {
method.strip_prefix("codex/event/").unwrap_or(method)
}
fn lagged_event_warning_message(skipped: usize) -> String {
format!("in-process app-server event stream lagged; dropped {skipped} events")
}
@@ -1303,56 +1442,6 @@ fn credits_snapshot_from_api(
}
}
struct DecodedLegacyNotification {
conversation_id: Option<String>,
event: Event,
}
fn decode_legacy_notification(
notification: JSONRPCNotification,
) -> Result<DecodedLegacyNotification, String> {
let value = notification
.params
.unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new()));
let method = notification.method;
let normalized_method = normalize_legacy_notification_method(&method).to_string();
let serde_json::Value::Object(mut object) = value else {
return Err(format!(
"legacy notification `{method}` params were not an object"
));
};
let conversation_id = object
.get("conversationId")
.and_then(serde_json::Value::as_str)
.map(str::to_owned);
let mut event_payload = if let Some(serde_json::Value::Object(msg_payload)) = object.get("msg")
{
serde_json::Value::Object(msg_payload.clone())
} else {
object.remove("conversationId");
serde_json::Value::Object(object)
};
let serde_json::Value::Object(ref mut object) = event_payload else {
return Err(format!(
"legacy notification `{method}` event payload was not an object"
));
};
object.insert(
"type".to_string(),
serde_json::Value::String(normalized_method),
);
let msg: EventMsg = serde_json::from_value(event_payload)
.map_err(|err| format!("failed to decode event: {err}"))?;
Ok(DecodedLegacyNotification {
conversation_id,
event: Event {
id: String::new(),
msg,
},
})
}
async fn resolve_server_request<T>(
client: &InProcessAppServerClient,
request_id: RequestId,
codex-rs/tui/src/app/pending_interactive_replay.rs
+136 -552
View File
@@ -1,5 +1,4 @@
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use crate::thread_update::ThreadUpdate;
use codex_protocol::protocol::Op;
use std::collections::HashMap;
use std::collections::HashSet;
@@ -7,11 +6,11 @@ use std::collections::HashSet;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct ElicitationRequestKey {
server_name: String,
request_id: codex_protocol::mcp::RequestId,
request_id: String,
}
impl ElicitationRequestKey {
fn new(server_name: String, request_id: codex_protocol::mcp::RequestId) -> Self {
fn new(server_name: String, request_id: String) -> Self {
Self {
server_name,
request_id,
@@ -20,44 +19,30 @@ impl ElicitationRequestKey {
}
#[derive(Debug, Default)]
// Tracks which interactive prompts are still unresolved in the thread-event buffer.
//
// Thread snapshots are replayed when switching threads/agents. Most events should replay
// verbatim, but interactive prompts (approvals, request_user_input, MCP elicitations) must
// only replay if they are still pending. This state is updated from:
// - inbound events (`note_event`)
// - outbound ops that resolve a prompt (`note_outbound_op`)
// - buffer eviction (`note_evicted_event`)
//
// We keep both fast lookup sets (for snapshot filtering by call_id/request key) and
// turn-indexed queues/vectors so `TurnComplete`/`TurnAborted` can clear stale prompts tied
// to a turn. `request_user_input` removal is FIFO because the overlay answers queued prompts
// in FIFO order for a shared `turn_id`.
pub(super) struct PendingInteractiveReplayState {
exec_approval_call_ids: HashSet<String>,
exec_approval_call_ids_by_turn_id: HashMap<String, Vec<String>>,
patch_approval_call_ids: HashSet<String>,
patch_approval_call_ids_by_turn_id: HashMap<String, Vec<String>>,
exec_approval_ids: HashSet<String>,
exec_approval_ids_by_turn_id: HashMap<String, Vec<String>>,
patch_approval_ids: HashSet<String>,
patch_approval_ids_by_turn_id: HashMap<String, Vec<String>>,
elicitation_requests: HashSet<ElicitationRequestKey>,
request_permissions_call_ids: HashSet<String>,
request_permissions_call_ids_by_turn_id: HashMap<String, Vec<String>>,
request_user_input_call_ids: HashSet<String>,
request_user_input_call_ids_by_turn_id: HashMap<String, Vec<String>>,
request_permissions_ids: HashSet<String>,
request_permissions_ids_by_turn_id: HashMap<String, Vec<String>>,
request_user_input_ids: HashSet<String>,
request_user_input_ids_by_turn_id: HashMap<String, Vec<String>>,
}
impl PendingInteractiveReplayState {
pub(super) fn event_can_change_pending_thread_approvals(event: &Event) -> bool {
pub(super) fn update_can_change_pending_thread_approvals(update: &ThreadUpdate) -> bool {
matches!(
&event.msg,
EventMsg::ExecApprovalRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::ElicitationRequest(_)
| EventMsg::RequestPermissions(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::PatchApplyBegin(_)
| EventMsg::TurnComplete(_)
| EventMsg::TurnAborted(_)
| EventMsg::ShutdownComplete
update,
ThreadUpdate::CommandExecutionRequestApproval { .. }
| ThreadUpdate::FileChangeRequestApproval { .. }
| ThreadUpdate::McpServerElicitationRequest { .. }
| ThreadUpdate::PermissionsRequestApproval { .. }
| ThreadUpdate::ToolRequestUserInput { .. }
| ThreadUpdate::TurnCompleted(_)
| ThreadUpdate::ThreadClosed(_)
| ThreadUpdate::ThreadRolledBack { .. }
)
}
@@ -76,21 +61,18 @@ impl PendingInteractiveReplayState {
pub(super) fn note_outbound_op(&mut self, op: &Op) {
match op {
Op::ExecApproval { id, turn_id, .. } => {
self.exec_approval_call_ids.remove(id);
self.exec_approval_ids.remove(id);
if let Some(turn_id) = turn_id {
Self::remove_call_id_from_turn_map_entry(
&mut self.exec_approval_call_ids_by_turn_id,
&mut self.exec_approval_ids_by_turn_id,
turn_id,
id,
);
}
}
Op::PatchApproval { id, .. } => {
self.patch_approval_call_ids.remove(id);
Self::remove_call_id_from_turn_map(
&mut self.patch_approval_call_ids_by_turn_id,
id,
);
self.patch_approval_ids.remove(id);
Self::remove_call_id_from_turn_map(&mut self.patch_approval_ids_by_turn_id, id);
}
Op::ResolveElicitation {
server_name,
@@ -100,32 +82,29 @@ impl PendingInteractiveReplayState {
self.elicitation_requests
.remove(&ElicitationRequestKey::new(
server_name.clone(),
request_id.clone(),
request_id.to_string(),
));
}
Op::RequestPermissionsResponse { id, .. } => {
self.request_permissions_call_ids.remove(id);
self.request_permissions_ids.remove(id);
Self::remove_call_id_from_turn_map(
&mut self.request_permissions_call_ids_by_turn_id,
&mut self.request_permissions_ids_by_turn_id,
id,
);
}
// `Op::UserInputAnswer` identifies the turn, not the prompt call_id. The UI
// answers queued prompts for the same turn in FIFO order, so remove the oldest
// queued call_id for that turn.
Op::UserInputAnswer { id, .. } => {
let mut remove_turn_entry = false;
if let Some(call_ids) = self.request_user_input_call_ids_by_turn_id.get_mut(id) {
if !call_ids.is_empty() {
let call_id = call_ids.remove(0);
self.request_user_input_call_ids.remove(&call_id);
if let Some(item_ids) = self.request_user_input_ids_by_turn_id.get_mut(id) {
if !item_ids.is_empty() {
let item_id = item_ids.remove(0);
self.request_user_input_ids.remove(&item_id);
}
if call_ids.is_empty() {
if item_ids.is_empty() {
remove_turn_entry = true;
}
}
if remove_turn_entry {
self.request_user_input_call_ids_by_turn_id.remove(id);
self.request_user_input_ids_by_turn_id.remove(id);
}
}
Op::Shutdown => self.clear(),
@@ -133,202 +112,169 @@ impl PendingInteractiveReplayState {
}
}
pub(super) fn note_event(&mut self, event: &Event) {
match &event.msg {
EventMsg::ExecApprovalRequest(ev) => {
let approval_id = ev.effective_approval_id();
self.exec_approval_call_ids.insert(approval_id.clone());
self.exec_approval_call_ids_by_turn_id
.entry(ev.turn_id.clone())
pub(super) fn note_update(&mut self, update: &ThreadUpdate) {
match update {
ThreadUpdate::CommandExecutionRequestApproval { params, .. } => {
let approval_id = params
.approval_id
.clone()
.unwrap_or_else(|| params.item_id.clone());
self.exec_approval_ids.insert(approval_id.clone());
self.exec_approval_ids_by_turn_id
.entry(params.turn_id.clone())
.or_default()
.push(approval_id);
}
EventMsg::ExecCommandBegin(ev) => {
self.exec_approval_call_ids.remove(&ev.call_id);
Self::remove_call_id_from_turn_map(
&mut self.exec_approval_call_ids_by_turn_id,
&ev.call_id,
);
}
EventMsg::ApplyPatchApprovalRequest(ev) => {
self.patch_approval_call_ids.insert(ev.call_id.clone());
self.patch_approval_call_ids_by_turn_id
.entry(ev.turn_id.clone())
ThreadUpdate::FileChangeRequestApproval { params, .. } => {
self.patch_approval_ids.insert(params.item_id.clone());
self.patch_approval_ids_by_turn_id
.entry(params.turn_id.clone())
.or_default()
.push(ev.call_id.clone());
.push(params.item_id.clone());
}
EventMsg::PatchApplyBegin(ev) => {
self.patch_approval_call_ids.remove(&ev.call_id);
Self::remove_call_id_from_turn_map(
&mut self.patch_approval_call_ids_by_turn_id,
&ev.call_id,
);
}
EventMsg::ElicitationRequest(ev) => {
ThreadUpdate::McpServerElicitationRequest { request_id, params } => {
self.elicitation_requests.insert(ElicitationRequestKey::new(
ev.server_name.clone(),
ev.id.clone(),
params.server_name.clone(),
request_id.to_string(),
));
}
EventMsg::RequestUserInput(ev) => {
self.request_user_input_call_ids.insert(ev.call_id.clone());
self.request_user_input_call_ids_by_turn_id
.entry(ev.turn_id.clone())
ThreadUpdate::ToolRequestUserInput { params, .. } => {
self.request_user_input_ids.insert(params.item_id.clone());
self.request_user_input_ids_by_turn_id
.entry(params.turn_id.clone())
.or_default()
.push(ev.call_id.clone());
.push(params.item_id.clone());
}
EventMsg::RequestPermissions(ev) => {
self.request_permissions_call_ids.insert(ev.call_id.clone());
self.request_permissions_call_ids_by_turn_id
.entry(ev.turn_id.clone())
ThreadUpdate::PermissionsRequestApproval { params, .. } => {
self.request_permissions_ids.insert(params.item_id.clone());
self.request_permissions_ids_by_turn_id
.entry(params.turn_id.clone())
.or_default()
.push(ev.call_id.clone());
.push(params.item_id.clone());
}
// A turn ending (normally or aborted/replaced) invalidates any unresolved
// turn-scoped approvals, permission prompts, and request_user_input prompts.
EventMsg::TurnComplete(ev) => {
self.clear_exec_approval_turn(&ev.turn_id);
self.clear_patch_approval_turn(&ev.turn_id);
self.clear_request_permissions_turn(&ev.turn_id);
self.clear_request_user_input_turn(&ev.turn_id);
ThreadUpdate::TurnCompleted(notification) => {
self.clear_exec_approval_turn(&notification.turn.id);
self.clear_patch_approval_turn(&notification.turn.id);
self.clear_request_permissions_turn(&notification.turn.id);
self.clear_request_user_input_turn(&notification.turn.id);
}
EventMsg::TurnAborted(ev) => {
if let Some(turn_id) = &ev.turn_id {
self.clear_exec_approval_turn(turn_id);
self.clear_patch_approval_turn(turn_id);
self.clear_request_permissions_turn(turn_id);
self.clear_request_user_input_turn(turn_id);
}
}
EventMsg::ShutdownComplete => self.clear(),
ThreadUpdate::ThreadClosed(_) | ThreadUpdate::ThreadRolledBack { .. } => self.clear(),
_ => {}
}
}
pub(super) fn note_evicted_event(&mut self, event: &Event) {
match &event.msg {
EventMsg::ExecApprovalRequest(ev) => {
let approval_id = ev.effective_approval_id();
self.exec_approval_call_ids.remove(&approval_id);
pub(super) fn note_evicted_update(&mut self, update: &ThreadUpdate) {
match update {
ThreadUpdate::CommandExecutionRequestApproval { params, .. } => {
let approval_id = params
.approval_id
.clone()
.unwrap_or_else(|| params.item_id.clone());
self.exec_approval_ids.remove(&approval_id);
Self::remove_call_id_from_turn_map_entry(
&mut self.exec_approval_call_ids_by_turn_id,
&ev.turn_id,
&mut self.exec_approval_ids_by_turn_id,
&params.turn_id,
&approval_id,
);
}
EventMsg::ApplyPatchApprovalRequest(ev) => {
self.patch_approval_call_ids.remove(&ev.call_id);
ThreadUpdate::FileChangeRequestApproval { params, .. } => {
self.patch_approval_ids.remove(&params.item_id);
Self::remove_call_id_from_turn_map_entry(
&mut self.patch_approval_call_ids_by_turn_id,
&ev.turn_id,
&ev.call_id,
&mut self.patch_approval_ids_by_turn_id,
&params.turn_id,
&params.item_id,
);
}
EventMsg::ElicitationRequest(ev) => {
ThreadUpdate::McpServerElicitationRequest { request_id, params } => {
self.elicitation_requests
.remove(&ElicitationRequestKey::new(
ev.server_name.clone(),
ev.id.clone(),
params.server_name.clone(),
request_id.to_string(),
));
}
EventMsg::RequestUserInput(ev) => {
self.request_user_input_call_ids.remove(&ev.call_id);
let mut remove_turn_entry = false;
if let Some(call_ids) = self
.request_user_input_call_ids_by_turn_id
.get_mut(&ev.turn_id)
{
call_ids.retain(|call_id| call_id != &ev.call_id);
if call_ids.is_empty() {
remove_turn_entry = true;
}
}
if remove_turn_entry {
self.request_user_input_call_ids_by_turn_id
.remove(&ev.turn_id);
}
ThreadUpdate::ToolRequestUserInput { params, .. } => {
self.request_user_input_ids.remove(&params.item_id);
Self::remove_call_id_from_turn_map_entry(
&mut self.request_user_input_ids_by_turn_id,
&params.turn_id,
&params.item_id,
);
}
EventMsg::RequestPermissions(ev) => {
self.request_permissions_call_ids.remove(&ev.call_id);
let mut remove_turn_entry = false;
if let Some(call_ids) = self
.request_permissions_call_ids_by_turn_id
.get_mut(&ev.turn_id)
{
call_ids.retain(|call_id| call_id != &ev.call_id);
if call_ids.is_empty() {
remove_turn_entry = true;
}
}
if remove_turn_entry {
self.request_permissions_call_ids_by_turn_id
.remove(&ev.turn_id);
}
ThreadUpdate::PermissionsRequestApproval { params, .. } => {
self.request_permissions_ids.remove(&params.item_id);
Self::remove_call_id_from_turn_map_entry(
&mut self.request_permissions_ids_by_turn_id,
&params.turn_id,
&params.item_id,
);
}
_ => {}
}
}
pub(super) fn should_replay_snapshot_event(&self, event: &Event) -> bool {
match &event.msg {
EventMsg::ExecApprovalRequest(ev) => self
.exec_approval_call_ids
.contains(&ev.effective_approval_id()),
EventMsg::ApplyPatchApprovalRequest(ev) => {
self.patch_approval_call_ids.contains(&ev.call_id)
pub(super) fn should_replay_snapshot_update(&self, update: &ThreadUpdate) -> bool {
match update {
ThreadUpdate::CommandExecutionRequestApproval { params, .. } => self
.exec_approval_ids
.contains(params.approval_id.as_ref().unwrap_or(&params.item_id)),
ThreadUpdate::FileChangeRequestApproval { params, .. } => {
self.patch_approval_ids.contains(&params.item_id)
}
EventMsg::ElicitationRequest(ev) => {
self.elicitation_requests
.contains(&ElicitationRequestKey::new(
ev.server_name.clone(),
ev.id.clone(),
))
ThreadUpdate::McpServerElicitationRequest { request_id, params } => self
.elicitation_requests
.contains(&ElicitationRequestKey::new(
params.server_name.clone(),
request_id.to_string(),
)),
ThreadUpdate::ToolRequestUserInput { params, .. } => {
self.request_user_input_ids.contains(&params.item_id)
}
EventMsg::RequestUserInput(ev) => {
self.request_user_input_call_ids.contains(&ev.call_id)
}
EventMsg::RequestPermissions(ev) => {
self.request_permissions_call_ids.contains(&ev.call_id)
ThreadUpdate::PermissionsRequestApproval { params, .. } => {
self.request_permissions_ids.contains(&params.item_id)
}
_ => true,
}
}
pub(super) fn has_pending_thread_approvals(&self) -> bool {
!self.exec_approval_call_ids.is_empty()
|| !self.patch_approval_call_ids.is_empty()
!self.exec_approval_ids.is_empty()
|| !self.patch_approval_ids.is_empty()
|| !self.elicitation_requests.is_empty()
|| !self.request_permissions_call_ids.is_empty()
|| !self.request_permissions_ids.is_empty()
}
fn clear(&mut self) {
*self = Self::default();
}
fn clear_request_user_input_turn(&mut self, turn_id: &str) {
if let Some(call_ids) = self.request_user_input_call_ids_by_turn_id.remove(turn_id) {
for call_id in call_ids {
self.request_user_input_call_ids.remove(&call_id);
if let Some(item_ids) = self.request_user_input_ids_by_turn_id.remove(turn_id) {
for item_id in item_ids {
self.request_user_input_ids.remove(&item_id);
}
}
}
fn clear_request_permissions_turn(&mut self, turn_id: &str) {
if let Some(call_ids) = self.request_permissions_call_ids_by_turn_id.remove(turn_id) {
for call_id in call_ids {
self.request_permissions_call_ids.remove(&call_id);
if let Some(item_ids) = self.request_permissions_ids_by_turn_id.remove(turn_id) {
for item_id in item_ids {
self.request_permissions_ids.remove(&item_id);
}
}
}
fn clear_exec_approval_turn(&mut self, turn_id: &str) {
if let Some(call_ids) = self.exec_approval_call_ids_by_turn_id.remove(turn_id) {
for call_id in call_ids {
self.exec_approval_call_ids.remove(&call_id);
if let Some(item_ids) = self.exec_approval_ids_by_turn_id.remove(turn_id) {
for item_id in item_ids {
self.exec_approval_ids.remove(&item_id);
}
}
}
fn clear_patch_approval_turn(&mut self, turn_id: &str) {
if let Some(call_ids) = self.patch_approval_call_ids_by_turn_id.remove(turn_id) {
for call_id in call_ids {
self.patch_approval_call_ids.remove(&call_id);
if let Some(item_ids) = self.patch_approval_ids_by_turn_id.remove(turn_id) {
for item_id in item_ids {
self.patch_approval_ids.remove(&item_id);
}
}
}
@@ -359,366 +305,4 @@ impl PendingInteractiveReplayState {
call_ids_by_turn_id.remove(turn_id);
}
}
fn clear(&mut self) {
self.exec_approval_call_ids.clear();
self.exec_approval_call_ids_by_turn_id.clear();
self.patch_approval_call_ids.clear();
self.patch_approval_call_ids_by_turn_id.clear();
self.elicitation_requests.clear();
self.request_permissions_call_ids.clear();
self.request_permissions_call_ids_by_turn_id.clear();
self.request_user_input_call_ids.clear();
self.request_user_input_call_ids_by_turn_id.clear();
}
}
#[cfg(test)]
mod tests {
use super::super::ThreadEventStore;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::TurnAbortReason;
use pretty_assertions::assert_eq;
use std::collections::HashMap;
use std::path::PathBuf;
#[test]
fn thread_event_snapshot_keeps_pending_request_user_input() {
let mut store = ThreadEventStore::new(8);
let request = Event {
id: "ev-1".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
};
store.push_event(request);
let snapshot = store.snapshot();
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first().map(|event| &event.msg),
Some(EventMsg::RequestUserInput(_))
));
}
#[test]
fn thread_event_snapshot_drops_resolved_request_user_input_after_user_answer() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
store.note_outbound_op(&Op::UserInputAnswer {
id: "turn-1".to_string(),
response: codex_protocol::request_user_input::RequestUserInputResponse {
answers: HashMap::new(),
},
});
let snapshot = store.snapshot();
assert!(
snapshot.events.is_empty(),
"resolved request_user_input prompt should not replay on thread switch"
);
}
#[test]
fn thread_event_snapshot_drops_resolved_exec_approval_after_outbound_approval_id() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::ExecApprovalRequest(
codex_protocol::protocol::ExecApprovalRequestEvent {
call_id: "call-1".to_string(),
approval_id: Some("approval-1".to_string()),
turn_id: "turn-1".to_string(),
command: vec!["echo".to_string(), "hi".to_string()],
cwd: PathBuf::from("/tmp"),
reason: None,
network_approval_context: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
additional_permissions: None,
skill_metadata: None,
available_decisions: None,
parsed_cmd: Vec::new(),
},
),
});
store.note_outbound_op(&Op::ExecApproval {
id: "approval-1".to_string(),
turn_id: Some("turn-1".to_string()),
decision: codex_protocol::protocol::ReviewDecision::Approved,
});
let snapshot = store.snapshot();
assert!(
snapshot.events.is_empty(),
"resolved exec approval prompt should not replay on thread switch"
);
}
#[test]
fn thread_event_snapshot_drops_answered_request_user_input_for_multi_prompt_turn() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
store.note_outbound_op(&Op::UserInputAnswer {
id: "turn-1".to_string(),
response: codex_protocol::request_user_input::RequestUserInputResponse {
answers: HashMap::new(),
},
});
store.push_event(Event {
id: "ev-2".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-2".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
let snapshot = store.snapshot();
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first().map(|event| &event.msg),
Some(EventMsg::RequestUserInput(ev)) if ev.call_id == "call-2"
));
}
#[test]
fn thread_event_snapshot_keeps_newer_request_user_input_pending_when_same_turn_has_queue() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
store.push_event(Event {
id: "ev-2".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-2".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
store.note_outbound_op(&Op::UserInputAnswer {
id: "turn-1".to_string(),
response: codex_protocol::request_user_input::RequestUserInputResponse {
answers: HashMap::new(),
},
});
let snapshot = store.snapshot();
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first().map(|event| &event.msg),
Some(EventMsg::RequestUserInput(ev)) if ev.call_id == "call-2"
));
}
#[test]
fn thread_event_snapshot_drops_resolved_patch_approval_after_outbound_approval() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::ApplyPatchApprovalRequest(
codex_protocol::protocol::ApplyPatchApprovalRequestEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
changes: HashMap::new(),
reason: None,
grant_root: None,
},
),
});
store.note_outbound_op(&Op::PatchApproval {
id: "call-1".to_string(),
decision: codex_protocol::protocol::ReviewDecision::Approved,
});
let snapshot = store.snapshot();
assert!(
snapshot.events.is_empty(),
"resolved patch approval prompt should not replay on thread switch"
);
}
#[test]
fn thread_event_snapshot_drops_pending_approvals_when_turn_aborts() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::ExecApprovalRequest(
codex_protocol::protocol::ExecApprovalRequestEvent {
call_id: "exec-call-1".to_string(),
approval_id: Some("approval-1".to_string()),
turn_id: "turn-1".to_string(),
command: vec!["echo".to_string(), "hi".to_string()],
cwd: PathBuf::from("/tmp"),
reason: None,
network_approval_context: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
additional_permissions: None,
skill_metadata: None,
available_decisions: None,
parsed_cmd: Vec::new(),
},
),
});
store.push_event(Event {
id: "ev-2".to_string(),
msg: EventMsg::ApplyPatchApprovalRequest(
codex_protocol::protocol::ApplyPatchApprovalRequestEvent {
call_id: "patch-call-1".to_string(),
turn_id: "turn-1".to_string(),
changes: HashMap::new(),
reason: None,
grant_root: None,
},
),
});
store.push_event(Event {
id: "ev-3".to_string(),
msg: EventMsg::TurnAborted(codex_protocol::protocol::TurnAbortedEvent {
turn_id: Some("turn-1".to_string()),
reason: TurnAbortReason::Replaced,
}),
});
let snapshot = store.snapshot();
assert!(snapshot.events.iter().all(|event| {
!matches!(
&event.msg,
EventMsg::ExecApprovalRequest(_) | EventMsg::ApplyPatchApprovalRequest(_)
)
}));
}
#[test]
fn thread_event_snapshot_drops_resolved_elicitation_after_outbound_resolution() {
let mut store = ThreadEventStore::new(8);
let request_id = codex_protocol::mcp::RequestId::String("request-1".to_string());
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::ElicitationRequest(codex_protocol::approvals::ElicitationRequestEvent {
turn_id: Some("turn-1".to_string()),
server_name: "server-1".to_string(),
id: request_id.clone(),
request: codex_protocol::approvals::ElicitationRequest::Form {
meta: None,
message: "Please confirm".to_string(),
requested_schema: serde_json::json!({
"type": "object",
"properties": {}
}),
},
}),
});
store.note_outbound_op(&Op::ResolveElicitation {
server_name: "server-1".to_string(),
request_id,
decision: codex_protocol::approvals::ElicitationAction::Accept,
content: None,
meta: None,
});
let snapshot = store.snapshot();
assert!(
snapshot.events.is_empty(),
"resolved elicitation prompt should not replay on thread switch"
);
}
#[test]
fn thread_event_store_reports_pending_thread_approvals() {
let mut store = ThreadEventStore::new(8);
assert_eq!(store.has_pending_thread_approvals(), false);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::ExecApprovalRequest(
codex_protocol::protocol::ExecApprovalRequestEvent {
call_id: "call-1".to_string(),
approval_id: None,
turn_id: "turn-1".to_string(),
command: vec!["echo".to_string(), "hi".to_string()],
cwd: PathBuf::from("/tmp"),
reason: None,
network_approval_context: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
additional_permissions: None,
skill_metadata: None,
available_decisions: None,
parsed_cmd: Vec::new(),
},
),
});
assert_eq!(store.has_pending_thread_approvals(), true);
store.note_outbound_op(&Op::ExecApproval {
id: "call-1".to_string(),
turn_id: Some("turn-1".to_string()),
decision: codex_protocol::protocol::ReviewDecision::Approved,
});
assert_eq!(store.has_pending_thread_approvals(), false);
}
#[test]
fn request_user_input_does_not_count_as_pending_thread_approval() {
let mut store = ThreadEventStore::new(8);
store.push_event(Event {
id: "ev-1".to_string(),
msg: EventMsg::RequestUserInput(
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
questions: Vec::new(),
},
),
});
assert_eq!(store.has_pending_thread_approvals(), false);
}
}
codex-rs/tui/src/bottom_pane/mcp_server_elicitation.rs
+44
View File
@@ -5,6 +5,9 @@ use std::path::PathBuf;
use codex_app_server_protocol::McpElicitationEnumSchema;
use codex_app_server_protocol::McpElicitationPrimitiveSchema;
use codex_app_server_protocol::McpElicitationSingleSelectEnumSchema;
use codex_app_server_protocol::McpServerElicitationRequest;
use codex_app_server_protocol::McpServerElicitationRequestParams;
use codex_app_server_protocol::RequestId as AppServerRequestId;
use codex_protocol::ThreadId;
use codex_protocol::approvals::ElicitationAction;
use codex_protocol::approvals::ElicitationRequest;
@@ -200,6 +203,47 @@ impl FooterTip {
}
impl McpServerElicitationFormRequest {
pub(crate) fn from_app_server_request(
thread_id: ThreadId,
request_id: AppServerRequestId,
params: McpServerElicitationRequestParams,
) -> Option<Self> {
let request = match params.request {
McpServerElicitationRequest::Form {
meta,
message,
requested_schema,
} => ElicitationRequest::Form {
meta,
message,
requested_schema: serde_json::to_value(requested_schema).ok()?,
},
McpServerElicitationRequest::Url {
meta,
message,
url,
elicitation_id,
} => ElicitationRequest::Url {
meta,
message,
url,
elicitation_id,
},
};
Self::from_event(
thread_id,
ElicitationRequestEvent {
turn_id: params.turn_id,
server_name: params.server_name,
id: match request_id {
AppServerRequestId::String(value) => McpRequestId::String(value),
AppServerRequestId::Integer(value) => McpRequestId::Integer(value),
},
request,
},
)
}
pub(crate) fn from_event(
thread_id: ThreadId,
request: ElicitationRequestEvent,
codex-rs/tui/src/chatwidget.rs
+763
View File
@@ -49,6 +49,7 @@ use crate::status::format_directory_display;
use crate::status::format_tokens_compact;
use crate::status::rate_limit_snapshot_display_for_limit;
use crate::text_formatting::proper_join;
use crate::thread_update::ThreadUpdate;
use crate::version::CODEX_CLI_VERSION;
use codex_app_server_protocol::ConfigLayerSource;
use codex_chatgpt::connectors;
@@ -77,6 +78,7 @@ use codex_otel::RuntimeMetricsSummary;
use codex_otel::SessionTelemetry;
use codex_protocol::ThreadId;
use codex_protocol::account::PlanType;
use codex_protocol::approvals::ElicitationRequest;
use codex_protocol::approvals::ElicitationRequestEvent;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::CollaborationModeMask;
@@ -4844,6 +4846,436 @@ impl ChatWidget {
self.dispatch_event_msg(None, msg, Some(ReplayKind::ThreadSnapshot));
}
pub(crate) fn handle_thread_update(&mut self, update: ThreadUpdate) {
self.dispatch_thread_update(update, None);
}
pub(crate) fn handle_thread_update_replay(&mut self, update: ThreadUpdate) {
self.dispatch_thread_update(update, Some(ReplayKind::ThreadSnapshot));
}
fn dispatch_thread_update(&mut self, update: ThreadUpdate, replay_kind: Option<ReplayKind>) {
let from_replay = replay_kind.is_some();
let is_resume_initial_replay =
matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages));
if !is_resume_initial_replay {
self.restore_retry_status_header_if_present();
}
match update {
ThreadUpdate::SessionConfigured(event) => self.on_session_configured(event),
ThreadUpdate::ThreadNameUpdated(event) => {
match ThreadId::from_string(&event.thread_id) {
Ok(thread_id) => self.on_thread_name_updated(
codex_protocol::protocol::ThreadNameUpdatedEvent {
thread_id,
thread_name: event.thread_name,
},
),
Err(err) => tracing::warn!(
thread_id = %event.thread_id,
"ignoring thread/nameUpdated notification with invalid thread id: {err}"
),
}
}
ThreadUpdate::ThreadTokenUsageUpdated(event) => {
let info = Some(TokenUsageInfo {
total_token_usage: token_usage_from_api(&event.token_usage.total),
last_token_usage: token_usage_from_api(&event.token_usage.last),
model_context_window: event.token_usage.model_context_window,
});
self.set_token_info(info);
}
ThreadUpdate::TurnStarted(_event) => {
if !is_resume_initial_replay {
self.on_task_started();
self.apply_turn_started_context_window(None);
}
}
ThreadUpdate::TurnCompleted(event) => {
let last_agent_message =
event.turn.items.iter().rev().find_map(|item| match item {
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } => {
Some(text.clone())
}
_ => None,
});
match event.turn.status {
codex_app_server_protocol::TurnStatus::Completed
| codex_app_server_protocol::TurnStatus::Failed => {
self.on_task_complete(last_agent_message, from_replay);
if let Some(error) = event.turn.error {
self.on_error(error.message);
}
}
codex_app_server_protocol::TurnStatus::Interrupted => {
self.on_interrupted_turn(TurnAbortReason::Interrupted);
}
codex_app_server_protocol::TurnStatus::InProgress => {}
}
}
ThreadUpdate::TurnDiffUpdated(event) => self.on_turn_diff(event.diff),
ThreadUpdate::TurnPlanUpdated(event) => {
self.on_plan_update(update_plan_args_from_api(event.explanation, event.plan))
}
ThreadUpdate::AgentMessageDelta(event) => self.on_agent_message_delta(event.delta),
ThreadUpdate::PlanDelta(event) => self.on_plan_delta(event.delta),
ThreadUpdate::ReasoningSummaryTextDelta(event) => {
self.on_agent_reasoning_delta(event.delta);
}
ThreadUpdate::ReasoningSummaryPartAdded(_) => self.on_reasoning_section_break(),
ThreadUpdate::ReasoningTextDelta(event) => self.on_agent_reasoning_delta(event.delta),
ThreadUpdate::TerminalInteraction(event) => {
self.on_terminal_interaction(codex_protocol::protocol::TerminalInteractionEvent {
call_id: event.item_id,
process_id: event.process_id,
stdin: event.stdin,
});
}
ThreadUpdate::CommandExecutionOutputDelta(event) => {
self.on_exec_command_output_delta(
codex_protocol::protocol::ExecCommandOutputDeltaEvent {
call_id: event.item_id,
stream: codex_protocol::protocol::ExecOutputStream::Stdout,
chunk: event.delta.into_bytes(),
},
);
}
ThreadUpdate::FileChangeOutputDelta(_) => {}
ThreadUpdate::McpToolCallProgress(_) => {}
ThreadUpdate::ItemStarted(event) => {
self.handle_thread_item_started(event.item);
}
ThreadUpdate::ItemCompleted(event) => {
self.handle_thread_item_completed(event.item, from_replay);
}
ThreadUpdate::ItemGuardianApprovalReviewStarted(event) => {
self.on_guardian_assessment(codex_protocol::approvals::GuardianAssessmentEvent {
id: event.target_item_id,
turn_id: event.turn_id,
status: guardian_status_from_api(event.review.status),
risk_score: event.review.risk_score,
risk_level: event.review.risk_level.map(guardian_risk_level_from_api),
rationale: event.review.rationale,
action: event.action,
});
}
ThreadUpdate::ItemGuardianApprovalReviewCompleted(event) => {
self.on_guardian_assessment(codex_protocol::approvals::GuardianAssessmentEvent {
id: event.target_item_id,
turn_id: event.turn_id,
status: guardian_status_from_api(event.review.status),
risk_score: event.review.risk_score,
risk_level: event.review.risk_level.map(guardian_risk_level_from_api),
rationale: event.review.rationale,
action: event.action,
});
}
ThreadUpdate::CommandExecutionRequestApproval { params, .. } => {
self.on_exec_approval_request(String::new(), exec_approval_event_from_api(params));
}
ThreadUpdate::FileChangeRequestApproval { params, .. } => {
self.on_apply_patch_approval_request(
String::new(),
patch_approval_event_from_api(params),
);
}
ThreadUpdate::McpServerElicitationRequest { request_id, params } => {
self.on_elicitation_request(elicitation_event_from_api(request_id, params));
}
ThreadUpdate::PermissionsRequestApproval { params, .. } => {
self.on_request_permissions(request_permissions_event_from_api(params));
}
ThreadUpdate::ToolRequestUserInput { params, .. } => {
self.on_request_user_input(request_user_input_event_from_api(params));
}
ThreadUpdate::Error(event) => self.on_error(event.error.message),
ThreadUpdate::ModelRerouted(_) => {}
ThreadUpdate::DeprecationNotice(event) => {
self.on_deprecation_notice(codex_protocol::protocol::DeprecationNoticeEvent {
summary: event.summary,
details: event.details,
});
}
ThreadUpdate::HookStarted(event) => {
self.on_hook_started(codex_protocol::protocol::HookStartedEvent {
turn_id: event.turn_id,
run: hook_run_summary_from_api(event.run),
});
}
ThreadUpdate::HookCompleted(event) => {
self.on_hook_completed(codex_protocol::protocol::HookCompletedEvent {
turn_id: event.turn_id,
run: hook_run_summary_from_api(event.run),
});
}
ThreadUpdate::ThreadRealtimeStarted(event) => {
if !from_replay {
self.on_realtime_conversation_started(
codex_protocol::protocol::RealtimeConversationStartedEvent {
session_id: event.session_id,
},
);
}
}
ThreadUpdate::ThreadRealtimeItemAdded(_) => {}
ThreadUpdate::ThreadRealtimeOutputAudioDelta(_) => {}
ThreadUpdate::ThreadRealtimeError(event) => {
if !from_replay {
self.on_stream_error(event.message, None);
}
}
ThreadUpdate::ThreadRealtimeClosed(event) => {
if !from_replay {
self.on_realtime_conversation_closed(
codex_protocol::protocol::RealtimeConversationClosedEvent {
reason: event.reason,
},
);
}
}
ThreadUpdate::DynamicToolCall { .. } => {}
ThreadUpdate::ThreadStarted(_)
| ThreadUpdate::ThreadStatusChanged(_)
| ThreadUpdate::ThreadClosed(_) => {}
ThreadUpdate::ThreadRolledBack { num_turns } => {
self.last_copyable_output = None;
if from_replay {
self.app_event_tx
.send(AppEvent::ApplyThreadRollback { num_turns });
}
}
}
if !from_replay && self.agent_turn_running {
self.refresh_runtime_metrics();
}
}
fn handle_thread_item_started(&mut self, item: codex_app_server_protocol::ThreadItem) {
match item {
codex_app_server_protocol::ThreadItem::CommandExecution {
id,
command,
cwd,
process_id,
..
} => {
let command = vec![command];
let parsed_cmd = codex_shell_command::parse_command::parse_command(&command);
self.on_exec_command_begin(ExecCommandBeginEvent {
call_id: id,
process_id,
turn_id: String::new(),
command,
cwd,
parsed_cmd,
source: ExecCommandSource::Agent,
interaction_input: None,
});
}
codex_app_server_protocol::ThreadItem::FileChange { id, changes, .. } => {
self.on_patch_apply_begin(PatchApplyBeginEvent {
call_id: id,
turn_id: String::new(),
auto_approved: false,
changes: file_changes_from_api(changes),
});
}
codex_app_server_protocol::ThreadItem::McpToolCall {
id,
server,
tool,
arguments,
..
} => {
self.on_mcp_tool_call_begin(McpToolCallBeginEvent {
call_id: id,
invocation: codex_protocol::protocol::McpInvocation {
server,
tool,
arguments: Some(arguments),
},
});
}
codex_app_server_protocol::ThreadItem::WebSearch { id, .. } => {
self.on_web_search_begin(codex_protocol::protocol::WebSearchBeginEvent {
call_id: id,
});
}
codex_app_server_protocol::ThreadItem::ImageView { id, path } => {
self.on_view_image_tool_call(codex_protocol::protocol::ViewImageToolCallEvent {
call_id: id,
path: PathBuf::from(path),
});
}
codex_app_server_protocol::ThreadItem::ImageGeneration { id, .. } => {
self.on_image_generation_begin(ImageGenerationBeginEvent { call_id: id });
}
_ => {}
}
}
fn handle_thread_item_completed(
&mut self,
item: codex_app_server_protocol::ThreadItem,
from_replay: bool,
) {
match item {
codex_app_server_protocol::ThreadItem::UserMessage { content, .. } => {
let event = user_message_event_from_api(content);
if from_replay || self.should_render_realtime_user_message_event(&event) {
self.on_user_message_event(event);
}
}
codex_app_server_protocol::ThreadItem::AgentMessage { id, text, phase } => {
if from_replay || self.is_review_mode {
self.on_agent_message(text.clone());
}
self.on_agent_message_item_completed(AgentMessageItem {
id,
content: vec![AgentMessageContent::Text { text }],
phase,
});
}
codex_app_server_protocol::ThreadItem::Plan { text, .. } => {
self.on_plan_item_completed(text);
}
codex_app_server_protocol::ThreadItem::Reasoning { content, .. } => {
for part in content {
self.on_agent_reasoning_delta(part);
}
self.on_agent_reasoning_final();
}
codex_app_server_protocol::ThreadItem::CommandExecution {
id,
command,
cwd,
process_id,
aggregated_output,
exit_code,
duration_ms,
status,
..
} => {
let command = vec![command];
let parsed_cmd = codex_shell_command::parse_command::parse_command(&command);
self.on_exec_command_end(ExecCommandEndEvent {
call_id: id,
process_id,
turn_id: String::new(),
command,
cwd,
parsed_cmd,
source: ExecCommandSource::Agent,
interaction_input: None,
stdout: aggregated_output.clone().unwrap_or_default(),
stderr: String::new(),
aggregated_output: aggregated_output.unwrap_or_default(),
exit_code: exit_code.unwrap_or_default(),
duration: std::time::Duration::from_millis(
duration_ms.unwrap_or_default().max(0) as u64,
),
formatted_output: String::new(),
status: exec_status_from_api(status),
});
}
codex_app_server_protocol::ThreadItem::FileChange {
id,
status,
changes,
..
} => {
self.on_patch_apply_end(codex_protocol::protocol::PatchApplyEndEvent {
call_id: id,
turn_id: String::new(),
stdout: String::new(),
stderr: String::new(),
success: matches!(
status,
codex_app_server_protocol::PatchApplyStatus::Completed
),
changes: file_changes_from_api(changes),
status: patch_apply_status_from_api(status),
});
}
codex_app_server_protocol::ThreadItem::McpToolCall {
id,
server,
tool,
arguments,
result,
error,
duration_ms,
..
} => {
self.on_mcp_tool_call_end(McpToolCallEndEvent {
call_id: id,
invocation: codex_protocol::protocol::McpInvocation {
server,
tool,
arguments: Some(arguments),
},
duration: std::time::Duration::from_millis(
duration_ms.unwrap_or_default().max(0) as u64,
),
result: match result {
Some(result) => Ok(mcp_result_from_api(result)),
None => Err(error.map(|err| err.message).unwrap_or_default()),
},
});
}
codex_app_server_protocol::ThreadItem::WebSearch { id, query, action } => {
self.on_web_search_end(codex_protocol::protocol::WebSearchEndEvent {
call_id: id,
query,
action: web_search_action_from_api(action),
});
}
codex_app_server_protocol::ThreadItem::ImageGeneration {
id,
status,
revised_prompt,
result,
} => {
self.on_image_generation_end(ImageGenerationEndEvent {
call_id: id,
status,
revised_prompt,
result,
saved_path: None,
});
}
codex_app_server_protocol::ThreadItem::EnteredReviewMode { review, .. } => {
self.on_entered_review_mode(
ReviewRequest {
target: ReviewTarget::Custom {
instructions: review.clone(),
},
user_facing_hint: Some(review),
},
from_replay,
);
}
codex_app_server_protocol::ThreadItem::ExitedReviewMode { review, .. } => {
self.on_exited_review_mode(ExitedReviewModeEvent {
review_output: Some(codex_protocol::protocol::ReviewOutputEvent {
findings: Vec::new(),
overall_correctness: String::new(),
overall_explanation: review,
overall_confidence_score: 0.0,
}),
});
}
codex_app_server_protocol::ThreadItem::ContextCompaction { .. } => {
self.on_agent_message("Context compacted".to_owned());
}
codex_app_server_protocol::ThreadItem::ImageView { .. }
| codex_app_server_protocol::ThreadItem::DynamicToolCall { .. }
| codex_app_server_protocol::ThreadItem::CollabAgentToolCall { .. } => {}
}
}
/// Dispatch a protocol `EventMsg` to the appropriate handler.
///
/// `id` is `Some` for live events and `None` for replayed events from
@@ -8984,6 +9416,337 @@ fn hook_event_label(event_name: codex_protocol::protocol::HookEventName) -> &'st
}
}
fn token_usage_from_api(
usage: &codex_app_server_protocol::TokenUsageBreakdown,
) -> codex_protocol::protocol::TokenUsage {
codex_protocol::protocol::TokenUsage {
total_tokens: usage.total_tokens,
input_tokens: usage.input_tokens,
cached_input_tokens: usage.cached_input_tokens,
output_tokens: usage.output_tokens,
reasoning_output_tokens: usage.reasoning_output_tokens,
}
}
fn update_plan_args_from_api(
explanation: Option<String>,
plan: Vec<codex_app_server_protocol::TurnPlanStep>,
) -> codex_protocol::plan_tool::UpdatePlanArgs {
codex_protocol::plan_tool::UpdatePlanArgs {
explanation,
plan: plan
.into_iter()
.map(|step| codex_protocol::plan_tool::PlanItemArg {
step: step.step,
status: match step.status {
codex_app_server_protocol::TurnPlanStepStatus::Pending => {
codex_protocol::plan_tool::StepStatus::Pending
}
codex_app_server_protocol::TurnPlanStepStatus::InProgress => {
codex_protocol::plan_tool::StepStatus::InProgress
}
codex_app_server_protocol::TurnPlanStepStatus::Completed => {
codex_protocol::plan_tool::StepStatus::Completed
}
},
})
.collect(),
}
}
fn guardian_status_from_api(
status: codex_app_server_protocol::GuardianApprovalReviewStatus,
) -> GuardianAssessmentStatus {
match status {
codex_app_server_protocol::GuardianApprovalReviewStatus::InProgress => {
GuardianAssessmentStatus::InProgress
}
codex_app_server_protocol::GuardianApprovalReviewStatus::Approved => {
GuardianAssessmentStatus::Approved
}
codex_app_server_protocol::GuardianApprovalReviewStatus::Denied => {
GuardianAssessmentStatus::Denied
}
codex_app_server_protocol::GuardianApprovalReviewStatus::Aborted => {
GuardianAssessmentStatus::Aborted
}
}
}
fn guardian_risk_level_from_api(
level: codex_app_server_protocol::GuardianRiskLevel,
) -> codex_protocol::approvals::GuardianRiskLevel {
match level {
codex_app_server_protocol::GuardianRiskLevel::Low => {
codex_protocol::approvals::GuardianRiskLevel::Low
}
codex_app_server_protocol::GuardianRiskLevel::Medium => {
codex_protocol::approvals::GuardianRiskLevel::Medium
}
codex_app_server_protocol::GuardianRiskLevel::High => {
codex_protocol::approvals::GuardianRiskLevel::High
}
}
}
fn exec_approval_event_from_api(
params: codex_app_server_protocol::CommandExecutionRequestApprovalParams,
) -> ExecApprovalRequestEvent {
let command = params
.command
.map(|command| vec![command])
.unwrap_or_default();
ExecApprovalRequestEvent {
call_id: params.item_id,
approval_id: params.approval_id,
turn_id: params.turn_id,
command: command.clone(),
cwd: params.cwd.unwrap_or_default(),
reason: params.reason,
network_approval_context: params
.network_approval_context
.and_then(|context| serde_json::from_value(serde_json::to_value(context).ok()?).ok()),
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
additional_permissions: params.additional_permissions.and_then(|permissions| {
serde_json::from_value(serde_json::to_value(permissions).ok()?).ok()
}),
skill_metadata: None,
available_decisions: None,
parsed_cmd: codex_shell_command::parse_command::parse_command(&command),
}
}
fn patch_approval_event_from_api(
params: codex_app_server_protocol::FileChangeRequestApprovalParams,
) -> ApplyPatchApprovalRequestEvent {
ApplyPatchApprovalRequestEvent {
call_id: params.item_id,
turn_id: params.turn_id,
changes: HashMap::new(),
reason: params.reason,
grant_root: params.grant_root,
}
}
fn elicitation_event_from_api(
request_id: codex_app_server_protocol::RequestId,
params: codex_app_server_protocol::McpServerElicitationRequestParams,
) -> ElicitationRequestEvent {
ElicitationRequestEvent {
turn_id: params.turn_id,
server_name: params.server_name,
id: match request_id {
codex_app_server_protocol::RequestId::String(value) => {
codex_protocol::mcp::RequestId::String(value)
}
codex_app_server_protocol::RequestId::Integer(value) => {
codex_protocol::mcp::RequestId::Integer(value)
}
},
request: match params.request {
codex_app_server_protocol::McpServerElicitationRequest::Form {
meta,
message,
requested_schema,
} => ElicitationRequest::Form {
meta,
message,
requested_schema: serde_json::to_value(requested_schema)
.unwrap_or(serde_json::Value::Null),
},
codex_app_server_protocol::McpServerElicitationRequest::Url {
meta,
message,
url,
elicitation_id,
} => ElicitationRequest::Url {
meta,
message,
url,
elicitation_id,
},
},
}
}
fn request_permissions_event_from_api(
params: codex_app_server_protocol::PermissionsRequestApprovalParams,
) -> codex_protocol::request_permissions::RequestPermissionsEvent {
codex_protocol::request_permissions::RequestPermissionsEvent {
call_id: params.item_id,
turn_id: params.turn_id,
reason: params.reason,
permissions: serde_json::from_value(
serde_json::to_value(params.permissions).unwrap_or(serde_json::Value::Null),
)
.unwrap_or_default(),
}
}
fn request_user_input_event_from_api(
params: codex_app_server_protocol::ToolRequestUserInputParams,
) -> codex_protocol::request_user_input::RequestUserInputEvent {
codex_protocol::request_user_input::RequestUserInputEvent {
call_id: params.item_id,
turn_id: params.turn_id,
questions: params
.questions
.into_iter()
.map(
|question| codex_protocol::request_user_input::RequestUserInputQuestion {
id: question.id,
header: question.header,
question: question.question,
is_other: question.is_other,
is_secret: question.is_secret,
options: question.options.map(|options| {
options
.into_iter()
.map(|option| {
codex_protocol::request_user_input::RequestUserInputQuestionOption {
label: option.label,
description: option.description,
}
})
.collect()
}),
},
)
.collect(),
}
}
fn file_changes_from_api(
changes: Vec<codex_app_server_protocol::FileUpdateChange>,
) -> HashMap<PathBuf, codex_protocol::protocol::FileChange> {
changes
.into_iter()
.map(|change| {
let file_change = match change.kind {
codex_app_server_protocol::PatchChangeKind::Add => {
codex_protocol::protocol::FileChange::Add {
content: change.diff,
}
}
codex_app_server_protocol::PatchChangeKind::Delete => {
codex_protocol::protocol::FileChange::Delete {
content: change.diff,
}
}
codex_app_server_protocol::PatchChangeKind::Update { move_path } => {
codex_protocol::protocol::FileChange::Update {
unified_diff: change.diff,
move_path,
}
}
};
(PathBuf::from(change.path), file_change)
})
.collect()
}
fn user_message_event_from_api(
content: Vec<codex_app_server_protocol::UserInput>,
) -> codex_protocol::protocol::UserMessageEvent {
let mut message = String::new();
let mut images = Vec::new();
let mut local_images = Vec::new();
let mut text_elements = Vec::new();
for item in content {
match item {
codex_app_server_protocol::UserInput::Text {
text,
text_elements: api_elements,
} => {
message.push_str(&text);
text_elements.extend(api_elements.into_iter().map(Into::into));
}
codex_app_server_protocol::UserInput::Image { url } => images.push(url),
codex_app_server_protocol::UserInput::LocalImage { path } => local_images.push(path),
codex_app_server_protocol::UserInput::Skill { .. }
| codex_app_server_protocol::UserInput::Mention { .. } => {}
}
}
codex_protocol::protocol::UserMessageEvent {
message,
images: Some(images),
local_images,
text_elements,
}
}
fn exec_status_from_api(
status: codex_app_server_protocol::CommandExecutionStatus,
) -> codex_protocol::protocol::ExecCommandStatus {
match status {
codex_app_server_protocol::CommandExecutionStatus::InProgress
| codex_app_server_protocol::CommandExecutionStatus::Completed => {
codex_protocol::protocol::ExecCommandStatus::Completed
}
codex_app_server_protocol::CommandExecutionStatus::Failed => {
codex_protocol::protocol::ExecCommandStatus::Failed
}
codex_app_server_protocol::CommandExecutionStatus::Declined => {
codex_protocol::protocol::ExecCommandStatus::Declined
}
}
}
fn patch_apply_status_from_api(
status: codex_app_server_protocol::PatchApplyStatus,
) -> codex_protocol::protocol::PatchApplyStatus {
match status {
codex_app_server_protocol::PatchApplyStatus::InProgress
| codex_app_server_protocol::PatchApplyStatus::Completed => {
codex_protocol::protocol::PatchApplyStatus::Completed
}
codex_app_server_protocol::PatchApplyStatus::Failed => {
codex_protocol::protocol::PatchApplyStatus::Failed
}
codex_app_server_protocol::PatchApplyStatus::Declined => {
codex_protocol::protocol::PatchApplyStatus::Declined
}
}
}
fn hook_run_summary_from_api(
run: codex_app_server_protocol::HookRunSummary,
) -> codex_protocol::protocol::HookRunSummary {
serde_json::from_value(serde_json::to_value(run).unwrap_or(serde_json::Value::Null))
.unwrap_or_else(|err| panic!("failed to convert app-server hook run summary: {err}"))
}
fn mcp_result_from_api(
result: codex_app_server_protocol::McpToolCallResult,
) -> codex_protocol::mcp::CallToolResult {
codex_protocol::mcp::CallToolResult {
content: result.content,
structured_content: result.structured_content,
is_error: None,
meta: None,
}
}
fn web_search_action_from_api(
action: Option<codex_app_server_protocol::WebSearchAction>,
) -> codex_protocol::models::WebSearchAction {
match action.unwrap_or(codex_app_server_protocol::WebSearchAction::Other) {
codex_app_server_protocol::WebSearchAction::Search { query, queries } => {
codex_protocol::models::WebSearchAction::Search { query, queries }
}
codex_app_server_protocol::WebSearchAction::OpenPage { url } => {
codex_protocol::models::WebSearchAction::OpenPage { url }
}
codex_app_server_protocol::WebSearchAction::FindInPage { url, pattern } => {
codex_protocol::models::WebSearchAction::FindInPage { url, pattern }
}
codex_app_server_protocol::WebSearchAction::Other => {
codex_protocol::models::WebSearchAction::Other
}
}
}
#[cfg(test)]
pub(crate) fn show_review_commit_picker_with_entries(
chat: &mut ChatWidget,
codex-rs/tui/src/lib.rs
+1
View File
@@ -119,6 +119,7 @@ mod style;
mod terminal_palette;
mod text_formatting;
mod theme_picker;
mod thread_update;
mod tooltips;
mod tui;
mod ui_consts;
codex-rs/tui/src/thread_update.rs
+170
View File
@@ -0,0 +1,170 @@
use codex_app_server_protocol::AgentMessageDeltaNotification;
use codex_app_server_protocol::CommandExecutionOutputDeltaNotification;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::DeprecationNoticeNotification;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::ErrorNotification;
use codex_app_server_protocol::FileChangeOutputDeltaNotification;
use codex_app_server_protocol::FileChangeRequestApprovalParams;
use codex_app_server_protocol::HookCompletedNotification;
use codex_app_server_protocol::HookStartedNotification;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification;
use codex_app_server_protocol::ItemGuardianApprovalReviewStartedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::McpServerElicitationRequestParams;
use codex_app_server_protocol::McpToolCallProgressNotification;
use codex_app_server_protocol::ModelReroutedNotification;
use codex_app_server_protocol::PermissionsRequestApprovalParams;
use codex_app_server_protocol::PlanDeltaNotification;
use codex_app_server_protocol::ReasoningSummaryPartAddedNotification;
use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
use codex_app_server_protocol::ReasoningTextDeltaNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::TerminalInteractionNotification;
use codex_app_server_protocol::ThreadClosedNotification;
use codex_app_server_protocol::ThreadNameUpdatedNotification;
use codex_app_server_protocol::ThreadRealtimeClosedNotification;
use codex_app_server_protocol::ThreadRealtimeErrorNotification;
use codex_app_server_protocol::ThreadRealtimeItemAddedNotification;
use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification;
use codex_app_server_protocol::ThreadRealtimeStartedNotification;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;
use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnDiffUpdatedNotification;
use codex_app_server_protocol::TurnPlanUpdatedNotification;
use codex_app_server_protocol::TurnStartedNotification;
use codex_protocol::protocol::SessionConfiguredEvent;
#[derive(Debug, Clone)]
pub(crate) enum ThreadUpdate {
SessionConfigured(SessionConfiguredEvent),
ThreadStarted(ThreadStartedNotification),
ThreadStatusChanged(ThreadStatusChangedNotification),
ThreadClosed(ThreadClosedNotification),
ThreadNameUpdated(ThreadNameUpdatedNotification),
ThreadTokenUsageUpdated(ThreadTokenUsageUpdatedNotification),
TurnStarted(TurnStartedNotification),
TurnCompleted(TurnCompletedNotification),
TurnDiffUpdated(TurnDiffUpdatedNotification),
TurnPlanUpdated(TurnPlanUpdatedNotification),
ItemStarted(ItemStartedNotification),
ItemGuardianApprovalReviewStarted(ItemGuardianApprovalReviewStartedNotification),
ItemGuardianApprovalReviewCompleted(ItemGuardianApprovalReviewCompletedNotification),
ItemCompleted(ItemCompletedNotification),
AgentMessageDelta(AgentMessageDeltaNotification),
PlanDelta(PlanDeltaNotification),
ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification),
ReasoningSummaryPartAdded(ReasoningSummaryPartAddedNotification),
ReasoningTextDelta(ReasoningTextDeltaNotification),
TerminalInteraction(TerminalInteractionNotification),
CommandExecutionOutputDelta(CommandExecutionOutputDeltaNotification),
FileChangeOutputDelta(FileChangeOutputDeltaNotification),
McpToolCallProgress(McpToolCallProgressNotification),
HookStarted(HookStartedNotification),
HookCompleted(HookCompletedNotification),
Error(ErrorNotification),
ModelRerouted(ModelReroutedNotification),
DeprecationNotice(DeprecationNoticeNotification),
ThreadRealtimeStarted(ThreadRealtimeStartedNotification),
ThreadRealtimeItemAdded(ThreadRealtimeItemAddedNotification),
ThreadRealtimeOutputAudioDelta(ThreadRealtimeOutputAudioDeltaNotification),
ThreadRealtimeError(ThreadRealtimeErrorNotification),
ThreadRealtimeClosed(ThreadRealtimeClosedNotification),
CommandExecutionRequestApproval {
request_id: RequestId,
params: CommandExecutionRequestApprovalParams,
},
FileChangeRequestApproval {
request_id: RequestId,
params: FileChangeRequestApprovalParams,
},
McpServerElicitationRequest {
request_id: RequestId,
params: McpServerElicitationRequestParams,
},
PermissionsRequestApproval {
request_id: RequestId,
params: PermissionsRequestApprovalParams,
},
ToolRequestUserInput {
request_id: RequestId,
params: ToolRequestUserInputParams,
},
DynamicToolCall {
request_id: RequestId,
params: DynamicToolCallParams,
},
ThreadRolledBack {
num_turns: u32,
},
}
impl ThreadUpdate {
pub(crate) fn thread_id(&self) -> Option<String> {
match self {
Self::SessionConfigured(session) => Some(session.session_id.to_string()),
Self::ThreadStarted(notification) => Some(notification.thread.id.clone()),
Self::ThreadStatusChanged(notification) => Some(notification.thread_id.clone()),
Self::ThreadClosed(notification) => Some(notification.thread_id.clone()),
Self::ThreadNameUpdated(notification) => Some(notification.thread_id.clone()),
Self::ThreadTokenUsageUpdated(notification) => Some(notification.thread_id.clone()),
Self::TurnStarted(notification) => Some(notification.thread_id.clone()),
Self::TurnCompleted(notification) => Some(notification.thread_id.clone()),
Self::TurnDiffUpdated(notification) => Some(notification.thread_id.clone()),
Self::TurnPlanUpdated(notification) => Some(notification.thread_id.clone()),
Self::ItemStarted(notification) => Some(notification.thread_id.clone()),
Self::ItemGuardianApprovalReviewStarted(notification) => {
Some(notification.thread_id.clone())
}
Self::ItemGuardianApprovalReviewCompleted(notification) => {
Some(notification.thread_id.clone())
}
Self::ItemCompleted(notification) => Some(notification.thread_id.clone()),
Self::AgentMessageDelta(notification) => Some(notification.thread_id.clone()),
Self::PlanDelta(notification) => Some(notification.thread_id.clone()),
Self::ReasoningSummaryTextDelta(notification) => Some(notification.thread_id.clone()),
Self::ReasoningSummaryPartAdded(notification) => Some(notification.thread_id.clone()),
Self::ReasoningTextDelta(notification) => Some(notification.thread_id.clone()),
Self::TerminalInteraction(notification) => Some(notification.thread_id.clone()),
Self::CommandExecutionOutputDelta(notification) => Some(notification.thread_id.clone()),
Self::FileChangeOutputDelta(notification) => Some(notification.thread_id.clone()),
Self::McpToolCallProgress(notification) => Some(notification.thread_id.clone()),
Self::HookStarted(notification) => Some(notification.thread_id.clone()),
Self::HookCompleted(notification) => Some(notification.thread_id.clone()),
Self::Error(notification) => Some(notification.thread_id.clone()),
Self::ModelRerouted(notification) => Some(notification.thread_id.clone()),
Self::ThreadRealtimeStarted(notification) => Some(notification.thread_id.clone()),
Self::ThreadRealtimeItemAdded(notification) => Some(notification.thread_id.clone()),
Self::ThreadRealtimeOutputAudioDelta(notification) => {
Some(notification.thread_id.clone())
}
Self::ThreadRealtimeError(notification) => Some(notification.thread_id.clone()),
Self::ThreadRealtimeClosed(notification) => Some(notification.thread_id.clone()),
Self::CommandExecutionRequestApproval { params, .. } => Some(params.thread_id.clone()),
Self::FileChangeRequestApproval { params, .. } => Some(params.thread_id.clone()),
Self::McpServerElicitationRequest { params, .. } => Some(params.thread_id.clone()),
Self::PermissionsRequestApproval { params, .. } => Some(params.thread_id.clone()),
Self::ToolRequestUserInput { params, .. } => Some(params.thread_id.clone()),
Self::DynamicToolCall { params, .. } => Some(params.thread_id.clone()),
Self::DeprecationNotice(_) | Self::ThreadRolledBack { .. } => None,
}
}
pub(crate) fn is_status_refresh_update(&self) -> bool {
matches!(
self,
Self::SessionConfigured(_)
| Self::TurnStarted(_)
| Self::ThreadTokenUsageUpdated(_)
| Self::TurnCompleted(_)
)
}
pub(crate) fn is_thread_closed(&self) -> bool {
matches!(self, Self::ThreadClosed(_))
}
}