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/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/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/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/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/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" + ); +} 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);