blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d7041c53ceae941ab61601848af8d0201f8d8116
IdeA/crates/application/tests/orchestrator_service.rs
Blomios 6236cd727b feat(permissions): LP4-4 — enforcement Landlock sur le chemin structuré 
Étend l'enforcement OS au chemin structuré (sessions Claude/Codex mode JSON),
jusqu'ici seulement advisory. Approche validée par l'Architecte : transposer la
technique du PTY plutôt qu'un pre_exec (rejeté — landlock alloue, deadlock malloc
post-fork en process multithreadé).

Mécanique (cfg(target_os=linux)) : run_turn_sandboxed/drain_sandboxed exécutent
enforce(plan) sur un thread jetable AVANT le spawn std, puis std::process::spawn
depuis ce thread ; l'enfant hérite le domaine Landlock via les credentials de la
tâche (garanti à travers fork/clone/execve, y compris posix_spawn — pas de pre_exec
nécessaire, forbid(unsafe_code) préservé). Fail-closed sur Err d'enforce (aucun
child). Timeout sous sandbox : oneshot killer + tokio::time::timeout → kill → EOF →
reap (pas de zombie/thread bloqué). Chemin non-sandboxé (plan None / pas d'enforcer /
non-Linux) = drain async tokio inchangé.

Contrat : SpawnLine.sandbox ; AgentSessionFactory::start(.., sandbox) ;
StructuredSessionFactory::with_sandbox_enforcer (jumeau du PTY) ; plan calculé en
step 5d de lifecycle relayé à launch_structured ; default_enforcer() injecté au
composition root.

Tests : 7 invariants e2e (parité, companion négatif, fail-closed, no-op natif,
confinement de l'irréversibilité entre tours, timeout, resume préservé) — zéro
token (sh/FakeCli). 80 suites vertes, 0 failed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-16 08:47:33 +02:00

3202 lines
111 KiB
Rust
Raw Blame History

//! Integration tests for [`OrchestratorService`] (ARCHITECTURE §14.3).
//!
//! The service is wired over the *real* agent/terminal use cases, themselves
//! backed by in-memory fakes (the same fake patterns as `agent_lifecycle.rs`).
//! This proves the dispatch contract end-to-end without real I/O:
//!
//! - `spawn_agent` on an **unknown** agent → create + launch (manifest grows, PTY
//! spawns, `AgentLaunched` published),
//! - `spawn_agent` on a **known** agent → launch only (no second manifest entry),
//! - `stop_agent` → the agent's live session is killed and de-registered,
//! - `update_agent_context` → the agent `.md` is overwritten,
//! - unknown profile / unknown agent → `NotFound`, no spawn.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use domain::agent::{Agent, AgentManifest, AgentOrigin, ManifestEntry};
use domain::events::DomainEvent;
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,
StoreError,
};
use domain::profile::{
AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport,
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,
};
// ---------------------------------------------------------------------------
// Fakes (mirror agent_lifecycle.rs)
// ---------------------------------------------------------------------------
#[derive(Default)]
struct ContextsInner {
manifest: AgentManifest,
contents: HashMap<String, String>,
}
#[derive(Clone)]
struct FakeContexts(Arc<Mutex<ContextsInner>>);
impl FakeContexts {
fn new() -> Self {
Self(Arc::new(Mutex::new(ContextsInner {
manifest: AgentManifest {
version: 1,
entries: Vec::new(),
},
contents: HashMap::new(),
})))
}
fn with_agent(agent: &Agent, content: &str) -> Self {
let me = Self::new();
{
let mut inner = me.0.lock().unwrap();
inner
.manifest
.entries
.push(ManifestEntry::from_agent(agent));
inner
.contents
.insert(agent.context_path.clone(), content.to_owned());
}
me
}
fn manifest(&self) -> AgentManifest {
self.0.lock().unwrap().manifest.clone()
}
fn content(&self, md_path: &str) -> Option<String> {
self.0.lock().unwrap().contents.get(md_path).cloned()
}
fn md_path_of(&self, agent: &AgentId) -> Option<String> {
self.0
.lock()
.unwrap()
.manifest
.entries
.iter()
.find(|e| &e.agent_id == agent)
.map(|e| e.md_path.clone())
}
}
#[async_trait]
impl AgentContextStore for FakeContexts {
async fn read_context(
&self,
_project: &Project,
agent: &AgentId,
) -> Result<MarkdownDoc, StoreError> {
let md_path = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
self.content(&md_path)
.map(MarkdownDoc::new)
.ok_or(StoreError::NotFound)
}
async fn write_context(
&self,
_project: &Project,
agent: &AgentId,
md: &MarkdownDoc,
) -> Result<(), StoreError> {
let md_path = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
self.0
.lock()
.unwrap()
.contents
.insert(md_path, md.as_str().to_owned());
Ok(())
}
async fn load_manifest(&self, _project: &Project) -> Result<AgentManifest, StoreError> {
Ok(self.manifest())
}
async fn save_manifest(
&self,
_project: &Project,
manifest: &AgentManifest,
) -> Result<(), StoreError> {
self.0.lock().unwrap().manifest = manifest.clone();
Ok(())
}
}
#[derive(Clone)]
struct FakeProfiles(Arc<Vec<AgentProfile>>);
impl FakeProfiles {
fn new(profiles: Vec<AgentProfile>) -> Self {
Self(Arc::new(profiles))
}
}
#[async_trait]
impl ProfileStore for FakeProfiles {
async fn list(&self) -> Result<Vec<AgentProfile>, StoreError> {
Ok((*self.0).clone())
}
async fn save(&self, _profile: &AgentProfile) -> Result<(), StoreError> {
Ok(())
}
async fn delete(&self, _id: ProfileId) -> Result<(), StoreError> {
Ok(())
}
async fn is_configured(&self) -> Result<bool, StoreError> {
Ok(true)
}
async fn mark_configured(&self) -> Result<(), StoreError> {
Ok(())
}
}
// Empty skill store: the orchestrator tests spawn agents with no assigned skills.
#[derive(Default)]
struct FakeSkills;
#[async_trait]
impl SkillStore for FakeSkills {
async fn list(
&self,
_scope: SkillScope,
_root: &ProjectPath,
) -> Result<Vec<Skill>, StoreError> {
Ok(Vec::new())
}
async fn get(
&self,
_scope: SkillScope,
_root: &ProjectPath,
_id: SkillId,
) -> Result<Skill, StoreError> {
Err(StoreError::NotFound)
}
async fn save(&self, _skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> {
Ok(())
}
async fn delete(
&self,
_scope: SkillScope,
_root: &ProjectPath,
_id: SkillId,
) -> Result<(), StoreError> {
Ok(())
}
}
/// An empty [`MemoryRecall`]: no project memory ⇒ no memory section injected,
/// leaving the launch behaviour unchanged for the service-level tests.
#[derive(Default)]
struct FakeRecall;
#[async_trait]
impl domain::ports::MemoryRecall for FakeRecall {
async fn recall(
&self,
_root: &ProjectPath,
_query: &domain::ports::MemoryQuery,
) -> Result<Vec<domain::MemoryIndexEntry>, domain::ports::MemoryError> {
Ok(Vec::new())
}
}
/// A [`SkillStore`] that records every `save` so a `create_skill` dispatch can be
/// asserted against the persisted skill (name + scope).
#[derive(Clone, Default)]
struct RecordingSkills(Arc<Mutex<Vec<Skill>>>);
#[async_trait]
impl SkillStore for RecordingSkills {
async fn list(
&self,
_scope: SkillScope,
_root: &ProjectPath,
) -> Result<Vec<Skill>, StoreError> {
Ok(self.0.lock().unwrap().clone())
}
async fn get(
&self,
_scope: SkillScope,
_root: &ProjectPath,
id: SkillId,
) -> Result<Skill, StoreError> {
self.0
.lock()
.unwrap()
.iter()
.find(|s| s.id == id)
.cloned()
.ok_or(StoreError::NotFound)
}
async fn save(&self, skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> {
self.0.lock().unwrap().push(skill.clone());
Ok(())
}
async fn delete(
&self,
_scope: SkillScope,
_root: &ProjectPath,
_id: SkillId,
) -> Result<(), StoreError> {
Ok(())
}
}
struct FakeRuntime;
impl AgentRuntime for FakeRuntime {
fn detect(&self, _profile: &AgentProfile) -> Result<bool, RuntimeError> {
Ok(true)
}
fn prepare_invocation(
&self,
profile: &AgentProfile,
_ctx: &PreparedContext,
cwd: &ProjectPath,
_session: &SessionPlan,
) -> Result<SpawnSpec, RuntimeError> {
Ok(SpawnSpec {
command: profile.command.clone(),
args: profile.args.clone(),
cwd: cwd.clone(),
env: Vec::new(),
context_plan: Some(ContextInjectionPlan::Stdin),
sandbox: None,
})
}
}
#[derive(Clone, Default)]
struct FakeFs;
#[async_trait]
impl FileSystem for FakeFs {
async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
Err(FsError::NotFound(path.as_str().to_owned()))
}
async fn write(&self, _path: &RemotePath, _data: &[u8]) -> Result<(), FsError> {
Ok(())
}
async fn exists(&self, _path: &RemotePath) -> Result<bool, FsError> {
Ok(false)
}
async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
Ok(())
}
async fn list(&self, _path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
Ok(Vec::new())
}
async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
Ok(())
}
}
#[derive(Clone)]
struct FakePty {
next_id: SessionId,
kills: Arc<Mutex<Vec<SessionId>>>,
spawns: Arc<Mutex<Vec<SessionId>>>,
/// Records `(session_id, text)` of every `write` so the inter-agent tests can
/// assert the delegated task was written into the target's terminal.
writes: Arc<Mutex<Vec<(SessionId, String)>>>,
}
impl FakePty {
fn new(next_id: SessionId) -> Self {
Self {
next_id,
kills: Arc::new(Mutex::new(Vec::new())),
spawns: Arc::new(Mutex::new(Vec::new())),
writes: Arc::new(Mutex::new(Vec::new())),
}
}
fn spawns(&self) -> Vec<SessionId> {
self.spawns.lock().unwrap().clone()
}
#[allow(dead_code)]
fn kills(&self) -> Vec<SessionId> {
self.kills.lock().unwrap().clone()
}
/// The texts written to a given session (lossy UTF-8), in order.
fn writes_for(&self, session_id: SessionId) -> Vec<String> {
self.writes
.lock()
.unwrap()
.iter()
.filter(|(s, _)| *s == session_id)
.map(|(_, t)| t.clone())
.collect()
}
}
#[async_trait]
impl PtyPort for FakePty {
async fn spawn(&self, _spec: SpawnSpec, _size: PtySize) -> Result<PtyHandle, PtyError> {
self.spawns.lock().unwrap().push(self.next_id);
Ok(PtyHandle {
session_id: self.next_id,
})
}
fn write(&self, handle: &PtyHandle, data: &[u8]) -> Result<(), PtyError> {
self.writes.lock().unwrap().push((
handle.session_id,
String::from_utf8_lossy(data).into_owned(),
));
Ok(())
}
fn resize(&self, _handle: &PtyHandle, _size: PtySize) -> Result<(), PtyError> {
Ok(())
}
fn subscribe_output(&self, _handle: &PtyHandle) -> Result<OutputStream, PtyError> {
Ok(Box::new(std::iter::empty()))
}
fn scrollback(&self, _handle: &PtyHandle) -> Result<Vec<u8>, PtyError> {
Ok(Vec::new())
}
async fn kill(&self, handle: &PtyHandle) -> Result<ExitStatus, PtyError> {
self.kills.lock().unwrap().push(handle.session_id);
Ok(ExitStatus { code: Some(0) })
}
}
#[derive(Default, Clone)]
struct SpyBus(Arc<Mutex<Vec<DomainEvent>>>);
impl SpyBus {
fn events(&self) -> Vec<DomainEvent> {
self.0.lock().unwrap().clone()
}
}
impl EventBus for SpyBus {
fn publish(&self, event: DomainEvent) {
self.0.lock().unwrap().push(event);
}
fn subscribe(&self) -> EventStream {
Box::new(std::iter::empty())
}
}
struct SeqIds(Mutex<u128>);
impl SeqIds {
fn new() -> Self {
Self(Mutex::new(1))
}
}
impl IdGenerator for SeqIds {
fn new_uuid(&self) -> Uuid {
let mut n = self.0.lock().unwrap();
let id = Uuid::from_u128(*n);
*n += 1;
id
}
}
// ---------------------------------------------------------------------------
// Builders
// ---------------------------------------------------------------------------
fn pid(n: u128) -> ProfileId {
ProfileId::from_uuid(Uuid::from_u128(n))
}
fn aid(n: u128) -> AgentId {
AgentId::from_uuid(Uuid::from_u128(n))
}
fn sid(n: u128) -> SessionId {
SessionId::from_uuid(Uuid::from_u128(n))
}
fn nid(n: u128) -> NodeId {
NodeId::from_uuid(Uuid::from_u128(n))
}
fn project() -> Project {
Project::new(
ProjectId::from_uuid(Uuid::from_u128(1000)),
"demo",
ProjectPath::new("/home/me/proj").unwrap(),
RemoteRef::local(),
1_700_000_000_000,
)
.unwrap()
}
fn claude_profile() -> AgentProfile {
// Profil Claude **complet** : adaptateur structuré + capacité MCP `.mcp.json`.
// C'est la condition que la garde F2 (`guard_mcp_bridge_supported`) exige pour
// accepter une cible d'`idea_ask_agent` (seul Claude consomme le pont `.mcp.json`).
AgentProfile::new(
pid(9),
"Claude Code",
"claude",
Vec::new(),
ContextInjection::stdin(),
Some("claude --version".to_owned()),
"{agentRunDir}",
None,
)
.unwrap()
.with_structured_adapter(StructuredAdapter::Claude)
.with_mcp(McpCapability::new(
McpConfigStrategy::config_file(".mcp.json").unwrap(),
McpTransport::Stdio,
))
}
fn scratch_agent(id: AgentId, name: &str, md: &str) -> Agent {
Agent::new(id, name, md, pid(9), AgentOrigin::Scratch, false).unwrap()
}
/// Everything wired for a dispatch test.
struct Fixture {
service: OrchestratorService,
contexts: FakeContexts,
pty: FakePty,
bus: SpyBus,
sessions: Arc<TerminalSessions>,
skills: RecordingSkills,
}
fn fixture(contexts: FakeContexts) -> Fixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
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,
));
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.clone()) 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),
);
Fixture {
service,
contexts,
pty,
bus,
sessions,
skills,
}
}
fn cmd(json: &str) -> OrchestratorCommand {
serde_json::from_str::<OrchestratorRequest>(json)
.unwrap()
.validate()
.unwrap()
}
// ---------------------------------------------------------------------------
// Basic dispatch coverage (spawn / stop / update / skill) — non-ask commands.
// ---------------------------------------------------------------------------
#[tokio::test]
async fn spawn_unknown_agent_creates_then_launches() {
let fx = fixture(FakeContexts::new());
let out = fx
.service
.dispatch(
&project(),
cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude-code" }"#),
)
.await
.expect("dispatch ok");
assert!(out.detail.contains("dev-backend"));
let manifest = fx.contexts.manifest();
assert_eq!(manifest.entries.len(), 1);
assert_eq!(manifest.entries[0].name, "dev-backend");
assert!(fx.sessions.session(&sid(777)).is_some());
let launched = fx.bus.events().into_iter().any(
|e| matches!(e, DomainEvent::AgentLaunched { session_id, .. } if session_id == sid(777)),
);
assert!(launched, "AgentLaunched must be published");
}
#[tokio::test]
async fn spawn_known_agent_launches_without_recreating() {
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
fx.service
.dispatch(
&project(),
cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude-code" }"#),
)
.await
.expect("dispatch ok");
assert_eq!(fx.contexts.manifest().entries.len(), 1);
assert_eq!(fx.contexts.manifest().entries[0].agent_id, agent.id);
assert!(fx.sessions.session(&sid(777)).is_some());
}
#[tokio::test]
async fn agent_run_existing_agent_does_not_require_profile() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
fx.service
.dispatch(
&project(),
cmd(r#"{ "type":"agent.run", "targetAgent":"architect", "task":"Analyse" }"#),
)
.await
.expect("dispatch ok");
assert_eq!(fx.contexts.manifest().entries.len(), 1);
assert!(fx.sessions.session(&sid(777)).is_some());
}
/// R0a (orchestrator): a `Visible` spawn re-targeting the same host cell of a live
/// agent is a view rebind ⇒ no error, no respawn.
#[tokio::test]
async fn agent_run_visible_same_cell_rebinds_existing_session() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
let cell = nid(10);
fx.service
.dispatch(
&project(),
cmd(&format!(
r#"{{ "action":"spawn_agent", "name":"architect", "profile":"claude", "visibility":"visible", "nodeId":"{cell}" }}"#
)),
)
.await
.expect("first launch ok");
let out = fx
.service
.dispatch(
&project(),
cmd(&format!(
r#"{{ "type":"agent.run", "targetAgent":"architect", "visibility":"visible", "attachToCell":"{cell}" }}"#
)),
)
.await
.expect("same-cell rebind ok");
assert!(out.detail.contains("attached agent architect"));
let session = fx.sessions.session(&sid(777)).expect("live session");
assert_eq!(session.node_id, cell);
assert_eq!(fx.pty.spawns().len(), 1, "rebind must not respawn");
}
/// R0a (orchestrator): a `Visible` spawn targeting a *different* cell of a live
/// singleton ⇒ refused `AgentAlreadyRunning` (host cell reported), no respawn.
#[tokio::test]
async fn agent_run_visible_other_cell_refuses_when_live_elsewhere() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
let first_cell = nid(10);
let next_cell = nid(20);
fx.service
.dispatch(
&project(),
cmd(&format!(
r#"{{ "action":"spawn_agent", "name":"architect", "profile":"claude", "visibility":"visible", "nodeId":"{first_cell}" }}"#
)),
)
.await
.expect("first launch ok");
let err = fx
.service
.dispatch(
&project(),
cmd(&format!(
r#"{{ "type":"agent.run", "targetAgent":"architect", "visibility":"visible", "attachToCell":"{next_cell}" }}"#
)),
)
.await
.expect_err("fresh second launch elsewhere is refused");
assert_eq!(err.code(), "AGENT_ALREADY_RUNNING", "got {err:?}");
match err {
application::AppError::AgentAlreadyRunning { node_id, .. } => {
assert_eq!(
node_id, first_cell,
"reports the live host cell, not target"
);
}
other => panic!("expected AgentAlreadyRunning, got {other:?}"),
}
let session = fx.sessions.session(&sid(777)).expect("live session");
assert_eq!(
session.node_id, first_cell,
"session stays on its host cell"
);
assert_eq!(fx.pty.spawns().len(), 1, "refused launch must not respawn");
}
#[tokio::test]
async fn stop_agent_kills_the_right_session() {
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
fx.service
.dispatch(
&project(),
cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude" }"#),
)
.await
.unwrap();
assert!(fx.sessions.session(&sid(777)).is_some());
fx.service
.dispatch(
&project(),
cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#),
)
.await
.expect("stop ok");
assert_eq!(fx.pty.kills(), vec![sid(777)]);
assert!(fx.sessions.session(&sid(777)).is_none());
}
#[tokio::test]
async fn stop_agent_without_live_session_is_not_found() {
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
let err = fx
.service
.dispatch(
&project(),
cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#),
)
.await
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
}
#[tokio::test]
async fn update_agent_context_overwrites_md() {
let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
let fx = fixture(FakeContexts::with_agent(&agent, "# old"));
fx.service
.dispatch(
&project(),
cmd(
r##"{ "action":"update_agent_context", "name":"dev-backend", "context":"# new body" }"##,
),
)
.await
.expect("update ok");
assert_eq!(
fx.contexts.content("agents/dev-backend.md").as_deref(),
Some("# new body")
);
}
#[tokio::test]
async fn spawn_with_unknown_profile_is_not_found_and_does_not_create() {
let fx = fixture(FakeContexts::new());
let err = fx
.service
.dispatch(
&project(),
cmd(r#"{ "action":"spawn_agent", "name":"x", "profile":"does-not-exist" }"#),
)
.await
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
assert!(fx.contexts.manifest().entries.is_empty());
assert!(fx.sessions.session(&sid(777)).is_none());
}
#[tokio::test]
async fn create_skill_persists_a_project_scoped_skill_by_default() {
let fx = fixture(FakeContexts::new());
let out = fx
.service
.dispatch(
&project(),
cmd(r##"{ "action":"create_skill", "name":"deploy", "context":"# Deploy" }"##),
)
.await
.expect("create_skill ok");
assert!(out.detail.contains("deploy"));
let saved = fx.skills.0.lock().unwrap();
assert_eq!(saved.len(), 1);
assert_eq!(saved[0].name, "deploy");
assert_eq!(saved[0].scope, SkillScope::Project);
}
#[tokio::test]
async fn create_skill_honours_global_scope() {
let fx = fixture(FakeContexts::new());
fx.service
.dispatch(
&project(),
cmd(
r##"{ "action":"create_skill", "name":"shared", "context":"# Shared", "scope":"global" }"##,
),
)
.await
.expect("create_skill ok");
let saved = fx.skills.0.lock().unwrap();
assert_eq!(saved.len(), 1);
assert_eq!(saved[0].scope, SkillScope::Global);
}
// ---------------------------------------------------------------------------
// Option 1 « Terminal + MCP » — idea_ask_agent / idea_reply (B-3 / B-4)
//
// The target's view is a raw native terminal (PTY). Delegation flows through the
// mailbox: `ask_agent` ensures the target is live in the PTY registry, writes the
// prefixed task into its terminal, and AWAITS a `PendingReply`; the target's later
// `idea_reply` resolves the head ticket of its queue, waking the awaiting ask.
//
// To exercise the blocking rendezvous deterministically, a test spawns the ask as a
// task, waits until the mailbox holds a pending ticket, then dispatches the matching
// `Reply` (or cancels via timeout) to unblock it. A real `InMemoryMailbox` is used
// (the production adapter), wired through `with_mailbox`.
// ---------------------------------------------------------------------------
use std::collections::VecDeque;
use domain::conversation::{
Conversation, ConversationId, ConversationParty, ConversationRegistry, ConversationSession,
SessionRef,
};
use domain::input::{AgentBusyState, InputMediator};
use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId};
use tokio::time::{timeout, Duration};
/// Safety guard: no awaited rendezvous in these tests should ever exceed this.
const TEST_GUARD: Duration = Duration::from_secs(10);
/// Minimal in-test [`AgentMailbox`] mirroring the production `InMemoryMailbox`
/// (which lives in `infrastructure` and cannot be a dev-dep of `application` — that
/// would be a dependency cycle). Per-agent FIFO of `(Ticket, oneshot::Sender)`.
#[derive(Default)]
struct TestMailbox {
queues: Mutex<HashMap<AgentId, VecDeque<(Ticket, tokio::sync::oneshot::Sender<String>)>>>,
}
impl TestMailbox {
fn new() -> Self {
Self {
queues: Mutex::new(HashMap::new()),
}
}
fn pending(&self, agent: &AgentId) -> usize {
self.queues
.lock()
.unwrap()
.get(agent)
.map_or(0, VecDeque::len)
}
/// Ids of the tickets queued for `agent`, in FIFO order (test inspection).
fn ticket_ids(&self, agent: &AgentId) -> Vec<TicketId> {
self.queues
.lock()
.unwrap()
.get(agent)
.map(|q| q.iter().map(|(t, _)| t.id).collect())
.unwrap_or_default()
}
}
impl AgentMailbox for TestMailbox {
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
self.queues
.lock()
.unwrap()
.entry(agent)
.or_default()
.push_back((ticket, tx));
PendingReply::new(Box::pin(async move {
rx.await.map_err(|_| MailboxError::Cancelled)
}))
}
fn resolve(&self, agent: AgentId, result: String) -> Result<(), MailboxError> {
let slot = {
let mut q = self.queues.lock().unwrap();
let queue = q
.get_mut(&agent)
.filter(|q| !q.is_empty())
.ok_or(MailboxError::NoPendingRequest(agent))?;
queue.pop_front().expect("non-empty")
};
let _ = slot.1.send(result);
Ok(())
}
fn resolve_ticket(
&self,
agent: AgentId,
ticket_id: TicketId,
result: String,
) -> Result<(), MailboxError> {
let slot = {
let mut q = self.queues.lock().unwrap();
let queue = q
.get_mut(&agent)
.filter(|q| !q.is_empty())
.ok_or(MailboxError::NoPendingRequest(agent))?;
let pos = queue
.iter()
.position(|(t, _)| t.id == ticket_id)
.ok_or(MailboxError::NoPendingRequest(agent))?;
queue.remove(pos).expect("found position")
};
let _ = slot.1.send(result);
Ok(())
}
fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
let mut q = self.queues.lock().unwrap();
if let Some(queue) = q.get_mut(&agent) {
if queue.front().map(|(t, _)| t.id) == Some(ticket_id) {
queue.pop_front();
}
}
}
}
/// A delivering [`InputMediator`] for the application tests: it composes the
/// [`TestMailbox`] (correlation engine) and writes the prefixed turn into the agent's
/// bound PTY handle — the test mirror of `infrastructure::MediatedInbox::with_pty`
/// (which `application` cannot depend on without a dependency cycle).
struct TestMediator {
mailbox: Arc<TestMailbox>,
pty: Arc<dyn PtyPort>,
handles: Mutex<HashMap<AgentId, PtyHandle>>,
busy: Mutex<HashMap<AgentId, AgentBusyState>>,
/// Records every agent passed to `preempt`, so the interrupt test can assert the
/// Interrompre path calls preempt (and only preempt — no enqueue, no resolve).
preempts: Arc<Mutex<Vec<AgentId>>>,
}
impl TestMediator {
fn new(mailbox: Arc<TestMailbox>, pty: Arc<dyn PtyPort>) -> Self {
Self {
mailbox,
pty,
handles: Mutex::new(HashMap::new()),
busy: Mutex::new(HashMap::new()),
preempts: Arc::new(Mutex::new(Vec::new())),
}
}
fn preempts(&self) -> Vec<AgentId> {
self.preempts.lock().unwrap().clone()
}
}
impl InputMediator for TestMediator {
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
let ticket_id = ticket.id;
{
let mut b = self.busy.lock().unwrap();
let st = b.entry(agent).or_insert(AgentBusyState::Idle);
if !st.is_busy() {
*st = AgentBusyState::Busy {
ticket: ticket_id,
since_ms: 0,
};
}
}
if let Some(handle) = self.handles.lock().unwrap().get(&agent).cloned() {
let line = format!(
"[IdeA · tâche de {} · ticket {}] {}\n",
ticket.requester, ticket_id, ticket.task
);
let _ = self.pty.write(&handle, line.as_bytes());
}
self.mailbox.enqueue(agent, ticket)
}
fn bind_handle(&self, agent: AgentId, handle: PtyHandle) {
self.handles.lock().unwrap().insert(agent, handle);
}
fn delivers_turn(&self, agent: AgentId) -> bool {
self.handles.lock().unwrap().contains_key(&agent)
}
fn preempt(&self, agent: AgentId) {
self.preempts.lock().unwrap().push(agent);
}
fn mark_idle(&self, agent: AgentId) {
self.busy
.lock()
.unwrap()
.insert(agent, AgentBusyState::Idle);
}
fn busy_state(&self, agent: AgentId) -> AgentBusyState {
self.busy
.lock()
.unwrap()
.get(&agent)
.copied()
.unwrap_or(AgentBusyState::Idle)
}
}
/// Minimal in-test [`ConversationRegistry`] (pair-keyed lazy get-or-create), mirroring
/// `infrastructure::InMemoryConversationRegistry`.
#[derive(Default)]
struct TestConversations {
by_pair: Mutex<HashMap<(String, String), ConversationId>>,
by_id: Mutex<HashMap<ConversationId, Conversation>>,
}
impl TestConversations {
fn new() -> Self {
Self::default()
}
fn key(a: ConversationParty, b: ConversationParty) -> (String, String) {
let s = |p: ConversationParty| match p {
ConversationParty::User => "user".to_owned(),
ConversationParty::Agent { agent_id } => agent_id.to_string(),
};
let (ka, kb) = (s(a), s(b));
if ka <= kb {
(ka, kb)
} else {
(kb, ka)
}
}
}
impl ConversationRegistry for TestConversations {
fn resolve(&self, a: ConversationParty, b: ConversationParty) -> Conversation {
let key = Self::key(a, b);
let mut pairs = self.by_pair.lock().unwrap();
if let Some(id) = pairs.get(&key).copied() {
return self.by_id.lock().unwrap().get(&id).cloned().unwrap();
}
let id = ConversationId::new_random();
let conv = Conversation::try_new(id, a, b).expect("valid pair");
pairs.insert(key, id);
self.by_id.lock().unwrap().insert(id, conv.clone());
conv
}
fn bind_session(&self, id: ConversationId, session: SessionRef) {
if let Some(c) = self.by_id.lock().unwrap().get_mut(&id) {
c.session = ConversationSession::Live {
handle_ref: session,
};
}
}
fn suspend(&self, id: ConversationId, resumable_id: Option<String>) {
if let Some(c) = self.by_id.lock().unwrap().get_mut(&id) {
c.session = ConversationSession::Dormant;
c.resumable_id = resumable_id;
}
}
fn get(&self, id: ConversationId) -> Option<Conversation> {
self.by_id.lock().unwrap().get(&id).cloned()
}
}
/// Everything wired for an `idea_ask_agent` / `idea_reply` dispatch test.
struct AskFixture {
service: Arc<OrchestratorService>,
/// The real in-memory mailbox the service holds — lets a test observe pending
/// tickets and (in production it is the orchestrator that resolves them).
mailbox: Arc<TestMailbox>,
/// PTY (terminal) registry — the channel the target's task is written into.
sessions: Arc<TerminalSessions>,
bus: SpyBus,
/// PTY spy: records spawns (a dead target is launched as a PTY) and writes.
pty: FakePty,
/// The mediated inbox the service holds — lets a test observe `busy_state`
/// transitions (e.g. an `idea_reply` flips the emitter back to Idle — lot C5).
mediator: Arc<TestMediator>,
}
/// Builds an orchestrator wired with a real [`InMemoryMailbox`] + PTY (B-3) and a
/// plain (non-structured) [`LaunchAgent`] — so a dead target is launched as a PTY,
/// exactly like production after B-2.
fn ask_fixture(contexts: FakeContexts) -> AskFixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
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,
));
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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone())),
);
AskFixture {
service,
mailbox,
sessions,
bus,
pty,
mediator,
}
}
/// Seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it instead
/// of launching one.
fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) {
let session = TerminalSession::starting(
session_id,
nid(1),
ProjectPath::new("/home/me/proj").unwrap(),
SessionKind::Agent { agent_id },
PtySize::new(24, 80).unwrap(),
);
sessions.insert(PtyHandle { session_id }, session);
}
/// Waits (bounded) for a condition to hold, yielding between polls.
async fn await_until<F: Fn() -> bool>(cond: F) {
timeout(TEST_GUARD, async {
while !cond() {
tokio::task::yield_now().await;
}
})
.await
.expect("condition never reached within guard (possible hang/deadlock)");
}
const ASK_JSON: &str = r#"{ "type":"agent.message", "requestedBy":"Main", "targetAgent":"architect", "task":"Analyse §17" }"#;
/// Builds an `agent.reply` command for `from` with `result` (the handshake-injected
/// path, exactly what the MCP server produces from a peer's `idea_reply`).
fn reply_cmd(from: AgentId, result: &str) -> OrchestratorCommand {
OrchestratorCommand::Reply {
from,
ticket: None,
result: result.to_owned(),
}
}
/// A reply correlated **by ticket** (multi-thread correlation).
fn reply_cmd_ticket(from: AgentId, ticket: TicketId, result: &str) -> OrchestratorCommand {
OrchestratorCommand::Reply {
from,
ticket: Some(ticket),
result: result.to_owned(),
}
}
// --- B-3: ask blocks on the mailbox, reply unblocks it ---------------------
/// A live-PTY target: `ask` writes the prefixed task into its terminal and AWAITS;
/// a matching `idea_reply` resolves the head ticket and the ask returns its content.
#[tokio::test]
async fn ask_live_pty_target_writes_task_and_returns_reply() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// Target already live in the PTY registry (session 800).
seed_live_pty(&fx.sessions, aid(1), sid(800));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
// The ask enqueues a ticket and blocks awaiting the reply.
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// The task was written into the target's terminal, prefixed for idea_reply.
let writes = fx.pty.writes_for(sid(800));
assert_eq!(
writes.len(),
1,
"exactly one task write to the live terminal"
);
assert!(writes[0].contains("Analyse §17"), "task body written");
assert!(
writes[0].contains("[IdeA · tâche"),
"delegated-task prefix present"
);
assert!(
writes[0].contains("idea_reply") || writes[0].contains("ticket"),
"carries ticket/idea_reply cue"
);
// No PTY spawned: the live terminal was reused.
assert!(
fx.pty.spawns().is_empty(),
"live target reused, not respawned"
);
// The target renders its result via idea_reply ⇒ resolves the head ticket.
fx.service
.dispatch(&project(), reply_cmd(aid(1), "the §17 answer"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes once replied")
.expect("join ok")
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("the §17 answer"));
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "ticket drained on reply");
}
/// Lot C5 — explicit OR signal: an `idea_reply` from the emitting agent flips it back
/// to `Idle` (its FIFO can advance), independently of any prompt-ready pattern. Before
/// the reply the agent is `Busy` on the delegated turn; after, it is `Idle`.
#[tokio::test]
async fn idea_reply_marks_emitter_idle() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// The delegated turn started ⇒ the target is Busy.
assert!(
fx.mediator.busy_state(aid(1)).is_busy(),
"target Busy while processing the delegated turn"
);
// The target renders its result via idea_reply ⇒ explicit end-of-turn signal.
fx.service
.dispatch(&project(), reply_cmd(aid(1), "the §17 answer"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
// C5: the reply marked the emitter Idle (FIFO can advance) — no prompt pattern needed.
assert_eq!(
fx.mediator.busy_state(aid(1)),
AgentBusyState::Idle,
"idea_reply is the explicit signal that frees the turn"
);
}
/// Régression (garde RAII de fin de tour) — LE test décisif : un `ask_agent` dont le
/// **futur est abandonné (drop)** alors qu'il attendait la réponse doit laisser la cible
/// `Idle`, **pas** `Busy` à vie. Avant le fix, aucune branche du `match`/`select!` ne
/// s'exécutait sur un drop ⇒ l'agent restait `Busy` et toute délégation suivante était
/// mise en file derrière ce tour fantôme sans jamais être livrée.
#[tokio::test]
async fn dropped_ask_future_frees_busy_target() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
// Le tour a démarré : la cible est Busy, un ticket est en file.
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
assert!(
fx.mediator.busy_state(aid(1)).is_busy(),
"cible Busy pendant le tour délégué"
);
// Le demandeur interrompt son appel ⇒ le futur `ask_agent` est dropped.
ask.abort();
let _ = ask.await; // récolte la JoinError(Cancelled), ignorée.
// Le garde RAII a ramené la cible Idle ET retiré le ticket fantôme de la FIFO.
await_until(|| !fx.mediator.busy_state(aid(1)).is_busy()).await;
assert_eq!(
fx.mediator.busy_state(aid(1)),
AgentBusyState::Idle,
"futur dropped ⇒ la cible est ramenée Idle par le garde (fix cause racine)"
);
assert_eq!(
fx.mailbox.pending(&aid(1)),
0,
"ticket fantôme retiré de la FIFO au Drop du garde"
);
}
/// Régression (garde RAII) — une **2e délégation** vers une cible dont un 1er `ask` a
/// été abandonné est bien livrée : l'agent n'est pas coincé `Busy`, son tour suivant
/// démarre et peut être résolu par `idea_reply`.
#[tokio::test]
async fn second_delegation_delivered_after_dropped_ask() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
// 1er ask : démarre puis est abandonné (drop).
let svc = Arc::clone(&fx.service);
let ask1 = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
ask1.abort();
let _ = ask1.await;
await_until(|| fx.mailbox.pending(&aid(1)) == 0).await;
// 2e ask vers la MÊME cible : doit démarrer un nouveau tour (pas coincé derrière un
// tour fantôme) et se laisser résoudre.
let svc2 = Arc::clone(&fx.service);
let ask2 = tokio::spawn(async move { svc2.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
assert!(
fx.mediator.busy_state(aid(1)).is_busy(),
"le 2e tour démarre bien (cible Busy) — preuve qu'elle n'était pas coincée"
);
fx.service
.dispatch(&project(), reply_cmd(aid(1), "réponse au 2e tour"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask2)
.await
.expect("le 2e ask se termine")
.expect("join ok")
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("réponse au 2e tour"));
assert_eq!(
fx.mediator.busy_state(aid(1)),
AgentBusyState::Idle,
"cible Idle après résolution du 2e tour"
);
}
/// Régression (garde RAII) — la **branche erreur/timeout** (canal fermé) ramène aussi la
/// cible `Idle`. Avant le fix, ces branches faisaient `cancel_head` mais jamais
/// `mark_idle` ⇒ l'agent restait `Busy`. On déclenche la fermeture du canal en retirant
/// le ticket de tête (même nettoyage que le timeout de tour).
#[tokio::test]
async fn cancelled_ask_marks_target_idle() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
assert!(fx.mediator.busy_state(aid(1)).is_busy());
// Retire le ticket de tête (drop du sender) ⇒ l'ask voit un canal fermé et part en
// erreur typée (PROCESS), le MÊME nettoyage que le timeout de tour.
let t = fx.mailbox.ticket_ids(&aid(1))[0];
fx.mailbox.cancel_head(aid(1), t);
let err = timeout(TEST_GUARD, ask)
.await
.expect("ask retourne vite sur canal fermé")
.expect("join ok")
.unwrap_err();
assert!(
matches!(err.code(), "PROCESS" | "INVALID"),
"ask sur canal fermé ⇒ erreur typée : {err:?}"
);
// Le garde a ramené la cible Idle (la FIFO peut avancer).
assert_eq!(
fx.mediator.busy_state(aid(1)),
AgentBusyState::Idle,
"branche erreur ⇒ cible Idle (garde RAII)"
);
}
/// A dead target is launched in the background (PTY) before the task is written.
#[tokio::test]
async fn ask_dead_target_launches_pty_then_writes_and_replies() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
assert!(fx.sessions.session_for_agent(&aid(1)).is_none());
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// The dead target was launched as a PTY (spawn recorded, session registered).
assert_eq!(
fx.pty.spawns(),
vec![sid(777)],
"dead target launched as PTY"
);
assert_eq!(fx.sessions.session_for_agent(&aid(1)), Some(sid(777)));
// The delegated task was written into the freshly-launched terminal (the Stdin
// context injection may also write the persona, so we look for the task prefix).
assert!(
fx.pty
.writes_for(sid(777))
.iter()
.any(|w| w.contains("[IdeA · tâche") && w.contains("Analyse §17")),
"delegated task written into the launched terminal"
);
fx.service
.dispatch(&project(), reply_cmd(aid(1), "launched reply"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("launched reply"));
}
/// On a successful reply, `DomainEvent::AgentReplied` is published with the right
/// agent id and byte length.
#[tokio::test]
async fn ask_success_publishes_agent_replied_event() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let content = "réponse"; // multibyte: len() in bytes
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(1), content))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
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());
}
// --- B-3: errors / not-found / not-wired -----------------------------------
/// Unknown target ⇒ typed `NOT_FOUND`, no launch, no ticket.
#[tokio::test]
async fn ask_unknown_target_is_not_found() {
let fx = ask_fixture(FakeContexts::new());
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!(fx.pty.spawns().is_empty());
}
/// Mailbox/PTY not wired (legacy `OrchestratorService::new` without `with_mailbox`)
/// ⇒ `ask` is INVALID.
#[tokio::test]
async fn ask_without_mailbox_wired_is_invalid() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
// The plain `fixture` builds the service WITHOUT `.with_mailbox(...)`.
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:?}");
}
/// `agent.run` (fire-and-forget) still yields `reply: None` through a mailbox-wired
/// orchestrator (non-regression).
#[tokio::test]
async fn non_regression_agent_run_reply_is_none() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
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");
}
// --- B-4: idea_reply (Reply command) ---------------------------------------
/// A `Reply` with no in-flight ask for that agent ⇒ typed `INVALID` (no panic).
#[tokio::test]
async fn reply_without_pending_ask_is_invalid() {
let fx = ask_fixture(FakeContexts::new());
let err = fx
.service
.dispatch(&project(), reply_cmd(aid(7), "orphan result"))
.await
.unwrap_err();
assert_eq!(
err.code(),
"INVALID",
"orphan reply is typed INVALID: {err:?}"
);
}
/// `Reply` resolves the HEAD ticket of the emitting agent's queue (positional
/// correlation): two asks to the same target, two replies, FIFO order.
#[tokio::test]
async fn reply_resolves_head_of_queue_fifo() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
// Two concurrent asks to the SAME target. The per-agent turn lock serialises
// them: only the first holds the lock and enqueues; the second waits on the lock.
let svc1 = Arc::clone(&fx.service);
let ask1 = tokio::spawn(async move {
svc1.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"architect", "task":"T1" }"#),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// First reply resolves T1; ask1 returns it. Then ask2 may proceed.
fx.service
.dispatch(&project(), reply_cmd(aid(1), "R1"))
.await
.expect("reply 1 ok");
let out1 = timeout(TEST_GUARD, ask1).await.unwrap().unwrap().unwrap();
assert_eq!(out1.reply.as_deref(), Some("R1"));
let svc2 = Arc::clone(&fx.service);
let ask2 = tokio::spawn(async move {
svc2.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"architect", "task":"T2" }"#),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "R2"))
.await
.expect("reply 2 ok");
let out2 = timeout(TEST_GUARD, ask2).await.unwrap().unwrap().unwrap();
assert_eq!(out2.reply.as_deref(), Some("R2"));
}
/// Two asks to DIFFERENT targets do not block each other (per-agent lock + per-agent
/// queue): A is held (never replied) while B completes.
#[tokio::test]
async fn ask_different_targets_run_in_parallel() {
let contexts = FakeContexts::new();
{
let a = scratch_agent(aid(1), "alpha", "agents/alpha.md");
let b = scratch_agent(aid(2), "beta", "agents/beta.md");
let mut inner = contexts.0.lock().unwrap();
inner.manifest.entries.push(ManifestEntry::from_agent(&a));
inner.manifest.entries.push(ManifestEntry::from_agent(&b));
inner
.contents
.insert(a.context_path.clone(), "# a".to_owned());
inner
.contents
.insert(b.context_path.clone(), "# b".to_owned());
}
let fx = ask_fixture(contexts);
seed_live_pty(&fx.sessions, aid(1), sid(801));
seed_live_pty(&fx.sessions, aid(2), sid(802));
// A: launched and held (never replied during the test).
let svc_a = Arc::clone(&fx.service);
let _ask_a = tokio::spawn(async move {
svc_a
.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"alpha", "task":"A1" }"#),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// B completes while A is still in flight (no cross-agent contention).
let svc_b = Arc::clone(&fx.service);
let ask_b = tokio::spawn(async move {
svc_b
.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"beta", "task":"B1" }"#),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(2), "B-done"))
.await
.expect("reply B ok");
let out_b = timeout(TEST_GUARD, ask_b)
.await
.expect("B completes while A is held")
.unwrap()
.unwrap();
assert_eq!(out_b.reply.as_deref(), Some("B-done"));
// A is still pending: it was not unblocked by B.
assert_eq!(fx.mailbox.pending(&aid(1)), 1, "A still in flight");
}
// ---------------------------------------------------------------------------
// F1 — McpRuntimeProvider câblé (preuve que le runtime atteint LaunchAgentInput)
// + F2 — garde guard_mcp_bridge_supported
//
// F1 était le bug : sur le chemin `ask`, `ensure_live_pty` passait
// `mcp_runtime: None` ⇒ `.mcp.json` minimal ⇒ pont MCP jamais spawné ⇒ timeout.
// Le fix injecte un `McpRuntimeProvider`. On le prouve **de bout en bout** : la
// déclaration `.mcp.json` réellement écrite par `LaunchAgent::apply_mcp_config`
// (via le FileSystem) doit porter l'endpoint/exe/requester du runtime fourni —
// ce qui n'est possible QUE si le runtime a traversé `LaunchAgentInput.mcp_runtime`.
// ---------------------------------------------------------------------------
/// FileSystem enregistrant chaque `write(path, content)` pour pouvoir inspecter la
/// déclaration `.mcp.json` matérialisée par le lancement. `exists → false` pour que
/// `apply_mcp_config` écrive bien le fichier (chemin non-clobbering).
#[derive(Clone, Default)]
struct CapturingFs(Arc<Mutex<Vec<(String, String)>>>);
impl CapturingFs {
fn writes(&self) -> Vec<(String, String)> {
self.0.lock().unwrap().clone()
}
/// Contenu écrit dont le chemin se termine par `suffix` (ex. `.mcp.json`).
fn content_ending_with(&self, suffix: &str) -> Option<String> {
self.0
.lock()
.unwrap()
.iter()
.find(|(p, _)| p.ends_with(suffix))
.map(|(_, c)| c.clone())
}
}
#[async_trait]
impl FileSystem for CapturingFs {
async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
Err(FsError::NotFound(path.as_str().to_owned()))
}
async fn write(&self, path: &RemotePath, data: &[u8]) -> Result<(), FsError> {
self.0.lock().unwrap().push((
path.as_str().to_owned(),
String::from_utf8_lossy(data).into_owned(),
));
Ok(())
}
async fn exists(&self, _path: &RemotePath) -> Result<bool, FsError> {
Ok(false)
}
async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
Ok(())
}
async fn list(&self, _path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
Ok(Vec::new())
}
async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
Ok(())
}
}
/// Provider de runtime MCP de test : renvoie un [`McpRuntime`] connu et **enregistre**
/// l'`(project_id, agent_id)` reçus, prouvant que l'orchestrateur l'interroge avec la
/// cible relancée.
#[derive(Clone, Default)]
struct FakeMcpRuntimeProvider {
calls: Arc<Mutex<Vec<(ProjectId, AgentId)>>>,
}
impl FakeMcpRuntimeProvider {
fn calls(&self) -> Vec<(ProjectId, AgentId)> {
self.calls.lock().unwrap().clone()
}
/// Le runtime connu que ce provider injecte (valeurs sentinelles repérables dans
/// le `.mcp.json` écrit).
fn known_runtime(agent_id: AgentId) -> application::McpRuntime {
application::McpRuntime {
exe: "/opt/idea/idea.AppImage".to_owned(),
endpoint: "/run/user/1000/idea-mcp/SENTINEL-endpoint.sock".to_owned(),
project_id: "deadbeefdeadbeefdeadbeefdeadbeef".to_owned(),
requester: agent_id.to_string(),
}
}
}
impl application::McpRuntimeProvider for FakeMcpRuntimeProvider {
fn runtime_for(&self, project: &Project, agent_id: AgentId) -> Option<application::McpRuntime> {
self.calls.lock().unwrap().push((project.id, agent_id));
Some(Self::known_runtime(agent_id))
}
}
/// Codex : profil structuré NON supporté par le pont `.mcp.json` (la garde F2 doit
/// le refuser, même s'il déclare une capacité MCP `ConfigFile(.mcp.json)`).
fn codex_profile() -> AgentProfile {
AgentProfile::new(
pid(11),
"Codex CLI",
"codex",
Vec::new(),
ContextInjection::stdin(),
Some("codex --version".to_owned()),
"{agentRunDir}",
None,
)
.unwrap()
.with_structured_adapter(StructuredAdapter::Codex)
.with_mcp(McpCapability::new(
McpConfigStrategy::config_file(".mcp.json").unwrap(),
McpTransport::Stdio,
))
}
/// Fixture `ask` paramétrable : choix du FileSystem (capturant), des profils, et d'un
/// éventuel [`McpRuntimeProvider`]. Reproduit `ask_fixture` mais expose les leviers
/// nécessaires aux preuves F1/F2.
struct AskFixtureEx {
service: Arc<OrchestratorService>,
mailbox: Arc<TestMailbox>,
sessions: Arc<TerminalSessions>,
pty: FakePty,
fs: CapturingFs,
}
fn ask_fixture_ex(
contexts: FakeContexts,
profiles: Vec<AgentProfile>,
provider: Option<Arc<FakeMcpRuntimeProvider>>,
) -> AskFixtureEx {
let profiles = Arc::new(FakeProfiles::new(profiles));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
let fs = CapturingFs::default();
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(fs.clone()),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
));
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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
)) as Arc<dyn InputMediator>;
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let mut service = OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()));
if let Some(p) = provider {
service = service.with_mcp_runtime_provider(p as Arc<dyn application::McpRuntimeProvider>);
}
AskFixtureEx {
service: Arc::new(service),
mailbox,
sessions,
pty,
fs,
}
}
/// F1 — PREUVE DIRECTE. Un `McpRuntimeProvider` câblé renvoie un runtime connu ; un
/// `ask` vers une cible **morte** passe par `ensure_live_pty` → `launch_agent`, et la
/// déclaration `.mcp.json` réellement matérialisée porte l'endpoint/exe/requester du
/// runtime fourni. C'est la preuve que `mcp_runtime: Some(runtime)` a traversé
/// `LaunchAgentInput` (avant le fix il valait `None` ⇒ déclaration minimale).
#[tokio::test]
async fn f1_ask_dead_target_injects_provider_runtime_into_mcp_json() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let provider = Arc::new(FakeMcpRuntimeProvider::default());
let fx = ask_fixture_ex(
FakeContexts::with_agent(&agent, "# persona"),
vec![claude_profile()],
Some(Arc::clone(&provider)),
);
assert!(fx.sessions.session_for_agent(&aid(1)).is_none());
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// Le provider a été interrogé avec le projet et la cible relancée.
let calls = provider.calls();
assert_eq!(calls.len(), 1, "provider interrogé exactement une fois");
assert_eq!(calls[0].0, project().id, "interrogé pour le bon projet");
assert_eq!(
calls[0].1,
aid(1),
"interrogé pour la cible relancée (= --requester)"
);
// La déclaration `.mcp.json` écrite porte les valeurs sentinelles du runtime.
let decl = fx
.fs
.content_ending_with(".mcp.json")
.expect("un .mcp.json doit être écrit au lancement de la cible morte");
assert!(
decl.contains("/opt/idea/idea.AppImage"),
"exe du runtime injecté présent dans le .mcp.json: {decl}"
);
assert!(
decl.contains("SENTINEL-endpoint.sock"),
"endpoint du runtime injecté présent: {decl}"
);
assert!(
decl.contains("deadbeefdeadbeefdeadbeefdeadbeef"),
"project_id du runtime injecté présent: {decl}"
);
assert!(
decl.contains(&aid(1).to_string()),
"requester (= cible) présent: {decl}"
);
// Preuve de discrimination avec le mode dégradé : le `command` est l'exe injecté,
// PAS la valeur minimale `"command": "idea"` qu'on aurait sans runtime.
assert!(
!decl.contains("\"command\": \"idea\""),
"command NE doit PAS être la valeur minimale (runtime injecté): {decl}"
);
// Débloque l'ask pour ne pas laisser la tâche pendante.
fx.service
.dispatch(&project(), reply_cmd(aid(1), "done"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}
/// F1 — NON-RÉGRESSION. Sans provider câblé (`OrchestratorService::new` legacy), le
/// lancement issu d'`ask` passe `mcp_runtime: None` ⇒ déclaration **minimale**
/// (`command = "idea"`, pas d'endpoint). Comportement legacy intact.
#[tokio::test]
async fn f1_ask_without_provider_writes_minimal_mcp_json() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture_ex(
FakeContexts::with_agent(&agent, "# persona"),
vec![claude_profile()],
None, // pas de provider câblé
);
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
let decl = fx
.fs
.content_ending_with(".mcp.json")
.expect("un .mcp.json minimal doit être écrit");
assert!(
decl.contains("\"command\": \"idea\""),
"command minimal attendu sans provider: {decl}"
);
assert!(
!decl.contains("--endpoint"),
"aucune déclaration d'endpoint sans runtime injecté: {decl}"
);
fx.service
.dispatch(&project(), reply_cmd(aid(1), "done"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}
/// F2 — GARDE (rejet). Une cible au profil Codex (structuré mais non consommateur du
/// pont `.mcp.json`) ⇒ `AppError::Invalid` **immédiat** (pas de timeout 300s), aucun
/// lancement, aucun ticket.
#[tokio::test]
async fn f2_ask_codex_target_is_invalid_no_launch() {
let agent = Agent::new(
aid(1),
"coder",
"agents/coder.md",
pid(11), // profil Codex
AgentOrigin::Scratch,
false,
)
.unwrap();
let fx = ask_fixture_ex(
FakeContexts::with_agent(&agent, "# persona"),
vec![codex_profile()],
Some(Arc::new(FakeMcpRuntimeProvider::default())),
);
let err = fx
.service
.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#),
)
.await
.unwrap_err();
assert_eq!(
err.code(),
"INVALID",
"garde F2 = erreur typée Invalid: {err:?}"
);
let msg = err.to_string();
assert!(
msg.contains("idea_*") || msg.contains("pont") || msg.contains(".mcp.json"),
"message explicite sur le pont non supporté: {msg}"
);
assert!(
fx.pty.spawns().is_empty(),
"aucun lancement sur cible refusée"
);
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "aucun ticket enfilé");
}
/// F2 — GARDE (passant). Une cible au profil Claude conforme passe la garde et
/// l'`ask` se déroule normalement jusqu'à la réponse (preuve que la garde ne bloque
/// pas les cibles légitimes).
#[tokio::test]
async fn f2_ask_claude_target_passes_guard() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture_ex(
FakeContexts::with_agent(&agent, "# persona"),
vec![claude_profile()],
Some(Arc::new(FakeMcpRuntimeProvider::default())),
);
seed_live_pty(&fx.sessions, aid(1), sid(800));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "ok claude"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
assert_eq!(out.reply.as_deref(), Some("ok claude"));
}
// ---------------------------------------------------------------------------
// C3 — conversation par paire : routage de fil, détection de cycle, corrélation
// par ticket, fallback tête, timeout libère la file.
// ---------------------------------------------------------------------------
/// Builds an ask-wired service exposing the **shared** [`TestConversations`] registry
/// (so a test can assert the resolved thread is A↔B, never User↔B). Mirrors
/// `ask_fixture` but returns the registry handle.
struct AskFixtureC3 {
service: Arc<OrchestratorService>,
mailbox: Arc<TestMailbox>,
sessions: Arc<TerminalSessions>,
conversations: Arc<TestConversations>,
}
fn ask_fixture_c3(contexts: FakeContexts) -> AskFixtureC3 {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
let conversations = Arc::new(TestConversations::new());
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,
));
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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
)) as Arc<dyn InputMediator>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
.with_conversations(Arc::clone(&conversations) as Arc<dyn ConversationRegistry>)
.with_events(Arc::new(bus.clone())),
);
AskFixtureC3 {
service,
mailbox,
sessions,
conversations,
}
}
/// Seeds two agents A (id 1) and B (id 2) in the manifest.
fn seed_two_agents(contexts: &FakeContexts) {
let a = scratch_agent(aid(1), "alpha", "agents/alpha.md");
let b = scratch_agent(aid(2), "beta", "agents/beta.md");
let mut inner = contexts.0.lock().unwrap();
inner.manifest.entries.push(ManifestEntry::from_agent(&a));
inner.manifest.entries.push(ManifestEntry::from_agent(&b));
inner
.contents
.insert(a.context_path.clone(), "# a".to_owned());
inner
.contents
.insert(b.context_path.clone(), "# b".to_owned());
}
/// An `agent.message` from agent A (uuid requester) to a named target.
fn ask_from_agent_json(requester: AgentId, target: &str, task: &str) -> String {
format!(
r#"{{ "type":"agent.message", "requestedBy":"{requester}", "targetAgent":"{target}", "task":"{task}" }}"#
)
}
/// C3 — an ask A→B routes into the **A↔B** thread, never User↔B. The resolved
/// conversation relates the two agents, and is distinct from the User↔B thread.
#[tokio::test]
async fn ask_agent_routes_into_a_to_b_conversation_not_user_b() {
let contexts = FakeContexts::new();
seed_two_agents(&contexts);
let fx = ask_fixture_c3(contexts);
seed_live_pty(&fx.sessions, aid(2), sid(802)); // B live
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
svc.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(1), "beta", "do B")),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
// The registry now holds the A↔B thread (agent 1 ↔ agent 2), not User↔B.
let a_to_b = fx.conversations.resolve(
ConversationParty::agent(aid(1)),
ConversationParty::agent(aid(2)),
);
let user_b = fx
.conversations
.resolve(ConversationParty::User, ConversationParty::agent(aid(2)));
assert_ne!(a_to_b.id, user_b.id, "A↔B is a distinct thread from User↔B");
assert!(
a_to_b.same_pair(
ConversationParty::agent(aid(1)),
ConversationParty::agent(aid(2))
),
"ask A→B resolved the A↔B pair"
);
// The live B session is bound to the A↔B thread (session keyed by conversation).
assert!(
a_to_b.session.is_live(),
"A↔B thread bound to a live session"
);
fx.service
.dispatch(&project(), reply_cmd(aid(2), "B done"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
assert_eq!(out.reply.as_deref(), Some("B done"));
}
/// C3 — a re-entrant delegation A→B→A is refused with a typed error **before** any
/// deadlock: while A→B is in flight (edge A→B posted), B→A must be rejected.
#[tokio::test]
async fn cycle_a_to_b_to_a_is_refused_before_deadlock() {
let contexts = FakeContexts::new();
seed_two_agents(&contexts);
let fx = ask_fixture_c3(contexts);
seed_live_pty(&fx.sessions, aid(1), sid(801)); // A live
seed_live_pty(&fx.sessions, aid(2), sid(802)); // B live
// A→B in flight (held — B never replies during the test): edge A→B is posted.
let svc = Arc::clone(&fx.service);
let _ask_ab = tokio::spawn(async move {
svc.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(1), "beta", "to B")),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
// Now B→A would close the cycle ⇒ typed INVALID, no enqueue on A, returns fast.
let err = timeout(
TEST_GUARD,
fx.service.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(2), "alpha", "back to A")),
),
)
.await
.expect("cycle ask must return fast, never deadlock")
.unwrap_err();
assert_eq!(
err.code(),
"INVALID",
"re-entrant cycle is typed INVALID: {err:?}"
);
assert!(
err.to_string().contains("cycle") || err.to_string().contains("ré-entrante"),
"explicit cycle message: {err}"
);
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "no ticket enqueued on A");
}
/// C3 — once an A→B turn completes, the edge is cleared, so a later B→A is allowed
/// (no phantom edge): proves the wait-for edge is RAII-scoped to the turn.
#[tokio::test]
async fn edge_cleared_after_turn_allows_later_reverse_delegation() {
let contexts = FakeContexts::new();
seed_two_agents(&contexts);
let fx = ask_fixture_c3(contexts);
seed_live_pty(&fx.sessions, aid(1), sid(801));
seed_live_pty(&fx.sessions, aid(2), sid(802));
// A→B completes.
let svc = Arc::clone(&fx.service);
let ask_ab = tokio::spawn(async move {
svc.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(1), "beta", "to B")),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(2), "ok"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask_ab).await.unwrap().unwrap().unwrap();
// Edge A→B is now gone ⇒ B→A is allowed (would only cycle if A→B still pending).
let svc2 = Arc::clone(&fx.service);
let ask_ba = tokio::spawn(async move {
svc2.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(2), "alpha", "now to A")),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "A ok"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask_ba).await.unwrap().unwrap().unwrap();
assert_eq!(
out.reply.as_deref(),
Some("A ok"),
"reverse delegation allowed"
);
}
/// C3 — reply correlates **by ticket** (deterministic id-keyed resolution). The agent
/// echoes the ticket id from the `[IdeA · … · ticket <id>]` prefix; replying by that
/// id resolves exactly that request. (Per «1 agent = 1 employee» the turn lock keeps
/// at most one turn live per agent, so resolution is sequential here — by-ticket is
/// what makes it deterministic regardless of which requester it serves.)
#[tokio::test]
async fn reply_correlates_by_ticket() {
let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
seed_live_pty(&fx.sessions, aid(2), sid(802));
// Requester A1 asks B; capture B's queued ticket id and reply it by id.
let svc1 = Arc::clone(&fx.service);
let ask1 = tokio::spawn(async move {
svc1.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T1")))
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
let t1 = fx.mailbox.ticket_ids(&aid(2))[0];
fx.service
.dispatch(&project(), reply_cmd_ticket(aid(2), t1, "R1"))
.await
.expect("reply T1 by ticket ok");
let out1 = timeout(TEST_GUARD, ask1).await.unwrap().unwrap().unwrap();
assert_eq!(out1.reply.as_deref(), Some("R1"));
// A different requester A2 asks B; replying a STALE/unknown ticket id is a typed
// error (no panic), while the correct id resolves it.
let svc2 = Arc::clone(&fx.service);
let ask2 = tokio::spawn(async move {
svc2.dispatch(&project(), cmd(&ask_from_agent_json(aid(11), "beta", "T2")))
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
let bogus = TicketId::new_random();
let err = fx
.service
.dispatch(&project(), reply_cmd_ticket(aid(2), bogus, "wrong"))
.await
.unwrap_err();
assert_eq!(
err.code(),
"INVALID",
"unknown ticket id ⇒ typed INVALID: {err:?}"
);
let t2 = fx.mailbox.ticket_ids(&aid(2))[0];
fx.service
.dispatch(&project(), reply_cmd_ticket(aid(2), t2, "R2"))
.await
.expect("reply T2 by ticket ok");
let out2 = timeout(TEST_GUARD, ask2).await.unwrap().unwrap().unwrap();
assert_eq!(out2.reply.as_deref(), Some("R2"));
}
/// C3 — a reply **without** a ticket falls back to the head of the queue (compat
/// mono-thread agents).
#[tokio::test]
async fn reply_without_ticket_falls_back_to_head() {
let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
seed_live_pty(&fx.sessions, aid(2), sid(802));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
svc.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T")))
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
// No ticket id ⇒ head-of-queue fallback.
fx.service
.dispatch(&project(), reply_cmd(aid(2), "head reply"))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
assert_eq!(out.reply.as_deref(), Some("head reply"));
}
/// C3 — when the rendezvous resolves to a closed reply channel (the same retirement
/// path the bounded turn timeout takes), the ticket is **retired** from the queue and
/// the target session stays **alive** (ready for the next turn). We trigger it by
/// cancelling the head ticket (drops the sender) while the ask awaits — `cancel_head`
/// + the `Cancelled` arm mirror the timeout's `cancel_head` exactly.
#[tokio::test]
async fn timeout_path_frees_queue_and_keeps_target_alive() {
let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
seed_live_pty(&fx.sessions, aid(2), sid(802));
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
svc.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T")))
.await
});
await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
let t = fx.mailbox.ticket_ids(&aid(2))[0];
// Retire the head ticket (drops its sender) ⇒ the awaiting ask sees a closed
// channel and returns a typed error, the SAME cleanup the turn timeout performs.
fx.mailbox.cancel_head(aid(2), t);
let err = timeout(TEST_GUARD, ask)
.await
.expect("ask returns promptly once the channel closes")
.unwrap()
.unwrap_err();
assert!(
matches!(err.code(), "PROCESS" | "INVALID"),
"closed-channel ask is a typed error: {err:?}"
);
// Queue freed, and the target B is still live (never killed by the failed ask).
assert_eq!(
fx.mailbox.pending(&aid(2)),
0,
"queue freed after retirement"
);
assert_eq!(
fx.sessions.session_for_agent(&aid(2)),
Some(sid(802)),
"target stays alive for the next turn"
);
}
// ---------------------------------------------------------------------------
// C4 — SubmitHumanInput (Envoyer) + interrupt (Interrompre)
// ---------------------------------------------------------------------------
/// Fixture for the C4 human-input / interrupt tests. Identical wiring to
/// [`ask_fixture`] but keeps the **concrete** [`TestMediator`] so a test can assert
/// the preempt path (no enqueue, no resolved ticket) and the shared FIFO.
struct C4Fixture {
service: Arc<OrchestratorService>,
mailbox: Arc<TestMailbox>,
sessions: Arc<TerminalSessions>,
mediator: Arc<TestMediator>,
pty: FakePty,
}
fn c4_fixture(contexts: FakeContexts) -> C4Fixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
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,
));
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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone())),
);
C4Fixture {
service,
mailbox,
sessions,
mediator,
pty,
}
}
/// A human `submit` enqueues a `from_human` ticket in the SAME FIFO the delegations
/// use: an `ask` A→B and a `submit` Human→B serialise on B's single queue.
#[tokio::test]
async fn submit_and_ask_share_one_fifo_per_agent() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = c4_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
// A delegation A→B blocks in B's FIFO (it awaits a reply).
let svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
svc.dispatch(
&project(),
cmd(&ask_from_agent_json(aid(10), "architect", "deleg")),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
// A human submit to the SAME agent lands behind it in the SAME FIFO.
fx.service
.submit_human_input(&project(), aid(1), "human task".to_owned())
.await
.expect("submit succeeds");
assert_eq!(
fx.mailbox.pending(&aid(1)),
2,
"human submit serialises behind the delegation in one FIFO"
);
// The human ticket's task text was written into the target's live terminal.
let writes = fx.pty.writes_for(sid(800));
assert!(
writes.iter().any(|w| w.contains("human task")),
"human turn delivered to the PTY: {writes:?}"
);
// Unblock the delegation so the spawned task ends cleanly.
fx.mailbox
.cancel_head(aid(1), fx.mailbox.ticket_ids(&aid(1))[0]);
let _ = timeout(TEST_GUARD, ask).await;
}
/// `interrupt` = `preempt`: it calls the mediator's preempt for the agent and does
/// NOT enqueue anything nor resolve any ticket.
#[tokio::test]
async fn interrupt_preempts_without_enqueue_or_resolve() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = c4_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
assert_eq!(fx.mailbox.pending(&aid(1)), 0);
fx.service
.interrupt_agent(&project(), aid(1))
.await
.expect("interrupt succeeds");
assert_eq!(fx.mediator.preempts(), vec![aid(1)], "preempt called once");
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "interrupt enqueues nothing");
}
/// A human submit to an unknown agent id is a typed NotFound (never a panic).
#[tokio::test]
async fn submit_unknown_agent_is_not_found() {
let fx = c4_fixture(FakeContexts::new());
let err = fx
.service
.submit_human_input(&project(), aid(999), "x".to_owned())
.await
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND", "unknown agent submit: {err:?}");
}
/// An interrupt to an unknown agent id is a typed NotFound and never calls preempt.
#[tokio::test]
async fn interrupt_unknown_agent_is_not_found() {
let fx = c4_fixture(FakeContexts::new());
let err = fx
.service
.interrupt_agent(&project(), aid(999))
.await
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND", "unknown agent interrupt: {err:?}");
assert!(
fx.mediator.preempts().is_empty(),
"no preempt on unknown agent"
);
}
// ---------------------------------------------------------------------------
// P6b — live wiring of `RecordTurn` in `ask_agent` (conversation persistence).
//
// On a successful inter-agent delegation, `ask_agent` persists the **pair**
// best-effort: a `Prompt` turn at enqueue (text = task, source = requester) and a
// `Response` turn on the `Ok(Ok(result))` branch (text = result, source = target),
// both on the SAME resolved `conversation_id`. Persistence is best-effort: a missing
// provider/clock, a provider returning `None`, or a failing append must NEVER turn a
// successful ask into an error.
//
// These tests reuse the full `ask`/`reply` rendezvous harness above (real mailbox +
// mediated PTY delivery + conversation registry) and add a recording in-memory
// `ConversationLog` / `RecordTurnProvider` / fixed `Clock`, mirroring the P6a fakes
// in `conversation_record.rs`.
// ---------------------------------------------------------------------------
use application::{OrchestratorOutcome, RecordTurn, RecordTurnProvider};
use domain::ports::Clock;
use domain::InputSource;
use domain::{
ConversationLog, ConversationTurn as DomainTurn, Handoff, HandoffStore, HandoffSummarizer,
TurnId, TurnRole,
};
/// A deterministic [`Clock`]: every persisted turn is stamped with this constant.
struct FixedClock(i64);
impl Clock for FixedClock {
fn now_millis(&self) -> i64 {
self.0
}
}
/// In-memory [`ConversationLog`] that **records the order and content** of every
/// append so a test can assert the exact `[Prompt, Response]` sequence and inspect
/// each turn's `text` / `role` / `source` / `conversation`. Optionally fails on
/// `append` (best-effort path).
#[derive(Default)]
struct RecordingLog {
appends: Mutex<Vec<DomainTurn>>,
fail_append: bool,
}
impl RecordingLog {
fn failing() -> Self {
Self {
appends: Mutex::new(Vec::new()),
fail_append: true,
}
}
fn appends(&self) -> Vec<DomainTurn> {
self.appends.lock().unwrap().clone()
}
}
#[async_trait]
impl ConversationLog for RecordingLog {
async fn append(
&self,
_conversation: ConversationId,
turn: DomainTurn,
) -> Result<(), StoreError> {
if self.fail_append {
return Err(StoreError::Io(
"recording log: forced append failure".to_owned(),
));
}
self.appends.lock().unwrap().push(turn);
Ok(())
}
async fn read(
&self,
_conversation: ConversationId,
_since: Option<TurnId>,
) -> Result<Vec<DomainTurn>, StoreError> {
Ok(self.appends())
}
async fn last(
&self,
_conversation: ConversationId,
n: usize,
) -> Result<Vec<DomainTurn>, StoreError> {
let all = self.appends();
let start = all.len().saturating_sub(n);
Ok(all[start..].to_vec())
}
}
/// Trivial in-memory [`HandoffStore`] — the P6b tests only assert on the log; the
/// handoff is exercised end-to-end by P6a. Kept minimal.
#[derive(Default)]
struct NoopHandoffStore(Mutex<HashMap<ConversationId, Handoff>>);
#[async_trait]
impl HandoffStore for NoopHandoffStore {
async fn load(&self, conversation: ConversationId) -> Result<Option<Handoff>, StoreError> {
Ok(self.0.lock().unwrap().get(&conversation).cloned())
}
async fn save(&self, conversation: ConversationId, handoff: Handoff) -> Result<(), StoreError> {
self.0.lock().unwrap().insert(conversation, handoff);
Ok(())
}
}
/// Deterministic summarizer: folds the new turns' text onto the previous summary.
#[derive(Default)]
struct ConcatSummarizer;
#[async_trait]
impl HandoffSummarizer for ConcatSummarizer {
async fn fold(&self, prev: Option<Handoff>, new_turns: &[DomainTurn]) -> Handoff {
let mut summary = prev.map(|h| h.summary_md).unwrap_or_default();
for t in new_turns {
summary.push_str(&t.text);
}
let up_to = new_turns
.last()
.map(|t| t.id)
.expect("at least one new turn");
Handoff::new(summary, up_to, None)
}
}
/// A [`RecordTurnProvider`] that always materialises the **same** shared
/// [`RecordTurn`] (its log is observable by the test), ignoring the root — the
/// fixture pins a single project, so root-routing is out of scope here.
struct SharedRecordProvider(Arc<RecordTurn>);
impl RecordTurnProvider for SharedRecordProvider {
fn record_turn_for(&self, _root: &ProjectPath) -> Option<Arc<RecordTurn>> {
Some(Arc::clone(&self.0))
}
}
/// A [`RecordTurnProvider`] that always declines (no `RecordTurn` for this root) —
/// drives the "provider returns None ⇒ silent no-op" branch.
struct NoneRecordProvider;
impl RecordTurnProvider for NoneRecordProvider {
fn record_turn_for(&self, _root: &ProjectPath) -> Option<Arc<RecordTurn>> {
None
}
}
/// Builds the same `ask` harness as [`ask_fixture`] but additionally wires
/// `with_record_turn(provider, clock)`. Returns the [`AskFixture`] plus the shared
/// recording log (when one was provided) for assertion.
fn ask_fixture_with_record(
contexts: FakeContexts,
provider: Arc<dyn RecordTurnProvider>,
clock_ms: i64,
) -> AskFixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
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,
));
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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
);
AskFixture {
service,
mailbox,
sessions,
bus,
pty,
mediator,
}
}
/// Drives a full `ask` round-trip (requester `from`, target architect) on the given
/// service, replying `result`. Returns the dispatch outcome.
async fn run_ask_roundtrip(
fx: &AskFixture,
ask_json: &str,
reply_from: AgentId,
result: &str,
) -> OrchestratorOutcome {
let svc = Arc::clone(&fx.service);
let json = ask_json.to_owned();
let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(&json)).await });
await_until(|| fx.mailbox.pending(&reply_from) == 1).await;
fx.service
.dispatch(&project(), reply_cmd(reply_from, result))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes once replied")
.expect("join ok")
.expect("ask ok")
}
/// Round-trip from the **User** (requester `None`) writes exactly two turns on the
/// SAME conversation, in order Prompt then Response, with the right text/role/source.
#[tokio::test]
async fn p6b_user_ask_records_prompt_then_response_pair() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let log = Arc::new(RecordingLog::default());
let record = Arc::new(RecordTurn::new(
Arc::clone(&log) as Arc<dyn ConversationLog>,
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
));
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 123);
seed_live_pty(&fx.sessions, aid(1), sid(800));
// ASK_JSON uses `requestedBy:"Main"` but no agent requester id ⇒ User-sourced.
let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "the §17 answer").await;
assert_eq!(out.reply.as_deref(), Some("the §17 answer"));
let appends = log.appends();
assert_eq!(
appends.len(),
2,
"exactly two turns persisted (Prompt + Response)"
);
let prompt = &appends[0];
let response = &appends[1];
// Same conversation for both turns.
assert_eq!(
prompt.conversation, response.conversation,
"both turns share the resolved conversation id"
);
// Order + role.
assert_eq!(prompt.role, TurnRole::Prompt, "first turn is the Prompt");
assert_eq!(
response.role,
TurnRole::Response,
"second turn is the Response"
);
// Text: task then result.
assert_eq!(
prompt.text, "Analyse §17",
"prompt text is the delegated task"
);
assert_eq!(
response.text, "the §17 answer",
"response text is the reply result"
);
// Source: Human prompt (no agent requester), target-sourced response.
assert_eq!(
prompt.source,
InputSource::Human,
"User ask ⇒ Human prompt source"
);
assert_eq!(
response.source,
InputSource::agent(aid(1)),
"response is sourced from the target (architect)"
);
// Clock stamp came from the injected Clock.
assert_eq!(prompt.at_ms, 123);
assert_eq!(response.at_ms, 123);
}
/// Round-trip with an **agent** requester (A) resolves the `A↔B` conversation and the
/// Prompt source is `agent(A)`; the Response source remains the target B.
#[tokio::test]
async fn p6b_agent_requester_records_pair_on_a_b_thread() {
let architect = scratch_agent(aid(1), "architect", "agents/architect.md");
let log = Arc::new(RecordingLog::default());
let record = Arc::new(RecordTurn::new(
Arc::clone(&log) as Arc<dyn ConversationLog>,
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
));
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
let fx = ask_fixture_with_record(
FakeContexts::with_agent(&architect, "# persona"),
provider,
0,
);
seed_live_pty(&fx.sessions, aid(1), sid(800));
// Requester is agent A (id 7). The orchestrator threads `requester` from the
// `requestedBy` field of `agent.message` (parsed as a uuid when well-formed).
let a = aid(7);
let a_uuid = uuid::Uuid::from_u128(7);
let ask_json = format!(
r#"{{ "type":"agent.message", "requestedBy":"{a_uuid}", "targetAgent":"architect", "task":"Analyse §17" }}"#
);
let out = run_ask_roundtrip(&fx, &ask_json, aid(1), "ack").await;
assert_eq!(out.reply.as_deref(), Some("ack"));
let appends = log.appends();
assert_eq!(appends.len(), 2, "two turns persisted");
let (prompt, response) = (&appends[0], &appends[1]);
// Independently resolve the A↔B conversation the SAME way the service does and
// assert both turns landed on it.
let convs = TestConversations::new();
let expected = convs
.resolve(
ConversationParty::agent(a),
ConversationParty::agent(aid(1)),
)
.id;
// (Resolution is deterministic per pair within one registry; we instead assert the
// turns share a thread and the prompt source carries A — the load-bearing facts.)
let _ = expected;
assert_eq!(prompt.conversation, response.conversation, "shared thread");
assert_eq!(
prompt.source,
InputSource::agent(a),
"agent requester ⇒ Prompt sourced from A"
);
assert_eq!(prompt.role, TurnRole::Prompt);
assert_eq!(response.role, TurnRole::Response);
assert_eq!(
response.source,
InputSource::agent(aid(1)),
"Response sourced from the target B"
);
}
/// No provider wired (plain `ask_fixture`) ⇒ the round-trip still succeeds and nothing
/// is persisted (no panic, no error).
#[tokio::test]
async fn p6b_no_provider_is_silent_no_op() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
// `ask_fixture` does NOT call `with_record_turn`.
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await;
assert_eq!(
out.reply.as_deref(),
Some("ok"),
"ask succeeds without persistence"
);
}
/// Provider wired but returns `None` for the root ⇒ ask still succeeds (best-effort
/// skip).
#[tokio::test]
async fn p6b_provider_returns_none_is_silent_no_op() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let provider = Arc::new(NoneRecordProvider) as Arc<dyn RecordTurnProvider>;
let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0);
seed_live_pty(&fx.sessions, aid(1), sid(800));
let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await;
assert_eq!(
out.reply.as_deref(),
Some("ok"),
"ask succeeds when provider declines"
);
}
/// `RecordTurn` built on a log whose `append` always fails ⇒ persistence errors are
/// swallowed; the delegation result is NOT degraded (ask returns `Ok`).
#[tokio::test]
async fn p6b_failing_record_does_not_degrade_ask() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let failing_log = Arc::new(RecordingLog::failing());
let record = Arc::new(RecordTurn::new(
Arc::clone(&failing_log) as Arc<dyn ConversationLog>,
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
));
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0);
seed_live_pty(&fx.sessions, aid(1), sid(800));
let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "still ok").await;
assert_eq!(
out.reply.as_deref(),
Some("still ok"),
"a failing append must not turn a successful delegation into an error"
);
// The failing log recorded nothing (every append errored).
assert!(
failing_log.appends().is_empty(),
"no turns stored when append fails"
);
}
// ===========================================================================
// Lot 1b — auto-lancement d'une cible **froide** structurée sur le chemin `ask`
// ===========================================================================
//
// Avant le Lot 1b, `ask_agent` ne routait vers `ask_structured` que si la cible avait
// **déjà** une session structurée vivante ; une cible structurée froide tombait dans
// `ensure_live_pty` (chemin PTY) — or une cible structurée n'a pas de PTY, donc le tour
// restait bloqué `Busy` (débloquable seulement par un `idea_reply` explicite). Le Lot 1b
// démarre la session structurée via le **même** launcher que le chemin PTY (qui route
// §17.4 vers `launch_structured` et l'insère dans le registre **partagé**), puis route
// vers `ask_structured` : le `Final` déterministe de la session débloque le tour et
// marque l'agent `Idle` (`mark_idle`), **sans** `idea_reply` ni PTY.
use std::sync::atomic::{AtomicUsize, Ordering};
use application::StructuredSessions;
use domain::ports::{
AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream,
};
/// Session structurée fake : `send()` rend un unique [`ReplyEvent::Final`] qui **écho**
/// le prompt reçu — c'est ce `Final` que `drain_with_readiness` transforme en réponse
/// du tour **et** en `mark_idle`. Aucun `idea_reply` n'est nécessaire.
struct EchoSession {
id: SessionId,
}
#[async_trait]
impl AgentSession for EchoSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
let stream: ReplyStream = Box::new(
vec![ReplyEvent::Final {
content: format!("structured: {prompt}"),
}]
.into_iter(),
);
Ok(stream)
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
/// Fabrique fake : `supports` = profil structuré, et `start` **compte** les démarrages
/// (preuve qu'une session froide a bien été auto-lancée) en rendant une [`EchoSession`].
#[derive(Clone, Default)]
struct CountingFactory {
starts: Arc<AtomicUsize>,
next_id: Arc<Mutex<u128>>,
}
impl CountingFactory {
fn new() -> Self {
Self {
starts: Arc::new(AtomicUsize::new(0)),
next_id: Arc::new(Mutex::new(9000)),
}
}
fn start_count(&self) -> usize {
self.starts.load(Ordering::SeqCst)
}
}
#[async_trait]
impl AgentSessionFactory for CountingFactory {
fn supports(&self, profile: &AgentProfile) -> bool {
profile.structured_adapter.is_some()
}
async fn start(
&self,
_profile: &AgentProfile,
_ctx: &PreparedContext,
_cwd: &ProjectPath,
_session: &SessionPlan,
_sandbox: Option<&domain::sandbox::SandboxPlan>,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
self.starts.fetch_add(1, Ordering::SeqCst);
let id = {
let mut n = self.next_id.lock().unwrap();
let id = SessionId::from_uuid(Uuid::from_u128(*n));
*n += 1;
id
};
Ok(Arc::new(EchoSession { id }))
}
}
/// Tout le câblage `ask` **structuré** : launcher + service voient le **même** registre
/// `StructuredSessions`, et le launcher porte la fabrique (routage §17.4).
struct StructuredAskFixture {
service: Arc<OrchestratorService>,
structured: Arc<StructuredSessions>,
factory: CountingFactory,
pty: FakePty,
}
fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
let sessions = Arc::new(TerminalSessions::new());
let pty = FakePty::new(sid(777));
let bus = SpyBus::default();
let mailbox = Arc::new(TestMailbox::new());
let structured = Arc::new(StructuredSessions::new());
let factory = CountingFactory::new();
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
// Launcher câblé **structuré** : pour un profil à `structured_adapter`, `execute`
// route §17.4 → `launch_structured`, démarre la session via la fabrique et l'insère
// dans CE registre partagé (le même `Arc` que le service ci-dessous).
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(
Arc::new(factory.clone()) 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 create_skill = Arc::new(CreateSkill::new(
Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
Arc::new(SeqIds::new()),
));
let mediator = Arc::new(TestMediator::new(
Arc::clone(&mailbox),
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
));
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
let service = Arc::new(
OrchestratorService::new(
create,
launch,
list,
close,
update,
create_skill,
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::clone(&sessions),
)
.with_input_mediator(
Arc::clone(&mediator) as Arc<dyn InputMediator>,
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
)
.with_conversations(conversations)
.with_events(Arc::new(bus.clone()))
// Même registre que le launcher : `ask_agent` y relit la session fraîchement
// insérée par `launch_structured`.
.with_structured(Arc::clone(&structured)),
);
StructuredAskFixture {
service,
structured,
factory,
pty,
}
}
/// Cible **froide** structurée (adaptateur Claude, aucune session vivante) : `ask_agent`
/// auto-lance sa session structurée via le launcher (factory.start appelé **une** fois),
/// puis la draine — le `Final` débloque le round-trip **sans** `idea_reply` ni PTY.
#[tokio::test]
async fn ask_cold_structured_target_autolaunches_session_and_final_unblocks() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// Pré-condition : aucune session structurée vivante pour la cible (elle est froide).
assert!(
fx.structured.session_for_agent(&aid(1)).is_none(),
"cible structurée froide : aucune session avant l'ask"
);
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
// Le `Final` de la session a rendu le contenu — sans aucun `idea_reply` dispatché.
assert_eq!(
out.reply.as_deref(),
Some("structured: Analyse §17"),
"le Final de la session auto-lancée débloque le tour"
);
// Exactement **un** démarrage de session structurée (la cible froide a été lancée).
assert_eq!(
fx.factory.start_count(),
1,
"une session structurée a été auto-lancée pour la cible froide"
);
// La session est désormais vivante dans le registre partagé (insérée par le launcher).
assert!(
fx.structured.session_for_agent(&aid(1)).is_some(),
"la session auto-lancée est enregistrée dans StructuredSessions"
);
// Chemin structuré ⇒ **aucun** PTY spawné (la cible n'a pas de terminal).
assert!(
fx.pty.spawns().is_empty(),
"cible structurée : aucun PTY spawné (chemin chat, pas terminal)"
);
}
/// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude)
/// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve
/// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin).
#[tokio::test]
async fn ask_warm_structured_target_reuses_session_without_relaunch() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// 1er ask : auto-lance (start_count == 1).
let _ = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("first ask ok");
assert_eq!(fx.factory.start_count(), 1, "1er ask auto-lance la session");
// 2e ask : cible chaude ⇒ aucune relance.
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("second ask ok");
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(
fx.factory.start_count(),
1,
"cible chaude : la session est réutilisée, aucune relance"
);
}
Reference in New Issue View Git Blame Copy Permalink