blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge(runtime): stop/restart d'agent visible robuste — kill réel, état stopped, relance froide (#196, QA verte)

Browse Source
This commit is contained in:
Blomios
2026-08-12 14:56:15 +02:00
parent 4e93fa2f20 bad7b4c2f3
commit f5e2e302ba
23 changed files with 440 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/app-tauri/src/commands.rs
View File

@ -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)?;

23
crates/application/src/agent/lifecycle.rs
View File

@ -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,

8
crates/application/src/lib.rs
View File

@ -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,

3
crates/application/src/orchestrator/service.rs
View File

@ -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?;

4
crates/application/src/terminal/mod.rs
View File

@ -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,

121
crates/application/src/terminal/registry.rs
View File

@ -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<LiveSessionSnapshot> 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<TerminalSessions>,
/// Registre des sessions IA structurées.
pub structured: Arc<StructuredSessions>,
runtime: Mutex<HashMap<LiveAgentKey, LiveRuntimeStatus>>,
}
impl LiveSessions {
/// Construit l'agrégateur à partir des deux registres partagés.
#[must_use]
pub fn new(pty: Arc<TerminalSessions>, structured: Arc<StructuredSessions>) -> 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<LiveSessionSnapshot> {
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<LiveRuntimeStatus> {
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 {

55
crates/application/src/terminal/usecases.rs
View File

@ -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<CloseTerminalOutput, AppError> {
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<Vec<StoppedAgentRuntime>, 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,
});
}

26
crates/application/src/workstate/actions.rs
View File

@ -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}")))
}

9
crates/application/src/workstate/mod.rs
View File

@ -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<LiveWorkSession>,
/// Last known runtime lifecycle state for the agent, distinct from `live`.
pub runtime_state: Option<LiveRuntimeState>,
/// 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,

28
crates/application/tests/agent_lifecycle.rs
View File

@ -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.

23
crates/application/tests/structured_launch_d3.rs
View File

@ -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).

35
crates/application/tests/workstate.rs
View File

@ -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<dyn InputMediator>,
Arc::clone(&f.queue) as Arc<dyn AgentQueueSnapshot>,
);
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");

35
crates/application/tests/workstate_actions.rs
View File

@ -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();

46
crates/backend/src/dto.rs
View File

@ -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<CellKind>,
/// 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<LaunchAgentOutput> 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<LiveSessionKind> 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<LiveRuntimeState> 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<LiveWorkSessionDto>,
/// Last known runtime lifecycle state, distinct from live-session presence.
pub runtime_state: Option<LiveRuntimeStateDto>,
/// Current mediated-input busy state.
pub busy: AgentBusyState,
/// Pending/in-progress delegation tickets, in FIFO order.
@ -3789,7 +3826,9 @@ impl From<ProjectWorkState> 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<StopLiveAgentOutput> for StopLiveAgentResponseDto {
@ -3984,6 +4025,7 @@ impl From<StopLiveAgentOutput> 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(),
}
}
}

2
crates/backend/src/lib.rs
View File

@ -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?;

1
crates/web-server/src/lib.rs
View File

@ -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)?;

2
frontend/src/adapters/agent.ts
View File

@ -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,

1
frontend/src/adapters/http/streamGateways.ts
View File

@ -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);

2
frontend/src/adapters/http/wsLiveClient.ts
View File

@ -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<TerminalAttachResult> {
@ -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 },
);

21
frontend/src/adapters/mock/index.ts
View File

@ -401,6 +401,8 @@ export class MockAgentGateway implements AgentGateway {
private liveSessionByAgent = new Map<string, string>();
/** Runtime kind per live agent (`agentId → kind`). */
private liveKindByAgent = new Map<string, "pty" | "structured">();
/** Last stopped runtime per agent for idempotent stop. */
private stoppedByAgent = new Map<string, StoppedLiveAgent>();
/** Retained structured reply chunks per live chat session. */
private chatScrollback = new Map<string, ReplyChunk[]>();
/** 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<Agent> {
@ -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}` }
: {}),
};

21
frontend/src/adapters/workStateNormalization.ts
View File

@ -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" ||

8
frontend/src/domain/index.ts
View File

@ -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;

11
frontend/src/ports/index.ts
View File

@ -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}.
*