//! [`TerminalSessions`] — the active-terminal registry (application service). //! //! Maps a [`SessionId`] to the live [`PtyHandle`] and the [`TerminalSession`] //! snapshot. Thread-safe (behind a [`Mutex`]); a single instance is shared //! (as `Arc`) by all terminal use cases via the composition root. See the module //! docs in `terminal/mod.rs` for the rationale of keeping this in the //! application layer rather than the domain or the adapter. use std::collections::HashMap; use std::sync::{Arc, Mutex}; use domain::conversation::ConversationId; use domain::ports::{AgentSession, PtyHandle}; use domain::{AgentId, IssueRef, NodeId, SessionId, SessionKind, TerminalSession}; /// Runtime family of a live agent session. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum LiveSessionKind { /// Raw PTY-backed CLI session. Pty, /// Structured agent-session backend. Structured, } /// Read-only coordinates of one live agent session. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct LiveSessionSnapshot { /// The agent owning the live session. pub agent_id: AgentId, /// The layout node currently hosting the session view. pub node_id: NodeId, /// The live session id. pub session_id: SessionId, /// Which runtime registry owns the session. pub kind: LiveSessionKind, } /// A registered, live terminal: its PTY handle plus the domain snapshot. #[derive(Debug, Clone)] struct Entry { handle: PtyHandle, session: TerminalSession, } /// Read-only liveness query over the agents that currently own a live PTY. /// /// Abstracted as a trait so use cases that only need to ask "is this agent /// running right now?" (e.g. the close-time snapshot of running agents) depend /// on the *capability*, not on the whole [`TerminalSessions`] registry — and can /// be tested against a trivial fake. [`TerminalSessions`] is the production /// implementation. pub trait LiveAgentRegistry: Send + Sync { /// Whether `agent_id` currently has a live session in the registry. fn is_agent_live(&self, agent_id: &AgentId) -> bool; /// Whether `node_id` (a layout leaf) currently hosts a live session. /// /// This is the per-cell liveness the close-time snapshot uses: with the /// "one live session per agent" invariant in place, the same agent can be /// pinned on several leaves but be *live* in at most one — so liveness must /// be keyed on the hosting node, not the agent (otherwise a duplicate leaf /// would be wrongly marked as still running). fn is_node_live(&self, node_id: &NodeId) -> bool; } /// In-memory registry of active terminal sessions. #[derive(Default)] pub struct TerminalSessions { entries: Mutex>, /// Conversation → live session binding (cadrage C3 §5.2): «1 session vivante / /// **conversation**» (remplace «1 / agent»). Populated by the orchestrator when an /// ask resolves/launches the session for a given thread. Separate from `entries` /// (the domain [`TerminalSession`] does not carry a conversation id). conversations: Mutex>, } impl LiveAgentRegistry for TerminalSessions { fn is_agent_live(&self, agent_id: &AgentId) -> bool { self.session_for_agent(agent_id).is_some() } fn is_node_live(&self, node_id: &NodeId) -> bool { self.entries .lock() .map(|m| m.values().any(|e| e.session.node_id == *node_id)) .unwrap_or(false) } } impl TerminalSessions { /// Creates an empty registry. #[must_use] pub fn new() -> Self { Self { entries: Mutex::new(HashMap::new()), conversations: Mutex::new(HashMap::new()), } } /// Binds `conversation` to the live `session` (cadrage C3 §5.2). Idempotent: /// re-binding the same conversation overwrites the target session. pub fn bind_conversation(&self, conversation: ConversationId, session: SessionId) { if let Ok(mut m) = self.conversations.lock() { m.insert(conversation, session); } } /// Returns the live [`SessionId`] bound to `conversation`, if any **and** still /// registered (a stale binding to a closed session resolves to `None`). /// /// «1 session vivante / conversation» — deterministic, replacing the ambiguous /// per-agent lookup for the orchestrator's ask path. #[must_use] pub fn session_for(&self, conversation: ConversationId) -> Option { let sid = self .conversations .lock() .ok()? .get(&conversation) .copied()?; // Only return it if the session is still live in `entries`. self.entries .lock() .ok() .filter(|m| m.contains_key(&sid)) .map(|_| sid) } /// Lists every live [`SessionId`] hosting `agent_id` (cadrage C3 §1.2 — an agent /// may take part in several threads, so this is the **plural** of /// [`Self::session_for_agent`]). #[must_use] pub fn sessions_for_agent(&self, agent_id: &AgentId) -> Vec { self.entries .lock() .map(|m| { m.values() .filter(|e| matches!(e.session.kind, SessionKind::Agent { agent_id: a } if &a == agent_id)) .map(|e| e.session.id) .collect() }) .unwrap_or_default() } /// Inserts a freshly-opened session. pub fn insert(&self, handle: PtyHandle, session: TerminalSession) { if let Ok(mut map) = self.entries.lock() { map.insert(session.id, Entry { handle, session }); } } /// Returns the [`PtyHandle`] for a session, if registered. #[must_use] pub fn handle(&self, id: &SessionId) -> Option { self.entries .lock() .ok() .and_then(|m| m.get(id).map(|e| e.handle.clone())) } /// Returns the [`TerminalSession`] snapshot for a session, if registered. #[must_use] pub fn session(&self, id: &SessionId) -> Option { self.entries .lock() .ok() .and_then(|m| m.get(id).map(|e| e.session.clone())) } /// Returns the [`SessionId`] of the live session hosting a given agent, if any. /// /// An agent runs in a session tagged [`SessionKind::Agent`]; this is the /// mapping the orchestrator's `stop_agent` uses to translate an agent id into /// the [`SessionId`] that `CloseTerminal` expects. Returns `None` when the /// agent has no live session (already stopped / never launched). /// /// **Unambiguous by construction**: the "one live session per agent" /// invariant (enforced in [`crate::agent::LaunchAgent`]) guarantees at most /// one live session per agent, so the first match is *the* match. `find` /// short-circuits on it. #[must_use] pub fn session_for_agent(&self, agent_id: &AgentId) -> Option { self.entries.lock().ok().and_then(|m| { m.values() .find(|e| matches!(e.session.kind, SessionKind::Agent { agent_id: a } if &a == agent_id)) .map(|e| e.session.id) }) } /// Returns the [`NodeId`] of the live cell hosting a given agent, if any. /// /// Companion to [`Self::session_for_agent`]: the launch guard needs the /// *host node* of an already-live agent to report it in /// [`crate::error::AppError::AgentAlreadyRunning`]. Unambiguous by the same /// one-live-session-per-agent invariant. #[must_use] pub fn node_for_agent(&self, agent_id: &AgentId) -> Option { self.entries.lock().ok().and_then(|m| { m.values() .find(|e| matches!(e.session.kind, SessionKind::Agent { agent_id: a } if &a == agent_id)) .map(|e| e.session.node_id) }) } /// Lists every currently-live agent, its current host cell and session id. /// /// One `(AgentId, NodeId, SessionId)` tuple per session tagged [`SessionKind::Agent`]. /// Used by the `list_live_agents` query so the UI can disable an agent that /// is already running elsewhere (it cannot be launched in a second cell). #[must_use] pub fn live_agents(&self) -> Vec<(AgentId, NodeId, SessionId)> { self.entries .lock() .map(|m| { m.values() .filter_map(|e| match e.session.kind { SessionKind::Agent { agent_id } => { Some((agent_id, e.session.node_id, e.session.id)) } SessionKind::Plain => None, }) .collect() }) .unwrap_or_default() } /// Rebinds a live agent session to a new visible layout node without /// respawning the CLI process. /// /// This is the application-level "cell is a view" operation: closing a cell /// can leave an agent running in the background, and later opening that agent /// in another cell updates only the view binding (`node_id`). The PTY handle, /// session id, scrollback and process stay untouched. #[must_use] pub fn rebind_agent_node( &self, agent_id: &AgentId, node_id: NodeId, ) -> Option { self.entries.lock().ok().and_then(|mut m| { let entry = m.values_mut().find( |e| matches!(e.session.kind, SessionKind::Agent { agent_id: a } if &a == agent_id), )?; entry.session.node_id = node_id; Some(entry.session.clone()) }) } /// Returns the [`PtyHandle`]s of every currently-registered session. /// /// Used at application shutdown to kill all live PTYs cleanly (the /// `CloseRequested` hook), independently of the frontend's per-view lifecycle. #[must_use] pub fn handles(&self) -> Vec { self.entries .lock() .map(|m| m.values().map(|e| e.handle.clone()).collect()) .unwrap_or_default() } /// Removes a session from the registry, returning its handle if present. Also /// drops any conversation binding pointing at it (no stale `session_for`). pub fn remove(&self, id: &SessionId) -> Option { if let Ok(mut c) = self.conversations.lock() { c.retain(|_, sid| sid != id); } self.entries .lock() .ok() .and_then(|mut m| m.remove(id).map(|e| e.handle)) } /// Number of currently-registered sessions. #[must_use] pub fn len(&self) -> usize { self.entries.lock().map(|m| m.len()).unwrap_or(0) } /// Whether the registry is empty. #[must_use] pub fn is_empty(&self) -> bool { self.len() == 0 } } // --------------------------------------------------------------------------- // StructuredSessions — le jumeau de TerminalSessions pour les sessions IA (§17.5) // --------------------------------------------------------------------------- /// Une session structurée enregistrée : la session vivante plus les coordonnées /// (agent + cellule hôte) que [`AgentSession`] ne porte pas lui-même. /// /// `AgentSession` n'expose que `id()`/`conversation_id()` ; comme le snapshot /// [`TerminalSession`] côté PTY, on associe ici l'`agent_id` (clé de liveness) et /// le `node_id` (cellule-vue, rebindable) pour offrir la **même** surface que /// [`TerminalSessions`]. struct StructuredEntry { /// La session vivante (ressource process/SDK), derrière le port domaine. session: Arc, /// L'agent IA pilotant cette session (invariant « 1 session vivante/agent »). agent_id: AgentId, /// La cellule (feuille de layout) qui héberge actuellement la vue. node_id: NodeId, } #[derive(Clone)] struct TicketAssistantEntry { session: Arc, requester: String, } /// Registre en mémoire des sessions IA structurées vivantes (ARCHITECTURE §17.5). /// /// **Jumeau de [`TerminalSessions`]** : même rôle (état d'exécution applicatif, pas /// du modèle métier — cf. les docs de [`TerminalSessions`]), même surface côté /// liveness/agent (`session_for_agent`, `node_for_agent`, `live_agents`, /// `rebind_agent_node`, `insert`/`remove`/`session`). La seule différence : il /// stocke des `Arc` (sessions programmatiques) au lieu de /// [`PtyHandle`]/[`TerminalSession`]. /// /// Respecte l'invariant produit **« 1 session vivante par agent »** : la garde /// d'unicité (généralisée sur les deux registres via [`LiveSessions`]) interroge /// `session_for_agent` avant tout lancement. #[derive(Default)] pub struct StructuredSessions { entries: Mutex>, ticket_assistants: Mutex>, } impl LiveAgentRegistry for StructuredSessions { fn is_agent_live(&self, agent_id: &AgentId) -> bool { self.session_for_agent(agent_id).is_some() } fn is_node_live(&self, node_id: &NodeId) -> bool { self.entries .lock() .map(|m| m.values().any(|e| e.node_id == *node_id)) .unwrap_or(false) } } impl StructuredSessions { /// Crée un registre vide. #[must_use] pub fn new() -> Self { Self { entries: Mutex::new(HashMap::new()), ticket_assistants: Mutex::new(HashMap::new()), } } /// Enregistre une session fraîchement démarrée pour `agent_id`, hébergée par /// la cellule `node_id`. Clé par l'id de session ([`AgentSession::id`]). pub fn insert(&self, session: Arc, agent_id: AgentId, node_id: NodeId) { if let Ok(mut map) = self.entries.lock() { let id = session.id(); map.insert( id, StructuredEntry { session, agent_id, node_id, }, ); } } /// Retourne la session enregistrée pour un id, si présente. #[must_use] pub fn session(&self, id: &SessionId) -> Option> { self.entries .lock() .ok() .and_then(|m| m.get(id).map(|e| Arc::clone(&e.session))) .or_else(|| { self.ticket_assistants.lock().ok().and_then(|m| { m.values() .find(|e| e.session.id() == *id) .map(|e| Arc::clone(&e.session)) }) }) } /// Enregistre une session assistant ticket éphémère, keyée par `issue_ref`. /// /// Retourne l'éventuelle session remplacée pour que l'appelant puisse la fermer /// hors verrou. Ces sessions ne sont pas des agents manifest et ne participent /// donc pas aux vues `live_agents`. pub fn insert_ticket_assistant( &self, issue_ref: IssueRef, requester: String, session: Arc, ) -> Option> { self.ticket_assistants .lock() .ok() .and_then(|mut m| m.insert(issue_ref, TicketAssistantEntry { session, requester })) .map(|entry| entry.session) } /// Retourne la session assistant vivante pour `issue_ref`, si présente. #[must_use] pub fn ticket_assistant_session(&self, issue_ref: IssueRef) -> Option> { self.ticket_assistants .lock() .ok() .and_then(|m| m.get(&issue_ref).map(|e| Arc::clone(&e.session))) } /// Retourne l'identité MCP requester de l'assistant ticket, si présent. #[must_use] pub fn ticket_assistant_requester(&self, issue_ref: IssueRef) -> Option { self.ticket_assistants .lock() .ok() .and_then(|m| m.get(&issue_ref).map(|e| e.requester.clone())) } /// Retire l'assistant ticket pour `issue_ref`. pub fn remove_ticket_assistant( &self, issue_ref: IssueRef, ) -> Option<(Arc, String)> { self.ticket_assistants .lock() .ok() .and_then(|mut m| m.remove(&issue_ref).map(|e| (e.session, e.requester))) } /// Retourne la session vivante hébergeant `agent_id`, si elle existe. /// /// Jumeau de [`TerminalSessions::session_for_agent`] : **non ambigu** par /// l'invariant « 1 session vivante/agent » — le premier match est *le* match. #[must_use] pub fn session_for_agent(&self, agent_id: &AgentId) -> Option> { self.entries.lock().ok().and_then(|m| { m.values() .find(|e| &e.agent_id == agent_id) .map(|e| Arc::clone(&e.session)) }) } /// Retourne l'[`SessionId`] de la session vivante hébergeant `agent_id`, si any. #[must_use] pub fn session_id_for_agent(&self, agent_id: &AgentId) -> Option { self.entries.lock().ok().and_then(|m| { m.values() .find(|e| &e.agent_id == agent_id) .map(|e| e.session.id()) }) } /// Retourne le [`NodeId`] de la cellule vivante hébergeant `agent_id`, si any. /// /// Jumeau de [`TerminalSessions::node_for_agent`]. #[must_use] pub fn node_for_agent(&self, agent_id: &AgentId) -> Option { self.entries.lock().ok().and_then(|m| { m.values() .find(|e| &e.agent_id == agent_id) .map(|e| e.node_id) }) } /// 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 /// `(AgentId, NodeId, SessionId)` par session structurée vivante. #[must_use] pub fn live_agents(&self) -> Vec<(AgentId, NodeId, SessionId)> { self.entries .lock() .map(|m| { m.values() .map(|e| (e.agent_id, e.node_id, e.session.id())) .collect() }) .unwrap_or_default() } /// Rebinde la session vivante d'un agent vers une nouvelle cellule-vue sans /// redémarrer la conversation (« la cellule est une vue », §17.6). /// /// Jumeau de [`TerminalSessions::rebind_agent_node`] : seul le `node_id` /// change ; la session, son id et sa conversation restent intacts. Retourne la /// session rebindée, ou `None` si l'agent n'a pas de session vivante. #[must_use] pub fn rebind_agent_node( &self, agent_id: &AgentId, node_id: NodeId, ) -> Option> { self.entries.lock().ok().and_then(|mut m| { let entry = m.values_mut().find(|e| &e.agent_id == agent_id)?; entry.node_id = node_id; Some(Arc::clone(&entry.session)) }) } /// Retire une session du registre, retournant la session si présente (pour que /// l'appelant la `shutdown` hors du verrou). pub fn remove(&self, id: &SessionId) -> Option> { self.entries .lock() .ok() .and_then(|mut m| m.remove(id).map(|e| e.session)) .or_else(|| { self.ticket_assistants.lock().ok().and_then(|mut m| { let issue_ref = m .iter() .find(|(_, e)| e.session.id() == *id) .map(|(issue_ref, _)| *issue_ref)?; m.remove(&issue_ref).map(|e| e.session) }) }) } /// Retourne toutes les sessions vivantes (pour un arrêt global propre au /// shutdown applicatif, jumeau de [`TerminalSessions::handles`]). #[must_use] pub fn sessions(&self) -> Vec> { self.entries .lock() .map(|m| { m.values() .map(|e| Arc::clone(&e.session)) .collect::>() }) .unwrap_or_default() .into_iter() .chain( self.ticket_assistants .lock() .map(|m| { m.values() .map(|e| Arc::clone(&e.session)) .collect::>() }) .unwrap_or_default(), ) .collect() } /// Nombre de sessions structurées vivantes. #[must_use] pub fn len(&self) -> usize { self.entries.lock().map(|m| m.len()).unwrap_or(0) + self.ticket_assistants.lock().map(|m| m.len()).unwrap_or(0) } /// Si le registre est vide. #[must_use] pub fn is_empty(&self) -> bool { self.len() == 0 } } // --------------------------------------------------------------------------- // LiveSessions — agrégateur des deux registres derrière LiveAgentRegistry (§17.5) // --------------------------------------------------------------------------- /// Agrégateur de liveness sur **les deux** registres (PTY + structuré). /// /// Un agent terminal brut vit dans [`TerminalSessions`] ; un agent IA structuré vit /// dans [`StructuredSessions`]. La garde d'unicité et l'orchestrateur (§17.4) /// dépendent de cet agrégateur (ISP : ils ne voient que la capacité « liveness + /// résolution »), et une requête de liveness/agent voit donc les deux registres : /// un agent est vivant s'il a une session vivante dans **l'un OU l'autre**. /// /// Implémente [`LiveAgentRegistry`] : le trait existant n'est **pas modifié** (ses /// implémenteurs et appelants actuels — `TerminalSessions`, le snapshot — restent /// inchangés), il est simplement **réalisé par un troisième implémenteur** qui /// agrège, ce qui *généralise* sa portée à l'ensemble PTY+structuré sans régression. pub struct LiveSessions { /// Registre des sessions terminal brut (PTY). pub pty: Arc, /// Registre des sessions IA structurées. pub structured: Arc, } impl LiveSessions { /// Construit l'agrégateur à partir des deux registres partagés. #[must_use] pub fn new(pty: Arc, structured: Arc) -> Self { Self { pty, structured } } /// L'[`SessionId`] de la session vivante d'un agent, PTY **ou** structurée. #[must_use] pub fn session_id_for_agent(&self, agent_id: &AgentId) -> Option { self.pty .session_for_agent(agent_id) .or_else(|| self.structured.session_id_for_agent(agent_id)) } /// La cellule hôte de la session vivante d'un agent, PTY **ou** structurée. #[must_use] pub fn node_for_agent(&self, agent_id: &AgentId) -> Option { self.pty .node_for_agent(agent_id) .or_else(|| self.structured.node_for_agent(agent_id)) } /// Tous les agents vivants des deux registres (PTY puis structurés). #[must_use] pub fn live_agents(&self) -> Vec<(AgentId, NodeId, SessionId)> { let mut all = self.pty.live_agents(); all.extend(self.structured.live_agents()); all } /// Tous les agents vivants avec le type de registre source (PTY puis structuré). #[must_use] pub fn live_agent_snapshots(&self) -> Vec { let mut all: Vec = self .pty .live_agents() .into_iter() .map(|(agent_id, node_id, session_id)| LiveSessionSnapshot { agent_id, node_id, session_id, kind: LiveSessionKind::Pty, }) .collect(); all.extend(self.structured.live_agents().into_iter().map( |(agent_id, node_id, session_id)| LiveSessionSnapshot { agent_id, node_id, session_id, kind: LiveSessionKind::Structured, }, )); all } } impl LiveAgentRegistry for LiveSessions { fn is_agent_live(&self, agent_id: &AgentId) -> bool { self.pty.is_agent_live(agent_id) || self.structured.is_agent_live(agent_id) } fn is_node_live(&self, node_id: &NodeId) -> bool { self.pty.is_node_live(node_id) || self.structured.is_node_live(node_id) } }