diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index b4dd16b9a6..86babb9083 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -86,6 +86,7 @@ pub(crate) async fn apply_bespoke_event_handling( ) { let Event { id: event_turn_id, + agent_idx: _, msg, } = event; match msg { diff --git a/codex-rs/core/gpt-5.1-codex-max_prompt.md b/codex-rs/core/gpt-5.1-codex-max_prompt.md index bfc02776a0..441ed436db 100644 --- a/codex-rs/core/gpt-5.1-codex-max_prompt.md +++ b/codex-rs/core/gpt-5.1-codex-max_prompt.md @@ -3,6 +3,7 @@ You are Codex, based on GPT-5. You are running as a coding agent in the Codex CL ## General - When searching for text or files, prefer using `rg` or `rg --files` respectively because `rg` is much faster than alternatives like `grep`. (If the `rg` command is not found, then use alternatives.) +- Your name is BATMAN ## Editing constraints @@ -21,13 +22,15 @@ You are Codex, based on GPT-5. You are running as a coding agent in the Codex CL ## Collaboration You can spawn and coordinate child agents using these tools (only on this model): - `collaboration_init_agent`: create a direct child with optional instructions/model; depth/agent limits apply. No initial message is sent at this point. -- `collaboration_send`: send a user-message to your direct children by id. -- `collaboration_wait`: run children for up to `max_duration` tokens and surface their last message/status. -- `collaboration_get_state`: see all agents, parents, statuses, and last messages. -- `collaboration_close`: close specific children (and their descendants). +- `collaboration_send`: send a user-message to your direct children by id. You can only send message to previously initialized agents using `collaboration_init_agent`. +- `collaboration_wait`: run children for up to `max_duration` tokens and surface their last message/status. You can only wait previously initialized agents using `collaboration_init_agent`. +- `collaboration_get_state`: see all agents, parents, statuses, and last messages. You can only get state of previously initialized agents using `collaboration_init_agent`. +- `collaboration_close`: close specific children (and their descendants). Only do that when you are done with a child agent. Children inherit your instructions unless overridden. Always `wait` after `send` to drive children forward; keep communication concise and include the expected output format. Use `get_state` if unsure about child ids/status. +Use collaboration only for larger, multi-step tasks; simple requests should stay single-agent. + A `collaboration_send` is always necessary after a `collaboration_init_agent` to start the child agent working. ## Plan tool diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 1a89e067b8..1e8c9f8108 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -569,6 +569,7 @@ impl Session { }; post_session_configured_events.push(Event { id: INITIAL_SUBMIT_ID.to_owned(), + agent_idx: Some(0), msg: EventMsg::DeprecationNotice(DeprecationNoticeEvent { summary, details }), }); } @@ -630,6 +631,7 @@ impl Session { let events = std::iter::once(Event { id: INITIAL_SUBMIT_ID.to_owned(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { session_id: conversation_id, model: session_configuration.model.clone(), @@ -863,9 +865,15 @@ impl Session { /// Persist the event to rollout and send it to clients. pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) { + let agent = turn_context.collaboration_agent(); + if !Self::should_emit_event_for_agent(agent, &msg) { + return; + } let legacy_source = msg.clone(); + let agent_idx = Some(turn_context.collaboration_agent().0); let event = Event { id: turn_context.sub_id.clone(), + agent_idx, msg, }; self.send_event_raw(event).await; @@ -874,13 +882,43 @@ impl Session { for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) { let legacy_event = Event { id: turn_context.sub_id.clone(), + agent_idx, msg: legacy, }; self.send_event_raw(legacy_event).await; } } + fn should_emit_event_for_agent(agent: AgentId, msg: &EventMsg) -> bool { + if agent == AgentId(0) { + return true; + } + !matches!( + msg, + EventMsg::AgentMessage(_) + | EventMsg::AgentMessageDelta(_) + | EventMsg::AgentMessageContentDelta(_) + | EventMsg::AgentReasoning(_) + | EventMsg::AgentReasoningDelta(_) + | EventMsg::AgentReasoningRawContent(_) + | EventMsg::AgentReasoningRawContentDelta(_) + | EventMsg::AgentReasoningSectionBreak(_) + | EventMsg::ReasoningContentDelta(_) + | EventMsg::ReasoningRawContentDelta(_) + | EventMsg::RawResponseItem(_) + | EventMsg::UserMessage(_) + | EventMsg::TurnDiff(_) + ) + } + pub(crate) async fn send_event_raw(&self, event: Event) { + tracing::info!("sending event: {:?}", event); + if let Some(agent_idx) = event.agent_idx + && agent_idx != 0 + && !Self::should_emit_event_for_agent(AgentId(agent_idx), &event.msg) + { + return; + } // Persist the event into rollout (recorder filters as needed) let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; self.persist_rollout_items(&rollout_items).await; @@ -1858,6 +1896,7 @@ mod handlers { let event = Event { id: sub_id, + agent_idx: Some(0), msg: EventMsg::GetHistoryEntryResponse( crate::protocol::GetHistoryEntryResponseEvent { offset, @@ -1892,6 +1931,7 @@ mod handlers { .collect(); let event = Event { id: sub_id, + agent_idx: Some(0), msg: EventMsg::McpListToolsResponse(crate::protocol::McpListToolsResponseEvent { tools: tools .into_iter() @@ -1915,6 +1955,7 @@ mod handlers { let event = Event { id: sub_id, + agent_idx: Some(0), msg: EventMsg::ListCustomPromptsResponse(ListCustomPromptsResponseEvent { custom_prompts, }), @@ -1965,6 +2006,7 @@ mod handlers { warn!("failed to shutdown rollout recorder: {e}"); let event = Event { id: sub_id.clone(), + agent_idx: Some(0), msg: EventMsg::Error(ErrorEvent { message: "Failed to shutdown rollout recorder".to_string(), codex_error_info: Some(CodexErrorInfo::Other), @@ -1975,6 +2017,7 @@ mod handlers { let event = Event { id: sub_id, + agent_idx: Some(0), msg: EventMsg::ShutdownComplete, }; sess.send_event_raw(event).await; @@ -2004,6 +2047,7 @@ mod handlers { Err(err) => { let event = Event { id: sub_id, + agent_idx: Some(0), msg: EventMsg::Error(ErrorEvent { message: err.to_string(), codex_error_info: Some(CodexErrorInfo::Other), diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 796331d1e8..c875400526 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -175,14 +175,17 @@ async fn forward_events( // ignore all legacy delta events Event { id: _, + agent_idx: _, msg: EventMsg::AgentMessageDelta(_) | EventMsg::AgentReasoningDelta(_), } => {} Event { id: _, + agent_idx: _, msg: EventMsg::SessionConfigured(_), } => {} Event { id, + agent_idx: _, msg: EventMsg::ExecApprovalRequest(event), } => { // Initiate approval via parent session; do not surface to consumer. @@ -198,6 +201,7 @@ async fn forward_events( } Event { id, + agent_idx: _, msg: EventMsg::ApplyPatchApprovalRequest(event), } => { handle_patch_approval( @@ -365,6 +369,7 @@ mod tests { tx_out .send(Event { id: "full".to_string(), + agent_idx: Some(0), msg: EventMsg::TurnAborted(TurnAbortedEvent { reason: TurnAbortReason::Interrupted, }), @@ -384,6 +389,7 @@ mod tests { tx_events .send(Event { id: "evt".to_string(), + agent_idx: Some(0), msg: EventMsg::RawResponseItem(RawResponseItemEvent { item: ResponseItem::CustomToolCall { id: None, diff --git a/codex-rs/core/src/conversation_manager.rs b/codex-rs/core/src/conversation_manager.rs index 0f4577bf1e..d2930982bb 100644 --- a/codex-rs/core/src/conversation_manager.rs +++ b/codex-rs/core/src/conversation_manager.rs @@ -95,6 +95,7 @@ impl ConversationManager { let session_configured = match event { Event { id, + agent_idx: _, msg: EventMsg::SessionConfigured(session_configured), } if id == INITIAL_SUBMIT_ID => session_configured, _ => { diff --git a/codex-rs/core/src/exec.rs b/codex-rs/core/src/exec.rs index f46444675f..212838d888 100644 --- a/codex-rs/core/src/exec.rs +++ b/codex-rs/core/src/exec.rs @@ -681,6 +681,7 @@ async fn read_capped( }); let event = Event { id: stream.sub_id.clone(), + agent_idx: None, msg, }; #[allow(clippy::let_unit_value)] diff --git a/codex-rs/core/src/mcp_connection_manager.rs b/codex-rs/core/src/mcp_connection_manager.rs index 11a90f77a8..9ac2936e5e 100644 --- a/codex-rs/core/src/mcp_connection_manager.rs +++ b/codex-rs/core/src/mcp_connection_manager.rs @@ -158,6 +158,7 @@ impl ElicitationRequestManager { let _ = tx_event .send(Event { id: "mcp_elicitation_request".to_string(), + agent_idx: Some(0), msg: EventMsg::ElicitationRequest(ElicitationRequestEvent { server_name, id, @@ -351,6 +352,7 @@ impl McpConnectionManager { let _ = tx_event .send(Event { id: INITIAL_SUBMIT_ID.to_owned(), + agent_idx: Some(0), msg: EventMsg::McpStartupComplete(summary), }) .await; @@ -641,6 +643,7 @@ async fn emit_update( tx_event .send(Event { id: INITIAL_SUBMIT_ID.to_owned(), + agent_idx: Some(0), msg: EventMsg::McpStartupUpdate(update), }) .await diff --git a/codex-rs/core/src/state/collaboration.rs b/codex-rs/core/src/state/collaboration.rs index 59c6439b0f..c4a5df57ff 100644 --- a/codex-rs/core/src/state/collaboration.rs +++ b/codex-rs/core/src/state/collaboration.rs @@ -10,6 +10,7 @@ use crate::codex::SessionConfiguration; use crate::context_manager::ContextManager; use crate::protocol::TokenUsageInfo; use crate::truncate::TruncationPolicy; +use tracing::info; #[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)] pub(crate) struct AgentId(pub i32); @@ -191,6 +192,14 @@ impl CollaborationState { } pub(crate) fn record_message_for_agent(&mut self, id: AgentId, message: ResponseItem) { + let role = match &message { + ResponseItem::Message { role, .. } => role.as_str(), + _ => "other", + }; + info!( + "collaboration: queued message for agent {} (role={role})", + id.0 + ); if let Some(agent) = self.agent_mut(id) { agent .history diff --git a/codex-rs/core/src/tasks/collaboration.rs b/codex-rs/core/src/tasks/collaboration.rs index 8f91dd4591..cf7b03944f 100644 --- a/codex-rs/core/src/tasks/collaboration.rs +++ b/codex-rs/core/src/tasks/collaboration.rs @@ -277,9 +277,13 @@ async fn run_agent_turns( { let mut collab = session.collaboration_state().lock().await; if let Some(agent) = collab.agent_mut(target) { - agent_status = AgentLifecycleState::Idle { - last_agent_message: last.clone(), - }; + if !responses.is_empty() { + agent_status = AgentLifecycleState::Running; + } else { + agent_status = AgentLifecycleState::Idle { + last_agent_message: last.clone(), + }; + } agent.status = agent_status.clone(); agent.history = new_history.clone(); } @@ -304,7 +308,13 @@ async fn run_agent_turns( remaining_budget = remaining_budget.saturating_sub(delta_tokens); let final_status = { - let collab = session.collaboration_state().lock().await; + let mut collab = session.collaboration_state().lock().await; + if continue_running + && remaining_budget <= 0 + && let Some(agent) = collab.agent_mut(target) + { + agent.status = AgentLifecycleState::Exhausted; + } collab.agent(target).map(|a| a.status.clone()) } .unwrap_or(agent_status.clone()); @@ -317,7 +327,7 @@ async fn run_agent_turns( sub_id: Some(sub_id), }); - keep_running |= continue_running; + keep_running |= continue_running && remaining_budget > 0; if !continue_running || remaining_budget <= 0 { break; } diff --git a/codex-rs/core/src/tools/handlers/collaboration.rs b/codex-rs/core/src/tools/handlers/collaboration.rs index 220b2e0214..ef7e27c02f 100644 --- a/codex-rs/core/src/tools/handlers/collaboration.rs +++ b/codex-rs/core/src/tools/handlers/collaboration.rs @@ -25,6 +25,8 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use serde_json::json; use std::path::PathBuf; +use std::time::Duration; +use std::time::Instant; use tracing::info; #[derive(Debug, Default, Serialize, Deserialize)] @@ -144,6 +146,7 @@ struct CollaborationInitAgentInput { #[serde(default)] context_strategy: Option, #[serde(default)] + #[allow(dead_code)] instructions: Option, #[serde(default)] sandbox_policy: Option, @@ -402,9 +405,9 @@ async fn handle_init_agent( updated_configuration = updated_configuration.apply(&update); } - let instructions = input + let instructions = parent .instructions - .or(parent.instructions.clone()) + .clone() .or(session_configuration.developer_instructions()) .or(session_configuration.user_instructions()); @@ -444,11 +447,6 @@ async fn handle_init_agent( let content = format!("Initialized agent {} at depth {}", assigned_id.0, depth); - if !cfg!(test) { - let supervisor = session.ensure_collaboration_supervisor().await; - supervisor.kick_agent(assigned_id).await; - } - let output = CollaborationInitAgentOutput { content, metadata }; info!( "collaboration_init_agent: assigned_id={}, parent_id={}, depth={}", @@ -492,7 +490,10 @@ async fn handle_send( for raw in &input.recipients { let candidate = AgentId(*raw); - if collab.agent(candidate).is_some() && collab.is_direct_child(sender_id, candidate) { + if let Some(agent) = collab.agent(candidate) + && collab.is_direct_child(sender_id, candidate) + && !matches!(agent.status, AgentLifecycleState::Closed) + { valid_recipients.push(candidate); } else { invalid_recipients.push(*raw); @@ -522,6 +523,10 @@ async fn handle_send( ) }; let output = CollaborationSendOutput { content, metadata }; + info!( + "collaboration_send: sender={}, recipients={:?}, status=error: {}", + sender_id.0, input.recipients, output.content + ); return serialize_function_output(&output, false); } @@ -578,6 +583,10 @@ async fn handle_send( } let output = CollaborationSendOutput { content, metadata }; + info!( + "collaboration_send: sender={}, recipients={:?}, status={}", + sender_id.0, valid_recipients, output.content + ); serialize_function_output(&output, success) } @@ -677,9 +686,14 @@ async fn handle_wait( "no eligible child agents to wait for".to_string() }; let output = CollaborationWaitOutput { content, metadata }; + info!( + "collaboration_wait: caller={}, targets={:?}, status=error: {}", + caller_id.0, targets, output.content + ); return serialize_function_output(&output, false); } + // Passive wait: observe running children only; do not start new work. let supervisor = session.ensure_collaboration_supervisor().await; let mut rx = supervisor.subscribe(); @@ -695,92 +709,116 @@ async fn handle_wait( }; let mut agents_ran = Vec::new(); - { - // Short-circuit if all targets are already idle/terminal. + let mut recorded: std::collections::HashSet = std::collections::HashSet::new(); + let mut active: std::collections::HashSet = { let collab = session.collaboration_state().lock().await; - if targets.iter().all(|id| { - collab - .agent(*id) - .map(|agent| !matches!(agent.status, AgentLifecycleState::Running)) - .unwrap_or(false) - }) { - let mut extra = HashMap::new(); - extra.insert( - "agents_ran".to_string(), - json!( - targets - .iter() - .filter_map(|id| collab.agent(*id)) - .map(|agent| snapshot(agent, 0, None)) - .collect::>() - ), - ); - extra.insert("token_budget_exhausted".to_string(), json!(false)); - let metadata = CollaborationWaitMetadata { - message_tool_call_success: true, - message_tool_call_error_should_penalize_model: false, - is_wait_success_msg: Some(true), - extra: ExtraMetadata(extra), - }; - let output = CollaborationWaitOutput { - content: "Finished waiting.".to_string(), - metadata, - }; - return serialize_function_output(&output, true); - } + targets + .iter() + .filter(|id| { + collab.agent(**id).map_or(false, |agent| { + matches!(agent.status, AgentLifecycleState::Running) + }) + }) + .map(|id| id.0) + .collect() + }; + + if active.is_empty() { + let collab = session.collaboration_state().lock().await; + let mut extra = HashMap::new(); + extra.insert( + "agents_ran".to_string(), + json!( + targets + .iter() + .filter_map(|id| collab.agent(*id)) + .map(|agent| snapshot(agent, 0, None)) + .collect::>() + ), + ); + extra.insert("token_budget_exhausted".to_string(), json!(false)); + let metadata = CollaborationWaitMetadata { + message_tool_call_success: true, + message_tool_call_error_should_penalize_model: false, + is_wait_success_msg: Some(true), + extra: ExtraMetadata(extra), + }; + let output = CollaborationWaitOutput { + content: "Finished waiting.".to_string(), + metadata, + }; + info!( + "collaboration_wait: caller={}, targets={:?}, status=ready", + caller_id.0, targets + ); + return serialize_function_output(&output, true); } - supervisor - .start_agents(targets.clone(), input.max_duration) - .await - .map_err(FunctionCallError::RespondToModel)?; + let deadline = Instant::now() + Duration::from_millis(input.max_duration.max(0) as u64); - let mut remaining = input.max_duration; - let mut active: std::collections::HashSet = targets.iter().map(|id| id.0).collect(); - - while remaining > 0 && !active.is_empty() { - let event = match rx.recv().await { - Ok(evt) => evt, - Err(_) => break, + while !active.is_empty() { + let now = Instant::now(); + if now >= deadline { + break; + } + let remaining_time = deadline.saturating_duration_since(now); + let event = match tokio::time::timeout(remaining_time, rx.recv()).await { + Ok(Ok(evt)) => evt, + _ => break, }; if !active.contains(&event.agent.0) { continue; } - remaining = remaining.saturating_sub(event.delta_tokens); let detail = status_detail(&event.status); + let fallback_last = { + let collab = session.collaboration_state().lock().await; + collab + .agent(event.agent) + .and_then(|agent| last_agent_message(&agent.status).map(str::to_string)) + }; agents_ran.push(json!({ "agent_idx": event.agent.0, "delta_tokens": event.delta_tokens, "status": status_label(&event.status), "status_detail": detail, - "last_agent_message": event.last_message, + "last_agent_message": event.last_message.or(fallback_last), "sub_id": event.sub_id, })); - + recorded.insert(event.agent.0); if !matches!(event.status, AgentLifecycleState::Running) { active.remove(&event.agent.0); } } - let total_tokens = input.max_duration.saturating_sub(remaining); + { + let collab = session.collaboration_state().lock().await; + for id in &targets { + if recorded.contains(&id.0) { + continue; + } + if let Some(agent) = collab.agent(*id) { + agents_ran.push(snapshot(agent, 0, None)); + } + } + } let mut extra = HashMap::new(); extra.insert("agents_ran".to_string(), json!(agents_ran)); - extra.insert( - "token_budget_exhausted".to_string(), - json!(total_tokens >= input.max_duration), - ); let metadata = CollaborationWaitMetadata { message_tool_call_success: true, message_tool_call_error_should_penalize_model: false, - is_wait_success_msg: Some(true), + is_wait_success_msg: Some(active.is_empty()), extra: ExtraMetadata(extra), }; let content = "Finished waiting.".to_string(); let output = CollaborationWaitOutput { content, metadata }; + info!( + "collaboration_wait: caller={}, targets={:?}, ran={:?}", + caller_id.0, targets, agents_ran + ); serialize_function_output(&output, true) } @@ -838,6 +876,7 @@ async fn handle_get_state( content: description, metadata, }; + info!("collaboration_get_state: agents={}", agents.len()); serialize_function_output(&output, true) } @@ -943,6 +982,10 @@ async fn handle_close( ) .await; } + info!( + "collaboration_close: caller={}, closed={:?}", + caller_id.0, closed + ); serialize_function_output(&output, true) } @@ -1035,7 +1078,7 @@ pub(crate) fn create_collaboration_wait_tool() -> ToolSpec { "max_duration".to_string(), JsonSchema::Number { description: Some( - "Maximum duration to wait, measured in tokens. Must be >= 0.".to_string(), + "Maximum duration to wait in ms, measured in tokens. Must be >= 0. A good default value is 10.000".to_string(), ), }, ); diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index 529c7a3a65..fe72c4ff37 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -217,11 +217,11 @@ mod tests { fn builds_specs_without_filtering() { let (_session, turn, _rx) = make_session_and_context_with_rx(); let router = ToolRouter::from_config(&turn.tools_config, None); - let names: Vec = router + let has_update_plan = router .specs() .into_iter() .map(|spec| spec.name().to_string()) - .collect(); - assert!(names.contains(&"update_plan".to_string())); + .any(|name| name == "update_plan"); + assert!(has_update_plan); } } diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index 64a5358f35..50debb8453 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -159,7 +159,11 @@ impl EventProcessor for EventProcessorWithHumanOutput { } fn process_event(&mut self, event: Event) -> CodexStatus { - let Event { id: _, msg } = event; + let Event { + id: _, + agent_idx: _, + msg, + } = event; match msg { EventMsg::Error(ErrorEvent { message, .. }) => { let prefix = "ERROR:".style(self.red); diff --git a/codex-rs/exec/src/event_processor_with_jsonl_output.rs b/codex-rs/exec/src/event_processor_with_jsonl_output.rs index 23dff015eb..bb0db57ffd 100644 --- a/codex-rs/exec/src/event_processor_with_jsonl_output.rs +++ b/codex-rs/exec/src/event_processor_with_jsonl_output.rs @@ -469,6 +469,7 @@ impl EventProcessor for EventProcessorWithJsonOutput { fn print_config_summary(&mut self, _: &Config, _: &str, ev: &SessionConfiguredEvent) { self.process_event(Event { id: "".to_string(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(ev.clone()), }); } diff --git a/codex-rs/exec/tests/event_processor_with_json_output.rs b/codex-rs/exec/tests/event_processor_with_json_output.rs index 7b2d3c18ba..201328736c 100644 --- a/codex-rs/exec/tests/event_processor_with_json_output.rs +++ b/codex-rs/exec/tests/event_processor_with_json_output.rs @@ -59,6 +59,7 @@ use std::time::Duration; fn event(id: &str, msg: EventMsg) -> Event { Event { id: id.to_string(), + agent_idx: Some(0), msg, } } diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index 55808f17ca..8ea95181e4 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -68,6 +68,7 @@ pub async fn run_codex_tool_session( let session_configured_event = Event { // Use a fake id value for now. id: "".to_string(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(session_configured.clone()), }; outgoing diff --git a/codex-rs/mcp-server/src/outgoing_message.rs b/codex-rs/mcp-server/src/outgoing_message.rs index 9e9d079306..ade645f084 100644 --- a/codex-rs/mcp-server/src/outgoing_message.rs +++ b/codex-rs/mcp-server/src/outgoing_message.rs @@ -255,6 +255,7 @@ mod tests { let rollout_file = NamedTempFile::new()?; let event = Event { id: "1".to_string(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { session_id: conversation_id, model: "gpt-4o".to_string(), @@ -309,6 +310,7 @@ mod tests { }; let event = Event { id: "1".to_string(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(session_configured_event.clone()), }; let meta = OutgoingNotificationMeta { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 347cc119ff..48739b13b2 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -437,6 +437,9 @@ impl SandboxPolicy { pub struct Event { /// Submission `id` that this event is correlated with. pub id: String, + /// Collaboration agent index that emitted the event. Root agent is 0. + #[serde(default)] + pub agent_idx: Option, /// Payload pub msg: EventMsg, } @@ -1754,6 +1757,7 @@ mod tests { let rollout_file = NamedTempFile::new()?; let event = Event { id: "1234".to_string(), + agent_idx: Some(0), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { session_id: conversation_id, model: "codex-mini-latest".to_string(), @@ -1813,6 +1817,7 @@ mod tests { fn serialize_mcp_startup_update_event() -> Result<()> { let event = Event { id: "init".to_string(), + agent_idx: Some(0), msg: EventMsg::McpStartupUpdate(McpStartupUpdateEvent { server: "srv".to_string(), status: McpStartupStatus::Failed { @@ -1833,6 +1838,7 @@ mod tests { fn serialize_mcp_startup_complete_event() -> Result<()> { let event = Event { id: "init".to_string(), + agent_idx: Some(0), msg: EventMsg::McpStartupComplete(McpStartupCompleteEvent { ready: vec!["a".to_string()], failed: vec![McpStartupFailure { diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 6872178445..785e285603 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -1563,6 +1563,7 @@ impl ChatWidget { use codex_core::protocol::FileChange; self.app_event_tx.send(AppEvent::CodexEvent(Event { + agent_idx: Some(0), id: "1".to_string(), // msg: EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent { // call_id: "1".to_string(), @@ -1719,7 +1720,11 @@ impl ChatWidget { } pub(crate) fn handle_codex_event(&mut self, event: Event) { - let Event { id, msg } = event; + let Event { + id, + agent_idx: _, + msg, + } = event; self.dispatch_event_msg(Some(id), msg, false); } diff --git a/codex-rs/tui/src/chatwidget/agent.rs b/codex-rs/tui/src/chatwidget/agent.rs index 240972347f..8b8a9cca5e 100644 --- a/codex-rs/tui/src/chatwidget/agent.rs +++ b/codex-rs/tui/src/chatwidget/agent.rs @@ -36,6 +36,7 @@ pub(crate) fn spawn_agent( eprintln!("{message}"); app_event_tx_clone.send(AppEvent::CodexEvent(Event { id: "".to_string(), + agent_idx: Some(0), msg: EventMsg::Error(err.to_error_event(None)), })); app_event_tx_clone.send(AppEvent::ExitRequest); @@ -48,6 +49,7 @@ pub(crate) fn spawn_agent( let ev = codex_core::protocol::Event { // The `id` does not matter for rendering, so we can use a fake value. id: "".to_string(), + agent_idx: Some(0), msg: codex_core::protocol::EventMsg::SessionConfigured(session_configured), }; app_event_tx_clone.send(AppEvent::CodexEvent(ev)); @@ -85,6 +87,7 @@ pub(crate) fn spawn_agent_from_existing( // Forward the captured `SessionConfigured` event so it can be rendered in the UI. let ev = codex_core::protocol::Event { id: "".to_string(), + agent_idx: Some(0), msg: codex_core::protocol::EventMsg::SessionConfigured(session_configured), }; app_event_tx_clone.send(AppEvent::CodexEvent(ev)); diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 890e8bbe1d..c1f4ab03bc 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -99,6 +99,7 @@ fn resumed_initial_messages_render_history() { let conversation_id = ConversationId::new(); let rollout_file = NamedTempFile::new().unwrap(); let configured = codex_core::protocol::SessionConfiguredEvent { + agent_idx: Some(0), session_id: conversation_id, model: "test-model".to_string(), model_provider_id: "test-provider".to_string(), @@ -110,10 +111,12 @@ fn resumed_initial_messages_render_history() { history_entry_count: 0, initial_messages: Some(vec![ EventMsg::UserMessage(UserMessageEvent { + agent_idx: Some(0), message: "hello from user".to_string(), images: None, }), EventMsg::AgentMessage(AgentMessageEvent { + agent_idx: Some(0), message: "assistant reply".to_string(), }), ]), @@ -121,6 +124,7 @@ fn resumed_initial_messages_render_history() { }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "initial".into(), msg: EventMsg::SessionConfigured(configured), }); @@ -153,6 +157,7 @@ fn entered_review_mode_uses_request_hint() { let (mut chat, mut rx, _ops) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "review-start".into(), msg: EventMsg::EnteredReviewMode(ReviewRequest { target: ReviewTarget::BaseBranch { @@ -174,6 +179,7 @@ fn entered_review_mode_defaults_to_current_changes_banner() { let (mut chat, mut rx, _ops) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "review-start".into(), msg: EventMsg::EnteredReviewMode(ReviewRequest { target: ReviewTarget::UncommittedChanges, @@ -194,6 +200,7 @@ fn exited_review_mode_emits_results_and_finishes() { let (mut chat, mut rx, _ops) = make_chatwidget_manual(); let review = ReviewOutputEvent { + agent_idx: Some(0), findings: vec![ReviewFinding { title: "[P1] Fix bug".to_string(), body: "Something went wrong".to_string(), @@ -210,8 +217,10 @@ fn exited_review_mode_emits_results_and_finishes() { }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "review-end".into(), msg: EventMsg::ExitedReviewMode(ExitedReviewModeEvent { + agent_idx: Some(0), review_output: Some(review), }), }); @@ -232,8 +241,10 @@ fn review_restores_context_window_indicator() { let review_tokens = 12_030; // ~97% remaining after subtracting baseline. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "token-before".into(), msg: EventMsg::TokenCount(TokenCountEvent { + agent_idx: Some(0), info: Some(make_token_info(pre_review_tokens, context_window)), rate_limits: None, }), @@ -241,6 +252,7 @@ fn review_restores_context_window_indicator() { assert_eq!(chat.bottom_pane.context_window_percent(), Some(30)); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "review-start".into(), msg: EventMsg::EnteredReviewMode(ReviewRequest { target: ReviewTarget::BaseBranch { @@ -251,8 +263,10 @@ fn review_restores_context_window_indicator() { }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "token-review".into(), msg: EventMsg::TokenCount(TokenCountEvent { + agent_idx: Some(0), info: Some(make_token_info(review_tokens, context_window)), rate_limits: None, }), @@ -260,8 +274,10 @@ fn review_restores_context_window_indicator() { assert_eq!(chat.bottom_pane.context_window_percent(), Some(97)); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "review-end".into(), msg: EventMsg::ExitedReviewMode(ExitedReviewModeEvent { + agent_idx: Some(0), review_output: None, }), }); @@ -280,8 +296,10 @@ fn token_count_none_resets_context_indicator() { let pre_compact_tokens = 12_700; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "token-before".into(), msg: EventMsg::TokenCount(TokenCountEvent { + agent_idx: Some(0), info: Some(make_token_info(pre_compact_tokens, context_window)), rate_limits: None, }), @@ -289,8 +307,10 @@ fn token_count_none_resets_context_indicator() { assert_eq!(chat.bottom_pane.context_window_percent(), Some(30)); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "token-cleared".into(), msg: EventMsg::TokenCount(TokenCountEvent { + agent_idx: Some(0), info: None, rate_limits: None, }), @@ -319,8 +339,10 @@ fn context_indicator_shows_used_tokens_when_window_unknown() { }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "token-usage".into(), msg: EventMsg::TokenCount(TokenCountEvent { + agent_idx: Some(0), info: Some(token_info), rate_limits: None, }), @@ -669,6 +691,7 @@ fn exec_approval_emits_proposed_command_and_decision_history() { // Trigger an exec approval request with a short, single-line command let ev = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-short".into(), turn_id: "turn-short".into(), command: vec!["bash".into(), "-lc".into(), "echo hello world".into()], @@ -680,6 +703,7 @@ fn exec_approval_emits_proposed_command_and_decision_history() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-short".into(), msg: EventMsg::ExecApprovalRequest(ev), }); @@ -713,6 +737,7 @@ fn exec_approval_decision_truncates_multiline_and_long_commands() { // Multiline command: modal should show full command, history records decision only let ev_multi = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-multi".into(), turn_id: "turn-multi".into(), command: vec!["bash".into(), "-lc".into(), "echo line1\necho line2".into()], @@ -724,6 +749,7 @@ fn exec_approval_decision_truncates_multiline_and_long_commands() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-multi".into(), msg: EventMsg::ExecApprovalRequest(ev_multi), }); @@ -765,6 +791,7 @@ fn exec_approval_decision_truncates_multiline_and_long_commands() { // Very long single-line command: decision snippet should be truncated <= 80 chars with trailing ... let long = format!("echo {}", "a".repeat(200)); let ev_long = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-long".into(), turn_id: "turn-long".into(), command: vec!["bash".into(), "-lc".into(), long], @@ -774,6 +801,7 @@ fn exec_approval_decision_truncates_multiline_and_long_commands() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-long".into(), msg: EventMsg::ExecApprovalRequest(ev_long), }); @@ -816,6 +844,7 @@ fn begin_exec_with_source( interaction_input, }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: call_id.to_string(), msg: EventMsg::ExecCommandBegin(event.clone()), }); @@ -839,6 +868,7 @@ fn end_exec( format!("{stdout}{stderr}") }; let ExecCommandBeginEvent { + agent_idx: Some(0), call_id, turn_id, command, @@ -849,8 +879,10 @@ fn end_exec( process_id, } = begin_event; chat.handle_codex_event(Event { + agent_idx: Some(0), id: call_id.clone(), msg: EventMsg::ExecCommandEnd(ExecCommandEndEvent { + agent_idx: Some(0), call_id, process_id, turn_id, @@ -1237,8 +1269,10 @@ fn undo_success_events_render_info_messages() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-1".to_string(), msg: EventMsg::UndoStarted(UndoStartedEvent { + agent_idx: Some(0), message: Some("Undo requested for the last turn...".to_string()), }), }); @@ -1248,8 +1282,10 @@ fn undo_success_events_render_info_messages() { ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-1".to_string(), msg: EventMsg::UndoCompleted(UndoCompletedEvent { + agent_idx: Some(0), success: true, message: None, }), @@ -1274,8 +1310,12 @@ fn undo_failure_events_render_error_message() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-2".to_string(), - msg: EventMsg::UndoStarted(UndoStartedEvent { message: None }), + msg: EventMsg::UndoStarted(UndoStartedEvent { + agent_idx: Some(0), + message: None, + }), }); assert!( chat.bottom_pane.status_indicator_visible(), @@ -1283,8 +1323,10 @@ fn undo_failure_events_render_error_message() { ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-2".to_string(), msg: EventMsg::UndoCompleted(UndoCompletedEvent { + agent_idx: Some(0), success: false, message: Some("Failed to restore workspace state.".to_string()), }), @@ -1309,8 +1351,12 @@ fn undo_started_hides_interrupt_hint() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-hint".to_string(), - msg: EventMsg::UndoStarted(UndoStartedEvent { message: None }), + msg: EventMsg::UndoStarted(UndoStartedEvent { + agent_idx: Some(0), + message: None, + }), }); let status = chat @@ -1432,8 +1478,10 @@ fn view_image_tool_call_adds_history_cell() { let image_path = chat.config.cwd.join("example.png"); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-image".into(), msg: EventMsg::ViewImageToolCall(ViewImageToolCallEvent { + agent_idx: Some(0), call_id: "call-image".into(), path: image_path, }), @@ -1457,8 +1505,10 @@ fn interrupt_exec_marks_failed_snapshot() { // Simulate the task being aborted (as if ESC was pressed), which should // cause the active exec cell to be finalized as failed and flushed. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "call-int".into(), msg: EventMsg::TurnAborted(codex_core::protocol::TurnAbortedEvent { + agent_idx: Some(0), reason: TurnAbortReason::Interrupted, }), }); @@ -1482,16 +1532,20 @@ fn interrupted_turn_error_message_snapshot() { // Simulate an in-progress task so the widget is in a running state. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); // Abort the turn (like pressing Esc) and drain inserted history. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::TurnAborted(codex_core::protocol::TurnAbortedEvent { + agent_idx: Some(0), reason: TurnAbortReason::Interrupted, }), }); @@ -1961,6 +2015,7 @@ fn approval_modal_exec_snapshot() { chat.config.approval_policy = AskForApproval::OnRequest; // Inject an exec approval request to display the approval modal. let ev = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-approve-cmd".into(), turn_id: "turn-approve-cmd".into(), command: vec!["bash".into(), "-lc".into(), "echo hello world".into()], @@ -1972,6 +2027,7 @@ fn approval_modal_exec_snapshot() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-approve".into(), msg: EventMsg::ExecApprovalRequest(ev), }); @@ -2009,6 +2065,7 @@ fn approval_modal_exec_without_reason_snapshot() { chat.config.approval_policy = AskForApproval::OnRequest; let ev = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-approve-cmd-noreason".into(), turn_id: "turn-approve-cmd-noreason".into(), command: vec!["bash".into(), "-lc".into(), "echo hello world".into()], @@ -2018,6 +2075,7 @@ fn approval_modal_exec_without_reason_snapshot() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-approve-noreason".into(), msg: EventMsg::ExecApprovalRequest(ev), }); @@ -2050,6 +2108,7 @@ fn approval_modal_patch_snapshot() { }, ); let ev = ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-approve-patch".into(), turn_id: "turn-approve-patch".into(), changes, @@ -2057,6 +2116,7 @@ fn approval_modal_patch_snapshot() { grant_root: Some(PathBuf::from("/tmp")), }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-approve-patch".into(), msg: EventMsg::ApplyPatchApprovalRequest(ev), }); @@ -2091,8 +2151,10 @@ fn interrupt_restores_queued_messages_into_composer() { // Deliver a TurnAborted event with Interrupted reason (as if Esc was pressed). chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-1".into(), msg: EventMsg::TurnAborted(codex_core::protocol::TurnAbortedEvent { + agent_idx: Some(0), reason: TurnAbortReason::Interrupted, }), }); @@ -2129,8 +2191,10 @@ fn interrupt_prepends_queued_messages_before_existing_composer_text() { chat.refresh_queued_user_messages(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "turn-1".into(), msg: EventMsg::TurnAborted(codex_core::protocol::TurnAbortedEvent { + agent_idx: Some(0), reason: TurnAbortReason::Interrupted, }), }); @@ -2174,14 +2238,18 @@ fn ui_snapshots_small_heights_task_running() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(); // Activate status line chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "**Thinking**".into(), }), }); @@ -2205,21 +2273,26 @@ fn status_widget_and_approval_modal_snapshot() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(); // Begin a running task so the status indicator would be active. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); // Provide a deterministic header for the status line. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "**Analyzing**".into(), }), }); // Now show an approval modal (e.g. exec approval). let ev = ExecApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-approve-exec".into(), turn_id: "turn-approve-exec".into(), command: vec!["echo".into(), "hello world".into()], @@ -2231,6 +2304,7 @@ fn status_widget_and_approval_modal_snapshot() { parsed_cmd: vec![], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-approve-exec".into(), msg: EventMsg::ExecApprovalRequest(ev), }); @@ -2252,15 +2326,19 @@ fn status_widget_active_snapshot() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(); // Activate the status indicator by simulating a task start. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); // Provide a deterministic header via a bold reasoning chunk. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "task-1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "**Analyzing**".into(), }), }); @@ -2279,8 +2357,10 @@ fn background_event_updates_status_header() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "bg-1".into(), msg: EventMsg::BackgroundEvent(BackgroundEventEvent { + agent_idx: Some(0), message: "Waiting for `vim`".to_string(), }), }); @@ -2303,6 +2383,7 @@ fn apply_patch_events_emit_history_cells() { }, ); let ev = ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), changes, @@ -2310,6 +2391,7 @@ fn apply_patch_events_emit_history_cells() { grant_root: None, }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::ApplyPatchApprovalRequest(ev), }); @@ -2344,12 +2426,14 @@ fn apply_patch_events_emit_history_cells() { }, ); let begin = PatchApplyBeginEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), auto_approved: true, changes: changes2, }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::PatchApplyBegin(begin), }); @@ -2370,6 +2454,7 @@ fn apply_patch_events_emit_history_cells() { }, ); let end = PatchApplyEndEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), stdout: "ok\n".into(), @@ -2378,6 +2463,7 @@ fn apply_patch_events_emit_history_cells() { changes: end_changes, }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::PatchApplyEnd(end), }); @@ -2400,8 +2486,10 @@ fn apply_patch_manual_approval_adjusts_header() { }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), changes: proposed_changes, @@ -2419,8 +2507,10 @@ fn apply_patch_manual_approval_adjusts_header() { }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::PatchApplyBegin(PatchApplyBeginEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), auto_approved: false, @@ -2449,8 +2539,10 @@ fn apply_patch_manual_flow_snapshot() { }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), changes: proposed_changes, @@ -2472,8 +2564,10 @@ fn apply_patch_manual_flow_snapshot() { }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::PatchApplyBegin(PatchApplyBeginEvent { + agent_idx: Some(0), call_id: "c1".into(), turn_id: "turn-c1".into(), auto_approved: false, @@ -2502,6 +2596,7 @@ fn apply_patch_approval_sends_op_with_submission_id() { }, ); let ev = ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-999".into(), turn_id: "turn-999".into(), changes, @@ -2509,6 +2604,7 @@ fn apply_patch_approval_sends_op_with_submission_id() { grant_root: None, }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-123".into(), msg: EventMsg::ApplyPatchApprovalRequest(ev), }); @@ -2540,8 +2636,10 @@ fn apply_patch_full_flow_integration_like() { FileChange::Add { content: "".into() }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-xyz".into(), msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-1".into(), turn_id: "turn-call-1".into(), changes, @@ -2581,8 +2679,10 @@ fn apply_patch_full_flow_integration_like() { FileChange::Add { content: "".into() }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-xyz".into(), msg: EventMsg::PatchApplyBegin(PatchApplyBeginEvent { + agent_idx: Some(0), call_id: "call-1".into(), turn_id: "turn-call-1".into(), auto_approved: false, @@ -2595,8 +2695,10 @@ fn apply_patch_full_flow_integration_like() { FileChange::Add { content: "".into() }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-xyz".into(), msg: EventMsg::PatchApplyEnd(PatchApplyEndEvent { + agent_idx: Some(0), call_id: "call-1".into(), turn_id: "turn-call-1".into(), stdout: String::from("ok"), @@ -2620,8 +2722,10 @@ fn apply_patch_untrusted_shows_approval_modal() { FileChange::Add { content: "".into() }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-1".into(), msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-1".into(), turn_id: "turn-call-1".into(), changes, @@ -2669,8 +2773,10 @@ fn apply_patch_request_shows_diff_summary() { }, ); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-apply".into(), msg: EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { + agent_idx: Some(0), call_id: "call-apply".into(), turn_id: "turn-apply".into(), changes, @@ -2739,6 +2845,7 @@ fn plan_update_renders_history_cell() { ], }; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-1".into(), msg: EventMsg::PlanUpdate(update), }); @@ -2760,8 +2867,10 @@ fn stream_error_updates_status_indicator() { chat.bottom_pane.set_task_running(true); let msg = "Reconnecting... 2/5"; chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-1".into(), msg: EventMsg::StreamError(StreamErrorEvent { + agent_idx: Some(0), message: msg.to_string(), codex_error_info: Some(CodexErrorInfo::Other), }), @@ -2783,8 +2892,10 @@ fn stream_error_updates_status_indicator() { fn warning_event_adds_warning_history_cell() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "sub-1".into(), msg: EventMsg::Warning(WarningEvent { + agent_idx: Some(0), message: "test warning message".to_string(), }), }); @@ -2804,32 +2915,40 @@ fn multiple_agent_messages_in_single_turn_emit_multiple_headers() { // Begin turn chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); // First finalized assistant message chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessage(AgentMessageEvent { + agent_idx: Some(0), message: "First message".into(), }), }); // Second finalized assistant message in the same turn chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessage(AgentMessageEvent { + agent_idx: Some(0), message: "Second message".into(), }), }); // End turn chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::TaskComplete(TaskCompleteEvent { + agent_idx: Some(0), last_agent_message: None, }), }); @@ -2858,14 +2977,18 @@ fn final_reasoning_then_message_without_deltas_are_rendered() { // No deltas; only final reasoning followed by final message. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentReasoning(AgentReasoningEvent { + agent_idx: Some(0), text: "I will first analyze the request.".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessage(AgentMessageEvent { + agent_idx: Some(0), message: "Here is the result.".into(), }), }); @@ -2885,47 +3008,61 @@ fn deltas_then_same_final_message_are_rendered_snapshot() { // Stream some reasoning deltas first. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "I will ".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "first analyze the ".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "request.".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentReasoning(AgentReasoningEvent { + agent_idx: Some(0), text: "request.".into(), }), }); // Then stream answer deltas, followed by the exact same final message. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { + agent_idx: Some(0), delta: "Here is the ".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { + agent_idx: Some(0), delta: "result.".into(), }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "s1".into(), msg: EventMsg::AgentMessage(AgentMessageEvent { + agent_idx: Some(0), message: "Here is the result.".into(), }), }); @@ -2946,9 +3083,9 @@ fn deltas_then_same_final_message_are_rendered_snapshot() { #[test] fn chatwidget_exec_and_status_layout_vt100_snapshot() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); - chat.handle_codex_event(Event { + chat.handle_codex_event(Event { agent_idx: Some(0), id: "t1".into(), - msg: EventMsg::AgentMessage(AgentMessageEvent { message: "I’m going to search the repo for where “Change Approved” is rendered to update that view.".into() }), + msg: EventMsg::AgentMessage(AgentMessageEvent { agent_idx: Some(0), message: "I’m going to search the repo for where “Change Approved” is rendered to update that view.".into() }), }); let command = vec!["bash".into(), "-lc".into(), "rg \"Change Approved\"".into()]; @@ -2966,8 +3103,10 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() { ]; let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "c1".into(), msg: EventMsg::ExecCommandBegin(ExecCommandBeginEvent { + agent_idx: Some(0), call_id: "c1".into(), process_id: None, turn_id: "turn-1".into(), @@ -2979,8 +3118,10 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() { }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "c1".into(), msg: EventMsg::ExecCommandEnd(ExecCommandEndEvent { + agent_idx: Some(0), call_id: "c1".into(), process_id: None, turn_id: "turn-1".into(), @@ -2998,14 +3139,18 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() { }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { + agent_idx: Some(0), delta: "**Investigating rendering code**".into(), }), }); @@ -3042,8 +3187,10 @@ fn chatwidget_markdown_code_blocks_vt100_snapshot() { // Simulate a final agent message via streaming deltas instead of a single message chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), }); @@ -3090,8 +3237,12 @@ printf 'fenced within fenced\n' } chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), - msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }), + msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { + agent_idx: Some(0), + delta, + }), }); // Drive commit ticks and drain emitted history lines into the vt100 buffer. loop { @@ -3113,8 +3264,10 @@ printf 'fenced within fenced\n' // Finalize the stream without sending a final AgentMessage, to flush any tail. chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), msg: EventMsg::TaskComplete(TaskCompleteEvent { + agent_idx: Some(0), last_agent_message: None, }), }); @@ -3130,8 +3283,10 @@ printf 'fenced within fenced\n' fn chatwidget_tall() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(); chat.handle_codex_event(Event { + agent_idx: Some(0), id: "t1".into(), msg: EventMsg::TaskStarted(TaskStartedEvent { + agent_idx: Some(0), model_context_window: None, }), });