diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 94e6fb8..6875ca0 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -1652,11 +1652,10 @@ pub async fn agent_send( // `Final` event (or stream end / superseded channel), then detaches *only its // own* generation so a concurrent re-attach is never torn down. let bridge = std::sync::Arc::clone(&state.chat_bridge); - // Level-1 session-limit tap (§21, niveau 1 — structuré). Resolve the agent + host - // cell of this session once; a `RateLimited` turn event then feeds the service - // (détecter→planifier). DORMANT en composition B-2 (aucune session structurée - // vivante) mais câblé pour forward-compat. `conversation_id` non disponible ici ⇒ - // `None` (acceptable LS7). + // Level-1 session-limit tap (§21, niveau 1 — structuré). Resolve the agent, host + // cell and engine conversation once from the live registry; a `RateLimited` turn + // event then feeds the service (détecter→planifier). Without a conversation id the + // service must not pretend an automatic resume is armed. let service = std::sync::Arc::clone(&state.session_limit_service); let meta = state.structured_sessions.meta_for_session(&sid); std::thread::spawn(move || { @@ -1665,8 +1664,17 @@ pub async fn agent_send( // (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on // continue à drainer comme pour un battement. if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event { - if let Some((agent_id, node_id)) = meta { - service.on_rate_limited(agent_id, node_id, None, *resets_at_ms); + if let Some((agent_id, node_id, conversation_id)) = &meta { + let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) { + (Some(conversation_id), resets_at_ms) => { + (Some(conversation_id.clone()), *resets_at_ms) + } + (None, None) => (None, None), + // Reset connu mais conversation moteur absente : ne pas armer + // une reprise non reprenable ; surfacer le fallback humain. + (None, Some(_)) => (None, None), + }; + service.on_rate_limited(*agent_id, *node_id, conversation_id, resets_at_ms); } } // Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire diff --git a/crates/app-tauri/tests/session_limit_wiring.rs b/crates/app-tauri/tests/session_limit_wiring.rs index 12fd076..5f5a249 100644 --- a/crates/app-tauri/tests/session_limit_wiring.rs +++ b/crates/app-tauri/tests/session_limit_wiring.rs @@ -11,12 +11,14 @@ //! unknown agent (never a panic). use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use app_tauri_lib::state::AppState; +use async_trait::async_trait; use domain::events::DomainEvent; -use domain::ids::NodeId; -use domain::ports::IdGenerator; +use domain::ids::{NodeId, SessionId}; +use domain::ports::{AgentSession, AgentSessionError, IdGenerator, ReplyStream}; use domain::AgentId; use infrastructure::UuidGenerator; @@ -33,6 +35,34 @@ fn node(n: u128) -> NodeId { NodeId::from_uuid(uuid::Uuid::from_u128(n)) } +fn session(n: u128) -> SessionId { + SessionId::from_uuid(uuid::Uuid::from_u128(n)) +} + +struct LiveStructuredSession { + id: SessionId, + conversation_id: Option, +} + +#[async_trait] +impl AgentSession for LiveStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + self.conversation_id.clone() + } + + async fn send(&self, _prompt: &str) -> Result { + Ok(Box::new(std::iter::empty())) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + /// `cancel_resume` on an agent with no armed resume returns `false` (and never /// panics): the service is wired and behaves as a clean no-op. #[tokio::test] @@ -97,6 +127,134 @@ async fn on_rate_limited_arms_a_cancellable_resume_over_the_real_bus() { assert!(saw_cancelled, "AgentResumeCancelled relayed on the bus"); } +/// B1/B2 direct : le tap structuré doit résoudre `agent_id` + `node_id` + +/// `conversation_id` depuis le registre vivant avant d'appeler `on_rate_limited`. +/// Avec un reset connu et une conversation moteur connue, le service publie +/// `AgentRateLimited` puis `AgentResumeScheduled`. +#[tokio::test] +async fn direct_structured_rate_limit_uses_live_meta_and_arms_resume() { + let state = AppState::build(temp_path("appdata")); + let mut rx = state.event_bus.raw_receiver(); + let sid = session(707); + state.structured_sessions.insert( + Arc::new(LiveStructuredSession { + id: sid, + conversation_id: Some("engine-main".to_owned()), + }), + agent(7), + node(70), + ); + + let (agent_id, node_id, conversation_id) = state + .structured_sessions + .meta_for_session(&sid) + .expect("live structured session metadata"); + assert_eq!(agent_id, agent(7)); + assert_eq!(node_id, node(70)); + assert_eq!(conversation_id.as_deref(), Some("engine-main")); + + let resets_at_ms = i64::MAX; + state.session_limit_service.on_rate_limited( + agent_id, + node_id, + conversation_id, + Some(resets_at_ms), + ); + + let mut saw_rate_limited = false; + let mut saw_scheduled = false; + for _ in 0..32 { + match tokio::time::timeout(Duration::from_secs(2), rx.recv()).await { + Ok(Ok(DomainEvent::AgentRateLimited { agent_id, .. })) if agent_id == agent(7) => { + saw_rate_limited = true; + } + Ok(Ok(DomainEvent::AgentResumeScheduled { agent_id, .. })) if agent_id == agent(7) => { + saw_scheduled = true; + break; + } + Ok(Ok(_)) => continue, + _ => break, + } + } + + assert!( + saw_rate_limited, + "AgentRateLimited emitted for direct agent" + ); + assert!( + saw_scheduled, + "AgentResumeScheduled emitted for direct agent" + ); + assert!(state.session_limit_service.cancel_resume(agent(7))); +} + +/// B1/B2 direct : si le moteur donne une heure de reset mais que la session vivante +/// ne porte pas de `conversation_id`, la commande doit normaliser vers le fallback +/// humain plutôt que d'armer une reprise impossible à reprendre. +#[tokio::test] +async fn direct_structured_rate_limit_without_conversation_uses_human_fallback() { + let state = AppState::build(temp_path("appdata")); + let mut rx = state.event_bus.raw_receiver(); + let sid = session(708); + state.structured_sessions.insert( + Arc::new(LiveStructuredSession { + id: sid, + conversation_id: None, + }), + agent(7), + node(70), + ); + + let (agent_id, node_id, conversation_id) = state + .structured_sessions + .meta_for_session(&sid) + .expect("live structured session metadata"); + assert_eq!(conversation_id, None); + + let resets_at_ms = Some(i64::MAX); + let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) { + (Some(conversation_id), resets_at_ms) => (Some(conversation_id), resets_at_ms), + (None, None) => (None, None), + (None, Some(_)) => (None, None), + }; + state + .session_limit_service + .on_rate_limited(agent_id, node_id, conversation_id, resets_at_ms); + + let mut saw_rate_limited_none = false; + let mut saw_suspected = false; + let mut saw_scheduled = false; + for _ in 0..32 { + match tokio::time::timeout(Duration::from_millis(50), rx.recv()).await { + Ok(Ok(DomainEvent::AgentRateLimited { + agent_id, + resets_at_ms: None, + })) if agent_id == agent(7) => saw_rate_limited_none = true, + Ok(Ok(DomainEvent::AgentRateLimitSuspected { agent_id, .. })) + if agent_id == agent(7) => + { + saw_suspected = true; + } + Ok(Ok(DomainEvent::AgentResumeScheduled { agent_id, .. })) if agent_id == agent(7) => { + saw_scheduled = true; + } + Ok(Ok(_)) => continue, + _ => break, + } + } + + assert!( + saw_rate_limited_none, + "fallback exposes AgentRateLimited(None)" + ); + assert!(saw_suspected, "fallback exposes human input state"); + assert!( + !saw_scheduled, + "no AgentResumeScheduled without engine conversation_id" + ); + assert!(!state.session_limit_service.cancel_resume(agent(7))); +} + /// `set_resume_at` (filet humain niveau 3) résout l'agent→cellule dans les registries /// des sessions vivantes ; **sans cellule vivante**, la résolution échoue ⇒ la commande /// renvoie `NOT_FOUND` sans armer quoi que ce soit (pas d'armement orphelin). diff --git a/crates/application/src/terminal/registry.rs b/crates/application/src/terminal/registry.rs index d357775..40cfb79 100644 --- a/crates/application/src/terminal/registry.rs +++ b/crates/application/src/terminal/registry.rs @@ -464,7 +464,7 @@ impl StructuredSessions { }) } - /// Résout les coordonnées `(agent_id, node_id)` d'une session structurée par son + /// Résout les coordonnées `(agent_id, node_id, conversation_id)` d'une session structurée par son /// [`SessionId`] (LS7, tap niveau 1 des limites de session, §21.10). /// /// Jumeau « inverse » de [`Self::live_agents`] : là où `live_agents` énumère tout, @@ -473,11 +473,11 @@ impl StructuredSessions { /// passer à [`SessionLimitService::on_rate_limited`](crate::SessionLimitService) sur /// un signal `RateLimited`. `None` si l'id n'est pas (ou plus) une session vivante. #[must_use] - pub fn meta_for_session(&self, id: &SessionId) -> Option<(AgentId, NodeId)> { - self.entries - .lock() - .ok() - .and_then(|m| m.get(id).map(|e| (e.agent_id, e.node_id))) + pub fn meta_for_session(&self, id: &SessionId) -> Option<(AgentId, NodeId, Option)> { + self.entries.lock().ok().and_then(|m| { + m.get(id) + .map(|e| (e.agent_id, e.node_id, e.session.conversation_id())) + }) } /// Liste chaque agent IA vivant, sa cellule hôte et son id de session. diff --git a/crates/application/tests/structured_registry_d1.rs b/crates/application/tests/structured_registry_d1.rs index a35a3c1..4f87679 100644 --- a/crates/application/tests/structured_registry_d1.rs +++ b/crates/application/tests/structured_registry_d1.rs @@ -41,6 +41,7 @@ fn nid(n: u128) -> NodeId { /// `send`/`shutdown` ne sont pas exercés ici. struct FakeSession { id: SessionId, + conversation_id: Option, } #[async_trait] @@ -49,7 +50,7 @@ impl AgentSession for FakeSession { self.id } fn conversation_id(&self) -> Option { - None + self.conversation_id.clone() } async fn send(&self, _prompt: &str) -> Result { let stream: ReplyStream = Box::new(std::iter::empty()); @@ -61,7 +62,17 @@ impl AgentSession for FakeSession { } fn fake(id: SessionId) -> Arc { - Arc::new(FakeSession { id }) + Arc::new(FakeSession { + id, + conversation_id: None, + }) +} + +fn fake_with_conversation(id: SessionId, conversation_id: &str) -> Arc { + Arc::new(FakeSession { + id, + conversation_id: Some(conversation_id.to_owned()), + }) } // =========================================================================== @@ -107,16 +118,19 @@ fn structured_insert_resolve_and_remove() { } #[test] -fn structured_meta_for_session_resolves_agent_and_node() { +fn structured_meta_for_session_resolves_agent_node_and_conversation() { // LS7 (§21.10) : le tap niveau 1 n'a que le `SessionId` du tour et doit retrouver // l'agent + la cellule hôte à passer au service de limite. Lookup direct par id. let reg = StructuredSessions::new(); let s = sid(1); let a = aid(10); let n = nid(100); - reg.insert(fake(s), a, n); + reg.insert(fake_with_conversation(s, "conv-live"), a, n); - assert_eq!(reg.meta_for_session(&s), Some((a, n))); + assert_eq!( + reg.meta_for_session(&s), + Some((a, n, Some("conv-live".to_owned()))) + ); // Id inconnu (ou retiré) ⇒ None (jamais de panique sur une session morte). assert_eq!(reg.meta_for_session(&sid(999)), None);