From bad7b4c2f388825d9c3d2203bfc79451a81d889d Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 12 Aug 2026 14:56:02 +0200 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20stop/restart=20d'agent=20visibl?= =?UTF-8?q?e=20robuste=20=E2=80=94=20kill=20r=C3=A9el,=20=C3=A9tat=20stopp?= =?UTF-8?q?ed,=20relance=20froide=20(#196,=20QA=20verte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Solidifie le cycle stop/restart d'un agent visible en cellule: - terminal/registry: kill effectif de la TUI native et de la CLI custom, arrêt idempotent du live runtime (PTY + structuré), distinction nette « agent stoppé » vs « pas de session ». - workstate + lifecycle: projection d'un état runtime explicite « stopped » (AgentRuntimeState: running/stopping/stopped/failedStart), réconcilié avec le live state; traitement des courses « agent not found ». - relance froide via freshStart: repart de zéro sans réattacher une session morte et recharge la configuration courante de l'agent (y compris changement de modèle). - contrat projeté côté adapters frontend (ports/domain/normalization/mock) pour la consommation UI future (#197/#198), sans surface UI ici. Tests QA verts: workstate_actions, orchestrator_service stop_agent, agent_lifecycle (reopen/fresh_start), workstate stopped runtime state, vitest adapter stop_live_agent. --- crates/app-tauri/src/commands.rs | 1 + crates/application/src/agent/lifecycle.rs | 23 +++- crates/application/src/lib.rs | 8 +- .../application/src/orchestrator/service.rs | 3 + crates/application/src/terminal/mod.rs | 4 +- crates/application/src/terminal/registry.rs | 121 +++++++++++++++++- crates/application/src/terminal/usecases.rs | 55 ++++---- crates/application/src/workstate/actions.rs | 26 +++- crates/application/src/workstate/mod.rs | 9 +- crates/application/tests/agent_lifecycle.rs | 28 ++++ .../application/tests/structured_launch_d3.rs | 23 ++++ crates/application/tests/workstate.rs | 35 ++++- crates/application/tests/workstate_actions.rs | 35 ++++- crates/backend/src/dto.rs | 46 ++++++- crates/backend/src/lib.rs | 2 + crates/web-server/src/lib.rs | 1 + frontend/src/adapters/agent.ts | 2 + frontend/src/adapters/http/streamGateways.ts | 1 + frontend/src/adapters/http/wsLiveClient.ts | 2 + frontend/src/adapters/mock/index.ts | 21 ++- .../src/adapters/workStateNormalization.ts | 21 +++ frontend/src/domain/index.ts | 8 ++ frontend/src/ports/index.ts | 11 ++ 23 files changed, 440 insertions(+), 46 deletions(-) diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 0299b63f..0a20663b 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -2074,6 +2074,7 @@ pub async fn launch_agent( mcp_runtime, allow_structured_alongside_pty: false, require_structured: request.cell_kind == Some(CellKind::Chat), + fresh_start: request.fresh_start, }) .await .map_err(ErrorDto::from)?; diff --git a/crates/application/src/agent/lifecycle.rs b/crates/application/src/agent/lifecycle.rs index 70cabd52..75b85400 100644 --- a/crates/application/src/agent/lifecycle.rs +++ b/crates/application/src/agent/lifecycle.rs @@ -963,6 +963,7 @@ impl ChangeAgentProfile { mcp_runtime: None, allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, }) .await?; Ok(Some(output.session)) @@ -1143,6 +1144,9 @@ pub struct LaunchAgentInput { /// /// Historical callers leave this `false`, preserving the PTY launch contract. pub require_structured: bool, + /// Cold restart request: ignore the cell's previous conversation id and create + /// a new provider/CLI conversation for this activation. + pub fresh_start: bool, } /// OS/runtime facts injected by the composition root (`app-tauri`) to materialise @@ -1863,6 +1867,11 @@ impl LaunchAgent { self.structured_routing_mode, StructuredRoutingMode::RequireStructured ); + let launch_conversation_id = if input.fresh_start { + None + } else { + input.conversation_id.clone() + }; // 1b. Enforce the "one live session per agent" invariant (decision: an // agent is a singleton that runs in a single cell at a time). This @@ -1908,7 +1917,7 @@ impl LaunchAgent { self.pty.kill(&handle).await?; } } else { - match reattach_decision(input.node_id, host_node, input.conversation_id.as_deref()) + match reattach_decision(input.node_id, host_node, launch_conversation_id.as_deref()) { ReattachDecision::Rebind { node_id } => { if let Some(session) = self.sessions.rebind_agent_node_in_project( @@ -1971,7 +1980,7 @@ impl LaunchAgent { reattach_decision( input.node_id, host_node, - input.conversation_id.as_deref(), + launch_conversation_id.as_deref(), ) }; let node_id = match decision { @@ -2061,7 +2070,11 @@ impl LaunchAgent { // the cell's current `conversation_id`. Any id this launch *assigns* is // returned in the output so the caller persists it on the leaf. let (session_plan, assigned_conversation_id) = self - .resolve_session_plan(&profile, input.conversation_id.clone(), &input.project.root) + .resolve_session_plan( + &profile, + launch_conversation_id.clone(), + &input.project.root, + ) .await; let mut spec = self.runtime @@ -2076,7 +2089,7 @@ impl LaunchAgent { &manifest, &agent, content.clone(), - input.conversation_id.as_deref(), + launch_conversation_id.as_deref(), ) .await?; // Best-effort contextual embedder suggestion (LOT C3, §14.5.5): the agent @@ -2186,7 +2199,7 @@ impl LaunchAgent { // pure partagée avec `resolve_conversation` (aucun couplage à // l'orchestrateur). C'est cet id — et **non** l'id de session moteur — // qui retrouve log + handoff au (re)lancement (P7) et survit au swap. - let pair_conversation_id = input.conversation_id.clone().unwrap_or_else(|| { + let pair_conversation_id = launch_conversation_id.clone().unwrap_or_else(|| { ConversationId::for_project_pair( input.project.id, ConversationParty::User, diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index a45f61ae..e783c567 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -228,10 +228,10 @@ pub use template::{ SyncAgentWithTemplateOutput, UpdateTemplate, UpdateTemplateInput, UpdateTemplateOutput, }; pub use terminal::{ - CloseTerminal, CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, LiveSessionKind, - LiveSessionSnapshot, LiveSessions, OpenTerminal, OpenTerminalInput, OpenTerminalOutput, - ResizeTerminal, ResizeTerminalInput, StructuredSessions, TerminalSessions, WriteToTerminal, - WriteToTerminalInput, + CloseTerminal, CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, LiveRuntimeState, + LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot, LiveSessions, OpenTerminal, + OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, StructuredSessions, + TerminalSessions, WriteToTerminal, WriteToTerminalInput, }; pub use ticket_assistant::{ CloseTicketAssistant, CloseTicketAssistantInput, OpenTicketAssistant, OpenTicketAssistantInput, diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 6059079d..5e79ff53 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -1718,6 +1718,7 @@ impl OrchestratorService { mcp_runtime: None, allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, }) .await?; @@ -2496,6 +2497,7 @@ impl OrchestratorService { .and_then(|p| p.runtime_for(project, agent_id)), allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, }) .await?; @@ -2577,6 +2579,7 @@ impl OrchestratorService { .and_then(|p| p.runtime_for(project, agent_id)), allow_structured_alongside_pty: true, require_structured: true, + fresh_start: false, }) .await?; diff --git a/crates/application/src/terminal/mod.rs b/crates/application/src/terminal/mod.rs index 02e9f220..7dfa5e7e 100644 --- a/crates/application/src/terminal/mod.rs +++ b/crates/application/src/terminal/mod.rs @@ -29,8 +29,8 @@ mod registry; mod usecases; pub use registry::{ - LiveAgentRegistry, LiveSessionKind, LiveSessionSnapshot, LiveSessions, StructuredSessions, - TerminalSessions, + LiveAgentRegistry, LiveRuntimeState, LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot, + LiveSessions, StructuredSessions, TerminalSessions, }; pub use usecases::{ stop_live_agent_runtime_in_project, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, diff --git a/crates/application/src/terminal/registry.rs b/crates/application/src/terminal/registry.rs index 9d278fb0..be8fbe8d 100644 --- a/crates/application/src/terminal/registry.rs +++ b/crates/application/src/terminal/registry.rs @@ -24,6 +24,19 @@ pub enum LiveSessionKind { Structured, } +/// Explicit lifecycle state of an agent runtime in a visible cell. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LiveRuntimeState { + /// The runtime has a live session in one of the registries. + Running, + /// A stop request is currently tearing the runtime down. + Stopping, + /// The runtime was torn down and no live session remains. + Stopped, + /// The last launch attempt failed before a live session was registered. + FailedStart, +} + /// Read-only coordinates of one live agent session. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct LiveSessionSnapshot { @@ -39,6 +52,51 @@ pub struct LiveSessionSnapshot { pub kind: LiveSessionKind, } +/// Last-known lifecycle projection for one agent runtime. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LiveRuntimeStatus { + /// The project owning the runtime. + pub project_id: ProjectId, + /// The agent owning the runtime. + pub agent_id: AgentId, + /// Last known layout node hosting the runtime view. + pub node_id: NodeId, + /// Last known session id. + pub session_id: SessionId, + /// Runtime family that owns or owned the session. + pub kind: LiveSessionKind, + /// Current lifecycle state. + pub state: LiveRuntimeState, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +struct LiveAgentKey { + project_id: ProjectId, + agent_id: AgentId, +} + +impl LiveAgentKey { + fn new(project_id: ProjectId, agent_id: AgentId) -> Self { + Self { + project_id, + agent_id, + } + } +} + +impl From for LiveRuntimeStatus { + fn from(snapshot: LiveSessionSnapshot) -> Self { + Self { + project_id: snapshot.project_id, + agent_id: snapshot.agent_id, + node_id: snapshot.node_id, + session_id: snapshot.session_id, + kind: snapshot.kind, + state: LiveRuntimeState::Running, + } + } +} + /// A registered, live terminal: its PTY handle plus the domain snapshot. #[derive(Debug)] struct Entry { @@ -849,13 +907,18 @@ pub struct LiveSessions { pub pty: Arc, /// Registre des sessions IA structurées. pub structured: Arc, + runtime: Mutex>, } 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 } + Self { + pty, + structured, + runtime: Mutex::new(HashMap::new()), + } } /// L'[`SessionId`] de la session vivante d'un agent, PTY **ou** structurée. @@ -899,6 +962,62 @@ impl LiveSessions { all.extend(self.structured.live_agent_snapshots()); all } + + /// All live runtime snapshots for one `(project, agent)` canonical key. + #[must_use] + pub fn live_agent_snapshots_for( + &self, + project_id: ProjectId, + agent_id: &AgentId, + ) -> Vec { + self.live_agent_snapshots() + .into_iter() + .filter(|snapshot| snapshot.project_id == project_id && &snapshot.agent_id == agent_id) + .collect() + } + + /// Current runtime projection for one canonical `(project, agent)` key. + #[must_use] + pub fn runtime_status_for( + &self, + project_id: ProjectId, + agent_id: &AgentId, + ) -> Option { + self.live_agent_snapshots_for(project_id, agent_id) + .into_iter() + .next() + .map(LiveRuntimeStatus::from) + .or_else(|| { + self.runtime + .lock() + .ok() + .and_then(|m| m.get(&LiveAgentKey::new(project_id, *agent_id)).copied()) + }) + } + + /// Records a non-running runtime state for the last known session coordinates. + pub fn mark_runtime_state(&self, snapshot: LiveSessionSnapshot, state: LiveRuntimeState) { + if let Ok(mut map) = self.runtime.lock() { + map.insert( + LiveAgentKey::new(snapshot.project_id, snapshot.agent_id), + LiveRuntimeStatus { + project_id: snapshot.project_id, + agent_id: snapshot.agent_id, + node_id: snapshot.node_id, + session_id: snapshot.session_id, + kind: snapshot.kind, + state, + }, + ); + } + } + + /// Drops stale stopped/failed state when a caller needs a clean runtime boundary. + pub fn clear_runtime_state(&self, project_id: ProjectId, agent_id: &AgentId) { + if let Ok(mut map) = self.runtime.lock() { + map.remove(&LiveAgentKey::new(project_id, *agent_id)); + } + } } impl LiveAgentRegistry for LiveSessions { diff --git a/crates/application/src/terminal/usecases.rs b/crates/application/src/terminal/usecases.rs index 357735e6..a66246ab 100644 --- a/crates/application/src/terminal/usecases.rs +++ b/crates/application/src/terminal/usecases.rs @@ -10,7 +10,7 @@ use domain::{ use crate::error::AppError; -use super::registry::{LiveSessionKind, LiveSessions, TerminalSessions}; +use super::registry::{LiveRuntimeState, LiveSessionKind, LiveSessions, TerminalSessions}; /// Input for [`OpenTerminal::execute`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -223,9 +223,10 @@ impl CloseTerminal { ) -> Result { let handle = self .sessions - .remove(&input.session_id) + .handle(&input.session_id) .ok_or_else(|| AppError::NotFound(format!("terminal session {}", input.session_id)))?; let status = self.pty.kill(&handle).await?; + self.sessions.remove(&input.session_id); Ok(CloseTerminalOutput { code: status.code }) } } @@ -237,8 +238,12 @@ pub struct StoppedAgentRuntime { pub agent_id: AgentId, /// Live session id removed from its registry. pub session_id: SessionId, + /// Last known layout node hosting the runtime view. + pub node_id: NodeId, /// Runtime registry that owned the session. pub kind: LiveSessionKind, + /// Lifecycle state reached by this stop operation. + pub state: LiveRuntimeState, } /// Stops one agent's live runtime across both registries. @@ -253,30 +258,34 @@ pub async fn stop_live_agent_runtime_in_project( close: &CloseTerminal, ) -> Result, AppError> { let mut stopped = Vec::new(); + let snapshots = live.live_agent_snapshots_for(project.id, &agent_id); - if let Some(session_id) = live - .structured - .session_id_for_agent_in_project(project.id, &agent_id) - { - if let Some(session) = live.structured.remove(&session_id) { - session - .shutdown() - .await - .map_err(|e| AppError::Process(e.to_string()))?; + for snapshot in snapshots { + live.mark_runtime_state(snapshot, LiveRuntimeState::Stopping); + match snapshot.kind { + LiveSessionKind::Structured => { + if let Some(session) = live.structured.remove(&snapshot.session_id) { + session + .shutdown() + .await + .map_err(|e| AppError::Process(e.to_string()))?; + } + } + LiveSessionKind::Pty => { + close + .execute(CloseTerminalInput { + session_id: snapshot.session_id, + }) + .await?; + } } + live.mark_runtime_state(snapshot, LiveRuntimeState::Stopped); stopped.push(StoppedAgentRuntime { - agent_id, - session_id, - kind: LiveSessionKind::Structured, - }); - } - - if let Some(session_id) = live.pty.session_for_agent_in_project(project.id, &agent_id) { - close.execute(CloseTerminalInput { session_id }).await?; - stopped.push(StoppedAgentRuntime { - agent_id, - session_id, - kind: LiveSessionKind::Pty, + agent_id: snapshot.agent_id, + session_id: snapshot.session_id, + node_id: snapshot.node_id, + kind: snapshot.kind, + state: LiveRuntimeState::Stopped, }); } diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index dcffcf7e..7b130086 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -16,7 +16,8 @@ use domain::{AgentId, NodeId, Project, RuntimeAgentKey, SessionId}; use crate::error::AppError; use crate::orchestrator::OrchestratorService; use crate::terminal::{ - stop_live_agent_runtime_in_project, CloseTerminal, LiveSessionKind, LiveSessions, + stop_live_agent_runtime_in_project, CloseTerminal, LiveRuntimeState, LiveSessionKind, + LiveSessions, }; /// Input for [`AttachLiveAgent::execute`]. @@ -115,8 +116,12 @@ pub struct StopLiveAgentOutput { pub agent_id: AgentId, /// The id of the session that was torn down. pub session_id: SessionId, + /// Last known layout node hosting the runtime view. + pub node_id: NodeId, /// Runtime family that owned the session. pub kind: LiveSessionKind, + /// Runtime lifecycle projection after the stop request. + pub runtime_state: LiveRuntimeState, } /// Best-effort stopper for execution resources owned by one live agent turn. @@ -232,7 +237,26 @@ impl StopLiveAgent { .map(|stopped| StopLiveAgentOutput { agent_id: stopped.agent_id, session_id: stopped.session_id, + node_id: stopped.node_id, kind: stopped.kind, + runtime_state: stopped.state, + }) + .or_else(|| { + self.live + .runtime_status_for(project.id, &agent_id) + .filter(|status| { + matches!( + status.state, + LiveRuntimeState::Stopping | LiveRuntimeState::Stopped + ) + }) + .map(|status| StopLiveAgentOutput { + agent_id: status.agent_id, + session_id: status.session_id, + node_id: status.node_id, + kind: status.kind, + runtime_state: status.state, + }) }) .ok_or_else(|| AppError::NotFound(format!("running session for agent {agent_id}"))) } diff --git a/crates/application/src/workstate/mod.rs b/crates/application/src/workstate/mod.rs index bc1a1d90..a1ef5b37 100644 --- a/crates/application/src/workstate/mod.rs +++ b/crates/application/src/workstate/mod.rs @@ -32,7 +32,7 @@ use domain::{ }; use crate::error::AppError; -use crate::terminal::{LiveSessionKind, LiveSessionSnapshot, LiveSessions}; +use crate::terminal::{LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, LiveSessions}; use crate::HandoffProvider; /// Maximum length (in characters) of a derived [`AgentTicketState::task_preview`]. @@ -202,6 +202,8 @@ pub struct AgentWorkState { pub profile_id: ProfileId, /// Current live session, if the agent is live. pub live: Option, + /// Last known runtime lifecycle state for the agent, distinct from `live`. + pub runtime_state: Option, /// Current FIFO/busy state. pub busy: AgentBusyState, /// Pending/in-progress delegation tickets, in FIFO order. @@ -490,6 +492,10 @@ impl GetProjectWorkState { session_id: snapshot.session_id, kind: snapshot.kind, }); + let runtime_state = self + .live + .runtime_status_for(input.project.id, &agent.id) + .map(|status| status.state); // Tickets are crossed with the busy state: only an agent absent from // the manifest is dropped (this loop only visits manifest entries), so // the manifest boundary is naturally preserved. @@ -510,6 +516,7 @@ impl GetProjectWorkState { name: agent.name, profile_id: agent.profile_id, live, + runtime_state, busy, tickets, background_tasks, diff --git a/crates/application/tests/agent_lifecycle.rs b/crates/application/tests/agent_lifecycle.rs index d7097c68..fee85f09 100644 --- a/crates/application/tests/agent_lifecycle.rs +++ b/crates/application/tests/agent_lifecycle.rs @@ -1193,6 +1193,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput { mcp_runtime: None, allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, } } @@ -2946,6 +2947,33 @@ async fn launch_reopen_with_existing_conversation_id_resumes_without_new_id() { ); } +#[tokio::test] +async fn launch_fresh_start_ignores_existing_conversation_id_and_mints_new_one() { + let (launch, agent, _fs, _pty, _bus, _sessions, _tr, session) = launch_fixture_with_profile( + profile_with_session(pid(9), Some("--session-id"), "--resume"), + Some(ContextInjectionPlan::Stdin), + ); + + let mut input = launch_input(agent.id); + input.conversation_id = Some("conv-dead-session".to_owned()); + input.fresh_start = true; + let out = launch.execute(input).await.expect("fresh restart launch"); + + let expected = Uuid::from_u128(1).to_string(); + assert_eq!( + out.assigned_conversation_id.as_deref(), + Some(expected.as_str()), + "fresh restart must mint a new conversation id" + ); + assert_eq!( + *session.lock().unwrap(), + Some(SessionPlan::Assign { + conversation_id: expected + }), + "fresh restart must not reuse the dead conversation id" + ); +} + #[tokio::test] async fn launch_profile_without_session_block_passes_none() { // Non-regression: a profile with no session block behaves exactly as before. diff --git a/crates/application/tests/structured_launch_d3.rs b/crates/application/tests/structured_launch_d3.rs index ac27ea5a..ec239df5 100644 --- a/crates/application/tests/structured_launch_d3.rs +++ b/crates/application/tests/structured_launch_d3.rs @@ -772,6 +772,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput { mcp_runtime: None, allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, } } @@ -1462,6 +1463,28 @@ async fn structured_profile_with_cell_conversation_resolves_to_resume() { ); } +/// Une relance froide ignore explicitement le `conversation_id` encore porté par +/// la cellule : le runtime structuré doit recevoir un plan neuf, pas une reprise. +#[tokio::test] +async fn structured_profile_fresh_start_ignores_cell_conversation() { + let factory = FakeFactory::new(500, Some("engine-fresh")); + let f = launch_fixture(structured_profile(pid(9)), factory); + + let mut input = launch_input(f.agent.id); + input.conversation_id = Some("conv-existing".to_owned()); + input.fresh_start = true; + f.launch.execute(input).await.expect("launch starts fresh"); + + assert_eq!( + *f.session_probe.lock().unwrap(), + Some(SessionPlan::None), + "fresh start must not pass the old cell conversation to the runtime" + ); + let starts = f.factory.starts(); + assert_eq!(starts.len(), 1); + assert_eq!(starts[0].1, SessionPlan::None); +} + /// Une cellule structurée **neuve** (pas de conversation) sur un profil sans bloc /// `session` ⇒ `SessionPlan::None` (rien à reprendre ; l'id moteur sera capté au 1er /// tour). diff --git a/crates/application/tests/workstate.rs b/crates/application/tests/workstate.rs index cb04393c..c6bbc578 100644 --- a/crates/application/tests/workstate.rs +++ b/crates/application/tests/workstate.rs @@ -10,8 +10,8 @@ use async_trait::async_trait; use application::{ AppExitWorkGuardDetail, ConversationLogProvider, ConversationPreviewStatus, GetAppExitWorkGuardState, GetAppExitWorkGuardStateInput, GetProjectWorkState, - GetProjectWorkStateInput, HandoffProvider, LiveSessionKind, LiveSessions, StructuredSessions, - TerminalSessions, TicketWorkSource, TicketWorkStatus, + GetProjectWorkStateInput, HandoffProvider, LiveRuntimeState, LiveSessionKind, LiveSessions, + StructuredSessions, TerminalSessions, TicketWorkSource, TicketWorkStatus, }; use domain::mailbox::{ AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, TurnResolution, @@ -702,6 +702,37 @@ async fn workstate_attaches_live_structured_session_to_manifest_agent() { assert_eq!(live.kind, LiveSessionKind::Structured); } +#[tokio::test] +async fn workstate_exposes_stopped_runtime_state_without_live_session() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + let snapshot = application::LiveSessionSnapshot { + project_id: f.project.id, + agent_id: a.id, + node_id: nid(300), + session_id: sid(3), + kind: LiveSessionKind::Pty, + }; + let live = LiveSessions::new(Arc::clone(&f.pty), Arc::clone(&f.structured)); + live.mark_runtime_state(snapshot, LiveRuntimeState::Stopped); + let usecase = GetProjectWorkState::new( + Arc::new(FakeContexts { + manifest: manifest(std::slice::from_ref(&a)), + }), + Arc::new(live), + Arc::clone(&f.input) as Arc, + Arc::clone(&f.queue) as Arc, + ); + + let out = usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + assert_eq!(out.agents[0].live, None); + assert_eq!(out.agents[0].runtime_state, Some(LiveRuntimeState::Stopped)); +} + #[tokio::test] async fn app_exit_guard_is_false_without_busy_agent_or_active_background_task() { let a = agent(10, "alpha"); diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index fc7d0f46..bd482c79 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -11,8 +11,8 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use application::{ - AttachLiveAgent, AttachLiveAgentInput, CloseTerminal, LiveSessionKind, LiveSessions, - StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput, StructuredSessions, + AttachLiveAgent, AttachLiveAgentInput, CloseTerminal, LiveRuntimeState, LiveSessionKind, + LiveSessions, StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput, StructuredSessions, TerminalSessions, }; use domain::ports::{ @@ -329,7 +329,9 @@ async fn stop_pty_kills_and_removes_session() { assert_eq!(out.agent_id, a); assert_eq!(out.session_id, sid(1)); + assert_eq!(out.node_id, nid(100)); assert_eq!(out.kind, LiveSessionKind::Pty); + assert_eq!(out.runtime_state, LiveRuntimeState::Stopped); // Delegated to the close primitive: process killed and registry emptied. assert_eq!(f.pty_port.kills(), vec![sid(1)]); assert!(f.pty.is_empty(), "live session removed from the registry"); @@ -353,6 +355,7 @@ async fn stop_structured_shuts_down_and_removes_session() { assert_eq!(out.session_id, sid(2)); assert_eq!(out.kind, LiveSessionKind::Structured); + assert_eq!(out.runtime_state, LiveRuntimeState::Stopped); assert!(flag.load(Ordering::SeqCst), "session.shutdown() was called"); assert!(f.structured.is_empty(), "live session removed"); assert_eq!( @@ -388,6 +391,34 @@ async fn stop_structured_runs_execution_cascade_best_effort() { assert!(f.structured.is_empty(), "live session removed"); } +#[tokio::test] +async fn stop_pty_is_idempotent_after_success() { + let f = fixture(); + let a = aid(10); + insert_pty(&f.pty, sid(1), a, nid(100)); + + let first = f + .stop + .execute(StopLiveAgentInput { + project: project(), + agent_id: a, + }) + .await + .unwrap(); + let second = f + .stop + .execute(StopLiveAgentInput { + project: project(), + agent_id: a, + }) + .await + .unwrap(); + + assert_eq!(first, second); + assert_eq!(second.runtime_state, LiveRuntimeState::Stopped); + assert_eq!(f.pty_port.kills(), vec![sid(1)], "kill is not repeated"); +} + #[tokio::test] async fn stop_unknown_agent_is_not_found() { let f = fixture(); diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index f4b3476c..89fd0473 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -15,8 +15,9 @@ use application::{ ConversationPreviewStatus, ConversationTurnWorkPreview, ConversationWorkSummary, CreateProjectInput, CreateProjectOutput, ExecuteSlashCommandOutput, GitGraphOutput, HealthInput, HealthReport, LayoutKind, ListProjectsOutput, ListSlashCommandsOutput, - LiveSessionKind, LiveSessionSnapshot, OpenProjectOutput, ProjectWorkState, SlashCommandEffect, - StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, TurnPage, TurnSource, TurnView, + LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, OpenProjectOutput, ProjectWorkState, + SlashCommandEffect, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, TurnPage, + TurnSource, TurnView, }; use domain::ports::{ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage}; use domain::{ @@ -2962,6 +2963,10 @@ pub struct LaunchAgentRequestDto { /// view sends `"chat"` to require a structured/headless session. #[serde(default)] pub cell_kind: Option, + /// Cold restart request: ignore any cell conversation id and create a fresh + /// provider/CLI conversation after the caller has stopped the previous runtime. + #[serde(default)] + pub fresh_start: bool, } impl From for TerminalSessionDto { @@ -3452,6 +3457,8 @@ pub struct LiveAgentDto { pub session_id: String, /// Runtime family that owns the session (`pty`/`structured`). pub kind: LiveWorkSessionKindDto, + /// Runtime lifecycle state. Live agents are always `running`. + pub runtime_state: LiveRuntimeStateDto, } /// Response DTO for `list_live_agents` (transparent array on the wire). @@ -3479,6 +3486,7 @@ impl LiveAgentListDto { node_id: snapshot.node_id.to_string(), session_id: snapshot.session_id.to_string(), kind: snapshot.kind.into(), + runtime_state: LiveRuntimeStateDto::Running, }) .collect(), ) @@ -3504,6 +3512,31 @@ impl From for LiveWorkSessionKindDto { } } +/// Explicit lifecycle state for a visible agent runtime. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum LiveRuntimeStateDto { + /// A live session exists in one runtime registry. + Running, + /// A stop request is in progress. + Stopping, + /// The last live session was torn down. + Stopped, + /// The last launch attempt failed before registration. + FailedStart, +} + +impl From for LiveRuntimeStateDto { + fn from(state: LiveRuntimeState) -> Self { + match state { + LiveRuntimeState::Running => Self::Running, + LiveRuntimeState::Stopping => Self::Stopping, + LiveRuntimeState::Stopped => Self::Stopped, + LiveRuntimeState::FailedStart => Self::FailedStart, + } + } +} + /// Live session coordinates in the project work-state read model. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -3514,6 +3547,8 @@ pub struct LiveWorkSessionDto { pub session_id: String, /// Runtime family that owns the session. pub kind: LiveWorkSessionKindDto, + /// Runtime lifecycle state. Live sessions are always `running`. + pub runtime_state: LiveRuntimeStateDto, } /// Derived processing status of a queued ticket. @@ -3665,6 +3700,8 @@ pub struct AgentWorkStateDto { pub profile_id: String, /// Live session, if any. pub live: Option, + /// Last known runtime lifecycle state, distinct from live-session presence. + pub runtime_state: Option, /// Current mediated-input busy state. pub busy: AgentBusyState, /// Pending/in-progress delegation tickets, in FIFO order. @@ -3789,7 +3826,9 @@ impl From for ProjectWorkStateDto { node_id: live.node_id.to_string(), session_id: live.session_id.to_string(), kind: live.kind.into(), + runtime_state: LiveRuntimeStateDto::Running, }), + runtime_state: agent.runtime_state.map(Into::into), busy: agent.busy, tickets: agent .tickets @@ -3976,6 +4015,8 @@ pub struct StopLiveAgentResponseDto { pub session_id: String, /// Runtime family that owned the session (`pty`/`structured`). pub kind: LiveWorkSessionKindDto, + /// Runtime lifecycle projection after the stop request. + pub runtime_state: LiveRuntimeStateDto, } impl From for StopLiveAgentResponseDto { @@ -3984,6 +4025,7 @@ impl From for StopLiveAgentResponseDto { agent_id: out.agent_id.to_string(), session_id: out.session_id.to_string(), kind: out.kind.into(), + runtime_state: out.runtime_state.into(), } } } diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 1564e40f..666eca2c 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -730,6 +730,7 @@ impl WakeSessionProvider for AppWakeSessionProvider { mcp_runtime: None, allow_structured_alongside_pty: true, require_structured: false, + fresh_start: false, }) .await .map_err(|err| WakeError::Session(err.to_string()))?; @@ -850,6 +851,7 @@ impl AgentResumer for AppAgentResumer { mcp_runtime, allow_structured_alongside_pty: false, require_structured: false, + fresh_start: false, }) .await?; diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index 2f1b3778..aec765ec 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -1517,6 +1517,7 @@ async fn execute_launch_agent_for_ws( mcp_runtime, allow_structured_alongside_pty: false, require_structured: request.cell_kind == Some(backend::dto::CellKind::Chat), + fresh_start: request.fresh_start, }) .await .map_err(ErrorDto::from)?; diff --git a/frontend/src/adapters/agent.ts b/frontend/src/adapters/agent.ts index 1a80a22f..fa192d6f 100644 --- a/frontend/src/adapters/agent.ts +++ b/frontend/src/adapters/agent.ts @@ -185,6 +185,7 @@ export class TauriAgentGateway implements AgentGateway { // Resume id: the leaf's persisted conversation id, when any (T4b). The // backend resumes it; absent ⇒ a fresh cell that may get a new id. conversationId: options.conversationId ?? null, + freshStart: options.freshStart ?? false, // Hosting cell: lets the backend enforce one-live-session-per-agent. nodeId: options.nodeId ?? null, }, @@ -220,6 +221,7 @@ export class TauriAgentGateway implements AgentGateway { cols: options.cols, cellKind: "chat", conversationId: options.conversationId ?? null, + freshStart: options.freshStart ?? false, nodeId: options.nodeId ?? null, }, onOutput: channel, diff --git a/frontend/src/adapters/http/streamGateways.ts b/frontend/src/adapters/http/streamGateways.ts index e4b3b4c8..bc462473 100644 --- a/frontend/src/adapters/http/streamGateways.ts +++ b/frontend/src/adapters/http/streamGateways.ts @@ -288,6 +288,7 @@ export class HttpAgentGateway implements AgentGateway { rows: options.rows, cols: options.cols, conversationId: options.conversationId ?? null, + freshStart: options.freshStart ?? false, onData, }); return makeWsTerminalHandle(res.sessionId, this.ws, res.assignedConversationId); diff --git a/frontend/src/adapters/http/wsLiveClient.ts b/frontend/src/adapters/http/wsLiveClient.ts index 585b4ab2..72c20f9e 100644 --- a/frontend/src/adapters/http/wsLiveClient.ts +++ b/frontend/src/adapters/http/wsLiveClient.ts @@ -382,6 +382,7 @@ export class WsLiveClient { cols: number; nodeId?: string | null; conversationId?: string | null; + freshStart?: boolean; onData: (bytes: Uint8Array) => void; onStatus?: (status: string, exitCode?: number | null) => void; }): Promise { @@ -394,6 +395,7 @@ export class WsLiveClient { rows: params.rows, cols: params.cols, conversationId: params.conversationId ?? null, + freshStart: params.freshStart ?? false, }, { onData: params.onData, onStatus: params.onStatus, lastSeq: 0 }, ); diff --git a/frontend/src/adapters/mock/index.ts b/frontend/src/adapters/mock/index.ts index b3f55c1f..9522b224 100644 --- a/frontend/src/adapters/mock/index.ts +++ b/frontend/src/adapters/mock/index.ts @@ -401,6 +401,8 @@ export class MockAgentGateway implements AgentGateway { private liveSessionByAgent = new Map(); /** Runtime kind per live agent (`agentId → kind`). */ private liveKindByAgent = new Map(); + /** Last stopped runtime per agent for idempotent stop. */ + private stoppedByAgent = new Map(); /** Retained structured reply chunks per live chat session. */ private chatScrollback = new Map(); /** Persisted custom chat drafts keyed by project/layout/node. */ @@ -460,6 +462,7 @@ export class MockAgentGateway implements AgentGateway { nodeId, sessionId: this.liveSessionByAgent.get(agentId)!, kind: this.liveKindByAgent.get(agentId) ?? "pty", + runtimeState: "running", })); } @@ -509,6 +512,7 @@ export class MockAgentGateway implements AgentGateway { nodeId, sessionId, kind: this.liveKindByAgent.get(agentId) ?? "pty", + runtimeState: "running", }; } @@ -523,6 +527,8 @@ export class MockAgentGateway implements AgentGateway { } const sessionId = this.liveSessionByAgent.get(agentId); if (!sessionId) { + const stopped = this.stoppedByAgent.get(agentId); + if (stopped) return stopped; const err: GatewayError = { code: "NOT_FOUND", message: `agent ${agentId} has no live session`, @@ -537,7 +543,14 @@ export class MockAgentGateway implements AgentGateway { this.liveByAgent.delete(agentId); const kind = this.liveKindByAgent.get(agentId) ?? "pty"; this.liveKindByAgent.delete(agentId); - return { agentId, sessionId, kind }; + const stopped: StoppedLiveAgent = { + agentId, + sessionId, + kind, + runtimeState: "stopped", + }; + this.stoppedByAgent.set(agentId, stopped); + return stopped; } async createAgent(projectId: string, input: CreateAgentInput): Promise { @@ -781,6 +794,7 @@ export class MockAgentGateway implements AgentGateway { this.liveByAgent.set(agentId, options.nodeId); this.liveSessionByAgent.set(agentId, sessionId); this.liveKindByAgent.set(agentId, "pty"); + this.stoppedByAgent.delete(agentId); } const clearLive = () => { if (this.liveByAgent.get(agentId) === options.nodeId) { @@ -807,7 +821,7 @@ export class MockAgentGateway implements AgentGateway { // Simulate session assignment (T4b): a fresh cell (no conversation id) gets a // newly-minted id surfaced on the handle so the caller persists it; a cell // that already carries an id resumes it and nothing new is assigned. - if (!options.conversationId) { + if (options.freshStart || !options.conversationId) { return { ...handle, assignedConversationId: `mock-conversation-${sessionId}`, @@ -848,11 +862,12 @@ export class MockAgentGateway implements AgentGateway { this.liveByAgent.set(agentId, options.nodeId); this.liveSessionByAgent.set(agentId, sessionId); this.liveKindByAgent.set(agentId, "structured"); + this.stoppedByAgent.delete(agentId); } return { sessionId, cellKind: "chat", - ...(!options.conversationId + ...(options.freshStart || !options.conversationId ? { assignedConversationId: `mock-conversation-${sessionId}` } : {}), }; diff --git a/frontend/src/adapters/workStateNormalization.ts b/frontend/src/adapters/workStateNormalization.ts index 3d76e09b..9ae54422 100644 --- a/frontend/src/adapters/workStateNormalization.ts +++ b/frontend/src/adapters/workStateNormalization.ts @@ -150,11 +150,19 @@ function normalizeAgent(value: unknown, index: number): AgentWorkState { const agent = isRecord(value) ? value : {}; const liveKind: "pty" | "structured" = isRecord(agent.live) && agent.live.kind === "structured" ? "structured" : "pty"; + const runtimeState = normalizeRuntimeState(agent.runtimeState ?? agent.runtime_state); const live = isRecord(agent.live) ? { nodeId: stringValue(agent.live.nodeId), sessionId: stringValue(agent.live.sessionId), kind: liveKind, + ...(normalizeRuntimeState(agent.live.runtimeState ?? agent.live.runtime_state) + ? { + runtimeState: normalizeRuntimeState( + agent.live.runtimeState ?? agent.live.runtime_state, + ), + } + : {}), } : undefined; @@ -173,6 +181,7 @@ function normalizeAgent(value: unknown, index: number): AgentWorkState { name: stringValue(agent.name, "Unnamed agent"), profileId: stringValue(agent.profileId, "unknown"), ...(live ? { live } : {}), + ...(runtimeState ? { runtimeState } : {}), busy: normalizeBusy(agent.busy), tickets: arrayValue(agent.tickets).map(normalizeTicket), inboxDepth: numberValue(agent.inboxDepth, inbox.length), @@ -181,6 +190,18 @@ function normalizeAgent(value: unknown, index: number): AgentWorkState { }; } +function normalizeRuntimeState(value: unknown): AgentWorkState["runtimeState"] { + if ( + value === "running" || + value === "stopping" || + value === "stopped" || + value === "failedStart" + ) { + return value; + } + return undefined; +} + function normalizeSummaryStatus(value: unknown): ConversationPreviewStatus { if ( value === "ready" || diff --git a/frontend/src/domain/index.ts b/frontend/src/domain/index.ts index 7c5ab8b3..06536921 100644 --- a/frontend/src/domain/index.ts +++ b/frontend/src/domain/index.ts @@ -572,8 +572,15 @@ export interface LiveWorkSession { nodeId: string; sessionId: string; kind: "pty" | "structured"; + runtimeState?: AgentRuntimeState; } +export type AgentRuntimeState = + | "running" + | "stopping" + | "stopped" + | "failedStart"; + /** Busy/idle status for an agent in the work-state read-model. */ export type WorkBusyState = | { state: "idle" } @@ -663,6 +670,7 @@ export interface AgentWorkState { name: string; profileId: string; live?: LiveWorkSession; + runtimeState?: AgentRuntimeState; busy: WorkBusyState; tickets: AgentTicketState[]; inboxDepth?: number; diff --git a/frontend/src/ports/index.ts b/frontend/src/ports/index.ts index 87f8b10f..1231778a 100644 --- a/frontend/src/ports/index.ts +++ b/frontend/src/ports/index.ts @@ -378,6 +378,8 @@ export interface OpenTerminalOptions { * meaningful for {@link AgentGateway.launchAgent}; ignored for plain terminals. */ conversationId?: string; + /** Ignore a previous conversation id and request a cold, clean launch. */ + freshStart?: boolean; /** * The layout leaf (node) hosting this launch. Drives the "one live session per * agent" invariant backend-side: launching an agent already running in a @@ -400,6 +402,8 @@ export interface LiveAgent { sessionId: string; /** Runtime kind backing the live agent. */ kind: "pty" | "structured"; + /** Runtime lifecycle state. Live agents are always running. */ + runtimeState?: AgentRuntimeState; } /** Result returned when the backend stops a live agent. */ @@ -407,8 +411,15 @@ export interface StoppedLiveAgent { agentId: string; sessionId: string; kind: "pty" | "structured"; + runtimeState?: AgentRuntimeState; } +export type AgentRuntimeState = + | "running" + | "stopping" + | "stopped" + | "failedStart"; + /** * A live terminal handle returned by {@link TerminalGateway.openTerminal}. *