From dd1194abe80fbcbc9d51394ddd790c716aeca861 Mon Sep 17 00:00:00 2001 From: Blomios Date: Tue, 9 Jun 2026 23:44:53 +0200 Subject: [PATCH] =?UTF-8?q?feat(agent):=20messagerie=20inter-agents=20via?= =?UTF-8?q?=20send=5Fblocking=20(D6)=20=E2=80=94=20=C2=A717?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. 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 --- .ideai/briefs/d6-messagerie-inter-agents.md | 104 ++++ crates/app-tauri/src/events.rs | 15 + crates/app-tauri/src/state.rs | 26 +- .../application/src/orchestrator/service.rs | 172 +++++- .../application/tests/orchestrator_service.rs | 569 +++++++++++++++++- crates/domain/src/events.rs | 12 + crates/domain/src/orchestrator.rs | 63 +- 7 files changed, 938 insertions(+), 23 deletions(-) create mode 100644 .ideai/briefs/d6-messagerie-inter-agents.md diff --git a/.ideai/briefs/d6-messagerie-inter-agents.md b/.ideai/briefs/d6-messagerie-inter-agents.md new file mode 100644 index 0000000..4ff1a99 --- /dev/null +++ b/.ideai/briefs/d6-messagerie-inter-agents.md @@ -0,0 +1,104 @@ +# Brief Dev — Lot D6 : messagerie inter-agents via `send_blocking` (§17.9) + +> Demandé par **Main** à **DevBackend** (dev) + **QA** (test). Cycle §3 : code → tests → vert. +> Périmètre **backend** (domaine + application). Pas de frontend. + +## 0. Le trou que D6 comble (important) + +Aujourd'hui, « Main demande à Architect » ne **retourne jamais le contenu** de la réponse : +- `OrchestratorCommand` n'a **pas** de variante « demander/attendre une réponse ». Le wire + `agent.run` replie `task` dans `context`, et `context` n'est utilisé que pour un agent **neuf** + (`.md` initial) ⇒ pour un agent **déjà existant**, le `task` est **silencieusement ignoré**. +- La réponse (`*.response.json`) ne porte qu'un **ACK de cycle de vie** + (`detail: "launched agent X"`), jamais la sortie produite par la cible. + +D6 = brancher la **messagerie synchrone** sur le port `AgentSession` : la cible est pilotée en +mode structuré, on **attend son tour** et on **renvoie son contenu**. La primitive existe déjà : +`crate::agent::structured::send_blocking(session, prompt, timeout) -> Result` +(livrée en D1 ; retourne le contenu du `Final`, `Timeout` typé **sans tuer la session**). + +## 1. Périmètre D6 (réf. tableau §17.9 ligne D6) + +### A. Domaine (`crates/domain/src/`) +1. **`OrchestratorCommand::AskAgent { target: String, task: String }`** (`orchestrator.rs`, + enum ~l.111). Nouvelle variante « j'attends une réponse ». +2. **Validation** : nouvelle action wire **`agent.message`** → `AskAgent` dans + `OrchestratorRequest::validate` (~l.175). `targetAgent` (ou `name`) **et** `task` requis + non-vides (sinon `OrchestratorError::MissingField`). Garde le mapping `agent.run` actuel + inchangé (lancement fire-and-forget). Ajoute un cas de test de round-trip JSON `agent.message`. +3. **`DomainEvent::AgentReplied { ... }`** (`events.rs`, à côté de `AgentLaunched`) pour + l'observabilité : au minimum le nom/id de l'agent cible et, si pertinent, la taille/preview + de la réponse (reste pur, pas de payload lourd imposé — calque le style des variantes voisines). + +### B. Application (`crates/application/src/orchestrator/service.rs`) +4. **`OrchestratorOutcome` gagne le contenu** : ajoute `reply: Option` (l'ACK actuel + `detail` reste). Les commandes existantes mettent `reply: None` ; `AskAgent` met + `reply: Some(contenu)`. (Champ additif ⇒ aucune régression des call sites/tests existants.) +5. **`dispatch` route `AskAgent`** : + - résous l'agent cible par nom (`find_agent_id_by_name`) → sinon `AppError::NotFound` typé. + - cherche sa **session structurée vivante** dans le registre **`StructuredSessions`** (PAS + `TerminalSessions`). Si vivante ⇒ `send_blocking(session, &task, timeout)`. + - si **pas vivante** ⇒ lance l'agent en mode structuré (via `LaunchAgent`, background) puis + `send_blocking`. Respecte l'invariant **1 session/agent**. + - si la cible est **PTY-only** (profil sans `structured_adapter` ⇒ pas adressable en `ask`) ⇒ + **erreur typée explicite** (`AppError::Invalid`/`NotFound` avec message clair « agent X n'est + pas pilotable en mode structuré »). C'est acceptable (le menu ne crée plus que des agents + structurés, cf. D7 à venir). + - **timeout** ⇒ remonte une erreur typée, **sans tuer la session** (déjà la sémantique de + `send_blocking`). + - en cas de succès ⇒ **publie `DomainEvent::AgentReplied`** sur l'`EventBus`, et retourne + `OrchestratorOutcome { detail, reply: Some(content) }`. +6. **Injection** : le service a besoin du registre `StructuredSessions` + de l'`EventBus` (et de + quoi piloter `send_blocking`). **Ajoute-les par builder additif** (`with_structured(...)` / + `with_events(...)` façon D3 sur `LaunchAgent`) pour que `OrchestratorService::new` reste + compatible et que les tests/call sites legacy restent verts. Câble au composition root + (`crates/app-tauri/src/state.rs`). + +### C. Nettoyage voie principale +7. **Aucun accès outbox/inbox** dans le chemin `AskAgent` (le rendez-vous est intrinsèque à + `send_blocking`). Les seules occurrences `outbox/inbox` actuelles sont des commentaires de doc + dans `agent/structured.rs` — ne ré-introduis rien. Vérifie qu'aucun `AgentReplyChannel`/outbox + n'est utilisé. + +## 2. Invariants à respecter + +- **1 session vivante par agent** across registres (PTY + structuré). +- **Timeout ne tue jamais la session** (retry possible). +- **Cible PTY non adressable** par `ask` ⇒ erreur typée, jamais un ACK trompeur ni un panic. +- Frontières hexagonales : le domaine reste pur (pas d'I/O dans `orchestrator.rs`/`events.rs`) ; + l'orchestration vit dans l'application ; aucun `new` d'adapter infra dans le service. +- **Zéro régression** : `agent.run`/`spawn_agent`/`stop_agent`/`update_agent_context`/`skill.create` + inchangés ; le watcher d'orchestration et ses tests restent verts. + +## 3. Tests attendus (QA — colonne « Tests attendus » D6 du §17.9) + +Unitaires avec **fakes** (fake `AgentSession`, fake registres, fake EventBus — aucun vrai CLI) : +- cible **vivante** (session structurée enregistrée) ⇒ `send_blocking` appelé, `reply: Some(...)` + porte le contenu du `Final`. +- cible **morte** ⇒ `LaunchAgent` invoqué (structuré) **puis** `send` ; `reply` renvoyé. +- **timeout** ⇒ erreur typée remontée, **session non tuée** (le fake atteste qu'aucun `shutdown` + n'a été appelé), pas de `reply`. +- **`AgentReplied`** publié sur l'EventBus en cas de succès. +- cible **PTY-only** (profil sans adapter / présente seulement dans `TerminalSessions`) ⇒ erreur + typée explicite, **pas** d'ACK « launched ». +- **validation** : `agent.message` sans `task` ⇒ `MissingField` ; round-trip JSON `agent.message`. +- **non-régression** : `agent.run` ne change pas de comportement ; aucun accès outbox. +- garde anti-always-green sur au moins l'invariant timeout-ne-tue-pas OU ask-retourne-le-contenu. + +## 4. Méthode + +DevBackend code → vérifie `cargo build -p domain -p application -p app-tauri`. **N'exécute pas +`cargo test --workspace`** si un build concurrent tient le lock (sinon, lance-le). QA écrit + exécute +les tests, produit un rapport clair si rouge. **Ne pas committer, ne pas push** — Main relit et commit. + +## 5. Références + +- Domaine : `crates/domain/src/orchestrator.rs` (enum `OrchestratorCommand` ~l.111, `validate` + ~l.175), `crates/domain/src/events.rs` (`DomainEvent`). +- Application : `crates/application/src/orchestrator/service.rs` (struct/deps ~l.37, `dispatch` + ~l.88, `OrchestratorOutcome` ~l.50, `spawn_agent` comme modèle de résolution d'agent). +- Primitive : `crates/application/src/agent/structured.rs` (`send_blocking`). +- Registre structuré : `StructuredSessions` (livré D1, jumeau de `TerminalSessions`) — repère-le + et lis son API avant de t'en servir. +- Composition root : `crates/app-tauri/src/state.rs`. +- Spec : `ARCHITECTURE.md` §17.4 (réconciliation §16) et tableau §17.9 ligne **D6**. diff --git a/crates/app-tauri/src/events.rs b/crates/app-tauri/src/events.rs index de32522..ee17c74 100644 --- a/crates/app-tauri/src/events.rs +++ b/crates/app-tauri/src/events.rs @@ -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, diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 2bc546a..57a9e64 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -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( diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index f646d0b..99dc717 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -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, @@ -43,6 +53,15 @@ pub struct OrchestratorService { create_skill: Arc, profiles: Arc, sessions: Arc, + /// 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>, + /// 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>, } /// 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, } 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) -> 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) -> 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 { + 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, }) } diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index 5d6b881..408dec8 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -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), + /// Sleep `delay` *before* emitting the stream — forces a `send_blocking` timeout. + Delayed { + delay: std::time::Duration, + events: Vec, + }, +} + +/// 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>, + sends: Arc>>, + shutdowns: Arc>, +} +impl FakeSession { + fn new(id: SessionId, script: SendScript) -> Arc { + 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 { + None + } + async fn send(&self, prompt: &str) -> Result { + 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, + starts: Arc>, +} +impl FakeFactory { + fn new(session: Arc) -> 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, AgentSessionError> { + *self.starts.lock().unwrap() += 1; + Ok(Arc::clone(&self.session) as Arc) + } +} + +fn final_(c: &str) -> ReplyEvent { + ReplyEvent::Final { + content: c.to_owned(), + } +} + struct SeqIds(Mutex); 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, + /// 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, + 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>, +} + +/// 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, + 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, + 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, 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, + Arc::new(SeqIds::new()), + )); + + let service = OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + 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, 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, 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, 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, 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"); +} diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 3b65f26..9cc9dd9 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -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. diff --git a/crates/domain/src/orchestrator.rs b/crates/domain/src/orchestrator.rs index cf52346..0a24fb4 100644 --- a/crates/domain/src/orchestrator.rs +++ b/crates/domain/src/orchestrator.rs @@ -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, - /// 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, /// 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, - /// 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, /// 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" }"#);