fix(cli): progress agent_message émis en live + anti-doublon running/done par progressId — #187 (QA verte)
- Codex: les items agent_message sont désormais tapés en live comme progress (Message/Delta) dès item.completed, au lieu d'être bufferisés dans pending_final jusqu'à l'événement tool/MCP suivant. - Corrélation stable progressId (codex:item:<id>) ajoutée sur ReplyProgress, propagée jusqu'au DTO (ReplyProgressDto.progressId) et au frontend ; le ProgressSettler et la vue procèdent par upsert running -> done au lieu d'empiler des lignes dupliquées. - Tests: chat_bridge (settler par progressId), dto_chat, session codex, CustomAgentChatView (upsert par progressId).
This commit is contained in:
@ -196,6 +196,7 @@ fn item_progress(
|
||||
stage: ReplyProgressStage,
|
||||
native_event: impl Into<String>,
|
||||
) -> ReplyProgress {
|
||||
let progress_id = item_progress_id(item);
|
||||
let progress = if label.starts_with("idea_") {
|
||||
ReplyProgress::new(
|
||||
ReplyProgressSource::IdeaLocal,
|
||||
@ -214,6 +215,11 @@ fn item_progress(
|
||||
)
|
||||
.with_tool_name(label.to_owned())
|
||||
};
|
||||
let progress = if let Some(progress_id) = progress_id {
|
||||
progress.with_progress_id(progress_id)
|
||||
} else {
|
||||
progress
|
||||
};
|
||||
if let Some(preview) = item_preview(item) {
|
||||
progress.with_text(preview)
|
||||
} else {
|
||||
@ -221,6 +227,29 @@ fn item_progress(
|
||||
}
|
||||
}
|
||||
|
||||
fn agent_message_progress(item: &Value, native_event: &str) -> Option<ReplyEvent> {
|
||||
let content = item_text(item)?;
|
||||
let mut progress = ReplyProgress::new(
|
||||
ReplyProgressSource::ProviderNative,
|
||||
ReplyProgressKind::Message,
|
||||
ReplyProgressStage::Delta,
|
||||
"message intermédiaire",
|
||||
)
|
||||
.with_provider_event("codex", native_event)
|
||||
.with_text(content);
|
||||
if let Some(progress_id) = item_progress_id(item) {
|
||||
progress = progress.with_progress_id(progress_id);
|
||||
}
|
||||
Some(ReplyEvent::Progress { progress })
|
||||
}
|
||||
|
||||
fn item_progress_id(item: &Value) -> Option<String> {
|
||||
non_blank_string(item.get("id"))
|
||||
.or_else(|| non_blank_string(item.get("item_id")))
|
||||
.or_else(|| non_blank_string(item.get("call_id")))
|
||||
.map(|id| format!("codex:item:{id}"))
|
||||
}
|
||||
|
||||
fn append_item_progress(
|
||||
events: &mut Vec<ReplyEvent>,
|
||||
item: &Value,
|
||||
@ -580,20 +609,18 @@ impl CodexExecSession {
|
||||
let parser = std::thread::spawn(move || {
|
||||
let mut pending_final: Option<String> = None;
|
||||
for line in line_rx {
|
||||
if let Ok(Some(progress)) = agent_message_progress_from_line(&line) {
|
||||
let _ = event_tap.send(progress);
|
||||
}
|
||||
if let Ok(parsed) = parse_event(&line) {
|
||||
for event in parsed.events {
|
||||
match event {
|
||||
ReplyEvent::Final { content } => {
|
||||
if let Some(text) = pending_final.replace(content) {
|
||||
let _ = event_tap.send(ReplyEvent::Announcement { text });
|
||||
}
|
||||
let _ = pending_final.replace(content);
|
||||
}
|
||||
other => {
|
||||
if event_supersedes_pending_final(&other) {
|
||||
if let Some(text) = pending_final.take() {
|
||||
let _ =
|
||||
event_tap.send(ReplyEvent::Announcement { text });
|
||||
}
|
||||
let _ = pending_final.take();
|
||||
}
|
||||
let _ = event_tap.send(other);
|
||||
}
|
||||
@ -684,3 +711,30 @@ fn event_supersedes_pending_final(event: &ReplyEvent) -> bool {
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
fn agent_message_progress_from_line(line: &str) -> Result<Option<ReplyEvent>, AgentSessionError> {
|
||||
let trimmed = line.trim();
|
||||
if trimmed.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
let value: Value = serde_json::from_str(trimmed)
|
||||
.map_err(|e| AgentSessionError::Decode(format!("ligne JSON illisible: {e}")))?;
|
||||
let Some("item.completed" | "conversation.item.done" | "response.output_item.done") =
|
||||
value.get("type").and_then(Value::as_str)
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
let Some(item) = value.get("item") else {
|
||||
return Ok(None);
|
||||
};
|
||||
if item_type(item).as_deref() != Some("agent_message") {
|
||||
return Ok(None);
|
||||
}
|
||||
Ok(agent_message_progress(
|
||||
item,
|
||||
value
|
||||
.get("type")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("item.completed"),
|
||||
))
|
||||
}
|
||||
|
||||
@ -1016,6 +1016,7 @@ mod tests {
|
||||
&& progress.stage == ReplyProgressStage::Completed
|
||||
&& progress.label == "reasoning"
|
||||
&& progress.tool_name.as_deref() == Some("reasoning")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:i0")
|
||||
),
|
||||
"reasoning item must be represented by one canonical progress event: {:?}",
|
||||
r.events
|
||||
@ -1032,6 +1033,7 @@ mod tests {
|
||||
&& progress.stage == ReplyProgressStage::Completed
|
||||
&& progress.label == "command"
|
||||
&& progress.tool_name.as_deref() == Some("command")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:i1")
|
||||
),
|
||||
"command item must be represented by one canonical progress event: {:?}",
|
||||
c.events
|
||||
@ -1055,6 +1057,7 @@ mod tests {
|
||||
&& progress.label == "read_file"
|
||||
&& progress.tool_name.as_deref() == Some("read_file")
|
||||
&& progress.native_event.as_deref() == Some("conversation.item.added")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:call_1")
|
||||
),
|
||||
"Codex item.added must surface tool start immediately: {:?}",
|
||||
r.events
|
||||
@ -1083,6 +1086,7 @@ mod tests {
|
||||
&& progress.label == "idea_ask_agent"
|
||||
&& progress.tool_name.as_deref() == Some("idea_ask_agent")
|
||||
&& progress.native_event.as_deref() == Some("conversation.item.added")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:call_1")
|
||||
),
|
||||
"Codex MCP tool start must be tagged as IdeaLocal/Mcp: {:?}",
|
||||
r.events
|
||||
@ -1106,6 +1110,7 @@ mod tests {
|
||||
&& progress.label == "idea_ask_agent"
|
||||
&& progress.tool_name.as_deref() == Some("idea_ask_agent")
|
||||
&& progress.native_event.as_deref() == Some("conversation.item.done")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:call_1")
|
||||
),
|
||||
"Codex MCP tool completion must keep the real tool name: {:?}",
|
||||
r.events
|
||||
@ -1436,12 +1441,20 @@ 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
|
||||
let message_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");
|
||||
.position(|event| {
|
||||
matches!(
|
||||
event,
|
||||
ReplyEvent::Progress { progress }
|
||||
if progress.source == ReplyProgressSource::ProviderNative
|
||||
&& progress.kind == ReplyProgressKind::Message
|
||||
&& progress.stage == ReplyProgressStage::Delta
|
||||
&& progress.text.as_deref() == Some("je délègue")
|
||||
&& progress.progress_id.as_deref() == Some("codex:item:i0")
|
||||
)
|
||||
})
|
||||
.expect("le message agent doit être tapé live comme progress");
|
||||
let mcp_idx = live
|
||||
.iter()
|
||||
.position(|event| {
|
||||
@ -1456,8 +1469,8 @@ mod tests {
|
||||
})
|
||||
.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: {:?}",
|
||||
message_idx < mcp_idx,
|
||||
"le message agent doit précéder le premier MCP dans le tap live: {:?}",
|
||||
live
|
||||
);
|
||||
assert!(
|
||||
|
||||
Reference in New Issue
Block a user