merge(orchestrator): délégation simple vers UX renvoie son Final structuré (#207, QA verte)
This commit is contained in:
@ -43,7 +43,10 @@ use crate::model_server::{
|
|||||||
};
|
};
|
||||||
use crate::project::project_context_path;
|
use crate::project::project_context_path;
|
||||||
use crate::skill::{AgentCapability, AssignedSkillResolver, ResolveAgentCapabilities};
|
use crate::skill::{AgentCapability, AssignedSkillResolver, ResolveAgentCapabilities};
|
||||||
use crate::terminal::{StructuredSessions, TerminalSessions};
|
use crate::terminal::{
|
||||||
|
stop_live_agent_runtime_in_project, CloseTerminal, LiveSessions, StructuredSessions,
|
||||||
|
TerminalSessions,
|
||||||
|
};
|
||||||
use crate::workstate::GetLiveStateLean;
|
use crate::workstate::GetLiveStateLean;
|
||||||
|
|
||||||
/// Directory (relative to `.ideai/`) under which agent contexts are written.
|
/// Directory (relative to `.ideai/`) under which agent contexts are written.
|
||||||
@ -1033,13 +1036,29 @@ pub struct DeleteAgentInput {
|
|||||||
pub struct DeleteAgent {
|
pub struct DeleteAgent {
|
||||||
contexts: Arc<dyn AgentContextStore>,
|
contexts: Arc<dyn AgentContextStore>,
|
||||||
events: Arc<dyn EventBus>,
|
events: Arc<dyn EventBus>,
|
||||||
|
runtime_cleanup: Option<(Arc<LiveSessions>, Arc<CloseTerminal>)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl DeleteAgent {
|
impl DeleteAgent {
|
||||||
/// Builds the use case.
|
/// Builds the use case.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn new(contexts: Arc<dyn AgentContextStore>, events: Arc<dyn EventBus>) -> Self {
|
pub fn new(contexts: Arc<dyn AgentContextStore>, events: Arc<dyn EventBus>) -> Self {
|
||||||
Self { contexts, events }
|
Self {
|
||||||
|
contexts,
|
||||||
|
events,
|
||||||
|
runtime_cleanup: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Wires live-runtime cleanup performed before the manifest entry is deleted.
|
||||||
|
#[must_use]
|
||||||
|
pub fn with_runtime_cleanup(
|
||||||
|
mut self,
|
||||||
|
live: Arc<LiveSessions>,
|
||||||
|
close: Arc<CloseTerminal>,
|
||||||
|
) -> Self {
|
||||||
|
self.runtime_cleanup = Some((live, close));
|
||||||
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Drops the manifest entry for the agent.
|
/// Drops the manifest entry for the agent.
|
||||||
@ -1058,6 +1077,11 @@ impl DeleteAgent {
|
|||||||
if entries.len() == before {
|
if entries.len() == before {
|
||||||
return Err(AppError::NotFound(format!("agent {}", input.agent_id)));
|
return Err(AppError::NotFound(format!("agent {}", input.agent_id)));
|
||||||
}
|
}
|
||||||
|
if let Some((live, close)) = &self.runtime_cleanup {
|
||||||
|
let _stopped =
|
||||||
|
stop_live_agent_runtime_in_project(&input.project, input.agent_id, live, close)
|
||||||
|
.await?;
|
||||||
|
}
|
||||||
let manifest = AgentManifest::new(manifest.version, entries)
|
let manifest = AgentManifest::new(manifest.version, entries)
|
||||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||||
self.contexts
|
self.contexts
|
||||||
|
|||||||
@ -73,15 +73,14 @@ pub enum AppError {
|
|||||||
node_id: NodeId,
|
node_id: NodeId,
|
||||||
},
|
},
|
||||||
|
|
||||||
/// The delegation target finished its turn and **returned to its prompt without
|
/// The delegation target finished its structured/headless turn without producing
|
||||||
/// calling `idea_reply`**, so the rendezvous could produce no answer. Surfaced
|
/// a [`ReplyEvent::Final`](domain::ports::ReplyEvent::Final), so the rendezvous
|
||||||
/// promptly (after a short grace window) instead of blocking the caller until the
|
/// could not capture an inline answer. Surfaced promptly instead of blocking the
|
||||||
/// long safety timeout. **Retryable**: the caller may re-ask, ideally reminding the
|
/// caller until the long safety timeout. **Retryable**: the caller may re-ask for a
|
||||||
/// target to render its result via `idea_reply` (never plain text). Carries the
|
/// concise final response. Carries the target's display name.
|
||||||
/// target's display name.
|
|
||||||
#[error(
|
#[error(
|
||||||
"agent {0} returned to its prompt without calling idea_reply (no answer was \
|
"agent {0} finished without a structured Final response (no inline answer was \
|
||||||
rendered); retry the request and ensure the agent replies via idea_reply"
|
captured); retry the request and ask for a concise final response"
|
||||||
)]
|
)]
|
||||||
TargetReturnedNoReply(String),
|
TargetReturnedNoReply(String),
|
||||||
|
|
||||||
|
|||||||
@ -6,7 +6,7 @@
|
|||||||
//! that turns a request into the *same* calls the UI makes:
|
//! that turns a request into the *same* calls the UI makes:
|
||||||
//!
|
//!
|
||||||
//! - `spawn_agent` → [`CreateAgentFromScratch`] (if unknown) then [`LaunchAgent`],
|
//! - `spawn_agent` → [`CreateAgentFromScratch`] (if unknown) then [`LaunchAgent`],
|
||||||
//! - `stop_agent` → resolve the agent's live session, then [`CloseTerminal`],
|
//! - `stop_agent` → stop the agent's live PTY or structured runtime,
|
||||||
//! - `update_agent_context` → [`UpdateAgentContext`].
|
//! - `update_agent_context` → [`UpdateAgentContext`].
|
||||||
//!
|
//!
|
||||||
//! It talks **only** to use cases and ports ([`ProfileStore`], [`TerminalSessions`]):
|
//! It talks **only** to use cases and ports ([`ProfileStore`], [`TerminalSessions`]):
|
||||||
@ -53,7 +53,10 @@ use crate::orchestrator::{
|
|||||||
WriteMemoryInput,
|
WriteMemoryInput,
|
||||||
};
|
};
|
||||||
use crate::skill::{CreateSkill, CreateSkillInput, ReadSkill, ReadSkillInput};
|
use crate::skill::{CreateSkill, CreateSkillInput, ReadSkill, ReadSkillInput};
|
||||||
use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions};
|
use crate::terminal::{
|
||||||
|
stop_live_agent_runtime_in_project, CloseTerminal, LiveSessions, StructuredSessions,
|
||||||
|
TerminalSessions,
|
||||||
|
};
|
||||||
use crate::workstate::{
|
use crate::workstate::{
|
||||||
GetLiveStateLean, StopAgentExecutionCascade, UpdateLiveState, UpdateLiveStateInput,
|
GetLiveStateLean, StopAgentExecutionCascade, UpdateLiveState, UpdateLiveStateInput,
|
||||||
};
|
};
|
||||||
@ -431,9 +434,9 @@ pub struct OrchestratorService {
|
|||||||
/// Séparé de [`WaitForGraph`] qui reste un objet domaine minimal de détection de
|
/// Séparé de [`WaitForGraph`] qui reste un objet domaine minimal de détection de
|
||||||
/// cycle, sans API de traversal.
|
/// cycle, sans API de traversal.
|
||||||
active_waits: StdMutex<Vec<(AgentId, AgentId)>>,
|
active_waits: StdMutex<Vec<(AgentId, AgentId)>>,
|
||||||
/// Composite business rendezvous currently driven by a target agent. Used to
|
/// Headless rendezvous currently driven by a target agent. Used to structurally
|
||||||
/// structurally attach `idea_run_in_background` tasks launched during
|
/// attach `idea_run_in_background` tasks for observability/cancellation while the
|
||||||
/// `idea_ask_agent` without parsing the model's textual `Final`.
|
/// synchronous answer remains the first structured `Final`.
|
||||||
active_rendezvous: StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
|
active_rendezvous: StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
|
||||||
/// Bus d'événements pour publier [`DomainEvent::AgentReplied`] à l'issue d'un
|
/// Bus d'événements pour publier [`DomainEvent::AgentReplied`] à l'issue d'un
|
||||||
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
|
/// `ask` réussi (§17.4). Injecté via [`Self::with_events`] ; `None` ⇒ pas de
|
||||||
@ -751,19 +754,6 @@ impl OrchestratorService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn active_rendezvous_has_background_tasks(
|
|
||||||
&self,
|
|
||||||
agent: RuntimeAgentKey,
|
|
||||||
rendezvous_id: RendezvousId,
|
|
||||||
) -> bool {
|
|
||||||
self.active_rendezvous
|
|
||||||
.lock()
|
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
||||||
.get(&agent)
|
|
||||||
.filter(|active| active.link.rendezvous_id == rendezvous_id)
|
|
||||||
.is_some_and(|active| !active.background_tasks.is_empty())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Returns the transitive set of agents currently waited on by `agent`.
|
/// Returns the transitive set of agents currently waited on by `agent`.
|
||||||
///
|
///
|
||||||
/// Used by user-driven cancellation: stopping A while A waits on B should also
|
/// Used by user-driven cancellation: stopping A while A waits on B should also
|
||||||
@ -893,27 +883,6 @@ impl OrchestratorService {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn mark_rendezvous_task_waiting(&self, task_id: Option<TaskId>) -> Result<(), AppError> {
|
|
||||||
let (Some(store), Some(task_id)) = (&self.background_tasks, task_id) else {
|
|
||||||
return Ok(());
|
|
||||||
};
|
|
||||||
let task = store
|
|
||||||
.get(task_id)
|
|
||||||
.await
|
|
||||||
.map_err(|err| AppError::Store(err.to_string()))?
|
|
||||||
.ok_or_else(|| AppError::Store(format!("background task {task_id} not found")))?;
|
|
||||||
if task.is_terminal() || task.state == BackgroundTaskState::Waiting {
|
|
||||||
return Ok(());
|
|
||||||
}
|
|
||||||
let waiting = task
|
|
||||||
.transition(BackgroundTaskState::Waiting, self.now_ms())
|
|
||||||
.map_err(|err| AppError::Internal(err.to_string()))?;
|
|
||||||
store
|
|
||||||
.save(&waiting)
|
|
||||||
.await
|
|
||||||
.map_err(|err| AppError::Store(err.to_string()))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn complete_rendezvous_task_failure(
|
async fn complete_rendezvous_task_failure(
|
||||||
&self,
|
&self,
|
||||||
project: &Project,
|
project: &Project,
|
||||||
@ -1963,7 +1932,7 @@ impl OrchestratorService {
|
|||||||
// Timeout de tour piloté par profil (lot 2) + armement du seuil de stall, AVANT
|
// Timeout de tour piloté par profil (lot 2) + armement du seuil de stall, AVANT
|
||||||
// l'enqueue qui démarre le tour (le médiateur arme alors sa fenêtre de vivacité).
|
// l'enqueue qui démarre le tour (le médiateur arme alors sa fenêtre de vivacité).
|
||||||
let turn_timeout = self.turn_timeout_for(project, agent_id).await;
|
let turn_timeout = self.turn_timeout_for(project, agent_id).await;
|
||||||
let pending = input.enqueue_silent(agent_key, ticket);
|
let _pending = input.enqueue_silent(agent_key, ticket);
|
||||||
let rendezvous_link = BackgroundTaskRendezvousLink {
|
let rendezvous_link = BackgroundTaskRendezvousLink {
|
||||||
rendezvous_id: RendezvousId::new_random(),
|
rendezvous_id: RendezvousId::new_random(),
|
||||||
ticket_id,
|
ticket_id,
|
||||||
@ -2071,7 +2040,7 @@ impl OrchestratorService {
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
|
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
|
||||||
let initial_result = match self
|
let result = match self
|
||||||
.run_ask_with_watchdog(wait, turn_timeout, &project.root, agent_id, target, started)
|
.run_ask_with_watchdog(wait, turn_timeout, &project.root, agent_id, target, started)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
@ -2163,64 +2132,6 @@ impl OrchestratorService {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
let result = if self
|
|
||||||
.active_rendezvous_has_background_tasks(agent_key, rendezvous_link.rendezvous_id)
|
|
||||||
{
|
|
||||||
self.mark_rendezvous_task_waiting(rendezvous_task).await?;
|
|
||||||
crate::diag!(
|
|
||||||
"[rendezvous] ask waiting on background: target={target} \
|
|
||||||
(agent {agent_id}) ticket={ticket_id} rendezvous={} first_final_len={}",
|
|
||||||
rendezvous_link.rendezvous_id,
|
|
||||||
initial_result.len(),
|
|
||||||
);
|
|
||||||
let remaining = self.ask_ceiling.saturating_sub(started.elapsed());
|
|
||||||
match tokio::time::timeout(remaining, pending).await {
|
|
||||||
Ok(Ok(domain::mailbox::TurnResolution::Replied(content))) => content,
|
|
||||||
Ok(Ok(domain::mailbox::TurnResolution::ReturnedToPromptNoReply)) => {
|
|
||||||
let err = AppError::TargetReturnedNoReply(target.to_owned());
|
|
||||||
self.complete_rendezvous_task_failure(
|
|
||||||
project,
|
|
||||||
agent_id,
|
|
||||||
rendezvous_task,
|
|
||||||
format!("NoReply: {err}"),
|
|
||||||
)
|
|
||||||
.await?;
|
|
||||||
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
|
|
||||||
.await;
|
|
||||||
return Err(err);
|
|
||||||
}
|
|
||||||
Ok(Err(err)) => {
|
|
||||||
let err = AppError::Process(err.to_string());
|
|
||||||
self.complete_rendezvous_task_failure(
|
|
||||||
project,
|
|
||||||
agent_id,
|
|
||||||
rendezvous_task,
|
|
||||||
format!("NoReply: {err}"),
|
|
||||||
)
|
|
||||||
.await?;
|
|
||||||
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
|
|
||||||
.await;
|
|
||||||
return Err(err);
|
|
||||||
}
|
|
||||||
Err(_elapsed) => {
|
|
||||||
self.complete_rendezvous_task_failure(
|
|
||||||
project,
|
|
||||||
agent_id,
|
|
||||||
rendezvous_task,
|
|
||||||
format!(
|
|
||||||
"Timeout: composite rendezvous ceiling reached for target {target}"
|
|
||||||
),
|
|
||||||
)
|
|
||||||
.await?;
|
|
||||||
self.mark_target_done_best_effort(&project.root, agent_id, ticket_id)
|
|
||||||
.await;
|
|
||||||
return Err(AppError::TargetCeilingActive(target.to_owned()));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
initial_result
|
|
||||||
};
|
|
||||||
|
|
||||||
self.complete_rendezvous_task_success(project, agent_id, rendezvous_task, &result)
|
self.complete_rendezvous_task_success(project, agent_id, rendezvous_task, &result)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
@ -2665,7 +2576,7 @@ impl OrchestratorService {
|
|||||||
.as_ref()
|
.as_ref()
|
||||||
.and_then(|p| p.runtime_for(project, agent_id)),
|
.and_then(|p| p.runtime_for(project, agent_id)),
|
||||||
allow_structured_alongside_pty: true,
|
allow_structured_alongside_pty: true,
|
||||||
require_structured: false,
|
require_structured: true,
|
||||||
})
|
})
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
@ -2774,7 +2685,7 @@ impl OrchestratorService {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
/// `stop_agent`: translate the agent name → its live session → `CloseTerminal`.
|
/// `stop_agent`: translate the agent name → its live runtime → stop it.
|
||||||
async fn stop_agent(
|
async fn stop_agent(
|
||||||
&self,
|
&self,
|
||||||
project: &Project,
|
project: &Project,
|
||||||
@ -2785,14 +2696,19 @@ impl OrchestratorService {
|
|||||||
.await?
|
.await?
|
||||||
.ok_or_else(|| AppError::NotFound(format!("agent {name}")))?;
|
.ok_or_else(|| AppError::NotFound(format!("agent {name}")))?;
|
||||||
|
|
||||||
let session_id = self
|
let structured = self
|
||||||
.sessions
|
.structured
|
||||||
.session_for_agent_in_project(project.id, &agent_id)
|
.clone()
|
||||||
.ok_or_else(|| AppError::NotFound(format!("running session for agent {name}")))?;
|
.unwrap_or_else(|| Arc::new(StructuredSessions::new()));
|
||||||
|
let live = LiveSessions::new(Arc::clone(&self.sessions), structured);
|
||||||
self.close_terminal
|
if stop_live_agent_runtime_in_project(project, agent_id, &live, &self.close_terminal)
|
||||||
.execute(CloseTerminalInput { session_id })
|
.await?
|
||||||
.await?;
|
.is_empty()
|
||||||
|
{
|
||||||
|
return Err(AppError::NotFound(format!(
|
||||||
|
"running session for agent {name}"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
Ok(OrchestratorOutcome {
|
Ok(OrchestratorOutcome {
|
||||||
detail: format!("stopped agent {name}"),
|
detail: format!("stopped agent {name}"),
|
||||||
|
|||||||
@ -33,6 +33,7 @@ pub use registry::{
|
|||||||
TerminalSessions,
|
TerminalSessions,
|
||||||
};
|
};
|
||||||
pub use usecases::{
|
pub use usecases::{
|
||||||
CloseTerminal, CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput,
|
stop_live_agent_runtime_in_project, CloseTerminal, CloseTerminalInput, CloseTerminalOutput,
|
||||||
OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, WriteToTerminal, WriteToTerminalInput,
|
OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput,
|
||||||
|
StoppedAgentRuntime, WriteToTerminal, WriteToTerminalInput,
|
||||||
};
|
};
|
||||||
|
|||||||
@ -4,13 +4,13 @@ use std::sync::Arc;
|
|||||||
|
|
||||||
use domain::ports::{EventBus, PtyPort, SpawnSpec};
|
use domain::ports::{EventBus, PtyPort, SpawnSpec};
|
||||||
use domain::{
|
use domain::{
|
||||||
DomainEvent, NodeId, ProjectPath, PtySize, SessionId, SessionKind, SessionStatus,
|
AgentId, DomainEvent, NodeId, Project, ProjectPath, PtySize, SessionId, SessionKind,
|
||||||
TerminalSession,
|
SessionStatus, TerminalSession,
|
||||||
};
|
};
|
||||||
|
|
||||||
use crate::error::AppError;
|
use crate::error::AppError;
|
||||||
|
|
||||||
use super::registry::TerminalSessions;
|
use super::registry::{LiveSessionKind, LiveSessions, TerminalSessions};
|
||||||
|
|
||||||
/// Input for [`OpenTerminal::execute`].
|
/// Input for [`OpenTerminal::execute`].
|
||||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
@ -230,6 +230,59 @@ impl CloseTerminal {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Runtime session stopped for one agent.
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
pub struct StoppedAgentRuntime {
|
||||||
|
/// Agent whose live runtime was stopped.
|
||||||
|
pub agent_id: AgentId,
|
||||||
|
/// Live session id removed from its registry.
|
||||||
|
pub session_id: SessionId,
|
||||||
|
/// Runtime registry that owned the session.
|
||||||
|
pub kind: LiveSessionKind,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Stops one agent's live runtime across both registries.
|
||||||
|
///
|
||||||
|
/// PTY sessions are closed through [`CloseTerminal`]. Structured sessions are
|
||||||
|
/// removed from their registry before `shutdown()` so relaunch guards stop seeing
|
||||||
|
/// them as live even while shutdown runs.
|
||||||
|
pub async fn stop_live_agent_runtime_in_project(
|
||||||
|
project: &Project,
|
||||||
|
agent_id: AgentId,
|
||||||
|
live: &LiveSessions,
|
||||||
|
close: &CloseTerminal,
|
||||||
|
) -> Result<Vec<StoppedAgentRuntime>, AppError> {
|
||||||
|
let mut stopped = Vec::new();
|
||||||
|
|
||||||
|
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()))?;
|
||||||
|
}
|
||||||
|
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,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(stopped)
|
||||||
|
}
|
||||||
|
|
||||||
/// The platform default interactive shell.
|
/// The platform default interactive shell.
|
||||||
///
|
///
|
||||||
/// Resolution policy lives in the application layer (a metier default), not the
|
/// Resolution policy lives in the application layer (a metier default), not the
|
||||||
|
|||||||
@ -15,7 +15,9 @@ use domain::{AgentId, NodeId, Project, RuntimeAgentKey, SessionId};
|
|||||||
|
|
||||||
use crate::error::AppError;
|
use crate::error::AppError;
|
||||||
use crate::orchestrator::OrchestratorService;
|
use crate::orchestrator::OrchestratorService;
|
||||||
use crate::terminal::{CloseTerminal, CloseTerminalInput, LiveSessionKind, LiveSessions};
|
use crate::terminal::{
|
||||||
|
stop_live_agent_runtime_in_project, CloseTerminal, LiveSessionKind, LiveSessions,
|
||||||
|
};
|
||||||
|
|
||||||
/// Input for [`AttachLiveAgent::execute`].
|
/// Input for [`AttachLiveAgent::execute`].
|
||||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
@ -223,42 +225,15 @@ impl StopLiveAgent {
|
|||||||
project: &Project,
|
project: &Project,
|
||||||
agent_id: AgentId,
|
agent_id: AgentId,
|
||||||
) -> Result<StopLiveAgentOutput, AppError> {
|
) -> Result<StopLiveAgentOutput, AppError> {
|
||||||
// PTY first: delegate to the existing close primitive (removes + kills).
|
stop_live_agent_runtime_in_project(project, agent_id, &self.live, &self.close)
|
||||||
if let Some(session_id) = self
|
.await?
|
||||||
.live
|
.into_iter()
|
||||||
.pty
|
.next()
|
||||||
.session_for_agent_in_project(project.id, &agent_id)
|
.map(|stopped| StopLiveAgentOutput {
|
||||||
{
|
agent_id: stopped.agent_id,
|
||||||
self.close
|
session_id: stopped.session_id,
|
||||||
.execute(CloseTerminalInput { session_id })
|
kind: stopped.kind,
|
||||||
.await?;
|
})
|
||||||
return Ok(StopLiveAgentOutput {
|
.ok_or_else(|| AppError::NotFound(format!("running session for agent {agent_id}")))
|
||||||
agent_id,
|
|
||||||
session_id,
|
|
||||||
kind: LiveSessionKind::Pty,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
// Structured: remove from the registry first (so the uniqueness guard no
|
|
||||||
// longer sees a live session), then shut the session down out of the lock.
|
|
||||||
if let Some(session_id) = self
|
|
||||||
.live
|
|
||||||
.structured
|
|
||||||
.session_id_for_agent_in_project(project.id, &agent_id)
|
|
||||||
{
|
|
||||||
if let Some(session) = self.live.structured.remove(&session_id) {
|
|
||||||
session
|
|
||||||
.shutdown()
|
|
||||||
.await
|
|
||||||
.map_err(|e| AppError::Process(e.to_string()))?;
|
|
||||||
}
|
|
||||||
return Ok(StopLiveAgentOutput {
|
|
||||||
agent_id,
|
|
||||||
session_id,
|
|
||||||
kind: LiveSessionKind::Structured,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
Err(AppError::NotFound(format!(
|
|
||||||
"running session for agent {agent_id}"
|
|
||||||
)))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -47,10 +47,11 @@ use domain::{PtySize, SessionId, SkillId, SkillRef};
|
|||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
use application::{
|
use application::{
|
||||||
CreateAgentFromScratch, CreateAgentInput, DeleteAgent, DeleteAgentInput, LaunchAgent,
|
CloseTerminal, CreateAgentFromScratch, CreateAgentInput, DeleteAgent, DeleteAgentInput,
|
||||||
LaunchAgentInput, ListAgents, ListAgentsInput, PermissionProjectorRegistry, ReadAgentContext,
|
LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, LiveSessions,
|
||||||
ReadAgentContextInput, StructuredRoutingMode, StructuredSessions, TerminalSessions,
|
PermissionProjectorRegistry, ReadAgentContext, ReadAgentContextInput, StructuredRoutingMode,
|
||||||
UpdateAgentContext, UpdateAgentContextInput, UpdateAgentEffort, UpdateAgentEffortInput,
|
StructuredSessions, TerminalSessions, UpdateAgentContext, UpdateAgentContextInput,
|
||||||
|
UpdateAgentEffort, UpdateAgentEffortInput,
|
||||||
};
|
};
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@ -585,6 +586,7 @@ impl PtyPort for FakePty {
|
|||||||
struct FakeSession {
|
struct FakeSession {
|
||||||
id: SessionId,
|
id: SessionId,
|
||||||
conversation_id: Option<String>,
|
conversation_id: Option<String>,
|
||||||
|
shutdowns: Arc<Mutex<Vec<SessionId>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
@ -602,6 +604,7 @@ impl AgentSession for FakeSession {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||||
|
self.shutdowns.lock().unwrap().push(self.id);
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -613,6 +616,7 @@ struct FakeStructuredFactory {
|
|||||||
efforts: Arc<Mutex<Vec<Option<String>>>>,
|
efforts: Arc<Mutex<Vec<Option<String>>>>,
|
||||||
envs: Arc<Mutex<Vec<Vec<(String, String)>>>>,
|
envs: Arc<Mutex<Vec<Vec<(String, String)>>>>,
|
||||||
policies: Arc<Mutex<Vec<Option<domain::ports::StructuredProviderLaunchPolicy>>>>,
|
policies: Arc<Mutex<Vec<Option<domain::ports::StructuredProviderLaunchPolicy>>>>,
|
||||||
|
shutdowns: Arc<Mutex<Vec<SessionId>>>,
|
||||||
next_session: SessionId,
|
next_session: SessionId,
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -624,6 +628,7 @@ impl FakeStructuredFactory {
|
|||||||
efforts: Arc::new(Mutex::new(Vec::new())),
|
efforts: Arc::new(Mutex::new(Vec::new())),
|
||||||
envs: Arc::new(Mutex::new(Vec::new())),
|
envs: Arc::new(Mutex::new(Vec::new())),
|
||||||
policies: Arc::new(Mutex::new(Vec::new())),
|
policies: Arc::new(Mutex::new(Vec::new())),
|
||||||
|
shutdowns: Arc::new(Mutex::new(Vec::new())),
|
||||||
next_session,
|
next_session,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -679,6 +684,7 @@ impl AgentSessionFactory for FakeStructuredFactory {
|
|||||||
Ok(Arc::new(FakeSession {
|
Ok(Arc::new(FakeSession {
|
||||||
id: self.next_session,
|
id: self.next_session,
|
||||||
conversation_id: Some("engine-session-1".to_owned()),
|
conversation_id: Some("engine-session-1".to_owned()),
|
||||||
|
shutdowns: Arc::clone(&self.shutdowns),
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -1048,6 +1054,56 @@ async fn delete_removes_entry_then_unknown_is_not_found() {
|
|||||||
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
|
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn delete_agent_cleans_structured_then_pty_runtime_before_manifest_removal() {
|
||||||
|
let a = scratch_agent(aid(1), "Backend", "agents/backend.md", pid(9));
|
||||||
|
let contexts = FakeContexts::with_agent(&a, "ctx");
|
||||||
|
let sessions = Arc::new(TerminalSessions::new());
|
||||||
|
let structured = Arc::new(StructuredSessions::new());
|
||||||
|
let live = Arc::new(LiveSessions::new(
|
||||||
|
Arc::clone(&sessions),
|
||||||
|
Arc::clone(&structured),
|
||||||
|
));
|
||||||
|
let pty = FakePty::new(trace(), sid(42));
|
||||||
|
let close = Arc::new(CloseTerminal::new(
|
||||||
|
Arc::new(pty.clone()),
|
||||||
|
Arc::clone(&sessions),
|
||||||
|
));
|
||||||
|
let shutdowns = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
structured.insert_in_project(
|
||||||
|
project().id,
|
||||||
|
Arc::new(FakeSession {
|
||||||
|
id: sid(555),
|
||||||
|
conversation_id: Some("engine-session-stale".to_owned()),
|
||||||
|
shutdowns: Arc::clone(&shutdowns),
|
||||||
|
}),
|
||||||
|
a.id,
|
||||||
|
nid(7),
|
||||||
|
);
|
||||||
|
seed_live_agent_session(&sessions, a.id, nid(8), sid(42));
|
||||||
|
|
||||||
|
let delete = DeleteAgent::new(Arc::new(contexts.clone()), Arc::new(SpyBus::default()))
|
||||||
|
.with_runtime_cleanup(live, close);
|
||||||
|
|
||||||
|
delete
|
||||||
|
.execute(DeleteAgentInput {
|
||||||
|
project: project(),
|
||||||
|
agent_id: a.id,
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(*shutdowns.lock().unwrap(), vec![sid(555)]);
|
||||||
|
assert_eq!(pty.kills(), vec![sid(42)]);
|
||||||
|
assert!(structured
|
||||||
|
.session_for_agent_in_project(project().id, &a.id)
|
||||||
|
.is_none());
|
||||||
|
assert!(sessions
|
||||||
|
.session_for_agent_in_project(project().id, &a.id)
|
||||||
|
.is_none());
|
||||||
|
assert!(contexts.manifest().entries.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// LaunchAgent
|
// LaunchAgent
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@ -1369,6 +1425,7 @@ async fn chat_intent_reattaches_existing_structured_session_after_stale_pty_tran
|
|||||||
Arc::new(FakeSession {
|
Arc::new(FakeSession {
|
||||||
id: sid(555),
|
id: sid(555),
|
||||||
conversation_id: Some("engine-session-stale".to_owned()),
|
conversation_id: Some("engine-session-stale".to_owned()),
|
||||||
|
shutdowns: Arc::new(Mutex::new(Vec::new())),
|
||||||
}),
|
}),
|
||||||
agent.id,
|
agent.id,
|
||||||
stale_custom_node,
|
stale_custom_node,
|
||||||
|
|||||||
@ -390,6 +390,10 @@ impl RecordingBackgroundStore {
|
|||||||
fn task(&self, id: TaskId) -> Option<BackgroundTask> {
|
fn task(&self, id: TaskId) -> Option<BackgroundTask> {
|
||||||
self.tasks.lock().unwrap().get(&id).cloned()
|
self.tasks.lock().unwrap().get(&id).cloned()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn tasks(&self) -> Vec<BackgroundTask> {
|
||||||
|
self.tasks.lock().unwrap().values().cloned().collect()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl BackgroundTaskStore for RecordingBackgroundStore {
|
impl BackgroundTaskStore for RecordingBackgroundStore {
|
||||||
@ -1009,6 +1013,41 @@ async fn stop_agent_kills_the_right_session() {
|
|||||||
assert!(fx.sessions.session(&sid(777)).is_none());
|
assert!(fx.sessions.session(&sid(777)).is_none());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn stop_agent_stops_structured_session() {
|
||||||
|
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
||||||
|
let contexts = FakeContexts::with_agent(&agent, "# persona");
|
||||||
|
let fx = ask_fixture(contexts);
|
||||||
|
let session = Arc::new(CompletionSession {
|
||||||
|
id: sid(9010),
|
||||||
|
agent: agent.id,
|
||||||
|
completions: fx.mailbox.completions(),
|
||||||
|
}) as Arc<dyn AgentSession>;
|
||||||
|
fx.structured
|
||||||
|
.insert_in_project(project_id(), session, agent.id, nid(9010));
|
||||||
|
assert!(fx
|
||||||
|
.structured
|
||||||
|
.session_for_agent_in_project(project_id(), &agent.id)
|
||||||
|
.is_some());
|
||||||
|
|
||||||
|
fx.service
|
||||||
|
.dispatch(
|
||||||
|
&project(),
|
||||||
|
cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("stop ok");
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
fx.pty.kills().is_empty(),
|
||||||
|
"structured stop must not kill PTY"
|
||||||
|
);
|
||||||
|
assert!(fx
|
||||||
|
.structured
|
||||||
|
.session_for_agent_in_project(project_id(), &agent.id)
|
||||||
|
.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn stop_agent_without_live_session_is_not_found() {
|
async fn stop_agent_without_live_session_is_not_found() {
|
||||||
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
||||||
@ -1025,6 +1064,47 @@ async fn stop_agent_without_live_session_is_not_found() {
|
|||||||
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
|
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_agent_after_delete_recreate_same_name_uses_new_agent_id() {
|
||||||
|
let old = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
||||||
|
let new = scratch_agent(aid(2), "dev-backend", "agents/dev-backend-2.md");
|
||||||
|
let contexts = FakeContexts::with_agent(&new, "# recreated persona");
|
||||||
|
let fx = ask_fixture(contexts);
|
||||||
|
|
||||||
|
let old_session = Arc::new(CompletionSession {
|
||||||
|
id: sid(9001),
|
||||||
|
agent: old.id,
|
||||||
|
completions: fx.mailbox.completions(),
|
||||||
|
}) as Arc<dyn AgentSession>;
|
||||||
|
fx.structured
|
||||||
|
.insert_in_project(project_id(), old_session, old.id, nid(9001));
|
||||||
|
fx.mailbox
|
||||||
|
.completions()
|
||||||
|
.push(old.id, TestCompletion::Replied("old-agent".to_owned()));
|
||||||
|
fx.mailbox
|
||||||
|
.completions()
|
||||||
|
.push(new.id, TestCompletion::Replied("new-agent".to_owned()));
|
||||||
|
|
||||||
|
let out = fx
|
||||||
|
.service
|
||||||
|
.dispatch(
|
||||||
|
&project(),
|
||||||
|
cmd(r#"{ "type":"agent.message", "targetAgent":"dev-backend", "task":"ping" }"#),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(out.reply.as_deref(), Some("new-agent"));
|
||||||
|
assert!(fx
|
||||||
|
.structured
|
||||||
|
.session_for_agent_in_project(project_id(), &old.id)
|
||||||
|
.is_some());
|
||||||
|
assert!(fx
|
||||||
|
.structured
|
||||||
|
.session_for_agent_in_project(project_id(), &new.id)
|
||||||
|
.is_some());
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn update_agent_context_overwrites_md() {
|
async fn update_agent_context_overwrites_md() {
|
||||||
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
|
||||||
@ -4198,6 +4278,96 @@ impl AgentSession for RateLimitedSession {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct EmptyStructuredSession {
|
||||||
|
id: SessionId,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentSession for EmptyStructuredSession {
|
||||||
|
fn id(&self) -> SessionId {
|
||||||
|
self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
fn conversation_id(&self) -> Option<String> {
|
||||||
|
Some(format!("engine-{}", self.id))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||||
|
Ok(Box::new(Vec::<ReplyEvent>::new().into_iter()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct FailingStructuredSession {
|
||||||
|
id: SessionId,
|
||||||
|
message: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentSession for FailingStructuredSession {
|
||||||
|
fn id(&self) -> SessionId {
|
||||||
|
self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
fn conversation_id(&self) -> Option<String> {
|
||||||
|
Some(format!("engine-{}", self.id))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||||
|
Err(AgentSessionError::Io(self.message.clone()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct BackgroundThenFinalSession {
|
||||||
|
id: SessionId,
|
||||||
|
service: Arc<OrchestratorService>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentSession for BackgroundThenFinalSession {
|
||||||
|
fn id(&self) -> SessionId {
|
||||||
|
self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
fn conversation_id(&self) -> Option<String> {
|
||||||
|
Some(format!("engine-{}", self.id))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||||
|
self.service
|
||||||
|
.dispatch(
|
||||||
|
&project(),
|
||||||
|
OrchestratorCommand::RunInBackground {
|
||||||
|
owner: aid(1),
|
||||||
|
label: "correlated child".to_owned(),
|
||||||
|
command: "bash".to_owned(),
|
||||||
|
args: vec!["-lc".to_owned(), "sleep 60".to_owned()],
|
||||||
|
cwd: None,
|
||||||
|
deadline_ms: None,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.map_err(|err| AgentSessionError::Io(err.to_string()))?;
|
||||||
|
Ok(Box::new(
|
||||||
|
vec![ReplyEvent::Final {
|
||||||
|
content: "business final".to_owned(),
|
||||||
|
}]
|
||||||
|
.into_iter(),
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
struct NoopScheduler {
|
struct NoopScheduler {
|
||||||
next: Mutex<u128>,
|
next: Mutex<u128>,
|
||||||
@ -4241,9 +4411,24 @@ struct StructuredAskFixture {
|
|||||||
pty: FakePty,
|
pty: FakePty,
|
||||||
bus: SpyBus,
|
bus: SpyBus,
|
||||||
session_limits: Arc<SessionLimitService>,
|
session_limits: Arc<SessionLimitService>,
|
||||||
|
background_store: Arc<RecordingBackgroundStore>,
|
||||||
|
background_runner: Arc<RecordingBackgroundRunner>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
|
fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
|
||||||
|
structured_ask_fixture_with_background(contexts, false)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn structured_ask_fixture_with_observable_background(
|
||||||
|
contexts: FakeContexts,
|
||||||
|
) -> StructuredAskFixture {
|
||||||
|
structured_ask_fixture_with_background(contexts, true)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn structured_ask_fixture_with_background(
|
||||||
|
contexts: FakeContexts,
|
||||||
|
enable_background: bool,
|
||||||
|
) -> StructuredAskFixture {
|
||||||
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
|
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
|
||||||
let sessions = Arc::new(TerminalSessions::new());
|
let sessions = Arc::new(TerminalSessions::new());
|
||||||
let pty = FakePty::new(sid(777));
|
let pty = FakePty::new(sid(777));
|
||||||
@ -4251,6 +4436,15 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
|
|||||||
let mailbox = Arc::new(TestMailbox::new());
|
let mailbox = Arc::new(TestMailbox::new());
|
||||||
let structured = Arc::new(StructuredSessions::new());
|
let structured = Arc::new(StructuredSessions::new());
|
||||||
let factory = CountingFactory::new();
|
let factory = CountingFactory::new();
|
||||||
|
let background_store = Arc::new(RecordingBackgroundStore::default());
|
||||||
|
let background_runner = Arc::new(RecordingBackgroundRunner::default());
|
||||||
|
let background_clock = Arc::new(FixedMillisClock(42)) as Arc<dyn domain::ports::Clock>;
|
||||||
|
let spawn_background = Arc::new(SpawnBackgroundCommand::new(
|
||||||
|
Arc::clone(&background_store) as Arc<dyn BackgroundTaskStore>,
|
||||||
|
Arc::clone(&background_runner) as Arc<dyn BackgroundTaskRunner>,
|
||||||
|
Arc::clone(&background_clock),
|
||||||
|
Arc::new(SeqIds::new()) as Arc<dyn IdGenerator>,
|
||||||
|
));
|
||||||
|
|
||||||
let create = Arc::new(CreateAgentFromScratch::new(
|
let create = Arc::new(CreateAgentFromScratch::new(
|
||||||
Arc::new(contexts.clone()),
|
Arc::new(contexts.clone()),
|
||||||
@ -4301,28 +4495,39 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
|
|||||||
Arc::new(NoopResumer),
|
Arc::new(NoopResumer),
|
||||||
));
|
));
|
||||||
|
|
||||||
let service = Arc::new(
|
let mut service = OrchestratorService::new(
|
||||||
OrchestratorService::new(
|
create,
|
||||||
create,
|
launch,
|
||||||
launch,
|
list,
|
||||||
list,
|
close,
|
||||||
close,
|
update,
|
||||||
update,
|
create_skill,
|
||||||
create_skill,
|
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
||||||
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
Arc::clone(&sessions),
|
||||||
Arc::clone(&sessions),
|
)
|
||||||
)
|
.with_input_mediator(
|
||||||
.with_input_mediator(
|
Arc::clone(&mediator) as Arc<dyn InputMediator>,
|
||||||
Arc::clone(&mediator) as Arc<dyn InputMediator>,
|
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
|
||||||
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
|
)
|
||||||
)
|
.with_conversations(conversations)
|
||||||
.with_conversations(conversations)
|
.with_events(Arc::new(bus.clone()))
|
||||||
.with_events(Arc::new(bus.clone()))
|
.with_session_limits(Arc::clone(&session_limits))
|
||||||
.with_session_limits(Arc::clone(&session_limits))
|
// Même registre que le launcher : `ask_agent` y relit la session fraîchement
|
||||||
// Même registre que le launcher : `ask_agent` y relit la session fraîchement
|
// insérée par `launch_structured`.
|
||||||
// insérée par `launch_structured`.
|
.with_structured(Arc::clone(&structured));
|
||||||
.with_structured(Arc::clone(&structured)),
|
|
||||||
);
|
if enable_background {
|
||||||
|
service = service
|
||||||
|
.with_background_tasks(
|
||||||
|
Arc::clone(&background_store) as Arc<dyn BackgroundTaskStore>,
|
||||||
|
Arc::clone(&background_clock),
|
||||||
|
)
|
||||||
|
.with_spawn_background_command(spawn_background)
|
||||||
|
.with_background_task_runner(
|
||||||
|
Arc::clone(&background_runner) as Arc<dyn BackgroundTaskRunner>
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let service = Arc::new(service);
|
||||||
|
|
||||||
StructuredAskFixture {
|
StructuredAskFixture {
|
||||||
service,
|
service,
|
||||||
@ -4332,6 +4537,8 @@ fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
|
|||||||
pty,
|
pty,
|
||||||
bus,
|
bus,
|
||||||
session_limits,
|
session_limits,
|
||||||
|
background_store,
|
||||||
|
background_runner,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -4446,6 +4653,138 @@ async fn ask_structured_target_live_as_pty_keeps_visible_terminal() {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Régression #201 : une tâche de fond lancée pendant le tour headless reste corrélée
|
||||||
|
/// au rendez-vous pour l'observabilité, mais elle ne prolonge plus la réponse
|
||||||
|
/// synchrone. Le premier `Final` structured est le succès métier.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_structured_final_wins_even_with_attached_background_task() {
|
||||||
|
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||||
|
let fx = structured_ask_fixture_with_observable_background(FakeContexts::with_agent(
|
||||||
|
&agent,
|
||||||
|
"# persona",
|
||||||
|
));
|
||||||
|
fx.structured.insert_in_project(
|
||||||
|
project_id(),
|
||||||
|
Arc::new(BackgroundThenFinalSession {
|
||||||
|
id: sid(9800),
|
||||||
|
service: Arc::clone(&fx.service),
|
||||||
|
}),
|
||||||
|
aid(1),
|
||||||
|
nid(700),
|
||||||
|
);
|
||||||
|
|
||||||
|
let out = fx
|
||||||
|
.service
|
||||||
|
.dispatch(&project(), cmd(ASK_JSON))
|
||||||
|
.await
|
||||||
|
.expect("Final must win even when a background task is attached");
|
||||||
|
|
||||||
|
assert_eq!(out.reply.as_deref(), Some("business final"));
|
||||||
|
assert_eq!(
|
||||||
|
fx.background_runner.specs().len(),
|
||||||
|
1,
|
||||||
|
"the child background command is still spawned for observability"
|
||||||
|
);
|
||||||
|
|
||||||
|
let tasks = fx.background_store.tasks();
|
||||||
|
let rendezvous_task = tasks
|
||||||
|
.iter()
|
||||||
|
.find(|task| {
|
||||||
|
matches!(
|
||||||
|
task.kind,
|
||||||
|
domain::BackgroundTaskKind::HeadlessRendezvous { .. }
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.expect("headless rendezvous task persisted");
|
||||||
|
assert_eq!(
|
||||||
|
rendezvous_task.state,
|
||||||
|
BackgroundTaskState::Completed,
|
||||||
|
"the rendezvous itself is marked successful inline"
|
||||||
|
);
|
||||||
|
|
||||||
|
let child_task = tasks
|
||||||
|
.iter()
|
||||||
|
.find(|task| matches!(task.kind, domain::BackgroundTaskKind::Command { .. }))
|
||||||
|
.expect("attached command task persisted");
|
||||||
|
assert_eq!(
|
||||||
|
child_task.state,
|
||||||
|
BackgroundTaskState::Running,
|
||||||
|
"the attached background task remains observable without gating the reply"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
child_task.rendezvous.is_some(),
|
||||||
|
"the child task keeps its rendezvous correlation"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Régression #201 : si le flux structuré se clôt sans `Final`, l'erreur publique
|
||||||
|
/// décrit l'absence de réponse inline exploitable, sans mentionner l'ancien protocole
|
||||||
|
/// `idea_reply`.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_structured_closed_stream_without_final_returns_inline_no_reply_error() {
|
||||||
|
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||||
|
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
|
||||||
|
fx.structured.insert_in_project(
|
||||||
|
project_id(),
|
||||||
|
Arc::new(EmptyStructuredSession { id: sid(9801) }),
|
||||||
|
aid(1),
|
||||||
|
nid(700),
|
||||||
|
);
|
||||||
|
|
||||||
|
let err = fx
|
||||||
|
.service
|
||||||
|
.dispatch(&project(), cmd(ASK_JSON))
|
||||||
|
.await
|
||||||
|
.expect_err("closed stream without Final must be a typed no-reply");
|
||||||
|
|
||||||
|
assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY");
|
||||||
|
let message = err.to_string();
|
||||||
|
assert!(
|
||||||
|
message.contains("structured Final response"),
|
||||||
|
"message must explain the missing inline Final, got: {message}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!message.contains("idea_reply"),
|
||||||
|
"message must not mention the legacy reply tool, got: {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Régression #207 : si la CLI structurée échoue avant tout stdout exploitable
|
||||||
|
/// (ex. `401 Unauthorized`), l'orchestrateur doit remonter l'erreur réelle en
|
||||||
|
/// `PROCESS` au lieu de maquiller l'échec en faux no-reply.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ask_structured_session_error_surfaces_process_message_not_no_reply() {
|
||||||
|
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
|
||||||
|
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
|
||||||
|
fx.structured.insert_in_project(
|
||||||
|
project_id(),
|
||||||
|
Arc::new(FailingStructuredSession {
|
||||||
|
id: sid(9802),
|
||||||
|
message: "codex exec exited with code 1: 401 Unauthorized: Missing bearer"
|
||||||
|
.to_owned(),
|
||||||
|
}),
|
||||||
|
aid(1),
|
||||||
|
nid(700),
|
||||||
|
);
|
||||||
|
|
||||||
|
let err = fx
|
||||||
|
.service
|
||||||
|
.dispatch(&project(), cmd(ASK_JSON))
|
||||||
|
.await
|
||||||
|
.expect_err("session startup failure must surface as a process error");
|
||||||
|
|
||||||
|
assert_eq!(err.code(), "PROCESS");
|
||||||
|
let message = err.to_string();
|
||||||
|
assert!(
|
||||||
|
message.contains("401 Unauthorized: Missing bearer"),
|
||||||
|
"public error must preserve stderr detail, got: {message}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!message.contains("structured Final response"),
|
||||||
|
"public error must not fall back to no-reply wording, got: {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/// Sous-lot adjacent #7/F2 : une limite détectée pendant un rendez-vous structuré
|
/// Sous-lot adjacent #7/F2 : une limite détectée pendant un rendez-vous structuré
|
||||||
/// doit aussi passer par `SessionLimitService::on_rate_limited` pour la cible, afin
|
/// doit aussi passer par `SessionLimitService::on_rate_limited` pour la cible, afin
|
||||||
/// de publier le flux mono-agent `AgentRateLimited` puis `AgentResumeScheduled`.
|
/// de publier le flux mono-agent `AgentRateLimited` puis `AgentResumeScheduled`.
|
||||||
|
|||||||
@ -1514,6 +1514,10 @@ impl BackendCore {
|
|||||||
// `profile.structured_adapter`. Injectés dans LaunchAgent (routage §17.4) et
|
// `profile.structured_adapter`. Injectés dans LaunchAgent (routage §17.4) et
|
||||||
// ChangeAgentProfile (shutdown polymorphe au hot-swap).
|
// ChangeAgentProfile (shutdown polymorphe au hot-swap).
|
||||||
let structured_sessions = Arc::new(StructuredSessions::new());
|
let structured_sessions = Arc::new(StructuredSessions::new());
|
||||||
|
let live_sessions = Arc::new(LiveSessions::new(
|
||||||
|
Arc::clone(&terminal_sessions),
|
||||||
|
Arc::clone(&structured_sessions),
|
||||||
|
));
|
||||||
let openai_tool_invoker = Arc::new(LateBoundOpenAiToolInvoker::new());
|
let openai_tool_invoker = Arc::new(LateBoundOpenAiToolInvoker::new());
|
||||||
let openai_tool_invoker_port = Arc::clone(&openai_tool_invoker) as Arc<dyn ToolInvoker>;
|
let openai_tool_invoker_port = Arc::clone(&openai_tool_invoker) as Arc<dyn ToolInvoker>;
|
||||||
let session_factory = Arc::new(
|
let session_factory = Arc::new(
|
||||||
@ -2065,10 +2069,10 @@ impl BackendCore {
|
|||||||
));
|
));
|
||||||
let read_agent_context = Arc::new(ReadAgentContext::new(Arc::clone(&contexts_port)));
|
let read_agent_context = Arc::new(ReadAgentContext::new(Arc::clone(&contexts_port)));
|
||||||
let update_agent_context = Arc::new(UpdateAgentContext::new(Arc::clone(&contexts_port)));
|
let update_agent_context = Arc::new(UpdateAgentContext::new(Arc::clone(&contexts_port)));
|
||||||
let delete_agent = Arc::new(DeleteAgent::new(
|
let delete_agent = Arc::new(
|
||||||
Arc::clone(&contexts_port),
|
DeleteAgent::new(Arc::clone(&contexts_port), Arc::clone(&events_port))
|
||||||
Arc::clone(&events_port),
|
.with_runtime_cleanup(Arc::clone(&live_sessions), Arc::clone(&close_terminal)),
|
||||||
));
|
);
|
||||||
// LaunchAgent shares the SAME pty_port and terminal_sessions as the terminal
|
// LaunchAgent shares the SAME pty_port and terminal_sessions as the terminal
|
||||||
// use cases — indispensable for transport bridges to work correctly.
|
// use cases — indispensable for transport bridges to work correctly.
|
||||||
//
|
//
|
||||||
@ -2661,10 +2665,6 @@ impl BackendCore {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
let live_sessions = Arc::new(LiveSessions::new(
|
|
||||||
Arc::clone(&terminal_sessions),
|
|
||||||
Arc::clone(&structured_sessions),
|
|
||||||
));
|
|
||||||
let delete_model_artifact = Arc::new(DeleteModelArtifact::new(
|
let delete_model_artifact = Arc::new(DeleteModelArtifact::new(
|
||||||
Arc::clone(&model_server_registry_port),
|
Arc::clone(&model_server_registry_port),
|
||||||
Arc::clone(&model_server_probe_port),
|
Arc::clone(&model_server_probe_port),
|
||||||
|
|||||||
@ -186,6 +186,58 @@ mod tests {
|
|||||||
assert!(matches!(err, AgentSessionError::Start(_)), "vu: {err:?}");
|
assert!(matches!(err, AgentSessionError::Start(_)), "vu: {err:?}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn run_turn_nonzero_without_stdout_surfaces_stderr() {
|
||||||
|
use std::io::Write as _;
|
||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
|
||||||
|
static C: AtomicU64 = AtomicU64::new(0);
|
||||||
|
let mut path = std::env::temp_dir();
|
||||||
|
path.push(format!(
|
||||||
|
"idea-empty-failing-cli-{}-{}",
|
||||||
|
std::process::id(),
|
||||||
|
C.fetch_add(1, Ordering::Relaxed)
|
||||||
|
));
|
||||||
|
{
|
||||||
|
let mut file = std::fs::File::create(&path).expect("write fake cli");
|
||||||
|
file.write_all(
|
||||||
|
b"#!/bin/sh\nprintf '%s\\n' '401 Unauthorized: Missing bearer' >&2\nexit 1\n",
|
||||||
|
)
|
||||||
|
.expect("write fake cli");
|
||||||
|
file.sync_all().expect("sync fake cli");
|
||||||
|
}
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
|
let mut perms = std::fs::metadata(&path).unwrap().permissions();
|
||||||
|
perms.set_mode(0o755);
|
||||||
|
std::fs::set_permissions(&path, perms).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
let spec = super::process::SpawnLine {
|
||||||
|
command: path.to_string_lossy().into_owned(),
|
||||||
|
args: Vec::new(),
|
||||||
|
cwd: "/".to_owned(),
|
||||||
|
env: Vec::new(),
|
||||||
|
stdin: None,
|
||||||
|
sandbox: None,
|
||||||
|
};
|
||||||
|
|
||||||
|
let err = run_turn(&spec, None, None, None)
|
||||||
|
.await
|
||||||
|
.expect_err("empty nonzero process must be a session error");
|
||||||
|
match err {
|
||||||
|
AgentSessionError::Io(message) => {
|
||||||
|
assert!(
|
||||||
|
message.contains("401 Unauthorized: Missing bearer"),
|
||||||
|
"stderr must be preserved, got: {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
other => panic!("expected Io, got {other:?}"),
|
||||||
|
}
|
||||||
|
let _ = std::fs::remove_file(&path);
|
||||||
|
}
|
||||||
|
|
||||||
// -- parse_event Claude (format RÉEL vérifié 2026-06-09) --------------
|
// -- parse_event Claude (format RÉEL vérifié 2026-06-09) --------------
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@ -32,7 +32,7 @@ use std::sync::mpsc::Sender;
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
|
||||||
use tokio::process::Command;
|
use tokio::process::Command;
|
||||||
|
|
||||||
use domain::ports::AgentSessionError;
|
use domain::ports::AgentSessionError;
|
||||||
@ -219,7 +219,7 @@ fn drain_sandboxed(
|
|||||||
killer_tx: tokio::sync::oneshot::Sender<Arc<std::sync::Mutex<std::process::Child>>>,
|
killer_tx: tokio::sync::oneshot::Sender<Arc<std::sync::Mutex<std::process::Child>>>,
|
||||||
line_tap: Option<Sender<String>>,
|
line_tap: Option<Sender<String>>,
|
||||||
) -> Result<Vec<String>, AgentSessionError> {
|
) -> Result<Vec<String>, AgentSessionError> {
|
||||||
use std::io::{BufRead, BufReader as StdBufReader, Write as _};
|
use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _};
|
||||||
use std::process::{Command as StdCommand, Stdio};
|
use std::process::{Command as StdCommand, Stdio};
|
||||||
use std::sync::Mutex as StdMutex;
|
use std::sync::Mutex as StdMutex;
|
||||||
|
|
||||||
@ -266,6 +266,17 @@ fn drain_sandboxed(
|
|||||||
.stdout
|
.stdout
|
||||||
.take()
|
.take()
|
||||||
.ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?;
|
.ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?;
|
||||||
|
let stderr = child
|
||||||
|
.stderr
|
||||||
|
.take()
|
||||||
|
.ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?;
|
||||||
|
let stderr_handle = std::thread::spawn(move || {
|
||||||
|
let mut bytes = Vec::new();
|
||||||
|
StdBufReader::new(stderr)
|
||||||
|
.read_to_end(&mut bytes)
|
||||||
|
.map(|_| bytes)
|
||||||
|
.map_err(|e| AgentSessionError::Io(e.to_string()))
|
||||||
|
});
|
||||||
let child = Arc::new(StdMutex::new(child));
|
let child = Arc::new(StdMutex::new(child));
|
||||||
// Donne au côté async de quoi tuer l'enfant en cas de timeout.
|
// Donne au côté async de quoi tuer l'enfant en cas de timeout.
|
||||||
let _ = killer_tx.send(Arc::clone(&child));
|
let _ = killer_tx.send(Arc::clone(&child));
|
||||||
@ -285,8 +296,20 @@ fn drain_sandboxed(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 6. Reap (pas de zombie). Lock tenu brièvement.
|
// 6. Reap (pas de zombie). Lock tenu brièvement.
|
||||||
if let Ok(mut c) = child.lock() {
|
let status = child
|
||||||
let _ = c.wait();
|
.lock()
|
||||||
|
.map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))?
|
||||||
|
.wait()
|
||||||
|
.map_err(|e| AgentSessionError::Io(e.to_string()))?;
|
||||||
|
let stderr_bytes = stderr_handle
|
||||||
|
.join()
|
||||||
|
.map_err(|_| AgentSessionError::Io("stderr thread terminé sans résultat".to_owned()))??;
|
||||||
|
if !status.success() && collected.is_empty() {
|
||||||
|
return Err(AgentSessionError::Io(process_exit_message(
|
||||||
|
&command,
|
||||||
|
status.code(),
|
||||||
|
&stderr_bytes,
|
||||||
|
)));
|
||||||
}
|
}
|
||||||
Ok(collected)
|
Ok(collected)
|
||||||
}
|
}
|
||||||
@ -333,6 +356,14 @@ async fn drain(
|
|||||||
.stdout
|
.stdout
|
||||||
.take()
|
.take()
|
||||||
.ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?;
|
.ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?;
|
||||||
|
let mut stderr = child
|
||||||
|
.stderr
|
||||||
|
.take()
|
||||||
|
.ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?;
|
||||||
|
let stderr_task = tokio::spawn(async move {
|
||||||
|
let mut bytes = Vec::new();
|
||||||
|
stderr.read_to_end(&mut bytes).await.map(|_| bytes)
|
||||||
|
});
|
||||||
let mut lines = BufReader::new(stdout).lines();
|
let mut lines = BufReader::new(stdout).lines();
|
||||||
|
|
||||||
let mut collected = Vec::new();
|
let mut collected = Vec::new();
|
||||||
@ -350,9 +381,21 @@ async fn drain(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Réifie le statut du process pour ne pas laisser de zombie ; un statut non nul
|
// Réifie le statut du process pour ne pas laisser de zombie ; un statut non nul
|
||||||
// n'est pas en soi une erreur de session (la CLI peut signaler une erreur
|
// n'est pas en soi une erreur de session si la CLI a imprimé une sortie
|
||||||
// *dans* sa sortie structurée, captée par le parser).
|
// structurée, captée ensuite par le parser. Sans stdout, il n'y a rien à parser :
|
||||||
let _ = child.wait().await.map_err(map_io)?;
|
// on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final.
|
||||||
|
let status = child.wait().await.map_err(map_io)?;
|
||||||
|
let stderr_bytes = stderr_task
|
||||||
|
.await
|
||||||
|
.map_err(|e| AgentSessionError::Io(format!("lecture stderr: {e}")))?
|
||||||
|
.map_err(map_io)?;
|
||||||
|
if !status.success() && collected.is_empty() {
|
||||||
|
return Err(AgentSessionError::Io(process_exit_message(
|
||||||
|
&spec.command,
|
||||||
|
status.code(),
|
||||||
|
&stderr_bytes,
|
||||||
|
)));
|
||||||
|
}
|
||||||
Ok(collected)
|
Ok(collected)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -360,3 +403,14 @@ async fn drain(
|
|||||||
fn map_io(e: io::Error) -> AgentSessionError {
|
fn map_io(e: io::Error) -> AgentSessionError {
|
||||||
AgentSessionError::Io(e.to_string())
|
AgentSessionError::Io(e.to_string())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn process_exit_message(command: &str, code: Option<i32>, stderr: &[u8]) -> String {
|
||||||
|
let status = code.map_or_else(|| "signal".to_owned(), |code| format!("code {code}"));
|
||||||
|
let stderr = String::from_utf8_lossy(stderr);
|
||||||
|
let stderr = stderr.trim();
|
||||||
|
if stderr.is_empty() {
|
||||||
|
format!("{command} exited with {status} without structured output")
|
||||||
|
} else {
|
||||||
|
format!("{command} exited with {status}: {stderr}")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user