valknar/llmx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert to 3f13ebce10 without rewriting history. Wrong merge

Browse Source
This commit is contained in:
Ahmed Ibrahim
2025-08-04 17:03:24 -07:00
parent 1a33de34b0
commit e38ce39c51
9 changed files with 55 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
codex-rs/core/src/chat_completions.rs
View File

@@ -207,7 +207,6 @@ async fn process_chat_sse<S>(
} }
let mut fn_call_state = FunctionCallState::default(); let mut fn_call_state = FunctionCallState::default();
let mut assistant_text = String::new();
loop { loop {
let sse = match timeout(idle_timeout, stream.next()).await { let sse = match timeout(idle_timeout, stream.next()).await {
@@ -255,42 +254,21 @@ async fn process_chat_sse<S>(
let choice_opt = chunk.get("choices").and_then(|c| c.get(0)); let choice_opt = chunk.get("choices").and_then(|c| c.get(0));
if let Some(choice) = choice_opt { if let Some(choice) = choice_opt {
// Handle assistant content tokens as streaming deltas. // Handle assistant content tokens.
if let Some(content) = choice if let Some(content) = choice
.get("delta") .get("delta")
.and_then(|d| d.get("content")) .and_then(|d| d.get("content"))
.and_then(|c| c.as_str()) .and_then(|c| c.as_str())
{ {
if !content.is_empty() { let item = ResponseItem::Message {
assistant_text.push_str(content); role: "assistant".to_string(),
let _ = tx_event content: vec![ContentItem::OutputText {
.send(Ok(ResponseEvent::OutputTextDelta(content.to_string()))) text: content.to_string(),
.await; }],
} id: None,
} };
// Forward any reasoning/thinking deltas if present. let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
if let Some(reasoning) = choice
.get("delta")
.and_then(|d| d.get("reasoning"))
.and_then(|c| c.as_str())
{
let _ = tx_event
.send(Ok(ResponseEvent::ReasoningSummaryDelta(
reasoning.to_string(),
)))
.await;
}
if let Some(reasoning_content) = choice
.get("delta")
.and_then(|d| d.get("reasoning_content"))
.and_then(|c| c.as_str())
{
let _ = tx_event
.send(Ok(ResponseEvent::ReasoningSummaryDelta(
reasoning_content.to_string(),
)))
.await;
} }
// Handle streaming function / tool calls. // Handle streaming function / tool calls.
@@ -339,18 +317,7 @@ async fn process_chat_sse<S>(
let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
} }
"stop" => { "stop" => {
// Regular turn without tool-call. Emit the final assistant message // Regular turn without tool-call.
// as a single OutputItemDone so non-delta consumers see the result.
if !assistant_text.is_empty() {
let item = ResponseItem::Message {
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: std::mem::take(&mut assistant_text),
}],
id: None,
};
let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
}
} }
_ => {} _ => {}
} }
@@ -391,10 +358,7 @@ async fn process_chat_sse<S>(
pub(crate) struct AggregatedChatStream<S> { pub(crate) struct AggregatedChatStream<S> {
inner: S, inner: S,
cumulative: String, cumulative: String,
cumulative_reasoning: String, pending_completed: Option<ResponseEvent>,
pending: std::collections::VecDeque<ResponseEvent>,
// When true, do not emit a cumulative assistant message at Completed.
streaming_mode: bool,
} }
impl<S> Stream for AggregatedChatStream<S> impl<S> Stream for AggregatedChatStream<S>
@@ -406,8 +370,8 @@ where
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut(); let this = self.get_mut();
// First, flush any buffered events from the previous call. // First, flush any buffered Completed event from the previous call.
if let Some(ev) = this.pending.pop_front() { if let Some(ev) = this.pending_completed.take() {
return Poll::Ready(Some(Ok(ev))); return Poll::Ready(Some(Ok(ev)));
} }
@@ -424,21 +388,16 @@ where
let is_assistant_delta = matches!(&item, crate::models::ResponseItem::Message { role, .. } if role == "assistant"); let is_assistant_delta = matches!(&item, crate::models::ResponseItem::Message { role, .. } if role == "assistant");
if is_assistant_delta { if is_assistant_delta {
// Only use the final assistant message if we have not if let crate::models::ResponseItem::Message { content, .. } = &item {
// seen any deltas; otherwise, deltas already built the if let Some(text) = content.iter().find_map(|c| match c {
// cumulative text and this would duplicate it. crate::models::ContentItem::OutputText { text } => Some(text),
if this.cumulative.is_empty() { _ => None,
if let crate::models::ResponseItem::Message { content, .. } = &item { }) {
if let Some(text) = content.iter().find_map(|c| match c { this.cumulative.push_str(text);
crate::models::ContentItem::OutputText { text } => Some(text),
_ => None,
}) {
this.cumulative.push_str(text);
}
} }
} }
// Swallow assistant message here; emit on Completed. // Swallow partial assistant chunk; keep polling.
continue; continue;
} }
@@ -449,48 +408,24 @@ where
response_id, response_id,
token_usage, token_usage,
}))) => { }))) => {
// Build any aggregated items in the correct order: Reasoning first, then Message.
let mut emitted_any = false;
if !this.cumulative_reasoning.is_empty() {
let aggregated_reasoning = crate::models::ResponseItem::Reasoning {
id: String::new(),
summary: vec![
crate::models::ReasoningItemReasoningSummary::SummaryText {
text: std::mem::take(&mut this.cumulative_reasoning),
},
],
content: None,
encrypted_content: None,
};
this.pending
.push_back(ResponseEvent::OutputItemDone(aggregated_reasoning));
emitted_any = true;
}
if !this.cumulative.is_empty() { if !this.cumulative.is_empty() {
let aggregated_message = crate::models::ResponseItem::Message { let aggregated_item = crate::models::ResponseItem::Message {
id: None, id: None,
role: "assistant".to_string(), role: "assistant".to_string(),
content: vec![crate::models::ContentItem::OutputText { content: vec![crate::models::ContentItem::OutputText {
text: std::mem::take(&mut this.cumulative), text: std::mem::take(&mut this.cumulative),
}], }],
}; };
this.pending
.push_back(ResponseEvent::OutputItemDone(aggregated_message));
emitted_any = true;
}
// Always emit Completed last when anything was aggregated. // Buffer Completed so it is returned *after* the aggregated message.
if emitted_any { this.pending_completed = Some(ResponseEvent::Completed {
this.pending.push_back(ResponseEvent::Completed { response_id,
response_id: response_id.clone(), token_usage,
token_usage: token_usage.clone(),
}); });
// Return the first pending event now.
if let Some(ev) = this.pending.pop_front() { return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(
return Poll::Ready(Some(Ok(ev))); aggregated_item,
} ))));
} }
// Nothing aggregated – forward Completed directly. // Nothing aggregated – forward Completed directly.
@@ -504,25 +439,11 @@ where
// will never appear in a Chat Completions stream. // will never appear in a Chat Completions stream.
continue; continue;
} }
Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))) => { Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(_))))
// Always accumulate deltas so we can emit a final OutputItemDone at Completed. | Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => {
this.cumulative.push_str(&delta); // Deltas are ignored here since aggregation waits for the
if this.streaming_mode { // final OutputItemDone.
// In streaming mode, also forward the delta immediately. continue;
return Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta))));
} else {
continue;
}
}
Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(delta)))) => {
// Always accumulate reasoning deltas so we can emit a final Reasoning item at Completed.
this.cumulative_reasoning.push_str(&delta);
if this.streaming_mode {
// In streaming mode, also forward the delta immediately.
return Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(delta))));
} else {
continue;
}
} }
} }
} }
@@ -554,23 +475,9 @@ pub(crate) trait AggregateStreamExt: Stream<Item = Result<ResponseEvent>> + Size
AggregatedChatStream { AggregatedChatStream {
inner: self, inner: self,
cumulative: String::new(), cumulative: String::new(),
cumulative_reasoning: String::new(), pending_completed: None,
pending: std::collections::VecDeque::new(),
streaming_mode: false,
} }
} }
} }
impl<T> AggregateStreamExt for T where T: Stream<Item = Result<ResponseEvent>> + Sized {} impl<T> AggregateStreamExt for T where T: Stream<Item = Result<ResponseEvent>> + Sized {}
impl<S> AggregatedChatStream<S> {
pub(crate) fn streaming_mode(inner: S) -> Self {
AggregatedChatStream {
inner,
cumulative: String::new(),
cumulative_reasoning: String::new(),
pending: std::collections::VecDeque::new(),
streaming_mode: true,
}
}
}

8
codex-rs/core/src/client.rs
View File

@@ -93,11 +93,7 @@ impl ModelClient {
// Wrap it with the aggregation adapter so callers see *only* // Wrap it with the aggregation adapter so callers see *only*
// the final assistant message per turn (matching the // the final assistant message per turn (matching the
// behaviour of the Responses API). // behaviour of the Responses API).
let mut aggregated = if !self.config.hide_agent_reasoning { let mut aggregated = response_stream.aggregate();
crate::chat_completions::AggregatedChatStream::streaming_mode(response_stream)
} else {
response_stream.aggregate()
};
// Bridge the aggregated stream back into a standard // Bridge the aggregated stream back into a standard
// `ResponseStream` by forwarding events through a channel. // `ResponseStream` by forwarding events through a channel.
@@ -442,7 +438,7 @@ async fn process_sse<S>(
} }
} }
} }
"response.reasoning_summary_text.delta" | "response.reasoning_text.delta" => { "response.reasoning_summary_text.delta" => {
if let Some(delta) = event.delta { if let Some(delta) = event.delta {
let event = ResponseEvent::ReasoningSummaryDelta(delta); let event = ResponseEvent::ReasoningSummaryDelta(delta);
if tx_event.send(Ok(event)).await.is_err() { if tx_event.send(Ok(event)).await.is_err() {

58
codex-rs/core/src/codex.rs
View File

@@ -56,7 +56,6 @@ use crate::mcp_tool_call::handle_mcp_tool_call;
use crate::models::ContentItem; use crate::models::ContentItem;
use crate::models::FunctionCallOutputPayload; use crate::models::FunctionCallOutputPayload;
use crate::models::LocalShellAction; use crate::models::LocalShellAction;
use crate::models::ReasoningItemContent;
use crate::models::ReasoningItemReasoningSummary; use crate::models::ReasoningItemReasoningSummary;
use crate::models::ResponseInputItem; use crate::models::ResponseInputItem;
use crate::models::ResponseItem; use crate::models::ResponseItem;
@@ -65,7 +64,6 @@ use crate::plan_tool::handle_update_plan;
use crate::project_doc::get_user_instructions; use crate::project_doc::get_user_instructions;
use crate::protocol::AgentMessageDeltaEvent; use crate::protocol::AgentMessageDeltaEvent;
use crate::protocol::AgentMessageEvent; use crate::protocol::AgentMessageEvent;
use crate::protocol::AgentReasoningContentEvent;
use crate::protocol::AgentReasoningDeltaEvent; use crate::protocol::AgentReasoningDeltaEvent;
use crate::protocol::AgentReasoningEvent; use crate::protocol::AgentReasoningEvent;
use crate::protocol::ApplyPatchApprovalRequestEvent; use crate::protocol::ApplyPatchApprovalRequestEvent;
@@ -229,7 +227,6 @@ pub(crate) struct Session {
state: Mutex<State>, state: Mutex<State>,
codex_linux_sandbox_exe: Option<PathBuf>, codex_linux_sandbox_exe: Option<PathBuf>,
user_shell: shell::Shell, user_shell: shell::Shell,
hide_agent_reasoning: bool,
} }
impl Session { impl Session {
@@ -825,7 +822,6 @@ async fn submission_loop(
codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(),
disable_response_storage, disable_response_storage,
user_shell: default_shell, user_shell: default_shell,
hide_agent_reasoning: config.hide_agent_reasoning,
})); }));
// Patch restored state into the newly created session. // Patch restored state into the newly created session.
@@ -1136,7 +1132,6 @@ async fn run_task(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
ResponseItem::Reasoning { ResponseItem::Reasoning {
id, id,
summary, summary,
content,
encrypted_content, encrypted_content,
}, },
None, None,
@@ -1144,7 +1139,6 @@ async fn run_task(sess: Arc<Session>, sub_id: String, input: Vec<InputItem>) {
items_to_record_in_conversation_history.push(ResponseItem::Reasoning { items_to_record_in_conversation_history.push(ResponseItem::Reasoning {
id: id.clone(), id: id.clone(),
summary: summary.clone(), summary: summary.clone(),
content: content.clone(),
encrypted_content: encrypted_content.clone(), encrypted_content: encrypted_content.clone(),
}); });
} }
@@ -1387,13 +1381,11 @@ async fn try_run_turn(
sess.tx_event.send(event).await.ok(); sess.tx_event.send(event).await.ok();
} }
ResponseEvent::ReasoningSummaryDelta(delta) => { ResponseEvent::ReasoningSummaryDelta(delta) => {
if !sess.hide_agent_reasoning { let event = Event {
let event = Event { id: sub_id.to_string(),
id: sub_id.to_string(), msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }),
msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }), };
}; sess.tx_event.send(event).await.ok();
sess.tx_event.send(event).await.ok();
}
} }
} }
} }
@@ -1501,36 +1493,16 @@ async fn handle_response_item(
} }
None None
} }
ResponseItem::Reasoning { ResponseItem::Reasoning { summary, .. } => {
id: _, for item in summary {
summary, let text = match item {
content, ReasoningItemReasoningSummary::SummaryText { text } => text,
encrypted_content: _, };
} => { let event = Event {
if !sess.hide_agent_reasoning { id: sub_id.to_string(),
for item in summary { msg: EventMsg::AgentReasoning(AgentReasoningEvent { text }),
let text = match item { };
ReasoningItemReasoningSummary::SummaryText { text } => text, sess.tx_event.send(event).await.ok();
};
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentReasoning(AgentReasoningEvent { text }),
};
sess.tx_event.send(event).await.ok();
}
}
if !sess.hide_agent_reasoning && content.is_some() {
let content = content.unwrap();
for item in content {
let text = match item {
ReasoningItemContent::ReasoningText { text } => text,
};
let event = Event {
id: sub_id.to_string(),
msg: EventMsg::AgentReasoningContent(AgentReasoningContentEvent { text }),
};
sess.tx_event.send(event).await.ok();
}
} }
None None
} }

6
codex-rs/core/src/config.rs
View File

@@ -488,10 +488,6 @@ impl Config {
Self::get_base_instructions(experimental_instructions_path, &resolved_cwd)?; Self::get_base_instructions(experimental_instructions_path, &resolved_cwd)?;
let base_instructions = base_instructions.or(file_base_instructions); let base_instructions = base_instructions.or(file_base_instructions);
// Resolve hide/show reasoning flags with consistent precedence:
// if hide is true, force show_reasoning_content to false.
let hide_agent_reasoning_val = cfg.hide_agent_reasoning.unwrap_or(false);
let config = Self { let config = Self {
model, model,
model_context_window, model_context_window,
@@ -521,7 +517,7 @@ impl Config {
tui: cfg.tui.unwrap_or_default(), tui: cfg.tui.unwrap_or_default(),
codex_linux_sandbox_exe, codex_linux_sandbox_exe,
hide_agent_reasoning: hide_agent_reasoning_val, hide_agent_reasoning: cfg.hide_agent_reasoning.unwrap_or(false),
model_reasoning_effort: config_profile model_reasoning_effort: config_profile
.model_reasoning_effort .model_reasoning_effort
.or(cfg.model_reasoning_effort) .or(cfg.model_reasoning_effort)

8
codex-rs/core/src/models.rs
View File

@@ -45,8 +45,6 @@ pub enum ResponseItem {
Reasoning { Reasoning {
id: String, id: String,
summary: Vec<ReasoningItemReasoningSummary>, summary: Vec<ReasoningItemReasoningSummary>,
#[serde(default, skip_serializing_if = "Option::is_none")]
content: Option<Vec<ReasoningItemContent>>,
encrypted_content: Option<String>, encrypted_content: Option<String>,
}, },
LocalShellCall { LocalShellCall {
@@ -138,12 +136,6 @@ pub enum ReasoningItemReasoningSummary {
SummaryText { text: String }, SummaryText { text: String },
} }
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ReasoningItemContent {
ReasoningText { text: String },
}
impl From<Vec<InputItem>> for ResponseInputItem { impl From<Vec<InputItem>> for ResponseInputItem {
fn from(items: Vec<InputItem>) -> Self { fn from(items: Vec<InputItem>) -> Self {
Self::Message { Self::Message {

8
codex-rs/core/src/protocol.rs
View File

@@ -359,9 +359,6 @@ pub enum EventMsg {
/// Agent reasoning delta event from agent. /// Agent reasoning delta event from agent.
AgentReasoningDelta(AgentReasoningDeltaEvent), AgentReasoningDelta(AgentReasoningDeltaEvent),
/// Raw chain-of-thought from agent.
AgentReasoningContent(AgentReasoningContentEvent),
/// Ack the client's configure message. /// Ack the client's configure message.
SessionConfigured(SessionConfiguredEvent), SessionConfigured(SessionConfiguredEvent),
@@ -467,11 +464,6 @@ pub struct AgentReasoningEvent {
pub text: String, pub text: String,
} }
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct AgentReasoningContentEvent {
pub text: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)] #[derive(Debug, Clone, Deserialize, Serialize)]
pub struct AgentReasoningDeltaEvent { pub struct AgentReasoningDeltaEvent {
pub delta: String, pub delta: String,

9
codex-rs/exec/src/event_processor_with_human_output.rs
View File

@@ -4,7 +4,6 @@ use codex_core::config::Config;
use codex_core::plan_tool::UpdatePlanArgs; use codex_core::plan_tool::UpdatePlanArgs;
use codex_core::protocol::AgentMessageDeltaEvent; use codex_core::protocol::AgentMessageDeltaEvent;
use codex_core::protocol::AgentMessageEvent; use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::AgentReasoningContentEvent;
use codex_core::protocol::AgentReasoningDeltaEvent; use codex_core::protocol::AgentReasoningDeltaEvent;
use codex_core::protocol::BackgroundEventEvent; use codex_core::protocol::BackgroundEventEvent;
use codex_core::protocol::ErrorEvent; use codex_core::protocol::ErrorEvent;
@@ -204,14 +203,6 @@ impl EventProcessor for EventProcessorWithHumanOutput {
#[allow(clippy::expect_used)] #[allow(clippy::expect_used)]
std::io::stdout().flush().expect("could not flush stdout"); std::io::stdout().flush().expect("could not flush stdout");
} }
EventMsg::AgentReasoningContent(AgentReasoningContentEvent { text }) => {
if !self.show_agent_reasoning {
return CodexStatus::Running;
}
print!("{text}");
#[allow(clippy::expect_used)]
std::io::stdout().flush().expect("could not flush stdout");
}
EventMsg::AgentMessage(AgentMessageEvent { message }) => { EventMsg::AgentMessage(AgentMessageEvent { message }) => {
// if answer_started is false, this means we haven't received any // if answer_started is false, this means we haven't received any
// delta. Thus, we need to print the message as a new answer. // delta. Thus, we need to print the message as a new answer.

3
codex-rs/mcp-server/src/codex_tool_runner.rs
View File

@@ -252,8 +252,7 @@ async fn run_codex_tool_session_inner(
EventMsg::AgentMessage(AgentMessageEvent { .. }) => { EventMsg::AgentMessage(AgentMessageEvent { .. }) => {
// TODO: think how we want to support this in the MCP // TODO: think how we want to support this in the MCP
} }
EventMsg::AgentReasoningContent(_) EventMsg::TaskStarted
| EventMsg::TaskStarted
| EventMsg::TokenCount(_) | EventMsg::TokenCount(_)
| EventMsg::AgentReasoning(_) | EventMsg::AgentReasoning(_)
| EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallBegin(_)

3
codex-rs/mcp-server/src/conversation_loop.rs
View File

@@ -90,8 +90,7 @@ pub async fn run_conversation_loop(
EventMsg::AgentMessage(AgentMessageEvent { .. }) => { EventMsg::AgentMessage(AgentMessageEvent { .. }) => {
// TODO: think how we want to support this in the MCP // TODO: think how we want to support this in the MCP
} }
EventMsg::AgentReasoningContent(_) EventMsg::TaskStarted
| EventMsg::TaskStarted
| EventMsg::TokenCount(_) | EventMsg::TokenCount(_)
| EventMsg::AgentReasoning(_) | EventMsg::AgentReasoning(_)
| EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallBegin(_)