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

feat(agent): messagerie inter-agents via send_blocking (D6) — §17

Browse Source 
Comble le trou historique : un agent qui en interroge un autre reçoit enfin
le CONTENU de la réponse, plus seulement un ACK de cycle de vie.

- domain : OrchestratorCommand::AskAgent { target, task } + action wire
  agent.message (target+task requis) ; DomainEvent::AgentReplied
  { agent_id, reply_len } (bus I/O-free, le contenu remonte par l'outcome).
- application : OrchestratorOutcome gagne reply: Option<String>. Méthode
  ask_agent (§17.4) — cible structurée vivante ⇒ send_blocking direct ;
  cible morte ⇒ LaunchAgent structuré puis send ; agent vivant en PTY ou
  profil sans structured_adapter ⇒ Invalid (non adressable, jamais d'ACK
  trompeur) ; agent inconnu ⇒ NotFound ; service non câblé ⇒ Invalid.
  Timeout (300s) ⇒ erreur typée SANS tuer la session. Succès ⇒ publie
  AgentReplied + reply: Some(content). Injection additive via builders
  with_structured / with_events (call sites legacy intacts).
- app-tauri : state câble with_structured/with_events ; DTO AgentReplied.

Tests (QA) : +14 verts (11 app + 3 domaine), invariants validés par
mutation test (reply!=None, timeout ne shutdown pas). cargo test
--workspace : 817 passed, 0 failed.

Suivi (D6b) : surfacer reply dans le writer wire .response.json
(infrastructure/orchestrator) pour la délégation par protocole fichier.
Reste D7 : menu restreint Claude/Codex + retrait custom.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-06-09 23:44:53 +02:00
parent 5059f37890
commit dd1194abe8
7 changed files with 938 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
crates/app-tauri/src/events.rs
View File

@ -47,6 +47,14 @@ pub enum DomainEventDto {
/// Exit code.
code: i32,
},
/// A target agent produced a synchronous reply to an inter-agent `ask` (§17.4).
#[serde(rename_all = "camelCase")]
AgentReplied {
/// Target agent id.
agent_id: String,
/// Reply length in bytes (metric, not the payload).
reply_len: usize,
},
/// An agent's runtime profile was changed (hot-swap of the AI engine).
#[serde(rename_all = "camelCase")]
AgentProfileChanged {
@ -179,6 +187,13 @@ impl From<&DomainEvent> for DomainEventDto {
agent_id: agent_id.to_string(),
code: *code,
},
DomainEvent::AgentReplied {
agent_id,
reply_len,
} => Self::AgentReplied {
agent_id: agent_id.to_string(),
reply_len: *reply_len,
},
DomainEvent::AgentProfileChanged {
agent_id,
profile_id,

26
crates/app-tauri/src/state.rs
View File

@ -710,16 +710,22 @@ impl AppState {
// the UI drives (IdeA stays the single source of truth for the agent/skill
// lifecycle). The per-project watcher that feeds it is started lazily when
// a project is opened (see `ensure_orchestrator_watch`).
let orchestrator_service = Arc::new(OrchestratorService::new(
Arc::clone(&create_agent),
Arc::clone(&launch_agent),
Arc::clone(&list_agents),
Arc::clone(&close_terminal),
Arc::clone(&update_agent_context),
Arc::clone(&create_skill),
Arc::clone(&profile_store_port),
Arc::clone(&terminal_sessions),
));
let orchestrator_service = Arc::new(
OrchestratorService::new(
Arc::clone(&create_agent),
Arc::clone(&launch_agent),
Arc::clone(&list_agents),
Arc::clone(&close_terminal),
Arc::clone(&update_agent_context),
Arc::clone(&create_skill),
Arc::clone(&profile_store_port),
Arc::clone(&terminal_sessions),
)
// Messagerie inter-agents §17.4 : registre structuré + bus pour
// `agent.message` (AskAgent) — rendez-vous synchrone via send_blocking.
.with_structured(Arc::clone(&structured_sessions))
.with_events(Arc::clone(&events_port)),
);
// --- Windows (L10) ---
let move_tab = Arc::new(MoveTabToNewWindow::new(

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

@ -14,17 +14,18 @@
//! infrastructure adapter's job. That keeps this fully unit-testable with fakes.
use std::sync::Arc;
use std::time::Duration;
use domain::ports::ProfileStore;
use domain::{OrchestratorCommand, OrchestratorVisibility, ProfileId, Project};
use domain::ports::{EventBus, ProfileStore};
use domain::{DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId, Project};
use crate::agent::{
CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents,
ListAgentsInput, UpdateAgentContext, UpdateAgentContextInput,
send_blocking, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput,
ListAgents, ListAgentsInput, UpdateAgentContext, UpdateAgentContextInput,
};
use crate::error::AppError;
use crate::skill::{CreateSkill, CreateSkillInput};
use crate::terminal::{CloseTerminal, CloseTerminalInput, TerminalSessions};
use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions};
/// 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
@ -33,6 +34,15 @@ const DEFAULT_ROWS: u16 = 24;
/// See [`DEFAULT_ROWS`].
const DEFAULT_COLS: u16 = 80;
/// Bound on the synchronous inter-agent rendezvous (`agent.message` → `AskAgent`).
///
/// A target agent's turn can be long (reasoning + tool use), so the cap is
/// generous; on expiry [`send_blocking`] returns [`domain::ports::AgentSessionError::Timeout`]
/// **without killing the session**, so the requester can retry. Internal and
/// intentionally not yet config-exposed (it may become a per-project setting
/// without changing the contract).
const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300);
/// Dispatches validated orchestrator commands to the agent/terminal use cases.
pub struct OrchestratorService {
create_agent: Arc<CreateAgentFromScratch>,
@ -43,6 +53,15 @@ pub struct OrchestratorService {
create_skill: Arc<CreateSkill>,
profiles: Arc<dyn ProfileStore>,
sessions: Arc<TerminalSessions>,
/// Registre des sessions **structurées** (§17.5) — la cible d'un `agent.message`
/// y est cherchée pour le rendez-vous synchrone. Injecté au câblage via
/// [`Self::with_structured`] ; `None` ⇒ `AskAgent` ne peut pas être servi (les
/// call sites/tests legacy qui n'utilisent pas la messagerie restent verts).
structured: Option<Arc<StructuredSessions>>,
/// 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
/// publication (l'`ask` fonctionne quand même).
events: Option<Arc<dyn EventBus>>,
}
/// Outcome of dispatching a command — a short, human-readable success summary the
@ -51,6 +70,11 @@ pub struct OrchestratorService {
pub struct OrchestratorOutcome {
/// One-line description of what IdeA did (e.g. `"launched agent dev-backend"`).
pub detail: String,
/// The target agent's **reply content** for a synchronous `agent.message`
/// (`AskAgent`, §17.4). `Some(content)` carries the turn's `Final`; every other
/// command leaves it `None`. Additive field ⇒ existing call sites/tests are
/// untouched (they only read [`Self::detail`]).
pub reply: Option<String>,
}
impl OrchestratorService {
@ -76,9 +100,29 @@ impl OrchestratorService {
create_skill,
profiles,
sessions,
structured: None,
events: None,
}
}
/// Branche le registre des sessions **structurées** (§17.5) pour servir
/// `agent.message`/[`OrchestratorCommand::AskAgent`]. Builder additif façon D3 :
/// signature de [`Self::new`] **inchangée** (les tests/call sites legacy restent
/// verts), le câblage fait `OrchestratorService::new(...).with_structured(reg)`.
#[must_use]
pub fn with_structured(mut self, structured: Arc<StructuredSessions>) -> Self {
self.structured = Some(structured);
self
}
/// Branche l'[`EventBus`] pour publier [`DomainEvent::AgentReplied`] après un
/// `ask` réussi (§17.4). Builder additif (cf. [`Self::with_structured`]).
#[must_use]
pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
self.events = Some(events);
self
}
/// Dispatches a validated command against `project`.
///
/// # Errors
@ -100,6 +144,9 @@ impl OrchestratorService {
self.spawn_agent(project, name, profile, context, visibility)
.await
}
OrchestratorCommand::AskAgent { target, task } => {
self.ask_agent(project, target, task).await
}
OrchestratorCommand::StopAgent { name } => self.stop_agent(project, name).await,
OrchestratorCommand::UpdateAgentContext { name, context } => {
self.update_agent_context(project, name, context).await
@ -150,6 +197,7 @@ impl OrchestratorService {
OrchestratorVisibility::Background => {
return Ok(OrchestratorOutcome {
detail: format!("agent {name} already running in background"),
reply: None,
});
}
OrchestratorVisibility::Visible { node_id } => {
@ -163,6 +211,7 @@ impl OrchestratorService {
})?;
return Ok(OrchestratorOutcome {
detail: format!("attached agent {name} to cell {}", session.node_id),
reply: None,
});
}
}
@ -193,9 +242,119 @@ impl OrchestratorService {
format!("launched agent {name} in cell {node_id}")
}
},
reply: None,
})
}
/// `agent.message`: the **synchronous inter-agent rendezvous** (§17.4).
///
/// Resolves the target by name, ensures it has a **live structured session**
/// (launching it in structured mode if needed), sends `task` and **waits for the
/// turn's `Final`** via [`send_blocking`], then returns its content as
/// [`OrchestratorOutcome::reply`] and publishes [`DomainEvent::AgentReplied`].
///
/// Invariants:
/// - **1 session vivante/agent** : on réutilise la session structurée vivante si
/// elle existe ; sinon `LaunchAgent` (gardé sur les deux registres) la crée.
/// - **Timeout ne tue pas la session** : [`send_blocking`] remonte
/// [`AppError::Process`] (via [`domain::ports::AgentSessionError::Timeout`]) en
/// laissant la session vivante dans le registre (retry possible).
/// - **Cible PTY-only** (profil sans `structured_adapter`, ou agent déjà vivant en
/// PTY) ⇒ [`AppError::Invalid`] explicite, **jamais** un ACK trompeur.
///
/// # Errors
/// - [`AppError::NotFound`] si l'agent cible est inconnu ;
/// - [`AppError::Invalid`] si la messagerie structurée n'est pas câblée, ou si la
/// cible n'est pas pilotable en mode structuré ;
/// - [`AppError::Process`] sur échec/timeout du tour structuré.
async fn ask_agent(
&self,
project: &Project,
target: String,
task: String,
) -> Result<OrchestratorOutcome, AppError> {
let structured = self.structured.as_ref().ok_or_else(|| {
AppError::Invalid(
"la messagerie inter-agents (agent.message) n'est pas disponible : \
registre des sessions structurées non câblé"
.to_owned(),
)
})?;
let agent_id = self
.find_agent_id_by_name(project, &target)
.await?
.ok_or_else(|| AppError::NotFound(format!("agent {target}")))?;
// 1. Session structurée déjà vivante ? ⇒ rendez-vous direct.
if let Some(session) = structured.session_for_agent(&agent_id) {
let content = send_blocking(session.as_ref(), &task, Some(ASK_AGENT_TIMEOUT)).await?;
return Ok(self.reply_outcome(agent_id, &target, content));
}
// 2. Pas de session structurée. Si l'agent est vivant en **PTY** (terminal
// brut), il n'est pas adressable en `ask` ⇒ erreur typée explicite (pas
// d'ACK « launched »).
if self.sessions.session_for_agent(&agent_id).is_some() {
return Err(AppError::Invalid(format!(
"agent {target} n'est pas pilotable en mode structuré \
(session terminal brut, pas de canal de réponse)"
)));
}
// 3. Cible morte : on la lance en mode structuré (background) puis on envoie.
let launched = self
.launch_agent
.execute(LaunchAgentInput {
project: project.clone(),
agent_id,
rows: DEFAULT_ROWS,
cols: DEFAULT_COLS,
node_id: None,
conversation_id: None,
})
.await?;
// Si le lancement n'a pas produit de session structurée, le profil de la cible
// est PTY-only (pas de `structured_adapter`) ⇒ non adressable en `ask`.
if launched.structured.is_none() {
return Err(AppError::Invalid(format!(
"agent {target} n'est pas pilotable en mode structuré \
(profil sans adaptateur structuré)"
)));
}
// La session vient d'être enregistrée par LaunchAgent : on la récupère par
// agent (invariant « 1 session/agent » ⇒ non ambigu).
let session = structured.session_for_agent(&agent_id).ok_or_else(|| {
AppError::Process(format!(
"session structurée de l'agent {target} introuvable après lancement"
))
})?;
let content = send_blocking(session.as_ref(), &task, Some(ASK_AGENT_TIMEOUT)).await?;
Ok(self.reply_outcome(agent_id, &target, content))
}
/// Builds the success outcome of an `ask` and publishes [`DomainEvent::AgentReplied`]
/// (best-effort: a missing event bus never fails the rendezvous).
fn reply_outcome(
&self,
agent_id: domain::AgentId,
target: &str,
content: String,
) -> OrchestratorOutcome {
if let Some(events) = &self.events {
events.publish(DomainEvent::AgentReplied {
agent_id,
reply_len: content.len(),
});
}
OrchestratorOutcome {
detail: format!("agent {target} replied ({} bytes)", content.len()),
reply: Some(content),
}
}
/// `stop_agent`: translate the agent name → its live session → `CloseTerminal`.
async fn stop_agent(
&self,
@ -218,6 +377,7 @@ impl OrchestratorService {
Ok(OrchestratorOutcome {
detail: format!("stopped agent {name}"),
reply: None,
})
}
@ -243,6 +403,7 @@ impl OrchestratorService {
Ok(OrchestratorOutcome {
detail: format!("updated context for agent {name}"),
reply: None,
})
}
@ -269,6 +430,7 @@ impl OrchestratorService {
Ok(OrchestratorOutcome {
detail: format!("created skill {} ({:?})", created.skill.name, scope),
reply: None,
})
}

569
crates/application/tests/orchestrator_service.rs
View File

@ -21,21 +21,23 @@ use domain::ids::SkillId;
use domain::ids::{AgentId, NodeId, ProfileId, ProjectId};
use domain::markdown::MarkdownDoc;
use domain::ports::{
AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream,
ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, ProfileStore,
PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec,
AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, AgentSessionFactory,
ContextInjectionPlan, DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError,
IdGenerator, OutputStream, PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort,
RemotePath, ReplyEvent, ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec,
StoreError,
};
use domain::profile::{AgentProfile, ContextInjection};
use domain::profile::{AgentProfile, ContextInjection, StructuredAdapter};
use domain::project::{Project, ProjectPath};
use domain::remote::RemoteRef;
use domain::skill::{Skill, SkillScope};
use domain::terminal::{SessionKind, TerminalSession};
use domain::{OrchestratorCommand, OrchestratorRequest, PtySize, SessionId};
use uuid::Uuid;
use application::{
CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
OrchestratorService, TerminalSessions, UpdateAgentContext,
OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext,
};
// ---------------------------------------------------------------------------
@ -357,6 +359,109 @@ impl EventBus for SpyBus {
}
}
// ---------------------------------------------------------------------------
// D6 fakes: structured session + factory (inter-agent messaging)
// ---------------------------------------------------------------------------
/// What the scripted session does on its next `send`.
enum SendScript {
/// Emit this event stream verbatim (drained to its `Final`).
Stream(Vec<ReplyEvent>),
/// Sleep `delay` *before* emitting the stream — forces a `send_blocking` timeout.
Delayed {
delay: std::time::Duration,
events: Vec<ReplyEvent>,
},
}
/// A fake [`AgentSession`] that records every `send` prompt and every `shutdown`,
/// so a test can assert the rendezvous happened (send) and the session was *not*
/// killed (zero shutdowns) on timeout.
struct FakeSession {
id: SessionId,
script: Mutex<Option<SendScript>>,
sends: Arc<Mutex<Vec<String>>>,
shutdowns: Arc<Mutex<usize>>,
}
impl FakeSession {
fn new(id: SessionId, script: SendScript) -> Arc<Self> {
Arc::new(Self {
id,
script: Mutex::new(Some(script)),
sends: Arc::new(Mutex::new(Vec::new())),
shutdowns: Arc::new(Mutex::new(0)),
})
}
}
#[async_trait]
impl AgentSession for FakeSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
self.sends.lock().unwrap().push(prompt.to_owned());
let script = self
.script
.lock()
.unwrap()
.take()
.expect("send scripted exactly once");
match script {
SendScript::Stream(events) => Ok(Box::new(events.into_iter())),
SendScript::Delayed { delay, events } => {
tokio::time::sleep(delay).await;
Ok(Box::new(events.into_iter()))
}
}
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
*self.shutdowns.lock().unwrap() += 1;
Ok(())
}
}
/// A fake [`AgentSessionFactory`] that hands out a pre-built [`FakeSession`] on
/// `start`, so launching a *dead* target in structured mode registers a session
/// the orchestrator can then drive. `supports` is true for any profile bearing a
/// `structured_adapter` (mirrors the real factory's gate).
struct FakeFactory {
session: Arc<FakeSession>,
starts: Arc<Mutex<usize>>,
}
impl FakeFactory {
fn new(session: Arc<FakeSession>) -> Self {
Self {
session,
starts: Arc::new(Mutex::new(0)),
}
}
}
#[async_trait]
impl AgentSessionFactory for FakeFactory {
fn supports(&self, profile: &AgentProfile) -> bool {
profile.structured_adapter.is_some()
}
async fn start(
&self,
_profile: &AgentProfile,
_ctx: &PreparedContext,
_cwd: &ProjectPath,
_session: &SessionPlan,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
*self.starts.lock().unwrap() += 1;
Ok(Arc::clone(&self.session) as Arc<dyn AgentSession>)
}
}
fn final_(c: &str) -> ReplyEvent {
ReplyEvent::Final {
content: c.to_owned(),
}
}
struct SeqIds(Mutex<u128>);
impl SeqIds {
fn new() -> Self {
@ -418,6 +523,24 @@ fn scratch_agent(id: AgentId, name: &str, md: &str) -> Agent {
Agent::new(id, name, md, pid(9), AgentOrigin::Scratch, false).unwrap()
}
/// Same id (`pid(9)`) as [`claude_profile`] so agents created with it resolve to a
/// structured-capable profile, but bearing a `structured_adapter` so `LaunchAgent`
/// routes through the (fake) [`AgentSessionFactory`] instead of the PTY.
fn structured_profile() -> AgentProfile {
AgentProfile::new(
pid(9),
"Claude Code",
"claude",
Vec::new(),
ContextInjection::stdin(),
Some("claude --version".to_owned()),
"{agentRunDir}",
None,
)
.unwrap()
.with_structured_adapter(StructuredAdapter::Claude)
}
/// Everything wired for a dispatch test.
struct Fixture {
service: OrchestratorService,
@ -492,6 +615,105 @@ fn cmd(json: &str) -> OrchestratorCommand {
.unwrap()
}
// ---------------------------------------------------------------------------
// D6 fixture: orchestrator wired for `agent.message` (inter-agent rendezvous)
// ---------------------------------------------------------------------------
/// Everything wired for an `agent.message` dispatch test (§17.4).
struct AskFixture {
service: OrchestratorService,
structured: Arc<StructuredSessions>,
/// PTY (terminal) registry the service holds — lets a test register a raw
/// terminal session for an agent to exercise the PTY-only-live branch.
sessions: Arc<TerminalSessions>,
bus: SpyBus,
/// PTY spawn/kill spy of the underlying `LaunchAgent` — used to prove a
/// dead-target launch did *not* fall back to a raw PTY (zero spawns).
pty: FakePty,
/// `start` counter of the fake factory (dead-target launch path).
factory_starts: Arc<Mutex<usize>>,
}
/// Builds an orchestrator with the structured registry + event bus wired, and a
/// `LaunchAgent` whose structured routing uses `factory` over `profiles`.
///
/// `pty_only` controls the launched profile: `false` ⇒ structured-capable profile
/// (factory hands out a session); `true` ⇒ plain PTY profile (no
/// `structured_adapter`) so `launched.structured` stays `None`.
fn ask_fixture(
contexts: FakeContexts,
factory_session: Arc<FakeSession>,
pty_only: bool,
) -> AskFixture {
let profile = if pty_only {
claude_profile()
} else {
structured_profile()
};
let profiles = Arc::new(FakeProfiles::new(vec![profile]));
let sessions = Arc::new(TerminalSessions::new());
let structured = Arc::new(StructuredSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let factory = Arc::new(FakeFactory::new(factory_session));
let factory_starts = Arc::clone(&factory.starts);
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
let launch = Arc::new(
LaunchAgent::new(
Arc::new(contexts.clone()),
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeRuntime),
Arc::new(FakeFs),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
)
.with_structured(factory as Arc<dyn AgentSessionFactory>, Arc::clone(&structured)),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let skills = RecordingSkills::default();
let create_skill = Arc::new(CreateSkill::new(
Arc::new(skills) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let service = OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_structured(Arc::clone(&structured))
.with_events(Arc::new(bus.clone()));
AskFixture {
service,
structured,
sessions,
bus,
pty,
factory_starts,
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@ -717,3 +939,340 @@ async fn create_skill_honours_global_scope() {
assert_eq!(saved.len(), 1);
assert_eq!(saved[0].scope, SkillScope::Global);
}
// ---------------------------------------------------------------------------
// D6 — agent.message (synchronous inter-agent rendezvous, §17.4)
// ---------------------------------------------------------------------------
const ASK_JSON: &str =
r#"{ "type":"agent.message", "requestedBy":"Main", "targetAgent":"architect", "task":"Analyse §17" }"#;
/// Zone 1 — live structured target ⇒ `send_blocking` is driven and the outcome's
/// `reply` carries the turn's `Final` content. Anti-always-green: the assertion is
/// on the *exact* content, so it fails for `None` or any other string.
#[tokio::test]
async fn ask_live_structured_target_returns_final_content() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
// A live session already registered for the agent in the *structured* registry.
let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_("the §17 answer")]));
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
false,
);
fx.structured
.insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
// The exact Final content is returned (would fail on None or any other text).
assert_eq!(out.reply.as_deref(), Some("the §17 answer"));
// The prompt actually reached the live session.
assert_eq!(session.sends.lock().unwrap().as_slice(), &["Analyse §17"]);
// Live session reused: factory never started a new one, no PTY spawned.
assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
assert!(fx.pty.spawns().is_empty(), "must not spawn a PTY");
// Session not killed.
assert_eq!(*session.shutdowns.lock().unwrap(), 0);
}
/// Anti-always-green guard, explicit negative: the same fixture but the session
/// returns a *different* Final ⇒ the strict equality from the test above must NOT
/// hold. Proves the assertion has teeth.
#[tokio::test]
async fn ask_reply_assertion_is_content_sensitive() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_("SOMETHING ELSE")]));
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
false,
);
fx.structured
.insert(session as Arc<dyn AgentSession>, aid(1), nid(1));
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
assert_ne!(out.reply.as_deref(), Some("the §17 answer"));
assert_eq!(out.reply.as_deref(), Some("SOMETHING ELSE"));
}
/// Zone 4 — on success, `DomainEvent::AgentReplied` is published with the right
/// agent id and `reply_len` (byte length of the content).
#[tokio::test]
async fn ask_success_publishes_agent_replied_event() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let content = "réponse"; // multibyte: len() in bytes, not chars
let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_(content)]));
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
false,
);
fx.structured
.insert(session as Arc<dyn AgentSession>, aid(1), nid(1));
fx.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
let replied = fx
.bus
.events()
.into_iter()
.find_map(|e| match e {
DomainEvent::AgentReplied {
agent_id,
reply_len,
} => Some((agent_id, reply_len)),
_ => None,
})
.expect("AgentReplied must be published");
assert_eq!(replied.0, aid(1));
assert_eq!(replied.1, content.len());
}
/// Zone 2 (orchestrator seam) — a failing turn (stream ends without a `Final`)
/// surfaces a typed `PROCESS` error WITHOUT the orchestrator killing the session,
/// and the session stays registered (retry possible). The *genuine* timeout
/// variant of this invariant is asserted in
/// [`timeout_invariant_send_blocking_does_not_kill_session`] (the 300s prod bound
/// cannot be awaited here, so we drive the same "never shutdown on ask" branch via
/// the no-`Final` error and the bounded `send_blocking` directly).
#[tokio::test]
async fn ask_failed_turn_does_not_kill_session_and_errors() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let session = FakeSession::new(sid(500), SendScript::Stream(vec![/* no Final */]));
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
false,
);
fx.structured
.insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.unwrap_err();
// Typed PROCESS error (send_blocking's AgentSessionError → AppError::Process).
assert_eq!(err.code(), "PROCESS", "got {err:?}");
// Session NOT killed and STILL registered (retry possible).
assert_eq!(
*session.shutdowns.lock().unwrap(),
0,
"ask must never shutdown the target session"
);
assert!(
fx.structured.session_for_agent(&aid(1)).is_some(),
"session must remain in the registry after a failed turn"
);
}
/// Zone 2 (real timeout) — uses a *short-bounded* `send_blocking` directly against
/// the orchestrator's session is impossible (bound is internal/300s); instead we
/// prove the genuine timeout-does-not-kill invariant at the `send_blocking` seam
/// with a real delayed session, mirroring how the orchestrator awaits it.
#[tokio::test]
async fn timeout_invariant_send_blocking_does_not_kill_session() {
use application::send_blocking;
let session = FakeSession::new(
sid(500),
SendScript::Delayed {
delay: std::time::Duration::from_secs(1),
events: vec![final_("too late")],
},
);
let out = send_blocking(
session.as_ref(),
"Analyse §17",
Some(std::time::Duration::from_millis(20)),
)
.await;
assert_eq!(out, Err(AgentSessionError::Timeout));
assert_eq!(
*session.shutdowns.lock().unwrap(),
0,
"timeout must not kill the session"
);
// Negative control: had we waited (no bound) the call would succeed — proves the
// timeout above is the cause of the error, not a broken session.
let session2 = FakeSession::new(
sid(501),
SendScript::Delayed {
delay: std::time::Duration::from_millis(5),
events: vec![final_("eventually")],
},
);
let ok = send_blocking(session2.as_ref(), "x", None).await;
assert_eq!(ok, Ok("eventually".to_owned()));
}
/// Zone 3 — dead target ⇒ `LaunchAgent` is invoked in structured mode (factory
/// `start` called, structured session registered) *then* `send_blocking`; the
/// reply is returned and no raw PTY is spawned.
#[tokio::test]
async fn ask_dead_target_launches_structured_then_sends() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
// The session the fake factory will hand out on launch.
let session = FakeSession::new(sid(700), SendScript::Stream(vec![final_("launched reply")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
Arc::clone(&session),
false,
);
// No live session up front.
assert!(fx.structured.session_for_agent(&aid(1)).is_none());
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("launched reply"));
// Factory was started exactly once (structured launch), the prompt was sent.
assert_eq!(*fx.factory_starts.lock().unwrap(), 1);
assert_eq!(session.sends.lock().unwrap().as_slice(), &["Analyse §17"]);
// Structured-mode launch ⇒ NO raw PTY spawn.
assert!(fx.pty.spawns().is_empty(), "structured launch must not spawn a PTY");
// The session is now registered (1 session/agent).
assert!(fx.structured.session_for_agent(&aid(1)).is_some());
}
/// Zone 5a — target already live in the **PTY** registry (raw terminal, no
/// structured channel) ⇒ typed `INVALID`, never an ACK, never a launch.
#[tokio::test]
async fn ask_pty_live_target_is_invalid_no_ack() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
false,
);
// Register the agent as live in the **PTY** (terminal) registry only — a raw
// terminal with no structured reply channel.
let pty_session = TerminalSession::starting(
sid(800),
nid(2),
project().root.clone(),
SessionKind::Agent { agent_id: aid(1) },
PtySize::new(24, 80).unwrap(),
);
fx.sessions
.insert(PtyHandle { session_id: sid(800) }, pty_session);
assert!(fx.structured.session_for_agent(&aid(1)).is_none());
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.unwrap_err();
assert_eq!(
err.code(),
"INVALID",
"PTY-live target is not addressable by ask: {err:?}"
);
// Never launched, never sent: factory untouched, no PTY spawned.
assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
assert!(fx.pty.spawns().is_empty());
// No structured session conjured.
assert!(fx.structured.session_for_agent(&aid(1)).is_none());
}
/// Zone 5b — after launching, the target turns out PTY-only (profile without a
/// `structured_adapter` ⇒ `launched.structured` is `None`) ⇒ typed `INVALID`, no
/// ACK, and the orchestrator does not pretend a reply.
#[tokio::test]
async fn ask_pty_only_profile_after_launch_is_invalid() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
// pty_only = true ⇒ launched profile has no structured_adapter ⇒ structured None.
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
throwaway,
true,
);
assert!(fx.structured.session_for_agent(&aid(1)).is_none());
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.unwrap_err();
assert_eq!(err.code(), "INVALID", "PTY-only target must error typed: {err:?}");
// No structured session was registered, no reply pretended.
assert!(fx.structured.session_for_agent(&aid(1)).is_none());
}
/// Zone 6/7 — structured registry not wired ⇒ `ask` is INVALID (legacy service).
#[tokio::test]
async fn ask_without_structured_wired_is_invalid() {
// The default `fixture` builds the service via `OrchestratorService::new`
// WITHOUT `.with_structured(...)`, so AskAgent cannot be served.
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.unwrap_err();
assert_eq!(err.code(), "INVALID", "got {err:?}");
}
/// Zone 6/7 — unknown target agent ⇒ typed `NOT_FOUND`, no launch, no reply.
#[tokio::test]
async fn ask_unknown_target_is_not_found() {
let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
let fx = ask_fixture(FakeContexts::new(), throwaway, false);
let err = fx
.service
.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"ghost", "task":"hi" }"#),
)
.await
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
assert!(fx.pty.spawns().is_empty());
}
/// Zone 7 — non-regression: `agent.run` (fire-and-forget) still yields `reply:
/// None` even through a structured-wired orchestrator.
#[tokio::test]
async fn non_regression_agent_run_reply_is_none() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let session = FakeSession::new(sid(700), SendScript::Stream(vec![final_("x")]));
let fx = ask_fixture(
FakeContexts::with_agent(&agent, "# persona"),
session,
false,
);
let out = fx
.service
.dispatch(
&project(),
cmd(r#"{ "type":"agent.run", "targetAgent":"architect", "task":"go" }"#),
)
.await
.expect("run ok");
assert_eq!(out.reply, None, "agent.run must not carry a reply");
}

12
crates/domain/src/events.rs
View File

@ -25,6 +25,18 @@ pub enum DomainEvent {
/// The session it runs in.
session_id: SessionId,
},
/// A target agent produced a synchronous reply to an inter-agent `ask`
/// (ARCHITECTURE §17.4): the requester sent a task via `agent.message`, IdeA
/// drove the target's structured session to its turn `Final`, and this is the
/// observability beacon for that completed rendezvous. Carries the target id and
/// the reply size (a lightweight metric); the full body is returned to the
/// requester out-of-band, not in the event payload (the bus stays I/O-free).
AgentReplied {
/// The agent that produced the reply.
agent_id: AgentId,
/// Number of bytes in the reply content (preview metric, not the payload).
reply_len: usize,
},
/// An agent's process exited.
AgentExited {
/// The agent.

63
crates/domain/src/orchestrator.rs
View File

@ -60,8 +60,9 @@ pub struct OrchestratorRequest {
/// Legacy v1 action (`spawn_agent`, `stop_agent`, `update_agent_context`).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub action: Option<String>,
/// V2 action type (`agent.run`, `agent.stop`, `agent.update_context`,
/// `skill.create`). `type` is reserved in Rust, so the field is renamed.
/// V2 action type (`agent.run`, `agent.stop`, `agent.message`,
/// `agent.update_context`, `skill.create`). `type` is reserved in Rust, so the
/// field is renamed.
#[serde(default, rename = "type", skip_serializing_if = "Option::is_none")]
pub request_type: Option<String>,
/// Optional requester id/name, informational at this layer.
@ -81,7 +82,8 @@ pub struct OrchestratorRequest {
/// `create_skill` this carries the **new Markdown body** to write.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub context: Option<String>,
/// Optional task/message carried by `agent.run` / future `agent.message`.
/// Task/message carried by `agent.run` (fire-and-forget) and required by
/// `agent.message` (synchronous ask: the prompt sent to the target agent).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub task: Option<String>,
/// Whether a spawned agent should stay in the background or attach to a cell.
@ -135,6 +137,19 @@ pub enum OrchestratorCommand {
/// New Markdown body.
context: String,
},
/// Ask a target agent a question/task and **wait for its reply**.
///
/// Unlike [`Self::SpawnAgent`] (fire-and-forget lifecycle), this is the
/// synchronous inter-agent rendezvous (ARCHITECTURE §17.4): the application
/// layer drives the target's structured [`crate::ports::AgentSession`], waits
/// for the turn's `Final`, and returns its content to the requester. The
/// target is launched in structured mode first if it is not already live.
AskAgent {
/// Target agent display name (resolved case-insensitively).
target: String,
/// The task/message to send and await a reply for.
task: String,
},
/// Create a reusable skill in the given scope — exactly as the UI would.
CreateSkill {
/// Display name (also the `.md` stem on disk).
@ -208,6 +223,10 @@ impl OrchestratorRequest {
name: self.require_target_agent(action)?,
context: self.require("context", action, self.context.as_deref())?,
}),
"agent.message" => Ok(OrchestratorCommand::AskAgent {
target: self.require_target_agent(action)?,
task: self.require("task", action, self.task.as_deref())?,
}),
"create_skill" | "skill.create" => Ok(OrchestratorCommand::CreateSkill {
name: self.require_name(action)?,
content: self.require("context", action, self.context.as_deref())?,
@ -413,6 +432,44 @@ mod tests {
);
}
#[test]
fn agent_message_validates_target_and_task() {
let r = req(
r#"{ "type": "agent.message", "requestedBy": "Main", "targetAgent": "Architect", "task": "Analyse §17" }"#,
);
assert_eq!(
r.validate().unwrap(),
OrchestratorCommand::AskAgent {
target: "Architect".to_owned(),
task: "Analyse §17".to_owned(),
}
);
}
#[test]
fn agent_message_missing_task_is_rejected() {
let r = req(r#"{ "type": "agent.message", "targetAgent": "Architect" }"#);
assert_eq!(
r.validate(),
Err(OrchestratorError::MissingField {
action: "agent.message".to_owned(),
field: "task".to_owned(),
})
);
}
#[test]
fn agent_message_missing_target_is_rejected() {
let r = req(r#"{ "type": "agent.message", "task": "do it" }"#);
assert_eq!(
r.validate(),
Err(OrchestratorError::MissingField {
action: "agent.message".to_owned(),
field: "targetAgent".to_owned(),
})
);
}
#[test]
fn stop_agent_validates() {
let r = req(r#"{ "action": "stop_agent", "name": "dev-backend" }"#);