From aa5a4f30aee1958fb67365e2b09f33e120ac7647 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sat, 4 Jul 2026 19:50:03 +0200 Subject: [PATCH 1/3] 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!( From 50b2adfde36a97a60f08229cf3e520e40cd26693 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sat, 4 Jul 2026 19:50:13 +0200 Subject: [PATCH 2/3] feat(inter-agent): surface frontend des annonces live sur les cellules (F1-F3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Affiche les annonces d'agent en direct au-dessus des cellules, sur la base develop : - announcements: nouveau feature module — store réactif, provider d'abonnement aux événements, overlay par cible et aperçu, avec tests (announcementsStore + provider). - App.tsx: montage du provider dans l'arbre applicatif. - domain/index.ts: types partagés de l'événement d'annonce côté front. - layout: composition de l'overlay dans LayoutGrid + règle d'exclusion couverte par overlayExclusion.test.ts. - AgentsPanel: intègre l'aperçu des annonces dans la surface existante. Co-Authored-By: Claude Opus 4.8 --- frontend/src/app/App.tsx | 3 + frontend/src/domain/index.ts | 19 ++ frontend/src/features/agents/AgentsPanel.tsx | 6 + .../announcements/AnnouncementsPreview.tsx | 61 +++++ .../AnnouncementsProvider.test.tsx | 232 ++++++++++++++++++ .../announcements/AnnouncementsProvider.tsx | 217 ++++++++++++++++ .../TargetAnnouncementsOverlay.tsx | 95 +++++++ .../announcements/announcementsStore.test.ts | 169 +++++++++++++ .../announcements/announcementsStore.ts | 161 ++++++++++++ frontend/src/features/announcements/index.ts | 19 ++ frontend/src/features/layout/LayoutGrid.tsx | 45 +++- .../features/layout/overlayExclusion.test.ts | 58 +++++ 12 files changed, 1083 insertions(+), 2 deletions(-) create mode 100644 frontend/src/features/announcements/AnnouncementsPreview.tsx create mode 100644 frontend/src/features/announcements/AnnouncementsProvider.test.tsx create mode 100644 frontend/src/features/announcements/AnnouncementsProvider.tsx create mode 100644 frontend/src/features/announcements/TargetAnnouncementsOverlay.tsx create mode 100644 frontend/src/features/announcements/announcementsStore.test.ts create mode 100644 frontend/src/features/announcements/announcementsStore.ts create mode 100644 frontend/src/features/announcements/index.ts create mode 100644 frontend/src/features/layout/overlayExclusion.test.ts diff --git a/frontend/src/app/App.tsx b/frontend/src/app/App.tsx index 910bb65..3769a88 100644 --- a/frontend/src/app/App.tsx +++ b/frontend/src/app/App.tsx @@ -9,6 +9,7 @@ import { useEffect, useState } from "react"; import type { DomainEvent, HealthReport } from "@/domain"; import { ProjectsView } from "@/features/projects"; import { FirstRunWizard, ProfilesSettings } from "@/features/first-run"; +import { AnnouncementsProvider } from "@/features/announcements"; import { Button, Panel, Spinner, Toolbar } from "@/shared"; import { useGateways, shouldUseMock } from "./di"; @@ -63,6 +64,7 @@ export function App() { }, [profile]); return ( +
{/* ── Header ── */}
@@ -129,6 +131,7 @@ export function App() { )}
+ ); } diff --git a/frontend/src/domain/index.ts b/frontend/src/domain/index.ts index dff739e..c344c22 100644 --- a/frontend/src/domain/index.ts +++ b/frontend/src/domain/index.ts @@ -165,6 +165,25 @@ export type DomainEvent = agentId: string; version: number; } + | { + /** + * An intermediate assistant announcement emitted **during** an inter-agent + * structured turn (ticket #4, "annonces inter-agent"). Mirror of the backend + * `DomainEventDto::AgentAnnouncement`. Ephemeral (never persisted): the front + * folds these into a live, bounded, per-`(target, ticketId)` index that feeds + * the target-cell overlay (F3) and the requester-cell preview (F2). + * + * `requester` is `"user"` or the requesting agent's id; `target` is the agent + * being contacted; `text` is the human-readable réflexion/annonce. + */ + type: "agentAnnouncement"; + projectId: string; + requester: string; + target: string; + ticketId: string; + text: string; + atMs: number; + } | { type: "ptyOutput"; sessionId: string; bytes: number[] }; /** diff --git a/frontend/src/features/agents/AgentsPanel.tsx b/frontend/src/features/agents/AgentsPanel.tsx index 6e0763e..0416f2e 100644 --- a/frontend/src/features/agents/AgentsPanel.tsx +++ b/frontend/src/features/agents/AgentsPanel.tsx @@ -18,6 +18,7 @@ import { useState } from "react"; import { Button, Input, Panel, Spinner, cn } from "@/shared"; import { TerminalView } from "@/features/terminals/TerminalView"; +import { AnnouncementsPreview } from "@/features/announcements"; import { useDrift } from "@/features/templates/useDrift"; import { useGateways } from "@/app/di"; import { useAgents } from "./useAgents"; @@ -387,6 +388,11 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) { /> )} + {/* F2 (ticket #4): announcements this agent is waiting on while + it talks to another agent (requester == this row's id). Sits + just above the agent drop-list. */} + +
{/* Profile hot-swap selector (Chantier A). Changing the engine abandons the conversation history → confirmation. */} diff --git a/frontend/src/features/announcements/AnnouncementsPreview.tsx b/frontend/src/features/announcements/AnnouncementsPreview.tsx new file mode 100644 index 0000000..b089a8d --- /dev/null +++ b/frontend/src/features/announcements/AnnouncementsPreview.tsx @@ -0,0 +1,61 @@ +/** + * `AnnouncementsPreview` — F2 of ticket #4. + * + * The **requester** side: a small frame, shown near the agent drop-list, that + * previews the announcements the given agent is currently waiting on (it has + * contacted another agent and is receiving its réflexions live). + * + * Contract (Architect, point 7): filter on `requester == self`. The target + * thread is shared across every requester, so previewing by target alone would + * leak another requester's announcements; here `requester` is pinned to this + * agent's id. Renders nothing when the agent has no announcements in flight. + */ + +import { useRequesterAnnouncements } from "./AnnouncementsProvider"; + +export interface AnnouncementsPreviewProps { + /** The requesting agent this preview belongs to (the announcement `requester`). */ + requester: string; + /** Optional ticket scope (a single inter-agent turn). */ + ticketId?: string; + /** How many most-recent announcements to show. */ + limit?: number; +} + +export function AnnouncementsPreview({ + requester, + ticketId, + limit = 3, +}: AnnouncementsPreviewProps) { + const announcements = useRequesterAnnouncements(requester, ticketId); + if (announcements.length === 0) return null; + + // Most-recent `limit`, newest last (the list is oldest → newest). + const recent = announcements.slice(Math.max(0, announcements.length - limit)); + + return ( +
+ + + En attente d'un agent + +
    + {recent.map((a) => ( +
  • + {a.text} +
    • + ))} +
+
+ ); +} diff --git a/frontend/src/features/announcements/AnnouncementsProvider.test.tsx b/frontend/src/features/announcements/AnnouncementsProvider.test.tsx new file mode 100644 index 0000000..d4d0d74 --- /dev/null +++ b/frontend/src/features/announcements/AnnouncementsProvider.test.tsx @@ -0,0 +1,232 @@ +/** + * F2/F3 integration (ticket #4, develop base — busy is the sole F3 authority): + * + * - {@link TargetAnnouncementsOverlay} (F3) is driven by the target's **busy** + * state, NOT by the presence of announcements. It mounts on + * `agentBusyChanged { busy:true }` (or a read-model hydration snapshot) and + * retracts on `busy:false`, **including when no completion event is ever + * emitted** — the interruption/crash/rate-limit case that used to stick the + * overlay. There is no `agentTurnEvent` on this base. + * - {@link AnnouncementsPreview} (F2) shows only the caller-requester's + * announcements on a shared target thread (`requester == self`). + */ + +import { describe, it, expect } from "vitest"; +import { render, screen, act } from "@testing-library/react"; + +import type { DomainEvent } from "@/domain"; +import type { Gateways } from "@/ports"; +import { MockSystemGateway, MockWorkStateGateway } from "@/adapters/mock"; +import { DIProvider } from "@/app/di"; +import { AnnouncementsProvider } from "./AnnouncementsProvider"; +import { AnnouncementsPreview } from "./AnnouncementsPreview"; +import { TargetAnnouncementsOverlay } from "./TargetAnnouncementsOverlay"; + +function setup( + node: React.ReactNode, + opts: { workState?: MockWorkStateGateway } = {}, +) { + const system = new MockSystemGateway(); + const workState = opts.workState; + const gateways = { system, workState } as unknown as Gateways; + render( + + {node} + , + ); + return { system }; +} + +type AnnouncementEvent = Extract; + +function announcement(over: Partial = {}): AnnouncementEvent { + return { + type: "agentAnnouncement", + projectId: "p1", + requester: "agent-A", + target: "agent-T", + ticketId: "t1", + text: "je réfléchis…", + atMs: 1, + ...over, + }; +} + +function busy(agentId: string, isBusy: boolean): DomainEvent { + return { type: "agentBusyChanged", agentId, busy: isBusy }; +} + +// The provider subscribes asynchronously (onDomainEvent returns a Promise) and +// hydration awaits the read-model; flush microtasks before/after emitting. +async function flush() { + await act(async () => { + await Promise.resolve(); + await Promise.resolve(); + }); +} + +const overlay = () => screen.queryByTestId("target-announcements-overlay"); + +describe("F3 — overlay lifecycle is the target's busy state", () => { + it("mounts on busy:true and shows the streamed announcement as content", async () => { + const { system } = setup( + , + ); + await flush(); + expect(overlay()).toBeNull(); + + act(() => { + system.emit(busy("agent-T", true)); + system.emit(announcement({ target: "agent-T", text: "en cours" })); + }); + + expect(overlay()).not.toBeNull(); + expect(screen.getByText("en cours")).not.toBeNull(); + }); + + it("mounts on busy:true even before any announcement (banner only)", async () => { + const { system } = setup( + , + ); + await flush(); + + act(() => system.emit(busy("agent-T", true))); + + expect(overlay()).not.toBeNull(); + }); + + it("RETRACTS on busy:false even when NO completion event is emitted", async () => { + // The regression the arbitrage targets: a turn that ends without any + // completion signal (interruption / error / crash) must still bring the + // overlay down. On develop the only signal is busy — and it suffices. + const { system } = setup( + , + ); + await flush(); + + act(() => { + system.emit(busy("agent-T", true)); + system.emit(announcement({ target: "agent-T", text: "orphan" })); + }); + expect(overlay()).not.toBeNull(); + + act(() => system.emit(busy("agent-T", false))); + + expect(overlay()).toBeNull(); + }); + + it("clears stale content on busy:false so a next turn starts fresh", async () => { + const { system } = setup( + , + ); + await flush(); + + act(() => { + system.emit(busy("agent-T", true)); + system.emit(announcement({ target: "agent-T", text: "old" })); + system.emit(busy("agent-T", false)); + system.emit(busy("agent-T", true)); + }); + + // New turn: the previous "old" réflexion must not be shown. + expect(overlay()).not.toBeNull(); + expect(screen.queryByText("old")).toBeNull(); + }); + + it("hydrates busy from the reconciled read-model at mount (no live event)", async () => { + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("p1", { + agents: [ + { + agentId: "agent-T", + name: "Target", + profileId: "prof", + busy: { state: "busy", ticket: "t9", sinceMs: 123 }, + tickets: [], + }, + ], + conversations: [], + }); + + setup(, { + workState, + }); + await flush(); + + expect(overlay()).not.toBeNull(); + }); + + it("stays down when the read-model reports the agent idle", async () => { + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("p1", { + agents: [ + { + agentId: "agent-T", + name: "Target", + profileId: "prof", + busy: { state: "idle" }, + tickets: [], + }, + ], + conversations: [], + }); + + setup(, { + workState, + }); + await flush(); + + expect(overlay()).toBeNull(); + }); + + it("lets a live busy:false win over a stale busy hydration snapshot", async () => { + // Hydration seeds only when the agent is unknown; a later live idle event must + // still retract (live authority wins). + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("p1", { + agents: [ + { + agentId: "agent-T", + name: "Target", + profileId: "prof", + busy: { state: "busy", ticket: "t9", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + + const { system } = setup( + , + { workState }, + ); + await flush(); + expect(overlay()).not.toBeNull(); + + act(() => system.emit(busy("agent-T", false))); + expect(overlay()).toBeNull(); + }); +}); + +describe("F2 — AnnouncementsPreview (requester filter)", () => { + it("shows the requester's own announcements and hides another's", async () => { + const { system } = setup(); + await flush(); + + act(() => { + system.emit(announcement({ requester: "agent-A", text: "for-A" })); + system.emit( + announcement({ requester: "agent-B", ticketId: "t2", text: "for-B" }), + ); + }); + + expect(screen.getByText("for-A")).not.toBeNull(); + expect(screen.queryByText("for-B")).toBeNull(); + }); + + it("renders nothing when the requester has no announcements", async () => { + setup(); + await flush(); + expect(screen.queryByTestId("announcements-preview")).toBeNull(); + }); +}); diff --git a/frontend/src/features/announcements/AnnouncementsProvider.tsx b/frontend/src/features/announcements/AnnouncementsProvider.tsx new file mode 100644 index 0000000..acc3371 --- /dev/null +++ b/frontend/src/features/announcements/AnnouncementsProvider.tsx @@ -0,0 +1,217 @@ +/** + * `AnnouncementsProvider` — the live, app-wide store for inter-agent announcements + * (ticket #4, F1 wiring). Subscribes **once** to the domain-event stream via the + * {@link SystemGateway} and folds two orthogonal signals: + * + * 1. **Announcement content** — `agentAnnouncement` events appended to the + * bounded `(target, ticketId)` index (the scrolling text of the overlay F3 + * and the requester preview F2). + * 2. **Overlay lifecycle** — the per-agent **busy/idle** state, the single + * authority for mounting/retracting the target overlay (Architect arbitrage + * 2026-07-04). Fed live by `agentBusyChanged { agentId, busy }` and hydrated + * from the reconciled read-model `ProjectWorkState.agents[].busy` at mount / + * reboot (via {@link useHydrateAgentBusy}). + * + * Why busy — not "there are announcements" — drives F3: a turn can end **without** + * any completion event (interruption / error / crash / rate-limit). The mediator's + * `agentBusyChanged` `busy:false` falls at idle in *every* case, and the read-model + * is reconciled on reboot, so the overlay can never stick. A rate-limited agent + * stays `busy:true`, so its overlay persists with no special-casing. + * + * (On this `develop` base there is no canonical `agentTurnEvent`/`final` signal; + * busy is the sole lifecycle authority, which is exactly what we want.) + * + * Consumers read via {@link useTargetAnnouncements} (F3) and + * {@link useRequesterAnnouncements} (F2). The store is never persisted — it is + * rebuilt live from the event stream and the read-model snapshot. + */ + +import { + createContext, + useCallback, + useContext, + useEffect, + useMemo, + useState, + type ReactNode, +} from "react"; + +import { useGateways } from "@/app/di"; +import { + announcementsForRequester, + announcementsForTarget, + appendAnnouncement, + emptyAnnouncementIndex, + purgeAllForTarget, + type Announcement, + type AnnouncementIndex, +} from "./announcementsStore"; + +/** Per-agent busy map: `true` while the agent owns a turn (drives the overlay). */ +type BusyMap = Record; + +interface AnnouncementsStore { + /** Bounded announcement content, indexed by `(target, ticketId)`. */ + index: AnnouncementIndex; + /** Per-agent busy state — the overlay's lifecycle authority. */ + busy: BusyMap; + /** + * Seeds an agent's busy state from the read-model, but only when no live signal + * is yet known for it: live `agentBusyChanged` events always win over a + * (possibly staler) hydration snapshot. + */ + seedBusy: (agentId: string, busy: boolean) => void; +} + +const noopSeed = () => {}; +const EMPTY_INDEX = emptyAnnouncementIndex(); +const EMPTY_BUSY: BusyMap = {}; + +/** Store shape read outside a provider — a silent, inert default (never throws). */ +const DEFAULT_STORE: AnnouncementsStore = { + index: EMPTY_INDEX, + busy: EMPTY_BUSY, + seedBusy: noopSeed, +}; + +const AnnouncementsContext = createContext(null); + +export function AnnouncementsProvider({ children }: { children: ReactNode }) { + const { system } = useGateways(); + const [index, setIndex] = useState(emptyAnnouncementIndex); + const [busy, setBusy] = useState({}); + + // Hydration seed: apply only when the agent is unknown, so a live event that + // already landed (or lands before the async read-model resolves) is not clobbered. + const seedBusy = useCallback((agentId: string, value: boolean) => { + setBusy((prev) => (agentId in prev ? prev : { ...prev, [agentId]: value })); + }, []); + + useEffect(() => { + // `system` may be absent in unit tests injecting a partial gateway set. + if (!system) return; + let unsubscribe: (() => void) | undefined; + let cancelled = false; + + void system + .onDomainEvent((event) => { + if (event.type === "agentAnnouncement") { + // Content of the overlay/preview. + setIndex((prev) => + appendAnnouncement(prev, { + requester: event.requester, + target: event.target, + ticketId: event.ticketId, + text: event.text, + atMs: event.atMs, + }), + ); + } else if (event.type === "agentBusyChanged") { + // Lifecycle authority. `true` mounts the target overlay; `false` retracts + // it (idle) and clears the target's content so a next turn starts fresh — + // even when no completion event was ever emitted (the sticking case). + const { agentId, busy: isBusy } = event; + setBusy((prev) => ({ ...prev, [agentId]: isBusy })); + if (!isBusy) setIndex((prev) => purgeAllForTarget(prev, agentId)); + } + }) + .then((un) => { + if (cancelled) un(); + else unsubscribe = un; + }) + .catch(() => { + // Event relay unavailable in this environment — the store stays empty. + }); + + return () => { + cancelled = true; + unsubscribe?.(); + }; + }, [system]); + + const store = useMemo( + () => ({ index, busy, seedBusy }), + [index, busy, seedBusy], + ); + + return ( + + {children} + + ); +} + +/** The store, or an inert default when read outside a provider. */ +function useStore(): AnnouncementsStore { + return useContext(AnnouncementsContext) ?? DEFAULT_STORE; +} + +/** Whether a real provider is mounted above (hydration only runs when it is). */ +function useWithinProvider(): boolean { + return useContext(AnnouncementsContext) !== null; +} + +/** + * Announcements destined to `target` (F3): the flattened, time-ordered content + * plus `active` — whether the overlay should be mounted. `active` is the target's + * **busy** state (the lifecycle authority), NOT the mere presence of announcements. + */ +export function useTargetAnnouncements(target: string): { + announcements: Announcement[]; + active: boolean; +} { + const { index, busy } = useStore(); + return useMemo( + () => ({ + announcements: announcementsForTarget(index, target), + active: busy[target] === true, + }), + [index, busy, target], + ); +} + +/** + * Hydrates the target's busy state from the reconciled read-model at mount/reboot, + * so an overlay mounts for a turn already in flight when its cell appears — no live + * `agentBusyChanged` will replay for it. Live events subsequently win (see + * `seedBusy`). Guarded: no-op outside a provider or without a `workState` gateway. + */ +export function useHydrateAgentBusy(projectId: string, agentId: string): void { + const { workState } = useGateways(); + const { seedBusy } = useStore(); + const within = useWithinProvider(); + + useEffect(() => { + if (!within || !workState) return; + let cancelled = false; + void workState + .getProjectWorkState(projectId) + .then((state) => { + if (cancelled) return; + const agent = state.agents.find((a) => a.agentId === agentId); + if (agent) seedBusy(agentId, agent.busy.state === "busy"); + }) + .catch(() => { + // Read-model unavailable — fall back to the live event stream only. + }); + return () => { + cancelled = true; + }; + }, [within, workState, projectId, agentId, seedBusy]); +} + +/** + * Announcements the given `requester` is waiting on (F2), filtered by + * `requester == self` so a shared target thread never leaks another requester's + * announcements. Optionally scoped to a single `ticketId`. + */ +export function useRequesterAnnouncements( + requester: string, + ticketId?: string, +): Announcement[] { + const { index } = useStore(); + return useMemo( + () => announcementsForRequester(index, requester, ticketId), + [index, requester, ticketId], + ); +} diff --git a/frontend/src/features/announcements/TargetAnnouncementsOverlay.tsx b/frontend/src/features/announcements/TargetAnnouncementsOverlay.tsx new file mode 100644 index 0000000..ecf0fb8 --- /dev/null +++ b/frontend/src/features/announcements/TargetAnnouncementsOverlay.tsx @@ -0,0 +1,95 @@ +/** + * `TargetAnnouncementsOverlay` — F3 of ticket #4. + * + * Overlays the **target** agent's cell while another agent is talking to it: a + * top banner ("un agent est en train de lui parler") and, below, its live + * réflexions/annonces scrolling by. Mount/retract is driven by the target's + * **busy** state (the single lifecycle authority — `agentBusyChanged` + + * read-model hydration), so the overlay comes down at idle even when a turn ends + * without any completion event. The announcements are its *content*. + * + * This is **not** a PTY block: the inter-agent contact runs on a separate + * structured session. The overlay is a pure, non-interactive availability veil + * (`pointer-events-none`) so it never traps focus or clicks. + */ + +import { useEffect, useRef } from "react"; + +import { Spinner } from "@/shared"; +import { + useHydrateAgentBusy, + useTargetAnnouncements, +} from "./AnnouncementsProvider"; + +export interface TargetAnnouncementsOverlayProps { + /** Project hosting the agent (for read-model busy hydration at mount/reboot). */ + projectId: string; + /** The agent whose cell this overlay guards (the announcement `target`). */ + agentId: string; +} + +function formatTime(atMs: number): string { + if (!Number.isFinite(atMs) || atMs <= 0) return ""; + try { + return new Date(atMs).toLocaleTimeString(); + } catch { + return ""; + } +} + +export function TargetAnnouncementsOverlay({ + projectId, + agentId, +}: TargetAnnouncementsOverlayProps) { + // Seed busy from the reconciled read-model so a turn already in flight surfaces + // when this cell mounts (reboot / late open); live events then take over. + useHydrateAgentBusy(projectId, agentId); + const { announcements, active } = useTargetAnnouncements(agentId); + + // Follow the tail as new announcements scroll in. + const feedRef = useRef(null); + useEffect(() => { + const el = feedRef.current; + if (el) el.scrollTop = el.scrollHeight; + }, [announcements]); + + if (!active) return null; + + return ( +
+ {/* ── Banner ── */} +
+ + + Un agent est en train de lui parler… + +
+ + {/* ── Scrolling réflexions ── */} +
+ {announcements.map((a) => { + const time = formatTime(a.atMs); + return ( +
+ {time && {time}} + + {a.text} + +
+ ); + })} +
+
+ ); +} diff --git a/frontend/src/features/announcements/announcementsStore.test.ts b/frontend/src/features/announcements/announcementsStore.test.ts new file mode 100644 index 0000000..8014699 --- /dev/null +++ b/frontend/src/features/announcements/announcementsStore.test.ts @@ -0,0 +1,169 @@ +/** + * F1 — pure announcements store (ticket #4): indexation, bornage, purge per + * busy-cycle, and the requester filter that backs F2 (`requester == self` on a + * shared thread). + */ + +import { describe, it, expect } from "vitest"; + +import { + DEFAULT_ANNOUNCEMENT_CAP, + announcementsForRequester, + announcementsForTarget, + appendAnnouncement, + emptyAnnouncementIndex, + purgeAllForTarget, + targetHasActiveAnnouncements, + type AnnouncementInput, +} from "./announcementsStore"; + +function ann(over: Partial = {}): AnnouncementInput { + return { + requester: "agent-req", + target: "agent-tgt", + ticketId: "ticket-1", + text: "réflexion", + atMs: 1000, + ...over, + }; +} + +describe("appendAnnouncement — indexation", () => { + it("indexes by (target, ticketId) and assigns monotonic seq", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement(idx, ann({ text: "a", atMs: 1 })); + idx = appendAnnouncement(idx, ann({ text: "b", atMs: 2 })); + + const list = announcementsForTarget(idx, "agent-tgt"); + expect(list.map((a) => a.text)).toEqual(["a", "b"]); + expect(list.map((a) => a.seq)).toEqual([1, 2]); + expect(idx.nextSeq).toBe(3); + }); + + it("keeps separate buckets per ticket under the same target", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement(idx, ann({ ticketId: "t1", text: "x" })); + idx = appendAnnouncement(idx, ann({ ticketId: "t2", text: "y" })); + + expect(Object.keys(idx.byTarget["agent-tgt"])).toEqual(["t1", "t2"]); + expect(announcementsForTarget(idx, "agent-tgt").map((a) => a.text)).toEqual([ + "x", + "y", + ]); + }); + + it("orders a flattened target list by time then seq", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement(idx, ann({ ticketId: "t1", text: "late", atMs: 30 })); + idx = appendAnnouncement(idx, ann({ ticketId: "t2", text: "early", atMs: 10 })); + + expect(announcementsForTarget(idx, "agent-tgt").map((a) => a.text)).toEqual([ + "early", + "late", + ]); + }); + + it("does not mutate the previous index (immutability)", () => { + const idx0 = emptyAnnouncementIndex(); + const idx1 = appendAnnouncement(idx0, ann()); + expect(idx0.byTarget).toEqual({}); + expect(idx1).not.toBe(idx0); + }); +}); + +describe("appendAnnouncement — bornage", () => { + it("bounds a bucket to the cap, dropping the oldest", () => { + let idx = emptyAnnouncementIndex(); + const cap = 3; + for (let i = 0; i < 5; i++) { + idx = appendAnnouncement(idx, ann({ text: `m${i}`, atMs: i }), cap); + } + const list = announcementsForTarget(idx, "agent-tgt"); + expect(list).toHaveLength(cap); + // Oldest (m0, m1) dropped; most recent kept. + expect(list.map((a) => a.text)).toEqual(["m2", "m3", "m4"]); + }); + + it("defaults to DEFAULT_ANNOUNCEMENT_CAP", () => { + let idx = emptyAnnouncementIndex(); + for (let i = 0; i < DEFAULT_ANNOUNCEMENT_CAP + 10; i++) { + idx = appendAnnouncement(idx, ann({ text: `m${i}`, atMs: i })); + } + expect(announcementsForTarget(idx, "agent-tgt")).toHaveLength( + DEFAULT_ANNOUNCEMENT_CAP, + ); + }); +}); + +describe("purgeAllForTarget — retract au busy:false (idle, sans final)", () => { + it("drops every ticket of the target at once", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement(idx, ann({ ticketId: "t1", text: "a" })); + idx = appendAnnouncement(idx, ann({ ticketId: "t2", text: "b" })); + + idx = purgeAllForTarget(idx, "agent-tgt"); + + expect(targetHasActiveAnnouncements(idx, "agent-tgt")).toBe(false); + expect(idx.byTarget["agent-tgt"]).toBeUndefined(); + }); + + it("leaves other targets untouched", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement(idx, ann({ target: "T1", text: "x" })); + idx = appendAnnouncement(idx, ann({ target: "T2", text: "y" })); + + idx = purgeAllForTarget(idx, "T1"); + + expect(targetHasActiveAnnouncements(idx, "T1")).toBe(false); + expect(announcementsForTarget(idx, "T2").map((a) => a.text)).toEqual(["y"]); + }); + + it("is a no-op (stable reference) for an unknown target", () => { + const idx = appendAnnouncement(emptyAnnouncementIndex(), ann()); + expect(purgeAllForTarget(idx, "unknown")).toBe(idx); + }); +}); + +describe("announcementsForRequester — F2 filter (requester == self)", () => { + it("returns only the caller's announcements on a shared target thread", () => { + // Two requesters (A, B) both talk to the same target on distinct tickets: + // the target thread is shared, so filtering must be by requester. + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement( + idx, + ann({ requester: "A", ticketId: "ta", text: "for-A" }), + ); + idx = appendAnnouncement( + idx, + ann({ requester: "B", ticketId: "tb", text: "for-B" }), + ); + + expect(announcementsForRequester(idx, "A").map((a) => a.text)).toEqual([ + "for-A", + ]); + expect(announcementsForRequester(idx, "B").map((a) => a.text)).toEqual([ + "for-B", + ]); + }); + + it("scopes to a single ticket when ticketId is given", () => { + let idx = emptyAnnouncementIndex(); + idx = appendAnnouncement( + idx, + ann({ requester: "A", ticketId: "ta", text: "ta-msg", atMs: 1 }), + ); + idx = appendAnnouncement( + idx, + ann({ requester: "A", ticketId: "tb", text: "tb-msg", atMs: 2 }), + ); + + expect( + announcementsForRequester(idx, "A", "ta").map((a) => a.text), + ).toEqual(["ta-msg"]); + }); + + it("returns nothing for a requester with no announcements", () => { + const idx = appendAnnouncement(emptyAnnouncementIndex(), ann({ requester: "A" })); + expect(announcementsForRequester(idx, "someone-else")).toEqual([]); + }); +}); diff --git a/frontend/src/features/announcements/announcementsStore.ts b/frontend/src/features/announcements/announcementsStore.ts new file mode 100644 index 0000000..21f36d7 --- /dev/null +++ b/frontend/src/features/announcements/announcementsStore.ts @@ -0,0 +1,161 @@ +/** + * Pure store logic for the **inter-agent announcements** feature (ticket #4, F1). + * + * When an agent contacts another via `idea_ask_agent`, the backend streams the + * target's intermediate réflexions as `agentAnnouncement` domain events. These + * are **ephemeral** live signals — never persisted — folded here into a bounded + * index keyed by `(target, ticketId)`: + * + * - the **target cell** overlays the announcements destined to it (F3); + * - the **requester cell** previews the announcements it is waiting on (F2), + * filtered by `requester == self` — the target thread is shared across + * requesters, so without that filter one requester would see another's + * announcements (Architect contract, point 7). + * + * The overlay's mount/retract is driven by the target's **busy/idle** state (the + * single lifecycle authority — see `AnnouncementsProvider`), not by this index; + * the index only carries the scrolling *content*. It is retracted per busy-cycle + * (`purgeAllForTarget` on `busy:false`) so a fresh turn starts from a clean feed, + * and each bucket is bounded (oldest dropped first) so a long turn cannot grow it + * without limit. + * + * This module is intentionally free of React/Tauri: it is a set of immutable + * reducers + selectors, unit-tested in isolation (`announcementsStore.test.ts`). + */ + +/** A single live announcement folded from an `agentAnnouncement` event. */ +export interface Announcement { + /** Requesting party: `"user"` or the requesting agent's id. */ + requester: string; + /** Target agent being contacted. */ + target: string; + /** FIFO ticket correlating the inter-agent turn. */ + ticketId: string; + /** Human-readable réflexion/annonce text. */ + text: string; + /** Wall-clock timestamp (epoch-ms) as emitted by the backend. */ + atMs: number; + /** + * Monotonic local sequence assigned on ingest. Provides a stable React key and + * a total order even when two announcements share the same `atMs`. + */ + seq: number; +} + +/** + * Immutable announcement index. `byTarget[target][ticketId]` is a bounded list of + * announcements ordered oldest → newest. Empty ticket/target buckets are pruned, + * so `byTarget[target]` existing implies the target has ≥1 active ticket. + */ +export interface AnnouncementIndex { + byTarget: Record>; + /** Next sequence number to assign (monotonic, never reused). */ + nextSeq: number; +} + +/** Default per-`(target, ticketId)` bucket cap (Architect: ~20–50). */ +export const DEFAULT_ANNOUNCEMENT_CAP = 50; + +/** An empty index. */ +export function emptyAnnouncementIndex(): AnnouncementIndex { + return { byTarget: {}, nextSeq: 1 }; +} + +/** Fields carried by an `agentAnnouncement` event (sans the local `seq`). */ +export type AnnouncementInput = Omit; + +/** + * Appends one announcement to its `(target, ticketId)` bucket, assigning the next + * sequence and trimming the bucket to `cap` (dropping the oldest). Returns a new + * index; the input is never mutated. + */ +export function appendAnnouncement( + index: AnnouncementIndex, + input: AnnouncementInput, + cap: number = DEFAULT_ANNOUNCEMENT_CAP, +): AnnouncementIndex { + const { target, ticketId } = input; + const announcement: Announcement = { ...input, seq: index.nextSeq }; + + const targetBuckets = index.byTarget[target] ?? {}; + const prevBucket = targetBuckets[ticketId] ?? []; + const grown = [...prevBucket, announcement]; + // Keep only the last `cap` (bound the live index; oldest fall off first). + const bounded = grown.length > cap ? grown.slice(grown.length - cap) : grown; + + return { + byTarget: { + ...index.byTarget, + [target]: { ...targetBuckets, [ticketId]: bounded }, + }, + nextSeq: index.nextSeq + 1, + }; +} + +/** + * Retracts **all** tickets of `target` at once — called when the target goes idle + * (`agentBusyChanged` `busy:false`), which has no ticket to scope by. Keeps the + * overlay content ephemeral per busy-cycle: a fresh turn starts from a clean feed + * rather than showing the previous turn's stale réflexions. No-op (stable + * reference) when the target has nothing indexed. + */ +export function purgeAllForTarget( + index: AnnouncementIndex, + target: string, +): AnnouncementIndex { + if (!(target in index.byTarget)) return index; + const { [target]: _dropped, ...rest } = index.byTarget; + return { byTarget: rest, nextSeq: index.nextSeq }; +} + +/** Orders two announcements oldest → newest (by `atMs`, tie-broken by `seq`). */ +function byTime(a: Announcement, b: Announcement): number { + return a.atMs - b.atMs || a.seq - b.seq; +} + +/** + * All active announcements destined to `target`, flattened across its tickets and + * ordered oldest → newest. Empty when the target has no active turn (F3 source). + */ +export function announcementsForTarget( + index: AnnouncementIndex, + target: string, +): Announcement[] { + const buckets = index.byTarget[target]; + if (!buckets) return []; + return Object.values(buckets).flat().sort(byTime); +} + +/** Whether `target` has ≥1 active announcement (drives the overlay's mount). */ +export function targetHasActiveAnnouncements( + index: AnnouncementIndex, + target: string, +): boolean { + const buckets = index.byTarget[target]; + if (!buckets) return false; + return Object.values(buckets).some((bucket) => bucket.length > 0); +} + +/** + * All active announcements whose `requester === requester`, across every target + * and ticket, ordered oldest → newest. This is the F2 filter: because the target + * thread is **shared** across requesters, filtering by `requester` prevents one + * requester from previewing another's announcements. When `ticketId` is given the + * result is further scoped to that ticket. + */ +export function announcementsForRequester( + index: AnnouncementIndex, + requester: string, + ticketId?: string, +): Announcement[] { + const out: Announcement[] = []; + for (const buckets of Object.values(index.byTarget)) { + for (const [tid, bucket] of Object.entries(buckets)) { + if (ticketId !== undefined && tid !== ticketId) continue; + for (const ann of bucket) { + if (ann.requester === requester) out.push(ann); + } + } + } + return out.sort(byTime); +} diff --git a/frontend/src/features/announcements/index.ts b/frontend/src/features/announcements/index.ts new file mode 100644 index 0000000..869e466 --- /dev/null +++ b/frontend/src/features/announcements/index.ts @@ -0,0 +1,19 @@ +/** + * Inter-agent announcements feature (ticket #4) — live, ephemeral overlay/preview + * of the réflexions streamed while one agent talks to another. + */ + +export { + AnnouncementsProvider, + useTargetAnnouncements, + useRequesterAnnouncements, +} from "./AnnouncementsProvider"; +export { + TargetAnnouncementsOverlay, + type TargetAnnouncementsOverlayProps, +} from "./TargetAnnouncementsOverlay"; +export { + AnnouncementsPreview, + type AnnouncementsPreviewProps, +} from "./AnnouncementsPreview"; +export * from "./announcementsStore"; diff --git a/frontend/src/features/layout/LayoutGrid.tsx b/frontend/src/features/layout/LayoutGrid.tsx index 2637afa..c9ee6f6 100644 --- a/frontend/src/features/layout/LayoutGrid.tsx +++ b/frontend/src/features/layout/LayoutGrid.tsx @@ -34,9 +34,37 @@ import { } from "@/features/terminals"; import { useGateways } from "@/app/di"; import { useProjectWorkState } from "@/features/workstate/useProjectWorkState"; +import { + TargetAnnouncementsOverlay, + useTargetAnnouncements, +} from "@/features/announcements"; import { leaves, normalizeWeights, resizeAdjacent } from "./layout"; import { useLayout, type LayoutViewModel } from "./useLayout"; +/** + * Full-cell veil exclusion (ticket #4, Architect arbitrage): the write-portal + * veil (delegation injection into the PTY) and the F3 target overlay (another + * agent is talking to this one) share the same relative container and both cover + * `inset:0`. On the inter-agent injection path the write-portal `overlay` flag and + * the target's `busy` state rise *together*, which would stack two veils with + * near-duplicate banners. + * + * Rule: exactly one veil at a time, F3 has priority. So the write-portal veil is + * shown only for a pinned agent, when the portal asks for it, **and** the target + * is not busy-active (F3 owns the cell then). When busy is not yet set — the brief + * window before the mediator marks it — the write-portal veil is the fallback. + * + * Purely visual: this gates rendering only. The keystroke suspension during + * injection (`portal.isSuspended()` in `TerminalView`) is untouched. + */ +export function shouldShowWritePortalVeil( + agentPinned: boolean, + writePortalOverlay: boolean, + busyActive: boolean, +): boolean { + return agentPinned && writePortalOverlay && !busyActive; +} + interface LayoutGridProps { /** Project whose layout to render. */ projectId: string; @@ -313,6 +341,10 @@ function LeafView({ // Build the terminal opener based on whether an agent is pinned. const agentId = agent ?? null; + // F3 lifecycle state (busy) for this cell's agent — shared here so the + // write-portal veil can be mutually excluded with the F3 overlay (exactly one + // full-cell veil at a time, F3 first). Same source the overlay itself reads. + const { active: busyActive } = useTargetAnnouncements(agentId ?? ""); const agentWork = agentId ? workState?.agents.find((row) => row.agentId === agentId) : undefined; @@ -836,8 +868,9 @@ function LeafView({ /> {/* Write-portal overlay (ARCHITECTURE §20.3 step b/e): while a delegation is being injected into the agent's PTY, a grey veil with a centred - message sits above the terminal. Only ever shown for an agent cell. */} - {agentId != null && overlay && ( + message sits above the terminal. Only ever shown for an agent cell — + and never together with the F3 overlay (exactly one veil, F3 first). */} + {shouldShowWritePortalVeil(agentId != null, Boolean(overlay), busyActive) && (
)} + {/* F3 (ticket #4): while another agent is talking to THIS agent (target), + an availability veil showing its live réflexions. Driven by the target's + busy state (agentBusyChanged + read-model hydration), never by the raw + PTY — it retracts at idle even when a turn ends without a completion + event. Self-guards on `active` (busy) and renders null otherwise. */} + {agentId != null && ( + + )}
{busyNotice && (
+ agentPinned && busyActive; + +describe("shouldShowWritePortalVeil — veil exclusion", () => { + it("hides the write-portal veil while F3 is active (busy), whatever the portal says", () => { + // The nominal inter-agent injection path: overlay AND busy rise together. + expect(shouldShowWritePortalVeil(true, true, true)).toBe(false); + // Busy without a portal request: still no write-portal veil. + expect(shouldShowWritePortalVeil(true, false, true)).toBe(false); + }); + + it("shows the write-portal veil as the fallback when busy is not (yet) set", () => { + expect(shouldShowWritePortalVeil(true, true, false)).toBe(true); + }); + + it("shows no write-portal veil when the portal is idle", () => { + expect(shouldShowWritePortalVeil(true, false, false)).toBe(false); + }); + + it("never shows the write-portal veil for a plain (agent-less) cell", () => { + expect(shouldShowWritePortalVeil(false, true, true)).toBe(false); + expect(shouldShowWritePortalVeil(false, true, false)).toBe(false); + }); + + it("NEVER renders both veils simultaneously (exhaustive truth table)", () => { + for (const agentPinned of [false, true]) { + for (const overlay of [false, true]) { + for (const busyActive of [false, true]) { + const writePortal = shouldShowWritePortalVeil( + agentPinned, + overlay, + busyActive, + ); + const f3 = showsF3(agentPinned, busyActive); + // The safety invariant: the two full-cell veils are mutually exclusive. + expect(writePortal && f3).toBe(false); + // And F3 has priority: whenever it is up, the write-portal veil is down. + if (f3) expect(writePortal).toBe(false); + } + } + } + }); +}); From f0adf2dcc400e16ed4cbd74e3c65bce1fc895984 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sat, 4 Jul 2026 19:50:22 +0200 Subject: [PATCH 3/3] =?UTF-8?q?docs(memory):=20notes=20projet=20du=20red?= =?UTF-8?q?=C3=A9marrage=20des=20annonces=20live=20(ticket=20#4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Notes du chantier #4 relancé sur develop, commit séparé du code (convention d'usage) : - ticket4-restart-on-develop-ebd992e: baseline de redémarrage. - ticket4-overlay-composition-leafview: décision de composition de l'overlay dans le layout (LeafView). - ticket4-announcements-frontend-f1f2f3: cadrage de la surface frontend F1-F3. Co-Authored-By: Claude Opus 4.8 --- .../ticket4-announcements-frontend-f1f2f3.md | 18 +++++++++++++ .../ticket4-overlay-composition-leafview.md | 25 +++++++++++++++++++ .../ticket4-restart-on-develop-ebd992e.md | 20 +++++++++++++++ 3 files changed, 63 insertions(+) create mode 100644 .ideai/memory/ticket4-announcements-frontend-f1f2f3.md create mode 100644 .ideai/memory/ticket4-overlay-composition-leafview.md create mode 100644 .ideai/memory/ticket4-restart-on-develop-ebd992e.md diff --git a/.ideai/memory/ticket4-announcements-frontend-f1f2f3.md b/.ideai/memory/ticket4-announcements-frontend-f1f2f3.md new file mode 100644 index 0000000..3c90ccb --- /dev/null +++ b/.ideai/memory/ticket4-announcements-frontend-f1f2f3.md @@ -0,0 +1,18 @@ +--- +name: ticket4-announcements-frontend-f1f2f3 +description: Topologie du frontend des annonces inter-agent (store borné, preview demandeur filtré requester, overlay cible piloté par le busy/idle par agent) — réintégré sur base develop. +metadata: + type: project +--- +Branche `feature/agent-live-announcements` (base develop `ebd992e`, non commité). Frontend réintégré depuis `backup/announcements-work-20260704`. Feature `frontend/src/features/announcements/` : +- `announcementsStore.ts` : index pur `byTarget[target][ticketId]->Announcement[]`, `appendAnnouncement` borné (cap 50), `purgeAllForTarget` (au busy:false), sélecteurs target/requester. Éphémère, jamais persisté. (`purgeTurn` du backup RETIRÉ — plus de signal final sur develop.) +- `AnnouncementsProvider.tsx` (monté dans App.tsx) : 1 abonnement `system.onDomainEvent`. Deux signaux : CONTENU = `agentAnnouncement`→append ; CYCLE DE VIE = map `busy` par agent, seule autorité de montage/retrait, alimentée par `agentBusyChanged{agentId,busy}` + hydratée du read-model `ProjectWorkState.agents[].busy` via `useHydrateAgentBusy(projectId,agentId)` (seed « si absent » → event live gagne). `busy:false`→purgeAllForTarget. Hooks `useTargetAnnouncements` (active=busy[target]), `useRequesterAnnouncements`, `useHydrateAgentBusy`. Hors provider = store inerte. +- F3 `TargetAnnouncementsOverlay(projectId,agentId)` monté dans **`LayoutGrid` LeafView** (PAS ConversationCell — absent sur develop ; la cellule agent est le PTY `TerminalView`). Overlay `pointer-events-none`, z-20, au-dessus du write-portal overlay existant. F2 `AnnouncementsPreview` par ligne d'`AgentsPanel` (requester==a.id). + +ADAPTATION develop appliquée : `agentTurnEvent{final}` N'EXISTE PAS sur develop → branche de purge retirée du provider, type NON réintroduit dans domain, 2 tests agentTurnEvent supprimés. Autorité unique F3 = busy. Confirmé présents sur develop : `agentBusyChanged{agentId,busy}` (domain/index.ts) et `AgentWorkState.busy: {state:"idle"}|{state:"busy";ticket;sinceMs}`. DTO annonce backend confirmé (events.rs, test JSON à events.rs:927) : `{type:"agentAnnouncement",projectId,requester("user"|agentId),target,ticketId,text,atMs}` — mirroir ajouté dans domain/index.ts. + +Exclusion des voiles (arbitrage [[ticket4-overlay-composition-leafview]]) : le voile write-portal (injection PTY) et l'overlay F3 partagent le conteneur relatif du LeafView. Prédicat pur exporté `shouldShowWritePortalVeil(agentPinned, writePortalOverlay, busyActive) = agentPinned && overlay && !busyActive` ; F3 se gate sur le même `busyActive` (`useTargetAnnouncements(agentId).active`, remonté au LeafView). Résultat : exactement UN voile, F3 prioritaire. Purement visuel — `portal.isSuspended()`/l'injection PTY (TerminalView) INCHANGÉS. Testé exhaustivement (table de vérité) dans `overlayExclusion.test.ts`. + +Écarts tranchés à signaler : (1) F3 hébergé dans la cellule PTY (LeafView) faute de ConversationCell ; (2) busy est l'autorité brute : sur develop busy vient du mediator (délégations/inter-agent), pas du typing PTF humain, donc proxy correct, mais surveiller un éventuel over-trigger ; (3) F2 filtre requester seul (pas de ticketId par ligne) = garde anti-fuite. + +Build vert, 51 fichiers / 481 tests vitest verts. diff --git a/.ideai/memory/ticket4-overlay-composition-leafview.md b/.ideai/memory/ticket4-overlay-composition-leafview.md new file mode 100644 index 0000000..5b61ee5 --- /dev/null +++ b/.ideai/memory/ticket4-overlay-composition-leafview.md @@ -0,0 +1,25 @@ +--- +name: ticket4-overlay-composition-leafview +description: Règle de composition des deux voiles plein-cellule de LeafView (write-portal injection PTY vs F3 annonces busy) — exclusion mutuelle avec priorité F3, arbitrée pour éviter le double voile. +metadata: + type: reference +--- +Arbitrage layout Architect 2026-07-04 (branche feature/agent-live-announcements, base develop ebd992e, 481 tests verts). + +**Problème vérifié :** LeafView héberge DEUX voiles plein-cellule inset:0 dans le même conteneur relatif : +- write-portal (LayoutGrid.tsx:841, zIndex:4, rgba(0,0,0,0.45), « Un agent est en train de parler… ») = délégation injectée dans le PTY de l'agent (§20.3, flag `overlay` de useWritePortal). +- F3 TargetAnnouncementsOverlay (z-20, bg-canvas/85, « …en train de LUI parler… » + réflexions) = agent busy (agentBusyChanged mediator). +Ils rendent le MÊME événement depuis deux signaux et montent ENSEMBLE sur le chemin d'injection inter-agent (overlay+busy) → double voile empilé + bannières dupliquées = rendu cassé, chemin nominal. + +**Décision (DevFrontend, layout, zéro backend) :** exclusion mutuelle, un seul voile à la fois, priorité F3 : +- busy(agent) → F3 seul ; voile write-portal gated `agentId!=null && overlay && !busyActive`. +- sinon overlay → write-portal seul (repli fenêtre sans busy). +- jamais les deux. +Sécurité : on ne touche que le visuel ; la suspension des frappes (portal.isSuspended(), TerminalView.tsx:182) reste ; F3 est pointer-events-none. + +**Couplage point 3 :** F3 DOIT rendre la bannière sur busy même à 0 annonce (déjà: if(!active) return null, TargetAnnouncementsOverlay.tsx:56), car F3 absorbe le voile write-portal — une garde « ≥1 annonce » rouvrirait le trou d'indication. Pas de garde. + +**F2 (point 4) :** filtre requester==id de ligne suffit en V1 (isolation fil partagé) ; ticketId optionnel dormant. +**F3 hébergement (point 1) :** conteneur LeafView validé (pas de ConversationCell sur develop). + +Liens : [[ticket4-restart-on-develop-ebd992e]], [[ticket4-f3-overlay-driven-by-agentbusychanged]], [[inter-agent-live-context-shared-per-agent]]. \ No newline at end of file diff --git a/.ideai/memory/ticket4-restart-on-develop-ebd992e.md b/.ideai/memory/ticket4-restart-on-develop-ebd992e.md new file mode 100644 index 0000000..e3ab2ce --- /dev/null +++ b/.ideai/memory/ticket4-restart-on-develop-ebd992e.md @@ -0,0 +1,20 @@ +--- +name: ticket4-restart-on-develop-ebd992e +description: Cadrage du redémarrage des annonces inter-agent sur la base pré-canonique develop ebd992e — ce qui existe, l'écart live, la réutilisabilité du backup et le découpage B/F. +metadata: + type: reference +--- +Redémarrage #4 sur base réalignée main=develop=feature/background=feature/agent-live-announcements=ebd992e (pré-canonique). Vérifié sur le dépôt le 2026-07-04. + +**Signaux sur develop :** +- Émission intermédiaire : AUCUNE. codex.rs draine via run_turn BATCH (process.rs:268/321 lit ligne-à-ligne mais collecte en Vec, renvoie à EOF ; codex.rs:260 Box::new(events.into_iter())). structured.rs:266-271 jette TextDelta/ToolActivity/Heartbeat, ne renvoie que Final. +- Busy PAR AGENT : PRÉSENT ET COMPLET. DomainEvent::AgentBusyChanged{agent_id,busy} (domain/events.rs:171), autorité mediator (infrastructure/input/mod.rs:433), relay (app-tauri/events.rs:484), front event domain/index.ts:22 + read-model agents[].busy {state,ticket,sinceMs} (:220,304) réconcilié reboot. ⇒ autorité overlay F3 dispo telle quelle. +- Fix Final : PRÉSENT (62915ee) mais dégrade les agent_message intermédiaires en ToolActivity{label} en JETANT leur texte, et reste batch → inexploitable pour annonces. + +**Écart live (faisable SANS moteur canonique) :** run_turn lit déjà incrémentalement ; le batch n'est qu'un artefact de sa signature ->Vec. Ajouter un tap Option> dans run_turn/drain/drain_sandboxed (boucle existante) ; l'orchestrateur (qui détient requester/target/ticket) publie DomainEvent::AgentAnnouncement au fil de l'eau. Thread Landlock n'envoie que des lignes brutes ; publication bus hors thread. Pas de spawn_turn/AgentTurnEvent/ConversationId. + +**Réutilisabilité backup (backup/announcements-work-20260704, 691b2ce) :** frontend features/announcements/ (store borné, provider, AnnouncementsPreview=F2, TargetAnnouncementsOverlay=F3) réutilisable ; F3 sur busy = intact sur develop. Dépend d'agentAnnouncement{requester,target,ticketId,text,atMs} (absent→B) et agentTurnEvent{final} (absent, secondaire → RETIRER du store). Backend backup non réutilisable (canonique) ; contrat/attribution repris. + +**Découpage :** B1 domaine/appli (ReplyEvent::Announcement + DomainEvent::AgentAnnouncement + publish au drain) ; B2 infra = CŒUR (tap mpsc process.rs + codex émet Announcement{text} live) ; B3 relay/DTO ; F1 store (–hook final) ; F2 preview (filtre ticketId+requester==self) ; F3 overlay (réutilisé, busy seul autorité). + +Liens : [[ticket4-f3-overlay-driven-by-agentbusychanged]], [[ticket4-announcements-reintegration-topology]], [[inter-agent-announcements-feature-and-codex-final-bug]], [[inter-agent-live-context-shared-per-agent]]. \ No newline at end of file