1118 lines
37 KiB
Rust
1118 lines
37 KiB
Rust
//! Integration tests for the orchestrator filesystem adapter (ARCHITECTURE §14.3).
|
|
//!
|
|
//! These drive [`process_request_file`] — the standalone "parse + dispatch + write
|
|
//! response + delete request" unit — against a real temp directory, with an
|
|
//! [`OrchestratorService`] wired over in-memory fakes. We assert:
|
|
//!
|
|
//! - a **valid** `spawn_agent` request → success response, request deleted, agent
|
|
//! created + launched,
|
|
//! - an **invalid JSON** request → error response (no panic, request deleted),
|
|
//! - an unknown action → error response carrying the rejection.
|
|
|
|
use std::collections::HashMap;
|
|
use std::path::PathBuf;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use async_trait::async_trait;
|
|
use domain::agent::{AgentManifest, ManifestEntry};
|
|
use domain::events::{DomainEvent, OrchestrationSource};
|
|
use domain::ids::SkillId;
|
|
use domain::ids::{AgentId, ProfileId, ProjectId};
|
|
use domain::markdown::MarkdownDoc;
|
|
use domain::ports::{
|
|
AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, AgentSessionFactory,
|
|
BackgroundTaskPortError, BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, EventBus,
|
|
EventStream, ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext,
|
|
ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, ReplyEvent, ReplyStream, RuntimeError,
|
|
SessionPlan, SkillStore, SpawnSpec, StoreError,
|
|
};
|
|
use domain::profile::{
|
|
AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport,
|
|
StructuredAdapter,
|
|
};
|
|
use domain::project::{Project, ProjectPath};
|
|
use domain::remote::RemoteRef;
|
|
use domain::skill::{Skill, SkillScope};
|
|
use domain::{
|
|
BackgroundTask, BackgroundTaskResult, BackgroundTaskState, PtySize, SessionId, TaskId,
|
|
};
|
|
use uuid::Uuid;
|
|
|
|
use application::{
|
|
CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
|
|
OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext,
|
|
};
|
|
use infrastructure::{
|
|
process_request_file, FsOrchestratorWatcher, InMemoryMailbox, OrchestratorResponse,
|
|
};
|
|
|
|
// --- temp dir (mirror local_fs.rs) ---
|
|
struct TempDir(PathBuf);
|
|
impl TempDir {
|
|
fn new() -> Self {
|
|
let p = std::env::temp_dir().join(format!("idea-orch-{}", Uuid::new_v4()));
|
|
std::fs::create_dir_all(&p).unwrap();
|
|
Self(p)
|
|
}
|
|
}
|
|
impl Drop for TempDir {
|
|
fn drop(&mut self) {
|
|
let _ = std::fs::remove_dir_all(&self.0);
|
|
}
|
|
}
|
|
|
|
// --- minimal fakes ---
|
|
#[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 entries(&self) -> Vec<ManifestEntry> {
|
|
self.0.lock().unwrap().manifest.entries.clone()
|
|
}
|
|
/// Seeds the manifest with an already-existing agent so `find_agent_id_by_name`
|
|
/// resolves it without going through `CreateAgentFromScratch`. Returns the id.
|
|
fn seed_agent(&self, name: &str) -> AgentId {
|
|
let id = AgentId::from_uuid(Uuid::new_v4());
|
|
let mut inner = self.0.lock().unwrap();
|
|
inner.manifest.entries.push(ManifestEntry {
|
|
agent_id: id,
|
|
name: name.to_owned(),
|
|
md_path: format!("agents/{name}.md"),
|
|
profile_id: ProfileId::from_uuid(Uuid::from_u128(9)),
|
|
template_id: None,
|
|
synchronized: false,
|
|
synced_template_version: None,
|
|
skills: Vec::new(),
|
|
});
|
|
id
|
|
}
|
|
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 = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
|
|
Ok(MarkdownDoc::new(
|
|
self.0
|
|
.lock()
|
|
.unwrap()
|
|
.contents
|
|
.get(&md)
|
|
.cloned()
|
|
.unwrap_or_default(),
|
|
))
|
|
}
|
|
async fn write_context(
|
|
&self,
|
|
_project: &Project,
|
|
agent: &AgentId,
|
|
md: &MarkdownDoc,
|
|
) -> Result<(), StoreError> {
|
|
let path = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
|
|
self.0
|
|
.lock()
|
|
.unwrap()
|
|
.contents
|
|
.insert(path, md.as_str().to_owned());
|
|
Ok(())
|
|
}
|
|
async fn load_manifest(&self, _project: &Project) -> Result<AgentManifest, StoreError> {
|
|
Ok(self.0.lock().unwrap().manifest.clone())
|
|
}
|
|
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>>);
|
|
#[async_trait]
|
|
impl ProfileStore for FakeProfiles {
|
|
async fn list(&self) -> Result<Vec<AgentProfile>, StoreError> {
|
|
Ok((*self.0).clone())
|
|
}
|
|
async fn save(&self, _p: &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 watcher 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 watcher-driven launch behaviour unchanged.
|
|
#[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())
|
|
}
|
|
}
|
|
|
|
struct FakeRuntime;
|
|
#[async_trait]
|
|
impl AgentRuntime for FakeRuntime {
|
|
async fn detect(&self, _p: &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, p: &RemotePath) -> Result<Vec<u8>, FsError> {
|
|
Err(FsError::NotFound(p.as_str().to_owned()))
|
|
}
|
|
async fn write(&self, _p: &RemotePath, _d: &[u8]) -> Result<(), FsError> {
|
|
Ok(())
|
|
}
|
|
async fn exists(&self, _p: &RemotePath) -> Result<bool, FsError> {
|
|
Ok(false)
|
|
}
|
|
async fn create_dir_all(&self, _p: &RemotePath) -> Result<(), FsError> {
|
|
Ok(())
|
|
}
|
|
async fn list(&self, _p: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
|
|
Ok(Vec::new())
|
|
}
|
|
async fn symlink(&self, _s: &RemotePath, _d: &RemotePath) -> Result<(), FsError> {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct FakePty;
|
|
#[async_trait]
|
|
impl PtyPort for FakePty {
|
|
async fn spawn(&self, _s: SpawnSpec, _z: PtySize) -> Result<PtyHandle, PtyError> {
|
|
Ok(PtyHandle {
|
|
session_id: SessionId::from_uuid(Uuid::from_u128(777)),
|
|
})
|
|
}
|
|
fn write(&self, _h: &PtyHandle, _d: &[u8]) -> Result<(), PtyError> {
|
|
Ok(())
|
|
}
|
|
fn resize(&self, _h: &PtyHandle, _z: PtySize) -> Result<(), PtyError> {
|
|
Ok(())
|
|
}
|
|
fn subscribe_output(&self, _h: &PtyHandle) -> Result<OutputStream, PtyError> {
|
|
Ok(Box::new(std::iter::empty()))
|
|
}
|
|
fn scrollback(&self, _h: &PtyHandle) -> Result<Vec<u8>, PtyError> {
|
|
Ok(Vec::new())
|
|
}
|
|
async fn kill(&self, _h: &PtyHandle) -> Result<ExitStatus, PtyError> {
|
|
Ok(ExitStatus { code: Some(0) })
|
|
}
|
|
}
|
|
|
|
#[derive(Default, Clone)]
|
|
struct NoopBus;
|
|
impl EventBus for NoopBus {
|
|
fn publish(&self, _e: DomainEvent) {}
|
|
fn subscribe(&self) -> EventStream {
|
|
Box::new(std::iter::empty())
|
|
}
|
|
}
|
|
|
|
struct SeqIds(Mutex<u128>);
|
|
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
|
|
}
|
|
}
|
|
|
|
struct FixedClock(i64);
|
|
impl Clock for FixedClock {
|
|
fn now_millis(&self) -> i64 {
|
|
self.0
|
|
}
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct InMemoryBackgroundTaskStore {
|
|
tasks: Mutex<HashMap<TaskId, BackgroundTask>>,
|
|
}
|
|
|
|
impl InMemoryBackgroundTaskStore {
|
|
fn all(&self) -> Vec<BackgroundTask> {
|
|
self.tasks.lock().unwrap().values().cloned().collect()
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl BackgroundTaskStore for InMemoryBackgroundTaskStore {
|
|
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.tasks.lock().unwrap().get(&id).cloned())
|
|
}
|
|
|
|
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 mut tasks = self.tasks.lock().unwrap();
|
|
let task = tasks
|
|
.get(&task_id)
|
|
.cloned()
|
|
.ok_or(BackgroundTaskPortError::NotFound)?;
|
|
let delivered = task
|
|
.mark_completion_delivered()
|
|
.map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?;
|
|
tasks.insert(task_id, delivered);
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct BlockingReplyFactory {
|
|
outcome: Arc<Mutex<BlockingReplyOutcome>>,
|
|
notify: Arc<tokio::sync::Notify>,
|
|
}
|
|
|
|
impl BlockingReplyFactory {
|
|
fn new(reply: impl Into<String>) -> Self {
|
|
Self {
|
|
outcome: Arc::new(Mutex::new(BlockingReplyOutcome::Final(reply.into()))),
|
|
notify: Arc::new(tokio::sync::Notify::new()),
|
|
}
|
|
}
|
|
|
|
fn no_reply(&self) {
|
|
*self.outcome.lock().unwrap() = BlockingReplyOutcome::NoReply;
|
|
}
|
|
|
|
fn release(&self) {
|
|
self.notify.notify_waiters();
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
enum BlockingReplyOutcome {
|
|
Final(String),
|
|
NoReply,
|
|
}
|
|
|
|
struct BlockingReplySession {
|
|
id: SessionId,
|
|
outcome: Arc<Mutex<BlockingReplyOutcome>>,
|
|
notify: Arc<tokio::sync::Notify>,
|
|
}
|
|
|
|
#[async_trait]
|
|
impl AgentSession for BlockingReplySession {
|
|
fn id(&self) -> SessionId {
|
|
self.id
|
|
}
|
|
|
|
fn conversation_id(&self) -> Option<String> {
|
|
None
|
|
}
|
|
|
|
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
|
self.notify.notified().await;
|
|
match self.outcome.lock().unwrap().clone() {
|
|
BlockingReplyOutcome::Final(content) => {
|
|
Ok(Box::new(vec![ReplyEvent::Final { content }].into_iter()))
|
|
}
|
|
BlockingReplyOutcome::NoReply => Err(AgentSessionError::Io(
|
|
"target returned without a structured final".to_owned(),
|
|
)),
|
|
}
|
|
}
|
|
|
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl AgentSessionFactory for BlockingReplyFactory {
|
|
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> {
|
|
Ok(Arc::new(BlockingReplySession {
|
|
id: SessionId::from_uuid(Uuid::new_v4()),
|
|
outcome: Arc::clone(&self.outcome),
|
|
notify: Arc::clone(&self.notify),
|
|
}))
|
|
}
|
|
}
|
|
|
|
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 build_service(contexts: FakeContexts) -> Arc<OrchestratorService> {
|
|
// Profil Claude **complet** (adaptateur structuré + capacité MCP `.mcp.json`) :
|
|
// seul profil que la garde F2 (`guard_mcp_bridge_supported`) accepte comme cible
|
|
// d'`idea_ask_agent`, car seul Claude consomme réellement le pont `.mcp.json`.
|
|
let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new(
|
|
ProfileId::from_uuid(Uuid::from_u128(9)),
|
|
"Claude Code",
|
|
"claude",
|
|
Vec::new(),
|
|
ContextInjection::stdin(),
|
|
None,
|
|
"{agentRunDir}",
|
|
None,
|
|
)
|
|
.unwrap()
|
|
.with_structured_adapter(StructuredAdapter::Claude)
|
|
.with_mcp(McpCapability::new(
|
|
McpConfigStrategy::config_file(".mcp.json").unwrap(),
|
|
McpTransport::Stdio,
|
|
))])));
|
|
let sessions = Arc::new(TerminalSessions::new());
|
|
let bus = Arc::new(NoopBus);
|
|
let create = Arc::new(CreateAgentFromScratch::new(
|
|
Arc::new(contexts.clone()),
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
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(FakePty),
|
|
Arc::new(FakeSkills),
|
|
Arc::clone(&sessions),
|
|
bus.clone(),
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
Arc::new(FakeRecall),
|
|
None,
|
|
));
|
|
let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
|
|
let close = Arc::new(CloseTerminal::new(Arc::new(FakePty), Arc::clone(&sessions)));
|
|
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts)));
|
|
let create_skill = Arc::new(CreateSkill::new(
|
|
Arc::new(FakeSkills) as Arc<dyn SkillStore>,
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
));
|
|
Arc::new(OrchestratorService::new(
|
|
create,
|
|
launch,
|
|
list,
|
|
close,
|
|
update,
|
|
create_skill,
|
|
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
|
sessions,
|
|
))
|
|
}
|
|
|
|
/// Like [`build_service`] but with the **inter-agent mailbox** + PTY wired (Option 1
|
|
/// « Terminal + MCP », B-3/B-4), so an `agent.message` blocks on the mailbox and an
|
|
/// `agent.reply` resolves it. Returns the service, the shared mailbox (to observe
|
|
/// pending tickets) and the PTY registry (to pre-seed a live target).
|
|
fn build_service_with_mailbox(
|
|
contexts: FakeContexts,
|
|
) -> (
|
|
Arc<OrchestratorService>,
|
|
Arc<InMemoryMailbox>,
|
|
Arc<TerminalSessions>,
|
|
BlockingReplyFactory,
|
|
Arc<InMemoryBackgroundTaskStore>,
|
|
) {
|
|
// Profil Claude **complet** (adaptateur structuré + capacité MCP `.mcp.json`) :
|
|
// seul profil que la garde F2 (`guard_mcp_bridge_supported`) accepte comme cible
|
|
// d'`idea_ask_agent`, car seul Claude consomme réellement le pont `.mcp.json`.
|
|
let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new(
|
|
ProfileId::from_uuid(Uuid::from_u128(9)),
|
|
"Claude Code",
|
|
"claude",
|
|
Vec::new(),
|
|
ContextInjection::stdin(),
|
|
None,
|
|
"{agentRunDir}",
|
|
None,
|
|
)
|
|
.unwrap()
|
|
.with_structured_adapter(StructuredAdapter::Claude)
|
|
.with_mcp(McpCapability::new(
|
|
McpConfigStrategy::config_file(".mcp.json").unwrap(),
|
|
McpTransport::Stdio,
|
|
))])));
|
|
let sessions = Arc::new(TerminalSessions::new());
|
|
let structured = Arc::new(StructuredSessions::new());
|
|
let structured_factory = BlockingReplyFactory::new("the answer is 42");
|
|
let background_tasks = Arc::new(InMemoryBackgroundTaskStore::default());
|
|
let mailbox = Arc::new(InMemoryMailbox::new());
|
|
let bus = Arc::new(NoopBus);
|
|
let create = Arc::new(CreateAgentFromScratch::new(
|
|
Arc::new(contexts.clone()),
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
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(FakePty),
|
|
Arc::new(FakeSkills),
|
|
Arc::clone(&sessions),
|
|
bus.clone(),
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
Arc::new(FakeRecall),
|
|
None,
|
|
)
|
|
.with_structured(
|
|
Arc::new(structured_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(FakePty), Arc::clone(&sessions)));
|
|
let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts)));
|
|
let create_skill = Arc::new(CreateSkill::new(
|
|
Arc::new(FakeSkills) as Arc<dyn SkillStore>,
|
|
Arc::new(SeqIds(Mutex::new(1))),
|
|
));
|
|
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::new(infrastructure::MediatedInbox::with_pty(
|
|
Arc::clone(&mailbox),
|
|
Arc::new(infrastructure::SystemMillisClock),
|
|
Arc::new(FakePty) as Arc<dyn PtyPort>,
|
|
)) as Arc<dyn domain::input::InputMediator>,
|
|
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
|
|
)
|
|
.with_conversations(
|
|
Arc::new(infrastructure::InMemoryConversationRegistry::new())
|
|
as Arc<dyn domain::conversation::ConversationRegistry>,
|
|
)
|
|
.with_structured(Arc::clone(&structured))
|
|
.with_background_tasks(
|
|
Arc::clone(&background_tasks) as Arc<dyn BackgroundTaskStore>,
|
|
Arc::new(FixedClock(1_700_000_000_000)) as Arc<dyn Clock>,
|
|
),
|
|
);
|
|
(
|
|
service,
|
|
mailbox,
|
|
sessions,
|
|
structured_factory,
|
|
background_tasks,
|
|
)
|
|
}
|
|
|
|
fn read_response(request_path: &std::path::Path) -> OrchestratorResponse {
|
|
let mut name = request_path
|
|
.file_name()
|
|
.unwrap()
|
|
.to_string_lossy()
|
|
.into_owned();
|
|
name.push_str(".response.json");
|
|
let response_path = request_path.with_file_name(name);
|
|
let bytes = std::fs::read(&response_path).expect("response file written");
|
|
serde_json::from_slice(&bytes).expect("response parses")
|
|
}
|
|
|
|
/// Reads the raw `*.response.json` bytes as a UTF-8 string, for assertions on the
|
|
/// *literal wire shape* (e.g. that the `"reply"` key is absent — not just `None`
|
|
/// after deserialisation).
|
|
fn read_response_raw(request_path: &std::path::Path) -> String {
|
|
let mut name = request_path
|
|
.file_name()
|
|
.unwrap()
|
|
.to_string_lossy()
|
|
.into_owned();
|
|
name.push_str(".response.json");
|
|
let response_path = request_path.with_file_name(name);
|
|
std::fs::read_to_string(&response_path).expect("response file written")
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn valid_spawn_request_succeeds_and_is_consumed() {
|
|
let tmp = TempDir::new();
|
|
let contexts = FakeContexts::new();
|
|
let service = build_service(contexts.clone());
|
|
|
|
let req = tmp.0.join("req-1.json");
|
|
std::fs::write(
|
|
&req,
|
|
br#"{ "action": "spawn_agent", "name": "dev-backend", "profile": "claude-code" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(response.ok, "expected ok, got {response:?}");
|
|
assert_eq!(response.action.as_deref(), Some("spawn_agent"));
|
|
// Request consumed; a response sibling written.
|
|
assert!(!req.exists(), "request file must be removed");
|
|
let on_disk = read_response(&req);
|
|
assert!(on_disk.ok);
|
|
// The agent was actually created through the use cases.
|
|
assert_eq!(contexts.entries().len(), 1);
|
|
assert_eq!(contexts.entries()[0].name, "dev-backend");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn valid_agent_run_request_succeeds_and_reports_type() {
|
|
let tmp = TempDir::new();
|
|
let contexts = FakeContexts::new();
|
|
let service = build_service(contexts.clone());
|
|
|
|
let req = tmp.0.join("req-agent-run.json");
|
|
std::fs::write(
|
|
&req,
|
|
br#"{ "type": "agent.run", "requestedBy": "Main", "targetAgent": "architect", "profile": "claude-code", "task": "Analyse", "visibility": "background" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(response.ok, "expected ok, got {response:?}");
|
|
assert_eq!(response.action.as_deref(), Some("agent.run"));
|
|
assert!(!req.exists(), "request file must be removed");
|
|
let on_disk = read_response(&req);
|
|
assert!(on_disk.ok);
|
|
assert_eq!(contexts.entries().len(), 1);
|
|
assert_eq!(contexts.entries()[0].name, "architect");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn valid_create_skill_request_succeeds_and_is_consumed() {
|
|
let tmp = TempDir::new();
|
|
let service = build_service(FakeContexts::new());
|
|
|
|
let req = tmp.0.join("skill-req.json");
|
|
std::fs::write(
|
|
&req,
|
|
br##"{ "action": "create_skill", "name": "deploy", "context": "# Deploy steps" }"##,
|
|
)
|
|
.unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(response.ok, "expected ok, got {response:?}");
|
|
assert_eq!(response.action.as_deref(), Some("create_skill"));
|
|
// Request consumed; a response sibling written and marked ok.
|
|
assert!(!req.exists(), "request file must be removed");
|
|
assert!(read_response(&req).ok);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn invalid_json_request_yields_error_response() {
|
|
let tmp = TempDir::new();
|
|
let service = build_service(FakeContexts::new());
|
|
|
|
let req = tmp.0.join("broken.json");
|
|
std::fs::write(&req, b"{ this is not json").unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(!response.ok);
|
|
assert!(
|
|
response
|
|
.error
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("invalid json"),
|
|
"got {response:?}"
|
|
);
|
|
assert!(!req.exists(), "poisoned request must still be removed");
|
|
assert!(!read_response(&req).ok);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn unknown_action_yields_error_response() {
|
|
let tmp = TempDir::new();
|
|
let service = build_service(FakeContexts::new());
|
|
|
|
let req = tmp.0.join("weird.json");
|
|
std::fs::write(&req, br#"{ "action": "explode", "name": "x" }"#).unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(!response.ok);
|
|
assert_eq!(response.action.as_deref(), Some("explode"));
|
|
assert!(response
|
|
.error
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("unknown orchestrator action"));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// LOT D6b — `reply` surfaced into the `*.response.json` wire format.
|
|
//
|
|
// D6b makes `OrchestratorResponse` carry an optional `reply` (the queried agent's
|
|
// content for `ask`/`agent.message`), `#[serde(skip_serializing_if = "Option::is_none")]`
|
|
// so a response with no reply keeps the exact legacy shape.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Point 1 — an `agent.message` (`ask`) whose target replies ⇒ the on-disk
|
|
/// `*.response.json` carries the `reply` content **and** `detail` (both coexist),
|
|
/// alongside `ok: true`.
|
|
#[tokio::test]
|
|
async fn ask_request_surfaces_reply_alongside_detail() {
|
|
// B6: an `agent.message` request is still synchronous for the caller, but the
|
|
// target turn is driven through the structured/headless session. The mailbox still
|
|
// receives exactly one silent ticket for turn accounting; the response comes from
|
|
// `ReplyEvent::Final`, not from a separate `agent.reply` request.
|
|
let tmp = TempDir::new();
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, mailbox, _sessions, structured_reply, tasks) =
|
|
build_service_with_mailbox(contexts.clone());
|
|
|
|
let req = tmp.0.join("ask-req.json");
|
|
std::fs::write(
|
|
&req,
|
|
br#"{ "type": "agent.message", "requestedBy": "Main", "targetAgent": "architect", "task": "What is the answer?" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let svc = Arc::clone(&service);
|
|
let proj = project();
|
|
let ask_req = req.clone();
|
|
let ask = tokio::spawn(async move { process_request_file(&ask_req, &proj, &svc).await });
|
|
|
|
// Wait until the ask has enqueued its ticket (blocked awaiting the reply).
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
while mailbox.pending(&agent_id) == 0 {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("ask must enqueue a ticket");
|
|
|
|
structured_reply.release();
|
|
|
|
let response = tokio::time::timeout(std::time::Duration::from_secs(10), ask)
|
|
.await
|
|
.expect("ask completes after reply")
|
|
.expect("join ok");
|
|
|
|
assert!(response.ok, "expected ok, got {response:?}");
|
|
assert_eq!(response.action.as_deref(), Some("agent.message"));
|
|
// reply carries the target's rendered result...
|
|
assert_eq!(response.reply.as_deref(), Some("the answer is 42"));
|
|
// ...and detail still summarises what IdeA did (the two coexist).
|
|
assert!(
|
|
response
|
|
.detail
|
|
.as_deref()
|
|
.unwrap_or_default()
|
|
.contains("architect"),
|
|
"detail should still be present: {response:?}"
|
|
);
|
|
|
|
// The on-disk wire format actually contains both keys with the right values.
|
|
let on_disk = read_response(&req);
|
|
assert!(on_disk.ok);
|
|
assert_eq!(on_disk.reply.as_deref(), Some("the answer is 42"));
|
|
assert!(on_disk.detail.is_some());
|
|
|
|
let raw = read_response_raw(&req);
|
|
let json: serde_json::Value = serde_json::from_str(&raw).unwrap();
|
|
assert_eq!(json["reply"], serde_json::json!("the answer is 42"));
|
|
assert!(json.get("detail").is_some());
|
|
assert!(!req.exists(), "request file must be removed");
|
|
|
|
let stored = tasks.all();
|
|
assert_eq!(stored.len(), 1, "one rendezvous task persisted: {stored:?}");
|
|
let task = &stored[0];
|
|
assert_eq!(task.owner_agent_id, agent_id);
|
|
assert_eq!(task.state, BackgroundTaskState::Completed);
|
|
assert!(matches!(
|
|
task.kind,
|
|
domain::BackgroundTaskKind::HeadlessRendezvous { .. }
|
|
));
|
|
assert!(matches!(
|
|
task.result,
|
|
Some(BackgroundTaskResult::Success { .. })
|
|
));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn ask_request_no_reply_persists_failed_rendezvous_task() {
|
|
let tmp = TempDir::new();
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, mailbox, _sessions, structured_reply, tasks) =
|
|
build_service_with_mailbox(contexts.clone());
|
|
structured_reply.no_reply();
|
|
|
|
let req = tmp.0.join("ask-no-reply.json");
|
|
std::fs::write(
|
|
&req,
|
|
br#"{ "type": "agent.message", "requestedBy": "Main", "targetAgent": "architect", "task": "Please forget to finalise" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let svc = Arc::clone(&service);
|
|
let proj = project();
|
|
let ask_req = req.clone();
|
|
let ask = tokio::spawn(async move { process_request_file(&ask_req, &proj, &svc).await });
|
|
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), async {
|
|
while mailbox.pending(&agent_id) == 0 {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("ask must enqueue a ticket before no-reply");
|
|
|
|
structured_reply.release();
|
|
|
|
let response = tokio::time::timeout(std::time::Duration::from_secs(10), ask)
|
|
.await
|
|
.expect("ask completes after no-reply")
|
|
.expect("join ok");
|
|
assert!(!response.ok, "expected no-reply error, got {response:?}");
|
|
|
|
let stored = tasks.all();
|
|
assert_eq!(stored.len(), 1, "one rendezvous task persisted: {stored:?}");
|
|
let task = &stored[0];
|
|
assert_eq!(task.owner_agent_id, agent_id);
|
|
assert_eq!(task.state, BackgroundTaskState::Failed);
|
|
match task.result.as_ref() {
|
|
Some(BackgroundTaskResult::Failure { error, .. }) => {
|
|
assert!(
|
|
error.contains("NoReply"),
|
|
"error should classify no-reply: {error}"
|
|
);
|
|
}
|
|
other => panic!("expected failure result, got {other:?}"),
|
|
}
|
|
assert_eq!(mailbox.pending(&agent_id), 0, "turn-lock mailbox is freed");
|
|
}
|
|
|
|
/// Point 2 — a non-`ask` command (here `spawn_agent`, reply `None`) ⇒ the `reply`
|
|
/// key is **absent** from the serialised JSON (`skip_serializing_if`). Anti-always-green:
|
|
/// we assert on the *literal absence* of the key in the raw bytes, so the test would
|
|
/// fail if `reply: null` were emitted instead.
|
|
#[tokio::test]
|
|
async fn non_ask_request_omits_reply_key() {
|
|
let tmp = TempDir::new();
|
|
let contexts = FakeContexts::new();
|
|
let service = build_service(contexts.clone());
|
|
|
|
let req = tmp.0.join("spawn-no-reply.json");
|
|
std::fs::write(
|
|
&req,
|
|
br#"{ "action": "spawn_agent", "name": "dev-backend", "profile": "claude-code" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(response.ok, "expected ok, got {response:?}");
|
|
assert!(response.reply.is_none(), "spawn must not produce a reply");
|
|
|
|
// Deserialised: reply is None.
|
|
let on_disk = read_response(&req);
|
|
assert!(on_disk.reply.is_none());
|
|
|
|
// Raw wire: the "reply" key must not appear at all. (Guard the assertion itself:
|
|
// confirm we ARE looking at the right file by checking a key we know is present.)
|
|
let raw = read_response_raw(&req);
|
|
assert!(
|
|
raw.contains("\"ok\""),
|
|
"sanity: response JSON should contain ok — got {raw}"
|
|
);
|
|
assert!(
|
|
!raw.contains("reply"),
|
|
"reply key must be skipped when None — got {raw}"
|
|
);
|
|
// And via the parsed tree, the key is structurally absent (not null).
|
|
let json: serde_json::Value = serde_json::from_str(&raw).unwrap();
|
|
assert!(
|
|
json.get("reply").is_none(),
|
|
"reply key must be structurally absent — got {json}"
|
|
);
|
|
}
|
|
|
|
/// Point 3 — round-trip a legacy `*.response.json` written **before** D6b (no
|
|
/// `reply` key): it deserialises to `reply == None`, and re-serialising does **not**
|
|
/// reintroduce the key (forward/backward wire compatibility).
|
|
#[test]
|
|
fn legacy_response_without_reply_round_trips_without_reintroducing_key() {
|
|
// A response payload exactly as a pre-D6b IdeA would have written it.
|
|
let legacy =
|
|
r#"{"ok":true,"action":"spawn_agent","detail":"launched agent dev-backend in background"}"#;
|
|
|
|
let parsed: OrchestratorResponse =
|
|
serde_json::from_str(legacy).expect("legacy response must still parse");
|
|
assert!(parsed.ok);
|
|
assert_eq!(parsed.action.as_deref(), Some("spawn_agent"));
|
|
assert!(
|
|
parsed.reply.is_none(),
|
|
"absent reply key must deserialise to None"
|
|
);
|
|
|
|
// Re-serialising must not bring a "reply" key back.
|
|
let reserialised = serde_json::to_string(&parsed).unwrap();
|
|
assert!(
|
|
!reserialised.contains("reply"),
|
|
"round-trip must not reintroduce the reply key — got {reserialised}"
|
|
);
|
|
let json: serde_json::Value = serde_json::from_str(&reserialised).unwrap();
|
|
assert!(json.get("reply").is_none());
|
|
}
|
|
|
|
/// Point 4 — a failure response carries no `reply` (the field is `None` for every
|
|
/// `failure(..)`), so the key is absent from the failing `*.response.json` too.
|
|
#[tokio::test]
|
|
async fn failure_response_omits_reply_key() {
|
|
let tmp = TempDir::new();
|
|
let service = build_service(FakeContexts::new());
|
|
|
|
let req = tmp.0.join("broken-no-reply.json");
|
|
std::fs::write(&req, b"{ not json at all").unwrap();
|
|
|
|
let response = process_request_file(&req, &project(), &service).await;
|
|
|
|
assert!(!response.ok);
|
|
assert!(response.reply.is_none(), "failure must not carry a reply");
|
|
|
|
let raw = read_response_raw(&req);
|
|
assert!(
|
|
raw.contains("\"error\""),
|
|
"sanity: a failure JSON should contain error — got {raw}"
|
|
);
|
|
assert!(
|
|
!raw.contains("reply"),
|
|
"reply key must be absent on failure — got {raw}"
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// M4 — observability: the filesystem watcher tags processed requests source=File.
|
|
//
|
|
// Drives the *public* `FsOrchestratorWatcher::start` against a real temp project
|
|
// root with a capturing events closure (the same sink shape the composition root
|
|
// wires), then polls (bounded) until the OrchestratorRequestProcessed beacon for
|
|
// the dropped request lands. Asserts the tag is `OrchestrationSource::File`.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn watcher_publishes_processed_event_tagged_file_source() {
|
|
use std::time::Duration;
|
|
|
|
let tmp = TempDir::new();
|
|
// A project rooted at the temp dir so the watcher scans <tmp>/.ideai/requests/.
|
|
let project = Project::new(
|
|
ProjectId::from_uuid(Uuid::from_u128(2000)),
|
|
"demo",
|
|
ProjectPath::new(tmp.0.to_str().unwrap()).unwrap(),
|
|
RemoteRef::local(),
|
|
1_700_000_000_000,
|
|
)
|
|
.unwrap();
|
|
let service = build_service(FakeContexts::new());
|
|
|
|
let captured = Arc::new(Mutex::new(Vec::<DomainEvent>::new()));
|
|
let sink = captured.clone();
|
|
let publish: Arc<dyn Fn(DomainEvent) + Send + Sync> =
|
|
Arc::new(move |e| sink.lock().unwrap().push(e));
|
|
|
|
let handle = FsOrchestratorWatcher::start(project, Arc::clone(&service), publish);
|
|
|
|
// Drop a valid request under .ideai/requests/<requester>/.
|
|
let req_dir = tmp.0.join(".ideai").join("requests").join("Main");
|
|
std::fs::create_dir_all(&req_dir).unwrap();
|
|
std::fs::write(
|
|
req_dir.join("req-1.json"),
|
|
br#"{ "action": "spawn_agent", "name": "dev-backend", "profile": "claude-code" }"#,
|
|
)
|
|
.unwrap();
|
|
|
|
// Poll (bounded) until the processed beacon for our request shows up.
|
|
let mut found: Option<DomainEvent> = None;
|
|
for _ in 0..50 {
|
|
if let Some(e) = captured.lock().unwrap().iter().find(|e| {
|
|
matches!(e, DomainEvent::OrchestratorRequestProcessed { action, .. } if action == "spawn_agent")
|
|
}) {
|
|
found = Some(e.clone());
|
|
break;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
handle.stop();
|
|
|
|
let event = found.expect("watcher must publish a processed beacon within the poll budget");
|
|
match event {
|
|
DomainEvent::OrchestratorRequestProcessed {
|
|
source,
|
|
ok,
|
|
requester_id,
|
|
..
|
|
} => {
|
|
assert_eq!(source, OrchestrationSource::File, "fs door must tag File");
|
|
assert!(ok, "the spawn request succeeded");
|
|
assert_eq!(requester_id, "Main", "requester id = request subdirectory");
|
|
}
|
|
other => panic!("expected OrchestratorRequestProcessed, got {other:?}"),
|
|
}
|
|
}
|