Files
IdeA/crates/application/tests/orchestrator_service.rs
Blomios 62b71776f5 fix(background-task): découpler la détection de fin de tâche du drain output PTY (#2)
Le runner de tâches de fond détectait la fin d'un process via l'EOF du drain de
sortie PTY, couplant le cycle de vie du process au flux d'output. Un output
encore ouvert (ou drainé ailleurs) pouvait masquer ou retarder la détection de
fin.

Ajoute wait/try_wait au port figé PtyPort :
- wait : bloquant, rend un ExitStatus idempotent ;
- try_wait : non bloquant, rend Option<ExitStatus> ;
- exit status mémorisé pour garder wait/try_wait/kill cohérents.
Implémentation dans PortablePtyAdapter (état d'exit mémorisé). Le
CommandBackgroundRunner détecte désormais la fin via pty.wait (select sur
cancel/deadline) au lieu de l'EOF du drain. Tous les fakes PtyPort du workspace
sont complétés en conséquence.

Le tee live UI reste hors périmètre (traité en #58). Aucun breaking IPC/front.

Tests : infrastructure/tests/background_task_runner.rs (nouveau) + pty_adapter.rs
verts, non-régression application/app-tauri OK (hors échecs réseau du sandbox).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-14 23:42:40 +02:00

4491 lines
157 KiB
Rust

//! 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, BackgroundCompletionStream, BackgroundTaskHandle,
BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore,
ContextInjectionPlan, DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError,
IdGenerator, MemoryError, MemoryStore, 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::{
BackgroundTask, BackgroundTaskState, BackgroundTaskWakePolicy, OrchestratorCommand,
OrchestratorRequest, PtySize, SessionId, TaskId,
};
use domain::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug};
use uuid::Uuid;
use application::{
CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, HarvestMemoryFromTurn,
LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, OrchestratorService,
SessionLimitService, SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext,
UpdateLiveState,
};
use domain::live_state::{LiveEntry, LiveState, WorkStatus};
use domain::ports::LiveStateStore;
// ---------------------------------------------------------------------------
// 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(),
orchestrator: None,
},
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;
#[async_trait]
impl AgentRuntime for FakeRuntime {
async 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 wait(&self, _handle: &PtyHandle) -> Result<ExitStatus, PtyError> {
Ok(ExitStatus { code: Some(0) })
}
fn try_wait(&self, _handle: &PtyHandle) -> Result<Option<ExitStatus>, PtyError> {
Ok(Some(ExitStatus { code: Some(0) }))
}
async fn kill(&self, handle: &PtyHandle) -> Result<ExitStatus, PtyError> {
self.kills.lock().unwrap().push(handle.session_id);
Ok(ExitStatus { code: Some(0) })
}
}
#[derive(Clone, Default)]
struct RecordingBackgroundStore {
tasks: Arc<Mutex<HashMap<TaskId, BackgroundTask>>>,
}
impl RecordingBackgroundStore {
fn task(&self, id: TaskId) -> Option<BackgroundTask> {
self.tasks.lock().unwrap().get(&id).cloned()
}
}
#[async_trait]
impl BackgroundTaskStore for RecordingBackgroundStore {
async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
let mut tasks = self.tasks.lock().unwrap();
if tasks.contains_key(&task.id) {
return Err(BackgroundTaskPortError::AlreadyExists);
}
tasks.insert(task.id, task.clone());
Ok(())
}
async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
Ok(self.task(id))
}
async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
self.tasks.lock().unwrap().insert(task.id, task.clone());
Ok(())
}
async fn list_open_for_agent(
&self,
agent_id: AgentId,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
Ok(self
.tasks
.lock()
.unwrap()
.values()
.filter(|task| task.owner_agent_id == agent_id && !task.is_terminal())
.cloned()
.collect())
}
async fn list_undelivered_completions(
&self,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
Ok(self
.tasks
.lock()
.unwrap()
.values()
.filter(|task| task.has_pending_completion_delivery())
.cloned()
.collect())
}
async fn mark_completion_delivered(
&self,
task_id: TaskId,
) -> Result<(), BackgroundTaskPortError> {
let task = self
.task(task_id)
.ok_or(BackgroundTaskPortError::NotFound)?
.mark_completion_delivered()
.map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?;
self.save(&task).await
}
}
#[derive(Clone, Default)]
struct RecordingBackgroundRunner {
specs: Arc<Mutex<Vec<BackgroundTaskSpec>>>,
}
impl RecordingBackgroundRunner {
fn specs(&self) -> Vec<BackgroundTaskSpec> {
self.specs.lock().unwrap().clone()
}
}
#[async_trait]
impl BackgroundTaskRunner for RecordingBackgroundRunner {
async fn spawn(
&self,
spec: BackgroundTaskSpec,
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
let task_id = spec.task_id;
self.specs.lock().unwrap().push(spec);
Ok(BackgroundTaskHandle { task_id })
}
async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
Ok(())
}
fn subscribe_completions(&self) -> BackgroundCompletionStream {
Box::new(std::iter::empty())
}
}
#[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())
}
}
/// In-memory [`MemoryStore`] for the auto-memory harvest hook (Lot E1): records the
/// saved slugs, with an optional forced `save` failure to prove harvest errors stay
/// best-effort (never break the reply).
#[derive(Default)]
struct HarvestMemories {
saved: Mutex<Vec<MemorySlug>>,
fail_save: bool,
}
impl HarvestMemories {
fn slugs(&self) -> Vec<String> {
self.saved
.lock()
.unwrap()
.iter()
.map(|s| s.as_str().to_owned())
.collect()
}
}
#[async_trait]
impl MemoryStore for HarvestMemories {
async fn list(&self, _root: &ProjectPath) -> Result<Vec<Memory>, MemoryError> {
Ok(Vec::new())
}
async fn get(&self, _root: &ProjectPath, _slug: &MemorySlug) -> Result<Memory, MemoryError> {
Err(MemoryError::NotFound)
}
async fn save(&self, _root: &ProjectPath, memory: &Memory) -> Result<(), MemoryError> {
if self.fail_save {
return Err(MemoryError::Frontmatter("forced failure".to_owned()));
}
self.saved.lock().unwrap().push(memory.slug().clone());
Ok(())
}
async fn delete(&self, _root: &ProjectPath, _slug: &MemorySlug) -> Result<(), MemoryError> {
Ok(())
}
async fn read_index(&self, _root: &ProjectPath) -> Result<Vec<MemoryIndexEntry>, MemoryError> {
Ok(Vec::new())
}
async fn resolve_links(
&self,
_root: &ProjectPath,
_slug: &MemorySlug,
) -> Result<Vec<MemoryLink>, MemoryError> {
Ok(Vec::new())
}
}
/// In-memory [`LiveStateStore`] for the live-state auto-update hook (lot LS3): keeps a
/// keyed last-writer-wins [`LiveState`] and counts upserts (to prove no write-storm),
/// with an optional forced failure to prove the auto-update stays best-effort.
#[derive(Default)]
struct RecordingLiveState {
state: Mutex<LiveState>,
upserts: Mutex<usize>,
fail: bool,
}
impl RecordingLiveState {
fn entry_for(&self, agent: AgentId) -> Option<LiveEntry> {
self.state
.lock()
.unwrap()
.entries
.iter()
.find(|e| e.agent_id == agent)
.cloned()
}
fn upsert_count(&self) -> usize {
*self.upserts.lock().unwrap()
}
}
#[async_trait]
impl LiveStateStore for RecordingLiveState {
async fn load(&self) -> Result<LiveState, StoreError> {
if self.fail {
return Err(StoreError::Io("forced failure".to_owned()));
}
Ok(self.state.lock().unwrap().clone())
}
async fn upsert(&self, entry: LiveEntry) -> Result<(), StoreError> {
if self.fail {
return Err(StoreError::Io("forced failure".to_owned()));
}
*self.upserts.lock().unwrap() += 1;
self.state.lock().unwrap().upsert(entry);
Ok(())
}
async fn prune(&self, now_ms: u64, ttl_ms: u64, max_n: usize) -> Result<(), StoreError> {
if self.fail {
return Err(StoreError::Io("forced failure".to_owned()));
}
self.state.lock().unwrap().prune(now_ms, ttl_ms, max_n);
Ok(())
}
}
/// A [`LiveStateProvider`] that always yields the **same** [`UpdateLiveState`] over a
/// shared recording store (root-agnostic for the test).
struct TestLiveStateProvider {
update: Arc<UpdateLiveState>,
}
impl LiveStateProvider for TestLiveStateProvider {
fn live_state_for(&self, _root: &ProjectPath) -> Option<Arc<UpdateLiveState>> {
Some(Arc::clone(&self.update))
}
}
/// A [`LiveStateReadProvider`] (lot LS4, `idea_workstate_read`) yielding the same
/// [`GetLiveStateLean`] over a shared recording store (root-agnostic for the test).
struct TestLiveStateReadProvider {
getter: Arc<GetLiveStateLean>,
}
impl LiveStateReadProvider for TestLiveStateReadProvider {
fn live_state_lean_for(&self, _root: &ProjectPath) -> Option<Arc<GetLiveStateLean>> {
Some(Arc::clone(&self.getter))
}
}
/// Fixed millis clock for deterministic `updated_at_ms`.
struct FixedMillisClock(i64);
impl Clock for FixedMillisClock {
fn now_millis(&self) -> i64 {
self.0
}
}
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 run_in_background_dispatch_creates_running_command_for_requester() {
let fx = fixture(FakeContexts::new());
let store = Arc::new(RecordingBackgroundStore::default());
let runner = Arc::new(RecordingBackgroundRunner::default());
let spawn = Arc::new(SpawnBackgroundCommand::new(
Arc::clone(&store) as Arc<dyn BackgroundTaskStore>,
Arc::clone(&runner) as Arc<dyn BackgroundTaskRunner>,
Arc::new(FixedMillisClock(42)) as Arc<dyn domain::ports::Clock>,
Arc::new(SeqIds::new()) as Arc<dyn IdGenerator>,
));
let service = fx.service.with_spawn_background_command(spawn);
let owner = aid(77);
let out = service
.dispatch(
&project(),
OrchestratorCommand::RunInBackground {
owner,
label: "B8 T1".to_owned(),
command: "bash".to_owned(),
args: vec![
"-lc".to_owned(),
"sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
],
cwd: Some("crates".to_owned()),
deadline_ms: None,
},
)
.await
.expect("dispatch ok");
let payload: serde_json::Value = serde_json::from_str(&out.detail).unwrap();
assert_eq!(payload["state"], "Running");
let task_id =
TaskId::from_uuid(uuid::Uuid::parse_str(payload["taskId"].as_str().unwrap()).unwrap());
let task = store.task(task_id).expect("task persisted");
assert_eq!(task.project_id, project().id);
assert_eq!(task.owner_agent_id, owner);
assert_eq!(task.state, BackgroundTaskState::Running);
assert_eq!(task.wake_policy, BackgroundTaskWakePolicy::WakeOwner);
assert_eq!(
task.kind,
domain::BackgroundTaskKind::Command {
label: "B8 T1".to_owned(),
}
);
let specs = runner.specs();
assert_eq!(specs.len(), 1);
assert_eq!(specs[0].task_id, task_id);
assert_eq!(specs[0].owner_agent_id, owner);
assert_eq!(specs[0].wake_policy, BackgroundTaskWakePolicy::WakeOwner);
let command = specs[0].command.as_ref().expect("command spec");
assert_eq!(command.command, "bash");
assert_eq!(
command.args,
vec![
"-lc".to_owned(),
"sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
]
);
assert_eq!(command.cwd.as_str(), "/home/me/proj/crates");
}
#[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, TurnResolution};
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<TurnResolution>)>>>,
completions: Arc<CompletionBus>,
}
impl TestMailbox {
fn new() -> Self {
Self {
queues: Mutex::new(HashMap::new()),
completions: Arc::new(CompletionBus::default()),
}
}
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()
}
fn completions(&self) -> Arc<CompletionBus> {
Arc::clone(&self.completions)
}
}
#[derive(Clone)]
enum TestCompletion {
Replied(String),
NoReply,
Cancelled,
}
#[derive(Default)]
struct CompletionBus {
by_agent: Mutex<HashMap<AgentId, VecDeque<TestCompletion>>>,
notify: tokio::sync::Notify,
}
impl CompletionBus {
fn push(&self, agent: AgentId, completion: TestCompletion) {
self.by_agent
.lock()
.unwrap()
.entry(agent)
.or_default()
.push_back(completion);
self.notify.notify_waiters();
}
async fn next(&self, agent: AgentId) -> TestCompletion {
loop {
if let Some(item) = self
.by_agent
.lock()
.unwrap()
.get_mut(&agent)
.and_then(VecDeque::pop_front)
{
return item;
}
self.notify.notified().await;
}
}
}
impl AgentMailbox for TestMailbox {
fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
let (tx, rx) = tokio::sync::oneshot::channel::<TurnResolution>();
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")
};
self.completions
.push(agent, TestCompletion::Replied(result.clone()));
let _ = slot.1.send(TurnResolution::Replied(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")
};
self.completions
.push(agent, TestCompletion::Replied(result.clone()));
let _ = slot.1.send(TurnResolution::Replied(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();
self.completions.push(agent, TestCompletion::Cancelled);
}
}
}
fn complete_without_reply(&self, agent: AgentId, ticket_id: TicketId) {
// Mirror the production adapter: head-only, idempotent, fire-and-forget-safe.
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) {
return;
}
if queue.front().is_some_and(|(_, tx)| tx.is_closed()) {
return; // receiver gone (human submit / timed-out caller): preserve head.
}
let (_, tx) = queue.pop_front().expect("head just matched");
self.completions.push(agent, TestCompletion::NoReply);
let _ = tx.send(TurnResolution::ReturnedToPromptNoReply);
}
}
}
/// 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>,
/// The auto-memory store wired into the harvest hook (Lot E1): lets a test
/// assert which notes a successful reply persisted.
memories: Arc<HarvestMemories>,
/// The live-state store wired into the auto-update hook (lot LS3): lets a test
/// assert the target's Working/Done transitions and the upsert count.
live: Arc<RecordingLiveState>,
structured: Arc<StructuredSessions>,
}
struct CompletionSession {
id: SessionId,
agent: AgentId,
completions: Arc<CompletionBus>,
}
#[async_trait]
impl AgentSession for CompletionSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
None
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
match self.completions.next(self.agent).await {
TestCompletion::Replied(content) => {
Ok(Box::new(vec![ReplyEvent::Final { content }].into_iter()))
}
TestCompletion::NoReply => Err(AgentSessionError::Io(
"target returned without a structured final".to_owned(),
)),
TestCompletion::Cancelled => Err(AgentSessionError::Io(
"structured test turn cancelled".to_owned(),
)),
}
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
#[derive(Clone)]
struct CompletionFactory {
completions: Arc<CompletionBus>,
agents_by_path: HashMap<String, AgentId>,
next_id: Arc<Mutex<u128>>,
}
impl CompletionFactory {
fn new(completions: Arc<CompletionBus>, agents_by_path: HashMap<String, AgentId>) -> Self {
Self {
completions,
agents_by_path,
next_id: Arc::new(Mutex::new(7000)),
}
}
}
#[async_trait]
impl AgentSessionFactory for CompletionFactory {
fn supports(&self, profile: &AgentProfile) -> bool {
profile.structured_adapter.is_some()
}
async fn start(
&self,
_profile: &AgentProfile,
ctx: &PreparedContext,
_cwd: &ProjectPath,
_session: &SessionPlan,
_env: &[(String, String)],
_sandbox: Option<&domain::sandbox::SandboxPlan>,
) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
let id = {
let mut n = self.next_id.lock().unwrap();
let id = SessionId::from_uuid(Uuid::from_u128(*n));
*n += 1;
id
};
let agent = self
.agents_by_path
.get(&ctx.relative_path)
.copied()
.ok_or_else(|| {
AgentSessionError::Start(format!(
"test completion session cannot resolve agent for {}",
ctx.relative_path
))
})?;
Ok(Arc::new(CompletionSession {
id,
agent,
completions: Arc::clone(&self.completions),
}))
}
}
fn completion_factory_for_contexts(
contexts: &FakeContexts,
completions: Arc<CompletionBus>,
) -> CompletionFactory {
let agents_by_path: HashMap<String, AgentId> = contexts
.manifest()
.entries
.iter()
.map(|e| (e.md_path.clone(), e.agent_id))
.collect();
CompletionFactory::new(completions, agents_by_path)
}
fn seed_structured_sessions(
structured: &StructuredSessions,
contexts: &FakeContexts,
completions: Arc<CompletionBus>,
) {
for (idx, entry) in contexts.manifest().entries.iter().enumerate() {
let session = Arc::new(CompletionSession {
id: SessionId::from_uuid(Uuid::from_u128(8_000 + idx as u128)),
agent: entry.agent_id,
completions: Arc::clone(&completions),
}) as Arc<dyn AgentSession>;
structured.insert(session, entry.agent_id, nid(8_000 + idx as u128));
}
}
/// 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 {
ask_fixture_full(contexts, false, false, true)
}
/// Like [`ask_fixture`] but lets a test force the auto-memory store to fail its
/// `save`, proving the harvest stays best-effort (the reply still succeeds).
fn ask_fixture_with_memory(contexts: FakeContexts, fail_memory: bool) -> AskFixture {
ask_fixture_full(contexts, fail_memory, false, true)
}
/// Like [`ask_fixture`] but lets a test force the live-state store to fail, proving
/// the auto-update stays best-effort (the ask/reply still succeeds).
fn ask_fixture_with_live(contexts: FakeContexts, fail_live: bool) -> AskFixture {
ask_fixture_full(contexts, false, fail_live, true)
}
/// Like [`ask_fixture`] but leaves the live-state auto-update **unwired** (`None`),
/// to prove no write happens when the hook is absent (zéro régression).
fn ask_fixture_without_live(contexts: FakeContexts) -> AskFixture {
ask_fixture_full(contexts, false, false, false)
}
/// Full builder: wires the auto-memory harvest (Lot E1) and, when `wire_live`, the
/// live-state auto-update (lot LS3), each with an optional forced failure.
fn ask_fixture_full(
contexts: FakeContexts,
fail_memory: bool,
fail_live: bool,
wire_live: bool,
) -> 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 structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
seed_structured_sessions(&structured, &contexts, mailbox.completions());
let memories = Arc::new(HarvestMemories {
saved: Mutex::new(Vec::new()),
fail_save: fail_memory,
});
let live = Arc::new(RecordingLiveState {
fail: fail_live,
..Default::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,
)
.with_structured(
Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let 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 mut builder = 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_structured(Arc::clone(&structured))
.with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
Arc::clone(&memories) as Arc<dyn MemoryStore>,
Arc::new(bus.clone()),
)));
if wire_live {
builder = builder.with_live_state(Arc::new(TestLiveStateProvider {
update: Arc::new(UpdateLiveState::new(
Arc::clone(&live) as Arc<dyn LiveStateStore>,
Arc::new(FixedMillisClock(1_234)) as Arc<dyn Clock>,
)),
}) as Arc<dyn LiveStateProvider>);
// LS4 read side over the SAME store, so `idea_workstate_set` then
// `idea_workstate_read` round-trips (no separate write provider would).
builder = builder.with_live_state_read(Arc::new(TestLiveStateReadProvider {
getter: Arc::new(GetLiveStateLean::new(
Arc::clone(&live) as Arc<dyn LiveStateStore>,
Arc::new(FixedMillisClock(1_234)) as Arc<dyn Clock>,
)),
}) as Arc<dyn LiveStateReadProvider>);
}
let service = Arc::new(builder);
AskFixture {
service,
mailbox,
sessions,
bus,
pty,
mediator,
memories,
live,
structured,
}
}
/// 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 structured target: `ask` enqueues a bookkeeping ticket, sends the turn to the
/// headless session, and returns the session's `Final`. The PTY is not used.
#[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 PTY is no longer part of the inter-agent conversation path.
let writes = fx.pty.writes_for(sid(800));
assert_eq!(
writes.len(),
0,
"inter-agent conversation must not write to the PTY"
);
// 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");
}
// --- LS3: live-state auto-update from the delegation cycle ------------------
/// A successful `ask` flips the **target** to `Working` with the distilled intent and
/// the current ticket — derived from the existing delegation transition (zéro token).
#[tokio::test]
async fn ask_sets_target_working_with_intent_and_ticket() {
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 });
// The Working transition is posted right after the enqueue, before any reply.
await_until(|| fx.live.entry_for(aid(1)).is_some()).await;
let entry = fx
.live
.entry_for(aid(1))
.expect("target live entry present");
assert_eq!(entry.status, WorkStatus::Working);
assert_eq!(
entry.intent, "Analyse §17",
"intent distilled from the task"
);
assert!(entry.ticket.is_some(), "current ticket recorded");
assert!(
entry.last_delegation.is_none(),
"no delegation resolved yet"
);
// The live-state path does not touch the memory store (canonical effects separate).
assert!(
fx.memories.slugs().is_empty(),
"no memory write on this path"
);
// Drain the ask so the spawned task completes cleanly.
fx.service
.dispatch(&project(), reply_cmd(aid(1), "done"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
}
/// A successful `reply` flips the target to `Done` and records `last_delegation` = the
/// resolved ticket.
#[tokio::test]
async fn reply_sets_target_done_with_last_delegation() {
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;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "the answer"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
let entry = fx
.live
.entry_for(aid(1))
.expect("target live entry present");
assert_eq!(
entry.status,
WorkStatus::Done,
"target marked Done on reply"
);
assert!(
entry.last_delegation.is_some(),
"last_delegation set to the resolved ticket"
);
// Working then Done ⇒ exactly two upserts for one delegation (no write-storm).
assert_eq!(
fx.live.upsert_count(),
2,
"one Working + one Done write per delegation"
);
}
/// A live-state store that fails must NOT fail the `ask`/`reply`: the delegation
/// still returns its reply (best-effort hook, exactly like the memory harvest).
#[tokio::test]
async fn live_state_failure_does_not_break_delegation() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture_with_live(FakeContexts::with_agent(&agent, "# persona"), true);
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), "still works"))
.await
.expect("reply ok despite failing live-state");
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok despite failing live-state");
assert_eq!(out.reply.as_deref(), Some("still works"));
// The failing store recorded nothing.
assert!(fx.live.entry_for(aid(1)).is_none());
}
/// When the live-state hook is **unwired** (`None`), the delegation behaves exactly as
/// before — no live-state write at all (zéro régression).
#[tokio::test]
async fn no_live_state_wired_means_no_write() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture_without_live(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;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "ok"))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
assert_eq!(fx.live.upsert_count(), 0, "no write when hook is None");
assert!(fx.live.entry_for(aid(1)).is_none());
}
// ---------------------------------------------------------------------------
// LS4 — `idea_workstate_set` / `idea_workstate_read` service round-trip.
// ---------------------------------------------------------------------------
/// `idea_workstate_set` writes the current agent's row, then `idea_workstate_read`
/// returns it enriched with the **resolved display name** (via `ListAgents`), with the
/// declared status/intent — and **never** the `progress` field.
#[tokio::test]
async fn workstate_set_then_read_round_trips_with_resolved_name_and_no_progress() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// set — keyed on the handshake agent id; carries a progress note (must not leak).
let ack = fx
.service
.dispatch(
&project(),
OrchestratorCommand::SetWorkState {
agent: aid(1),
status: WorkStatus::Working,
intent: Some("ship LS4".to_owned()),
progress: Some("secret internal note".to_owned()),
ticket: None,
last_delegation: None,
},
)
.await
.expect("set ok");
// ACK only — no inline reply on the write path.
assert!(ack.reply.is_none(), "set is ACK only, no reply: {ack:?}");
// read — the JSON array carries one row for the architect.
let out = fx
.service
.dispatch(
&project(),
OrchestratorCommand::ReadWorkState {
requester: ConversationParty::User,
},
)
.await
.expect("read ok");
let json: serde_json::Value =
serde_json::from_str(out.reply.as_deref().expect("read carries a reply")).unwrap();
let rows = json.as_array().expect("array");
assert_eq!(rows.len(), 1, "one live row expected: {json}");
let row = &rows[0];
// Name resolved from the manifest (not the bare UUID).
assert_eq!(row["agent"], serde_json::json!("architect"));
assert_eq!(row["status"], serde_json::json!("working"));
assert_eq!(row["intent"], serde_json::json!("ship LS4"));
// progress is NEVER exposed on read (the lean DTO has no such field).
assert!(
row.get("progress").is_none(),
"progress must never surface on read: {row}"
);
}
/// `idea_workstate_set` is last-writer-wins: a second set for the same agent replaces
/// the row in place (never a duplicate), and the read reflects the latest write.
#[tokio::test]
async fn workstate_set_is_last_writer_wins() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
for (status, intent) in [
(WorkStatus::Working, "first"),
(WorkStatus::Blocked, "second"),
] {
fx.service
.dispatch(
&project(),
OrchestratorCommand::SetWorkState {
agent: aid(1),
status,
intent: Some(intent.to_owned()),
progress: None,
ticket: None,
last_delegation: None,
},
)
.await
.expect("set ok");
}
let out = fx
.service
.dispatch(
&project(),
OrchestratorCommand::ReadWorkState {
requester: ConversationParty::User,
},
)
.await
.expect("read ok");
let json: serde_json::Value = serde_json::from_str(out.reply.as_deref().unwrap()).unwrap();
let rows = json.as_array().unwrap();
assert_eq!(rows.len(), 1, "LWW ⇒ exactly one row, no duplicate: {json}");
assert_eq!(rows[0]["status"], serde_json::json!("blocked"));
assert_eq!(rows[0]["intent"], serde_json::json!("second"));
}
/// `idea_workstate_read` drops rows whose agent left (or never was in) the manifest —
/// the manifest boundary is authoritative.
#[tokio::test]
async fn workstate_read_drops_off_manifest_rows() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
// aid(1) is in the manifest; aid(2) is NOT.
for who in [aid(1), aid(2)] {
fx.service
.dispatch(
&project(),
OrchestratorCommand::SetWorkState {
agent: who,
status: WorkStatus::Working,
intent: Some("x".to_owned()),
progress: None,
ticket: None,
last_delegation: None,
},
)
.await
.expect("set ok");
}
let out = fx
.service
.dispatch(
&project(),
OrchestratorCommand::ReadWorkState {
requester: ConversationParty::User,
},
)
.await
.expect("read ok");
let json: serde_json::Value = serde_json::from_str(out.reply.as_deref().unwrap()).unwrap();
let rows = json.as_array().unwrap();
assert_eq!(rows.len(), 1, "the off-manifest row is dropped: {json}");
assert_eq!(rows[0]["agent"], serde_json::json!("architect"));
}
/// An oversize `intent` (above the domain hard byte threshold) is **rejected** by the
/// write path (the domain bound surfaces as `AppError::Invalid`), never silently stored.
#[tokio::test]
async fn workstate_set_oversize_intent_is_rejected() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
let huge = "x".repeat(domain::live_state::FIELD_MAX_BYTES + 1);
let err = fx
.service
.dispatch(
&project(),
OrchestratorCommand::SetWorkState {
agent: aid(1),
status: WorkStatus::Working,
intent: Some(huge),
progress: None,
ticket: None,
last_delegation: None,
},
)
.await
.expect_err("oversize intent must be rejected");
assert!(
matches!(err, application::AppError::Invalid(_)),
"oversize ⇒ AppError::Invalid, got {err:?}"
);
}
/// The target returned to its prompt **without** `idea_reply`: the mediator's grace
/// window expired and completed the turn "no reply". The awaiting ask must error out
/// promptly with the typed, retryable [`AppError::TargetReturnedNoReply`] — never hang
/// until the long safety timeout. (The grace **timer** lives in the infrastructure
/// mediator; here we drive the resulting mailbox completion directly.)
#[tokio::test]
async fn ask_target_returns_to_prompt_without_reply_is_a_typed_error() {
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;
// Simulate the grace-window completion (target back at prompt, no idea_reply).
let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
fx.mailbox.complete_without_reply(aid(1), ticket);
let err = timeout(TEST_GUARD, ask)
.await
.expect("ask completes promptly, not after the long timeout")
.expect("join ok")
.expect_err("a no-reply turn is a typed error");
assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY", "got {err:?}");
assert_eq!(fx.mailbox.pending(&aid(1)), 0, "head retired by completion");
}
/// An `idea_reply` that lands **first** still wins over a (later) no-reply completion:
/// the completion then finds the head gone and is a no-op (idempotent race).
#[tokio::test]
async fn ask_reply_wins_then_late_completion_is_noop() {
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;
let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
// idea_reply resolves the head first…
fx.service
.dispatch(&project(), reply_cmd(aid(1), "the real answer"))
.await
.expect("reply ok");
// …then a late grace completion for the same ticket is a no-op.
fx.mailbox.complete_without_reply(aid(1), ticket);
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("the real answer"));
}
// --- Lot E1: auto-memory harvest on a successful ask reply -----------------
/// A reply carrying an `idea-memory` block is harvested **after** the reply
/// resolves: the note is persisted via the store and a `MemorySaved` event fires.
#[tokio::test]
async fn ask_reply_with_directive_harvests_memory_after_reply() {
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;
let reply = "Here is the answer.\n\n```idea-memory\nslug: section-17-rule\ntitle: §17 rule\ntype: project\ndescription: How §17 routing works\n---\nRoute structured asks straight to the session.\n```";
fx.service
.dispatch(&project(), reply_cmd(aid(1), reply))
.await
.expect("reply ok");
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
// The reply is returned unchanged …
assert_eq!(out.reply.as_deref(), Some(reply));
// … and the embedded note was harvested into the store + announced.
assert_eq!(fx.memories.slugs(), vec!["section-17-rule"]);
let saved: Vec<_> = fx
.bus
.events()
.into_iter()
.filter_map(|e| match e {
DomainEvent::MemorySaved { slug } => Some(slug.as_str().to_owned()),
_ => None,
})
.collect();
assert!(
saved.contains(&"section-17-rule".to_owned()),
"MemorySaved announced, got {saved:?}"
);
}
/// A store failure during harvest must not break the reply: the ask still returns
/// its content, and nothing is persisted.
#[tokio::test]
async fn ask_reply_harvest_store_failure_does_not_break_reply() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = ask_fixture_with_memory(FakeContexts::with_agent(&agent, "# persona"), true);
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;
let reply = "Reply.\n\n```idea-memory\nslug: doomed-note\ntitle: T\ntype: project\ndescription: hook\n---\nBody.\n```";
fx.service
.dispatch(&project(), reply_cmd(aid(1), reply))
.await
.expect("reply ok despite failing memory store");
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask still ok despite harvest store failure");
assert_eq!(
out.reply.as_deref(),
Some(reply),
"reply unaffected by harvest"
);
assert!(
fx.memories.slugs().is_empty(),
"failed save persisted nothing"
);
}
/// A reply with no directive harvests nothing (and the reply is unchanged) — the
/// common case must not touch the memory store.
#[tokio::test]
async fn ask_reply_without_directive_harvests_nothing() {
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;
fx.service
.dispatch(&project(), reply_cmd(aid(1), "Just a plain answer."))
.await
.expect("reply ok");
timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok")
.expect("ask ok");
assert!(fx.memories.slugs().is_empty());
}
/// A cancelled turn (head ticket dropped before any reply) yields no reply, so the
/// harvest never runs — nothing is persisted even with a would-be directive pending.
#[tokio::test]
async fn ask_cancelled_turn_does_not_harvest() {
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;
// Cancel the head ticket (drops the reply sender) ⇒ the ask resolves as a
// channel-closed error: no Response turn, hence no harvest.
let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
fx.mailbox.cancel_head(aid(1), ticket);
let out = timeout(TEST_GUARD, ask)
.await
.expect("ask completes")
.expect("join ok");
assert!(out.is_err(), "cancelled turn ⇒ ask errors");
assert!(fx.memories.slugs().is_empty(), "no harvest on cancel");
}
/// 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;
let _ = fx.mailbox.completions().next(aid(1)).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 structured target is launched headless before the task is sent through
/// `AgentSession::send`; the PTY is not used.
#[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());
assert!(fx.structured.session_for_agent(&aid(1)).is_some());
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.pty.spawns().is_empty(), "AskAgent does not spawn a PTY");
assert!(
fx.pty.writes_for(sid(777)).is_empty(),
"AskAgent does not write into a PTY"
);
fx.mailbox
.resolve(aid(1), "launched reply".to_owned())
.expect("structured Final");
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 {
/// 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 structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
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,
)
.with_structured(
Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let 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()))
.with_structured(Arc::clone(&structured));
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 via le Final structured.
fx.mailbox
.resolve(aid(1), "done".to_owned())
.expect("structured completion 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.mailbox
.resolve(aid(1), "done".to_owned())
.expect("structured completion ok");
timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}
/// F2 — GARDE (passant en mode structured). Une cible Codex passe si l'adaptateur
/// structured est câblé par la fixture : pas de fallback PTY, réponse par `Final`.
#[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 svc = Arc::clone(&fx.service);
let ask = tokio::spawn(async move {
svc.dispatch(
&project(),
cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#),
)
.await
});
await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
fx.mailbox
.resolve(aid(1), "codex ok".to_owned())
.expect("structured completion ok");
let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
assert_eq!(out.reply.as_deref(), Some("codex ok"));
assert!(fx.pty.spawns().is_empty(), "no PTY fallback");
}
/// 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.mailbox
.resolve(aid(1), "ok claude".to_owned())
.expect("structured completion 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 structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
let launch = Arc::new(
LaunchAgent::new(
Arc::new(contexts.clone()),
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeRuntime),
Arc::new(FakeFs),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
)
.with_structured(
Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let 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()))
.with_structured(Arc::clone(&structured)),
);
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 structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
seed_structured_sessions(&structured, &contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
let launch = Arc::new(
LaunchAgent::new(
Arc::new(contexts.clone()),
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeRuntime),
Arc::new(FakeFs),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
)
.with_structured(
Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let 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_structured(Arc::clone(&structured)),
);
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 structured = Arc::new(StructuredSessions::new());
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
seed_structured_sessions(&structured, &contexts, mailbox.completions());
let create = Arc::new(CreateAgentFromScratch::new(
Arc::new(contexts.clone()),
Arc::new(SeqIds::new()),
Arc::new(bus.clone()),
));
let launch = Arc::new(
LaunchAgent::new(
Arc::new(contexts.clone()),
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
Arc::new(FakeRuntime),
Arc::new(FakeFs),
Arc::new(pty.clone()),
Arc::new(FakeSkills),
Arc::clone(&sessions),
Arc::new(bus.clone()),
Arc::new(SeqIds::new()),
Arc::new(FakeRecall),
None,
)
.with_structured(
Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
Arc::clone(&structured),
),
);
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
let close = Arc::new(CloseTerminal::new(
Arc::new(pty.clone()),
Arc::clone(&sessions),
));
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
let 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_structured(Arc::clone(&structured))
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
);
AskFixture {
service,
mailbox,
sessions,
bus,
pty,
mediator,
// Harvest is intentionally not wired here: these fixtures exercise the
// RecordTurn checkpoint in isolation (proving it stays unchanged).
memories: Arc::new(HarvestMemories::default()),
// Live-state likewise unwired here (RecordTurn-focused fixtures).
live: Arc::new(RecordingLiveState::default()),
structured,
}
}
/// 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.mailbox
.resolve(reply_from, result.to_owned())
.expect("structured completion 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::{AgentResumer, StructuredSessions};
use domain::ids::ScheduleId;
use domain::ports::{
AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream, ScheduledTask,
Scheduler,
};
/// 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,
conversation_id: Option<String>,
}
#[async_trait]
impl AgentSession for EchoSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
self.conversation_id.clone()
}
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,
_env: &[(String, String)],
_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,
conversation_id: Some(format!("engine-{id}")),
}))
}
}
struct RateLimitedSession {
id: SessionId,
conversation_id: Option<String>,
resets_at_ms: i64,
}
#[async_trait]
impl AgentSession for RateLimitedSession {
fn id(&self) -> SessionId {
self.id
}
fn conversation_id(&self) -> Option<String> {
self.conversation_id.clone()
}
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
Ok(Box::new(
vec![ReplyEvent::RateLimited {
resets_at_ms: Some(self.resets_at_ms),
}]
.into_iter(),
))
}
async fn shutdown(&self) -> Result<(), AgentSessionError> {
Ok(())
}
}
#[derive(Default)]
struct NoopScheduler {
next: Mutex<u128>,
}
impl Scheduler for NoopScheduler {
fn arm(&self, _fire_at_ms: i64, _task: ScheduledTask) -> ScheduleId {
let mut next = self.next.lock().unwrap();
*next += 1;
ScheduleId::from_uuid(Uuid::from_u128(*next))
}
fn cancel(&self, _id: ScheduleId) -> bool {
true
}
}
struct NoopResumer;
#[async_trait]
impl AgentResumer for NoopResumer {
async fn resume(
&self,
_agent_id: AgentId,
_node_id: NodeId,
_conversation_id: Option<String>,
_resume_prompt: &str,
) -> Result<(), application::AppError> {
Ok(())
}
}
/// 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>,
sessions: Arc<TerminalSessions>,
factory: CountingFactory,
pty: FakePty,
bus: SpyBus,
session_limits: Arc<SessionLimitService>,
}
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 session_limits = Arc::new(SessionLimitService::new(
Arc::new(FixedMillisClock(1_000)) as Arc<dyn Clock>,
Arc::new(NoopScheduler::default()) as Arc<dyn Scheduler>,
Arc::new(bus.clone()),
Arc::new(NoopResumer),
));
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_session_limits(Arc::clone(&session_limits))
// 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,
sessions,
factory,
pty,
bus,
session_limits,
}
}
/// 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"
);
}
/// Régression live : une cible structurée peut déjà être vivante en PTY parce qu'elle
/// a été ouverte depuis la surface humaine/menu historique. Le chemin `ask` headless
/// doit alors créer une session structurée **sans fermer** ce PTY visible : les deux
/// canaux d'entrée coexistent, et la délégation passe par `AgentSession::send`.
#[tokio::test]
async fn ask_structured_target_live_as_pty_keeps_visible_terminal() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
seed_live_pty(&fx.sessions, aid(1), sid(800));
let out = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect("ask ok");
assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
assert_eq!(fx.factory.start_count(), 1, "headless session started");
assert!(fx.pty.kills().is_empty(), "visible PTY must not be stopped");
assert!(
fx.structured.session_for_agent(&aid(1)).is_some(),
"target now has a structured session"
);
assert!(
fx.sessions.session_for_agent(&aid(1)).is_some(),
"target still has its visible PTY session"
);
}
/// Sous-lot adjacent #7/F2 : une limite détectée pendant un rendez-vous structuré
/// doit aussi passer par `SessionLimitService::on_rate_limited` pour la cible, afin
/// de publier le flux mono-agent `AgentRateLimited` puis `AgentResumeScheduled`.
#[tokio::test]
async fn ask_structured_rate_limit_arms_target_resume_events() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
let resets_at_ms = 9_999_999;
fx.structured.insert(
Arc::new(RateLimitedSession {
id: sid(9700),
conversation_id: Some("engine-target".to_owned()),
resets_at_ms,
}),
aid(1),
nid(700),
);
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect_err("a rate-limited delegated turn has no Final reply");
assert_eq!(err.code(), "PROCESS");
let events = fx.bus.events();
assert!(
events.iter().any(|event| matches!(
event,
DomainEvent::AgentRateLimited {
agent_id,
resets_at_ms: Some(t),
} if *agent_id == aid(1) && *t == resets_at_ms
)),
"target AgentRateLimited must be published, got {events:?}"
);
assert!(
events.iter().any(|event| matches!(
event,
DomainEvent::AgentResumeScheduled {
agent_id,
fire_at_ms,
} if *agent_id == aid(1) && *fire_at_ms == resets_at_ms
)),
"target AgentResumeScheduled must be published, got {events:?}"
);
assert!(
fx.session_limits.cancel_resume(aid(1)),
"delegated target resume must be cancellable"
);
}
/// Sous-lot adjacent #7/F2 : si la cible structurée est limitée avec un reset connu
/// mais sans `conversation_id` moteur, l'orchestrateur doit exposer le fallback humain
/// et ne jamais armer une reprise non reprenable.
#[tokio::test]
async fn ask_structured_rate_limit_without_conversation_uses_human_fallback_no_resume() {
let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
fx.structured.insert(
Arc::new(RateLimitedSession {
id: sid(9701),
conversation_id: None,
resets_at_ms: 9_999_999,
}),
aid(1),
nid(700),
);
let err = fx
.service
.dispatch(&project(), cmd(ASK_JSON))
.await
.expect_err("a rate-limited delegated turn has no Final reply");
assert_eq!(err.code(), "PROCESS");
let events = fx.bus.events();
assert!(
events.iter().any(|event| matches!(
event,
DomainEvent::AgentRateLimited {
agent_id,
resets_at_ms: None,
} if *agent_id == aid(1)
)),
"target AgentRateLimited(None) must be published, got {events:?}"
);
assert!(
events.iter().any(|event| matches!(
event,
DomainEvent::AgentRateLimitSuspected { agent_id, .. } if *agent_id == aid(1)
)),
"target AgentRateLimitSuspected must be published, got {events:?}"
);
assert!(
!events.iter().any(|event| matches!(
event,
DomainEvent::AgentResumeScheduled { agent_id, .. } if *agent_id == aid(1)
)),
"target resume must not be scheduled without engine conversation_id, got {events:?}"
);
assert!(
!fx.session_limits.cancel_resume(aid(1)),
"no delegated resume should be armed without engine conversation_id"
);
}