fix(opencode): streaming progress + bulle MCP pour la CLI custom (#157)

Override send_with_tap: progress/text/tool désormais projetés en live via LiveTurnProjector (au lieu d'un batch en EOF), en suivant le pattern de codex.rs.

Tagging des tool_use idea_* en Progress Mcp/IdeaLocal/Completed (tool_progress) pour faire apparaître la bulle MCP IdeA locale au moment de l'invocation.

Protection anti-replay de session continuée: hold_until_new_message + buffer pending n'émet que le message courant, jamais l'historique rejoué.

send() batch inchangé: stream_events ne filtre le Final que lorsqu'un tap est présent.

crates/infrastructure/src/session/opencode.rs

@@ -184,58 +184,73 @@ fn parse_jsonl_turn_scoped(
     expected_session_id: Option<&str>,
 ) -> Result<Vec<ReplyEvent>, AgentSessionError> {
     let records = current_turn_records(lines, expected_session_id)?;
+    events_from_records(records)
+}
+
+fn events_from_records(records: Vec<ParsedRecord>) -> Result<Vec<ReplyEvent>, AgentSessionError> {
     let mut events = Vec::new();
     let mut final_text = String::new();
     let mut error_seen = false;
     for record in records {
-        match record.event {
+        append_record_events(record, &mut events, &mut final_text, &mut error_seen);
-            ParsedEvent::StepStart => {
-                events.push(ReplyEvent::Progress {
-                    progress: provider_progress(
-                        ReplyProgressKind::Turn,
-                        ReplyProgressStage::Started,
-                        "étape démarrée",
-                        "step_start",
-                    ),
-                });
-                events.push(ReplyEvent::Heartbeat);
-            }
-            ParsedEvent::StepFinish => {
-                events.push(ReplyEvent::Progress {
-                    progress: provider_progress(
-                        ReplyProgressKind::Turn,
-                        ReplyProgressStage::Completed,
-                        "étape terminée",
-                        "step_finish",
-                    ),
-                });
-                events.push(ReplyEvent::Heartbeat);
-            }
-            ParsedEvent::ToolActivity(label) => {
-                events.push(ReplyEvent::Progress {
-                    progress: provider_progress(
-                        ReplyProgressKind::Tool,
-                        ReplyProgressStage::Started,
-                        label.clone(),
-                        "tool_use",
-                    )
-                    .with_tool_name(label.clone()),
-                });
-                events.push(ReplyEvent::ToolActivity { label });
-            }
-            ParsedEvent::Text(text) => {
-                if !text.is_empty() {
-                    final_text.push_str(&text);
-                    events.push(ReplyEvent::TextDelta { text });
-                }
-            }
-            ParsedEvent::Error(message) => {
-                error_seen = true;
-                events.push(ReplyEvent::Error { message });
-            }
-            ParsedEvent::Ignored => {}
-        }
     }
+    append_final_event(events, final_text, error_seen)
+}
+
+fn append_record_events(
+    record: ParsedRecord,
+    events: &mut Vec<ReplyEvent>,
+    final_text: &mut String,
+    error_seen: &mut bool,
+) {
+    match record.event {
+        ParsedEvent::StepStart => {
+            events.push(ReplyEvent::Progress {
+                progress: provider_progress(
+                    ReplyProgressKind::Turn,
+                    ReplyProgressStage::Started,
+                    "étape démarrée",
+                    "step_start",
+                ),
+            });
+            events.push(ReplyEvent::Heartbeat);
+        }
+        ParsedEvent::StepFinish => {
+            events.push(ReplyEvent::Progress {
+                progress: provider_progress(
+                    ReplyProgressKind::Turn,
+                    ReplyProgressStage::Completed,
+                    "étape terminée",
+                    "step_finish",
+                ),
+            });
+            events.push(ReplyEvent::Heartbeat);
+        }
+        ParsedEvent::ToolActivity(label) => {
+            events.push(ReplyEvent::Progress {
+                progress: tool_progress(&label),
+            });
+            events.push(ReplyEvent::ToolActivity { label });
+        }
+        ParsedEvent::Text(text) => {
+            if !text.is_empty() {
+                final_text.push_str(&text);
+                events.push(ReplyEvent::TextDelta { text });
+            }
+        }
+        ParsedEvent::Error(message) => {
+            *error_seen = true;
+            events.push(ReplyEvent::Error { message });
+        }
+        ParsedEvent::Ignored => {}
+    }
+}
+
+fn append_final_event(
+    mut events: Vec<ReplyEvent>,
+    final_text: String,
+    error_seen: bool,
+) -> Result<Vec<ReplyEvent>, AgentSessionError> {
     if final_text.trim().is_empty() {
         if error_seen {
             return Ok(events);
@@ -260,6 +275,106 @@ fn provider_progress(
         .with_provider_event("opencode", native_event)
 }
 
+fn tool_progress(label: &str) -> ReplyProgress {
+    if label.starts_with("idea_") {
+        ReplyProgress::new(
+            ReplyProgressSource::IdeaLocal,
+            ReplyProgressKind::Mcp,
+            ReplyProgressStage::Completed,
+            label.to_owned(),
+        )
+        .with_tool_name(label.to_owned())
+    } else {
+        provider_progress(
+            ReplyProgressKind::Tool,
+            ReplyProgressStage::Started,
+            label.to_owned(),
+            "tool_use",
+        )
+        .with_tool_name(label.to_owned())
+    }
+}
+
+fn send_tap(tap: &std::sync::mpsc::Sender<ReplyEvent>, event: ReplyEvent) {
+    if !matches!(event, ReplyEvent::Final { .. }) {
+        let _ = tap.send(event);
+    }
+}
+
+#[derive(Default)]
+struct LiveTurnProjector {
+    expected_session_id: Option<String>,
+    active_message_id: Option<String>,
+    seen_message_id: bool,
+    hold_until_new_message: bool,
+    pending: Vec<ParsedRecord>,
+    ready: Vec<ReplyEvent>,
+    final_text: String,
+    error_seen: bool,
+}
+
+impl LiveTurnProjector {
+    fn new(expected_session_id: Option<String>) -> Self {
+        Self {
+            hold_until_new_message: expected_session_id.is_some(),
+            expected_session_id,
+            ..Self::default()
+        }
+    }
+
+    fn ingest(&mut self, record: ParsedRecord) -> Vec<ReplyEvent> {
+        if let (Some(expected), Some(actual)) = (
+            self.expected_session_id.as_deref(),
+            record.session_id.as_deref(),
+        ) {
+            if actual != expected {
+                return Vec::new();
+            }
+        }
+        if matches!(record.event, ParsedEvent::Ignored) {
+            return Vec::new();
+        }
+
+        if let Some(message_id) = record.message_id.clone() {
+            if self.active_message_id.as_deref() != Some(message_id.as_str()) {
+                if self.seen_message_id {
+                    self.pending.clear();
+                    self.hold_until_new_message = false;
+                }
+                self.active_message_id = Some(message_id);
+                self.seen_message_id = true;
+            }
+        }
+
+        if self.hold_until_new_message && self.seen_message_id {
+            self.pending.push(record);
+            return Vec::new();
+        }
+
+        self.emit_record(record)
+    }
+
+    fn finish(mut self) -> Vec<ReplyEvent> {
+        let pending = std::mem::take(&mut self.pending);
+        let mut out = Vec::new();
+        for record in pending {
+            out.extend(self.emit_record(record));
+        }
+        out
+    }
+
+    fn emit_record(&mut self, record: ParsedRecord) -> Vec<ReplyEvent> {
+        let before = self.ready.len();
+        append_record_events(
+            record,
+            &mut self.ready,
+            &mut self.final_text,
+            &mut self.error_seen,
+        );
+        self.ready.split_off(before)
+    }
+}
+
 fn current_turn_records(
     lines: &[String],
     expected_session_id: Option<&str>,
@@ -309,6 +424,16 @@ fn has_structured_terminal_event(events: &[ReplyEvent]) -> bool {
         .any(|event| matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. }))
 }
 
+fn stream_events(events: Vec<ReplyEvent>, has_tap: bool) -> Vec<ReplyEvent> {
+    if !has_tap {
+        return events;
+    }
+    events
+        .into_iter()
+        .filter(|event| matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. }))
+        .collect()
+}
+
 /// Découpe une commande utilisateur en argv sans shell implicite.
 ///
 /// Supporte les guillemets simples/doubles et les antislashs. Les expansions shell,
@@ -440,6 +565,45 @@ impl AgentSession for OpenCodeSession {
     }
 
     async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
+        self.send_inner(prompt, None).await
+    }
+
+    async fn send_with_tap(
+        &self,
+        prompt: &str,
+        tap: std::sync::mpsc::Sender<ReplyEvent>,
+    ) -> Result<ReplyStream, AgentSessionError> {
+        self.send_inner(prompt, Some(tap)).await
+    }
+
+    async fn shutdown(&self) -> Result<(), AgentSessionError> {
+        Ok(())
+    }
+
+    async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
+        let child = self
+            .current_child
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone();
+        if let Some(child) = child {
+            child
+                .lock()
+                .await
+                .kill()
+                .await
+                .map_err(|e| AgentSessionError::Io(format!("annulation OpenCode: {e}")))?;
+        }
+        Ok(())
+    }
+}
+
+impl OpenCodeSession {
+    async fn send_inner(
+        &self,
+        prompt: &str,
+        tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
+    ) -> Result<ReplyStream, AgentSessionError> {
         let mut cmd = Command::new(&self.command);
         cmd.args(self.build_args(prompt))
             .stdin(Stdio::null())
@@ -482,12 +646,16 @@ impl AgentSession for OpenCodeSession {
             let expected_session = self.conversation_id();
             let mut lines = BufReader::new(stdout).lines();
             let mut collected = Vec::new();
+            let mut live = tap
+                .as_ref()
+                .map(|_| LiveTurnProjector::new(expected_session.clone()));
             while let Some(line) = lines
                 .next_line()
                 .await
                 .map_err(|e| AgentSessionError::Io(e.to_string()))?
             {
-                if let Some(engine_id) = extract_session_id(&line)? {
+                let record = parse_jsonl_record(&line)?;
+                if let Some(engine_id) = record.session_id.clone() {
                     if expected_session
                         .as_deref()
                         .is_none_or(|expected| expected == engine_id)
@@ -495,8 +663,18 @@ impl AgentSession for OpenCodeSession {
                         self.capture_session_id(engine_id);
                     }
                 }
+                if let (Some(tap), Some(live)) = (&tap, live.as_mut()) {
+                    for event in live.ingest(record) {
+                        send_tap(tap, event);
+                    }
+                }
                 collected.push(line);
             }
+            if let (Some(tap), Some(live)) = (&tap, live) {
+                for event in live.finish() {
+                    send_tap(tap, event);
+                }
+            }
 
             let mut stderr_bytes = Vec::new();
             stderr_pipe
@@ -517,10 +695,11 @@ impl AgentSession for OpenCodeSession {
         let stderr = String::from_utf8_lossy(&stderr_bytes);
         let parsed_session = expected_session.or_else(|| self.conversation_id());
         let parsed = parse_jsonl_turn_scoped(&collected, parsed_session.as_deref());
+        let has_tap = tap.is_some();
         if !status.success() {
             if let Ok(events) = parsed {
                 if has_structured_terminal_event(&events) {
-                    return Ok(Box::new(events.into_iter()));
+                    return Ok(Box::new(stream_events(events, has_tap).into_iter()));
                 }
             }
             if stderr.contains("server unavailable") && stderr.contains("key=idea") {
@@ -541,28 +720,7 @@ impl AgentSession for OpenCodeSession {
         }
 
         let events = parsed?;
-        Ok(Box::new(events.into_iter()))
+        Ok(Box::new(stream_events(events, has_tap).into_iter()))
-    }
-
-    async fn shutdown(&self) -> Result<(), AgentSessionError> {
-        Ok(())
-    }
-
-    async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> {
-        let child = self
-            .current_child
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .clone();
-        if let Some(child) = child {
-            child
-                .lock()
-                .await
-                .kill()
-                .await
-                .map_err(|e| AgentSessionError::Io(format!("annulation OpenCode: {e}")))?;
-        }
-        Ok(())
     }
 }
 
@@ -574,6 +732,8 @@ mod tests {
     use std::os::unix::fs::PermissionsExt;
     #[cfg(unix)]
     use std::path::{Path, PathBuf};
+    #[cfg(unix)]
+    use std::time::Duration;
 
     use super::*;
 
@@ -714,6 +874,36 @@ mod tests {
         assert_eq!(event, ParsedEvent::ToolActivity("bash".to_owned()));
     }
 
+    #[test]
+    fn parse_jsonl_turn_tags_idea_tools_as_mcp_progress() {
+        let events = parse_jsonl_turn(&[
+            r#"{"type":"tool_use","part":{"tool":"idea_ask_agent"}}"#.to_owned(),
+            r#"{"type":"text","part":{"text":"ok"}}"#.to_owned(),
+        ])
+        .unwrap();
+
+        assert!(
+            events.iter().any(|event| matches!(
+                event,
+                ReplyEvent::Progress { progress }
+                    if progress.source == ReplyProgressSource::IdeaLocal
+                        && progress.kind == ReplyProgressKind::Mcp
+                        && progress.stage == ReplyProgressStage::Completed
+                        && progress.tool_name.as_deref() == Some("idea_ask_agent")
+            )),
+            "les tools idea_* doivent être surfacés comme MCP IdeA local: {events:?}"
+        );
+        assert!(
+            !events.iter().any(|event| matches!(
+                event,
+                ReplyEvent::Progress { progress }
+                    if progress.source == ReplyProgressSource::ProviderNative
+                        && progress.tool_name.as_deref() == Some("idea_ask_agent")
+            )),
+            "le tool MCP ne doit pas produire une double carte provider-native: {events:?}"
+        );
+    }
+
     #[test]
     fn parse_jsonl_turn_keeps_latest_message_when_opencode_replays_history() {
         let events = parse_jsonl_turn(&[
@@ -790,6 +980,14 @@ mod tests {
         }
     }
 
+    #[cfg(unix)]
+    fn write_executable_script(path: &Path, body: impl AsRef<str>) {
+        fs::write(path, body.as_ref()).unwrap();
+        let mut permissions = fs::metadata(path).unwrap().permissions();
+        permissions.set_mode(0o755);
+        fs::set_permissions(path, permissions).unwrap();
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn send_prefers_parseable_stdout_over_nonzero_exit_status() {
@@ -838,6 +1036,147 @@ exit 1
         );
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn send_with_tap_emits_progress_live_before_final_and_tags_idea_tool_as_mcp() {
+        let tmp = TempDir::new("tap-live");
+        let script = tmp.path().join("opencode-fixture.sh");
+        write_executable_script(
+            &script,
+            r#"#!/bin/sh
+printf '%s\n' '{"type":"step_start","sessionID":"fresh-engine","part":{"messageID":"m1","type":"step-start"}}'
+sleep 1
+printf '%s\n' '{"type":"tool_use","sessionID":"fresh-engine","part":{"messageID":"m1","type":"tool","tool":"idea_ask_agent"}}'
+printf '%s\n' '{"type":"text","sessionID":"fresh-engine","part":{"messageID":"m1","type":"text","text":"réponse live"}}'
+printf '%s\n' '{"type":"step_finish","sessionID":"fresh-engine","part":{"messageID":"m1","type":"step-finish"}}'
+"#,
+        );
+        let session = Arc::new(
+            OpenCodeSession::new(
+                SessionId::new_random(),
+                script.to_string_lossy(),
+                Vec::new(),
+                tmp.path().to_string_lossy(),
+                None,
+                Vec::new(),
+                None,
+                None,
+            )
+            .unwrap(),
+        );
+        let (tx, rx) = std::sync::mpsc::channel();
+        let worker = {
+            let session = Arc::clone(&session);
+            tokio::spawn(async move {
+                session
+                    .send_with_tap("prompt", tx)
+                    .await
+                    .map(|stream| stream.collect::<Vec<_>>())
+            })
+        };
+
+        let (first_live, rx) = tokio::task::spawn_blocking(move || {
+            let first_live = rx
+                .recv_timeout(Duration::from_millis(500))
+                .expect("un événement de progress doit atteindre le tap avant le Final");
+            (first_live, rx)
+        })
+        .await
+        .unwrap();
+        assert!(
+            matches!(
+                first_live,
+                ReplyEvent::Progress { ref progress }
+                    if progress.native_event.as_deref() == Some("step_start")
+            ),
+            "le premier événement live doit être le step_start provider: {first_live:?}"
+        );
+
+        let returned = worker.await.unwrap().unwrap();
+        let live: Vec<_> = std::iter::once(first_live).chain(rx.into_iter()).collect();
+
+        assert!(
+            live.iter()
+                .all(|event| !matches!(event, ReplyEvent::Final { .. })),
+            "le tap ne doit pas recevoir le Final autoritaire: {live:?}"
+        );
+        assert!(
+            live.iter().any(|event| matches!(
+                event,
+                ReplyEvent::Progress { progress }
+                    if progress.source == ReplyProgressSource::IdeaLocal
+                        && progress.kind == ReplyProgressKind::Mcp
+                        && progress.stage == ReplyProgressStage::Completed
+                        && progress.tool_name.as_deref() == Some("idea_ask_agent")
+            )),
+            "le tool_use idea_* doit produire une progression MCP locale: {live:?}"
+        );
+        assert_eq!(
+            returned,
+            vec![ReplyEvent::Final {
+                content: "réponse live".to_owned()
+            }]
+        );
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn send_with_tap_does_not_replay_previous_message_into_live_stream() {
+        let tmp = TempDir::new("tap-replay");
+        let script = tmp.path().join("opencode-fixture.sh");
+        write_executable_script(
+            &script,
+            r#"#!/bin/sh
+printf '%s\n' '{"type":"step_start","sessionID":"seeded-engine","part":{"messageID":"old","type":"step-start"}}'
+printf '%s\n' '{"type":"text","sessionID":"seeded-engine","part":{"messageID":"old","type":"text","text":"ancienne réponse"}}'
+printf '%s\n' '{"type":"step_finish","sessionID":"seeded-engine","part":{"messageID":"old","type":"step-finish"}}'
+printf '%s\n' '{"type":"step_start","sessionID":"seeded-engine","part":{"messageID":"new","type":"step-start"}}'
+printf '%s\n' '{"type":"text","sessionID":"seeded-engine","part":{"messageID":"new","type":"text","text":"réponse courante"}}'
+printf '%s\n' '{"type":"step_finish","sessionID":"seeded-engine","part":{"messageID":"new","type":"step-finish"}}'
+"#,
+        );
+        let session = OpenCodeSession::new(
+            SessionId::new_random(),
+            script.to_string_lossy(),
+            Vec::new(),
+            tmp.path().to_string_lossy(),
+            Some("seeded-engine".to_owned()),
+            Vec::new(),
+            None,
+            None,
+        )
+        .unwrap();
+        let (tx, rx) = std::sync::mpsc::channel();
+
+        let returned = session
+            .send_with_tap("prompt", tx)
+            .await
+            .unwrap()
+            .collect::<Vec<_>>();
+        let live: Vec<_> = rx.into_iter().collect();
+
+        assert!(
+            !live.iter().any(|event| matches!(
+                event,
+                ReplyEvent::TextDelta { text } if text.contains("ancienne")
+            )),
+            "le texte rejoué ne doit pas polluer le tap live: {live:?}"
+        );
+        assert!(
+            live.iter().any(|event| matches!(
+                event,
+                ReplyEvent::TextDelta { text } if text == "réponse courante"
+            )),
+            "le message courant doit rester visible dans le tap live: {live:?}"
+        );
+        assert_eq!(
+            returned,
+            vec![ReplyEvent::Final {
+                content: "réponse courante".to_owned()
+            }]
+        );
+    }
+
     #[test]
     fn extract_session_id_accepts_opencode_jsonl_session_id_variants() {
         assert_eq!(