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 <noreply@anthropic.com>
This commit is contained in:
2026-08-04 09:50:02 +02:00
parent 11c405efea
commit b88d500d89
3 changed files with 301 additions and 58 deletions

View File

@ -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 <prompt>`. 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<Option<String>, 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<ParsedEvent, AgentSessionError> {
parse_jsonl_record(line).map(|record| record.event)
}
#[derive(Debug, Clone)]
struct ParsedRecord {
event: ParsedEvent,
session_id: Option<String>,
message_id: Option<String>,
}
fn parse_jsonl_record(line: &str) -> Result<ParsedRecord, AgentSessionError> {
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<String> {
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<Vec<ReplyEvent>, AgentSessionError> {
parse_jsonl_turn_scoped(lines, None)
}
fn parse_jsonl_turn_scoped(
lines: &[String],
expected_session_id: Option<&str>,
) -> Result<Vec<ReplyEvent>, 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<Vec<ReplyEvent>, 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<Vec<ReplyEvent>, AgentSessio
Ok(events)
}
fn current_turn_records(
lines: &[String],
expected_session_id: Option<&str>,
) -> Result<Vec<ParsedRecord>, 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::<Vec<_>>();
assert_eq!(
events.last(),
Some(&ReplyEvent::Final {
content: "réponse seed".to_owned()
})
);
assert_eq!(session.conversation_id().as_deref(), Some("seeded-engine"));
}
}