hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add debug logging system and improve TUI conversation history

Browse Source 
- Add comprehensive debug logging for request/response tracking with unique IDs
- Implement SSE event logging for better streaming response debugging
- Add deduplication logic for function calls in conversation history
- Introduce CustomToolCall events for browser and agent operations
- Enhance TUI with tool call display in history cells
- Add item_id tracking for correlating deltas with output items
- Improve conversation history management with assistant text tracking

This improves debugging capabilities and provides better visibility into
the streaming response flow and tool execution in the TUI.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>
This commit is contained in:
James Peter
2025-08-14 06:53:33 +10:00
co-authored by Claude
parent dc73e51158
commit 92793b3ca9
12 changed files with 760 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/chat_completions.rs
+95 -25
View File
@@ -153,10 +153,12 @@ pub(crate) async fn stream_chat_completions(
serde_json::to_string_pretty(&payload).unwrap_or_default()
);
// Log the request if debug is enabled
if let Ok(logger) = debug_logger.lock() {
let _ = logger.log_request(&endpoint, &payload);
}
// Start logging the request and get a request_id to track the response
let request_id = if let Ok(logger) = debug_logger.lock() {
logger.start_request_log(&endpoint, &payload).unwrap_or_default()
} else {
String::new()
};
let mut attempt = 0;
let max_retries = provider.request_max_retries();
@@ -173,20 +175,23 @@ pub(crate) async fn stream_chat_completions(
match res {
Ok(resp) if resp.status().is_success() => {
// Log successful response initiation (streaming)
// Log successful response initiation
if let Ok(logger) = debug_logger.lock() {
let _ = logger.log_response(&endpoint, &serde_json::json!({
let _ = logger.append_response_event(&request_id, "stream_initiated", &serde_json::json!({
"status": "success",
"type": "stream_initiated",
"status_code": resp.status().as_u16()
}));
}
let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
let debug_logger_clone = Arc::clone(&debug_logger);
let request_id_clone = request_id.clone();
tokio::spawn(process_chat_sse(
stream,
tx_event,
provider.stream_idle_timeout(),
debug_logger_clone,
request_id_clone,
));
return Ok(ResponseStream { rx_event });
}
@@ -196,7 +201,11 @@ pub(crate) async fn stream_chat_completions(
let body = (res.text().await).unwrap_or_default();
// Log error response
if let Ok(logger) = debug_logger.lock() {
let _ = logger.log_error(&endpoint, &format!("Status: {}, Body: {}", status, body));
let _ = logger.append_response_event(&request_id, "error", &serde_json::json!({
"status": status.as_u16(),
"body": body
}));
let _ = logger.end_request_log(&request_id);
}
return Err(CodexErr::UnexpectedStatus(status, body));
}
@@ -220,7 +229,10 @@ pub(crate) async fn stream_chat_completions(
if attempt > max_retries {
// Log network error
if let Ok(logger) = debug_logger.lock() {
let _ = logger.log_error(&endpoint, &format!("Network error: {}", e));
let _ = logger.append_response_event(&request_id, "network_error", &serde_json::json!({
"error": e.to_string()
}));
let _ = logger.end_request_log(&request_id);
}
return Err(e.into());
}
@@ -238,6 +250,8 @@ async fn process_chat_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
debug_logger: Arc<Mutex<DebugLogger>>,
request_id: String,
) where
S: Stream<Item = Result<Bytes>> + Unpin,
{
@@ -259,6 +273,7 @@ async fn process_chat_sse<S>(
let mut fn_call_state = FunctionCallState::default();
let mut assistant_text = String::new();
let mut reasoning_text = String::new();
let mut current_item_id: Option<String> = None;
loop {
let sse = match timeout(idle_timeout, stream.next()).await {
@@ -275,6 +290,10 @@ async fn process_chat_sse<S>(
token_usage: None,
}))
.await;
// Mark the request log as complete
if let Ok(logger) = debug_logger.lock() {
let _ = logger.end_request_log(&request_id);
}
return;
}
Err(_) => {
@@ -295,14 +314,14 @@ async fn process_chat_sse<S>(
content: vec![ContentItem::OutputText {
text: std::mem::take(&mut assistant_text),
}],
id: None,
id: current_item_id.clone(),
};
let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
}
if !reasoning_text.is_empty() {
let item = ResponseItem::Reasoning {
id: String::new(),
id: current_item_id.clone().unwrap_or_else(String::new),
summary: Vec::new(),
content: Some(vec![ReasoningItemContent::ReasoningText {
text: std::mem::take(&mut reasoning_text),
@@ -318,6 +337,10 @@ async fn process_chat_sse<S>(
token_usage: None,
}))
.await;
// Mark the request log as complete
if let Ok(logger) = debug_logger.lock() {
let _ = logger.end_request_log(&request_id);
}
return;
}
@@ -327,10 +350,25 @@ async fn process_chat_sse<S>(
Err(_) => continue,
};
trace!("chat_completions received SSE chunk: {chunk:?}");
// Log the SSE chunk to debug log
if let Ok(logger) = debug_logger.lock() {
let _ = logger.append_response_event(&request_id, "sse_event", &chunk);
}
// Extract item_id if present at the top level or in choice
if let Some(item_id) = chunk.get("item_id").and_then(|id| id.as_str()) {
current_item_id = Some(item_id.to_string());
}
let choice_opt = chunk.get("choices").and_then(|c| c.get(0));
if let Some(choice) = choice_opt {
// Check for item_id in the choice as well
if let Some(item_id) = choice.get("item_id").and_then(|id| id.as_str()) {
current_item_id = Some(item_id.to_string());
}
// Handle assistant content tokens as streaming deltas.
if let Some(content) = choice
.get("delta")
@@ -340,7 +378,10 @@ async fn process_chat_sse<S>(
if !content.is_empty() {
assistant_text.push_str(content);
let _ = tx_event
.send(Ok(ResponseEvent::OutputTextDelta(content.to_string())))
.send(Ok(ResponseEvent::OutputTextDelta {
delta: content.to_string(),
item_id: current_item_id.clone(),
}))
.await;
}
}
@@ -369,7 +410,10 @@ async fn process_chat_sse<S>(
if let Some(reasoning) = maybe_text {
let _ = tx_event
.send(Ok(ResponseEvent::ReasoningContentDelta(reasoning)))
.send(Ok(ResponseEvent::ReasoningContentDelta {
delta: reasoning,
item_id: current_item_id.clone(),
}))
.await;
}
}
@@ -412,7 +456,7 @@ async fn process_chat_sse<S>(
// the reasoning stream before any exec/tool events begin.
if !reasoning_text.is_empty() {
let item = ResponseItem::Reasoning {
id: String::new(),
id: current_item_id.clone().unwrap_or_else(String::new),
summary: Vec::new(),
content: Some(vec![ReasoningItemContent::ReasoningText {
text: std::mem::take(&mut reasoning_text),
@@ -424,7 +468,7 @@ async fn process_chat_sse<S>(
// Then emit the FunctionCall response item.
let item = ResponseItem::FunctionCall {
id: None,
id: current_item_id.clone(),
name: fn_call_state.name.clone().unwrap_or_else(|| "".to_string()),
arguments: fn_call_state.arguments.clone(),
call_id: fn_call_state.call_id.clone().unwrap_or_else(String::new),
@@ -441,14 +485,14 @@ async fn process_chat_sse<S>(
content: vec![ContentItem::OutputText {
text: std::mem::take(&mut assistant_text),
}],
id: None,
id: current_item_id.clone(),
};
let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
}
// Also emit a terminal Reasoning item so UIs can finalize raw reasoning.
if !reasoning_text.is_empty() {
let item = ResponseItem::Reasoning {
id: String::new(),
id: current_item_id.clone().unwrap_or_else(String::new),
summary: Vec::new(),
content: Some(vec![ReasoningItemContent::ReasoningText {
text: std::mem::take(&mut reasoning_text),
@@ -472,6 +516,11 @@ async fn process_chat_sse<S>(
// Prepare for potential next turn (should not happen in same stream).
// fn_call_state = FunctionCallState::default();
// Mark the request log as complete
if let Ok(logger) = debug_logger.lock() {
let _ = logger.end_request_log(&request_id);
}
return; // End processing for this SSE stream.
}
}
@@ -503,6 +552,7 @@ pub(crate) struct AggregatedChatStream<S> {
inner: S,
cumulative: String,
cumulative_reasoning: String,
cumulative_item_id: Option<String>,
pending: std::collections::VecDeque<ResponseEvent>,
mode: AggregateMode,
}
@@ -538,7 +588,11 @@ where
// seen any deltas; otherwise, deltas already built the
// cumulative text and this would duplicate it.
if this.cumulative.is_empty() {
if let crate::models::ResponseItem::Message { content, .. } = &item {
if let crate::models::ResponseItem::Message { content, id, .. } = &item {
// Capture the item_id if present
if let Some(item_id) = id {
this.cumulative_item_id = Some(item_id.clone());
}
if let Some(text) = content.iter().find_map(|c| match c {
crate::models::ContentItem::OutputText { text } => Some(text),
_ => None,
@@ -551,6 +605,13 @@ where
// Swallow assistant message here; emit on Completed.
continue;
}
// Also capture item_id from Reasoning items
if let crate::models::ResponseItem::Reasoning { id, .. } = &item {
if !id.is_empty() {
this.cumulative_item_id = Some(id.clone());
}
}
// Not an assistant message – forward immediately.
return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item))));
@@ -566,7 +627,7 @@ where
&& matches!(this.mode, AggregateMode::AggregatedOnly)
{
let aggregated_reasoning = crate::models::ResponseItem::Reasoning {
id: String::new(),
id: this.cumulative_item_id.clone().unwrap_or_else(String::new),
summary: Vec::new(),
content: Some(vec![
crate::models::ReasoningItemContent::ReasoningText {
@@ -582,7 +643,7 @@ where
if !this.cumulative.is_empty() {
let aggregated_message = crate::models::ResponseItem::Message {
id: None,
id: this.cumulative_item_id.clone(),
role: "assistant".to_string(),
content: vec![crate::models::ContentItem::OutputText {
text: std::mem::take(&mut this.cumulative),
@@ -616,27 +677,35 @@ where
// will never appear in a Chat Completions stream.
continue;
}
Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))) => {
Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta { delta, item_id }))) => {
// Always accumulate deltas so we can emit a final OutputItemDone at Completed.
this.cumulative.push_str(&delta);
// Capture the item_id if we haven't already
if item_id.is_some() && this.cumulative_item_id.is_none() {
this.cumulative_item_id = item_id.clone();
}
if matches!(this.mode, AggregateMode::Streaming) {
// In streaming mode, also forward the delta immediately.
return Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta))));
return Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta { delta, item_id })));
} else {
continue;
}
}
Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => {
Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta { delta, item_id }))) => {
// Always accumulate reasoning deltas so we can emit a final Reasoning item at Completed.
this.cumulative_reasoning.push_str(&delta);
// Capture the item_id if we haven't already
if item_id.is_some() && this.cumulative_item_id.is_none() {
this.cumulative_item_id = item_id.clone();
}
if matches!(this.mode, AggregateMode::Streaming) {
// In streaming mode, also forward the delta immediately.
return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta))));
return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta { delta, item_id })));
} else {
continue;
}
}
Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => {
Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta { .. }))) => {
continue;
}
}
@@ -678,6 +747,7 @@ impl<S> AggregatedChatStream<S> {
inner,
cumulative: String::new(),
cumulative_reasoning: String::new(),
cumulative_item_id: None,
pending: std::collections::VecDeque::new(),
mode,
}
codex-rs/core/src/client.rs
+74 -15
View File
@@ -193,10 +193,12 @@ impl ModelClient {
serde_json::to_string(&payload)?
);
// Log the request if debug is enabled
if let Ok(logger) = self.debug_logger.lock() {
let _ = logger.log_request(&endpoint, &serde_json::to_value(&payload)?);
}
// Start logging the request and get a request_id to track the response
let request_id = if let Ok(logger) = self.debug_logger.lock() {
logger.start_request_log(&endpoint, &serde_json::to_value(&payload)?).unwrap_or_default()
} else {
String::new()
};
loop {
attempt += 1;
@@ -242,13 +244,12 @@ impl ModelClient {
match res {
Ok(resp) if resp.status().is_success() => {
// Log successful response initiation (streaming)
// Log successful response initiation
if let Ok(logger) = self.debug_logger.lock() {
let _ = logger.log_response(&endpoint, &serde_json::json!({
let _ = logger.append_response_event(&request_id, "stream_initiated", &serde_json::json!({
"status": "success",
"type": "stream_initiated",
"status_code": resp.status().as_u16(),
"request_id": resp.headers()
"x_request_id": resp.headers()
.get("x-request-id")
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
@@ -258,10 +259,14 @@ impl ModelClient {
// spawn task to process SSE
let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
let debug_logger = Arc::clone(&self.debug_logger);
let request_id_clone = request_id.clone();
tokio::spawn(process_sse(
stream,
tx_event,
self.provider.stream_idle_timeout(),
debug_logger,
request_id_clone,
));
return Ok(ResponseStream { rx_event });
@@ -288,7 +293,11 @@ impl ModelClient {
let body = res.text().await.unwrap_or_default();
// Log error response
if let Ok(logger) = self.debug_logger.lock() {
let _ = logger.log_error(&endpoint, &format!("Status: {}, Body: {}", status, body));
let _ = logger.append_response_event(&request_id, "error", &serde_json::json!({
"status": status.as_u16(),
"body": body
}));
let _ = logger.end_request_log(&request_id);
}
return Err(CodexErr::UnexpectedStatus(status, body));
}
@@ -395,6 +404,8 @@ async fn process_sse<S>(
stream: S,
tx_event: mpsc::Sender<Result<ResponseEvent>>,
idle_timeout: Duration,
debug_logger: Arc<Mutex<DebugLogger>>,
request_id: String,
) where
S: Stream<Item = Result<Bytes>> + Unpin,
{
@@ -403,6 +414,8 @@ async fn process_sse<S>(
// If the stream stays completely silent for an extended period treat it as disconnected.
// The response id returned from the "complete" message.
let mut response_completed: Option<ResponseCompleted> = None;
// Track the current item_id to include with delta events
let mut current_item_id: Option<String> = None;
loop {
let sse = match timeout(idle_timeout, stream.next()).await {
@@ -433,6 +446,10 @@ async fn process_sse<S>(
.await;
}
}
// Mark the request log as complete
if let Ok(logger) = debug_logger.lock() {
let _ = logger.end_request_log(&request_id);
}
return;
}
Err(_) => {
@@ -445,6 +462,13 @@ async fn process_sse<S>(
trace!("SSE event: {}", sse.data);
// Log the raw SSE event data
if let Ok(logger) = debug_logger.lock() {
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(&sse.data) {
let _ = logger.append_response_event(&request_id, "sse_event", &json_value);
}
}
let event: SseEvent = match serde_json::from_str(&sse.data) {
Ok(event) => event,
Err(e) => {
@@ -474,11 +498,31 @@ async fn process_sse<S>(
// drop the duplicated list inside `response.completed`.
"response.output_item.done" => {
let Some(item_val) = event.item else { continue };
let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) else {
let Ok(item) = serde_json::from_value::<ResponseItem>(item_val.clone()) else {
debug!("failed to parse ResponseItem from output_item.done");
continue;
};
// Extract item_id if present
if let Some(id) = item_val.get("id").and_then(|v| v.as_str()) {
current_item_id = Some(id.to_string());
} else {
// Check within the parsed item structure
match &item {
ResponseItem::Message { id, .. } |
ResponseItem::FunctionCall { id, .. } |
ResponseItem::LocalShellCall { id, .. } => {
if let Some(item_id) = id {
current_item_id = Some(item_id.clone());
}
}
ResponseItem::Reasoning { id, .. } => {
current_item_id = Some(id.clone());
}
_ => {}
}
}
let event = ResponseEvent::OutputItemDone(item);
if tx_event.send(Ok(event)).await.is_err() {
return;
@@ -486,7 +530,10 @@ async fn process_sse<S>(
}
"response.output_text.delta" => {
if let Some(delta) = event.delta {
let event = ResponseEvent::OutputTextDelta(delta);
let event = ResponseEvent::OutputTextDelta {
delta,
item_id: current_item_id.clone()
};
if tx_event.send(Ok(event)).await.is_err() {
return;
}
@@ -494,7 +541,10 @@ async fn process_sse<S>(
}
"response.reasoning_summary_text.delta" => {
if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningSummaryDelta(delta);
let event = ResponseEvent::ReasoningSummaryDelta {
delta,
item_id: current_item_id.clone()
};
if tx_event.send(Ok(event)).await.is_err() {
return;
}
@@ -502,7 +552,10 @@ async fn process_sse<S>(
}
"response.reasoning_text.delta" => {
if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningContentDelta(delta);
let event = ResponseEvent::ReasoningContentDelta {
delta,
item_id: current_item_id.clone()
};
if tx_event.send(Ok(event)).await.is_err() {
return;
}
@@ -573,10 +626,14 @@ async fn stream_from_fixture(
let rdr = std::io::Cursor::new(content);
let stream = ReaderStream::new(rdr).map_err(CodexErr::Io);
// Create a dummy debug logger for testing
let debug_logger = Arc::new(Mutex::new(DebugLogger::new(false).unwrap()));
tokio::spawn(process_sse(
stream,
tx_event,
provider.stream_idle_timeout(),
debug_logger,
String::new(), // Empty request_id for test fixture
));
Ok(ResponseStream { rx_event })
}
@@ -609,7 +666,8 @@ mod tests {
let reader = builder.build();
let stream = ReaderStream::new(reader).map_err(CodexErr::Io);
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(16);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
let debug_logger = Arc::new(Mutex::new(DebugLogger::new(false).unwrap()));
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout(), debug_logger, String::new()));
let mut events = Vec::new();
while let Some(ev) = rx.recv().await {
@@ -639,7 +697,8 @@ mod tests {
let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(8);
let stream = ReaderStream::new(std::io::Cursor::new(body)).map_err(CodexErr::Io);
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
let debug_logger = Arc::new(Mutex::new(DebugLogger::new(false).unwrap()));
tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout(), debug_logger, String::new()));
let mut out = Vec::new();
while let Some(ev) = rx.recv().await {
codex-rs/core/src/client_common.rs
+19 -4
View File
@@ -140,7 +140,22 @@ impl Prompt {
content: vec![ContentItem::InputText { text: ui }],
});
}
input_with_instructions.extend(self.input.clone());
// Deduplicate function call outputs before adding to input
let mut seen_call_ids = std::collections::HashSet::new();
for item in &self.input {
match item {
ResponseItem::FunctionCallOutput { call_id, .. } => {
if !seen_call_ids.insert(call_id.clone()) {
// Skip duplicate function call output
tracing::debug!("Filtering duplicate FunctionCallOutput with call_id: {} from input", call_id);
continue;
}
}
_ => {}
}
input_with_instructions.push(item.clone());
}
// Add ephemeral items at the end so they're fresh for each request
input_with_instructions.extend(self.ephemeral_items.clone());
input_with_instructions
@@ -155,9 +170,9 @@ pub enum ResponseEvent {
response_id: String,
token_usage: Option<TokenUsage>,
},
OutputTextDelta(String),
ReasoningSummaryDelta(String),
ReasoningContentDelta(String),
OutputTextDelta { delta: String, item_id: Option<String> },
ReasoningSummaryDelta { delta: String, item_id: Option<String> },
ReasoningContentDelta { delta: String, item_id: Option<String> },
}
#[derive(Debug, Serialize)]
codex-rs/core/src/codex.rs
+213 -77
View File
@@ -130,8 +130,7 @@ async fn build_turn_ephemeral_items(sess: &Session) -> Vec<ResponseItem> {
let mut content: Vec<ContentItem> = Vec::new();
content.push(ContentItem::InputText {
text: format!(r#"== System Status ==
This ephemeral message is added to the current turn to provide context about your environment. You only see the current version of it.
Use this message only as context and work from the previous user or assistant message.
[ephemeral message - added automatically by system]
cwd: {}
branch: {}
@@ -684,6 +683,14 @@ impl Session {
/// Browser screenshots are filtered out from history to keep them ephemeral.
pub fn turn_input_with_history(&self, extra: Vec<ResponseItem>) -> Vec<ResponseItem> {
let history = self.state.lock().unwrap().history.contents();
// Debug: Count function call outputs in history
let fc_output_count = history.iter().filter(|item| {
matches!(item, ResponseItem::FunctionCallOutput { .. })
}).count();
if fc_output_count > 0 {
debug!("History contains {} FunctionCallOutput items", fc_output_count);
}
// Count images in extra for debugging (we can't distinguish ephemeral at this level anymore)
let images_in_extra = extra
@@ -1464,6 +1471,7 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
ResponseItem::FunctionCall { .. },
Some(ResponseInputItem::FunctionCallOutput { call_id, output }),
) => {
debug!("Recording function call and output for call_id: {}", call_id);
items_to_record_in_conversation_history.push(item);
items_to_record_in_conversation_history.push(
ResponseItem::FunctionCallOutput {
@@ -1525,26 +1533,10 @@ async fn run_agent(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
// Only attempt to take the lock if there is something to record.
if !items_to_record_in_conversation_history.is_empty() {
// To ensure streamed assistant deltas (already appended into
// history during OutputTextDelta) merge cleanly with the final
// assistant message, record assistant messages first. This
// preserves merge-adjacency even if the provider delivered
// other output items (e.g., reasoning) before the final
// assistant message.
let mut assistants = Vec::new();
let mut others = Vec::new();
for item in items_to_record_in_conversation_history.drain(..) {
match &item {
ResponseItem::Message { role, .. } if role == "assistant" => {
assistants.push(item)
}
_ => others.push(item),
}
}
// Maintain relative order within each group while ensuring
// assistants are recorded first for proper merge behaviour.
assistants.extend(others.into_iter());
sess.record_conversation_items(&assistants).await;
// Record items in their original chronological order to maintain
// proper sequence of events. This ensures function calls and their
// outputs appear in the correct order in conversation history.
sess.record_conversation_items(&items_to_record_in_conversation_history).await;
}
// If there are responses, add them to pending input for the next iteration
@@ -1789,29 +1781,34 @@ async fn try_run_turn(
return Ok(output);
}
ResponseEvent::OutputTextDelta(delta) => {
{
let mut st = sess.state.lock().unwrap();
st.history.append_assistant_text(&delta);
}
ResponseEvent::OutputTextDelta { delta, item_id } => {
// Don't append to history during streaming - only send UI events.
// The complete message will be added to history when OutputItemDone arrives.
// This ensures items are recorded in the correct chronological order.
// Use the item_id if present, otherwise fall back to sub_id
let event_id = item_id.unwrap_or_else(|| sub_id.to_string());
let event = Event {
id: sub_id.to_string(),
id: event_id,
msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }),
};
sess.tx_event.send(event).await.ok();
}
ResponseEvent::ReasoningSummaryDelta(delta) => {
ResponseEvent::ReasoningSummaryDelta { delta, item_id } => {
// Use the item_id if present, otherwise fall back to sub_id
let event_id = item_id.unwrap_or_else(|| sub_id.to_string());
let event = Event {
id: sub_id.to_string(),
id: event_id,
msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }),
};
sess.tx_event.send(event).await.ok();
}
ResponseEvent::ReasoningContentDelta(delta) => {
ResponseEvent::ReasoningContentDelta { delta, item_id } => {
if sess.show_raw_agent_reasoning {
// Use the item_id if present, otherwise fall back to sub_id
let event_id = item_id.unwrap_or_else(|| sub_id.to_string());
let event = Event {
id: sub_id.to_string(),
id: event_id,
msg: EventMsg::AgentReasoningRawContentDelta(
AgentReasoningRawContentDeltaEvent { delta },
),
@@ -1917,11 +1914,13 @@ async fn handle_response_item(
) -> CodexResult<Option<ResponseInputItem>> {
debug!(?item, "Output item");
let output = match item {
ResponseItem::Message { content, .. } => {
ResponseItem::Message { content, id, .. } => {
// Use the item_id if present, otherwise fall back to sub_id
let event_id = id.unwrap_or_else(|| sub_id.to_string());
for item in content {
if let ContentItem::OutputText { text } = item {
let event = Event {
id: sub_id.to_string(),
id: event_id.clone(),
msg: EventMsg::AgentMessage(AgentMessageEvent { message: text }),
};
sess.tx_event.send(event).await.ok();
@@ -1930,17 +1929,19 @@ async fn handle_response_item(
None
}
ResponseItem::Reasoning {
id: _,
id,
summary,
content,
encrypted_content: _,
} => {
// Use the item_id if present and not empty, otherwise fall back to sub_id
let event_id = if !id.is_empty() { id.clone() } else { sub_id.to_string() };
for item in summary {
let text = match item {
ReasoningItemReasoningSummary::SummaryText { text } => text,
};
let event = Event {
id: sub_id.to_string(),
id: event_id.clone(),
msg: EventMsg::AgentReasoning(AgentReasoningEvent { text }),
};
sess.tx_event.send(event).await.ok();
@@ -1952,7 +1953,7 @@ async fn handle_response_item(
ReasoningItemContent::ReasoningText { text } => text,
};
let event = Event {
id: sub_id.to_string(),
id: event_id.clone(),
msg: EventMsg::AgentReasoningRawContent(AgentReasoningRawContentEvent {
text,
}),
@@ -3248,16 +3249,92 @@ fn consume_pending_screenshots(sess: &Session) -> Vec<ResponseInputItem> {
.collect()
}
/// Helper function to wrap custom tool calls with events
async fn execute_custom_tool<F, Fut>(
sess: &Session,
sub_id: &str,
call_id: String,
tool_name: String,
parameters: Option<serde_json::Value>,
tool_fn: F,
) -> ResponseInputItem
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = ResponseInputItem>,
{
use std::time::Instant;
use crate::protocol::{CustomToolCallBeginEvent, CustomToolCallEndEvent};
// Send begin event
let begin_event = Event {
id: sub_id.to_string(),
msg: EventMsg::CustomToolCallBegin(CustomToolCallBeginEvent {
call_id: call_id.clone(),
tool_name: tool_name.clone(),
parameters: parameters.clone(),
}),
};
sess.send_event(begin_event).await;
// Execute the tool
let start = Instant::now();
let result = tool_fn().await;
let duration = start.elapsed();
// Extract success/failure from result
let (success, message) = match &result {
ResponseInputItem::FunctionCallOutput { output, .. } => {
let content = &output.content;
let is_error = content.contains("error") || content.contains("failed");
(!is_error, content.clone())
}
_ => (true, String::from("Tool completed")),
};
// Send end event
let end_event = Event {
id: sub_id.to_string(),
msg: EventMsg::CustomToolCallEnd(CustomToolCallEndEvent {
call_id: call_id.clone(),
tool_name,
parameters,
duration,
result: if success {
Ok(message)
} else {
Err(message)
},
}),
};
sess.send_event(end_event).await;
result
}
async fn handle_browser_open(
sess: &Session,
arguments: String,
_sub_id: String,
sub_id: String,
call_id: String,
) -> ResponseInputItem {
// Parse the URL from arguments
let args: Result<Value, _> = serde_json::from_str(&arguments);
// Parse arguments as JSON for the event
let params = serde_json::from_str(&arguments).ok();
let sess_clone = sess;
let arguments_clone = arguments.clone();
let call_id_clone = call_id.clone();
execute_custom_tool(
sess,
&sub_id,
call_id,
"browser_open".to_string(),
params,
|| async move {
// Parse the URL from arguments
let args: Result<Value, _> = serde_json::from_str(&arguments_clone);
match args {
match args {
Ok(json) => {
let url = json
.get("url")
@@ -3285,14 +3362,14 @@ async fn handle_browser_open(
Ok(_) => {
tracing::info!("Browser navigation to {} completed successfully", url);
// Capture screenshot after navigation
match capture_browser_screenshot(sess).await {
match capture_browser_screenshot(sess_clone).await {
Ok((screenshot_path, updated_url)) => {
tracing::info!(
"Screenshot captured after navigation: {} at URL: {}",
screenshot_path.display(),
updated_url
);
add_pending_screenshot(sess, screenshot_path, updated_url);
add_pending_screenshot(sess_clone, screenshot_path, updated_url);
}
Err(e) => {
tracing::warn!("Screenshot capture failed after navigation: {}", e);
@@ -3300,7 +3377,7 @@ async fn handle_browser_open(
}
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!(
"Browser opened to: {}. Screenshots will be automatically attached to messages.",
@@ -3311,7 +3388,7 @@ async fn handle_browser_open(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to navigate browser to {}: {}", url, e),
success: Some(false),
@@ -3320,7 +3397,7 @@ async fn handle_browser_open(
}
} else {
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: "Failed to initialize browser manager.".to_string(),
success: Some(false),
@@ -3329,13 +3406,16 @@ async fn handle_browser_open(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone,
output: FunctionCallOutputPayload {
content: format!("Failed to parse browser_open arguments: {}", e),
success: Some(false),
},
},
}
},
)
.await
}
/// Get the browser manager for the session (always uses global)
@@ -3348,17 +3428,27 @@ async fn get_browser_manager_for_session(
async fn handle_browser_close(
sess: &Session,
_sub_id: String,
sub_id: String,
call_id: String,
) -> ResponseInputItem {
let browser_manager = get_browser_manager_for_session(sess).await;
let sess_clone = sess;
let call_id_clone = call_id.clone();
execute_custom_tool(
sess,
&sub_id,
call_id,
"browser_close".to_string(),
None,
|| async move {
let browser_manager = get_browser_manager_for_session(sess_clone).await;
if let Some(browser_manager) = browser_manager {
match browser_manager.stop().await {
Ok(_) => {
// Clear the browser manager from global
codex_browser::global::clear_browser_manager().await;
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: "Browser closed. Screenshot capture disabled.".to_string(),
success: Some(true),
@@ -3366,7 +3456,7 @@ async fn handle_browser_close(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to close browser: {}", e),
success: Some(false),
@@ -3375,21 +3465,34 @@ async fn handle_browser_close(
}
} else {
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone,
output: FunctionCallOutputPayload {
content: "Browser is not currently open.".to_string(),
success: Some(false),
},
}
}
},
)
.await
}
async fn handle_browser_status(
sess: &Session,
_sub_id: String,
sub_id: String,
call_id: String,
) -> ResponseInputItem {
let browser_manager = get_browser_manager_for_session(sess).await;
let sess_clone = sess;
let call_id_clone = call_id.clone();
execute_custom_tool(
sess,
&sub_id,
call_id,
"browser_status".to_string(),
None,
|| async move {
let browser_manager = get_browser_manager_for_session(sess_clone).await;
if let Some(browser_manager) = browser_manager {
let status = browser_manager.get_status().await;
let status_msg = if status.enabled {
@@ -3403,7 +3506,7 @@ async fn handle_browser_status(
};
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: status_msg,
success: Some(true),
@@ -3411,7 +3514,7 @@ async fn handle_browser_status(
}
} else {
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone,
output: FunctionCallOutputPayload {
content: "Browser is not initialized. Use browser_open to start the browser."
.to_string(),
@@ -3419,18 +3522,33 @@ async fn handle_browser_status(
},
}
}
},
)
.await
}
async fn handle_browser_click(
sess: &Session,
arguments: String,
_sub_id: String,
sub_id: String,
call_id: String,
) -> ResponseInputItem {
let browser_manager = get_browser_manager_for_session(sess).await;
let params = serde_json::from_str(&arguments).ok();
let sess_clone = sess;
let arguments_clone = arguments.clone();
let call_id_clone = call_id.clone();
execute_custom_tool(
sess,
&sub_id,
call_id,
"browser_click".to_string(),
params,
|| async move {
let browser_manager = get_browser_manager_for_session(sess_clone).await;
if let Some(browser_manager) = browser_manager {
let args: Result<Value, _> = serde_json::from_str(&arguments);
if let Some(browser_manager) = browser_manager {
let args: Result<Value, _> = serde_json::from_str(&arguments_clone);
match args {
Ok(json) => {
let x = json.get("x").and_then(|v| v.as_f64()).unwrap_or(0.0);
@@ -3439,13 +3557,13 @@ async fn handle_browser_click(
match browser_manager.click(x, y).await {
Ok(_) => {
// Capture screenshot after clicking
if let Ok((screenshot_path, url)) = capture_browser_screenshot(sess).await
if let Ok((screenshot_path, url)) = capture_browser_screenshot(sess_clone).await
{
add_pending_screenshot(sess, screenshot_path, url);
add_pending_screenshot(sess_clone, screenshot_path, url);
}
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Clicked at ({}, {})", x, y),
success: Some(true),
@@ -3453,7 +3571,7 @@ async fn handle_browser_click(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to click: {}", e),
success: Some(false),
@@ -3462,7 +3580,7 @@ async fn handle_browser_click(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to parse browser_click arguments: {}", e),
success: Some(false),
@@ -3471,7 +3589,7 @@ async fn handle_browser_click(
}
} else {
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone,
output: FunctionCallOutputPayload {
content: "Browser is not initialized. Use browser_open to start the browser."
.to_string(),
@@ -3479,17 +3597,32 @@ async fn handle_browser_click(
},
}
}
},
)
.await
}
async fn handle_browser_type(
sess: &Session,
arguments: String,
_sub_id: String,
sub_id: String,
call_id: String,
) -> ResponseInputItem {
let browser_manager = get_browser_manager_for_session(sess).await;
if let Some(browser_manager) = browser_manager {
let args: Result<Value, _> = serde_json::from_str(&arguments);
let params = serde_json::from_str(&arguments).ok();
let sess_clone = sess;
let arguments_clone = arguments.clone();
let call_id_clone = call_id.clone();
execute_custom_tool(
sess,
&sub_id,
call_id,
"browser_type".to_string(),
params,
|| async move {
let browser_manager = get_browser_manager_for_session(sess_clone).await;
if let Some(browser_manager) = browser_manager {
let args: Result<Value, _> = serde_json::from_str(&arguments_clone);
match args {
Ok(json) => {
let text = json.get("text").and_then(|v| v.as_str()).unwrap_or("");
@@ -3497,13 +3630,13 @@ async fn handle_browser_type(
match browser_manager.type_text(text).await {
Ok(_) => {
// Capture screenshot after typing
if let Ok((screenshot_path, url)) = capture_browser_screenshot(sess).await
if let Ok((screenshot_path, url)) = capture_browser_screenshot(sess_clone).await
{
add_pending_screenshot(sess, screenshot_path, url);
add_pending_screenshot(sess_clone, screenshot_path, url);
}
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Typed: {}", text),
success: Some(true),
@@ -3511,7 +3644,7 @@ async fn handle_browser_type(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to type text: {}", e),
success: Some(false),
@@ -3520,7 +3653,7 @@ async fn handle_browser_type(
}
}
Err(e) => ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone.clone(),
output: FunctionCallOutputPayload {
content: format!("Failed to parse browser_type arguments: {}", e),
success: Some(false),
@@ -3529,7 +3662,7 @@ async fn handle_browser_type(
}
} else {
ResponseInputItem::FunctionCallOutput {
call_id,
call_id: call_id_clone,
output: FunctionCallOutputPayload {
content: "Browser is not initialized. Use browser_open to start the browser."
.to_string(),
@@ -3537,6 +3670,9 @@ async fn handle_browser_type(
},
}
}
},
)
.await
}
async fn handle_browser_key(
codex-rs/core/src/conversation_history.rs
+26
View File
@@ -1,4 +1,5 @@
use crate::models::ResponseItem;
use tracing::debug;
/// Transcript of conversation history
#[derive(Debug, Clone, Default)]
@@ -68,6 +69,31 @@ impl ConversationHistory {
// 3) No merge opportunity – push as a new assistant message.
self.items.push(item.clone());
}
ResponseItem::FunctionCallOutput { call_id, .. } => {
// Check if we already have an output for this call_id to prevent duplicates
let already_exists = self.items.iter().any(|existing| {
matches!(existing, ResponseItem::FunctionCallOutput { call_id: existing_id, .. } if existing_id == call_id)
});
if already_exists {
debug!("Skipping duplicate FunctionCallOutput for call_id: {} (already in history)", call_id);
} else {
debug!("Recording FunctionCallOutput to history for call_id: {}", call_id);
self.items.push(item.clone());
}
}
ResponseItem::FunctionCall { call_id, .. } => {
// Check if we already have this function call to prevent duplicates during retries
let already_exists = self.items.iter().any(|existing| {
matches!(existing, ResponseItem::FunctionCall { call_id: existing_id, .. } if existing_id == call_id)
});
if already_exists {
debug!("Skipping duplicate FunctionCall for call_id: {} (already in history)", call_id);
} else {
self.items.push(item.clone());
}
}
_ => {
self.items.push(item.clone());
}
codex-rs/core/src/debug_logger.rs
+141 -22
View File
@@ -1,11 +1,21 @@
use std::fs;
use std::path::PathBuf;
use std::collections::HashMap;
use std::sync::Mutex;
use chrono::Local;
use serde_json::Value;
use uuid::Uuid;
struct StreamInfo {
response_file: PathBuf,
events: Vec<Value>,
}
pub struct DebugLogger {
enabled: bool,
log_dir: PathBuf,
// Maps request_id to stream info for collecting events
active_streams: Mutex<HashMap<String, StreamInfo>>,
}
impl DebugLogger {
@@ -14,6 +24,7 @@ impl DebugLogger {
return Ok(Self {
enabled: false,
log_dir: PathBuf::new(),
active_streams: Mutex::new(HashMap::new()),
});
}
@@ -27,9 +38,105 @@ impl DebugLogger {
Ok(Self {
enabled,
log_dir,
active_streams: Mutex::new(HashMap::new()),
})
}
/// Start a new request/response log file and return the request ID
pub fn start_request_log(&self, endpoint: &str, payload: &Value) -> Result<String, std::io::Error> {
if !self.enabled {
return Ok(String::new());
}
let timestamp = Local::now();
let request_id = Uuid::new_v4().to_string();
let request_id_short = &request_id[..8]; // Use first 8 chars of UUID for brevity
// Create request file with pretty-printed JSON
let request_filename = format!(
"{}_{}_request.json",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
request_id_short
);
let request_file_path = self.log_dir.join(request_filename);
// Create request object with metadata
let request_entry = serde_json::json!({
"timestamp": timestamp.to_rfc3339(),
"request_id": request_id,
"endpoint": endpoint,
"payload": payload
});
// Write pretty-printed JSON to request file
let formatted_request = serde_json::to_string_pretty(&request_entry)?;
fs::write(&request_file_path, formatted_request)?;
// Prepare response file path
let response_filename = format!(
"{}_{}_response.json",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
request_id_short
);
let response_file_path = self.log_dir.join(response_filename);
// Store the stream info for this request_id
if let Ok(mut streams) = self.active_streams.lock() {
streams.insert(request_id.clone(), StreamInfo {
response_file: response_file_path,
events: Vec::new(),
});
}
Ok(request_id)
}
/// Append a response event to the in-memory event list
pub fn append_response_event(&self, request_id: &str, event_type: &str, data: &Value) -> Result<(), std::io::Error> {
if !self.enabled || request_id.is_empty() {
return Ok(());
}
if let Ok(mut streams) = self.active_streams.lock() {
if let Some(stream_info) = streams.get_mut(request_id) {
let timestamp = Local::now();
let event_entry = serde_json::json!({
"timestamp": timestamp.to_rfc3339(),
"type": event_type,
"data": data
});
stream_info.events.push(event_entry);
}
}
Ok(())
}
/// Mark a stream as completed and write all collected events to response file
pub fn end_request_log(&self, request_id: &str) -> Result<(), std::io::Error> {
if !self.enabled || request_id.is_empty() {
return Ok(());
}
if let Ok(mut streams) = self.active_streams.lock() {
if let Some(stream_info) = streams.remove(request_id) {
// Create the response object with all events as an array
let response_data = serde_json::json!({
"request_id": request_id,
"completed_at": Local::now().to_rfc3339(),
"events": stream_info.events
});
// Write pretty-printed JSON to response file
let formatted_response = serde_json::to_string_pretty(&response_data)?;
fs::write(&stream_info.response_file, formatted_response)?;
}
}
Ok(())
}
// Legacy methods for backward compatibility - they now create standalone files
pub fn log_request(&self, endpoint: &str, payload: &Value) -> Result<(), std::io::Error> {
if !self.enabled {
return Ok(());
@@ -37,9 +144,8 @@ impl DebugLogger {
let timestamp = Local::now();
let filename = format!(
"{}_request_{}.json",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
sanitize_endpoint(endpoint)
"{}_request.json",
timestamp.format("%Y%m%d_%H%M%S%.3f")
);
let file_path = self.log_dir.join(filename);
@@ -64,9 +170,8 @@ impl DebugLogger {
let timestamp = Local::now();
let filename = format!(
"{}_response_{}.json",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
sanitize_endpoint(endpoint)
"{}_response.json",
timestamp.format("%Y%m%d_%H%M%S%.3f")
);
let file_path = self.log_dir.join(filename);
@@ -91,9 +196,8 @@ impl DebugLogger {
let timestamp = Local::now();
let filename = format!(
"{}_stream_{}.txt",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
sanitize_endpoint(endpoint)
"{}_stream.txt",
timestamp.format("%Y%m%d_%H%M%S%.3f")
);
let file_path = self.log_dir.join(filename);
@@ -117,9 +221,8 @@ impl DebugLogger {
let timestamp = Local::now();
let filename = format!(
"{}_error_{}.txt",
timestamp.format("%Y%m%d_%H%M%S%.3f"),
sanitize_endpoint(endpoint)
"{}_error.txt",
timestamp.format("%Y%m%d_%H%M%S%.3f")
);
let file_path = self.log_dir.join(filename);
@@ -139,15 +242,31 @@ impl DebugLogger {
pub fn is_enabled(&self) -> bool {
self.enabled
}
}
fn sanitize_endpoint(endpoint: &str) -> String {
endpoint
.replace('/', "_")
.replace('\\', "_")
.replace(':', "_")
.replace(' ', "_")
.chars()
.filter(|c| c.is_alphanumeric() || *c == '_' || *c == '-')
.collect()
pub fn log_sse_event(&self, endpoint: &str, event_data: &Value) -> Result<(), std::io::Error> {
// Legacy method - now creates standalone files
if !self.enabled {
return Ok(());
}
let timestamp = Local::now();
let filename = format!(
"{}_sse.json",
timestamp.format("%Y%m%d_%H%M%S%.3f")
);
let file_path = self.log_dir.join(filename);
let log_entry = serde_json::json!({
"timestamp": timestamp.to_rfc3339(),
"type": "sse_event",
"endpoint": endpoint,
"event": event_data
});
let formatted = serde_json::to_string_pretty(&log_entry)?;
fs::write(file_path, formatted)?;
Ok(())
}
}
codex-rs/core/src/protocol.rs
+28
View File
@@ -401,6 +401,10 @@ pub enum EventMsg {
McpToolCallEnd(McpToolCallEndEvent),
/// Custom tool call events for non-MCP tools (browser, agent, etc)
CustomToolCallBegin(CustomToolCallBeginEvent),
CustomToolCallEnd(CustomToolCallEndEvent),
/// Notification that the server is about to execute a command.
ExecCommandBegin(ExecCommandBeginEvent),
@@ -587,6 +591,30 @@ impl McpToolCallEndEvent {
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct CustomToolCallBeginEvent {
/// Identifier so this can be paired with the CustomToolCallEnd event.
pub call_id: String,
/// Name of the tool (e.g., "browser_navigate", "run_agent")
pub tool_name: String,
/// Parameters passed to the tool as JSON
pub parameters: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct CustomToolCallEndEvent {
/// Identifier for the corresponding CustomToolCallBegin that finished.
pub call_id: String,
/// Name of the tool
pub tool_name: String,
/// Parameters passed to the tool as JSON
pub parameters: Option<serde_json::Value>,
/// Duration of the tool call
pub duration: Duration,
/// Result of the tool call (success message or error)
pub result: Result<String, String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ExecCommandBeginEvent {
/// Identifier so this can be paired with the ExecCommandEnd event.
codex-rs/exec/src/event_processor_with_human_output.rs
+29
View File
@@ -521,6 +521,35 @@ impl EventProcessor for EventProcessorWithHumanOutput {
// Currently ignored in exec output.
}
EventMsg::ShutdownComplete => return CodexStatus::Shutdown,
EventMsg::CustomToolCallBegin(event) => {
ts_println!(
self,
"{} {}",
"tool".style(self.magenta),
event.tool_name.style(self.bold),
);
if let Some(params) = &event.parameters {
if let Ok(formatted) = serde_json::to_string_pretty(params) {
for line in formatted.lines() {
println!("{}", line.style(self.dimmed));
}
}
}
}
EventMsg::CustomToolCallEnd(event) => {
let status = if event.result.is_ok() {
"success".style(self.green)
} else {
"failed".style(self.red)
};
ts_println!(
self,
"{} {} {}",
"tool".style(self.magenta),
event.tool_name.style(self.bold),
status,
);
}
}
CodexStatus::Running
}
codex-rs/mcp-server/src/codex_tool_runner.rs
+3 -1
View File
@@ -270,7 +270,9 @@ async fn run_codex_tool_session_inner(
| EventMsg::PlanUpdate(_)
| EventMsg::BrowserScreenshotUpdate(_)
| EventMsg::AgentStatusUpdate(_)
| EventMsg::ShutdownComplete => {
| EventMsg::ShutdownComplete
| EventMsg::CustomToolCallBegin(_)
| EventMsg::CustomToolCallEnd(_) => {
// For now, we do not do anything extra for these
// events. Note that
// send(codex_event_to_notification(&event)) above has
codex-rs/mcp-server/src/conversation_loop.rs
+3 -1
View File
@@ -108,7 +108,9 @@ pub async fn run_conversation_loop(
| EventMsg::PlanUpdate(_)
| EventMsg::BrowserScreenshotUpdate(_)
| EventMsg::AgentStatusUpdate(_)
| EventMsg::ShutdownComplete => {
| EventMsg::ShutdownComplete
| EventMsg::CustomToolCallBegin(_)
| EventMsg::CustomToolCallEnd(_) => {
// For now, we do not do anything extra for these
// events. Note that
// send(codex_event_to_notification(&event)) above has
codex-rs/tui/src/chatwidget.rs
+28
View File
@@ -27,6 +27,8 @@ use codex_core::protocol::ExecApprovalRequestEvent;
use codex_core::protocol::ExecCommandBeginEvent;
use codex_core::protocol::ExecCommandEndEvent;
use codex_core::protocol::InputItem;
use codex_core::protocol::CustomToolCallBeginEvent;
use codex_core::protocol::CustomToolCallEndEvent;
use codex_core::protocol::McpToolCallBeginEvent;
use codex_core::protocol::McpToolCallEndEvent;
use codex_core::protocol::Op;
@@ -1367,6 +1369,32 @@ impl ChatWidget<'_> {
result,
));
}
EventMsg::CustomToolCallBegin(CustomToolCallBeginEvent {
call_id: _,
tool_name,
parameters,
}) => {
self.finalize_active_stream();
self.add_to_history(HistoryCell::new_active_custom_tool_call(
tool_name,
parameters,
));
}
EventMsg::CustomToolCallEnd(CustomToolCallEndEvent {
call_id: _,
tool_name,
parameters,
duration,
result,
}) => {
self.add_to_history(HistoryCell::new_completed_custom_tool_call(
80,
tool_name,
parameters,
duration,
result,
));
}
EventMsg::GetHistoryEntryResponse(event) => {
let codex_core::protocol::GetHistoryEntryResponseEvent {
offset,
codex-rs/tui/src/history_cell.rs
+101 -3
View File
@@ -100,6 +100,12 @@ pub(crate) enum HistoryCell {
/// Completed MCP tool call where we show the result serialized as JSON.
CompletedMcpToolCall { view: TextBlock },
/// An active custom tool call (browser, agent, etc) that has not finished yet.
ActiveCustomToolCall { view: TextBlock },
/// Completed custom tool call with result
CompletedCustomToolCall { view: TextBlock },
/// Completed MCP tool call where the result is an image.
/// Admittedly, [mcp_types::CallToolResult] can have multiple content types,
/// which could be a mix of text and images, so we need to tighten this up.
@@ -291,7 +297,9 @@ impl HistoryCell {
| HistoryCell::PendingPatch { view }
| HistoryCell::PlanUpdate { view }
| HistoryCell::PatchApplyResult { view }
| HistoryCell::ActiveMcpToolCall { view, .. } => {
| HistoryCell::ActiveMcpToolCall { view, .. }
| HistoryCell::ActiveCustomToolCall { view, .. }
| HistoryCell::CompletedCustomToolCall { view, .. } => {
view.lines.iter().map(line_to_static).collect()
}
HistoryCell::Exec(ExecCell {
@@ -543,6 +551,33 @@ impl HistoryCell {
}
}
pub(crate) fn new_active_custom_tool_call(
tool_name: String,
parameters: Option<serde_json::Value>,
) -> Self {
let title_line = Line::from(vec!["tool".magenta(), " running...".dim()]);
let mut lines: Vec<Line> = vec![title_line];
// Add tool name
lines.push(Line::from(vec![tool_name.bold()]));
// Add parameters if present
if let Some(params) = parameters {
if let Ok(formatted) = serde_json::to_string_pretty(&params) {
lines.push(Line::from(""));
for line in formatted.lines() {
lines.push(Line::from(line.to_string().dim()));
}
}
}
lines.push(Line::from(""));
HistoryCell::ActiveCustomToolCall {
view: TextBlock::new(lines),
}
}
/// If the first content is an image, return a new cell with the image.
/// TODO(rgwood-dd): Handle images properly even if they're not the first result.
fn try_new_completed_mcp_tool_call_with_image_output(
@@ -672,6 +707,65 @@ impl HistoryCell {
}
// allow dead code for now. maybe we'll use it again.
#[allow(dead_code)]
pub(crate) fn new_completed_custom_tool_call(
num_cols: usize,
tool_name: String,
parameters: Option<serde_json::Value>,
duration: Duration,
result: Result<String, String>,
) -> Self {
let duration_str = format_duration(duration);
let success = result.is_ok();
let status_str = if success { "success" } else { "failed" };
let title_line = Line::from(vec![
"tool".magenta(),
" ".into(),
if success {
status_str.green()
} else {
status_str.red()
},
format!(", duration: {duration_str}").gray(),
]);
let mut lines: Vec<Line<'static>> = Vec::new();
lines.push(title_line);
lines.push(Line::from(vec![tool_name.bold()]));
// Add parameters if present
if let Some(params) = parameters {
if let Ok(formatted) = serde_json::to_string_pretty(&params) {
lines.push(Line::from(""));
for line in formatted.lines().take(10) {
lines.push(Line::from(line.to_string().dim()));
}
if formatted.lines().count() > 10 {
lines.push(Line::from("...".dim()));
}
}
}
// Add result
lines.push(Line::from(""));
match result {
Ok(msg) => {
let truncated = format_and_truncate_tool_result(&msg, TOOL_CALL_MAX_LINES, num_cols);
for line in truncated.lines() {
lines.push(Line::from(line.to_string()));
}
}
Err(err) => {
lines.push(Line::from(format!("Error: {}", err).red()));
}
}
lines.push(Line::from(""));
HistoryCell::CompletedCustomToolCall {
view: TextBlock::new(lines),
}
}
pub(crate) fn new_background_event(message: String) -> Self {
let mut lines: Vec<Line<'static>> = Vec::new();
lines.push(Line::from("event".dim()));
@@ -1261,7 +1355,9 @@ impl HistoryCell {
| HistoryCell::PlanUpdate { .. }
| HistoryCell::PatchApplyResult { .. }
| HistoryCell::CompletedMcpToolCall { .. }
| HistoryCell::ActiveMcpToolCall { .. } => {
| HistoryCell::ActiveMcpToolCall { .. }
| HistoryCell::ActiveCustomToolCall { .. }
| HistoryCell::CompletedCustomToolCall { .. } => {
// Use the same processed/plain lines as desired_height/render_ref but apply a vertical scroll.
let lines = match self {
HistoryCell::BackgroundEvent { .. }
@@ -1278,7 +1374,9 @@ impl HistoryCell {
| HistoryCell::PlanUpdate { view }
| HistoryCell::PatchApplyResult { view }
| HistoryCell::CompletedMcpToolCall { view }
| HistoryCell::ActiveMcpToolCall { view } => view.lines.clone(),
| HistoryCell::ActiveMcpToolCall { view }
| HistoryCell::ActiveCustomToolCall { view }
| HistoryCell::CompletedCustomToolCall { view } => view.lines.clone(),
_ => self.plain_lines(),
};
Paragraph::new(Text::from(lines))