blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(codex): activité tool/MCP émise en live pendant le tour (#58 #93 #171)

Browse Source 
L'adaptateur Codex n'émettait Progress/ToolActivity qu'à item.completed. On capte désormais conversation.item.added / item.started / response.output_item.added (stage Started), item.updated (Delta) et conversation.item.done / response.output_item.done (Completed), en préservant le nom réel de l'outil (name / function.name / function_call.name).

Les outils idea_* sont taggés ReplyProgressSource::IdeaLocal / Kind::Mcp pour les distinguer du tooling provider. Corrige #93 (l'activité MCP idea_ask_agent devient visible avant le Final, nom non corrompu) et #171 (pipeline progress -> UI temps réel) ; base du rendu live #58.

Tests : 3 unitaires (started live, tag IdeaLocal/Mcp, done avec nom) + assertion d'intégration (MCP started tapé avant le Final). QA : verte (live).
This commit is contained in:
Blomios
2026-08-10 16:46:17 +02:00
parent 47bc2c4fb6
commit c5a38df2af
2 changed files with 219 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
crates/infrastructure/src/session/codex.rs
View File

@ -45,15 +45,21 @@ pub struct ParsedLine {
///
/// Le flux est du **JSONL**. Types réels :
///
/// - `{"type":"thread.started","thread_id":"<id>"}` ⇒ capte le `thread_id`
/// - `{"type":"thread.started","thread_id":"<id>"}` /
/// `{"type":"codex.thread.started","thread_id":"<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<ParsedLine, AgentSessionError> {
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<ParsedLine, AgentSessionError> {
});
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<ParsedLine, AgentSessionError> {
}
// 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<String>,
) -> 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<ReplyEvent>,
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<String> {
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<String> {
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<String> {
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<String> {
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<String> {

97
crates/infrastructure/src/session/mod.rs
View File

@ -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")