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

fix(core/reasoning): separate reasoning summary streams by summary_index and raw content by content_index; include index in stream id (item_id#s<idx>/#c<idx>) and de-dup keys; propagate indices through events

Browse Source
This commit is contained in:
James Peter
2025-09-04 08:50:19 +10:00
parent c25146b596
commit d362d096e9
4 changed files with 26 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/chat_completions.rs
+4
View File
@@ -586,6 +586,7 @@ async fn process_chat_sse<S>(
item_id: current_item_id.clone(),
sequence_number: None,
output_index: None,
content_index: None,
}))
.await;
}
@@ -604,6 +605,7 @@ async fn process_chat_sse<S>(
item_id: current_item_id.clone(),
sequence_number: None,
output_index: None,
content_index: None,
}))
.await;
}
@@ -621,6 +623,7 @@ async fn process_chat_sse<S>(
item_id: current_item_id.clone(),
sequence_number: None,
output_index: None,
content_index: None,
}))
.await;
}
@@ -923,6 +926,7 @@ where
item_id,
sequence_number,
output_index: None,
content_index: None,
})));
} else {
continue;
codex-rs/core/src/client.rs
+10 -6
View File
@@ -465,6 +465,7 @@ struct SseEvent {
sequence_number: Option<u64>,
output_index: Option<u32>,
content_index: Option<u32>,
summary_index: Option<u32>,
}
#[derive(Debug, Deserialize)]
@@ -530,10 +531,10 @@ async fn process_sse<S>(
use std::collections::HashMap;
// Track last sequence_number per (item_id, output_index[, content_index])
// Default indices to 0 when absent for robustness across providers.
let mut last_seq_reasoning_summary: HashMap<(String, u32), u64> = HashMap::new();
let mut last_seq_reasoning_summary: HashMap<(String, u32, u32), u64> = HashMap::new();
let mut last_seq_reasoning_content: HashMap<(String, u32, u32), u64> = HashMap::new();
// Best-effort duplicate text guard when sequence_number is unavailable.
let mut last_text_reasoning_summary: HashMap<(String, u32), String> = HashMap::new();
let mut last_text_reasoning_summary: HashMap<(String, u32, u32), String> = HashMap::new();
let mut last_text_reasoning_content: HashMap<(String, u32, u32), String> = HashMap::new();
loop {
@@ -698,15 +699,16 @@ async fn process_sse<S>(
}
// Compose key using item_id + output_index
let out_idx: u32 = event.output_index.unwrap_or(0);
let sum_idx: u32 = event.summary_index.unwrap_or(0);
if let Some(ref id) = current_item_id {
// Drop duplicates/out‑of‑order by sequence_number when available
if let Some(sn) = event.sequence_number {
let last = last_seq_reasoning_summary.entry((id.clone(), out_idx)).or_insert(0);
let last = last_seq_reasoning_summary.entry((id.clone(), out_idx, sum_idx)).or_insert(0);
if *last >= sn { continue; }
*last = sn;
} else {
// Best-effort: drop exact duplicate text for same key when seq is missing
let key = (id.clone(), out_idx);
let key = (id.clone(), out_idx, sum_idx);
if last_text_reasoning_summary.get(&key).map_or(false, |prev| prev == &delta) {
continue;
}
@@ -714,8 +716,8 @@ async fn process_sse<S>(
}
}
tracing::debug!(
"sse.delta reasoning_summary id={:?} out_idx={} len={} seq={:?}",
current_item_id, out_idx,
"sse.delta reasoning_summary id={:?} out_idx={} sum_idx={} len={} seq={:?}",
current_item_id, out_idx, sum_idx,
delta.len(),
event.sequence_number
);
@@ -724,6 +726,7 @@ async fn process_sse<S>(
item_id: event.item_id.or_else(|| current_item_id.clone()),
sequence_number: event.sequence_number,
output_index: event.output_index,
summary_index: event.summary_index,
};
if tx_event.send(Ok(ev)).await.is_err() {
return;
@@ -764,6 +767,7 @@ async fn process_sse<S>(
item_id: event.item_id.or_else(|| current_item_id.clone()),
sequence_number: event.sequence_number,
output_index: event.output_index,
content_index: event.content_index,
};
if tx_event.send(Ok(ev)).await.is_err() {
return;
codex-rs/core/src/client_common.rs
+2
View File
@@ -183,12 +183,14 @@ pub enum ResponseEvent {
item_id: Option<String>,
sequence_number: Option<u64>,
output_index: Option<u32>,
summary_index: Option<u32>,
},
ReasoningContentDelta {
delta: String,
item_id: Option<String>,
sequence_number: Option<u64>,
output_index: Option<u32>,
content_index: Option<u32>,
},
ReasoningSummaryPartAdded,
WebSearchCallBegin {
codex-rs/core/src/codex.rs
+10 -7
View File
@@ -1546,7 +1546,7 @@ async fn submission_loop(
let mut restored_items: Option<Vec<ResponseItem>> = None;
let rollout_recorder: Option<RolloutRecorder> =
if let Some(path) = resume_path.as_ref() {
match RolloutRecorder::resume(path, cwd.clone()).await {
match RolloutRecorder::resume(&config, path).await {
Ok((rec, saved)) => {
session_id = saved.session_id;
if !saved.items.is_empty() {
@@ -2440,9 +2440,10 @@ async fn try_run_turn(
// We deliberately do not scope by item_id to keep implementation simple.
sess.scratchpad_add_text_delta(&delta);
}
ResponseEvent::ReasoningSummaryDelta { delta, item_id, sequence_number, output_index } => {
ResponseEvent::ReasoningSummaryDelta { delta, item_id, sequence_number, output_index, summary_index } => {
// 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 mut event_id = item_id.unwrap_or_else(|| sub_id.to_string());
if let Some(si) = summary_index { event_id = format!("{}#s{}", event_id, si); }
let order = crate::protocol::OrderMeta { request_ordinal: sess.current_request_ordinal(), output_index, sequence_number };
let stamped = sess.make_event_with_order(&event_id, EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta: delta.clone() }), order, sequence_number);
sess.tx_event.send(stamped).await.ok();
@@ -2454,10 +2455,11 @@ async fn try_run_turn(
let stamped = sess.make_event(&sub_id, EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent {}));
sess.tx_event.send(stamped).await.ok();
}
ResponseEvent::ReasoningContentDelta { delta, item_id, sequence_number, output_index } => {
ResponseEvent::ReasoningContentDelta { delta, item_id, sequence_number, output_index, content_index } => {
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 mut event_id = item_id.unwrap_or_else(|| sub_id.to_string());
if let Some(ci) = content_index { event_id = format!("{}#c{}", event_id, ci); }
let order = crate::protocol::OrderMeta { request_ordinal: sess.current_request_ordinal(), output_index, sequence_number };
let stamped = sess.make_event_with_order(&event_id, EventMsg::AgentReasoningRawContentDelta(AgentReasoningRawContentDeltaEvent { delta }), order, sequence_number);
sess.tx_event.send(stamped).await.ok();
@@ -2586,12 +2588,13 @@ async fn handle_response_item(
} else {
sub_id.to_string()
};
for item in summary {
for (i, item) in summary.into_iter().enumerate() {
let text = match item {
ReasoningItemReasoningSummary::SummaryText { text } => text,
};
let eid = format!("{}#s{}", event_id, i);
let order = crate::protocol::OrderMeta { request_ordinal: sess.current_request_ordinal(), output_index, sequence_number: seq_hint };
let stamped = sess.make_event_with_order(&event_id, EventMsg::AgentReasoning(AgentReasoningEvent { text }), order, seq_hint);
let stamped = sess.make_event_with_order(&eid, EventMsg::AgentReasoning(AgentReasoningEvent { text }), order, seq_hint);
sess.tx_event.send(stamped).await.ok();
}
if sess.show_raw_agent_reasoning && content.is_some() {