Un agent_message d'intention arrivant avant un tool (ex idea_ask_agent) n'est plus avalé comme faux candidat Final: tamponné puis reclassé en Announcement dès qu'un événement le détrône (helper event_supersedes_pending_final), sur le tap live comme sur le flux autoritatif. Les items reasoning/command n'émettent plus de ToolActivity de compatibilité en doublon du Progress canonique Completed -> supprime les doubles cartes tool/MCP à la source. Tests mod.rs alignés + nouveau cas agent_message_before_tool_becomes_announcement_before_tool_progress.
This commit is contained in:
@ -57,7 +57,8 @@ pub struct ParsedLine {
|
||||
/// c'est la réponse) ; si `item.type=="error"` ⇒ [`ReplyEvent::Error`] avec message
|
||||
/// extrait de `item.text`, `item.message`, `item.error.message`, `item.error`, puis
|
||||
/// des champs top-level `stderr`/`message`/`error.message`/`error` ; sinon
|
||||
/// (`reasoning`/`command`/autre) ⇒ [`ReplyEvent::ToolActivity`] (`label` = `item.type`).
|
||||
/// (`reasoning`/`command`/autre) ⇒ progression canonique `Completed` (`label` =
|
||||
/// `item.type`).
|
||||
/// - `{"type":"conversation.item.done","item":{…}}` est traité comme `item.completed`
|
||||
/// (Codex CLI ≥ 0.147).
|
||||
/// - `{"type":"turn.completed","usage":{…}}` ⇒ [`ReplyEvent::Heartbeat`] (le `Final`
|
||||
@ -150,7 +151,7 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
|
||||
message: codex_error_message(&value, item),
|
||||
});
|
||||
}
|
||||
// reasoning / command / tout autre item ⇒ activité (label = type).
|
||||
// reasoning / command / tout autre item ⇒ progression canonique.
|
||||
Some(kind) => {
|
||||
let label = item_label(item).unwrap_or_else(|| kind.to_owned());
|
||||
events.push(ReplyEvent::Progress {
|
||||
@ -163,7 +164,6 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
|
||||
.unwrap_or("item.completed"),
|
||||
),
|
||||
});
|
||||
events.push(ReplyEvent::ToolActivity { label });
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
@ -230,9 +230,6 @@ fn append_item_progress(
|
||||
events.push(ReplyEvent::Progress {
|
||||
progress: item_progress(&label, stage, native_event),
|
||||
});
|
||||
if stage == ReplyProgressStage::Started {
|
||||
events.push(ReplyEvent::ToolActivity { label });
|
||||
}
|
||||
}
|
||||
|
||||
fn item_type(item: &Value) -> Option<String> {
|
||||
@ -554,11 +551,25 @@ impl CodexExecSession {
|
||||
let (line_tap, parser_thread) = tap.map_or((None, None), |event_tap| {
|
||||
let (line_tx, line_rx) = std::sync::mpsc::channel::<String>();
|
||||
let parser = std::thread::spawn(move || {
|
||||
let mut pending_final: Option<String> = None;
|
||||
for line in line_rx {
|
||||
if let Ok(parsed) = parse_event(&line) {
|
||||
for event in parsed.events {
|
||||
if !matches!(event, ReplyEvent::Final { .. }) {
|
||||
let _ = event_tap.send(event);
|
||||
match event {
|
||||
ReplyEvent::Final { content } => {
|
||||
if let Some(text) = pending_final.replace(content) {
|
||||
let _ = event_tap.send(ReplyEvent::Announcement { text });
|
||||
}
|
||||
}
|
||||
other => {
|
||||
if event_supersedes_pending_final(&other) {
|
||||
if let Some(text) = pending_final.take() {
|
||||
let _ =
|
||||
event_tap.send(ReplyEvent::Announcement { text });
|
||||
}
|
||||
}
|
||||
let _ = event_tap.send(other);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -597,11 +608,22 @@ impl CodexExecSession {
|
||||
events.push(ReplyEvent::Announcement {
|
||||
text: last_final.take().expect("présent"),
|
||||
});
|
||||
} else {
|
||||
let _ = last_final.take();
|
||||
}
|
||||
}
|
||||
last_final = Some(content);
|
||||
}
|
||||
other => {
|
||||
if last_final.is_some() && event_supersedes_pending_final(&other) {
|
||||
if !has_tap {
|
||||
events.push(ReplyEvent::Announcement {
|
||||
text: last_final.take().expect("présent"),
|
||||
});
|
||||
} else {
|
||||
let _ = last_final.take();
|
||||
}
|
||||
}
|
||||
if !has_tap {
|
||||
events.push(other);
|
||||
}
|
||||
@ -621,3 +643,17 @@ impl CodexExecSession {
|
||||
Ok(Box::new(events.into_iter()))
|
||||
}
|
||||
}
|
||||
|
||||
fn event_supersedes_pending_final(event: &ReplyEvent) -> bool {
|
||||
!matches!(
|
||||
event,
|
||||
ReplyEvent::Heartbeat
|
||||
| ReplyEvent::RateLimited { .. }
|
||||
| ReplyEvent::Progress {
|
||||
progress: ReplyProgress {
|
||||
kind: ReplyProgressKind::Turn,
|
||||
..
|
||||
}
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
@ -942,27 +942,43 @@ mod tests {
|
||||
|
||||
// ---- Codex parse_event (format RÉEL vérifié 2026-06-09) -------------
|
||||
|
||||
/// Un item non-`agent_message` (reasoning/command/…) ⇒ ToolActivity (label = type).
|
||||
/// Un item non-`agent_message` (reasoning/command/…) ⇒ Progress canonique
|
||||
/// (label = type), sans ToolActivity de compatibilité qui créerait une double
|
||||
/// carte UI.
|
||||
#[test]
|
||||
fn codex_non_agent_message_item_is_tool_activity() {
|
||||
fn codex_non_agent_message_item_is_progress_only() {
|
||||
let r = codex::parse_event(
|
||||
r#"{"type":"item.completed","item":{"id":"i0","type":"reasoning","text":"…"}}"#,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
without_progress(r.events),
|
||||
vec![ReplyEvent::ToolActivity {
|
||||
label: "reasoning".into()
|
||||
}]
|
||||
assert!(
|
||||
matches!(
|
||||
r.events.as_slice(),
|
||||
[ReplyEvent::Progress { progress }]
|
||||
if progress.source == ReplyProgressSource::ProviderNative
|
||||
&& progress.kind == ReplyProgressKind::Tool
|
||||
&& progress.stage == ReplyProgressStage::Completed
|
||||
&& progress.label == "reasoning"
|
||||
&& progress.tool_name.as_deref() == Some("reasoning")
|
||||
),
|
||||
"reasoning item must be represented by one canonical progress event: {:?}",
|
||||
r.events
|
||||
);
|
||||
let c =
|
||||
codex::parse_event(r#"{"type":"item.completed","item":{"id":"i1","type":"command"}}"#)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
without_progress(c.events),
|
||||
vec![ReplyEvent::ToolActivity {
|
||||
label: "command".into()
|
||||
}]
|
||||
assert!(
|
||||
matches!(
|
||||
c.events.as_slice(),
|
||||
[ReplyEvent::Progress { progress }]
|
||||
if progress.source == ReplyProgressSource::ProviderNative
|
||||
&& progress.kind == ReplyProgressKind::Tool
|
||||
&& progress.stage == ReplyProgressStage::Completed
|
||||
&& progress.label == "command"
|
||||
&& progress.tool_name.as_deref() == Some("command")
|
||||
),
|
||||
"command item must be represented by one canonical progress event: {:?}",
|
||||
c.events
|
||||
);
|
||||
}
|
||||
|
||||
@ -987,11 +1003,10 @@ mod tests {
|
||||
"Codex item.added must surface tool start immediately: {:?}",
|
||||
r.events
|
||||
);
|
||||
assert!(
|
||||
r.events.iter().any(
|
||||
|event| matches!(event, ReplyEvent::ToolActivity { label } if label == "read_file")
|
||||
),
|
||||
"tool activity remains available for existing UI mapping"
|
||||
assert_eq!(
|
||||
r.events.len(),
|
||||
1,
|
||||
"tool start must not emit a duplicate ToolActivity compatibility event"
|
||||
);
|
||||
}
|
||||
|
||||
@ -1039,12 +1054,10 @@ mod tests {
|
||||
"Codex MCP tool completion must keep the real tool name: {:?}",
|
||||
r.events
|
||||
);
|
||||
assert!(
|
||||
matches!(
|
||||
r.events.get(1),
|
||||
Some(ReplyEvent::ToolActivity { label }) if label == "idea_ask_agent"
|
||||
),
|
||||
"completion still emits ToolActivity compatibility event"
|
||||
assert_eq!(
|
||||
r.events.len(),
|
||||
1,
|
||||
"MCP tool completion must not emit a duplicate ToolActivity compatibility event"
|
||||
);
|
||||
}
|
||||
|
||||
@ -1246,6 +1259,45 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn codex_agent_message_before_tool_becomes_announcement_before_tool_progress() {
|
||||
let fake = FakeCli::printing(&[
|
||||
r#"{"type":"thread.started","thread_id":"c"}"#,
|
||||
r#"{"type":"item.completed","item":{"id":"i0","type":"agent_message","text":"je délègue"}}"#,
|
||||
r#"{"type":"conversation.item.added","item":{"id":"tool0","type":"function_call","name":"idea_ask_agent","arguments":"{}"}}"#,
|
||||
r#"{"type":"item.completed","item":{"id":"i1","type":"agent_message","text":"résultat"}}"#,
|
||||
]);
|
||||
let s = CodexExecSession::new(
|
||||
SessionId::new_random(),
|
||||
fake.command(),
|
||||
"/",
|
||||
None,
|
||||
Vec::new(),
|
||||
Vec::new(),
|
||||
None,
|
||||
None,
|
||||
);
|
||||
|
||||
let events: Vec<_> = s.send("x").await.expect("send").collect();
|
||||
assert!(
|
||||
matches!(
|
||||
events.as_slice(),
|
||||
[
|
||||
ReplyEvent::Announcement { text },
|
||||
ReplyEvent::Progress { progress },
|
||||
ReplyEvent::Final { content }
|
||||
] if text == "je délègue"
|
||||
&& progress.source == ReplyProgressSource::IdeaLocal
|
||||
&& progress.kind == ReplyProgressKind::Mcp
|
||||
&& progress.stage == ReplyProgressStage::Started
|
||||
&& progress.tool_name.as_deref() == Some("idea_ask_agent")
|
||||
&& content == "résultat"
|
||||
),
|
||||
"l'annonce d'intention doit précéder le premier MCP dans le flux: {:?}",
|
||||
events
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn codex_single_agent_message_is_final_without_announcement() {
|
||||
let fake = FakeCli::printing(&[
|
||||
@ -1303,6 +1355,7 @@ mod tests {
|
||||
let fake = FakeCli::printing(&[
|
||||
r#"{"type":"thread.started","thread_id":"c"}"#,
|
||||
r#"{"type":"turn.started"}"#,
|
||||
r#"{"type":"item.completed","item":{"id":"i0","type":"agent_message","text":"je délègue"}}"#,
|
||||
r#"{"type":"conversation.item.added","item":{"id":"tool0","type":"function_call","name":"idea_ask_agent","arguments":"{}"}}"#,
|
||||
r#"{"type":"item.completed","item":{"id":"tool0","type":"command","text":"cargo test"}}"#,
|
||||
r#"{"type":"turn.completed"}"#,
|
||||
@ -1327,6 +1380,30 @@ mod tests {
|
||||
matches!(live.first(), Some(ReplyEvent::Progress { progress }) if progress.native_event.as_deref() == Some("turn.started")),
|
||||
"le démarrage du tour provider doit être tapé live en premier"
|
||||
);
|
||||
let announcement_idx = live
|
||||
.iter()
|
||||
.position(
|
||||
|event| matches!(event, ReplyEvent::Announcement { text } if text == "je délègue"),
|
||||
)
|
||||
.expect("l'annonce d'intention doit être tapée live");
|
||||
let mcp_idx = live
|
||||
.iter()
|
||||
.position(|event| {
|
||||
matches!(
|
||||
event,
|
||||
ReplyEvent::Progress { progress }
|
||||
if progress.source == ReplyProgressSource::IdeaLocal
|
||||
&& progress.kind == ReplyProgressKind::Mcp
|
||||
&& progress.stage == ReplyProgressStage::Started
|
||||
&& progress.tool_name.as_deref() == Some("idea_ask_agent")
|
||||
)
|
||||
})
|
||||
.expect("le début MCP doit être tapé live");
|
||||
assert!(
|
||||
announcement_idx < mcp_idx,
|
||||
"l'annonce d'intention doit précéder le premier MCP dans le tap live: {:?}",
|
||||
live
|
||||
);
|
||||
assert!(
|
||||
live.iter().any(|event| matches!(
|
||||
event,
|
||||
@ -1339,10 +1416,15 @@ mod tests {
|
||||
"le début MCP Codex doit être tapé live avant le Final"
|
||||
);
|
||||
assert!(
|
||||
live.iter().any(
|
||||
|event| matches!(event, ReplyEvent::ToolActivity { label } if label == "command")
|
||||
),
|
||||
"l'activité outil provider doit être tapée live"
|
||||
live.iter().any(|event| matches!(
|
||||
event,
|
||||
ReplyEvent::Progress { progress }
|
||||
if progress.source == ReplyProgressSource::ProviderNative
|
||||
&& progress.kind == ReplyProgressKind::Tool
|
||||
&& progress.stage == ReplyProgressStage::Completed
|
||||
&& progress.tool_name.as_deref() == Some("command")
|
||||
)),
|
||||
"l'activité outil provider doit être tapée live comme progression canonique"
|
||||
);
|
||||
assert!(
|
||||
live.iter().any(|event| matches!(event, ReplyEvent::Progress { progress } if progress.native_event.as_deref() == Some("turn.completed"))),
|
||||
|
||||
Reference in New Issue
Block a user