diff --git a/crates/infrastructure/src/session/opencode.rs b/crates/infrastructure/src/session/opencode.rs index 6191c6d7..0251e374 100644 --- a/crates/infrastructure/src/session/opencode.rs +++ b/crates/infrastructure/src/session/opencode.rs @@ -184,58 +184,73 @@ fn parse_jsonl_turn_scoped( expected_session_id: Option<&str>, ) -> Result, AgentSessionError> { let records = current_turn_records(lines, expected_session_id)?; + events_from_records(records) +} + +fn events_from_records(records: Vec) -> Result, AgentSessionError> { let mut events = Vec::new(); let mut final_text = String::new(); let mut error_seen = false; for record in records { - 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: 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_record_events(record, &mut events, &mut final_text, &mut error_seen); } + append_final_event(events, final_text, error_seen) +} + +fn append_record_events( + record: ParsedRecord, + events: &mut Vec, + 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, + final_text: String, + error_seen: bool, +) -> Result, 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, event: ReplyEvent) { + if !matches!(event, ReplyEvent::Final { .. }) { + let _ = tap.send(event); + } +} + +#[derive(Default)] +struct LiveTurnProjector { + expected_session_id: Option, + active_message_id: Option, + seen_message_id: bool, + hold_until_new_message: bool, + pending: Vec, + ready: Vec, + final_text: String, + error_seen: bool, +} + +impl LiveTurnProjector { + fn new(expected_session_id: Option) -> Self { + Self { + hold_until_new_message: expected_session_id.is_some(), + expected_session_id, + ..Self::default() + } + } + + fn ingest(&mut self, record: ParsedRecord) -> Vec { + 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 { + 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 { + 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, has_tap: bool) -> Vec { + 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 { + self.send_inner(prompt, None).await + } + + async fn send_with_tap( + &self, + prompt: &str, + tap: std::sync::mpsc::Sender, + ) -> Result { + 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>, + ) -> Result { 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())) - } - - 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(()) + Ok(Box::new(stream_events(events, has_tap).into_iter())) } } @@ -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) { + 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::>()) + }) + }; + + 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::>(); + 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!(