diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index 070b3fa..2eedfbe 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -30,7 +30,9 @@ use backend::stream::ReplayOutputBridge; use tauri::ipc::Channel; use domain::ids::SessionId; -use domain::ports::ReplyEvent; +use domain::ports::{ + ReplyEvent, ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage, +}; use crate::dto::ReplyChunk; use crate::stream::TauriChannelSink; @@ -136,6 +138,13 @@ fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize { ReplyChunk::UserPrompt { text } => text.len(), ReplyChunk::TextDelta { text } => text.len(), ReplyChunk::ToolActivity { label } => label.len(), + ReplyChunk::Progress { progress } => { + progress.label.len() + + progress.text.as_ref().map_or(0, String::len) + + progress.provider.as_ref().map_or(0, String::len) + + progress.native_event.as_ref().map_or(0, String::len) + + progress.tool_name.as_ref().map_or(0, String::len) + } ReplyChunk::Final { content } => content.len(), ReplyChunk::Error { message } => message.len(), } @@ -154,8 +163,20 @@ fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize { #[must_use] pub fn chunk_from_event(event: ReplyEvent) -> Option { match event { + ReplyEvent::Progress { progress } => Some(ReplyChunk::Progress { + progress: progress.into(), + }), ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text }), - ReplyEvent::ToolActivity { label } => Some(ReplyChunk::ToolActivity { label }), + ReplyEvent::ToolActivity { label } => Some(ReplyChunk::Progress { + progress: ReplyProgress::new( + ReplyProgressSource::ProviderNative, + ReplyProgressKind::Tool, + ReplyProgressStage::Info, + label.clone(), + ) + .with_tool_name(label) + .into(), + }), ReplyEvent::Error { message } => Some(ReplyChunk::Error { message }), ReplyEvent::Final { content } => { if content.trim().is_empty() { @@ -166,7 +187,16 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option { Some(ReplyChunk::Final { content }) } } - ReplyEvent::Announcement { .. } => None, + ReplyEvent::Announcement { text } => Some(ReplyChunk::Progress { + progress: ReplyProgress::new( + ReplyProgressSource::ProviderNative, + ReplyProgressKind::Message, + ReplyProgressStage::Delta, + "message intermédiaire", + ) + .with_text(text) + .into(), + }), ReplyEvent::Heartbeat => None, ReplyEvent::RateLimited { .. } => None, } diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 6ebc6cd..d0eb710 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -2195,14 +2195,32 @@ pub async fn agent_send( }, ); + // Open the turn stream with a live progress tap. The returned `ReplyStream` + // remains the authoritative drain to `Final`; the tap is best-effort + // observability for provider-native/local progress produced while `send` is + // still running. + let (tap_tx, tap_rx) = std::sync::mpsc::channel(); + let tap_bridge = std::sync::Arc::clone(&state.chat_bridge); + let tap_sid = sid; + let tap_pump = std::thread::spawn(move || { + for event in tap_rx { + let Some(chunk) = crate::chat::chunk_from_event(event) else { + continue; + }; + if matches!(chunk, ReplyChunk::Final { .. } | ReplyChunk::Error { .. }) { + continue; + } + let _ = tap_bridge.send_output(&tap_sid, chunk); + } + }); + // Open the turn stream. A start failure leaves the just-registered channel in // place (the cell stays attached, ready for a retry) — mirrors the PTY pump, // which only unregisters on a hard subscribe failure; here the session is // still live, so we keep the attach and surface the error. - let stream = session - .send(&prompt_for_model) - .await - .map_err(|e| ErrorDto::from(AppError::from(e)))?; + let stream_result = session.send_with_tap(&prompt_for_model, tap_tx).await; + let _ = tap_pump.join(); + let stream = stream_result.map_err(|e| ErrorDto::from(AppError::from(e)))?; // Drain the blocking reply iterator on a dedicated OS thread (the stream is a // synchronous `Iterator`, exactly like the PTY byte stream). It runs to the diff --git a/crates/app-tauri/tests/chat_bridge.rs b/crates/app-tauri/tests/chat_bridge.rs index a680d87..685bfed 100644 --- a/crates/app-tauri/tests/chat_bridge.rs +++ b/crates/app-tauri/tests/chat_bridge.rs @@ -18,7 +18,9 @@ use app_tauri_lib::chat::{ }; use app_tauri_lib::dto::ReplyChunk; use domain::ids::SessionId; -use domain::ports::ReplyEvent; +use domain::ports::{ + ReplyEvent, ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage, +}; use uuid::Uuid; /// Builds a `Channel` whose sent chunks are recorded into `sink`. @@ -62,6 +64,9 @@ fn chunk_bytes(chunk: &ReplyChunk) -> usize { ReplyChunk::UserPrompt { text } => text.len(), ReplyChunk::TextDelta { text } => text.len(), ReplyChunk::ToolActivity { label } => label.len(), + ReplyChunk::Progress { progress } => { + progress.label.len() + progress.text.as_ref().map_or(0, String::len) + } ReplyChunk::Final { content } => content.len(), ReplyChunk::Error { message } => message.len(), } @@ -81,13 +86,41 @@ fn chunk_from_event_maps_text_delta() { #[test] fn chunk_from_event_maps_tool_activity() { + let Some(ReplyChunk::Progress { progress }) = chunk_from_event(ReplyEvent::ToolActivity { + label: "reads file".into(), + }) else { + panic!("tool activity must map to canonical progress") + }; + assert_eq!(progress.label, "reads file"); assert_eq!( - chunk_from_event(ReplyEvent::ToolActivity { - label: "reads file".into() - }), - Some(ReplyChunk::ToolActivity { - label: "reads file".into() - }) + progress.kind, + app_tauri_lib::dto::ReplyProgressKindDto::Tool + ); + assert_eq!( + progress.source, + app_tauri_lib::dto::ReplyProgressSourceDto::ProviderNative + ); +} + +#[test] +fn chunk_from_event_maps_canonical_progress() { + let event = ReplyEvent::Progress { + progress: ReplyProgress::new( + ReplyProgressSource::IdeaLocal, + ReplyProgressKind::Mcp, + ReplyProgressStage::Started, + "idea_ask_agent", + ) + .with_tool_name("idea_ask_agent"), + }; + let Some(ReplyChunk::Progress { progress }) = chunk_from_event(event) else { + panic!("progress event must map to progress chunk") + }; + assert_eq!(progress.label, "idea_ask_agent"); + assert_eq!(progress.kind, app_tauri_lib::dto::ReplyProgressKindDto::Mcp); + assert_eq!( + progress.source, + app_tauri_lib::dto::ReplyProgressSourceDto::IdeaLocal ); } @@ -357,7 +390,18 @@ fn scrollback_accumulates_every_routed_chunk_in_order() { ); assert_eq!( - bridge.scrollback(&session), + bridge + .scrollback(&session) + .into_iter() + .map(|chunk| match chunk { + ReplyChunk::Progress { progress } => { + ReplyChunk::ToolActivity { + label: progress.label, + } + } + other => other, + }) + .collect::>(), vec![ delta("x"), ReplyChunk::ToolActivity { diff --git a/crates/app-tauri/tests/dto_chat.rs b/crates/app-tauri/tests/dto_chat.rs index ff140cd..ee27c35 100644 --- a/crates/app-tauri/tests/dto_chat.rs +++ b/crates/app-tauri/tests/dto_chat.rs @@ -8,7 +8,8 @@ use app_tauri_lib::dto::{ CellKind, ChatAttachmentDto, ChatAttachmentInputDto, ImportChatAttachmentsRequestDto, - ImportChatAttachmentsResponseDto, ReattachChatDto, ReplyChunk, TerminalSessionDto, + ImportChatAttachmentsResponseDto, ReattachChatDto, ReplyChunk, ReplyProgressDto, + ReplyProgressKindDto, ReplyProgressSourceDto, ReplyProgressStageDto, TerminalSessionDto, }; use application::{LaunchAgentOutput, StructuredSessionDescriptor}; use domain::project::ProjectPath; @@ -48,6 +49,37 @@ fn reply_chunk_tool_activity_serialises_exact_camel_case() { assert_eq!(v, json!({ "kind": "toolActivity", "label": "reads file" })); } +#[test] +fn reply_chunk_progress_serialises_exact_camel_case() { + let v = serde_json::to_value(ReplyChunk::Progress { + progress: ReplyProgressDto { + source: ReplyProgressSourceDto::IdeaLocal, + kind: ReplyProgressKindDto::Mcp, + stage: ReplyProgressStageDto::Started, + label: "idea_ask_agent".into(), + text: Some("vers QA".into()), + provider: None, + native_event: None, + tool_name: Some("idea_ask_agent".into()), + }, + }) + .unwrap(); + assert_eq!( + v, + json!({ + "kind": "progress", + "progress": { + "source": "ideaLocal", + "kind": "mcp", + "stage": "started", + "label": "idea_ask_agent", + "text": "vers QA", + "toolName": "idea_ask_agent" + } + }) + ); +} + #[test] fn reply_chunk_final_serialises_exact_camel_case() { let v = serde_json::to_value(ReplyChunk::Final { @@ -77,6 +109,18 @@ fn reply_chunk_round_trips_through_json_for_every_variant() { ReplyChunk::ToolActivity { label: "runs".into(), }, + ReplyChunk::Progress { + progress: ReplyProgressDto { + source: ReplyProgressSourceDto::ProviderNative, + kind: ReplyProgressKindDto::Turn, + stage: ReplyProgressStageDto::Started, + label: "tour démarré".into(), + text: None, + provider: Some("codex".into()), + native_event: Some("turn.started".into()), + tool_name: None, + }, + }, ReplyChunk::Final { content: "y".into(), }, diff --git a/crates/application/src/agent/structured.rs b/crates/application/src/agent/structured.rs index 9df7bb4..98d0af0 100644 --- a/crates/application/src/agent/structured.rs +++ b/crates/application/src/agent/structured.rs @@ -66,16 +66,25 @@ pub struct AnnouncementPublisher { impl AnnouncementPublisher { fn publish_event(&self, event: &ReplyEvent) { - if let ReplyEvent::Announcement { text } = event { - self.bus.publish(DomainEvent::AgentAnnouncement { - project_id: self.project_id, - requester: self.requester, - target: self.target, - ticket: self.ticket, - text: text.clone(), - at_ms: now_epoch_ms(), - }); - } + let text = match event { + ReplyEvent::Announcement { text } => Some(text.clone()), + ReplyEvent::Progress { progress } => progress + .text + .clone() + .or_else(|| (!progress.label.trim().is_empty()).then(|| progress.label.clone())), + _ => None, + }; + let Some(text) = text else { + return; + }; + self.bus.publish(DomainEvent::AgentAnnouncement { + project_id: self.project_id, + requester: self.requester, + target: self.target, + ticket: self.ticket, + text, + at_ms: now_epoch_ms(), + }); } } @@ -383,9 +392,11 @@ fn drain_stream_to_final( match event { ReplyEvent::Final { content } => return Ok(TurnOutcome::Completed(content)), ReplyEvent::RateLimited { resets_at_ms } => last_rate_limit = Some(resets_at_ms), - // TextDelta / Announcement / ToolActivity / Error / Heartbeat : non terminaux + // Progress / TextDelta / Announcement / ToolActivity / Error / Heartbeat : + // non terminaux // pour ce drain synchrone ; seul Final est une réponse réussie. - ReplyEvent::TextDelta { .. } + ReplyEvent::Progress { .. } + | ReplyEvent::TextDelta { .. } | ReplyEvent::Announcement { .. } | ReplyEvent::Error { .. } | ReplyEvent::ToolActivity { .. } diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index 22e9647..98fbe4f 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -17,6 +17,7 @@ use application::{ ListProjectsOutput, LiveSessionKind, LiveSessionSnapshot, OpenProjectOutput, ProjectWorkState, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, TurnPage, TurnSource, TurnView, }; +use domain::ports::{ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage}; use domain::{ AgentBusyState, PageCursor, PageDirection, Project, ProjectId, ProjectSystemPermissions, ResolvedAgentSystemPermissions, SystemPermissionSet, TurnRole, @@ -3046,6 +3047,119 @@ impl From for ChatAttachmentDto { } } +/// DTO source of a canonical non-terminal reply progress event. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum ReplyProgressSourceDto { + /// Native provider event. + ProviderNative, + /// Local IdeA observability event. + IdeaLocal, +} + +impl From for ReplyProgressSourceDto { + fn from(value: ReplyProgressSource) -> Self { + match value { + ReplyProgressSource::ProviderNative => Self::ProviderNative, + ReplyProgressSource::IdeaLocal => Self::IdeaLocal, + } + } +} + +/// DTO kind of a canonical non-terminal reply progress event. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum ReplyProgressKindDto { + /// Turn lifecycle/progress. + Turn, + /// Assistant message/progress. + Message, + /// Provider tool activity. + Tool, + /// IdeA MCP/tool activity. + Mcp, + /// Unclassified event. + Other, +} + +impl From for ReplyProgressKindDto { + fn from(value: ReplyProgressKind) -> Self { + match value { + ReplyProgressKind::Turn => Self::Turn, + ReplyProgressKind::Message => Self::Message, + ReplyProgressKind::Tool => Self::Tool, + ReplyProgressKind::Mcp => Self::Mcp, + ReplyProgressKind::Other => Self::Other, + } + } +} + +/// DTO stage of a canonical non-terminal reply progress event. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum ReplyProgressStageDto { + /// Activity started. + Started, + /// Incremental update. + Delta, + /// Activity completed. + Completed, + /// Informational point event. + Info, +} + +impl From for ReplyProgressStageDto { + fn from(value: ReplyProgressStage) -> Self { + match value { + ReplyProgressStage::Started => Self::Started, + ReplyProgressStage::Delta => Self::Delta, + ReplyProgressStage::Completed => Self::Completed, + ReplyProgressStage::Info => Self::Info, + } + } +} + +/// Canonical, provider-agnostic non-terminal progress/event DTO. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReplyProgressDto { + /// Event source. + pub source: ReplyProgressSourceDto, + /// Canonical event family. + pub kind: ReplyProgressKindDto, + /// Canonical stage. + pub stage: ReplyProgressStageDto, + /// Short display label. + pub label: String, + /// Optional display excerpt. + #[serde(skip_serializing_if = "Option::is_none")] + pub text: Option, + /// Optional provider identity. + #[serde(skip_serializing_if = "Option::is_none")] + pub provider: Option, + /// Optional native provider event name. + #[serde(skip_serializing_if = "Option::is_none")] + pub native_event: Option, + /// Optional tool/MCP name. + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_name: Option, +} + +impl From for ReplyProgressDto { + fn from(value: ReplyProgress) -> Self { + Self { + source: value.source.into(), + kind: value.kind.into(), + stage: value.stage.into(), + label: value.label, + text: value.text, + provider: value.provider, + native_event: value.native_event, + tool_name: value.tool_name, + } + } +} + /// One incremental chunk of a structured agent reply, streamed over the chat /// session's adapter-owned channel. The serialised wire twin of a /// [`domain::ports::ReplyEvent`]: the `agent_send` pump maps each turn event to @@ -3077,6 +3191,12 @@ pub enum ReplyChunk { /// The human-readable activity label. label: String, }, + /// Canonical non-terminal progress/event emitted before `Final`. + #[serde(rename_all = "camelCase")] + Progress { + /// Canonical event payload. + progress: ReplyProgressDto, + }, /// The deterministic end-of-turn chunk carrying the aggregated final content. #[serde(rename_all = "camelCase")] Final { diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 7a8cd94..7ac3ca7 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -596,6 +596,121 @@ pub enum WakeReason { }, } +/// Origine canonique d'un événement de progression intermédiaire. +/// +/// La séparation est volontairement métier-neutre : un adapter peut relayer un fait +/// natif du provider, tandis qu'IdeA peut produire sa propre observabilité locale +/// (MCP/outils orchestrés) sans prétendre que le provider l'a streamée. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplyProgressSource { + /// Événement observé dans le flux/documentation du provider. + ProviderNative, + /// Événement produit par IdeA autour d'un outil, d'un appel MCP ou d'une + /// orchestration locale. + IdeaLocal, +} + +/// Taxonomie stable des progress/events non terminaux projetables avant `Final`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplyProgressKind { + /// Début/fin/étape de tour sans contenu assistant exploitable. + Turn, + /// Message/progression textuelle intermédiaire. + Message, + /// Appel ou activité d'outil du provider. + Tool, + /// Appel MCP ou outil orchestré par IdeA. + Mcp, + /// Événement conservé sans spécialisation plus fine. + Other, +} + +/// Étape canonique d'un événement intermédiaire. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplyProgressStage { + /// Démarrage d'une activité. + Started, + /// Fragment ou mise à jour en cours. + Delta, + /// Fin d'une activité. + Completed, + /// Information ponctuelle. + Info, +} + +/// Événement de progression canonique, provider-agnostic et best-effort. +/// +/// Aucun champ n'a autorité sur la fin de tour : seul [`ReplyEvent::Final`] reste +/// terminal. `provider` / `native_event` sont des métadonnées d'observabilité, pas +/// un contrat métier à parser côté application. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ReplyProgress { + /// Origine de l'événement (provider natif ou observabilité locale IdeA). + pub source: ReplyProgressSource, + /// Famille canonique. + pub kind: ReplyProgressKind, + /// Étape canonique. + pub stage: ReplyProgressStage, + /// Libellé court affichable. + pub label: String, + /// Texte/extrait optionnel affichable, déjà nettoyé par l'adapter. + pub text: Option, + /// Identité provider optionnelle (`codex`, `claude`, `openai-compatible`, ...). + pub provider: Option, + /// Nom du type natif observé, quand disponible (`turn.started`, `tool_use`, ...). + pub native_event: Option, + /// Nom de l'outil/MCP, quand applicable. + pub tool_name: Option, +} + +impl ReplyProgress { + /// Construit un événement de progression avec les champs obligatoires. + #[must_use] + pub fn new( + source: ReplyProgressSource, + kind: ReplyProgressKind, + stage: ReplyProgressStage, + label: impl Into, + ) -> Self { + Self { + source, + kind, + stage, + label: label.into(), + text: None, + provider: None, + native_event: None, + tool_name: None, + } + } + + /// Ajoute un extrait textuel. + #[must_use] + pub fn with_text(mut self, text: impl Into) -> Self { + self.text = Some(text.into()); + self + } + + /// Ajoute les métadonnées provider natives. + #[must_use] + pub fn with_provider_event( + mut self, + provider: impl Into, + native_event: impl Into, + ) -> Self { + self.provider = Some(provider.into()); + self.native_event = Some(native_event.into()); + self + } + + /// Ajoute le nom d'outil/MCP associé. + #[must_use] + pub fn with_tool_name(mut self, tool_name: impl Into) -> Self { + self.tool_name = Some(tool_name.into()); + self + } +} + /// Un événement incrémental d'un tour de réponse d'un agent IA (ARCHITECTURE §17.1). /// /// Universel : l'adapter (Claude/Codex) traduit SON format structuré documenté @@ -604,6 +719,13 @@ pub enum WakeReason { /// frontière domaine. #[derive(Debug, Clone, PartialEq, Eq)] pub enum ReplyEvent { + /// Progress/event intermédiaire canonique et non terminal, utilisable pour + /// relayer les événements provider natifs et l'observabilité locale IdeA avant + /// le `Final`. + Progress { + /// Événement de progression normalisé. + progress: ReplyProgress, + }, /// Un fragment de texte assistant (rendu incrémental côté UI chat). TextDelta { /// Le fragment de texte. diff --git a/crates/domain/src/readiness.rs b/crates/domain/src/readiness.rs index 9749103..4a3e889 100644 --- a/crates/domain/src/readiness.rs +++ b/crates/domain/src/readiness.rs @@ -84,7 +84,8 @@ impl ReadinessPolicy { /// n'a pas rendu son `Final`), mais porteur d'un signal exploitable par /// l'application (planifier la reprise à `resets_at_ms`). L'heure de reset est /// propagée telle quelle. - /// - [`ReplyEvent::TextDelta`] / [`ReplyEvent::Announcement`] / + /// - [`ReplyEvent::Progress`] / + /// [`ReplyEvent::TextDelta`] / [`ReplyEvent::Announcement`] / /// [`ReplyEvent::Error`] / /// [`ReplyEvent::ToolActivity`] / /// [`ReplyEvent::Heartbeat`] ⇒ `None` : tous **non terminaux** (le flux @@ -96,7 +97,8 @@ impl ReadinessPolicy { ReplyEvent::RateLimited { resets_at_ms } => Some(ReadinessSignal::RateLimited { resets_at_ms: *resets_at_ms, }), - ReplyEvent::TextDelta { .. } + ReplyEvent::Progress { .. } + | ReplyEvent::TextDelta { .. } | ReplyEvent::Announcement { .. } | ReplyEvent::Error { .. } | ReplyEvent::ToolActivity { .. } @@ -108,6 +110,7 @@ impl ReadinessPolicy { #[cfg(test)] mod tests { use super::*; + use crate::ports::{ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage}; #[test] fn final_classifies_as_turn_ended() { @@ -122,6 +125,18 @@ mod tests { #[test] fn deltas_activities_and_heartbeats_are_non_terminal() { + assert_eq!( + ReadinessPolicy::classify(&ReplyEvent::Progress { + progress: ReplyProgress::new( + ReplyProgressSource::ProviderNative, + ReplyProgressKind::Turn, + ReplyProgressStage::Started, + "tour démarré", + ) + }), + None, + "un progress provider est projetable avant Final mais ne termine JAMAIS le tour" + ); assert_eq!( ReadinessPolicy::classify(&ReplyEvent::TextDelta { text: "x".into() }), None diff --git a/crates/infrastructure/src/session/claude.rs b/crates/infrastructure/src/session/claude.rs index f6dbeac..26271c7 100644 --- a/crates/infrastructure/src/session/claude.rs +++ b/crates/infrastructure/src/session/claude.rs @@ -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 { 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 { .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 { events } +fn provider_progress( + kind: ReplyProgressKind, + stage: ReplyProgressStage, + label: impl Into, + native_event: impl Into, +) -> 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 diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index b7b2f49..7cbea48 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -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 { // 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 { }); } // 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 { }) } +fn provider_progress( + kind: ReplyProgressKind, + stage: ReplyProgressStage, + label: impl Into, + native_event: impl Into, +) -> 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 { diff --git a/crates/infrastructure/src/session/conformance.rs b/crates/infrastructure/src/session/conformance.rs index 7b01e8f..e68862f 100644 --- a/crates/infrastructure/src/session/conformance.rs +++ b/crates/infrastructure/src/session/conformance.rs @@ -216,7 +216,8 @@ pub(crate) mod harness { assert!( matches!( e, - ReplyEvent::TextDelta { .. } + ReplyEvent::Progress { .. } + | ReplyEvent::TextDelta { .. } | ReplyEvent::ToolActivity { .. } | ReplyEvent::Announcement { .. } | ReplyEvent::Heartbeat diff --git a/crates/infrastructure/src/session/mod.rs b/crates/infrastructure/src/session/mod.rs index e5b464f..a1a481b 100644 --- a/crates/infrastructure/src/session/mod.rs +++ b/crates/infrastructure/src/session/mod.rs @@ -65,6 +65,13 @@ mod tests { // -- Helpers ---------------------------------------------------------- + fn without_progress(events: Vec) -> Vec { + 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 = 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 = session.send("x").await.expect("send ok").collect(); assert_eq!( - events, + without_progress(events.clone()), vec![ ReplyEvent::Heartbeat, ReplyEvent::RateLimited { diff --git a/crates/infrastructure/src/session/openai_compat.rs b/crates/infrastructure/src/session/openai_compat.rs index 1d8f45e..a65999f 100644 --- a/crates/infrastructure/src/session/openai_compat.rs +++ b/crates/infrastructure/src/session/openai_compat.rs @@ -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>, event: &ReplyEven } } +fn local_tool_progress( + stage: ReplyProgressStage, + label: impl Into, + 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(_)) } diff --git a/crates/infrastructure/src/session/opencode.rs b/crates/infrastructure/src/session/opencode.rs index a7d7e2a..6191c6d 100644 --- a/crates/infrastructure/src/session/opencode.rs +++ b/crates/infrastructure/src/session/opencode.rs @@ -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, + native_event: impl Into, +) -> 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) -> Vec { + 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 = 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::>(); assert_eq!( - events, + without_progress(events), vec![ ReplyEvent::Heartbeat, ReplyEvent::TextDelta {