From b88d500d89ff6c4bf6828930d8a747f023f42deb Mon Sep 17 00:00:00 2001 From: Blomios Date: Tue, 4 Aug 2026 09:50:02 +0200 Subject: [PATCH] fix(orchestrator): corrige la delegation inter-agent GLM/OpenCode sans reponse Final La detection d'absence de reponse structuree finale (structured_no_reply_error) ne couvrait pas AgentSessionError::Decode ni le message "aucun final textuel exploitable" emis par l'adaptateur OpenCode, ce qui laissait idea_ask_agent planter silencieusement sur les profils GLM/OpenCode au lieu de retomber sur le chemin de secours prevu. Co-Authored-By: Claude Sonnet 5 --- .../application/src/orchestrator/service.rs | 13 +- .../application/tests/orchestrator_service.rs | 26 ++ crates/infrastructure/src/session/opencode.rs | 320 +++++++++++++++--- 3 files changed, 301 insertions(+), 58 deletions(-) diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 0cd943c..d41a6a4 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -79,12 +79,17 @@ fn submit_config_for_profile(profile: &AgentProfile) -> SubmitConfig { } fn structured_no_reply_error(err: &domain::ports::AgentSessionError) -> bool { - matches!( - err, + match err { domain::ports::AgentSessionError::Io(message) - if message.contains("sans événement Final") + | domain::ports::AgentSessionError::Decode(message) => { + message.contains("sans événement Final") || message.contains("without a structured final") - ) + || message.contains("aucun final textuel exploitable") + } + domain::ports::AgentSessionError::Start(_) | domain::ports::AgentSessionError::Timeout => { + false + } + } } fn bound_task_text(value: &str, max_bytes: usize) -> String { diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index dcc8666..f00f573 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -1170,6 +1170,7 @@ impl TestMailbox { enum TestCompletion { Replied(String), NoReply, + DecodeNoFinal, Cancelled, } @@ -1467,6 +1468,9 @@ impl AgentSession for CompletionSession { TestCompletion::NoReply => Err(AgentSessionError::Io( "target returned without a structured final".to_owned(), )), + TestCompletion::DecodeNoFinal => Err(AgentSessionError::Decode( + "OpenCode n'a produit aucun final textuel exploitable".to_owned(), + )), TestCompletion::Cancelled => Err(AgentSessionError::Io( "structured test turn cancelled".to_owned(), )), @@ -2134,6 +2138,28 @@ async fn ask_target_returns_to_prompt_without_reply_is_a_typed_error() { assert_eq!(fx.mailbox.pending(&aid(1)), 0, "head retired by completion"); } +#[tokio::test] +async fn structured_opencode_decode_without_final_is_a_typed_no_reply_error() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + let svc = Arc::clone(&fx.service); + let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await }); + await_until(|| fx.mailbox.pending(&aid(1)) == 1).await; + + fx.mailbox + .completions() + .push(aid(1), TestCompletion::DecodeNoFinal); + + let err = timeout(TEST_GUARD, ask) + .await + .expect("ask completes promptly, not after the long timeout") + .expect("join ok") + .expect_err("OpenCode no-final decode is a typed no-reply error"); + assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY", "got {err:?}"); +} + /// An `idea_reply` that lands **first** still wins over a (later) no-reply completion: /// the completion then finds the head gone and is a no-op (idempotent race). #[tokio::test] diff --git a/crates/infrastructure/src/session/opencode.rs b/crates/infrastructure/src/session/opencode.rs index 1e6419b..bdc3502 100644 --- a/crates/infrastructure/src/session/opencode.rs +++ b/crates/infrastructure/src/session/opencode.rs @@ -3,7 +3,8 @@ //! OpenCode est piloté comme host process local : IdeA génère un `opencode.json` //! isolé dans le run dir, OpenCode lance le bridge MCP `idea`, puis chaque tour est //! un `opencode run --format json `. L'adapter ne connaît que le contrat -//! JSONL minimal observé/cadré : `step_start`, `text`, `step_finish`. +//! JSONL observé/cadré : `step_start`, `text(part.text)`, `tool_use`, +//! `step_finish`, `error`. use std::process::Stdio; use std::sync::Arc; @@ -25,6 +26,8 @@ pub enum ParsedEvent { StepStart, /// Fragment texte assistant. Text(String), + /// Activité outil OpenCode. + ToolActivity(String), /// Fin d'étape OpenCode. StepFinish, /// Erreur terminale remontée par OpenCode. @@ -41,14 +44,13 @@ pub fn extract_session_id(line: &str) -> Result, AgentSessionErro } let value: Value = serde_json::from_str(trimmed) .map_err(|e| AgentSessionError::Decode(format!("ligne JSON OpenCode illisible: {e}")))?; - Ok(value - .get("sessionID") - .or_else(|| value.get("session_id")) - .or_else(|| value.get("sessionId")) - .and_then(Value::as_str) - .map(str::trim) - .filter(|id| !id.is_empty()) - .map(str::to_owned)) + Ok( + json_string_field(&value, &["sessionID", "session_id", "sessionId"]).or_else(|| { + value + .get("part") + .and_then(|part| json_string_field(part, &["sessionID", "session_id", "sessionId"])) + }), + ) } /// Message d'erreur par défaut quand `error` n'est ni une chaîne ni un objet exploitable. @@ -59,34 +61,77 @@ const OPENCODE_ERROR_FALLBACK: &str = "OpenCode a signalé une erreur sans déta /// # Errors /// [`AgentSessionError::Decode`] si la ligne non vide n'est pas du JSON valide. pub fn parse_jsonl_event(line: &str) -> Result { + parse_jsonl_record(line).map(|record| record.event) +} + +#[derive(Debug, Clone)] +struct ParsedRecord { + event: ParsedEvent, + session_id: Option, + message_id: Option, +} + +fn parse_jsonl_record(line: &str) -> Result { let trimmed = line.trim(); if trimmed.is_empty() { - return Ok(ParsedEvent::Ignored); + return Ok(ParsedRecord { + event: ParsedEvent::Ignored, + session_id: None, + message_id: None, + }); } let value: Value = serde_json::from_str(trimmed) .map_err(|e| AgentSessionError::Decode(format!("ligne JSON OpenCode illisible: {e}")))?; - match value.get("type").and_then(Value::as_str) { - Some("step_start") | Some("step.start") => Ok(ParsedEvent::StepStart), - Some("step_finish") | Some("step.finish") => Ok(ParsedEvent::StepFinish), + let part = value.get("part"); + let session_id = + json_string_field(&value, &["sessionID", "session_id", "sessionId"]).or_else(|| { + part.and_then(|part| json_string_field(part, &["sessionID", "session_id", "sessionId"])) + }); + let message_id = + json_string_field(&value, &["messageID", "message_id", "messageId"]).or_else(|| { + part.and_then(|part| json_string_field(part, &["messageID", "message_id", "messageId"])) + }); + let event = match value.get("type").and_then(Value::as_str).or_else(|| { + part.and_then(|part| part.get("type")) + .and_then(Value::as_str) + }) { + Some("step_start" | "step.start" | "step-start") => ParsedEvent::StepStart, + Some("step_finish" | "step.finish" | "step-finish") => ParsedEvent::StepFinish, // Le texte réel est niché sous `.part.text` (cf. `packages/opencode/src/cli/cmd/run.ts`) ; // un seul événement `text` par part, déjà finalisé — pas de delta à dédupliquer. - Some("text") => Ok(ParsedEvent::Text( - value - .get("part") - .and_then(|part| part.get("text")) - .and_then(Value::as_str) - .unwrap_or_default() - .to_owned(), - )), - Some("error") => Ok(ParsedEvent::Error(opencode_error_message(&value))), - _ => Ok(ParsedEvent::Ignored), - } + Some("text") => part + .and_then(|part| part.get("text")) + .or_else(|| value.get("text")) + .or_else(|| value.get("content")) + .and_then(Value::as_str) + .map(|text| ParsedEvent::Text(text.to_owned())) + .unwrap_or(ParsedEvent::Ignored), + Some("tool_use" | "tool.use" | "tool") => ParsedEvent::ToolActivity(tool_label(&value)), + Some("error" | "session.error") => ParsedEvent::Error(opencode_error_message(&value)), + _ => ParsedEvent::Ignored, + }; + Ok(ParsedRecord { + event, + session_id, + message_id, + }) +} + +fn json_string_field(value: &Value, names: &[&str]) -> Option { + names + .iter() + .find_map(|name| value.get(*name).and_then(Value::as_str)) + .map(str::trim) + .filter(|id| !id.is_empty()) + .map(str::to_owned) } /// Extrait un message d'erreur exploitable depuis `value.error`, qu'il s'agisse d'une /// chaîne brute ou d'un objet `{message: ...}` (forme exacte non confirmée empiriquement). fn opencode_error_message(value: &Value) -> String { - let error = value.get("error"); + let error = value + .get("error") + .or_else(|| value.get("properties").and_then(|props| props.get("error"))); error .and_then(Value::as_str) .map(str::to_owned) @@ -96,29 +141,59 @@ fn opencode_error_message(value: &Value) -> String { .and_then(Value::as_str) .map(str::to_owned) }) + .or_else(|| { + error + .and_then(|e| e.get("data")) + .and_then(|data| data.get("message")) + .and_then(Value::as_str) + .map(str::to_owned) + }) + .or_else(|| { + error + .and_then(|e| e.get("name")) + .and_then(Value::as_str) + .map(str::to_owned) + }) .unwrap_or_else(|| OPENCODE_ERROR_FALLBACK.to_owned()) } +fn tool_label(value: &Value) -> String { + let part = value.get("part").unwrap_or(value); + json_string_field(part, &["tool", "name"]) + .or_else(|| json_string_field(value, &["tool", "name"])) + .unwrap_or_else(|| "outil OpenCode".to_owned()) +} + /// Convertit des lignes JSONL OpenCode en événements domaine. /// -/// Le `Final` est la concaténation ordonnée des événements `text`. Un tour sans -/// texte mais terminé proprement (`step_finish`) émet un `Final` vide : cela -/// couvre les tours tool-call-only tout en gardant un hard-fail pour un vrai flux -/// vide, uniquement ignoré, ou seulement démarré. +/// Le `Final` est la concaténation ordonnée des événements `text` du tour courant. +/// Si OpenCode rejoue des événements d'une même session, les identifiants de +/// message (`messageID`/`messageId`) servent à conserver seulement le dernier +/// message assistant observable. Un tour terminé sans texte exploitable est une +/// erreur de décodage, pas un succès vide. pub fn parse_jsonl_turn(lines: &[String]) -> Result, AgentSessionError> { + parse_jsonl_turn_scoped(lines, None) +} + +fn parse_jsonl_turn_scoped( + lines: &[String], + expected_session_id: Option<&str>, +) -> Result, AgentSessionError> { + let records = current_turn_records(lines, expected_session_id)?; let mut events = Vec::new(); let mut final_text = String::new(); let mut error_seen = false; - let mut step_finish_seen = false; - for line in lines { - match parse_jsonl_event(line)? { + for record in records { + match record.event { ParsedEvent::StepStart => { events.push(ReplyEvent::Heartbeat); } ParsedEvent::StepFinish => { - step_finish_seen = true; events.push(ReplyEvent::Heartbeat); } + ParsedEvent::ToolActivity(label) => { + events.push(ReplyEvent::ToolActivity { label }); + } ParsedEvent::Text(text) => { if !text.is_empty() { final_text.push_str(&text); @@ -136,14 +211,8 @@ pub fn parse_jsonl_turn(lines: &[String]) -> Result, AgentSessio if error_seen { return Ok(events); } - if step_finish_seen { - events.push(ReplyEvent::Final { - content: String::new(), - }); - return Ok(events); - } return Err(AgentSessionError::Decode( - "OpenCode n'a produit aucun final textuel".to_owned(), + "OpenCode n'a produit aucun final textuel exploitable".to_owned(), )); } events.push(ReplyEvent::Final { @@ -152,6 +221,49 @@ pub fn parse_jsonl_turn(lines: &[String]) -> Result, AgentSessio Ok(events) } +fn current_turn_records( + lines: &[String], + expected_session_id: Option<&str>, +) -> Result, AgentSessionError> { + let mut records = Vec::new(); + for line in lines { + let record = parse_jsonl_record(line)?; + if let (Some(expected), Some(actual)) = (expected_session_id, record.session_id.as_deref()) + { + if actual != expected { + continue; + } + } + if !matches!(record.event, ParsedEvent::Ignored) { + records.push(record); + } + } + let Some(current_message_id) = records.iter().rev().find_map(|record| { + matches!( + record.event, + ParsedEvent::Text(_) + | ParsedEvent::ToolActivity(_) + | ParsedEvent::StepStart + | ParsedEvent::StepFinish + | ParsedEvent::Error(_) + ) + .then(|| record.message_id.as_deref()) + .flatten() + .map(str::to_owned) + }) else { + return Ok(records); + }; + Ok(records + .into_iter() + .filter(|record| { + record + .message_id + .as_deref() + .is_none_or(|message_id| message_id == current_message_id.as_str()) + }) + .collect()) +} + fn has_structured_terminal_event(events: &[ReplyEvent]) -> bool { events .iter() @@ -301,6 +413,7 @@ impl AgentSession for OpenCodeSession { .take() .ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?; + let expected_session = self.conversation_id(); let mut lines = BufReader::new(stdout).lines(); let mut collected = Vec::new(); while let Some(line) = lines @@ -309,7 +422,12 @@ impl AgentSession for OpenCodeSession { .map_err(|e| AgentSessionError::Io(e.to_string()))? { if let Some(engine_id) = extract_session_id(&line)? { - self.capture_session_id(engine_id); + if expected_session + .as_deref() + .is_none_or(|expected| expected == engine_id) + { + self.capture_session_id(engine_id); + } } collected.push(line); } @@ -324,7 +442,8 @@ impl AgentSession for OpenCodeSession { .await .map_err(|e| AgentSessionError::Io(e.to_string()))?; let stderr = String::from_utf8_lossy(&stderr_bytes); - let parsed = parse_jsonl_turn(&collected); + let parsed_session = expected_session.or_else(|| self.conversation_id()); + let parsed = parse_jsonl_turn_scoped(&collected, parsed_session.as_deref()); if !status.success() { if let Ok(events) = parsed { if has_structured_terminal_event(&events) { @@ -410,21 +529,15 @@ mod tests { } #[test] - fn parse_jsonl_turn_accepts_tool_only_finished_turn() { - let events = parse_jsonl_turn(&[ + fn parse_jsonl_turn_rejects_tool_only_finished_turn() { + let err = parse_jsonl_turn(&[ r#"{"type":"step_start"}"#.to_owned(), r#"{"type":"step_finish"}"#.to_owned(), ]) - .unwrap(); - assert_eq!( - events, - vec![ - ReplyEvent::Heartbeat, - ReplyEvent::Heartbeat, - ReplyEvent::Final { - content: String::new() - } - ] + .unwrap_err(); + assert!( + matches!(err, AgentSessionError::Decode(ref message) if message.contains("aucun final textuel exploitable")), + "tool-only turn must be actionable, not a fake empty success: {err:?}" ); } @@ -498,6 +611,64 @@ mod tests { assert_eq!(event, ParsedEvent::Error("panne réseau".to_owned())); } + #[test] + fn parse_jsonl_event_accepts_tool_use() { + let event = parse_jsonl_event(r#"{"type":"tool_use","part":{"tool":"bash"}}"#).unwrap(); + assert_eq!(event, ParsedEvent::ToolActivity("bash".to_owned())); + } + + #[test] + fn parse_jsonl_turn_keeps_latest_message_when_opencode_replays_history() { + let events = parse_jsonl_turn(&[ + r#"{"type":"step_start","sessionID":"s1","part":{"messageID":"old","type":"step-start"}}"# + .to_owned(), + r#"{"type":"text","sessionID":"s1","part":{"messageID":"old","type":"text","text":"ancienne réponse"}}"# + .to_owned(), + r#"{"type":"step_finish","sessionID":"s1","part":{"messageID":"old","type":"step-finish"}}"# + .to_owned(), + r#"{"type":"step_start","sessionID":"s1","part":{"messageID":"new","type":"step-start"}}"# + .to_owned(), + r#"{"type":"text","sessionID":"s1","part":{"messageID":"new","type":"text","text":"réponse courante"}}"# + .to_owned(), + r#"{"type":"step_finish","sessionID":"s1","part":{"messageID":"new","type":"step-finish"}}"# + .to_owned(), + ]) + .unwrap(); + assert_eq!( + events.last(), + Some(&ReplyEvent::Final { + content: "réponse courante".to_owned() + }) + ); + assert!( + !events.iter().any(|event| matches!( + event, + ReplyEvent::TextDelta { text } if text.contains("ancienne") + )), + "replayed text from a previous message must not pollute the current final: {events:?}" + ); + } + + #[test] + fn parse_jsonl_turn_scoped_ignores_other_sessions_when_known() { + let events = parse_jsonl_turn_scoped( + &[ + r#"{"type":"text","sessionID":"other","part":{"messageID":"m1","text":"pollution"}}"# + .to_owned(), + r#"{"type":"text","sessionID":"current","part":{"messageID":"m2","text":"ok"}}"# + .to_owned(), + ], + Some("current"), + ) + .unwrap(); + assert_eq!( + events.last(), + Some(&ReplyEvent::Final { + content: "ok".to_owned() + }) + ); + } + #[cfg(unix)] struct TempDir(PathBuf); @@ -674,4 +845,45 @@ printf '%s\n' '{{"type":"step_finish","sessionID":"seeded-engine"}}' ); assert_eq!(argv.last(), Some(&"prompt")); } + + #[cfg(unix)] + #[tokio::test] + async fn seeded_session_ignores_and_does_not_capture_other_session_events() { + let tmp = TempDir::new("ignore-other-session"); + let script = tmp.path().join("opencode-fixture.sh"); + fs::write( + &script, + r#"#!/bin/sh +printf '%s\n' '{"type":"step_start","sessionID":"seeded-engine","part":{"messageID":"m1"}}' +printf '%s\n' '{"type":"text","sessionID":"seeded-engine","part":{"messageID":"m1","text":"réponse seed"}}' +printf '%s\n' '{"type":"step_finish","sessionID":"seeded-engine","part":{"messageID":"m1"}}' +printf '%s\n' '{"type":"text","sessionID":"other-engine","part":{"messageID":"m2","text":"pollution"}}' +"#, + ) + .unwrap(); + let mut permissions = fs::metadata(&script).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(&script, permissions).unwrap(); + 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 events = session.send("prompt").await.unwrap().collect::>(); + + assert_eq!( + events.last(), + Some(&ReplyEvent::Final { + content: "réponse seed".to_owned() + }) + ); + assert_eq!(session.conversation_id().as_deref(), Some("seeded-engine")); + } }