From 9430c6505099e065210d33ae6f26cdb98d6a3d66 Mon Sep 17 00:00:00 2001 From: Blomios Date: Mon, 13 Jul 2026 13:19:11 +0200 Subject: [PATCH 1/2] fix(session-limit): brancher le handle de limite sur le chemin direct (#30) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Le handle de limite de session ne se déclenchait pas quand un agent directement adressé (dont Main) touchait sa propre limite : le ReplyEvent ::RateLimited du chemin structuré direct n'était pas relayé au service de limite. On tap désormais ReplyEvent::RateLimited dans le registre terminal vers SessionLimitService::on_rate_limited, qui émet AgentRateLimited puis AgentResumeScheduled et arme la reprise auto annulable, exactement comme le chemin délégué. Couverture QA (sortie réelle) : structured_registry_d1 12 passed, session_limit_wiring 6 passed. Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/commands.rs | 22 ++- .../app-tauri/tests/session_limit_wiring.rs | 162 +++++++++++++++++- crates/application/src/terminal/registry.rs | 12 +- .../tests/structured_registry_d1.rs | 24 ++- 4 files changed, 200 insertions(+), 20 deletions(-) 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); From 225890b57a4dfb09ec9c6f71e42522d06e672ca0 Mon Sep 17 00:00:00 2001 From: Blomios Date: Mon, 13 Jul 2026 13:19:21 +0200 Subject: [PATCH 2/2] =?UTF-8?q?fix(session-limit):=20=C3=A9mettre=20AgentR?= =?UTF-8?q?ateLimited=20sur=20le=20chemin=20d=C3=A9l=C3=A9gu=C3=A9=20(#7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sur le rendez-vous délégué (idea_ask_agent), quand la cible touchait sa limite de session, l'événement AgentRateLimited n'était pas émis : l'écart backend documenté (ticket #7/F2) laissait la surface UI sans signal de limite pour la cible déléguée. Le service orchestrateur relaie désormais la limite de la cible vers le service de limite de session, fermant l'écart en cohérence avec le chemin direct (#30). Couverture QA (sortie réelle) : orchestrator_service 63 passed, session_limit_service 15 passed, session_limit_t4 7 passed, cargo test -p application 0 failed. Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/state.rs | 4 + crates/application/src/agent/mod.rs | 4 +- crates/application/src/agent/structured.rs | 38 +++- .../application/src/orchestrator/service.rs | 58 +++++- .../application/tests/orchestrator_service.rs | 197 +++++++++++++++++- 5 files changed, 277 insertions(+), 24 deletions(-) diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index c9fa950..21b5a01 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -2232,6 +2232,10 @@ impl AppState { Arc::clone(&background_tasks_port), Arc::clone(&clock) as Arc, ) + // Limites de session sur le chemin délégué headless : quand le drain + // structuré retourne `RateLimited`, l'orchestrateur arme la reprise pour + // la cible limitée via le même service que les chemins directs. + .with_session_limits(Arc::clone(&session_limit_service)) // Producteur B8 côté agent MCP : `idea_run_in_background` passe par le // même use case que la commande Tauri, sans aller-retour frontend. .with_spawn_background_command(Arc::clone(&spawn_background_command)), diff --git a/crates/application/src/agent/mod.rs b/crates/application/src/agent/mod.rs index 7a11e5b..2da7679 100644 --- a/crates/application/src/agent/mod.rs +++ b/crates/application/src/agent/mod.rs @@ -20,8 +20,8 @@ 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_and_announcements, drain_with_readiness_outcome, send_blocking, - AnnouncementPublisher, TurnOutcome, + drain_with_readiness_and_announcements, drain_with_readiness_and_announcements_outcome, + 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 0d13d37..9f6b948 100644 --- a/crates/application/src/agent/structured.rs +++ b/crates/application/src/agent/structured.rs @@ -167,8 +167,35 @@ pub async fn drain_with_readiness_and_announcements( agent: AgentId, announcements: Option, ) -> Result { + match drain_with_readiness_and_announcements_outcome( + session, + prompt, + timeout, + mediator, + agent, + announcements, + ) + .await? + { + TurnOutcome::Completed(content) => Ok(content), + TurnOutcome::RateLimited { .. } => Err(AgentSessionError::Io( + "le tour s'est clos en limite de session, sans contenu Final".to_string(), + )), + } +} + +/// Comme [`drain_with_readiness_and_announcements`], mais conserve l'issue riche +/// [`TurnOutcome::RateLimited`] pour les appelants capables d'armer la reprise. +pub async fn drain_with_readiness_and_announcements_outcome( + 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; + return drain_with_readiness_outcome(session, prompt, timeout, mediator, agent).await; }; let (tap_tx, tap_rx) = std::sync::mpsc::channel(); @@ -188,7 +215,7 @@ pub async fn drain_with_readiness_and_announcements( let _ = live_pump.join(); let stream = stream_result?; - match drain_stream_to_final( + drain_stream_to_final( stream, |event| { if !matches!(event, ReplyEvent::Final { .. }) { @@ -201,12 +228,7 @@ pub async fn drain_with_readiness_and_announcements( 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 diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 4d2d954..003352b 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -36,9 +36,10 @@ use crate::conversation::RecordTurn; use crate::memory::HarvestMemoryFromTurn; use crate::agent::{ - drain_with_readiness_and_announcements, AnnouncementPublisher, CreateAgentFromScratch, + drain_with_readiness_and_announcements_outcome, AnnouncementPublisher, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, McpRuntime, - ReattachDecision, UpdateAgentContext, UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, + ReattachDecision, SessionLimitService, TurnOutcome, UpdateAgentContext, + UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, }; use crate::background::{SpawnBackgroundCommand, SpawnBackgroundCommandInput}; use crate::error::AppError; @@ -484,6 +485,10 @@ pub struct OrchestratorService { /// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒ /// comportement historique sans persistance de tâche (call sites non câblés). background_tasks: Option>, + /// Service applicatif des limites de session, injecté depuis la composition root. + /// `None` conserve les call sites/tests legacy ; les chemins conscients des limites + /// publient alors seulement leur erreur de tour historique. + session_limits: Option>, /// Use case B8 qui produit une tâche de fond command-backed depuis la surface /// agent MCP `idea_run_in_background`. `None` ⇒ outil non configuré. spawn_background_command: Option>, @@ -558,10 +563,18 @@ impl OrchestratorService { ask_liveness_probe: None, ask_ceiling: ASK_AGENT_CEILING, background_tasks: None, + session_limits: None, spawn_background_command: None, } } + /// Branche le service de reprise automatique des limites de session. + #[must_use] + pub fn with_session_limits(mut self, service: Arc) -> Self { + self.session_limits = Some(service); + self + } + /// Branche le producteur B8 command-backed pour `idea_run_in_background`. #[must_use] pub fn with_spawn_background_command(mut self, spawn: Arc) -> Self { @@ -1822,7 +1835,7 @@ impl OrchestratorService { target: agent_id, ticket: ticket_id, }); - let drain = drain_with_readiness_and_announcements( + let drain = drain_with_readiness_and_announcements_outcome( session, &task, None, @@ -1836,13 +1849,40 @@ impl OrchestratorService { // au no-reply du chemin headless : la cible a rendu la main sans réponse // exploitable, on expose donc l'erreur métier retryable plutôt qu'un PROCESS. let wait = async { - drain.await.map_err(|err| { - if structured_no_reply_error(&err) { - AppError::TargetReturnedNoReply(target.to_owned()) - } else { - AppError::from(err) + match drain.await { + Ok(TurnOutcome::Completed(content)) => Ok(content), + Ok(TurnOutcome::RateLimited { resets_at_ms }) => { + let conversation_id = session.conversation_id(); + 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), + // Reset connu mais conversation moteur absente : ne pas armer + // une reprise non reprenable ; surfacer le fallback humain. + (None, Some(_)) => (None, None), + }; + if let (Some(service), Some(structured)) = + (&self.session_limits, &self.structured) + { + if let Some(node_id) = structured.node_for_agent(&agent_id) { + service.on_rate_limited( + agent_id, + node_id, + conversation_id, + resets_at_ms, + ); + } + } + Err(AppError::Process( + "le tour s'est clos en limite de session, sans contenu Final".to_owned(), + )) } - }) + Err(err) if structured_no_reply_error(&err) => { + Err(AppError::TargetReturnedNoReply(target.to_owned())) + } + Err(err) => Err(AppError::from(err)), + } }; // Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu. diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index b8343f9..a4bd2d6 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -45,7 +45,8 @@ use uuid::Uuid; use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, HarvestMemoryFromTurn, LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, OrchestratorService, - SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext, UpdateLiveState, + SessionLimitService, SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext, + UpdateLiveState, }; use domain::live_state::{LiveEntry, LiveState, WorkStatus}; use domain::ports::LiveStateStore; @@ -4024,9 +4025,11 @@ async fn p6b_failing_record_does_not_degrade_ask() { use std::sync::atomic::{AtomicUsize, Ordering}; -use application::StructuredSessions; +use application::{AgentResumer, StructuredSessions}; +use domain::ids::ScheduleId; use domain::ports::{ - AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream, + AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream, ScheduledTask, + Scheduler, }; /// Session structurée fake : `send()` rend un unique [`ReplyEvent::Final`] qui **écho** @@ -4034,6 +4037,7 @@ use domain::ports::{ /// du tour **et** en `mark_idle`. Aucun `idea_reply` n'est nécessaire. struct EchoSession { id: SessionId, + conversation_id: Option, } #[async_trait] impl AgentSession for EchoSession { @@ -4041,7 +4045,7 @@ impl AgentSession for EchoSession { self.id } fn conversation_id(&self) -> Option { - None + self.conversation_id.clone() } async fn send(&self, prompt: &str) -> Result { let stream: ReplyStream = Box::new( @@ -4096,7 +4100,72 @@ impl AgentSessionFactory for CountingFactory { *n += 1; id }; - Ok(Arc::new(EchoSession { id })) + Ok(Arc::new(EchoSession { + id, + conversation_id: Some(format!("engine-{id}")), + })) + } +} + +struct RateLimitedSession { + id: SessionId, + conversation_id: Option, + resets_at_ms: i64, +} + +#[async_trait] +impl AgentSession for RateLimitedSession { + 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( + vec![ReplyEvent::RateLimited { + resets_at_ms: Some(self.resets_at_ms), + }] + .into_iter(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +#[derive(Default)] +struct NoopScheduler { + next: Mutex, +} + +impl Scheduler for NoopScheduler { + fn arm(&self, _fire_at_ms: i64, _task: ScheduledTask) -> ScheduleId { + let mut next = self.next.lock().unwrap(); + *next += 1; + ScheduleId::from_uuid(Uuid::from_u128(*next)) + } + + fn cancel(&self, _id: ScheduleId) -> bool { + true + } +} + +struct NoopResumer; + +#[async_trait] +impl AgentResumer for NoopResumer { + async fn resume( + &self, + _agent_id: AgentId, + _node_id: NodeId, + _conversation_id: Option, + _resume_prompt: &str, + ) -> Result<(), application::AppError> { + Ok(()) } } @@ -4108,6 +4177,8 @@ struct StructuredAskFixture { sessions: Arc, factory: CountingFactory, pty: FakePty, + bus: SpyBus, + session_limits: Arc, } fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { @@ -4161,6 +4232,12 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { Arc::new(pty.clone()) as Arc, )); let conversations = Arc::new(TestConversations::new()) as Arc; + let session_limits = Arc::new(SessionLimitService::new( + Arc::new(FixedMillisClock(1_000)) as Arc, + Arc::new(NoopScheduler::default()) as Arc, + Arc::new(bus.clone()), + Arc::new(NoopResumer), + )); let service = Arc::new( OrchestratorService::new( @@ -4179,6 +4256,7 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { ) .with_conversations(conversations) .with_events(Arc::new(bus.clone())) + .with_session_limits(Arc::clone(&session_limits)) // Même registre que le launcher : `ask_agent` y relit la session fraîchement // insérée par `launch_structured`. .with_structured(Arc::clone(&structured)), @@ -4190,6 +4268,8 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { sessions, factory, pty, + bus, + session_limits, } } @@ -4295,3 +4375,110 @@ async fn ask_structured_target_live_as_pty_keeps_visible_terminal() { "target still has its visible PTY session" ); } + +/// Sous-lot adjacent #7/F2 : une limite détectée pendant un rendez-vous structuré +/// doit aussi passer par `SessionLimitService::on_rate_limited` pour la cible, afin +/// de publier le flux mono-agent `AgentRateLimited` puis `AgentResumeScheduled`. +#[tokio::test] +async fn ask_structured_rate_limit_arms_target_resume_events() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + let resets_at_ms = 9_999_999; + fx.structured.insert( + Arc::new(RateLimitedSession { + id: sid(9700), + conversation_id: Some("engine-target".to_owned()), + resets_at_ms, + }), + aid(1), + nid(700), + ); + + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("a rate-limited delegated turn has no Final reply"); + assert_eq!(err.code(), "PROCESS"); + + let events = fx.bus.events(); + assert!( + events.iter().any(|event| matches!( + event, + DomainEvent::AgentRateLimited { + agent_id, + resets_at_ms: Some(t), + } if *agent_id == aid(1) && *t == resets_at_ms + )), + "target AgentRateLimited must be published, got {events:?}" + ); + assert!( + events.iter().any(|event| matches!( + event, + DomainEvent::AgentResumeScheduled { + agent_id, + fire_at_ms, + } if *agent_id == aid(1) && *fire_at_ms == resets_at_ms + )), + "target AgentResumeScheduled must be published, got {events:?}" + ); + assert!( + fx.session_limits.cancel_resume(aid(1)), + "delegated target resume must be cancellable" + ); +} + +/// Sous-lot adjacent #7/F2 : si la cible structurée est limitée avec un reset connu +/// mais sans `conversation_id` moteur, l'orchestrateur doit exposer le fallback humain +/// et ne jamais armer une reprise non reprenable. +#[tokio::test] +async fn ask_structured_rate_limit_without_conversation_uses_human_fallback_no_resume() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + fx.structured.insert( + Arc::new(RateLimitedSession { + id: sid(9701), + conversation_id: None, + resets_at_ms: 9_999_999, + }), + aid(1), + nid(700), + ); + + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("a rate-limited delegated turn has no Final reply"); + assert_eq!(err.code(), "PROCESS"); + + let events = fx.bus.events(); + assert!( + events.iter().any(|event| matches!( + event, + DomainEvent::AgentRateLimited { + agent_id, + resets_at_ms: None, + } if *agent_id == aid(1) + )), + "target AgentRateLimited(None) must be published, got {events:?}" + ); + assert!( + events.iter().any(|event| matches!( + event, + DomainEvent::AgentRateLimitSuspected { agent_id, .. } if *agent_id == aid(1) + )), + "target AgentRateLimitSuspected must be published, got {events:?}" + ); + assert!( + !events.iter().any(|event| matches!( + event, + DomainEvent::AgentResumeScheduled { agent_id, .. } if *agent_id == aid(1) + )), + "target resume must not be scheduled without engine conversation_id, got {events:?}" + ); + assert!( + !fx.session_limits.cancel_resume(aid(1)), + "no delegated resume should be armed without engine conversation_id" + ); +}