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

merge(chat): stop en cascade pour la CLI custom (#187, QA verte)

Browse Source
This commit is contained in:
Blomios
2026-08-11 18:15:01 +02:00
parent 2413aa7473 22ce2f62fd
commit bc9a034249
9 changed files with 293 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2764,29 +2764,39 @@ pub async fn interrupt_agent(
.map_err(ErrorDto::from)
}
/// `cancel_agent_chat` — interrupt the current turn of a live structured chat
/// session without shutting the session down.
/// `cancel_agent_chat` — stop a live structured chat session and cascade its
/// running execution children best-effort.
///
/// This is the session-id based twin of [`interrupt_agent`] for a custom chat
/// This is the session-id based twin of [`stop_live_agent`] for a custom chat
/// surface that already owns the structured `sessionId`.
///
/// # Errors
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `NOT_FOUND` if no live
/// structured session owns the id, `PROCESS` if a concrete cancel fails).
/// structured session owns the id, `PROCESS` if a concrete stop fails).
#[tauri::command]
pub async fn cancel_agent_chat(
session_id: String,
state: State<'_, AppState>,
) -> Result<(), ErrorDto> {
let sid = parse_session_id(&session_id)?;
let session = state
let (project_id, agent_id, _, _) = state
.structured_sessions
.session(&sid)
.meta_for_session(&sid)
.ok_or_else(|| ErrorDto::from(AppError::NotFound(format!("structured session {sid}"))))?;
session
.cancel_current_turn()
let project = resolve_project(&project_id.to_string(), &state).await?;
let dependencies = state
.orchestrator_service
.active_wait_dependencies(agent_id);
state.stop_turn_watch(project.id, agent_id);
for dependency in dependencies {
state.stop_turn_watch(project.id, dependency);
}
state
.stop_live_agent
.execute(StopLiveAgentInput { project, agent_id })
.await
.map_err(|e| ErrorDto::from(AppError::from(e)))
.map(|_| ())
.map_err(ErrorDto::from)
}
async fn preferred_view_for_agent(

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

@ -249,7 +249,7 @@ pub use workstate::{
ConversationTurnWorkPreview, ConversationWorkSummary, GetAppExitWorkGuardState,
GetAppExitWorkGuardStateInput, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, ReconcileLiveState,
ReconcileLiveStateInput, StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
ReconcileLiveStateInput, StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput,
StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
};

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

@ -24,7 +24,9 @@ use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef,
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
use domain::input::{InputMediator, InputSource, SubmitConfig};
use domain::mailbox::{Ticket, TicketId};
use domain::ports::{BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle};
use domain::ports::{
BackgroundTaskRunner, BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle,
};
use domain::project::ProjectPath;
use domain::{
AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskRendezvousLink,
@ -52,7 +54,9 @@ use crate::orchestrator::{
};
use crate::skill::{CreateSkill, CreateSkillInput, ReadSkill, ReadSkillInput};
use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions};
use crate::workstate::{GetLiveStateLean, UpdateLiveState, UpdateLiveStateInput};
use crate::workstate::{
GetLiveStateLean, StopAgentExecutionCascade, UpdateLiveState, UpdateLiveStateInput,
};
/// Default terminal geometry for an orchestrator-launched agent cell. The UI
/// resizes the PTY to the real cell size on attach; these are sane starting rows
@ -523,6 +527,9 @@ pub struct OrchestratorService {
/// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒
/// comportement historique sans persistance de tâche (call sites non câblés).
background_tasks: Option<Arc<dyn BackgroundTaskStore>>,
/// Runner runtime des tâches command-backed, utilisé ici uniquement pour la cascade
/// d'arrêt best-effort des `idea_run_in_background` attachés au rendez-vous actif.
background_runner: Option<Arc<dyn BackgroundTaskRunner>>,
/// Service applicatif des limites de session, injecté depuis la composition root.
/// `None` conserve les call sites/tests legacy ; les chemins conscients des limites
/// publient alors seulement leur erreur de tour historique.
@ -610,6 +617,7 @@ impl OrchestratorService {
ask_liveness_probe: None,
ask_ceiling: ASK_AGENT_CEILING,
background_tasks: None,
background_runner: None,
session_limits: None,
spawn_background_command: None,
}
@ -646,6 +654,13 @@ impl OrchestratorService {
self
}
/// Branche le runner command-backed pour annuler les tâches enfants lors d'un stop.
#[must_use]
pub fn with_background_task_runner(mut self, runner: Arc<dyn BackgroundTaskRunner>) -> Self {
self.background_runner = Some(runner);
self
}
/// Branche la **sonde de vivacité** (signe de vie) du rendez-vous délégué : la borne
/// de tour devient une **fenêtre d'inactivité** réarmée à chaque progrès observé,
/// sous le plafond [`Self::with_ask_ceiling`]. Builder additif (signature de
@ -3101,6 +3116,40 @@ impl<'a> WaitEdgeGuard<'a> {
}
}
#[async_trait::async_trait]
impl StopAgentExecutionCascade for OrchestratorService {
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) {
let Some(runner) = &self.background_runner else {
return;
};
let agent = runtime_key(project, agent_id);
stop_active_rendezvous_background_tasks(&self.active_rendezvous, agent, runner.as_ref())
.await;
}
}
fn active_rendezvous_background_tasks(
active: &StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
agent: RuntimeAgentKey,
) -> Vec<TaskId> {
active
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(&agent)
.map(|active| active.background_tasks.clone())
.unwrap_or_default()
}
async fn stop_active_rendezvous_background_tasks(
active: &StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
agent: RuntimeAgentKey,
runner: &dyn BackgroundTaskRunner,
) {
for task_id in active_rendezvous_background_tasks(active, agent) {
let _ = runner.cancel(task_id).await;
}
}
impl Drop for WaitEdgeGuard<'_> {
fn drop(&mut self) {
let mut g = self
@ -3375,6 +3424,77 @@ mod tests {
RuntimeAgentKey::new(domain::ProjectId::from_uuid(uuid::Uuid::nil()), aid(n))
}
#[derive(Default)]
struct SpyBackgroundRunner {
cancelled: TestMutex<Vec<TaskId>>,
}
#[async_trait::async_trait]
impl BackgroundTaskRunner for SpyBackgroundRunner {
async fn spawn(
&self,
spec: domain::ports::BackgroundTaskSpec,
) -> Result<domain::ports::BackgroundTaskHandle, domain::ports::BackgroundTaskPortError>
{
Ok(domain::ports::BackgroundTaskHandle {
task_id: spec.task_id,
})
}
async fn cancel(
&self,
task_id: TaskId,
) -> Result<(), domain::ports::BackgroundTaskPortError> {
self.cancelled.lock().unwrap().push(task_id);
Ok(())
}
fn subscribe_completions(&self) -> domain::ports::BackgroundCompletionStream {
Box::new(std::iter::empty())
}
}
#[tokio::test]
async fn stop_active_rendezvous_background_tasks_cancels_attached_tasks_only() {
let active = StdMutex::new(HashMap::new());
let target = rkey(2);
let unrelated = rkey(9);
let first = TaskId::from_uuid(uuid::Uuid::from_u128(101));
let second = TaskId::from_uuid(uuid::Uuid::from_u128(102));
let other = TaskId::from_uuid(uuid::Uuid::from_u128(999));
active.lock().unwrap().insert(
target,
ActiveRendezvous {
link: BackgroundTaskRendezvousLink {
rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(201)),
ticket_id: tid(202),
requester_agent_id: Some(aid(1)),
target_agent_id: target.agent_id,
conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(203)),
},
background_tasks: vec![first, second],
},
);
active.lock().unwrap().insert(
unrelated,
ActiveRendezvous {
link: BackgroundTaskRendezvousLink {
rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(301)),
ticket_id: tid(302),
requester_agent_id: Some(aid(1)),
target_agent_id: unrelated.agent_id,
conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(303)),
},
background_tasks: vec![other],
},
);
let runner = SpyBackgroundRunner::default();
stop_active_rendezvous_background_tasks(&active, target, &runner).await;
assert_eq!(*runner.cancelled.lock().unwrap(), vec![first, second]);
}
/// Drop d'un garde **armé** ⇒ `cancel_head` + `mark_idle` sur la cible (c'est le
/// comportement qui débloque un agent resté `Busy` sur un futur abandonné).
#[test]

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

@ -9,6 +9,7 @@
use std::sync::Arc;
use async_trait::async_trait;
use domain::input::InputMediator;
use domain::{AgentId, NodeId, Project, RuntimeAgentKey, SessionId};
@ -116,6 +117,18 @@ pub struct StopLiveAgentOutput {
pub kind: LiveSessionKind,
}
/// Best-effort stopper for execution resources owned by one live agent turn.
///
/// Implementations must be idempotent: a missing, already-finished or already-stopped
/// child resource is a successful no-op. Failures are deliberately not propagated by
/// [`StopLiveAgent`], because the product contract is "stop the agent; cascade related
/// work best-effort".
#[async_trait]
pub trait StopAgentExecutionCascade: Send + Sync {
/// Requests cancellation of child execution resources for `agent_id` in `project`.
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId);
}
/// Stops an already-running agent's live session, polymorphically.
///
/// Resolves the session by agent (PTY first, then structured). A PTY session is
@ -129,6 +142,7 @@ pub struct StopLiveAgent {
close: Arc<CloseTerminal>,
input: Option<Arc<dyn InputMediator>>,
waits: Option<Arc<OrchestratorService>>,
cascade: Option<Arc<dyn StopAgentExecutionCascade>>,
}
impl StopLiveAgent {
@ -140,6 +154,7 @@ impl StopLiveAgent {
close,
input: None,
waits: None,
cascade: None,
}
}
@ -159,6 +174,13 @@ impl StopLiveAgent {
self
}
/// Wires best-effort cancellation of child resources spawned by the running turn.
#[must_use]
pub fn with_execution_cascade(mut self, cascade: Arc<dyn StopAgentExecutionCascade>) -> Self {
self.cascade = Some(cascade);
self
}
/// Tears down the agent's live session.
///
/// # Errors
@ -177,10 +199,18 @@ impl StopLiveAgent {
}
async fn stop_dependencies(&self, project: &Project, agent_id: AgentId) {
if let Some(cascade) = &self.cascade {
cascade
.stop_agent_execution_children(project, agent_id)
.await;
}
let Some(waits) = &self.waits else {
return;
};
for dep in waits.active_wait_dependencies(agent_id) {
if let Some(cascade) = &self.cascade {
cascade.stop_agent_execution_children(project, dep).await;
}
if let Some(mediator) = &self.input {
mediator.preempt(RuntimeAgentKey::new(project.id, dep));
}

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

@ -10,8 +10,8 @@ mod live;
mod reconcile;
pub use actions::{
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
StopLiveAgentInput, StopLiveAgentOutput,
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopAgentExecutionCascade,
StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
};
pub use live::{
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,

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

@ -12,7 +12,8 @@ use async_trait::async_trait;
use application::{
AttachLiveAgent, AttachLiveAgentInput, CloseTerminal, LiveSessionKind, LiveSessions,
StopLiveAgent, StopLiveAgentInput, StructuredSessions, TerminalSessions,
StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput, StructuredSessions,
TerminalSessions,
};
use domain::ports::{
AgentSession, AgentSessionError, ExitStatus, OutputStream, PtyError, PtyHandle, PtyPort,
@ -120,6 +121,18 @@ impl AgentSession for FakeSession {
}
}
#[derive(Default)]
struct FakeCascade {
calls: Mutex<Vec<(ProjectId, AgentId)>>,
}
#[async_trait]
impl StopAgentExecutionCascade for FakeCascade {
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) {
self.calls.lock().unwrap().push((project.id, agent_id));
}
}
// ---------------------------------------------------------------------------
// wiring helpers
// ---------------------------------------------------------------------------
@ -351,6 +364,30 @@ async fn stop_structured_shuts_down_and_removes_session() {
assert!(f.pty_port.kills().is_empty());
}
#[tokio::test]
async fn stop_structured_runs_execution_cascade_best_effort() {
let f = fixture();
let a = aid(10);
let flag = insert_structured(&f.structured, sid(2), a, nid(100));
let cascade = Arc::new(FakeCascade::default());
let stop = f
.stop
.with_execution_cascade(Arc::clone(&cascade) as Arc<dyn StopAgentExecutionCascade>);
let out = stop
.execute(StopLiveAgentInput {
project: project(),
agent_id: a,
})
.await
.unwrap();
assert_eq!(out.session_id, sid(2));
assert_eq!(*cascade.calls.lock().unwrap(), vec![(project().id, a)]);
assert!(flag.load(Ordering::SeqCst), "session.shutdown() was called");
assert!(f.structured.is_empty(), "live session removed");
}
#[tokio::test]
async fn stop_unknown_agent_is_not_found() {
let f = fixture();

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

@ -2907,6 +2907,7 @@ impl BackendCore {
Arc::clone(&background_tasks_port),
Arc::clone(&clock) as Arc<dyn Clock>,
)
.with_background_task_runner(Arc::clone(&background_runner_port))
// Limites de session sur le chemin délégué headless : quand le drain
// structuré retourne `RateLimited`, l'orchestrateur arme la reprise pour
// la cible limitée via le même service que les chemins directs.
@ -2929,7 +2930,9 @@ impl BackendCore {
.with_cascade(
Arc::clone(&input_mediator),
Arc::clone(&orchestrator_service),
),
)
.with_execution_cascade(Arc::clone(&orchestrator_service)
as Arc<dyn application::StopAgentExecutionCascade>),
);
// --- Windows (L10) ---

56
frontend/src/features/agents/CustomAgentChatView.test.tsx
View File

@ -101,7 +101,12 @@ describe("CustomAgentChatView", () => {
expect(screen.getByText("Fichier: clipboard.png")).toBeTruthy();
});
it("cancels only the current turn and keeps the structured session alive", async () => {
it("stops the live agent with cascade feedback and releases the structured session", async () => {
let resolveStop: (value: {
agentId: string;
sessionId: string;
kind: "structured";
}) => void = () => {};
const agent = {
launchAgentChat: vi.fn(),
reattachAgentChat: vi.fn(async (sessionId: string) => ({
@ -109,6 +114,14 @@ describe("CustomAgentChatView", () => {
scrollback: [],
})),
sendAgentChat: vi.fn(() => new Promise<void>(() => {})),
stopLiveAgent: vi.fn(
() =>
new Promise<{ agentId: string; sessionId: string; kind: "structured" }>(
(resolve) => {
resolveStop = resolve;
},
),
),
cancelAgentChat: vi.fn(async () => {}),
closeAgentChat: vi.fn(async () => {}),
};
@ -159,24 +172,27 @@ describe("CustomAgentChatView", () => {
fireEvent.click(await screen.findByRole("button", { name: "Stop" }));
const stoppingLabel = await screen.findByText("Arrêt en cours...");
expect(stoppingLabel.closest("button")?.hasAttribute("disabled")).toBe(true);
await waitFor(() =>
expect(agent.cancelAgentChat).toHaveBeenCalledWith("chat-session-1"),
expect(agent.stopLiveAgent).toHaveBeenCalledWith("project-1", "agent-1"),
);
expect(agent.closeAgentChat).not.toHaveBeenCalled();
expect(onSessionId).not.toHaveBeenCalledWith(null);
expect(screen.getByText("Tour interrompu.")).toBeTruthy();
fireEvent.change(screen.getByLabelText(/message CLI custom/), {
target: { value: "second turn" },
resolveStop({
agentId: "agent-1",
sessionId: "chat-session-1",
kind: "structured",
});
fireEvent.click(screen.getByRole("button", { name: "Envoyer" }));
await waitFor(() => expect(agent.sendAgentChat).toHaveBeenCalledTimes(2));
expect(agent.sendAgentChat).toHaveBeenLastCalledWith(
"chat-session-1",
"second turn",
expect.any(Function),
);
await waitFor(() => expect(onSessionId).toHaveBeenCalledWith(null));
expect(agent.cancelAgentChat).not.toHaveBeenCalled();
expect(agent.closeAgentChat).not.toHaveBeenCalled();
expect(
screen.getByText(
"Agent arrêté. Cascade d'arrêt demandée pour les tâches et appels MCP liés.",
),
).toBeTruthy();
expect(
screen.getByRole("button", { name: "Envoyer" }).hasAttribute("disabled"),
).toBe(true);
});
it("falls back to a fresh launch when reattach reports NOT_FOUND", async () => {
@ -2569,10 +2585,11 @@ describe("CustomAgentChatView", () => {
scrollback: [],
})),
sendAgentChat: vi.fn(() => new Promise<void>(() => {})),
cancelAgentChat: vi.fn().mockRejectedValueOnce({
stopLiveAgent: vi.fn().mockRejectedValueOnce({
code: "NOT_FOUND",
message: "not found: structured session gone-session",
message: "agent agent-1 has no live session",
}),
cancelAgentChat: vi.fn(async () => {}),
closeAgentChat: vi.fn(async () => {}),
};
const onSessionId = vi.fn();
@ -2614,8 +2631,9 @@ describe("CustomAgentChatView", () => {
fireEvent.click(await screen.findByRole("button", { name: "Stop" }));
expect(await screen.findByText("Session déjà fermée.")).toBeTruthy();
expect(await screen.findByText("Session déjà arrêtée.")).toBeTruthy();
expect(onSessionId).toHaveBeenCalledWith(null);
expect(agent.cancelAgentChat).not.toHaveBeenCalled();
expect(agent.closeAgentChat).not.toHaveBeenCalled();
expect(screen.queryByText(/not found: structured session/)).toBeNull();
});

49
frontend/src/features/agents/CustomAgentChatView.tsx
View File

@ -534,6 +534,7 @@ export function CustomAgentChatView({
useState<ProfileCommandDialogState | null>(null);
const [opening, setOpening] = useState(false);
const [busy, setBusy] = useState(false);
const [stopping, setStopping] = useState(false);
const [error, setError] = useState<string | null>(null);
const scrollRef = useRef<HTMLDivElement | null>(null);
const composerRef = useRef<HTMLTextAreaElement | null>(null);
@ -1209,23 +1210,46 @@ export function CustomAgentChatView({
}
}
async function cancel() {
async function stopAgent() {
const sid = sessionRef.current;
if (!sid || !agent.cancelAgentChat) return;
setBusy(false);
setOpening(false);
if (!sid || !agent.stopLiveAgent || stopping) return;
setStopping(true);
setError(null);
try {
await agent.cancelAgentChat(sid);
setTurns((prev) => [...prev, { role: "tool", label: "Tour interrompu." }]);
const stopped = await agent.stopLiveAgent(projectId, agentId);
setBusy(false);
setOpening(false);
setCurrentSession(null);
publishSessionId(null);
setTurns((prev) => [
...completeRunningProgress(prev),
{
role: "tool",
label:
stopped.kind === "structured"
? "Agent arrêté. Cascade d'arrêt demandée pour les tâches et appels MCP liés."
: "Agent arrêté.",
},
]);
} catch (e) {
if (isNotFound(e)) {
setBusy(false);
setOpening(false);
setCurrentSession(null);
publishSessionId(null);
setTurns((prev) => [...prev, { role: "tool", label: "Session déjà fermée." }]);
setTurns((prev) => [
...completeRunningProgress(prev),
{ role: "tool", label: "Session déjà arrêtée." },
]);
return;
}
setError(describe(e));
setTurns((prev) => [
...prev,
{ role: "error", text: `Arrêt de l'agent échoué: ${describe(e)}` },
]);
} finally {
setStopping(false);
}
}
@ -1511,10 +1535,15 @@ export function CustomAgentChatView({
size="sm"
variant={busy ? "danger" : undefined}
className="shrink-0 whitespace-nowrap"
disabled={busy ? !supported || !currentSession : !canSend}
onClick={() => (busy ? void cancel() : void send())}
disabled={
busy
? !supported || !currentSession || !agent.stopLiveAgent || stopping
: !canSend
}
loading={stopping}
onClick={() => (busy ? void stopAgent() : void send())}
>
{busy ? "Stop" : "Envoyer"}
{stopping ? "Arrêt en cours..." : busy ? "Stop" : "Envoyer"}
</Button>
</div>
</div>