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

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

Browse Source 
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.
This commit is contained in:
Blomios
2026-08-09 23:06:41 +02:00
parent a3d94d4d5e
commit 836137a36d
1 changed files with 410 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

403
crates/infrastructure/src/session/opencode.rs
View File

@ -184,10 +184,25 @@ 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 {
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<ReplyEvent>,
final_text: &mut String,
error_seen: &mut bool,
) {
match record.event {
ParsedEvent::StepStart => {
events.push(ReplyEvent::Progress {
@ -213,13 +228,7 @@ fn parse_jsonl_turn_scoped(
}
ParsedEvent::ToolActivity(label) => {
events.push(ReplyEvent::Progress {
progress: provider_progress(
ReplyProgressKind::Tool,
ReplyProgressStage::Started,
label.clone(),
"tool_use",
)
.with_tool_name(label.clone()),
progress: tool_progress(&label),
});
events.push(ReplyEvent::ToolActivity { label });
}
@ -230,12 +239,18 @@ fn parse_jsonl_turn_scoped(
}
}
ParsedEvent::Error(message) => {
error_seen = true;
*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()))
}
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<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!(