diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index 679b49b5..df976ebe 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -45,15 +45,21 @@ pub struct ParsedLine { /// /// Le flux est du **JSONL**. Types réels : /// -/// - `{"type":"thread.started","thread_id":""}` ⇒ capte le `thread_id` +/// - `{"type":"thread.started","thread_id":""}` / +/// `{"type":"codex.thread.started","thread_id":""}` ⇒ capte le `thread_id` /// (= id de conversation pour la reprise), **aucun** événement émis. /// - `{"type":"turn.started"}` ⇒ [`ReplyEvent::Heartbeat`] (vivacité non terminale). +/// - `{"type":"conversation.item.added","item":{…}}` / +/// `{"type":"item.started","item":{…}}` ⇒ progression non terminale `Started` +/// pour les items outil/MCP dès que Codex annonce leur existence. /// - `{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"…"}}` /// ⇒ si `item.type=="agent_message"` ⇒ [`ReplyEvent::Final`] (`content` = `item.text`, /// 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`). +/// - `{"type":"conversation.item.done","item":{…}}` est traité comme `item.completed` +/// (Codex CLI ≥ 0.147). /// - `{"type":"turn.completed","usage":{…}}` ⇒ [`ReplyEvent::Heartbeat`] (le `Final` /// vient de l'`agent_message`, pas de `turn.completed`). /// @@ -74,7 +80,7 @@ pub fn parse_event(line: &str) -> Result { let mut events = Vec::new(); match value.get("type").and_then(Value::as_str) { - Some("thread.started") => { + Some("thread.started" | "codex.thread.started") => { // Handshake : on ne capte que le thread_id (= id de conversation). conversation_id = value .get("thread_id") @@ -106,15 +112,37 @@ pub fn parse_event(line: &str) -> Result { }); events.push(ReplyEvent::Heartbeat); } - Some("item.completed") => { + Some("item.started" | "conversation.item.added" | "response.output_item.added") => { if let Some(item) = value.get("item") { - match item.get("type").and_then(Value::as_str) { + append_item_progress( + &mut events, + item, + ReplyProgressStage::Started, + value + .get("type") + .and_then(Value::as_str) + .unwrap_or("item.started"), + ); + } + } + Some("item.updated" | "conversation.item.updated") => { + if let Some(item) = value.get("item") { + append_item_progress( + &mut events, + item, + ReplyProgressStage::Delta, + value + .get("type") + .and_then(Value::as_str) + .unwrap_or("item.updated"), + ); + } + } + Some("item.completed" | "conversation.item.done" | "response.output_item.done") => { + if let Some(item) = value.get("item") { + match item_type(item).as_deref() { Some("agent_message") => { - let content = item - .get("text") - .and_then(Value::as_str) - .unwrap_or_default() - .to_owned(); + let content = item_text(item).unwrap_or_default(); events.push(ReplyEvent::Final { content }); } Some("error") => { @@ -124,18 +152,18 @@ pub fn parse_event(line: &str) -> Result { } // reasoning / command / tout autre item ⇒ activité (label = type). Some(kind) => { + let label = item_label(item).unwrap_or_else(|| kind.to_owned()); events.push(ReplyEvent::Progress { - progress: provider_progress( - ReplyProgressKind::Tool, + progress: item_progress( + &label, ReplyProgressStage::Completed, - kind, - "item.completed", - ) - .with_tool_name(kind.to_owned()), - }); - events.push(ReplyEvent::ToolActivity { - label: kind.to_owned(), + value + .get("type") + .and_then(Value::as_str) + .unwrap_or("item.completed"), + ), }); + events.push(ReplyEvent::ToolActivity { label }); } None => {} } @@ -161,6 +189,83 @@ fn provider_progress( .with_provider_event("codex", native_event) } +fn item_progress( + label: &str, + stage: ReplyProgressStage, + native_event: impl Into, +) -> ReplyProgress { + if label.starts_with("idea_") { + ReplyProgress::new( + ReplyProgressSource::IdeaLocal, + ReplyProgressKind::Mcp, + stage, + label.to_owned(), + ) + .with_provider_event("codex", native_event) + .with_tool_name(label.to_owned()) + } else { + provider_progress( + ReplyProgressKind::Tool, + stage, + label.to_owned(), + native_event, + ) + .with_tool_name(label.to_owned()) + } +} + +fn append_item_progress( + events: &mut Vec, + item: &Value, + stage: ReplyProgressStage, + native_event: &str, +) { + let Some(kind) = item_type(item) else { + return; + }; + if matches!(kind.as_str(), "agent_message" | "error") { + return; + } + let label = item_label(item).unwrap_or(kind); + 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 { + non_blank_string(item.get("type")) + .or_else(|| non_blank_string(item.get("item_type"))) + .or_else(|| non_blank_string(item.get("kind"))) +} + +fn item_label(item: &Value) -> Option { + non_blank_string(item.get("name")) + .or_else(|| non_blank_string(item.get("tool_name"))) + .or_else(|| non_blank_string(item.get("tool"))) + .or_else(|| non_blank_string(item.get("action"))) + .or_else(|| nested_non_blank(item, &["call", "name"])) + .or_else(|| nested_non_blank(item, &["function", "name"])) + .or_else(|| nested_non_blank(item, &["function_call", "name"])) + .or_else(|| item_type(item)) +} + +fn item_text(item: &Value) -> Option { + non_blank_string(item.get("text")) + .or_else(|| non_blank_string(item.get("message"))) + .or_else(|| nested_non_blank(item, &["content", "text"])) +} + +fn nested_non_blank(root: &Value, path: &[&str]) -> Option { + let mut cursor = root; + for key in path { + cursor = cursor.get(*key)?; + } + non_blank_string(Some(cursor)) +} + const CODEX_ERROR_FALLBACK: &str = "Codex a renvoyé une erreur."; fn non_blank_string(value: Option<&Value>) -> Option { diff --git a/crates/infrastructure/src/session/mod.rs b/crates/infrastructure/src/session/mod.rs index 90e335cc..0572eda2 100644 --- a/crates/infrastructure/src/session/mod.rs +++ b/crates/infrastructure/src/session/mod.rs @@ -50,7 +50,8 @@ mod tests { use domain::ids::ProfileId; use domain::ports::{ AgentSession, AgentSessionError, AgentSessionFactory, ContextInjectionPlan, - PreparedContext, ReplyEvent, SessionPlan, + PreparedContext, ReplyEvent, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage, + SessionPlan, }; use domain::profile::{AgentProfile, ContextInjection, HttpChatConfig, StructuredAdapter}; use domain::project::ProjectPath; @@ -965,6 +966,88 @@ mod tests { ); } + #[test] + fn codex_conversation_item_added_emits_live_started_progress() { + let r = codex::parse_event( + r#"{"type":"conversation.item.added","item":{"id":"call_1","type":"function_call","name":"read_file","arguments":"{}"}}"#, + ) + .unwrap(); + + assert!( + matches!( + r.events.first(), + Some(ReplyEvent::Progress { progress }) + if progress.source == ReplyProgressSource::ProviderNative + && progress.kind == ReplyProgressKind::Tool + && progress.stage == ReplyProgressStage::Started + && progress.label == "read_file" + && progress.tool_name.as_deref() == Some("read_file") + && progress.native_event.as_deref() == Some("conversation.item.added") + ), + "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" + ); + } + + #[test] + fn codex_idea_tool_is_tagged_as_local_mcp_progress() { + let r = codex::parse_event( + r#"{"type":"conversation.item.added","item":{"id":"call_1","type":"function_call","name":"idea_ask_agent","arguments":"{}"}}"#, + ) + .unwrap(); + + assert!( + matches!( + r.events.first(), + Some(ReplyEvent::Progress { progress }) + if progress.source == ReplyProgressSource::IdeaLocal + && progress.kind == ReplyProgressKind::Mcp + && progress.stage == ReplyProgressStage::Started + && progress.label == "idea_ask_agent" + && progress.tool_name.as_deref() == Some("idea_ask_agent") + && progress.native_event.as_deref() == Some("conversation.item.added") + ), + "Codex MCP tool start must be tagged as IdeaLocal/Mcp: {:?}", + r.events + ); + } + + #[test] + fn codex_conversation_item_done_completes_tool_with_name() { + let r = codex::parse_event( + r#"{"type":"conversation.item.done","item":{"id":"call_1","type":"function_call","name":"idea_ask_agent","status":"completed"}}"#, + ) + .unwrap(); + + assert!( + matches!( + r.events.first(), + Some(ReplyEvent::Progress { progress }) + if progress.source == ReplyProgressSource::IdeaLocal + && progress.kind == ReplyProgressKind::Mcp + && progress.stage == ReplyProgressStage::Completed + && progress.label == "idea_ask_agent" + && progress.tool_name.as_deref() == Some("idea_ask_agent") + && progress.native_event.as_deref() == Some("conversation.item.done") + ), + "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" + ); + } + /// `item.type == error` ⇒ Error terminal visible, avec message réel. #[test] fn codex_error_item_is_error_event_with_real_message() { @@ -1220,6 +1303,7 @@ mod tests { let fake = FakeCli::printing(&[ r#"{"type":"thread.started","thread_id":"c"}"#, r#"{"type":"turn.started"}"#, + 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"}"#, r#"{"type":"item.completed","item":{"id":"i1","type":"agent_message","text":"résultat"}}"#, @@ -1243,6 +1327,17 @@ 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" ); + assert!( + live.iter().any(|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") + )), + "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")