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

feat(backend): modèle unifié streaming/progress/events provider-agnostic + pont app-tauri — foundation #156 (QA verte)

Browse Source
This commit is contained in:
Blomios
2026-08-06 13:13:17 +02:00
parent 40fa551f32
commit 918116664c
14 changed files with 630 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
crates/infrastructure/src/session/claude.rs
View File

@ -18,7 +18,10 @@ use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use serde_json::Value;
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::ports::{
AgentSession, AgentSessionError, ReplyEvent, ReplyProgress, ReplyProgressKind,
ReplyProgressSource, ReplyProgressStage, ReplyStream,
};
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
@ -88,7 +91,17 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
let events = match value.get("type").and_then(Value::as_str) {
// init/handshake : on capte le session_id ET on émet un battement de cœur
// (preuve de vivacité non terminale : la CLI a démarré et répond).
Some("system") => vec![ReplyEvent::Heartbeat],
Some("system") => vec![
ReplyEvent::Progress {
progress: provider_progress(
ReplyProgressKind::Turn,
ReplyProgressStage::Started,
"session initialisée",
"system",
),
},
ReplyEvent::Heartbeat,
],
// Limite de session/débit (ARCHITECTURE §21, niveau 1) : on lit l'heure de
// reset dans `rate_limit_info` (au lieu de la jeter) et on émet un
// `RateLimited{resets_at_ms}` **non terminal**. Robuste : absence/illisibilité
@ -140,6 +153,15 @@ fn assistant_events(value: &Value) -> Vec<ReplyEvent> {
.and_then(Value::as_str)
.unwrap_or("outil")
.to_owned();
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 });
}
_ => {}
@ -148,6 +170,16 @@ fn assistant_events(value: &Value) -> Vec<ReplyEvent> {
events
}
fn provider_progress(
kind: ReplyProgressKind,
stage: ReplyProgressStage,
label: impl Into<String>,
native_event: impl Into<String>,
) -> ReplyProgress {
ReplyProgress::new(ReplyProgressSource::ProviderNative, kind, stage, label)
.with_provider_event("claude", native_event)
}
/// **Extrait l'heure de reset d'une limite de débit** depuis l'objet
/// `rate_limit_info` d'un `rate_limit_event` Claude, **normalisée en époche-ms**
/// (ARCHITECTURE §21, niveau 1 structuré). Fonction **pure** (aucune I/O, aucun

55
crates/infrastructure/src/session/codex.rs
View File

@ -15,7 +15,10 @@ use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use serde_json::Value;
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::ports::{
AgentSession, AgentSessionError, ReplyEvent, ReplyProgress, ReplyProgressKind,
ReplyProgressSource, ReplyProgressStage, ReplyStream,
};
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
@ -81,7 +84,28 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
// Début/fin de tour côté moteur : pas de contenu, mais preuve de vivacité ⇒
// battement de cœur non terminal (readiness/heartbeat lot 1). Le `Final` vient
// toujours de l'`agent_message`, jamais de `turn.completed`.
Some("turn.started") | Some("turn.completed") => events.push(ReplyEvent::Heartbeat),
Some("turn.started") => {
events.push(ReplyEvent::Progress {
progress: provider_progress(
ReplyProgressKind::Turn,
ReplyProgressStage::Started,
"tour démarré",
"turn.started",
),
});
events.push(ReplyEvent::Heartbeat);
}
Some("turn.completed") => {
events.push(ReplyEvent::Progress {
progress: provider_progress(
ReplyProgressKind::Turn,
ReplyProgressStage::Completed,
"tour terminé côté provider",
"turn.completed",
),
});
events.push(ReplyEvent::Heartbeat);
}
Some("item.completed") => {
if let Some(item) = value.get("item") {
match item.get("type").and_then(Value::as_str) {
@ -99,9 +123,20 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
});
}
// reasoning / command / tout autre item ⇒ activité (label = type).
Some(kind) => events.push(ReplyEvent::ToolActivity {
label: kind.to_owned(),
}),
Some(kind) => {
events.push(ReplyEvent::Progress {
progress: provider_progress(
ReplyProgressKind::Tool,
ReplyProgressStage::Completed,
kind,
"item.completed",
)
.with_tool_name(kind.to_owned()),
});
events.push(ReplyEvent::ToolActivity {
label: kind.to_owned(),
});
}
None => {}
}
}
@ -116,6 +151,16 @@ pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
})
}
fn provider_progress(
kind: ReplyProgressKind,
stage: ReplyProgressStage,
label: impl Into<String>,
native_event: impl Into<String>,
) -> ReplyProgress {
ReplyProgress::new(ReplyProgressSource::ProviderNative, kind, stage, label)
.with_provider_event("codex", native_event)
}
const CODEX_ERROR_FALLBACK: &str = "Codex a renvoyé une erreur.";
fn non_blank_string(value: Option<&Value>) -> Option<String> {

3
crates/infrastructure/src/session/conformance.rs
View File

@ -216,7 +216,8 @@ pub(crate) mod harness {
assert!(
matches!(
e,
ReplyEvent::TextDelta { .. }
ReplyEvent::Progress { .. }
| ReplyEvent::TextDelta { .. }
| ReplyEvent::ToolActivity { .. }
| ReplyEvent::Announcement { .. }
| ReplyEvent::Heartbeat

43
crates/infrastructure/src/session/mod.rs
View File

@ -65,6 +65,13 @@ mod tests {
// -- Helpers ----------------------------------------------------------
fn without_progress(events: Vec<ReplyEvent>) -> Vec<ReplyEvent> {
events
.into_iter()
.filter(|event| !matches!(event, ReplyEvent::Progress { .. }))
.collect()
}
fn prepared_ctx() -> PreparedContext {
PreparedContext {
content: MarkdownDoc::new("# ctx"),
@ -188,7 +195,7 @@ mod tests {
.expect("parse ok");
assert_eq!(parsed.session_id.as_deref(), Some("conv-123"));
// L'init capte le session_id ET émet un heartbeat (vivacité non terminale, lot 1).
assert_eq!(parsed.events, vec![ReplyEvent::Heartbeat]);
assert_eq!(without_progress(parsed.events), vec![ReplyEvent::Heartbeat]);
}
/// §21 (LS2) : un `rate_limit_event` n'est PLUS un heartbeat — il porte désormais
@ -226,7 +233,7 @@ mod tests {
)
.expect("parse ok");
assert_eq!(
tool.events,
without_progress(tool.events),
vec![ReplyEvent::ToolActivity {
label: "Read".to_owned()
}]
@ -245,7 +252,7 @@ mod tests {
)
.expect("parse ok");
assert_eq!(
parsed.events,
without_progress(parsed.events),
vec![
ReplyEvent::TextDelta {
text: "un".to_owned()
@ -308,9 +315,15 @@ mod tests {
// turn.started / turn.completed ⇒ heartbeat (vivacité non terminale, lot 1).
let started = codex::parse_event(r#"{"type":"turn.started"}"#).expect("ok");
assert_eq!(started.events, vec![ReplyEvent::Heartbeat]);
assert_eq!(
without_progress(started.events),
vec![ReplyEvent::Heartbeat]
);
let completed = codex::parse_event(r#"{"type":"turn.completed","usage":{}}"#).expect("ok");
assert_eq!(completed.events, vec![ReplyEvent::Heartbeat]);
assert_eq!(
without_progress(completed.events),
vec![ReplyEvent::Heartbeat]
);
let msg = codex::parse_event(
r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"fini"}}"#,
@ -857,7 +870,7 @@ mod tests {
.unwrap();
assert_eq!(t1.events, vec![ReplyEvent::TextDelta { text: "a".into() }]);
assert_eq!(
t2.events,
without_progress(t2.events),
vec![ReplyEvent::ToolActivity {
label: "Bash".into()
}]
@ -876,7 +889,7 @@ mod tests {
)
.unwrap();
assert_eq!(
parsed.events,
without_progress(parsed.events),
vec![
ReplyEvent::TextDelta { text: "un".into() },
ReplyEvent::ToolActivity {
@ -897,7 +910,7 @@ mod tests {
)
.unwrap();
assert_eq!(
parsed.events,
without_progress(parsed.events),
vec![ReplyEvent::ToolActivity {
label: "outil".into()
}]
@ -936,7 +949,7 @@ mod tests {
)
.unwrap();
assert_eq!(
r.events,
without_progress(r.events),
vec![ReplyEvent::ToolActivity {
label: "reasoning".into()
}]
@ -945,7 +958,7 @@ mod tests {
codex::parse_event(r#"{"type":"item.completed","item":{"id":"i1","type":"command"}}"#)
.unwrap();
assert_eq!(
c.events,
without_progress(c.events),
vec![ReplyEvent::ToolActivity {
label: "command".into()
}]
@ -1138,7 +1151,7 @@ mod tests {
let events: Vec<_> = s.send("x").await.expect("send").collect();
assert_eq!(
events,
without_progress(events.clone()),
vec![
ReplyEvent::Announcement {
text: "je regarde".into()
@ -1169,7 +1182,7 @@ mod tests {
let events: Vec<_> = s.send("x").await.expect("send").collect();
assert_eq!(
events,
without_progress(events.clone()),
vec![ReplyEvent::Final {
content: "résultat".into()
}]
@ -1237,7 +1250,7 @@ mod tests {
"le tap live publie chaque agent_message, y compris celui qui deviendra Final"
);
assert_eq!(
events,
without_progress(events.clone()),
vec![ReplyEvent::Final {
content: "résultat".into()
}],
@ -1630,7 +1643,7 @@ mod tests {
);
let events: Vec<ReplyEvent> = session.send("x").await.expect("send ok").collect();
assert_eq!(
events,
without_progress(events.clone()),
vec![
// L'init `system` émet un heartbeat (vivacité non terminale, lot 1).
ReplyEvent::Heartbeat,
@ -2449,7 +2462,7 @@ mod tests {
);
let events: Vec<ReplyEvent> = session.send("x").await.expect("send ok").collect();
assert_eq!(
events,
without_progress(events.clone()),
vec![
ReplyEvent::Heartbeat,
ReplyEvent::RateLimited {

33
crates/infrastructure/src/session/openai_compat.rs
View File

@ -15,7 +15,8 @@ use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use domain::ports::{
AgentSession, AgentSessionError, ReplyEvent, ReplyStream, ToolInvoker, ToolSpec,
AgentSession, AgentSessionError, ReplyEvent, ReplyProgress, ReplyProgressKind,
ReplyProgressSource, ReplyProgressStage, ReplyStream, ToolInvoker, ToolSpec,
};
use domain::profile::{HttpChatConfig, StructuredAdapter};
use domain::SessionId;
@ -461,6 +462,13 @@ impl OpenAiCompatibleSession {
}));
for call in tool_calls {
let started = local_tool_progress(
ReplyProgressStage::Started,
format!("appel MCP {}", call.name),
&call.name,
);
send_tap(tap, &started);
events.push(started);
let event = ReplyEvent::ToolActivity {
label: call.name.clone(),
};
@ -473,6 +481,13 @@ impl OpenAiCompatibleSession {
.unwrap_or_else(|e| format!("Tool invocation failed: {e}")),
None => "Tool invocation unavailable".to_owned(),
};
let completed = local_tool_progress(
ReplyProgressStage::Completed,
format!("MCP {} terminé", call.name),
&call.name,
);
send_tap(tap, &completed);
events.push(completed);
self.transcript.lock().expect("mutex sain").push(json!({
"role": "tool",
"tool_call_id": call.id,
@ -639,6 +654,22 @@ fn send_tap(tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>, event: &ReplyEven
}
}
fn local_tool_progress(
stage: ReplyProgressStage,
label: impl Into<String>,
tool_name: &str,
) -> ReplyEvent {
ReplyEvent::Progress {
progress: ReplyProgress::new(
ReplyProgressSource::IdeaLocal,
ReplyProgressKind::Mcp,
stage,
label,
)
.with_tool_name(tool_name.to_owned()),
}
}
fn should_retry_without_tools(err: &AgentSessionError, using_tools: bool) -> bool {
using_tools && matches!(err, AgentSessionError::Start(_))
}

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

@ -16,7 +16,10 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::Mutex as AsyncMutex;
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::ports::{
AgentSession, AgentSessionError, ReplyEvent, ReplyProgress, ReplyProgressKind,
ReplyProgressSource, ReplyProgressStage, ReplyStream,
};
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
use domain::SessionId;
@ -187,12 +190,37 @@ fn parse_jsonl_turn_scoped(
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) => {
@ -222,6 +250,16 @@ fn parse_jsonl_turn_scoped(
Ok(events)
}
fn provider_progress(
kind: ReplyProgressKind,
stage: ReplyProgressStage,
label: impl Into<String>,
native_event: impl Into<String>,
) -> ReplyProgress {
ReplyProgress::new(ReplyProgressSource::ProviderNative, kind, stage, label)
.with_provider_event("opencode", native_event)
}
fn current_turn_records(
lines: &[String],
expected_session_id: Option<&str>,
@ -539,6 +577,13 @@ mod tests {
use super::*;
fn without_progress(events: Vec<ReplyEvent>) -> Vec<ReplyEvent> {
events
.into_iter()
.filter(|event| !matches!(event, ReplyEvent::Progress { .. }))
.collect()
}
#[test]
fn split_command_prefix_handles_quotes_without_shell() {
assert_eq!(
@ -557,7 +602,7 @@ mod tests {
])
.unwrap();
assert_eq!(
events,
without_progress(events),
vec![
ReplyEvent::Heartbeat,
ReplyEvent::TextDelta {
@ -624,7 +669,7 @@ mod tests {
let lines: Vec<String> = opencode_script().into_iter().map(str::to_owned).collect();
let events = parse_jsonl_turn(&lines).unwrap();
assert_eq!(
events,
without_progress(events),
vec![
ReplyEvent::Heartbeat,
ReplyEvent::TextDelta {
@ -647,7 +692,7 @@ mod tests {
])
.unwrap();
assert_eq!(
events,
without_progress(events),
vec![
ReplyEvent::Heartbeat,
ReplyEvent::Error {
@ -779,7 +824,7 @@ exit 1
let events = session.send("prompt").await.unwrap().collect::<Vec<_>>();
assert_eq!(
events,
without_progress(events),
vec![
ReplyEvent::Heartbeat,
ReplyEvent::TextDelta {