Blomios
287681c198
Introduit le modèle AgentManifest { version, entries, orchestrator } et la
garde d'écriture directe may_write_directly(..., &OrchestratorDesignation) :
seul l'orchestrateur désigné peut écrire directement, les autres passent par
le rendez-vous médié. Câble la désignation à travers domain → application →
infrastructure → app-tauri (context_guard, service, lifecycle, ports).
Ajoute crates/application/src/diag.rs : sink de diagnostic best-effort, sans
dépendance, qui miroite les traces du rendez-vous inter-agents de
l'orchestrateur vers un fichier de log persistant (utile au lancement via
AppImage où stderr est jeté), avec la même discipline « zéro dépendance,
ne casse jamais le rendez-vous ».
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1252 lines
46 KiB
Rust
1252 lines
46 KiB
Rust
//! Functional unit tests for the IdeA **MCP server** adapter (lot **M2**,
|
|
//! cadrage `orchestration-v3` §3).
|
|
//!
|
|
//! These drive [`McpServer`] **entirely off-network** — no socket, no child
|
|
//! process — over either [`McpServer::handle_raw`] (one raw JSON-RPC message →
|
|
//! its response) or [`MemoryTransport`] (a scripted in-memory session). The server
|
|
//! is wired over the **same** in-memory fakes the filesystem-watcher tests already
|
|
//! use (`orchestrator_watcher.rs`), so we assert MCP behaviour against a real
|
|
//! [`OrchestratorService`] without any I/O:
|
|
//!
|
|
//! 1. `tools/list` advertises the six `idea_*` tools, each with an input schema.
|
|
//! 2. `tools/call` maps every tool to the right `OrchestratorCommand` (observed
|
|
//! through the fakes: spawn creates an agent, stop closes the session, …).
|
|
//! 3. `idea_ask_agent` returns the target's `reply` **inline**.
|
|
//! 4. `idea_list_agents` returns the agent list **inline** as a JSON array.
|
|
//! 5. A failed IdeA command ⇒ tool result `isError: true` (not a transport error,
|
|
//! not a panic); the server stays alive for the next call.
|
|
//! 6. Malformed JSON-RPC ⇒ a typed `PARSE_ERROR`, never a panic.
|
|
//! 7. Unknown method / unknown tool ⇒ typed errors, never a panic.
|
|
//! 8. `initialize` answers the minimal handshake.
|
|
//! 9. A `tools/call` missing a required argument is rejected by the **same**
|
|
//! `OrchestratorRequest::validate` (typed error, **no** dispatch).
|
|
|
|
use std::collections::HashMap;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use async_trait::async_trait;
|
|
use domain::agent::{AgentManifest, ManifestEntry};
|
|
use domain::events::{DomainEvent, OrchestrationSource};
|
|
use domain::ids::NodeId;
|
|
use domain::ids::SkillId;
|
|
use domain::ids::{AgentId, ProfileId, ProjectId};
|
|
use domain::markdown::MarkdownDoc;
|
|
use domain::ports::{
|
|
AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream,
|
|
ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, ProfileStore,
|
|
PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec,
|
|
StoreError,
|
|
};
|
|
use domain::profile::{
|
|
AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport,
|
|
StructuredAdapter,
|
|
};
|
|
use domain::project::{Project, ProjectPath};
|
|
use domain::remote::RemoteRef;
|
|
use domain::skill::{Skill, SkillScope};
|
|
use domain::terminal::{SessionKind, TerminalSession};
|
|
use domain::{PtySize, SessionId};
|
|
use uuid::Uuid;
|
|
|
|
use application::{
|
|
CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
|
|
OrchestratorService, TerminalSessions, UpdateAgentContext,
|
|
};
|
|
use infrastructure::orchestrator::mcp::jsonrpc::{error_codes, Transport, TransportError};
|
|
use infrastructure::{
|
|
InMemoryConversationRegistry, InMemoryMailbox, McpServer, MediatedInbox, MemoryTransport,
|
|
SystemMillisClock,
|
|
};
|
|
|
|
use serde_json::{json, Value};
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Fakes — mirrored from `orchestrator_watcher.rs` so the MCP adapter is tested
|
|
// against the exact same OrchestratorService wiring (no use-case is re-invented).
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[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`.
|
|
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(())
|
|
}
|
|
}
|
|
|
|
#[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(())
|
|
}
|
|
}
|
|
|
|
#[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;
|
|
impl AgentRuntime for FakeRuntime {
|
|
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
|
|
}
|
|
}
|
|
|
|
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()
|
|
}
|
|
|
|
/// Builds an [`OrchestratorService`] over the in-memory fakes (no structured
|
|
/// registry). Returns the shared `TerminalSessions` so a test can pre-bind a live
|
|
/// PTY session for an agent (needed to make `stop_agent` succeed).
|
|
fn build_service(contexts: FakeContexts) -> (Arc<OrchestratorService>, Arc<TerminalSessions>) {
|
|
// 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))),
|
|
));
|
|
let service = Arc::new(OrchestratorService::new(
|
|
create,
|
|
launch,
|
|
list,
|
|
close,
|
|
update,
|
|
create_skill,
|
|
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
|
Arc::clone(&sessions),
|
|
));
|
|
(service, sessions)
|
|
}
|
|
|
|
/// Like [`build_service`] but with the **inter-agent mailbox** + PTY wired (Option 1
|
|
/// « Terminal + MCP », B-3/B-4), so `idea_ask_agent` blocks on the mailbox and
|
|
/// `idea_reply` resolves it. Returns the service, the shared mailbox (to observe
|
|
/// pending tickets) and the PTY session registry (to pre-seed a live target).
|
|
fn build_service_with_mailbox(
|
|
contexts: FakeContexts,
|
|
) -> (
|
|
Arc<OrchestratorService>,
|
|
Arc<InMemoryMailbox>,
|
|
Arc<TerminalSessions>,
|
|
) {
|
|
let mailbox = Arc::new(InMemoryMailbox::new());
|
|
let (base, sessions) = build_service(contexts);
|
|
let input = Arc::new(MediatedInbox::with_pty(
|
|
Arc::clone(&mailbox),
|
|
Arc::new(SystemMillisClock),
|
|
Arc::new(FakePty) as Arc<dyn PtyPort>,
|
|
)) as Arc<dyn domain::input::InputMediator>;
|
|
let conversations = Arc::new(InMemoryConversationRegistry::new())
|
|
as Arc<dyn domain::conversation::ConversationRegistry>;
|
|
let service = Arc::try_unwrap(base)
|
|
.unwrap_or_else(|_| panic!("freshly built service must be uniquely owned"))
|
|
.with_input_mediator(
|
|
input,
|
|
Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
|
|
)
|
|
.with_conversations(conversations);
|
|
(Arc::new(service), mailbox, sessions)
|
|
}
|
|
|
|
/// Pre-seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it.
|
|
fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) {
|
|
sessions.insert(
|
|
PtyHandle { session_id },
|
|
TerminalSession::starting(
|
|
session_id,
|
|
NodeId::from_uuid(Uuid::from_u128(7)),
|
|
ProjectPath::new("/home/me/proj").unwrap(),
|
|
SessionKind::Agent { agent_id },
|
|
PtySize { rows: 24, cols: 80 },
|
|
),
|
|
);
|
|
}
|
|
|
|
fn server(service: Arc<OrchestratorService>) -> McpServer {
|
|
McpServer::new(service, project())
|
|
}
|
|
|
|
/// A capturing event sink (the MCP twin of the file watcher's publish closure):
|
|
/// records every [`DomainEvent`] the server emits so a test can assert the
|
|
/// `OrchestratorRequestProcessed` source tag. Returns the closure to wire via
|
|
/// `with_events` and the shared buffer to inspect afterwards.
|
|
fn capturing_events() -> (
|
|
Arc<dyn Fn(DomainEvent) + Send + Sync>,
|
|
Arc<Mutex<Vec<DomainEvent>>>,
|
|
) {
|
|
let captured = Arc::new(Mutex::new(Vec::new()));
|
|
let sink = captured.clone();
|
|
let publish: Arc<dyn Fn(DomainEvent) + Send + Sync> =
|
|
Arc::new(move |e: DomainEvent| sink.lock().unwrap().push(e));
|
|
(publish, captured)
|
|
}
|
|
|
|
// --- JSON-RPC framing helpers ---------------------------------------------
|
|
|
|
/// A `tools/call` request envelope for `tool` with `arguments`.
|
|
fn tools_call(id: i64, tool: &str, arguments: Value) -> Vec<u8> {
|
|
serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0",
|
|
"id": id,
|
|
"method": "tools/call",
|
|
"params": { "name": tool, "arguments": arguments }
|
|
}))
|
|
.unwrap()
|
|
}
|
|
|
|
/// Extracts the single text content block of a successful `tools/call` result.
|
|
fn result_text(result: &Value) -> &str {
|
|
result["content"][0]["text"]
|
|
.as_str()
|
|
.expect("text content block")
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 1. tools/list catalogue
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn tools_list_advertises_the_seven_idea_tools_with_schemas() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let server = server(service);
|
|
|
|
let raw = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 1, "method": "tools/list"
|
|
}))
|
|
.unwrap();
|
|
|
|
let response = server
|
|
.handle_raw(&raw)
|
|
.await
|
|
.expect("tools/list owes a reply");
|
|
assert!(response.error.is_none(), "got error: {:?}", response.error);
|
|
let result = response.result.expect("result");
|
|
|
|
let tools = result["tools"].as_array().expect("tools array");
|
|
let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect();
|
|
|
|
for expected in [
|
|
"idea_list_agents",
|
|
"idea_ask_agent",
|
|
"idea_reply",
|
|
"idea_launch_agent",
|
|
"idea_stop_agent",
|
|
"idea_update_context",
|
|
"idea_create_skill",
|
|
// FileGuard-mediated context/memory tools (cadrage C7).
|
|
"idea_context_read",
|
|
"idea_context_propose",
|
|
"idea_memory_read",
|
|
"idea_memory_write",
|
|
] {
|
|
assert!(
|
|
names.contains(&expected),
|
|
"missing tool {expected}; got {names:?}"
|
|
);
|
|
}
|
|
assert_eq!(
|
|
tools.len(),
|
|
11,
|
|
"exactly the eleven idea_* tools (7 base + 4 FileGuard C7); got {names:?}"
|
|
);
|
|
|
|
// Every tool advertises an object input schema.
|
|
for t in tools {
|
|
assert_eq!(
|
|
t["inputSchema"]["type"],
|
|
json!("object"),
|
|
"tool {} must declare an object input schema",
|
|
t["name"]
|
|
);
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 2. tools/call → the right OrchestratorCommand (observed through the fakes)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn launch_agent_call_creates_and_launches_the_agent() {
|
|
let contexts = FakeContexts::new();
|
|
let (service, _s) = build_service(contexts.clone());
|
|
let server = server(service);
|
|
|
|
// Agent unknown → SpawnAgent creates it from scratch then launches it.
|
|
let raw = tools_call(
|
|
1,
|
|
"idea_launch_agent",
|
|
json!({ "target": "dev-backend", "profile": "claude-code" }),
|
|
);
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert!(
|
|
response.error.is_none(),
|
|
"transport error: {:?}",
|
|
response.error
|
|
);
|
|
let result = response.result.expect("result");
|
|
assert_eq!(result["isError"], json!(false), "got {result}");
|
|
|
|
// The command really reached the use cases: the agent now exists in the manifest.
|
|
assert_eq!(contexts.entries().len(), 1);
|
|
assert_eq!(contexts.entries()[0].name, "dev-backend");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn stop_agent_call_closes_the_live_session() {
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("dev");
|
|
let (service, sessions) = build_service(contexts);
|
|
// Pre-bind a live PTY session for the agent so StopAgent has something to close.
|
|
let session_id = SessionId::from_uuid(Uuid::from_u128(555));
|
|
sessions.insert(
|
|
PtyHandle { session_id },
|
|
TerminalSession::starting(
|
|
session_id,
|
|
NodeId::from_uuid(Uuid::from_u128(8)),
|
|
ProjectPath::new("/home/me/proj").unwrap(),
|
|
SessionKind::Agent { agent_id },
|
|
PtySize { rows: 24, cols: 80 },
|
|
),
|
|
);
|
|
assert!(sessions.session_for_agent(&agent_id).is_some());
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(1, "idea_stop_agent", json!({ "target": "dev" }));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
let result = response.result.expect("result");
|
|
assert_eq!(result["isError"], json!(false), "got {result}");
|
|
|
|
// CloseTerminal ran → the agent no longer has a live session.
|
|
assert!(
|
|
sessions.session_for_agent(&agent_id).is_none(),
|
|
"stop_agent should have removed the session"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn update_context_and_create_skill_calls_succeed() {
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("dev");
|
|
// Seed an existing body so write_context finds the md path.
|
|
contexts
|
|
.0
|
|
.lock()
|
|
.unwrap()
|
|
.contents
|
|
.insert(format!("agents/dev.md"), "# old".to_owned());
|
|
let _ = agent_id;
|
|
let (service, _s) = build_service(contexts.clone());
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(
|
|
1,
|
|
"idea_update_context",
|
|
json!({ "target": "dev", "context": "# new body" }),
|
|
);
|
|
let r = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert_eq!(r.result.expect("result")["isError"], json!(false));
|
|
|
|
let raw = tools_call(
|
|
2,
|
|
"idea_create_skill",
|
|
json!({ "name": "deploy", "context": "# steps" }),
|
|
);
|
|
let r = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert_eq!(r.result.expect("result")["isError"], json!(false));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 3. idea_ask_agent returns the reply inline
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn ask_agent_returns_target_reply_inline() {
|
|
// Option 1 (B-3/B-4): `idea_ask_agent` writes the task into the target's live
|
|
// terminal and blocks on the mailbox; the target's `idea_reply` (carrying its
|
|
// handshake identity as requester) resolves it, and the ask returns the result
|
|
// inline. We drive both over the MCP server, the ask on a spawned task.
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, mailbox, sessions) = build_service_with_mailbox(contexts);
|
|
seed_live_pty(
|
|
&sessions,
|
|
agent_id,
|
|
SessionId::from_uuid(Uuid::from_u128(4242)),
|
|
);
|
|
|
|
let ask_server = server(Arc::clone(&service));
|
|
let ask = tokio::spawn(async move {
|
|
let raw = tools_call(
|
|
7,
|
|
"idea_ask_agent",
|
|
json!({ "target": "architect", "task": "What is the answer?" }),
|
|
);
|
|
ask_server.handle_raw(&raw).await.expect("reply owed")
|
|
});
|
|
|
|
// Wait until the ask has enqueued its ticket (it is now blocked awaiting a 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");
|
|
|
|
// The target replies via idea_reply — its identity comes from the handshake
|
|
// requester, which `for_requester` injects as the `from` of the Reply command.
|
|
let reply_server = server(Arc::clone(&service)).for_requester(agent_id.to_string());
|
|
let reply_raw = tools_call(8, "idea_reply", json!({ "result": "the answer is 42" }));
|
|
let reply_resp = reply_server
|
|
.handle_raw(&reply_raw)
|
|
.await
|
|
.expect("reply owed");
|
|
assert_eq!(reply_resp.result.expect("result")["isError"], json!(false));
|
|
|
|
let response = tokio::time::timeout(std::time::Duration::from_secs(10), ask)
|
|
.await
|
|
.expect("ask completes after reply")
|
|
.expect("join ok");
|
|
assert_eq!(response.id, json!(7), "id must be echoed");
|
|
assert!(
|
|
response.error.is_none(),
|
|
"transport error: {:?}",
|
|
response.error
|
|
);
|
|
let result = response.result.expect("result");
|
|
assert_eq!(result["isError"], json!(false), "got {result}");
|
|
assert_eq!(
|
|
result_text(&result),
|
|
"the answer is 42",
|
|
"ask reply must be returned inline; got {result}"
|
|
);
|
|
}
|
|
|
|
/// `idea_reply` with no in-flight ask for the connected peer ⇒ the IdeA command
|
|
/// fails, surfaced as a tool execution error (`isError: true`), never a panic.
|
|
#[tokio::test]
|
|
async fn reply_without_pending_ask_is_a_tool_error() {
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, _mailbox, _sessions) = build_service_with_mailbox(contexts);
|
|
let reply_server = server(service).for_requester(agent_id.to_string());
|
|
|
|
let raw = tools_call(9, "idea_reply", json!({ "result": "orphan" }));
|
|
let resp = reply_server.handle_raw(&raw).await.expect("reply owed");
|
|
// Mapped + dispatched, but the command failed ⇒ tool error, transport intact.
|
|
assert!(resp.error.is_none(), "no transport error: {:?}", resp.error);
|
|
assert_eq!(resp.result.expect("result")["isError"], json!(true));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 4. idea_list_agents returns the agent list inline (JSON array)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn list_agents_returns_the_agents_inline_as_json_array() {
|
|
let contexts = FakeContexts::new();
|
|
contexts.seed_agent("architect");
|
|
contexts.seed_agent("dev-backend");
|
|
let (service, _s) = build_service(contexts);
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(3, "idea_list_agents", json!({}));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
let result = response.result.expect("result");
|
|
assert_eq!(result["isError"], json!(false), "got {result}");
|
|
|
|
// The inline text is the JSON array of the project's agents.
|
|
let text = result_text(&result);
|
|
let agents: Value = serde_json::from_str(text).expect("reply must be a JSON array");
|
|
let arr = agents.as_array().expect("array");
|
|
assert_eq!(arr.len(), 2, "two seeded agents expected; got {text}");
|
|
let names: Vec<&str> = arr.iter().map(|a| a["name"].as_str().unwrap()).collect();
|
|
assert!(names.contains(&"architect"));
|
|
assert!(names.contains(&"dev-backend"));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 5. A failed IdeA command ⇒ isError: true (not a transport error, no panic)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn failed_command_is_tool_error_not_transport_error_and_server_survives() {
|
|
// stop_agent on a seeded-but-not-running agent → AppError::NotFound from the
|
|
// use case → the tool result is an execution error, the JSON-RPC layer is fine.
|
|
let contexts = FakeContexts::new();
|
|
contexts.seed_agent("idle");
|
|
let (service, _s) = build_service(contexts);
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(1, "idea_stop_agent", json!({ "target": "idle" }));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
// Healthy connection: no JSON-RPC error...
|
|
assert!(
|
|
response.error.is_none(),
|
|
"a failed command must NOT be a JSON-RPC transport error: {:?}",
|
|
response.error
|
|
);
|
|
// ...but the tool result is flagged as an error.
|
|
let result = response.result.expect("result");
|
|
assert_eq!(result["isError"], json!(true), "got {result}");
|
|
assert!(
|
|
!result_text(&result).is_empty(),
|
|
"error text should describe the failure"
|
|
);
|
|
|
|
// The server is still alive: a subsequent valid call still answers.
|
|
let raw = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 2, "method": "tools/list"
|
|
}))
|
|
.unwrap();
|
|
let again = server.handle_raw(&raw).await.expect("server still serves");
|
|
assert!(
|
|
again.result.is_some(),
|
|
"server must survive a failed command"
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 6. Malformed JSON-RPC ⇒ typed PARSE_ERROR, never a panic
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn malformed_json_yields_parse_error_no_panic() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let server = server(service);
|
|
|
|
let response = server
|
|
.handle_raw(b"{ this is not json")
|
|
.await
|
|
.expect("a parse error still owes a response");
|
|
let error = response.error.expect("parse error expected");
|
|
assert_eq!(error.code, error_codes::PARSE_ERROR);
|
|
// JSON-RPC mandates a null id when the request could not be correlated.
|
|
assert_eq!(response.id, Value::Null);
|
|
assert!(response.result.is_none());
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 7. Unknown method / unknown tool ⇒ typed errors, no panic
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn unknown_method_yields_method_not_found() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let server = server(service);
|
|
|
|
let raw = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 9, "method": "resources/list"
|
|
}))
|
|
.unwrap();
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
let error = response.error.expect("method-not-found expected");
|
|
assert_eq!(error.code, error_codes::METHOD_NOT_FOUND);
|
|
assert_eq!(response.id, json!(9), "id echoed even on error");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn unknown_tool_yields_typed_error() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(4, "idea_made_up_tool", json!({}));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
let error = response.error.expect("unknown-tool expected");
|
|
// An unknown tool maps to METHOD_NOT_FOUND (see server::map_err_to_jsonrpc).
|
|
assert_eq!(error.code, error_codes::METHOD_NOT_FOUND);
|
|
assert!(
|
|
error.message.contains("unknown tool"),
|
|
"message should name the cause; got {}",
|
|
error.message
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 8. initialize handshake
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn initialize_answers_minimal_handshake() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let server = server(service);
|
|
|
|
let raw = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 0, "method": "initialize",
|
|
"params": { "protocolVersion": "2024-11-05" }
|
|
}))
|
|
.unwrap();
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert!(response.error.is_none());
|
|
let result = response.result.expect("result");
|
|
assert!(
|
|
result["protocolVersion"].is_string(),
|
|
"handshake must advertise a protocol version; got {result}"
|
|
);
|
|
// Capability for tools is advertised, and the server identifies itself.
|
|
assert!(result["capabilities"]["tools"].is_object());
|
|
assert_eq!(result["serverInfo"]["name"], json!("idea-orchestrator"));
|
|
}
|
|
|
|
/// Readiness de démarrage (fix race cold-launch, signal MCP) : un `initialize` reçu sur
|
|
/// un serveur per-connexion (`for_requester(id)`) déclenche le `ready_sink` avec l'id du
|
|
/// peer. Et un serveur SANS requester (anonyme) ne déclenche PAS le sink (peer legacy).
|
|
#[tokio::test]
|
|
async fn initialize_fires_ready_sink_with_requester() {
|
|
let (service, _s) = build_service(FakeContexts::new());
|
|
let captured: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
|
|
let sink_buf = Arc::clone(&captured);
|
|
let ready_sink: Arc<dyn Fn(&str) + Send + Sync> =
|
|
Arc::new(move |req: &str| sink_buf.lock().unwrap().push(req.to_owned()));
|
|
|
|
let base = McpServer::new(service, project()).with_ready_sink(ready_sink);
|
|
|
|
let init = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 0, "method": "initialize",
|
|
"params": { "protocolVersion": "2024-11-05" }
|
|
}))
|
|
.unwrap();
|
|
|
|
// Cas 1 : peer identifié (requester non vide) ⇒ le sink reçoit son id.
|
|
let peer = base.for_requester("agent-42");
|
|
let response = peer.handle_raw(&init).await.expect("reply owed");
|
|
assert!(response.error.is_none(), "initialize must still answer");
|
|
assert_eq!(
|
|
*captured.lock().unwrap(),
|
|
vec!["agent-42".to_owned()],
|
|
"initialize d'un peer identifié ⇒ ready_sink appelé avec son requester"
|
|
);
|
|
|
|
// Cas 2 : peer anonyme (requester vide, le serveur de base) ⇒ sink NON appelé.
|
|
captured.lock().unwrap().clear();
|
|
let response = base.handle_raw(&init).await.expect("reply owed");
|
|
assert!(response.error.is_none());
|
|
assert!(
|
|
captured.lock().unwrap().is_empty(),
|
|
"requester vide ⇒ ready_sink jamais appelé (peer legacy/anonyme)"
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 9. Shared validation: a tools/call missing a required argument is rejected by
|
|
// the SAME OrchestratorRequest::validate, with no dispatch.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn ask_agent_missing_task_is_rejected_by_shared_validation_no_dispatch() {
|
|
let contexts = FakeContexts::new();
|
|
contexts.seed_agent("architect");
|
|
// If validation wrongly let this through to dispatch, ask_agent would try to
|
|
// launch/contact the target and the error would differ. A validation rejection
|
|
// here is INVALID_PARAMS, before dispatch.
|
|
let (service, _mailbox, _sessions) = build_service_with_mailbox(contexts);
|
|
let server = server(service);
|
|
|
|
let raw = tools_call(5, "idea_ask_agent", json!({ "target": "architect" }));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
|
|
// Rejected at the mapping/validation seam → a JSON-RPC INVALID_PARAMS error,
|
|
// NOT a tool result (which would mean dispatch happened).
|
|
let error = response.error.expect("validation error expected");
|
|
assert_eq!(error.code, error_codes::INVALID_PARAMS, "got {error:?}");
|
|
assert!(response.result.is_none(), "no dispatch ⇒ no tool result");
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Bonus: drive a whole scripted session over MemoryTransport (zero I/O), proving
|
|
// the serve loop handles a sequence (notification + calls) and closes cleanly.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn scripted_session_over_memory_transport_round_trips() {
|
|
let contexts = FakeContexts::new();
|
|
contexts.seed_agent("architect");
|
|
let (service, _s) = build_service(contexts);
|
|
let server = server(service);
|
|
|
|
let init = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 1, "method": "initialize"
|
|
}))
|
|
.unwrap();
|
|
// A notification (no id) must produce NO reply.
|
|
let note = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "method": "notifications/initialized"
|
|
}))
|
|
.unwrap();
|
|
let list = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 2, "method": "tools/list"
|
|
}))
|
|
.unwrap();
|
|
|
|
let (mut transport, mut rx) = MemoryTransport::scripted(vec![init, note, list]);
|
|
server.serve(&mut transport).await;
|
|
// The serve loop has returned (clean EOF), but `transport` still owns the
|
|
// outbound sender; drop it so the channel closes and `rx` can reach EOF.
|
|
// Without this, `rx.recv()` below would block forever on the single-thread
|
|
// test runtime (the sender would still be alive).
|
|
drop(transport);
|
|
|
|
// Exactly two responses (init + list); the notification produced none.
|
|
let first = rx.recv().await.expect("init response");
|
|
let second = rx.recv().await.expect("list response");
|
|
assert!(
|
|
rx.recv().await.is_none(),
|
|
"notification must not be answered"
|
|
);
|
|
|
|
let first: Value = serde_json::from_slice(&first).unwrap();
|
|
let second: Value = serde_json::from_slice(&second).unwrap();
|
|
assert_eq!(first["id"], json!(1));
|
|
assert!(first["result"]["protocolVersion"].is_string());
|
|
assert_eq!(second["id"], json!(2));
|
|
assert!(second["result"]["tools"].is_array());
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// M4 — observability: a processed `tools/call` publishes
|
|
// OrchestratorRequestProcessed tagged source = Mcp when an event sink is wired.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[tokio::test]
|
|
async fn processed_tools_call_publishes_event_tagged_mcp_source() {
|
|
let contexts = FakeContexts::new();
|
|
let (service, _s) = build_service(contexts.clone());
|
|
let (publish, captured) = capturing_events();
|
|
let server = McpServer::new(service, project()).with_events(publish);
|
|
|
|
// A successful launch (creates + launches an agent from scratch).
|
|
let raw = tools_call(
|
|
1,
|
|
"idea_launch_agent",
|
|
json!({ "target": "dev-backend", "profile": "claude-code" }),
|
|
);
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert!(
|
|
response.error.is_none(),
|
|
"transport error: {:?}",
|
|
response.error
|
|
);
|
|
assert_eq!(response.result.expect("result")["isError"], json!(false));
|
|
|
|
// Exactly one OrchestratorRequestProcessed event, tagged as the MCP door.
|
|
let events = captured.lock().unwrap();
|
|
let processed: Vec<&DomainEvent> = events
|
|
.iter()
|
|
.filter(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. }))
|
|
.collect();
|
|
assert_eq!(
|
|
processed.len(),
|
|
1,
|
|
"expected exactly one processed event; got {events:?}"
|
|
);
|
|
match processed[0] {
|
|
DomainEvent::OrchestratorRequestProcessed {
|
|
requester_id,
|
|
action,
|
|
ok,
|
|
source,
|
|
} => {
|
|
assert_eq!(*source, OrchestrationSource::Mcp, "MCP door must tag Mcp");
|
|
assert_eq!(action, "idea_launch_agent", "action mirrors the tool name");
|
|
assert!(*ok, "the launch succeeded");
|
|
assert_eq!(
|
|
requester_id, "mcp",
|
|
"stable mcp label until per-session bind"
|
|
);
|
|
}
|
|
other => panic!("expected OrchestratorRequestProcessed, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn failed_tools_call_still_publishes_event_with_ok_false_and_mcp_source() {
|
|
// stop_agent on a seeded-but-not-running agent → command fails (isError), yet
|
|
// the observability beacon is still emitted, tagged Mcp with ok = false.
|
|
let contexts = FakeContexts::new();
|
|
contexts.seed_agent("idle");
|
|
let (service, _s) = build_service(contexts);
|
|
let (publish, captured) = capturing_events();
|
|
let server = McpServer::new(service, project()).with_events(publish);
|
|
|
|
let raw = tools_call(1, "idea_stop_agent", json!({ "target": "idle" }));
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert_eq!(response.result.expect("result")["isError"], json!(true));
|
|
|
|
let events = captured.lock().unwrap();
|
|
let processed: Vec<&DomainEvent> = events
|
|
.iter()
|
|
.filter(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. }))
|
|
.collect();
|
|
assert_eq!(
|
|
processed.len(),
|
|
1,
|
|
"a failed call still beacons; got {events:?}"
|
|
);
|
|
match processed[0] {
|
|
DomainEvent::OrchestratorRequestProcessed {
|
|
ok, source, action, ..
|
|
} => {
|
|
assert!(!*ok, "the dispatch failed → ok = false");
|
|
assert_eq!(*source, OrchestrationSource::Mcp);
|
|
assert_eq!(action, "idea_stop_agent");
|
|
}
|
|
other => panic!("expected OrchestratorRequestProcessed, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn server_without_event_sink_emits_nothing_and_does_not_panic() {
|
|
// Non-regression: McpServer::new (no with_events) keeps working and publishes
|
|
// no event — existing call sites stay valid.
|
|
let contexts = FakeContexts::new();
|
|
let (service, _s) = build_service(contexts.clone());
|
|
let server = McpServer::new(service, project()); // no event sink
|
|
|
|
let raw = tools_call(
|
|
1,
|
|
"idea_launch_agent",
|
|
json!({ "target": "dev-backend", "profile": "claude-code" }),
|
|
);
|
|
let response = server.handle_raw(&raw).await.expect("reply owed");
|
|
assert!(
|
|
response.error.is_none(),
|
|
"transport error: {:?}",
|
|
response.error
|
|
);
|
|
assert_eq!(response.result.expect("result")["isError"], json!(false));
|
|
// The command really ran (agent created) — proof the no-op sink path is live.
|
|
assert_eq!(contexts.entries().len(), 1);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 10. Anti-wedge (régression du bug serveur lockstep) + timeout du rendezvous.
|
|
//
|
|
// Cœur de la régression : un `idea_ask_agent` en attente de son `idea_reply`
|
|
// ne doit PLUS parquer toute la connexion. La preuve : pendant qu'un ask est
|
|
// bloqué (rendezvous jamais résolu), un second appel (`tools/list`) sur la
|
|
// MÊME connexion reçoit sa réponse. L'ancien `serve` lockstep ne lisait plus
|
|
// rien tant que `handle_raw` de l'ask n'avait pas rendu → ce test bouclait.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Transport scriptable qui **reste ouvert** : il livre `inbound` dans l'ordre,
|
|
/// puis, une fois la file vide, `recv` reste en attente (au lieu de fermer) tant
|
|
/// que le test n'a pas appelé `close()`. Cela laisse la boucle `serve` vivante —
|
|
/// donc capable de drainer les réponses des tâches encore en vol — pendant qu'un
|
|
/// `idea_ask_agent` est parqué. Les `send` sont capturés sur un canal `mpsc`.
|
|
struct GatedTransport {
|
|
inbound: std::collections::VecDeque<Vec<u8>>,
|
|
outbound: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
|
|
close: Arc<tokio::sync::Notify>,
|
|
}
|
|
|
|
impl GatedTransport {
|
|
fn new(
|
|
messages: Vec<Vec<u8>>,
|
|
) -> (
|
|
Self,
|
|
tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
|
|
Arc<tokio::sync::Notify>,
|
|
) {
|
|
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
|
|
let close = Arc::new(tokio::sync::Notify::new());
|
|
(
|
|
Self {
|
|
inbound: messages.into(),
|
|
outbound: tx,
|
|
close: Arc::clone(&close),
|
|
},
|
|
rx,
|
|
close,
|
|
)
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Transport for GatedTransport {
|
|
async fn recv(&mut self) -> Result<Vec<u8>, TransportError> {
|
|
if let Some(next) = self.inbound.pop_front() {
|
|
return Ok(next);
|
|
}
|
|
// File vide : on n'émet PAS Closed tout de suite — on attend le signal de
|
|
// fermeture pour ne pas tuer la boucle pendant qu'un ask est encore parqué.
|
|
self.close.notified().await;
|
|
Err(TransportError::Closed)
|
|
}
|
|
|
|
async fn send(&mut self, message: &[u8]) -> Result<(), TransportError> {
|
|
self.outbound
|
|
.send(message.to_vec())
|
|
.map_err(|_| TransportError::Closed)
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pending_ask_does_not_wedge_the_connection_concurrent_call_still_answered() {
|
|
// Un `idea_ask_agent` vers une cible vivante bloque sur la mailbox (aucun
|
|
// `idea_reply` ne viendra). Sur la MÊME connexion, un `tools/list` qui suit
|
|
// doit recevoir sa réponse SANS attendre la résolution de l'ask.
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, mailbox, sessions) = build_service_with_mailbox(contexts);
|
|
seed_live_pty(
|
|
&sessions,
|
|
agent_id,
|
|
SessionId::from_uuid(Uuid::from_u128(909)),
|
|
);
|
|
let server = server(service);
|
|
|
|
let ask = tools_call(
|
|
1,
|
|
"idea_ask_agent",
|
|
json!({ "target": "architect", "task": "blocking..." }),
|
|
);
|
|
let list = serde_json::to_vec(&json!({
|
|
"jsonrpc": "2.0", "id": 2, "method": "tools/list"
|
|
}))
|
|
.unwrap();
|
|
|
|
let (transport, mut rx, close) = GatedTransport::new(vec![ask, list]);
|
|
let serve = tokio::spawn(async move {
|
|
let mut transport = transport;
|
|
server.serve(&mut transport).await;
|
|
});
|
|
|
|
// L'ask a bien été enqueué (donc il est parqué en attente de 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");
|
|
|
|
// La réponse au `tools/list` arrive AVANT que l'ask soit résolu : c'est la
|
|
// preuve anti-wedge. (Sur l'ancien code lockstep, ce recv timeout-ait.)
|
|
let response = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv())
|
|
.await
|
|
.expect("tools/list must be answered while the ask is still pending")
|
|
.expect("a response payload");
|
|
let response: Value = serde_json::from_slice(&response).unwrap();
|
|
assert_eq!(response["id"], json!(2), "the answered call is tools/list");
|
|
assert!(
|
|
response["result"]["tools"].is_array(),
|
|
"tools/list result, got {response}"
|
|
);
|
|
|
|
// L'ask est toujours en vol (non résolu, aucun `idea_reply` ne viendra) : la
|
|
// boucle restera en attente de sa réponse, c'est attendu. On la ferme et on
|
|
// abandonne la tâche serve (le rendezvous parqué ne se résoudra jamais ici).
|
|
assert_eq!(mailbox.pending(&agent_id), 1, "ask still pending");
|
|
close.notify_one();
|
|
serve.abort();
|
|
let _ = serve.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn ask_agent_rendezvous_times_out_with_a_jsonrpc_error() {
|
|
// La cible ne répondra jamais. Avec une borne courte injectée, l'ask doit
|
|
// finir par renvoyer une ERREUR JSON-RPC propre (filet de sécurité serveur),
|
|
// au lieu de pendre indéfiniment.
|
|
let contexts = FakeContexts::new();
|
|
let agent_id = contexts.seed_agent("architect");
|
|
let (service, _mailbox, sessions) = build_service_with_mailbox(contexts);
|
|
seed_live_pty(
|
|
&sessions,
|
|
agent_id,
|
|
SessionId::from_uuid(Uuid::from_u128(910)),
|
|
);
|
|
let server = server(service).with_ask_rendezvous_timeout(std::time::Duration::from_millis(50));
|
|
|
|
let raw = tools_call(
|
|
1,
|
|
"idea_ask_agent",
|
|
json!({ "target": "architect", "task": "never answered" }),
|
|
);
|
|
let response =
|
|
tokio::time::timeout(std::time::Duration::from_secs(10), server.handle_raw(&raw))
|
|
.await
|
|
.expect("must not hang past the injected timeout")
|
|
.expect("reply owed");
|
|
|
|
let error = response.error.expect("a JSON-RPC error on timeout");
|
|
assert_eq!(error.code, error_codes::INTERNAL_ERROR, "got {error:?}");
|
|
assert!(
|
|
error.message.contains("timeout"),
|
|
"explicit timeout message, got {error:?}"
|
|
);
|
|
assert!(response.result.is_none(), "error responses carry no result");
|
|
}
|