feat(inter-agent): backend des annonces live d'agent (B0-B3)

Redémarre le ticket #4 sur la base develop. Émission d'annonces live d'un
agent vers l'UI, distinctes du Final de délégation :

- domain: ReplyEvent::Announcement/Final et DomainEvent::AgentAnnouncement
  (events.rs), port d'émission (ports.rs), gating de readiness (readiness.rs).
- application: mapping des événements structurés en annonces
  (agent/structured.rs, agent/mod.rs, lib.rs) et relais côté orchestrateur
  (orchestrator/service.rs).
- infrastructure/session: parse des annonces + fix du Final pour Claude et
  Codex, propagé aux adaptateurs et à la conformance
  (claude.rs, codex.rs, conformance.rs, mod.rs, process.rs, sandbox_e2e.rs).
- app-tauri: relais Tauri des annonces vers le front (events.rs, chat.rs).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

crates/app-tauri/src/chat.rs

@@ -188,6 +188,7 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option<ReplyChunk> {
         ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text }),
         ReplyEvent::ToolActivity { label } => Some(ReplyChunk::ToolActivity { label }),
         ReplyEvent::Final { content } => Some(ReplyChunk::Final { content }),
+        ReplyEvent::Announcement { .. } => None,
         ReplyEvent::Heartbeat => None,
         ReplyEvent::RateLimited { .. } => None,
     }

crates/app-tauri/src/events.rs

@@ -12,6 +12,7 @@
 use serde::Serialize;
 use tauri::{AppHandle, Emitter};
 
+use domain::conversation::ConversationParty;
 use domain::events::{DomainEvent, OrchestrationSource};
 use domain::input::AgentLiveness;
 use domain::{IssueLinkKind, IssuePriority, IssueStatus};
@@ -86,6 +87,22 @@ pub enum DomainEventDto {
         /// Reply length in bytes (metric, not the payload).
         reply_len: usize,
     },
+    /// Intermediate assistant announcement emitted live during an inter-agent turn.
+    #[serde(rename_all = "camelCase")]
+    AgentAnnouncement {
+        /// Project id.
+        project_id: String,
+        /// Requester party (`"user"` or requester agent id).
+        requester: String,
+        /// Target agent id.
+        target: String,
+        /// FIFO ticket id.
+        ticket_id: String,
+        /// Announcement text.
+        text: String,
+        /// Wall-clock timestamp in epoch milliseconds.
+        at_ms: u64,
+    },
     /// An agent's liveness (alive/stalled) changed (lot 2, readiness/heartbeat). The
     /// frontend can badge a frozen agent; `"stalled"` while no proof of liveness arrived
     /// for longer than the profile's `stallAfterMs`, back to `"alive"` on a late
@@ -464,6 +481,13 @@ fn issue_link_kind_wire(kind: IssueLinkKind) -> &'static str {
     }
 }
 
+fn conversation_party_wire(party: ConversationParty) -> String {
+    match party {
+        ConversationParty::User => "user".to_owned(),
+        ConversationParty::Agent { agent_id } => agent_id.to_string(),
+    }
+}
+
 impl From<&DomainEvent> for DomainEventDto {
     fn from(e: &DomainEvent) -> Self {
         match e {
@@ -505,6 +529,21 @@ impl From<&DomainEvent> for DomainEventDto {
                 agent_id: agent_id.to_string(),
                 reply_len: *reply_len,
             },
+            DomainEvent::AgentAnnouncement {
+                project_id,
+                requester,
+                target,
+                ticket,
+                text,
+                at_ms,
+            } => Self::AgentAnnouncement {
+                project_id: project_id.to_string(),
+                requester: conversation_party_wire(*requester),
+                target: target.to_string(),
+                ticket_id: ticket.to_string(),
+                text: text.clone(),
+                at_ms: *at_ms,
+            },
             DomainEvent::AgentLivenessChanged { agent_id, liveness } => {
                 Self::AgentLivenessChanged {
                     agent_id: agent_id.to_string(),
@@ -843,6 +882,9 @@ pub fn spawn_relay(app: AppHandle, bus: &TokioBroadcastEventBus) {
 mod tests {
     use super::*;
     use domain::ids::AgentId;
+    use domain::mailbox::TicketId;
+    use domain::ProjectId;
+    use serde_json::json;
 
     fn agent(n: u128) -> AgentId {
         AgentId::from_uuid(uuid::Uuid::from_u128(n))
@@ -875,6 +917,56 @@ mod tests {
         assert_eq!(json["liveness"], "alive");
     }
 
+    #[test]
+    fn agent_announcement_relays_to_camel_case_wire_with_requester_party() {
+        let project_id = ProjectId::from_uuid(uuid::Uuid::from_u128(1));
+        let requester_agent = agent(2);
+        let target = agent(3);
+        let ticket = TicketId::from_uuid(uuid::Uuid::from_u128(4));
+
+        let human = DomainEventDto::from(&DomainEvent::AgentAnnouncement {
+            project_id,
+            requester: ConversationParty::User,
+            target,
+            ticket,
+            text: "statut humain".into(),
+            at_ms: 123_456,
+        });
+        assert_eq!(
+            serde_json::to_value(&human).unwrap(),
+            json!({
+                "type": "agentAnnouncement",
+                "projectId": project_id.to_string(),
+                "requester": "user",
+                "target": target.to_string(),
+                "ticketId": ticket.to_string(),
+                "text": "statut humain",
+                "atMs": 123456,
+            })
+        );
+
+        let agent_requester = DomainEventDto::from(&DomainEvent::AgentAnnouncement {
+            project_id,
+            requester: ConversationParty::agent(requester_agent),
+            target,
+            ticket,
+            text: "statut agent".into(),
+            at_ms: 654_321,
+        });
+        assert_eq!(
+            serde_json::to_value(&agent_requester).unwrap(),
+            json!({
+                "type": "agentAnnouncement",
+                "projectId": project_id.to_string(),
+                "requester": requester_agent.to_string(),
+                "target": target.to_string(),
+                "ticketId": ticket.to_string(),
+                "text": "statut agent",
+                "atMs": 654321,
+            })
+        );
+    }
+
     /// LS6 : un `AgentRateLimited` du domaine se relaie en DTO portant le même agent
     /// et l'heure de reset (époche-ms), et se sérialise en `"agentRateLimited"` avec
     /// `resetsAtMs` — le fait neutre que le front badge « limité jusqu'à HH:MM ».

crates/application/src/agent/mod.rs

@@ -19,8 +19,9 @@ pub(crate) use lifecycle::ReattachDecision;
 
 pub use session_limit::{AgentResumer, SessionLimitService, RESUME_PROMPT};
 pub use structured::{
-    drain_reply_stream_with_readiness, drain_with_readiness, drain_with_readiness_outcome,
+    drain_reply_stream_with_readiness, drain_with_readiness,
-    send_blocking, TurnOutcome,
+    drain_with_readiness_and_announcements, drain_with_readiness_outcome, send_blocking,
+    AnnouncementPublisher, TurnOutcome,
 };
 
 pub use catalogue::{

crates/application/src/agent/structured.rs

@@ -12,12 +12,17 @@
 //! DRY : un seul chemin de lecture (le flux). `send_blocking` n'est qu'un *consom-
 //! mateur* du même flux que la cellule chat utilise pour le rendu incrémental.
 
-use std::time::Duration;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use domain::conversation::ConversationParty;
+use domain::events::DomainEvent;
 use domain::ids::AgentId;
 use domain::input::InputMediator;
-use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
+use domain::mailbox::TicketId;
+use domain::ports::{AgentSession, AgentSessionError, EventBus, ReplyEvent, ReplyStream};
 use domain::readiness::{ReadinessPolicy, ReadinessSignal};
+use domain::ProjectId;
 
 /// Issue d'un tour drainé (ARCHITECTURE §21.2-T4, réconciliation de la limite de
 /// session avec le contrat « seul `Final` est terminal »).
@@ -44,6 +49,45 @@ pub enum TurnOutcome {
     },
 }
 
+/// Attribution applicative des annonces live d'un tour inter-agent structuré.
+#[derive(Clone)]
+pub struct AnnouncementPublisher {
+    /// Bus d'événements domaine.
+    pub bus: Arc<dyn EventBus>,
+    /// Projet hôte du rendez-vous.
+    pub project_id: ProjectId,
+    /// Partie qui a demandé le tour.
+    pub requester: ConversationParty,
+    /// Agent cible qui produit les annonces.
+    pub target: AgentId,
+    /// Ticket FIFO corrélant le tour.
+    pub ticket: TicketId,
+}
+
+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(),
+            });
+        }
+    }
+}
+
+fn now_epoch_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis()
+        .try_into()
+        .unwrap_or(u64::MAX)
+}
+
 /// Envoie `prompt` à la session vivante puis **draine le flux de réponse jusqu'au
 /// [`ReplyEvent::Final`]**, et retourne son contenu agrégé.
 ///
@@ -113,6 +157,58 @@ pub async fn drain_with_readiness(
     }
 }
 
+/// Comme [`drain_with_readiness`], mais publie les [`ReplyEvent::Announcement`]
+/// produits live par la session avec l'attribution applicative complète.
+pub async fn drain_with_readiness_and_announcements(
+    session: &dyn AgentSession,
+    prompt: &str,
+    timeout: Option<Duration>,
+    mediator: &dyn InputMediator,
+    agent: AgentId,
+    announcements: Option<AnnouncementPublisher>,
+) -> Result<String, AgentSessionError> {
+    let Some(publisher) = announcements else {
+        return drain_with_readiness(session, prompt, timeout, mediator, agent).await;
+    };
+
+    let (tap_tx, tap_rx) = std::sync::mpsc::channel();
+    let live_publisher = publisher.clone();
+    let live_pump = std::thread::spawn(move || {
+        for event in tap_rx {
+            live_publisher.publish_event(&event);
+        }
+    });
+
+    let stream_result = match timeout {
+        Some(dur) => tokio::time::timeout(dur, session.send_with_tap(prompt, tap_tx))
+            .await
+            .map_err(|_elapsed| AgentSessionError::Timeout)?,
+        None => session.send_with_tap(prompt, tap_tx).await,
+    };
+    let _ = live_pump.join();
+    let stream = stream_result?;
+
+    match drain_stream_to_final(
+        stream,
+        |event| {
+            if !matches!(event, ReplyEvent::Final { .. }) {
+                mediator.mark_alive(agent);
+            }
+            publisher.publish_event(event);
+        },
+        |signal| {
+            if signal == ReadinessSignal::TurnEnded {
+                mediator.mark_idle(agent);
+            }
+        },
+    )? {
+        TurnOutcome::Completed(content) => Ok(content),
+        TurnOutcome::RateLimited { .. } => Err(AgentSessionError::Io(
+            "le tour s'est clos en limite de session, sans contenu Final".to_string(),
+        )),
+    }
+}
+
 /// Draine un flux de réponse déjà ouvert, avec la même politique de readiness que
 /// [`drain_with_readiness`].
 ///
@@ -265,8 +361,9 @@ 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 / ToolActivity / Heartbeat : non terminaux, ignorés ici.
+            // TextDelta / Announcement / ToolActivity / Heartbeat : non terminaux.
             ReplyEvent::TextDelta { .. }
+            | ReplyEvent::Announcement { .. }
             | ReplyEvent::ToolActivity { .. }
             | ReplyEvent::Heartbeat => {}
         }
@@ -348,6 +445,20 @@ mod tests {
         }
     }
 
+    #[derive(Default)]
+    struct RecordingBus {
+        events: Mutex<Vec<DomainEvent>>,
+    }
+    impl EventBus for RecordingBus {
+        fn publish(&self, event: DomainEvent) {
+            self.events.lock().unwrap().push(event);
+        }
+
+        fn subscribe(&self) -> domain::ports::EventStream {
+            Box::new(std::iter::empty())
+        }
+    }
+
     #[tokio::test]
     async fn drain_marks_alive_on_each_non_terminal_then_idle_on_final() {
         let session = FakeSession {
@@ -374,4 +485,68 @@ mod tests {
             "un battement par événement non terminal, idle au Final (pas de battement sur le Final)"
         );
     }
+
+    #[tokio::test]
+    async fn drain_publishes_announcement_but_only_final_marks_idle_and_resolves() {
+        let session = FakeSession {
+            events: vec![
+                ReplyEvent::Announcement {
+                    text: "je travaille".into(),
+                },
+                ReplyEvent::Final {
+                    content: "fini".into(),
+                },
+            ],
+        };
+        let mediator = RecordingMediator::default();
+        let bus = Arc::new(RecordingBus::default());
+        let project_id = ProjectId::from_uuid(uuid::Uuid::from_u128(10));
+        let requester = ConversationParty::User;
+        let target = agent(2);
+        let ticket = TicketId::from_uuid(uuid::Uuid::from_u128(11));
+
+        let out = drain_with_readiness_and_announcements(
+            &session,
+            "go",
+            None,
+            &mediator,
+            target,
+            Some(AnnouncementPublisher {
+                bus: bus.clone(),
+                project_id,
+                requester,
+                target,
+                ticket,
+            }),
+        )
+        .await
+        .expect("drain ok");
+
+        assert_eq!(out, "fini");
+        assert_eq!(
+            *mediator.calls.lock().unwrap(),
+            vec!["alive", "idle"],
+            "Announcement est non terminale ; seul Final marque Idle"
+        );
+        let events = bus.events.lock().unwrap();
+        assert_eq!(events.len(), 1);
+        match &events[0] {
+            DomainEvent::AgentAnnouncement {
+                project_id: p,
+                requester: r,
+                target: t,
+                ticket: tk,
+                text,
+                at_ms,
+            } => {
+                assert_eq!(*p, project_id);
+                assert_eq!(*r, requester);
+                assert_eq!(*t, target);
+                assert_eq!(*tk, ticket);
+                assert_eq!(text, "je travaille");
+                assert!(*at_ms > 0);
+            }
+            other => panic!("événement inattendu: {other:?}"),
+        }
+    }
 }

crates/application/src/lib.rs

@@ -33,9 +33,10 @@ pub mod window;
 pub mod workstate;
 
 pub use agent::{
-    drain_reply_stream_with_readiness, drain_with_readiness, drain_with_readiness_outcome,
+    drain_reply_stream_with_readiness, drain_with_readiness,
-    reference_profile_id, reference_profiles, selectable_reference_profiles, send_blocking,
+    drain_with_readiness_and_announcements, drain_with_readiness_outcome, reference_profile_id,
-    AgentResumer, ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput,
+    reference_profiles, selectable_reference_profiles, send_blocking, AgentResumer,
+    AnnouncementPublisher, ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput,
     ConfigureProfiles, ConfigureProfilesInput, ConfigureProfilesOutput, CreateAgentFromScratch,
     CreateAgentInput, CreateAgentOutput, DeleteAgent, DeleteAgentInput, DeleteProfile,
     DeleteProfileInput, DetectProfiles, DetectProfilesInput, DetectProfilesOutput, FirstRunState,

crates/application/src/orchestrator/service.rs

@@ -36,9 +36,9 @@ use crate::conversation::RecordTurn;
 use crate::memory::HarvestMemoryFromTurn;
 
 use crate::agent::{
-    drain_with_readiness, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput,
+    drain_with_readiness_and_announcements, AnnouncementPublisher, CreateAgentFromScratch,
-    ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext,
+    CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, McpRuntime,
-    UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS,
+    ReattachDecision, UpdateAgentContext, UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS,
 };
 use crate::background::{SpawnBackgroundCommand, SpawnBackgroundCommandInput};
 use crate::error::AppError;
@@ -1815,7 +1815,21 @@ impl OrchestratorService {
         // **fenêtre d'inactivité réarmable** autour de l'attente (signe de vie = sonde de
         // transcript), pas par un timeout plat interne — un long tour unique qui progresse
         // n'est plus coupé à `turn_timeout`. Aucun `idea_reply` ne peut résoudre ce tour.
-        let drain = drain_with_readiness(session, &task, None, input.as_ref(), agent_id);
+        let announcement_publisher = self.events.as_ref().map(|bus| AnnouncementPublisher {
+            bus: Arc::clone(bus),
+            project_id: project.id,
+            requester: requester.map_or(ConversationParty::User, ConversationParty::agent),
+            target: agent_id,
+            ticket: ticket_id,
+        });
+        let drain = drain_with_readiness_and_announcements(
+            session,
+            &task,
+            None,
+            input.as_ref(),
+            agent_id,
+            announcement_publisher,
+        );
 
         // L'attente du rendez-vous structured, rendue comme `Result<String, AppError>`
         // pour être enveloppée par le watchdog. Un flux clos sans `Final` correspond

crates/domain/src/events.rs

@@ -1,6 +1,7 @@
 //! Domain events published on the [`crate::ports::EventBus`] and relayed to the
 //! presentation layer (ARCHITECTURE §3.2).
 
+use crate::conversation::ConversationParty;
 use crate::ids::{AgentId, IssueId, ProfileId, ProjectId, SessionId, SkillId, TaskId, TemplateId};
 use crate::issue::{IssueLinkKind, IssuePriority, IssueRef, IssueStatus, IssueVersion};
 use crate::mailbox::TicketId;
@@ -156,6 +157,22 @@ pub enum DomainEvent {
         /// Number of bytes in the reply content (preview metric, not the payload).
         reply_len: usize,
     },
+    /// Intermediate assistant announcement emitted live during an inter-agent
+    /// structured turn. Ephemeral observability event: never persisted, never terminal.
+    AgentAnnouncement {
+        /// Owning project.
+        project_id: ProjectId,
+        /// Human or source agent that requested the turn.
+        requester: ConversationParty,
+        /// Target agent producing the announcement.
+        target: AgentId,
+        /// FIFO ticket correlating this turn.
+        ticket: TicketId,
+        /// Announcement text.
+        text: String,
+        /// Wall-clock timestamp in epoch milliseconds.
+        at_ms: u64,
+    },
     /// An agent's process exited.
     AgentExited {
         /// The agent.

crates/domain/src/ports.rs

@@ -275,6 +275,12 @@ pub enum ReplyEvent {
         /// Libellé humain-lisible de l'activité.
         label: String,
     },
+    /// Message assistant intermédiaire complet, destiné à l'observabilité live d'un
+    /// rendez-vous inter-agent. Non terminal : seul [`ReplyEvent::Final`] clôt le tour.
+    Announcement {
+        /// Texte complet de l'annonce.
+        text: String,
+    },
     /// **Preuve de vivacité non terminale** (model-agnostique) : l'agent est
     /// toujours en train de travailler. Émis par l'adapter quand le moteur signale
     /// un battement de cœur natif **sans contenu utile** (handshake/init, fenêtre de
@@ -688,6 +694,20 @@ pub trait AgentSession: Send + Sync {
     /// communication/décodage.
     async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError>;
 
+    /// Comme [`AgentSession::send`], avec un tap optionnel d'événements live produits
+    /// pendant le tour avant que le flux batch final soit rendu. L'implémentation par
+    /// défaut préserve les sessions/fakes existants : aucun tap, puis [`send`](Self::send).
+    ///
+    /// # Errors
+    /// Identiques à [`AgentSession::send`].
+    async fn send_with_tap(
+        &self,
+        prompt: &str,
+        _tap: std::sync::mpsc::Sender<ReplyEvent>,
+    ) -> Result<ReplyStream, AgentSessionError> {
+        self.send(prompt).await
+    }
+
     /// Termine proprement la session (tue le process/SDK sous-jacent). Idempotent.
     ///
     /// # Errors

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::ToolActivity`] /
+    /// - [`ReplyEvent::TextDelta`] / [`ReplyEvent::Announcement`] /
+    ///   [`ReplyEvent::ToolActivity`] /
     ///   [`ReplyEvent::Heartbeat`] ⇒ `None` : tous **non terminaux** (le flux
     ///   continue). Un heartbeat prouve la vivacité mais ne termine pas le tour.
     #[must_use]
@@ -95,6 +96,7 @@ impl ReadinessPolicy {
                 resets_at_ms: *resets_at_ms,
             }),
             ReplyEvent::TextDelta { .. }
+            | ReplyEvent::Announcement { .. }
             | ReplyEvent::ToolActivity { .. }
             | ReplyEvent::Heartbeat => None,
         }

crates/infrastructure/src/session/claude.rs

@@ -303,8 +303,51 @@ impl AgentSession for ClaudeSdkSession {
     }
 
     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> {
+        // Incarnation « un run par tour » : aucun process long ne survit entre les
+        // tours, donc `shutdown` est intrinsèquement idempotent et sans effet.
+        Ok(())
+    }
+}
+
+impl ClaudeSdkSession {
+    async fn send_inner(
+        &self,
+        prompt: &str,
+        tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
+    ) -> Result<ReplyStream, AgentSessionError> {
         let spec = self.build_spawn_line(prompt);
-        let raw_lines = run_turn(&spec, None, self.sandbox_enforcer.as_ref()).await?;
+        let (line_tap, parser_thread) = tap.map_or((None, None), |event_tap| {
+            let (line_tx, line_rx) = std::sync::mpsc::channel::<String>();
+            let parser = std::thread::spawn(move || {
+                for line in line_rx {
+                    if let Ok(parsed) = parse_event(&line) {
+                        for event in parsed.events {
+                            if let ReplyEvent::TextDelta { text } = event {
+                                let _ = event_tap.send(ReplyEvent::Announcement { text });
+                            }
+                        }
+                    }
+                }
+            });
+            (Some(line_tx), Some(parser))
+        });
+        let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await;
+        if let Some(parser) = parser_thread {
+            let _ = parser.join();
+        }
+        let raw_lines = raw_result?;
 
         let mut events = Vec::new();
         let mut captured_id = None;
@@ -334,10 +377,4 @@ impl AgentSession for ClaudeSdkSession {
 
         Ok(Box::new(events.into_iter()))
     }
-
-    async fn shutdown(&self) -> Result<(), AgentSessionError> {
-        // Incarnation « un run par tour » : aucun process long ne survit entre les
-        // tours, donc `shutdown` est intrinsèquement idempotent et sans effet.
-        Ok(())
-    }
 }

crates/infrastructure/src/session/codex.rs

@@ -214,8 +214,51 @@ impl AgentSession for CodexExecSession {
     }
 
     async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
+        self.send_inner(prompt, None, true).await
+    }
+
+    async fn send_with_tap(
+        &self,
+        prompt: &str,
+        tap: std::sync::mpsc::Sender<ReplyEvent>,
+    ) -> Result<ReplyStream, AgentSessionError> {
+        self.send_inner(prompt, Some(tap), false).await
+    }
+
+    async fn shutdown(&self) -> Result<(), AgentSessionError> {
+        // « un run par tour » ⇒ pas de process long survivant : idempotent, no-op.
+        Ok(())
+    }
+}
+
+impl CodexExecSession {
+    async fn send_inner(
+        &self,
+        prompt: &str,
+        tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
+        include_stream_announcements: bool,
+    ) -> Result<ReplyStream, AgentSessionError> {
         let spec = self.build_spawn_line(prompt);
-        let raw_lines = run_turn(&spec, None, self.sandbox_enforcer.as_ref()).await?;
+        let (line_tap, parser_thread) = tap.map_or((None, None), |event_tap| {
+            let (line_tx, line_rx) = std::sync::mpsc::channel::<String>();
+            let parser = std::thread::spawn(move || {
+                for line in line_rx {
+                    if let Ok(parsed) = parse_event(&line) {
+                        for event in parsed.events {
+                            if let ReplyEvent::Final { content } = event {
+                                let _ = event_tap.send(ReplyEvent::Announcement { text: content });
+                            }
+                        }
+                    }
+                }
+            });
+            (Some(line_tx), Some(parser))
+        });
+        let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await;
+        if let Some(parser) = parser_thread {
+            let _ = parser.join();
+        }
+        let raw_lines = raw_result?;
 
         let mut events = Vec::new();
         let mut captured_id = None;
@@ -236,12 +279,14 @@ impl AgentSession for CodexExecSession {
                 match event {
                     ReplyEvent::Final { content } => {
                         // Un `agent_message` précédemment retenu est supersédé : il devient
-                        // une activité non terminale, on garde le plus récent en conclusion.
+                        // une annonce non terminale, on garde le plus récent en conclusion.
                         if last_final.is_some() {
-                            events.push(ReplyEvent::ToolActivity {
+                            if include_stream_announcements {
-                                label: "agent_message".to_owned(),
+                                events.push(ReplyEvent::Announcement {
+                                    text: last_final.take().expect("présent"),
                                 });
                             }
+                        }
                         last_final = Some(content);
                     }
                     other => events.push(other),
@@ -259,9 +304,4 @@ impl AgentSession for CodexExecSession {
 
         Ok(Box::new(events.into_iter()))
     }
-
-    async fn shutdown(&self) -> Result<(), AgentSessionError> {
-        // « un run par tour » ⇒ pas de process long survivant : idempotent, no-op.
-        Ok(())
-    }
 }

crates/infrastructure/src/session/conformance.rs

@@ -218,6 +218,7 @@ pub(crate) mod harness {
                     e,
                     ReplyEvent::TextDelta { .. }
                         | ReplyEvent::ToolActivity { .. }
+                        | ReplyEvent::Announcement { .. }
                         | ReplyEvent::Heartbeat
                         | ReplyEvent::RateLimited { .. }
                 ),

crates/infrastructure/src/session/mod.rs

@@ -90,7 +90,7 @@ mod tests {
     #[tokio::test]
     async fn run_turn_drains_every_line_in_order() {
         let fake = FakeCli::printing(&["ligne-1", "ligne-2", "ligne-3"]);
-        let lines = run_turn(&fake.spawn_line(), None, None)
+        let lines = run_turn(&fake.spawn_line(), None, None, None)
             .await
             .expect("run_turn réussit");
         assert_eq!(lines, vec!["ligne-1", "ligne-2", "ligne-3"]);
@@ -106,7 +106,9 @@ mod tests {
             stdin: None,
             sandbox: None,
         };
-        let err = run_turn(&spec, None, None).await.expect_err("doit échouer");
+        let err = run_turn(&spec, None, None, None)
+            .await
+            .expect_err("doit échouer");
         assert!(matches!(err, AgentSessionError::Start(_)), "vu: {err:?}");
     }
 
@@ -595,7 +597,7 @@ mod tests {
             stdin: None,
             sandbox: None,
         };
-        let err = run_turn(&spec, Some(Duration::from_millis(50)), None)
+        let err = run_turn(&spec, Some(Duration::from_millis(50)), None, None)
             .await
             .expect_err("doit expirer");
         assert!(matches!(err, AgentSessionError::Timeout), "vu: {err:?}");
@@ -858,6 +860,129 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn codex_two_agent_messages_preserve_first_as_announcement_and_last_as_final() {
+        let fake = FakeCli::printing(&[
+            r#"{"type":"thread.started","thread_id":"c"}"#,
+            r#"{"type":"item.completed","item":{"id":"i0","type":"agent_message","text":"je regarde"}}"#,
+            r#"{"type":"item.completed","item":{"id":"i1","type":"agent_message","text":"résultat"}}"#,
+        ]);
+        let s = CodexExecSession::new(
+            SessionId::new_random(),
+            fake.command(),
+            "/",
+            None,
+            Vec::new(),
+            None,
+            None,
+        );
+
+        let events: Vec<_> = s.send("x").await.expect("send").collect();
+        assert_eq!(
+            events,
+            vec![
+                ReplyEvent::Announcement {
+                    text: "je regarde".into()
+                },
+                ReplyEvent::Final {
+                    content: "résultat".into()
+                }
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn codex_single_agent_message_is_final_without_announcement() {
+        let fake = FakeCli::printing(&[
+            r#"{"type":"thread.started","thread_id":"c"}"#,
+            r#"{"type":"item.completed","item":{"id":"i0","type":"agent_message","text":"résultat"}}"#,
+        ]);
+        let s = CodexExecSession::new(
+            SessionId::new_random(),
+            fake.command(),
+            "/",
+            None,
+            Vec::new(),
+            None,
+            None,
+        );
+
+        let events: Vec<_> = s.send("x").await.expect("send").collect();
+        assert_eq!(
+            events,
+            vec![ReplyEvent::Final {
+                content: "résultat".into()
+            }]
+        );
+    }
+
+    #[tokio::test]
+    async fn codex_zero_agent_message_has_no_final() {
+        let fake = FakeCli::printing(&[
+            r#"{"type":"thread.started","thread_id":"c"}"#,
+            r#"{"type":"item.completed","item":{"id":"i0","type":"reasoning","text":"analyse"}}"#,
+        ]);
+        let s = CodexExecSession::new(
+            SessionId::new_random(),
+            fake.command(),
+            "/",
+            None,
+            Vec::new(),
+            None,
+            None,
+        );
+
+        let events: Vec<_> = s.send("x").await.expect("send").collect();
+        assert!(
+            events
+                .iter()
+                .all(|e| !matches!(e, ReplyEvent::Final { .. })),
+            "aucun agent_message => aucun Final"
+        );
+    }
+
+    #[tokio::test]
+    async fn codex_send_with_tap_emits_each_agent_message_live_as_announcement() {
+        let fake = FakeCli::printing(&[
+            r#"{"type":"thread.started","thread_id":"c"}"#,
+            r#"{"type":"item.completed","item":{"id":"i0","type":"agent_message","text":"je regarde"}}"#,
+            r#"{"type":"item.completed","item":{"id":"i1","type":"agent_message","text":"résultat"}}"#,
+        ]);
+        let s = CodexExecSession::new(
+            SessionId::new_random(),
+            fake.command(),
+            "/",
+            None,
+            Vec::new(),
+            None,
+            None,
+        );
+        let (tx, rx) = std::sync::mpsc::channel();
+
+        let events: Vec<_> = s.send_with_tap("x", tx).await.expect("send").collect();
+        let live: Vec<_> = rx.into_iter().collect();
+
+        assert_eq!(
+            live,
+            vec![
+                ReplyEvent::Announcement {
+                    text: "je regarde".into()
+                },
+                ReplyEvent::Announcement {
+                    text: "résultat".into()
+                },
+            ],
+            "le tap live publie chaque agent_message, y compris celui qui deviendra Final"
+        );
+        assert_eq!(
+            events,
+            vec![ReplyEvent::Final {
+                content: "résultat".into()
+            }],
+            "le flux final du chemin tap garde seulement le dernier Final pour le demandeur"
+        );
+    }
+
     /// LIMITE/ÉCART (à arbitrer) : un flux SANS `Final` ne provoque PAS d'erreur au
     /// niveau de l'adapter — `send()` renvoie Ok avec uniquement des deltas et AUCUN
     /// `Final`. Le §17.9 D2 mentionne « flux sans Final ⇒ Io » ; l'adapter actuel ne
@@ -892,7 +1017,9 @@ mod tests {
     #[tokio::test]
     async fn run_turn_empty_output_is_ok() {
         let fake = FakeCli::printing(&[]);
-        let lines = run_turn(&fake.spawn_line(), None, None).await.expect("ok");
+        let lines = run_turn(&fake.spawn_line(), None, None, None)
+            .await
+            .expect("ok");
         assert!(lines.is_empty());
     }
 
@@ -902,7 +1029,7 @@ mod tests {
         let fake = FakeCli::printing(&["pong"]);
         let mut spec = fake.spawn_line();
         spec.stdin = Some("ping".to_owned());
-        let lines = run_turn(&spec, None, None).await.expect("ok");
+        let lines = run_turn(&spec, None, None, None).await.expect("ok");
         assert_eq!(lines, vec!["pong"]);
     }
 
@@ -965,7 +1092,7 @@ mod tests {
             stdin: None,
             sandbox: None,
         };
-        let err = run_turn(&spec, Some(Duration::from_millis(50)), None)
+        let err = run_turn(&spec, Some(Duration::from_millis(50)), None, None)
             .await
             .expect_err("doit expirer");
         assert!(matches!(err, AgentSessionError::Timeout), "vu: {err:?}");

crates/infrastructure/src/session/process.rs

@@ -28,6 +28,7 @@
 
 use std::io;
 use std::process::Stdio;
+use std::sync::mpsc::Sender;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -82,13 +83,15 @@ pub async fn run_turn(
     spec: &SpawnLine,
     timeout: Option<Duration>,
     enforcer: Option<&Arc<dyn SandboxEnforcer>>,
+    line_tap: Option<Sender<String>>,
 ) -> Result<Vec<String>, AgentSessionError> {
     // Chemin SANDBOXÉ (Linux + plan posé + enforcer câblé) : transpose la technique
     // du PTY (`spawn_command_sandboxed`) — enforce sur un thread jetable AVANT le fork,
     // l'enfant hérite le domaine via fork+exec.
     #[cfg(target_os = "linux")]
     if let (Some(plan), Some(enforcer)) = (spec.sandbox.as_ref(), enforcer) {
-        return run_turn_sandboxed(spec, plan.clone(), Arc::clone(enforcer), timeout).await;
+        return run_turn_sandboxed(spec, plan.clone(), Arc::clone(enforcer), timeout, line_tap)
+            .await;
     }
     // Hors Linux : aucun sandboxing OS ⇒ on ignore l'enforcer (Noop de toute façon) et
     // on garde le drain async historique. `let _` évite l'avertissement « unused ».
@@ -96,11 +99,11 @@ pub async fn run_turn(
     let _ = enforcer;
 
     match timeout {
-        Some(dur) => match tokio::time::timeout(dur, drain(spec)).await {
+        Some(dur) => match tokio::time::timeout(dur, drain(spec, line_tap)).await {
             Ok(result) => result,
             Err(_elapsed) => Err(AgentSessionError::Timeout),
         },
-        None => drain(spec).await,
+        None => drain(spec, line_tap).await,
     }
 }
 
@@ -141,6 +144,7 @@ async fn run_turn_sandboxed(
     plan: SandboxPlan,
     enforcer: Arc<dyn SandboxEnforcer>,
     timeout: Option<Duration>,
+    line_tap: Option<Sender<String>>,
 ) -> Result<Vec<String>, AgentSessionError> {
     use std::sync::Mutex as StdMutex;
 
@@ -158,7 +162,9 @@ async fn run_turn_sandboxed(
 
     // Thread JETABLE : sa restriction Landlock meurt avec lui.
     std::thread::spawn(move || {
-        let result = drain_sandboxed(command, args, cwd, env, stdin, &enforcer, &plan, killer_tx);
+        let result = drain_sandboxed(
+            command, args, cwd, env, stdin, &enforcer, &plan, killer_tx, line_tap,
+        );
         // Le récepteur peut avoir abandonné (timeout) : on ignore l'erreur d'envoi.
         let _ = done_tx.send(result);
     });
@@ -211,6 +217,7 @@ fn drain_sandboxed(
     enforcer: &Arc<dyn SandboxEnforcer>,
     plan: &SandboxPlan,
     killer_tx: tokio::sync::oneshot::Sender<Arc<std::sync::Mutex<std::process::Child>>>,
+    line_tap: Option<Sender<String>>,
 ) -> Result<Vec<String>, AgentSessionError> {
     use std::io::{BufRead, BufReader as StdBufReader, Write as _};
     use std::process::{Command as StdCommand, Stdio};
@@ -267,7 +274,12 @@ fn drain_sandboxed(
     let mut collected = Vec::new();
     for line in StdBufReader::new(stdout).lines() {
         match line {
-            Ok(l) => collected.push(l),
+            Ok(l) => {
+                if let Some(tap) = &line_tap {
+                    let _ = tap.send(l.clone());
+                }
+                collected.push(l);
+            }
             Err(e) => return Err(AgentSessionError::Io(e.to_string())),
         }
     }
@@ -280,7 +292,10 @@ fn drain_sandboxed(
 }
 
 /// Cœur du drain : spawn → écriture stdin → lecture ligne-à-ligne → wait.
-async fn drain(spec: &SpawnLine) -> Result<Vec<String>, AgentSessionError> {
+async fn drain(
+    spec: &SpawnLine,
+    line_tap: Option<Sender<String>>,
+) -> Result<Vec<String>, AgentSessionError> {
     let mut cmd = Command::new(&spec.command);
     cmd.args(&spec.args)
         .stdin(Stdio::piped())
@@ -323,7 +338,12 @@ async fn drain(spec: &SpawnLine) -> Result<Vec<String>, AgentSessionError> {
     let mut collected = Vec::new();
     loop {
         match lines.next_line().await {
-            Ok(Some(line)) => collected.push(line),
+            Ok(Some(line)) => {
+                if let Some(tap) = &line_tap {
+                    let _ = tap.send(line.clone());
+                }
+                collected.push(line);
+            }
             Ok(None) => break,
             Err(e) => return Err(map_io(e)),
         }

crates/infrastructure/src/session/sandbox_e2e.rs

@@ -139,7 +139,7 @@ async fn pty_structured_run_turn_enforces_plan_end_to_end() {
     );
     let enforcer = crate::sandbox::default_enforcer();
 
-    let lines = run_turn(&spec, None, Some(&enforcer))
+    let lines = run_turn(&spec, None, Some(&enforcer), None)
         .await
         .expect("run_turn sandboxé réussit sous posture Ask");
 
@@ -182,7 +182,7 @@ async fn structured_run_turn_without_plan_does_not_sandbox() {
     let spec = writing_spawn_line(RESULT_LINE, &denied_marker, &allowed_marker, None);
     let enforcer = crate::sandbox::default_enforcer();
 
-    let lines = run_turn(&spec, None, Some(&enforcer))
+    let lines = run_turn(&spec, None, Some(&enforcer), None)
         .await
         .expect("run_turn natif réussit");
 
@@ -253,7 +253,7 @@ async fn structured_run_turn_fail_closed_no_child_on_enforce_err() {
     };
     let enforcer: Arc<dyn SandboxEnforcer> = Arc::new(AlwaysFailEnforcer);
 
-    let err = run_turn(&spec, None, Some(&enforcer))
+    let err = run_turn(&spec, None, Some(&enforcer), None)
         .await
         .expect_err("enforce Err ⇒ run_turn doit échouer (fail-closed)");
     assert!(
@@ -296,7 +296,7 @@ async fn structured_run_turn_none_plan_is_native_path() {
     let enforcer = crate::sandbox::default_enforcer();
 
     // Enforcer fourni MAIS plan None ⇒ la branche sandboxée n'est pas prise.
-    let lines = run_turn(&spec, None, Some(&enforcer))
+    let lines = run_turn(&spec, None, Some(&enforcer), None)
         .await
         .expect("chemin natif réussit");
     assert_eq!(lines, vec![RESULT_LINE.to_owned()]);
@@ -330,7 +330,7 @@ async fn structured_two_turns_disjoint_grants_are_confined() {
     let a_in = dir_a.join("in.txt");
     let a_into_b = dir_b.join("from-a.txt");
     let spec_a = writing_spawn_line(RESULT_LINE, &a_into_b, &a_in, Some(rw_plan(&dir_a)));
-    run_turn(&spec_a, None, Some(&enforcer))
+    run_turn(&spec_a, None, Some(&enforcer), None)
         .await
         .expect("tour A ok");
     assert!(
@@ -347,7 +347,7 @@ async fn structured_two_turns_disjoint_grants_are_confined() {
     let b_in = dir_b.join("in.txt");
     let b_into_a = dir_a.join("from-b.txt");
     let spec_b = writing_spawn_line(RESULT_LINE, &b_into_a, &b_in, Some(rw_plan(&dir_b)));
-    run_turn(&spec_b, None, Some(&enforcer))
+    run_turn(&spec_b, None, Some(&enforcer), None)
         .await
         .expect("tour B ok");
     assert!(
@@ -388,7 +388,12 @@ async fn structured_run_turn_timeout_under_sandbox() {
     let enforcer = crate::sandbox::default_enforcer();
 
     let started = Instant::now();
-    let err = run_turn(&spec, Some(Duration::from_millis(250)), Some(&enforcer))
+    let err = run_turn(
+        &spec,
+        Some(Duration::from_millis(250)),
+        Some(&enforcer),
+        None,
+    )
     .await
     .expect_err("doit expirer");
     let elapsed = started.elapsed();