diff --git a/crates/app-tauri/Cargo.toml b/crates/app-tauri/Cargo.toml index 96040c3..e3fa15d 100644 --- a/crates/app-tauri/Cargo.toml +++ b/crates/app-tauri/Cargo.toml @@ -32,6 +32,8 @@ serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } uuid = { workspace = true } +# `AppAgentResumer` implements the application's async `AgentResumer` port (LS7). +async-trait = { workspace = true } # Cross-OS local IPC for the MCP loopback transport (M5a): Unix domain socket # (Linux/macOS) + Windows named pipe behind one async API, no network port. Pulled # in only here (the transport is an app-tauri/infra concern); its `tokio` feature diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 38c8b90..77c94b7 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -1137,6 +1137,11 @@ pub async fn launch_agent( // artefacts from previous AppImages do not survive into this activation. state.reconcile_claude_run_dirs(&project).await; + // Session-limit resume context (LS7, §21.5): record the Project + cell size keyed by + // agent so an auto-resume (which only carries agent/node/conversation) can recompose a + // full `LaunchAgentInput`. Cloned here — `project` is moved into the launch below. + let resume_project = project.clone(); + let output = state .launch_agent .execute(LaunchAgentInput { @@ -1153,7 +1158,21 @@ pub async fn launch_agent( .await .map_err(ErrorDto::from)?; + if let Ok(mut contexts) = state.resume_contexts.lock() { + contexts.insert( + agent_id, + crate::state::ResumeContext { + project: resume_project, + rows: request.rows, + cols: request.cols, + }, + ); + } + let session_id = output.session.id; + // Host cell of the freshly (re)launched session — the pivot the level-2 tap reports + // to the service alongside the agent. + let host_node_id = output.session.node_id; // §17.4/§17.6: a structured launch routes to an `AgentSession`, registered // **only** in `StructuredSessions` — its id is never a live PTY. Such a cell is @@ -1166,6 +1185,18 @@ pub async fn launch_agent( // Register the xterm output channel for this session. let gen = state.pty_bridge.register(session_id, on_output); + // Level-2 session-limit detector (§21, niveau 2 — PTY/TUI sans adapter + // structuré). Selection rule (§21.10-4) is the single infra source `applies`: + // arm a parser **only** for a profile without a structured adapter that declares + // a `rate_limit_pattern` (anti-double-détection avec le niveau 1). A `None` + // profile (reattach/idempotent) or an invalid regex ⇒ no parser, jamais de panique. + let rate_limit_parser = output + .profile + .as_ref() + .filter(|p| infrastructure::ratelimit::applies(p)) + .and_then(|p| p.rate_limit_pattern.as_ref()) + .and_then(infrastructure::RateLimitParser::new); + // Subscribe to the PTY's byte stream and pump it to the channel. // The stream is a blocking iterator; it runs on a dedicated OS thread and // ends when the PTY hits EOF or this attach is superseded. @@ -1173,8 +1204,30 @@ pub async fn launch_agent( match state.pty_port.subscribe_output(&handle) { Ok(stream) => { let bridge: std::sync::Arc = std::sync::Arc::clone(&state.pty_bridge); + // Captured for the level-2 tap (moved into the pump thread). + let service = std::sync::Arc::clone(&state.session_limit_service); + let detect_clock = std::sync::Arc::new(infrastructure::SystemClock::new()); + let conversation_id = request.conversation_id.clone(); std::thread::spawn(move || { for chunk in stream { + // §21.10-4 tap (best-effort par fragment) : avant de consommer le + // fragment, on cherche le motif de limite. Une détection arme la + // reprise via le service (détecter→planifier). Dormant si pas de + // parser. La fragmentation PTV (motif coupé entre deux fragments) + // est un raté best-effort connu pour LS7. + if let Some(parser) = &rate_limit_parser { + let text = String::from_utf8_lossy(&chunk); + if let Some(limit) = + parser.detect(&text, domain::ports::Clock::now_millis(&*detect_clock)) + { + service.on_rate_limited( + agent_id, + host_node_id, + conversation_id.clone(), + limit.resets_at_ms, + ); + } + } if !bridge.send_output(&session_id, chunk) { break; } @@ -1279,8 +1332,23 @@ 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). + let service = std::sync::Arc::clone(&state.session_limit_service); + let meta = state.structured_sessions.meta_for_session(&sid); std::thread::spawn(move || { for event in stream { + // Non-terminal, sans contenu chat : un signal de limite alimente le service + // (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); + } + } // Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire // chunk; skip them while still draining so the turn runs to its `Final`. let Some(chunk) = crate::chat::chunk_from_event(event) else { @@ -1298,6 +1366,26 @@ pub async fn agent_send( Ok(()) } +/// `cancel_resume` — annule la **reprise automatique** armée pour un agent limité +/// (ARCHITECTURE §21.1-4, fenêtre annulable). +/// +/// Délègue à [`SessionLimitService::cancel_resume`](application::SessionLimitService::cancel_resume) : +/// désarme le réveil et, **seulement si** l'annulation a réussi (le réveil n'avait pas +/// encore tiré), publie `AgentResumeCancelled`. Renvoie `true` ssi une reprise a +/// effectivement été annulée (`false` si aucune n'était armée, ou si elle venait de +/// tirer — auquel cas la reprise suit son cours). +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed agent id). +#[tauri::command] +pub async fn cancel_resume( + agent_id: String, + state: State<'_, AppState>, +) -> Result { + let id = parse_agent_id(&agent_id)?; + Ok(state.session_limit_service.cancel_resume(id)) +} + /// `interrupt_agent` — the **Interrompre** path (cadrage C4 §4.2). /// /// Routes to [`OrchestratorService::interrupt_agent`], which `preempt`s the agent's diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index 59fd150..939d79d 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -165,6 +165,7 @@ pub fn run() { commands::launch_agent, commands::change_agent_profile, commands::agent_send, + commands::cancel_resume, commands::interrupt_agent, commands::delegation_delivered, commands::set_front_attached, diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 93ac7a2..fdf9831 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -12,9 +12,11 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use application::{ - AssignSkillToAgent, ChangeAgentProfile, CheckEmbedderSuggestion, CloseProject, CloseTab, + AgentResumer, AppError, AssignSkillToAgent, ChangeAgentProfile, CheckEmbedderSuggestion, + CloseProject, CloseTab, CloseTerminal, ConfigureProfiles, CreateAgentFromScratch, CreateAgentFromTemplate, CreateLayout, CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent, + LaunchAgentInput, SessionLimitService, DeleteEmbedderProfile, DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate, DescribeEmbedderEngines, DetectAgentDrift, DetectProfiles, DismissEmbedderSuggestion, FirstRunState, GetMemory, GetProjectPermissions, GitBranches, GitCheckout, GitCommit, GitGraph, @@ -35,7 +37,7 @@ use domain::ports::{ AgentContextStore, AgentRuntime, AgentSessionFactory, Clock, Embedder, EmbedderEnvInspector, EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator, MemoryRecall, MemoryStore, PermissionStore, ProcessSpawner, ProfileStore, ProjectStore, - PtyPort, SkillStore, TemplateStore, + PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, }; use domain::profile::{ AgentProfile, ContextInjection, McpConfigStrategy, McpTransport, StructuredAdapter, @@ -54,7 +56,7 @@ use infrastructure::{ HeuristicHandoffSummarizer, IdeaiContextStore, InMemoryConversationRegistry, InMemoryMailbox, LocalFileSystem, LocalProcessSpawner, McpServer, MediatedInbox, NaiveMemoryRecall, OrchestratorWatchHandle, PortablePtyAdapter, StructuredSessionFactory, SystemClock, - SystemMillisClock, TokioBroadcastEventBus, UuidGenerator, VectorMemoryRecall, + SystemMillisClock, TokioBroadcastEventBus, TokioScheduler, UuidGenerator, VectorMemoryRecall, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR, RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED, }; @@ -127,6 +129,108 @@ impl application::ProviderSessionProvider for AppProviderSessionProvider { } } +/// Contexte minimal de **relance** d'un agent (LS7, ARCHITECTURE §21.5). +/// +/// [`AgentResumer::resume`] et [`ScheduledTask::ResumeAgent`] ne portent **pas** le +/// `Project` ni la taille de la cellule, alors que [`LaunchAgentInput`] les exige. +/// La commande `launch_agent` (seul endroit où ces faits sont en main) alimente ce +/// contexte par `agent_id` ; [`AppAgentResumer`] le relit à l'échéance pour +/// recomposer un lancement complet. +#[derive(Clone)] +pub struct ResumeContext { + /// Le projet hôte de l'agent (pour recomposer `LaunchAgentInput`). + pub project: Project, + /// Hauteur de la cellule au dernier lancement (lignes PTY). + pub rows: u16, + /// Largeur de la cellule au dernier lancement (colonnes PTY). + pub cols: u16, +} + +/// Registre partagé `agent_id → ResumeContext` (composition root ↔ commande +/// `launch_agent`). Le **même** `Arc` est injecté dans [`AppAgentResumer`] et conservé +/// sur [`AppState`] pour que la commande l'alimente à chaque lancement. +pub type ResumeContexts = Arc>>; + +/// Implémente le port applicatif [`AgentResumer`] (LS7) **par-dessus** le mécanisme de +/// lancement existant ([`LaunchAgent`]). +/// +/// À l'échéance d'une limite de session, [`SessionLimitService::execute_resume`] +/// délègue ici : on relit le [`ResumeContext`] alimenté par `launch_agent`, on +/// recompose un [`LaunchAgentInput`] (avec le `conversation_id` de la cellule ⇒ +/// `LaunchAgent` applique [`domain::ports::SessionPlan::Resume`]), puis on transmet le +/// `resume_prompt` comme **premier tour** via le portail d'entrée unique +/// ([`InputMediator`](domain::input::InputMediator)) — jamais un write PTY brut +/// (ARCHITECTURE §20). Sans contexte connu (resume « à l'aveugle »), on échoue +/// proprement (l'erreur empêche `AgentResumed` d'être publié), jamais de panique. +struct AppAgentResumer { + /// Le **même** `Arc` que la commande `launch_agent` (relance/réattache). + launch_agent: Arc, + /// Portail d'entrée unique : injecte le prompt de reprise comme premier tour. + input_mediator: Arc, + /// Contexte de relance alimenté par la commande `launch_agent`. + contexts: ResumeContexts, +} + +#[async_trait::async_trait] +impl AgentResumer for AppAgentResumer { + async fn resume( + &self, + agent_id: AgentId, + node_id: domain::NodeId, + conversation_id: Option, + resume_prompt: &str, + ) -> Result<(), AppError> { + // Repli propre (jamais de panique) : sans contexte de relance connu, on ne + // reprend pas à l'aveugle. L'erreur remonte ⇒ `AgentResumed` n'est pas publié. + let ctx = self + .contexts + .lock() + .ok() + .and_then(|m| m.get(&agent_id).cloned()); + let Some(ctx) = ctx else { + return Err(AppError::NotFound(format!( + "resume context for agent {agent_id}" + ))); + }; + + // Recompose la déclaration MCP réelle (même recette que la commande + // `launch_agent`) pour que l'agent repris retrouve ses outils `idea_*`. + let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { + exe, + endpoint: crate::mcp_endpoint::mcp_endpoint(&ctx.project.id) + .as_cli_arg() + .to_owned(), + project_id: ctx.project.id.as_uuid().simple().to_string(), + requester: agent_id.to_string(), + }); + + self.launch_agent + .execute(LaunchAgentInput { + project: ctx.project, + agent_id, + rows: ctx.rows, + cols: ctx.cols, + node_id: Some(node_id), + conversation_id, + mcp_runtime, + }) + .await?; + + // Premier tour de reprise par le **portail d'entrée** (pas de write brut, §20) : + // l'enqueue publie `DelegationReady`, la cellule l'écrit quand le prompt est prêt. + // On ne corrèle aucune réponse (reprise, pas une délégation) ⇒ on lâche le + // `PendingReply`. + let ticket = domain::mailbox::Ticket::new( + domain::mailbox::TicketId::new_random(), + "IdeA", + resume_prompt, + ); + let _ = self.input_mediator.enqueue(agent_id, ticket); + + Ok(()) + } +} + /// Everything the IPC layer needs at runtime, managed by Tauri. /// /// Use cases are stored behind `Arc` so handlers clone cheaply. The concrete @@ -339,6 +443,15 @@ pub struct AppState { /// stopped on close, exactly like the watcher, via the same `Mutex`-guarded /// per-project registry. pub mcp_servers: Mutex>, + /// Service de gestion des **limites de session** des agents (ARCHITECTURE §21) : + /// détecte → planifie → reprend, et annule. Alimenté par les taps niveau 1 + /// (structuré, `agent_send`) et niveau 2 (PTY, `launch_agent`) ; sa reprise auto est + /// annulable via la commande `cancel_resume`. + pub session_limit_service: Arc, + /// Registre `agent_id → ResumeContext` (LS7) partagé avec [`AppAgentResumer`] : + /// la commande `launch_agent` y dépose le `Project`/taille du dernier lancement pour + /// que la reprise auto puisse recomposer un `LaunchAgentInput` complet. + pub resume_contexts: ResumeContexts, } impl AppState { @@ -910,6 +1023,45 @@ impl AppState { }); } let input_mediator = Arc::clone(&mediated_inbox) as Arc; + + // --- Limites de session des agents (ARCHITECTURE §21, LS7) --- + // Service pur-ports « détecter → planifier → reprendre » câblé sur l'existant : + // l'horloge système, le bus partagé, un `TokioScheduler` (minuterie one-shot + // annulable) dont les tâches échues sont drainées plus bas, et un `AppAgentResumer` + // qui relance via le *même* `LaunchAgent` que la commande `launch_agent`. + let resume_contexts: ResumeContexts = Arc::new(Mutex::new(HashMap::new())); + let (resume_tx, mut resume_rx) = tokio::sync::mpsc::unbounded_channel::(); + let scheduler = Arc::new(TokioScheduler::new( + resume_tx, + Arc::clone(&clock) as Arc, + )) as Arc; + let resumer = Arc::new(AppAgentResumer { + launch_agent: Arc::clone(&launch_agent), + input_mediator: Arc::clone(&input_mediator), + contexts: Arc::clone(&resume_contexts), + }) as Arc; + let session_limit_service = Arc::new(SessionLimitService::new( + Arc::clone(&clock) as Arc, + scheduler, + Arc::clone(&events_port), + resumer, + )); + // Drain du scheduler (§21.5-b) : à chaque réveil tiré, `TokioScheduler` pousse une + // `ScheduledTask::ResumeAgent` ; on l'exécute via le service (relance + AgentResumed). + // Patron `sweep_stalled` : `tauri::async_runtime::spawn` (pas `tokio::spawn` — `build` + // tourne dans le hook `setup` sans runtime ambiant). Détaché ⇒ vit autant que l'app. + { + let service = Arc::clone(&session_limit_service); + tauri::async_runtime::spawn(async move { + while let Some(task) = resume_rx.recv().await { + if let Err(e) = service.execute_resume(task).await { + // Best-effort : une relance qui échoue ne fige pas le drain. + eprintln!("[session-limit] reprise auto échouée : {e}"); + } + } + }); + } + // Registre des conversations par paire (cadrage C3) : un fil par paire, session // vivante keyée par conversation (lève l'ambiguïté session/agent). let conversation_registry = Arc::new(InMemoryConversationRegistry::new()) @@ -1051,6 +1203,8 @@ impl AppState { orchestrator_service, orchestrator_watchers: Mutex::new(HashMap::new()), mcp_servers: Mutex::new(HashMap::new()), + session_limit_service, + resume_contexts, move_tab, } } diff --git a/crates/app-tauri/tests/dto_agents.rs b/crates/app-tauri/tests/dto_agents.rs index 74bdb5d..fa48c98 100644 --- a/crates/app-tauri/tests/dto_agents.rs +++ b/crates/app-tauri/tests/dto_agents.rs @@ -309,6 +309,7 @@ fn launch_agent_output_maps_to_terminal_session_dto() { assigned_conversation_id: None, engine_session_id: None, structured: None, + profile: None, }; let dto = TerminalSessionDto::from(out); @@ -354,6 +355,7 @@ fn launch_agent_output_propagates_assigned_conversation_id() { assigned_conversation_id: Some("conv-xyz".to_owned()), engine_session_id: None, structured: None, + profile: None, }; let dto = TerminalSessionDto::from(out); diff --git a/crates/app-tauri/tests/dto_chat.rs b/crates/app-tauri/tests/dto_chat.rs index 4297333..b9fb1e0 100644 --- a/crates/app-tauri/tests/dto_chat.rs +++ b/crates/app-tauri/tests/dto_chat.rs @@ -157,6 +157,7 @@ fn launch_output_with_structured_descriptor_derives_chat_cell_kind() { assigned_conversation_id: None, engine_session_id: None, structured: Some(descriptor), + profile: None, }; let dto = TerminalSessionDto::from(out); assert_eq!(dto.cell_kind, CellKind::Chat); @@ -173,6 +174,7 @@ fn launch_output_without_structured_descriptor_derives_pty_cell_kind() { assigned_conversation_id: None, engine_session_id: None, structured: None, + profile: None, }; let dto = TerminalSessionDto::from(out); assert_eq!(dto.cell_kind, CellKind::Pty, "no descriptor ⇒ pty"); @@ -207,6 +209,7 @@ fn pty_launch_output_serialises_cellkind_pty_without_breaking_existing_keys() { assigned_conversation_id: None, engine_session_id: None, structured: None, + profile: None, }; let v = serde_json::to_value(TerminalSessionDto::from(out)).unwrap(); assert_eq!(v["sessionId"], sid.to_string()); diff --git a/crates/app-tauri/tests/session_limit_wiring.rs b/crates/app-tauri/tests/session_limit_wiring.rs new file mode 100644 index 0000000..a4ad52d --- /dev/null +++ b/crates/app-tauri/tests/session_limit_wiring.rs @@ -0,0 +1,98 @@ +//! Integration test for the **session-limit** wiring in the composition root +//! (ARCHITECTURE §21, LS7). +//! +//! Proves that [`AppState`] actually *constructs and wires* the +//! `SessionLimitService` (the gap LS7 closes: the LS1–LS4 domain/application +//! machinery existed but was never instantiated at runtime). The detect→plan→ +//! resume→cancel logic itself is covered by the application/infrastructure unit +//! tests (LS3/LS4); here we assert the **composition** the commands rely on: +//! the service is present, `on_rate_limited` arms a cancellable resume that emits +//! the domain events on the real bus, and `cancel_resume` is a clean no-op for an +//! unknown agent (never a panic). + +use std::path::PathBuf; +use std::time::Duration; + +use app_tauri_lib::state::AppState; +use domain::events::DomainEvent; +use domain::ids::NodeId; +use domain::AgentId; +use infrastructure::UuidGenerator; +use domain::ports::IdGenerator; + +fn temp_path(tag: &str) -> PathBuf { + let ids = UuidGenerator::new(); + std::env::temp_dir().join(format!("idea-sl-test-{tag}-{}", ids.new_uuid())) +} + +fn agent(n: u128) -> AgentId { + AgentId::from_uuid(uuid::Uuid::from_u128(n)) +} + +fn node(n: u128) -> NodeId { + NodeId::from_uuid(uuid::Uuid::from_u128(n)) +} + +/// `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] +async fn cancel_resume_is_a_clean_noop_for_unknown_agent() { + let state = AppState::build(temp_path("appdata")); + assert!( + !state.session_limit_service.cancel_resume(agent(1)), + "no armed resume ⇒ cancel returns false" + ); +} + +/// A structured rate-limit signal with a **future** reset arms a cancellable +/// resume: the wired service publishes `AgentRateLimited` then +/// `AgentResumeScheduled` on the **real** event bus, and `cancel_resume` then +/// publishes `AgentResumeCancelled` and returns `true`. +#[tokio::test] +async fn on_rate_limited_arms_a_cancellable_resume_over_the_real_bus() { + let state = AppState::build(temp_path("appdata")); + let mut rx = state.event_bus.raw_receiver(); + + // Reset far in the future so the scheduler does NOT fire before we cancel. + let resets_at_ms = i64::MAX; + state + .session_limit_service + .on_rate_limited(agent(7), node(70), None, Some(resets_at_ms)); + + // The two arming events, in order, on the bus. + 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 relayed on the bus"); + assert!(saw_scheduled, "AgentResumeScheduled relayed on the bus"); + + // The window is cancellable: cancel succeeds and emits AgentResumeCancelled. + assert!( + state.session_limit_service.cancel_resume(agent(7)), + "an armed resume is cancellable" + ); + let mut saw_cancelled = false; + for _ in 0..32 { + match tokio::time::timeout(Duration::from_secs(2), rx.recv()).await { + Ok(Ok(DomainEvent::AgentResumeCancelled { agent_id })) if agent_id == agent(7) => { + saw_cancelled = true; + break; + } + Ok(Ok(_)) => continue, + _ => break, + } + } + assert!(saw_cancelled, "AgentResumeCancelled relayed on the bus"); +} diff --git a/crates/application/src/agent/lifecycle.rs b/crates/application/src/agent/lifecycle.rs index 4e8bf29..f20c539 100644 --- a/crates/application/src/agent/lifecycle.rs +++ b/crates/application/src/agent/lifecycle.rs @@ -936,6 +936,15 @@ pub struct LaunchAgentOutput { /// avec les use cases A/B et le câblage qui ne lisent que `session` / /// `assigned_conversation_id`. pub structured: Option, + /// Le **profil résolu** de l'agent lors d'un (re)lancement *effectif* (LS7, + /// ARCHITECTURE §21.10-4). `LaunchAgent::execute` le résout déjà en interne (zéro + /// I/O supplémentaire) ; on l'expose pour que le câblage app-tauri décide d'armer — + /// ou non — le détecteur de limite **niveau 2** (PTY) via la règle de sélection + /// `infrastructure::ratelimit::applies` (anti-double-détection N1/N2), sans + /// reconstruire la frontière infra côté domaine. `None` sur une **réattache** + /// (rebind/idempotent) où aucun profil n'est re-résolu — le câblage n'arme alors + /// rien (best-effort, le tour est sans nouvelle session). + pub profile: Option, } /// Registry mapping each [`ProjectorKey`] to its concrete @@ -1331,6 +1340,8 @@ impl LaunchAgent { assigned_conversation_id: None, engine_session_id: None, structured: None, + // Réattache (rebind de vue) : aucun profil re-résolu. + profile: None, }); } } @@ -1350,6 +1361,8 @@ impl LaunchAgent { assigned_conversation_id: None, engine_session_id: None, structured: None, + // Idempotent (pas de respawn) : aucun profil re-résolu. + profile: None, }); } } @@ -1388,6 +1401,8 @@ impl LaunchAgent { node_id, conversation_id: existing.conversation_id(), }), + // Réattache structurée (rebind de vue) : aucun profil re-résolu. + profile: None, }); } } @@ -1605,6 +1620,9 @@ impl LaunchAgent { // Chemin PTY/terminal brut : pas de session moteur structurée à cacher. engine_session_id: None, structured: None, + // Lancement effectif : profil résolu exposé pour la sélection du détecteur + // de limite niveau 2 (§21.10-4) côté câblage. + profile: Some(profile.clone()), }) } @@ -1690,6 +1708,9 @@ impl LaunchAgent { node_id, conversation_id: engine_session_id, }), + // Lancement structuré effectif : profil résolu exposé (le tap niveau 2 PTY ne + // l'arme pas — `applies` est faux pour un profil structuré, §21.10-4). + profile: Some(profile.clone()), }) } diff --git a/crates/application/src/terminal/registry.rs b/crates/application/src/terminal/registry.rs index b86e632..6c14c6e 100644 --- a/crates/application/src/terminal/registry.rs +++ b/crates/application/src/terminal/registry.rs @@ -380,6 +380,22 @@ impl StructuredSessions { }) } + /// Résout les coordonnées `(agent_id, node_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, + /// celui-ci fait un lookup direct par id. Le pump structuré (`agent_send`) n'a en + /// main que le `SessionId` du tour ; ce lookup lui rend l'agent et la cellule à + /// 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))) + } + /// Liste chaque agent IA vivant, sa cellule hôte et son id de session. /// /// Jumeau de [`TerminalSessions::live_agents`] : un tuple diff --git a/crates/application/tests/structured_registry_d1.rs b/crates/application/tests/structured_registry_d1.rs index 0f74913..a35a3c1 100644 --- a/crates/application/tests/structured_registry_d1.rs +++ b/crates/application/tests/structured_registry_d1.rs @@ -106,6 +106,24 @@ fn structured_insert_resolve_and_remove() { assert!(reg.session_for_agent(&a).is_none()); } +#[test] +fn structured_meta_for_session_resolves_agent_and_node() { + // 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); + + assert_eq!(reg.meta_for_session(&s), Some((a, n))); + + // Id inconnu (ou retiré) ⇒ None (jamais de panique sur une session morte). + assert_eq!(reg.meta_for_session(&sid(999)), None); + reg.remove(&s); + assert_eq!(reg.meta_for_session(&s), None); +} + #[test] fn structured_live_agents_lists_triples() { let reg = StructuredSessions::new();