From d553d137ea6fdabebf9f7a6111d73edbcc9078e1 Mon Sep 17 00:00:00 2001 From: Blomios Date: Mon, 24 Aug 2026 08:04:49 +0200 Subject: [PATCH] =?UTF-8?q?fix(custom-cli):=20progression=20et=20messages?= =?UTF-8?q?=20d'outils=20restaur=C3=A9s=20au=20second=20tour=20=E2=80=94?= =?UTF-8?q?=20le=20projecteur=20ne=20retient=20plus=20les=20events=20de=20?= =?UTF-8?q?progress=20apr=C3=A8s=20le=20premier=20message,=20seuls=20Text/?= =?UTF-8?q?Error=20restent=20bufferis=C3=A9s=20=E2=80=94=20#252=20(QA=20ve?= =?UTF-8?q?rte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/infrastructure/src/session/opencode.rs | 85 ++++++++++++++++++- 1 file changed, 83 insertions(+), 2 deletions(-) diff --git a/crates/infrastructure/src/session/opencode.rs b/crates/infrastructure/src/session/opencode.rs index 8dad5951..5c0cf11b 100644 --- a/crates/infrastructure/src/session/opencode.rs +++ b/crates/infrastructure/src/session/opencode.rs @@ -390,8 +390,11 @@ impl LiveTurnProjector { } if self.hold_until_new_message && self.seen_message_id { - self.pending.push(record); - return Vec::new(); + if matches!(record.event, ParsedEvent::Text(_) | ParsedEvent::Error(_)) { + self.pending.push(record); + return Vec::new(); + } + return self.emit_record(record); } self.emit_record(record) @@ -1226,6 +1229,84 @@ printf '%s\n' '{"type":"step_finish","sessionID":"seeded-engine","part":{"messag ); } + #[cfg(unix)] + #[tokio::test] + async fn send_with_tap_streams_progress_for_seeded_session_with_single_current_message_id() { + let tmp = TempDir::new("tap-seeded-single-message"); + 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":"current","type":"step-start"}}' +sleep 1 +printf '%s\n' '{"type":"tool_use","sessionID":"seeded-engine","part":{"messageID":"current","type":"tool","tool":"idea_ask_agent"}}' +printf '%s\n' '{"type":"text","sessionID":"seeded-engine","part":{"messageID":"current","type":"text","text":"réponse courante"}}' +printf '%s\n' '{"type":"step_finish","sessionID":"seeded-engine","part":{"messageID":"current","type":"step-finish"}}' +"#, + ); + let session = Arc::new( + 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 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("la session existante doit streamer le progress du tour courant"); + (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 du message courant: {first_live:?}" + ); + + let returned = worker.await.unwrap().unwrap(); + let live: Vec<_> = std::iter::once(first_live).chain(rx.into_iter()).collect(); + + 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 du seul message courant doit atteindre 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!(