From aa5a4f30aee1958fb67365e2b09f33e120ac7647 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sat, 4 Jul 2026 19:50:03 +0200 Subject: [PATCH] feat(inter-agent): backend des annonces live d'agent (B0-B3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/app-tauri/src/chat.rs | 1 + crates/app-tauri/src/events.rs | 92 +++++++++ crates/application/src/agent/mod.rs | 5 +- crates/application/src/agent/structured.rs | 181 +++++++++++++++++- crates/application/src/lib.rs | 7 +- .../application/src/orchestrator/service.rs | 22 ++- crates/domain/src/events.rs | 17 ++ crates/domain/src/ports.rs | 20 ++ crates/domain/src/readiness.rs | 4 +- crates/infrastructure/src/session/claude.rs | 51 ++++- crates/infrastructure/src/session/codex.rs | 60 +++++- .../infrastructure/src/session/conformance.rs | 1 + crates/infrastructure/src/session/mod.rs | 139 +++++++++++++- crates/infrastructure/src/session/process.rs | 34 +++- .../infrastructure/src/session/sandbox_e2e.rs | 23 ++- 15 files changed, 605 insertions(+), 52 deletions(-) diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index daa7977..a8d8525 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -188,6 +188,7 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option { 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, } diff --git a/crates/app-tauri/src/events.rs b/crates/app-tauri/src/events.rs index e5ada69..1430a94 100644 --- a/crates/app-tauri/src/events.rs +++ b/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 ». diff --git a/crates/application/src/agent/mod.rs b/crates/application/src/agent/mod.rs index 130500a..8ba986d 100644 --- a/crates/application/src/agent/mod.rs +++ b/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, - send_blocking, TurnOutcome, + drain_reply_stream_with_readiness, drain_with_readiness, + drain_with_readiness_and_announcements, drain_with_readiness_outcome, send_blocking, + AnnouncementPublisher, TurnOutcome, }; pub use catalogue::{ diff --git a/crates/application/src/agent/structured.rs b/crates/application/src/agent/structured.rs index ab8085e..0d13d37 100644 --- a/crates/application/src/agent/structured.rs +++ b/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, + /// 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, + mediator: &dyn InputMediator, + agent: AgentId, + announcements: Option, +) -> Result { + 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>, + } + 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:?}"), + } + } } diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index f9ecc17..08d0529 100644 --- a/crates/application/src/lib.rs +++ b/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, - reference_profile_id, reference_profiles, selectable_reference_profiles, send_blocking, - AgentResumer, ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, + drain_reply_stream_with_readiness, drain_with_readiness, + drain_with_readiness_and_announcements, drain_with_readiness_outcome, reference_profile_id, + 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, diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 619211b..4d2d954 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/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, - ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext, - UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, + drain_with_readiness_and_announcements, AnnouncementPublisher, CreateAgentFromScratch, + CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, McpRuntime, + 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` // pour être enveloppée par le watchdog. Un flux clos sans `Final` correspond diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 73f4e0c..a25a0ee 100644 --- a/crates/domain/src/events.rs +++ b/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. diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index e877a23..6a588d5 100644 --- a/crates/domain/src/ports.rs +++ b/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; + /// 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, + ) -> Result { + self.send(prompt).await + } + /// Termine proprement la session (tue le process/SDK sous-jacent). Idempotent. /// /// # Errors diff --git a/crates/domain/src/readiness.rs b/crates/domain/src/readiness.rs index b44bacc..0a8078d 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::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, } diff --git a/crates/infrastructure/src/session/claude.rs b/crates/infrastructure/src/session/claude.rs index 470e9b7..f6dbeac 100644 --- a/crates/infrastructure/src/session/claude.rs +++ b/crates/infrastructure/src/session/claude.rs @@ -303,8 +303,51 @@ impl AgentSession for ClaudeSdkSession { } async fn send(&self, prompt: &str) -> Result { + self.send_inner(prompt, None).await + } + + async fn send_with_tap( + &self, + prompt: &str, + tap: std::sync::mpsc::Sender, + ) -> Result { + 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>, + ) -> Result { 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::(); + 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(()) - } } diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index 721c1fd..b3856f6 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -214,8 +214,51 @@ impl AgentSession for CodexExecSession { } async fn send(&self, prompt: &str) -> Result { + self.send_inner(prompt, None, true).await + } + + async fn send_with_tap( + &self, + prompt: &str, + tap: std::sync::mpsc::Sender, + ) -> Result { + 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>, + include_stream_announcements: bool, + ) -> Result { 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::(); + 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,11 +279,13 @@ 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 { - label: "agent_message".to_owned(), - }); + if include_stream_announcements { + events.push(ReplyEvent::Announcement { + text: last_final.take().expect("présent"), + }); + } } last_final = Some(content); } @@ -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(()) - } } diff --git a/crates/infrastructure/src/session/conformance.rs b/crates/infrastructure/src/session/conformance.rs index 5fe8d19..7b01e8f 100644 --- a/crates/infrastructure/src/session/conformance.rs +++ b/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 { .. } ), diff --git a/crates/infrastructure/src/session/mod.rs b/crates/infrastructure/src/session/mod.rs index 62c931d..391420a 100644 --- a/crates/infrastructure/src/session/mod.rs +++ b/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:?}"); diff --git a/crates/infrastructure/src/session/process.rs b/crates/infrastructure/src/session/process.rs index 9c5a6d9..9a82fa6 100644 --- a/crates/infrastructure/src/session/process.rs +++ b/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, enforcer: Option<&Arc>, + line_tap: Option>, ) -> Result, 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, timeout: Option, + line_tap: Option>, ) -> Result, 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, plan: &SandboxPlan, killer_tx: tokio::sync::oneshot::Sender>>, + line_tap: Option>, ) -> Result, 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, AgentSessionError> { +async fn drain( + spec: &SpawnLine, + line_tap: Option>, +) -> Result, 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, 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)), } diff --git a/crates/infrastructure/src/session/sandbox_e2e.rs b/crates/infrastructure/src/session/sandbox_e2e.rs index c5b5f89..f1e9556 100644 --- a/crates/infrastructure/src/session/sandbox_e2e.rs +++ b/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 = 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,9 +388,14 @@ 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)) - .await - .expect_err("doit expirer"); + let err = run_turn( + &spec, + Some(Duration::from_millis(250)), + Some(&enforcer), + None, + ) + .await + .expect_err("doit expirer"); let elapsed = started.elapsed(); assert!(