fix(session-limit): brancher le handle de limite sur le chemin direct (#30)

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 <noreply@anthropic.com>
This commit is contained in:
2026-07-13 13:19:11 +02:00
parent 631dd4eccf
commit 9430c65050
4 changed files with 200 additions and 20 deletions

View File

@ -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<String>,
}
#[async_trait]
impl AgentSession for LiveStructuredSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
self.conversation_id.clone()
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
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).