//! Managed application state — the product of the **composition root**. //! //! The composition root ([`build_app_state`]) is the *single* place that //! constructs concrete adapters (`new ConcreteAdapter`) and injects them as //! `Arc` into the use cases (ARCHITECTURE §1.1, §10). The use cases //! are then exposed through `tauri::State` to the command handlers. use std::collections::{HashMap, HashSet}; use std::ffi::OsString; use std::path::Path; use std::path::PathBuf; use std::sync::{Arc, Mutex}; use application::{ AgentResumer, AppError, AssignSkillToAgent, ChangeAgentProfile, CheckEmbedderSuggestion, CloseProject, CloseTab, CloseTerminal, ConfigureProfiles, CreateAgentFromScratch, CreateAgentFromTemplate, CreateLayout, CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent, LaunchAgentInput, SessionLimitService, DeleteEmbedderProfile, DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate, DescribeEmbedderEngines, DetectAgentDrift, DetectProfiles, DismissEmbedderSuggestion, FirstRunState, GetMemory, GetProjectPermissions, GitBranches, GitCheckout, GitCommit, GitGraph, GitInit, GitLog, GitStage, GitStatus, GitUnstage, HealthUseCase, InspectConversation, LaunchAgent, ListAgents, ListAgentsInput, ListEmbedderProfiles, ListLayouts, ListMemories, ListProfiles, ListProjects, ListResumableAgents, ListSkills, ListTemplates, LiveAgentRegistry, LoadLayout, McpRuntime, MoveTabToNewWindow, MutateLayout, OnnxModelView, OpenProject, PermissionProjectorRegistry, OpenTerminal, OrchestratorService, ReadAgentContext, ReadMemoryIndex, ReadProjectContext, RecallMemory, ReconcileLayouts, RecordTurn, RecordTurnProvider, ReferenceProfiles, RenameLayout, ResizeTerminal, ResolveAgentPermissions, ResolveMemoryLinks, SaveEmbedderProfile, SaveProfile, SetActiveLayout, SnapshotRunningAgents, StructuredSessions, SuggestedThisSession, SyncAgentWithTemplate, TerminalSessions, UnassignSkillFromAgent, UpdateAgentContext, UpdateAgentPermissions, UpdateMemory, UpdateProjectContext, UpdateProjectPermissions, UpdateSkill, UpdateTemplate, WriteToTerminal, AGENT_MEMORY_RECALL_BUDGET, }; use domain::ports::{ AgentContextStore, AgentRuntime, AgentSessionFactory, Clock, Embedder, EmbedderEnvInspector, EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator, MemoryRecall, MemoryStore, PermissionStore, ProcessSpawner, ProfileStore, ProjectStore, PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, }; use domain::profile::{ AgentProfile, ContextInjection, McpConfigStrategy, McpTransport, StructuredAdapter, }; use domain::remote::RemoteKind; use domain::{AgentId, DomainEvent, EmbedderProfile, Project, ProjectId}; use serde_json::{json, Map, Value}; use uuid::Uuid; use infrastructure::{ embedder_from_profile, AdaptiveMemoryRecall, ClaudePermissionProjector, ClaudeTranscriptInspector, CliAgentRuntime, CodexPermissionProjector, EmbedderEnvProbe, FsConversationLog, FsEmbedderProfileStore, FsEmbedderPromptStore, FsHandoffStore, FsMemoryStore, FsOrchestratorWatcher, FsPermissionStore, FsProfileStore, FsProjectStore, FsProviderSessionStore, FsSkillStore, FsTemplateStore, Git2Repository, HeuristicHandoffSummarizer, IdeaiContextStore, InMemoryConversationRegistry, InMemoryMailbox, LocalFileSystem, LocalProcessSpawner, McpServer, MediatedInbox, NaiveMemoryRecall, OrchestratorWatchHandle, PortablePtyAdapter, StructuredSessionFactory, SystemClock, SystemMillisClock, TokioBroadcastEventBus, TokioScheduler, UuidGenerator, VectorMemoryRecall, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR, RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED, }; use crate::chat::ChatBridge; use crate::mcp_endpoint::{mcp_endpoint, AppMcpRuntimeProvider, McpEndpoint}; use crate::pty::PtyBridge; use infrastructure::StdioTransport; use interprocess::local_socket::tokio::Listener as LocalSocketListener; use interprocess::local_socket::traits::tokio::Listener as _; use interprocess::local_socket::{GenericFilePath, ListenerOptions, ToFsName}; use tokio::io::{AsyncBufReadExt, BufReader}; /// Implémente [`RecordTurnProvider`] (lot P6b) en matérialisant un [`RecordTurn`] /// ciblant le **project root** du tour en cours. /// /// L'[`OrchestratorService`] est unique pour tous les projets, alors que le log/handoff /// conversationnel est **par project root** (`/.ideai/conversations/`). Les /// adapters `Fs*` fixent leur racine à la construction et ne sont que des jointures de /// chemin : on en construit donc un jeu frais par tour, ciblant le bon dossier. Sans /// état (zéro champ), partagé via un simple `Arc`. struct AppRecordTurnProvider; impl RecordTurnProvider for AppRecordTurnProvider { fn record_turn_for(&self, root: &domain::project::ProjectPath) -> Option> { let log = Arc::new(FsConversationLog::new(root)); let handoffs = Arc::new(FsHandoffStore::new(root)); let summarizer = Arc::new(HeuristicHandoffSummarizer::new()); Some(Arc::new(RecordTurn::new(log, handoffs, summarizer))) } } /// Implémente [`HandoffProvider`](application::HandoffProvider) (lot P7) en /// matérialisant un [`FsHandoffStore`] ciblant le **project root** du lancement en /// cours. /// /// Même raison d'être que [`AppRecordTurnProvider`] : [`LaunchAgent`] est unique pour /// tous les projets, alors que le handoff est **par project root** /// (`/.ideai/conversations/`). On construit donc un store frais par lancement, /// ciblant le bon dossier. Sans état (zéro champ), partagé via un simple `Arc`. struct AppHandoffProvider; impl application::HandoffProvider for AppHandoffProvider { fn handoff_store_for( &self, root: &domain::project::ProjectPath, ) -> Option> { Some(Arc::new(FsHandoffStore::new(root)) as Arc) } } /// Implémente [`ProviderSessionProvider`](application::ProviderSessionProvider) (lot /// P8b) en matérialisant un [`FsProviderSessionStore`] ciblant le **project root** du /// lancement en cours. /// /// Jumeau stateless de [`AppHandoffProvider`] : [`LaunchAgent`] est unique pour tous /// les projets, alors que `providers.json` est **par project root** /// (`/.ideai/conversations/providers.json`). On construit donc un store frais /// par lancement, ciblant le bon dossier. Sans état (zéro champ), partagé via `Arc`. struct AppProviderSessionProvider; impl application::ProviderSessionProvider for AppProviderSessionProvider { fn provider_session_store_for( &self, root: &domain::project::ProjectPath, ) -> Option> { Some(Arc::new(FsProviderSessionStore::new(root)) as Arc) } } /// Contexte minimal de **relance** d'un agent (LS7, ARCHITECTURE §21.5). /// /// [`AgentResumer::resume`] et [`ScheduledTask::ResumeAgent`] ne portent **pas** le /// `Project` ni la taille de la cellule, alors que [`LaunchAgentInput`] les exige. /// La commande `launch_agent` (seul endroit où ces faits sont en main) alimente ce /// contexte par `agent_id` ; [`AppAgentResumer`] le relit à l'échéance pour /// recomposer un lancement complet. #[derive(Clone)] pub struct ResumeContext { /// Le projet hôte de l'agent (pour recomposer `LaunchAgentInput`). pub project: Project, /// Hauteur de la cellule au dernier lancement (lignes PTY). pub rows: u16, /// Largeur de la cellule au dernier lancement (colonnes PTY). pub cols: u16, } /// Registre partagé `agent_id → ResumeContext` (composition root ↔ commande /// `launch_agent`). Le **même** `Arc` est injecté dans [`AppAgentResumer`] et conservé /// sur [`AppState`] pour que la commande l'alimente à chaque lancement. pub type ResumeContexts = Arc>>; /// Implémente le port applicatif [`AgentResumer`] (LS7) **par-dessus** le mécanisme de /// lancement existant ([`LaunchAgent`]). /// /// À l'échéance d'une limite de session, [`SessionLimitService::execute_resume`] /// délègue ici : on relit le [`ResumeContext`] alimenté par `launch_agent`, on /// recompose un [`LaunchAgentInput`] (avec le `conversation_id` de la cellule ⇒ /// `LaunchAgent` applique [`domain::ports::SessionPlan::Resume`]), puis on transmet le /// `resume_prompt` comme **premier tour** via le portail d'entrée unique /// ([`InputMediator`](domain::input::InputMediator)) — jamais un write PTY brut /// (ARCHITECTURE §20). Sans contexte connu (resume « à l'aveugle »), on échoue /// proprement (l'erreur empêche `AgentResumed` d'être publié), jamais de panique. struct AppAgentResumer { /// Le **même** `Arc` que la commande `launch_agent` (relance/réattache). launch_agent: Arc, /// Portail d'entrée unique : injecte le prompt de reprise comme premier tour. input_mediator: Arc, /// Contexte de relance alimenté par la commande `launch_agent`. contexts: ResumeContexts, } #[async_trait::async_trait] impl AgentResumer for AppAgentResumer { async fn resume( &self, agent_id: AgentId, node_id: domain::NodeId, conversation_id: Option, resume_prompt: &str, ) -> Result<(), AppError> { // Repli propre (jamais de panique) : sans contexte de relance connu, on ne // reprend pas à l'aveugle. L'erreur remonte ⇒ `AgentResumed` n'est pas publié. let ctx = self .contexts .lock() .ok() .and_then(|m| m.get(&agent_id).cloned()); let Some(ctx) = ctx else { return Err(AppError::NotFound(format!( "resume context for agent {agent_id}" ))); }; // Recompose la déclaration MCP réelle (même recette que la commande // `launch_agent`) pour que l'agent repris retrouve ses outils `idea_*`. let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { exe, endpoint: crate::mcp_endpoint::mcp_endpoint(&ctx.project.id) .as_cli_arg() .to_owned(), project_id: ctx.project.id.as_uuid().simple().to_string(), requester: agent_id.to_string(), }); self.launch_agent .execute(LaunchAgentInput { project: ctx.project, agent_id, rows: ctx.rows, cols: ctx.cols, node_id: Some(node_id), conversation_id, mcp_runtime, }) .await?; // Premier tour de reprise par le **portail d'entrée** (pas de write brut, §20) : // l'enqueue publie `DelegationReady`, la cellule l'écrit quand le prompt est prêt. // On ne corrèle aucune réponse (reprise, pas une délégation) ⇒ on lâche le // `PendingReply`. let ticket = domain::mailbox::Ticket::new( domain::mailbox::TicketId::new_random(), "IdeA", resume_prompt, ); let _ = self.input_mediator.enqueue(agent_id, ticket); Ok(()) } } /// Everything the IPC layer needs at runtime, managed by Tauri. /// /// Use cases are stored behind `Arc` so handlers clone cheaply. The concrete /// adapters are owned here and never leak past the composition root as concrete /// types — downstream code only sees the `Arc` held inside the use /// cases. pub struct AppState { /// Trivial health use case validating the end-to-end wiring. pub health: Arc, /// Create a project (init `.ideai/`, register it). pub create_project: Arc, /// Open a project (load meta + manifest). pub open_project: Arc, /// Close a project (persist state). pub close_project: Arc, /// Close a tab. pub close_tab: Arc, /// List known projects. pub list_projects: Arc, /// Read `.ideai/CONTEXT.md`. pub read_project_context: Arc, /// Overwrite `.ideai/CONTEXT.md`. pub update_project_context: Arc, /// Open a terminal (spawn PTY, register session). pub open_terminal: Arc, /// Write keystrokes to a terminal. pub write_terminal: Arc, /// Resize a terminal. pub resize_terminal: Arc, /// Close a terminal (kill PTY). pub close_terminal: Arc, /// Load a project's persisted layout tree. pub load_layout: Arc, /// Mutate + persist a project's layout tree. pub mutate_layout: Arc, /// List all named layouts for a project (#4). pub list_layouts: Arc, /// Create a new named layout (#4). pub create_layout: Arc, /// Rename a named layout (#4). pub rename_layout: Arc, /// Delete a named layout (#4). pub delete_layout: Arc, /// Set the active named layout (#4). pub set_active_layout: Arc, /// Freeze `agent_was_running` on every agent leaf before a PTY kill (T5). pub snapshot_running_agents: Arc, /// Dé-doublonne, à l'ouverture, les feuilles d'agent en double d'un même /// agent dans `layouts.json` (R0c). Idempotent : no-op sans doublon. pub reconcile_layouts: Arc, /// Detect which candidate profiles' CLIs are installed (first-run). pub detect_profiles: Arc, /// List configured profiles. pub list_profiles: Arc, /// Save (upsert) a profile. pub save_profile: Arc, /// Delete a profile. pub delete_profile: Arc, /// Persist the batch of chosen profiles (closes the first run). pub configure_profiles: Arc, /// Expose the pre-filled reference catalogue. pub reference_profiles: Arc, /// Whether the first-run wizard should show + the reference catalogue. pub first_run_state: Arc, /// The local PTY adapter, kept port-typed so the presentation layer can /// `subscribe_output` to pump bytes into the [`PtyBridge`] (it owns transport). pub pty_port: Arc, /// Active-terminal registry shared by the terminal use cases. pub terminal_sessions: Arc, /// The domain event bus (also handed to the event relay). pub event_bus: Arc, /// Generic PTY↔Channel bridge registry (consumed by L3). pub pty_bridge: Arc, /// Registre des sessions structurées (IA / cellules chat, §17.5). Partagé avec /// `LaunchAgent`/`ChangeAgentProfile` ; consommé par les commandes de chat (D4) /// pour résoudre la session vivante d'un `sessionId` et l'arrêter à la fermeture. pub structured_sessions: Arc, /// Pont réponses structurées ↔ Channel (jumeau de [`PtyBridge`], §17.7). Route /// les [`ReplyChunk`](crate::dto::ReplyChunk) d'un tour vers la bonne cellule /// chat et retient le scrollback de conversation pour la ré-attache. pub chat_bridge: Arc, // --- Agents (L6) --- /// Create a project agent from scratch. pub create_agent: Arc, /// List a project's agents. pub list_agents: Arc, /// Read an agent's Markdown context. pub read_agent_context: Arc, /// Overwrite an agent's Markdown context. pub update_agent_context: Arc, /// Delete an agent from the manifest. pub delete_agent: Arc, /// Launch an agent (spawn PTY, apply injection strategy). pub launch_agent: Arc, /// Hot-swap an agent's runtime profile, relaunching its live session in place (§15.1). pub change_agent_profile: Arc, /// Read-only inventory of a project's resumable agent cells, for the reopen /// panel (§15.2). pub list_resumable_agents: Arc, /// Best-effort inspection of a conversation (last topic + token indicator) /// for the resume popup (T7). Optional/extensible: backed by a `Vec` of /// [`domain::ports::SessionInspector`]s; an empty/missing match yields empty /// details, never an error. pub inspect_conversation: Arc, /// Project registry — used by agent commands to resolve a `Project` from an id. pub project_store: Arc, /// Read the project permission document. pub get_project_permissions: Arc, /// Update project-level default permissions. pub update_project_permissions: Arc, /// Update one agent permission override. pub update_agent_permissions: Arc, /// Resolve effective permissions for one agent. pub resolve_agent_permissions: Arc, // --- Windows (L10) --- /// Detach a tab into a new OS window (persists the workspace topology). pub move_tab: Arc, // --- Templates & sync (L7) --- /// Create a template in the global store. pub create_template: Arc, /// Update a template's content (bumps version). pub update_template: Arc, /// List all templates in the global store. pub list_templates: Arc, /// Delete a template from the global store. pub delete_template: Arc, /// Create an agent from a template. pub create_agent_from_template: Arc, /// Detect which synchronized agents are behind their template. pub detect_agent_drift: Arc, /// Apply a template update to a synchronized agent. pub sync_agent_with_template: Arc, // --- Git (L8) --- /// Report the working-tree status of a repository. pub git_status: Arc, /// Stage a path. pub git_stage: Arc, /// Unstage a path. pub git_unstage: Arc, /// Create a commit. pub git_commit: Arc, /// List branches. pub git_branches: Arc, /// Check out a branch. pub git_checkout: Arc, /// Return the recent commit log. pub git_log: Arc, /// Initialise a repository. pub git_init: Arc, /// Return the commit graph for all local branches. pub git_graph: Arc, // --- Skills (L12) --- /// Create a skill in a scope's store. pub create_skill: Arc, /// Update a skill's content. pub update_skill: Arc, /// List skills in a scope. pub list_skills: Arc, /// Delete a skill from its scope's store. pub delete_skill: Arc, /// Assign a skill to an agent (records a `SkillRef`). pub assign_skill: Arc, /// Unassign a skill from an agent. pub unassign_skill: Arc, // --- Memory (LOT A — §14.5.1) --- /// Create a memory note in the project's store. pub create_memory: Arc, /// Replace an existing memory note. pub update_memory: Arc, /// List the project's memory notes. pub list_memories: Arc, /// Read one memory note by slug. pub get_memory: Arc, /// Delete a memory note. pub delete_memory: Arc, /// Read the structured `MEMORY.md` index. pub read_memory_index: Arc, /// Resolve a note's outgoing `[[slug]]` links. pub resolve_memory_links: Arc, /// Recall the most relevant memory entries for a query within a budget /// (LOT B — §14.5.2). pub recall_memory: Arc, // --- Embedder config (LOT C2 — §14.5.3) --- /// List the configured embedder profiles (`embedder.json`). pub list_embedder_profiles: Arc, /// Save (upsert, validating) an embedder profile. pub save_embedder_profile: Arc, /// Delete an embedder profile by id. pub delete_embedder_profile: Arc, /// Describe the embedding engines available to the configuration UI (catalogue /// + detected local environment + compiled-in capabilities). pub describe_embedder_engines: Arc, // --- Embedder suggestion (LOT C3 — §14.5.5) --- /// Persist the user's response to the embedder suggestion (`later`/`never`). pub dismiss_embedder_suggestion: Arc, // --- Orchestrator (§14.3) --- /// Dispatches validated orchestrator requests to the agent/skill use cases. /// Shared by every per-project filesystem watcher. pub orchestrator_service: Arc, /// Live orchestrator request watchers, keyed by project. One watcher per open /// project tails its `.ideai/requests/` tree; dropping the handle stops it. /// Guarded by a `Mutex` so the open/close commands can register/unregister /// watchers concurrently. pub orchestrator_watchers: Mutex>, /// Live IdeA MCP servers, keyed by project — the **twin** of /// [`orchestrator_watchers`](Self::orchestrator_watchers). One server per open /// project is the MCP entry door onto the *same* [`OrchestratorService`] the /// file watcher feeds; both coexist (Décision 4). Started on open/create and /// stopped on close, exactly like the watcher, via the same `Mutex`-guarded /// per-project registry. pub mcp_servers: Mutex>, /// Service de gestion des **limites de session** des agents (ARCHITECTURE §21) : /// détecte → planifie → reprend, et annule. Alimenté par les taps niveau 1 /// (structuré, `agent_send`) et niveau 2 (PTY, `launch_agent`) ; sa reprise auto est /// annulable via la commande `cancel_resume`. pub session_limit_service: Arc, /// Registre `agent_id → ResumeContext` (LS7) partagé avec [`AppAgentResumer`] : /// la commande `launch_agent` y dépose le `Project`/taille du dernier lancement pour /// que la reprise auto puisse recomposer un `LaunchAgentInput` complet. pub resume_contexts: ResumeContexts, } impl AppState { /// **Composition root.** Builds all adapters and use cases. /// /// `app_data_dir` is the machine-local IDE data directory (ARCHITECTURE /// §9.2), resolved by the caller via the Tauri path API and injected here so /// the stores never touch Tauri themselves (Dependency Inversion). /// /// This is the only function that constructs concrete adapters; every other /// layer depends on ports. Adapters added in later lots (PTY, git, remote) /// are wired in here. #[must_use] pub fn build(app_data_dir: PathBuf) -> Self { // --- Concrete adapters (driven adapters) --- let event_bus = Arc::new(TokioBroadcastEventBus::new()); let clock = Arc::new(SystemClock::new()); let ids = Arc::new(UuidGenerator::new()); let fs = Arc::new(LocalFileSystem::new()); let store = Arc::new(FsProjectStore::new( Arc::clone(&fs) as Arc, app_data_dir.to_string_lossy().into_owned(), )); // Port-typed handles for injection. let fs_port = Arc::clone(&fs) as Arc; let store_port = Arc::clone(&store) as Arc; let events_port = Arc::clone(&event_bus) as Arc; // --- Use cases (ports injected as Arc) --- let health = Arc::new(HealthUseCase::new( Arc::clone(&clock) as Arc, Arc::clone(&ids) as Arc, Arc::clone(&events_port), )); let create_project = Arc::new(CreateProject::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&ids) as Arc, Arc::clone(&clock) as Arc, Arc::clone(&events_port), )); let open_project = Arc::new(OpenProject::new( Arc::clone(&store_port), Arc::clone(&fs_port), )); let close_project = Arc::new(CloseProject::new(Arc::clone(&store_port))); let close_tab = Arc::new(CloseTab::new(Arc::clone(&store_port))); let list_projects = Arc::new(ListProjects::new(Arc::clone(&store_port))); let read_project_context = Arc::new(ReadProjectContext::new(Arc::clone(&fs_port))); let update_project_context = Arc::new(UpdateProjectContext::new(Arc::clone(&fs_port))); // --- PTY adapter + terminal use cases (L3) --- let pty = Arc::new( PortablePtyAdapter::new().with_sandbox_enforcer(infrastructure::default_enforcer()), ); let pty_port = Arc::clone(&pty) as Arc; let terminal_sessions = Arc::new(TerminalSessions::new()); // --- Sessions structurées (IA / cellules chat, §17) --- // Registre jumeau de TerminalSessions + fabrique infra routée par // `profile.structured_adapter`. Injectés dans LaunchAgent (routage §17.4) et // ChangeAgentProfile (shutdown polymorphe au hot-swap). let structured_sessions = Arc::new(StructuredSessions::new()); let session_factory = Arc::new( StructuredSessionFactory::new() .with_sandbox_enforcer(infrastructure::default_enforcer()), ) as Arc; let open_terminal = Arc::new(OpenTerminal::new( Arc::clone(&pty_port), Arc::clone(&terminal_sessions), Arc::clone(&events_port), )); let write_terminal = Arc::new(WriteToTerminal::new( Arc::clone(&pty_port), Arc::clone(&terminal_sessions), )); let resize_terminal = Arc::new(ResizeTerminal::new( Arc::clone(&pty_port), Arc::clone(&terminal_sessions), )); let close_terminal = Arc::new(CloseTerminal::new( Arc::clone(&pty_port), Arc::clone(&terminal_sessions), )); // --- Layout use cases (L4 + #4) --- let load_layout = Arc::new(LoadLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), )); let mutate_layout = Arc::new(MutateLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&events_port), )); let list_layouts = Arc::new(ListLayouts::new( Arc::clone(&store_port), Arc::clone(&fs_port), )); let create_layout = Arc::new(CreateLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&ids) as Arc, Arc::clone(&events_port), )); let rename_layout = Arc::new(RenameLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&events_port), )); let delete_layout = Arc::new(DeleteLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&events_port), )); let set_active_layout = Arc::new(SetActiveLayout::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&events_port), )); // Close-time snapshot of running agents (T5). Shares the SAME live-session // registry as the terminal/agent use cases, so its liveness check reflects // the very PTYs the shutdown hook is about to kill — it must run *before*. let snapshot_running_agents = Arc::new(SnapshotRunningAgents::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&terminal_sessions) as Arc, )); // Twin of the snapshot above, but at *open* time: dé-doublonne les // `layouts.json` portant plusieurs feuilles sur le même agent (R0c, §3.4 // « Trou C »). Idempotent : aucun doublon ⇒ aucune écriture. let reconcile_layouts = Arc::new(ReconcileLayouts::new( Arc::clone(&store_port), Arc::clone(&fs_port), )); // --- Profiles & AI runtime (L5) --- // One generic, profile-driven runtime adapter (Open/Closed): it holds the // process spawner used for detection. The profile store persists // `profiles.json` in the same machine-local app-data dir as the project // registry. let spawner = Arc::new(LocalProcessSpawner::new()); let spawner_port = Arc::clone(&spawner) as Arc; let runtime = Arc::new(CliAgentRuntime::new(Arc::clone(&spawner_port))); let runtime_port = Arc::clone(&runtime) as Arc; let profile_store = Arc::new(FsProfileStore::new( Arc::clone(&fs_port), app_data_dir.to_string_lossy().into_owned(), )); let profile_store_port = Arc::clone(&profile_store) as Arc; let detect_profiles = Arc::new(DetectProfiles::new(Arc::clone(&runtime_port))); let list_profiles = Arc::new(ListProfiles::new(Arc::clone(&profile_store_port))); let save_profile = Arc::new(SaveProfile::new(Arc::clone(&profile_store_port))); let delete_profile = Arc::new(DeleteProfile::new(Arc::clone(&profile_store_port))); let configure_profiles = Arc::new(ConfigureProfiles::new(Arc::clone(&profile_store_port))); let reference_profiles = Arc::new(ReferenceProfiles::new()); let first_run_state = Arc::new(FirstRunState::new(Arc::clone(&profile_store_port))); let pty_bridge = Arc::new(PtyBridge::new()); // Twin of the PTY bridge for structured chat sessions (§17.7): routes a // turn's ReplyChunks to the owning chat cell and retains the conversation // scrollback for re-attach. let chat_bridge = Arc::new(ChatBridge::new()); // --- Agent context store + use cases (L6) --- let contexts = Arc::new(IdeaiContextStore::new(Arc::clone(&fs_port))); let contexts_port = Arc::clone(&contexts) as Arc; // --- Project permissions (LP1) --- let permission_store = Arc::new(FsPermissionStore::new(Arc::clone(&fs_port))); let permission_store_port = Arc::clone(&permission_store) as Arc; // --- Skill store (L12) --- // Global skills live in the machine-local app-data dir; project skills are // resolved per call from each project's `.ideai/` (so one store serves all // open projects). Shared by the skill use cases and the agent launcher // (assigned-skill injection into the convention file, §14.2). let skill_store = Arc::new(FsSkillStore::new( Arc::clone(&fs_port), app_data_dir.to_string_lossy().into_owned(), )); let skill_store_port = Arc::clone(&skill_store) as Arc; // Memory store + naïve recall (LOT A/B — §14.5.1/§14.5.4). Built here so the // recall port can be injected into LaunchAgent below (it composes the project // memory recall into the convention file at activation); the memory use cases // are wired further down from these same instances. let memory_store = Arc::new(FsMemoryStore::new(Arc::clone(&fs_port))); let memory_store_port = Arc::clone(&memory_store) as Arc; // Load the configured embedder profile (mono-profile for now: take the last // listed, fall back to `none`). A multi-profile selector is a follow-up; the // `none` default keeps recall strictly naïve and dependency-free. let embedder_store = Arc::new(FsEmbedderProfileStore::new( Arc::clone(&fs_port), app_data_dir.to_string_lossy().into_owned(), )); let embedder_store_port = Arc::clone(&embedder_store) as Arc; // `build` may run inside an ambient async runtime (Tauri's `setup`, or // `#[tokio::test]`), so blocking the current thread on a future panics. // Drive the one-shot load on a dedicated thread with its own runtime. let embedder_profile = std::thread::scope(|s| { s.spawn(|| { tokio::runtime::Builder::new_current_thread() .enable_all() .build() .ok() .and_then(|rt| rt.block_on(embedder_store.list()).ok()) .and_then(|mut v| v.pop()) }) .join() .ok() .flatten() }) .unwrap_or_else(EmbedderProfile::none); let onnx_cache_dir = app_data_dir.join(ONNX_CACHE_SUBDIR); let memory_recall_port = build_memory_recall( Arc::clone(&fs_port), Arc::clone(&memory_store_port), &embedder_profile, &onnx_cache_dir, ); // --- Embedder configuration use cases (LOT C2 — §14.5.3) --- // CRUD over `embedder.json` (port-typed store, built above) + a read-only // description of the available engines. The environment probe shares the SAME // `onnx_cache_dir` the recall uses, so "is this model cached?" stays coherent. // The static ONNX catalogue and the compiled-capability flags are owned by // infrastructure and injected here (the application stays infra-free, DIP). let env_inspector = Arc::new(EmbedderEnvProbe::new( onnx_cache_dir.clone(), DEFAULT_OLLAMA_BASE_URL, )) as Arc; // Cloned for the suggestion check below (the original is moved into // `describe_embedder_engines`). Both share the same probe behaviour. let env_inspector_for_suggestion = Arc::clone(&env_inspector); let recommended_onnx: Vec = RECOMMENDED_ONNX_MODELS .iter() .map(|m| OnnxModelView { id: m.id.to_owned(), display_name: m.display_name.to_owned(), dimension: m.dimension, approx_size_mb: m.approx_size_mb, recommended: m.recommended, }) .collect(); let list_embedder_profiles = Arc::new(ListEmbedderProfiles::new(Arc::clone(&embedder_store_port))); let save_embedder_profile = Arc::new(SaveEmbedderProfile::new(Arc::clone(&embedder_store_port))); let delete_embedder_profile = Arc::new(DeleteEmbedderProfile::new(Arc::clone(&embedder_store_port))); let describe_embedder_engines = Arc::new(DescribeEmbedderEngines::new( env_inspector, recommended_onnx, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED, )); // --- Embedder suggestion (LOT C3 — §14.5.5) --- // Per-project dismissal state (`.ideai/memory/.embedder-prompt.json`) + // the in-memory "already suggested this session" guard, shared with the // launcher's best-effort check. The check publishes EmbedderSuggested at // most once per session per project, only while strategy is `none` and the // memory has outgrown the recall budget. let prompt_store = Arc::new(FsEmbedderPromptStore::new(Arc::clone(&fs_port))); let prompt_store_port = Arc::clone(&prompt_store) as Arc; let suggested_this_session: SuggestedThisSession = SuggestedThisSession::default(); let check_embedder_suggestion = Arc::new(CheckEmbedderSuggestion::new( Arc::clone(&embedder_store_port), Arc::clone(&memory_store_port), Arc::clone(&prompt_store_port), env_inspector_for_suggestion, Arc::clone(&events_port), Arc::clone(&suggested_this_session), AGENT_MEMORY_RECALL_BUDGET, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED, )); let dismiss_embedder_suggestion = Arc::new(DismissEmbedderSuggestion::new(Arc::clone( &prompt_store_port, ))); let create_agent = Arc::new(CreateAgentFromScratch::new( Arc::clone(&contexts_port), Arc::clone(&ids) as Arc, Arc::clone(&events_port), )); let list_agents = Arc::new(ListAgents::new(Arc::clone(&contexts_port))); let read_agent_context = Arc::new(ReadAgentContext::new(Arc::clone(&contexts_port))); let update_agent_context = Arc::new(UpdateAgentContext::new(Arc::clone(&contexts_port))); let delete_agent = Arc::new(DeleteAgent::new( Arc::clone(&contexts_port), Arc::clone(&events_port), )); // LaunchAgent shares the SAME pty_port and terminal_sessions as the terminal // use cases — indispensable for the PtyBridge to work correctly. // // Option 1 « Terminal + MCP » (lot B-2) : on **ne câble plus** la fabrique // structurée. La vue humaine d'un agent est désormais le **terminal brut // natif** (PTY interactif) — réflexion live + Échap natifs CLI, zéro parsing — // et la délégation inter-agents passe par les outils MCP (`idea_ask_agent` / // `idea_reply`), pas par une `AgentChatView`. Sans `with_structured`, le point // de routage §17.4 de `LaunchAgent::execute` retombe **toujours** sur le chemin // PTH : tout profil (Claude/Codex inclus) ouvre une cellule terminal. Le code // `launch_structured` reste en place (mort-code retiré au lot de nettoyage B-6). let _session_factory = session_factory; // décâblé en B-2 (nettoyage B-6) // --- Permission projectors (lot LP3-5) --- // UN seul registre, source unique de vérité, injecté à l'identique dans // `LaunchAgent` (projection au lancement) ET `ChangeAgentProfile` (nettoyage des // fichiers orphelins au swap). Les deux projecteurs concrets vivent dans // l'infrastructure, keyés par leur `ProjectorKey` via `with(...)`. let permission_projectors = Arc::new( PermissionProjectorRegistry::new() .with(Arc::new(ClaudePermissionProjector) as Arc) .with(Arc::new(CodexPermissionProjector) as Arc), ); let launch_agent = Arc::new( LaunchAgent::new( Arc::clone(&contexts_port), Arc::clone(&profile_store_port), Arc::clone(&runtime_port), Arc::clone(&fs_port), Arc::clone(&pty_port), Arc::clone(&skill_store_port), Arc::clone(&terminal_sessions), Arc::clone(&events_port), Arc::clone(&ids) as Arc, Arc::clone(&memory_recall_port), Some(Arc::clone(&check_embedder_suggestion)), ) .with_permission_store(Arc::clone(&permission_store_port)) // Reprise conversationnelle (lot P7) : à chaque (re)lancement, si la cellule // porte une conversation et qu'un handoff existe (`/.ideai/conversations/`), // son résumé est réinjecté dans le convention file. Best-effort, additif : // un handoff absent/illisible ⇒ lancement normal sans section. .with_handoff_provider( Arc::new(AppHandoffProvider) as Arc ) // Resumable moteur par provider (lot P8b) : après un lancement structuré // exposant un id de session moteur, rangé sous la clé de paire dans // `/.ideai/conversations/providers.json`. Best-effort, additif : une // écriture en échec / pas d'id moteur ⇒ lancement normal, aucune écriture. .with_provider_session_provider(Arc::new(AppProviderSessionProvider) as Arc) // Projection des permissions au (re)lancement (lot LP3-5) : avec le // permission store câblé ci-dessus, `resolve` a une source ⇒ le projecteur // du profil matérialise la config de permission de la CLI dans le run dir. .with_permission_projectors(Arc::clone(&permission_projectors)), ); // Hot-swap an agent's runtime profile (§15.1). Reuses the shared context/ // profile/project/fs stores, the live-session registry and PTY port, and // *composes* the launcher above for the in-place relaunch (no duplication). // Voit aussi le registre structuré pour un « kill » polymorphe (§17.4). let change_agent_profile = Arc::new( ChangeAgentProfile::new( Arc::clone(&contexts_port), Arc::clone(&profile_store_port), Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&terminal_sessions), Arc::clone(&pty_port), Arc::clone(&launch_agent), Arc::clone(&events_port), ) .with_structured(Arc::clone(&structured_sessions)) // Même registre que `LaunchAgent` (lot LP3-5) : au swap cross-profile, on // nettoie les fichiers `Replace` orphelins de l'ancien profil avant relance. .with_permission_projectors(Arc::clone(&permission_projectors)), ); // Read-only inventory of resumable agent cells (§15.2). Reuses the shared // project/fs/context/profile stores already injected above — no new port. let list_resumable_agents = Arc::new(ListResumableAgents::new( Arc::clone(&store_port), Arc::clone(&fs_port), Arc::clone(&contexts_port), Arc::clone(&profile_store_port), )); // --- Conversation inspection (T7) --- // Best-effort, optional, extensible: a `Vec` of SessionInspectors routed // by profile. Adding an inspectable CLI = pushing one more adapter here. // The Claude inspector reads `/.claude/projects//.jsonl`; // `$HOME` is resolved from the environment (empty string if unset — the // inspector then simply finds nothing and yields empty details). let home_dir = std::env::var("HOME") .or_else(|_| std::env::var("USERPROFILE")) .unwrap_or_default(); let inspectors: Vec> = vec![Arc::new( ClaudeTranscriptInspector::new(Arc::clone(&fs_port), home_dir), )]; let inspect_conversation = Arc::new(InspectConversation::new( Arc::clone(&contexts_port), Arc::clone(&profile_store_port), inspectors, )); let project_store = Arc::clone(&store_port); let get_project_permissions = Arc::new(GetProjectPermissions::new(Arc::clone( &permission_store_port, ))); let update_project_permissions = Arc::new(UpdateProjectPermissions::new(Arc::clone( &permission_store_port, ))); let update_agent_permissions = Arc::new(UpdateAgentPermissions::new(Arc::clone( &permission_store_port, ))); let resolve_agent_permissions = Arc::new(ResolveAgentPermissions::new(Arc::clone( &permission_store_port, ))); // --- Template store + use cases (L7) --- let template_store = Arc::new(FsTemplateStore::new( Arc::clone(&fs_port), app_data_dir.to_string_lossy().into_owned(), )); let template_store_port = Arc::clone(&template_store) as Arc; let create_template = Arc::new(CreateTemplate::new( Arc::clone(&template_store_port), Arc::clone(&ids) as Arc, )); let update_template = Arc::new(UpdateTemplate::new( Arc::clone(&template_store_port), Arc::clone(&events_port), )); let list_templates = Arc::new(ListTemplates::new(Arc::clone(&template_store_port))); let delete_template = Arc::new(DeleteTemplate::new(Arc::clone(&template_store_port))); let create_agent_from_template = Arc::new(CreateAgentFromTemplate::new( Arc::clone(&template_store_port), Arc::clone(&contexts_port), Arc::clone(&ids) as Arc, Arc::clone(&events_port), )); let detect_agent_drift = Arc::new(DetectAgentDrift::new( Arc::clone(&template_store_port), Arc::clone(&contexts_port), Arc::clone(&events_port), )); let sync_agent_with_template = Arc::new(SyncAgentWithTemplate::new( Arc::clone(&template_store_port), Arc::clone(&contexts_port), Arc::clone(&events_port), )); // --- Git adapter + use cases (L8) --- let git = Arc::new(Git2Repository::new()); let git_port = Arc::clone(&git) as Arc; let git_status = Arc::new(GitStatus::new(Arc::clone(&git_port))); let git_stage = Arc::new(GitStage::new(Arc::clone(&git_port))); let git_unstage = Arc::new(GitUnstage::new(Arc::clone(&git_port))); let git_commit = Arc::new(GitCommit::new( Arc::clone(&git_port), Arc::clone(&events_port), )); let git_branches = Arc::new(GitBranches::new(Arc::clone(&git_port))); let git_checkout = Arc::new(GitCheckout::new( Arc::clone(&git_port), Arc::clone(&events_port), )); let git_log = Arc::new(GitLog::new(Arc::clone(&git_port))); let git_init = Arc::new(GitInit::new( Arc::clone(&git_port), Arc::clone(&events_port), )); let git_graph = Arc::new(GitGraph::new(Arc::clone(&git_port))); // --- Skill use cases (L12) --- // Reuse the skill store (built above for the launcher) and the shared // agent context store for the agent↔skill assignment. let create_skill = Arc::new(CreateSkill::new( Arc::clone(&skill_store_port), Arc::clone(&ids) as Arc, )); let update_skill = Arc::new(UpdateSkill::new(Arc::clone(&skill_store_port))); let list_skills = Arc::new(ListSkills::new(Arc::clone(&skill_store_port))); let delete_skill = Arc::new(DeleteSkill::new(Arc::clone(&skill_store_port))); let assign_skill = Arc::new(AssignSkillToAgent::new( Arc::clone(&contexts_port), Arc::clone(&events_port), )); let unassign_skill = Arc::new(UnassignSkillFromAgent::new( Arc::clone(&contexts_port), Arc::clone(&events_port), )); // --- Memory use cases (LOT A — §14.5.1) --- // `memory_store` / `memory_store_port` / `memory_recall_port` are built // earlier (above LaunchAgent, which needs the recall port). A single // FsMemoryStore takes the project root per call (like the skill store), so // one instance serves every open project. The mutating use cases share the // event bus to announce Memory{Saved,Deleted}. let create_memory = Arc::new(CreateMemory::new( Arc::clone(&memory_store_port), Arc::clone(&events_port), )); let update_memory = Arc::new(UpdateMemory::new( Arc::clone(&memory_store_port), Arc::clone(&events_port), )); let list_memories = Arc::new(ListMemories::new(Arc::clone(&memory_store_port))); let get_memory = Arc::new(GetMemory::new(Arc::clone(&memory_store_port))); let delete_memory = Arc::new(DeleteMemory::new( Arc::clone(&memory_store_port), Arc::clone(&events_port), )); let read_memory_index = Arc::new(ReadMemoryIndex::new(Arc::clone(&memory_store_port))); let resolve_memory_links = Arc::new(ResolveMemoryLinks::new(Arc::clone(&memory_store_port))); // Naïve recall (LOT B): the same instance injected into LaunchAgent above — // composes the store, returns index entries in order truncated to the token // budget. The default, dependency-free MemoryRecall; substitutable by a // VectorMemoryRecall (LOT C). let recall_memory = Arc::new(RecallMemory::new(Arc::clone(&memory_recall_port))); // --- Orchestrator service (§14.3) --- // Dispatches file-based orchestrator requests through the *same* use cases // the UI drives (IdeA stays the single source of truth for the agent/skill // lifecycle). The per-project watcher that feeds it is started lazily when // a project is opened (see `ensure_orchestrator_watch`). // File inter-agents (Option 1 « Terminal + MCP », B-3) : un ticket par tâche // déléguée, résolu par `idea_reply`. Une instance par AppState ⇒ partagée par // tous les projets (clé interne = AgentId, jamais de collision cross-projet). // Médiateur d'entrée (cadrage C3) : enveloppe l'`InMemoryMailbox` (moteur de // corrélation par ticket) + porte la **livraison** du tour dans le PTY de la // cible (écriture sérialisée, une seule voie d'entrée). Le mailbox concret est // partagé pour `resolve`/`resolve_ticket`/`cancel_head` côté orchestrateur. let inmemory_mailbox = Arc::new(InMemoryMailbox::new()); let mailbox = Arc::clone(&inmemory_mailbox) as Arc; let mediated_inbox = Arc::new( MediatedInbox::with_pty( Arc::clone(&inmemory_mailbox), Arc::new(SystemMillisClock), Arc::clone(&pty_port), ) // Émet `AgentBusyChanged` à la source (Busy à l'enqueue qui démarre un // tour, Idle au mark_idle) ⇒ relayé au front en event Tauri (cadrage C4). .with_events(Arc::clone(&events_port)), ); // Détection de stagnation (lot 2, readiness/heartbeat) : une tâche périodique // détenue au composition root appelle `sweep_stalled` (logique de décision pure, // `now` fourni par l'horloge injectée). Bascule `Alive→Stalled` les agents `Busy` // sans battement depuis `stall_after_ms` (profil) et émet `AgentLivenessChanged` // une fois par transition. Tick d'1 s : largement assez fin pour des seuils en // dizaines de secondes, négligeable en charge. Détaché ⇒ vit autant que l'app. { let sweeper = Arc::clone(&mediated_inbox); // `build` runs in Tauri's `setup` hook (main thread, *no* ambient Tokio // runtime) — `tokio::spawn` would panic with "there is no reactor running". // Use Tauri's global async runtime, like `events::spawn_relay` does. tauri::async_runtime::spawn(async move { let mut tick = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tick.tick().await; sweeper.sweep_stalled(); } }); } let input_mediator = Arc::clone(&mediated_inbox) as Arc; // --- Limites de session des agents (ARCHITECTURE §21, LS7) --- // Service pur-ports « détecter → planifier → reprendre » câblé sur l'existant : // l'horloge système, le bus partagé, un `TokioScheduler` (minuterie one-shot // annulable) dont les tâches échues sont drainées plus bas, et un `AppAgentResumer` // qui relance via le *même* `LaunchAgent` que la commande `launch_agent`. let resume_contexts: ResumeContexts = Arc::new(Mutex::new(HashMap::new())); let (resume_tx, mut resume_rx) = tokio::sync::mpsc::unbounded_channel::(); let scheduler = Arc::new(TokioScheduler::new( resume_tx, Arc::clone(&clock) as Arc, )) as Arc; let resumer = Arc::new(AppAgentResumer { launch_agent: Arc::clone(&launch_agent), input_mediator: Arc::clone(&input_mediator), contexts: Arc::clone(&resume_contexts), }) as Arc; let session_limit_service = Arc::new(SessionLimitService::new( Arc::clone(&clock) as Arc, scheduler, Arc::clone(&events_port), resumer, )); // Drain du scheduler (§21.5-b) : à chaque réveil tiré, `TokioScheduler` pousse une // `ScheduledTask::ResumeAgent` ; on l'exécute via le service (relance + AgentResumed). // Patron `sweep_stalled` : `tauri::async_runtime::spawn` (pas `tokio::spawn` — `build` // tourne dans le hook `setup` sans runtime ambiant). Détaché ⇒ vit autant que l'app. { let service = Arc::clone(&session_limit_service); tauri::async_runtime::spawn(async move { while let Some(task) = resume_rx.recv().await { if let Err(e) = service.execute_resume(task).await { // Best-effort : une relance qui échoue ne fige pas le drain. eprintln!("[session-limit] reprise auto échouée : {e}"); } } }); } // Registre des conversations par paire (cadrage C3) : un fil par paire, session // vivante keyée par conversation (lève l'ambiguïté session/agent). let conversation_registry = Arc::new(InMemoryConversationRegistry::new()) as Arc; let orchestrator_service = Arc::new( OrchestratorService::new( Arc::clone(&create_agent), Arc::clone(&launch_agent), Arc::clone(&list_agents), Arc::clone(&close_terminal), Arc::clone(&update_agent_context), Arc::clone(&create_skill), Arc::clone(&profile_store_port), Arc::clone(&terminal_sessions), ) // Messagerie inter-agents (cadrage C3) : médiateur d'entrée (file + livraison // PTV sérialisée) + registre de conversations + bus pour AgentReplied. .with_input_mediator(Arc::clone(&input_mediator), Arc::clone(&mailbox)) .with_conversations(Arc::clone(&conversation_registry)) .with_events(Arc::clone(&events_port)) // Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour // que les (re)lancements issus du chemin `ask` écrivent la déclaration MCP // réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip. .with_mcp_runtime_provider( Arc::new(AppMcpRuntimeProvider) as Arc ) // Persistance conversationnelle best-effort (lot P6b) : Prompt + Response de // chaque paire déléguée écrits dans `/.ideai/conversations/` via // le provider per-root + l'horloge millis (port Clock) déjà câblée. Un échec // n'affecte jamais la délégation live. .with_record_turn( Arc::new(AppRecordTurnProvider) as Arc, Arc::clone(&clock) as Arc, ), // NB (régression corrigée) : on ne câble PAS `.with_structured(...)` ici. // Décision produit lot B-2 (« Option 1 Terminal + MCP », cf. construction // de `LaunchAgent` plus haut) : la fabrique structurée est décâblée, donc // AUCUN agent n'a de session structurée vivante — tous tournent en PTY brut // et la délégation passe par les outils MCP (`idea_ask_agent`/`idea_reply`). // Si l'orchestrateur recevait `with_structured`, `ask_agent` emprunterait la // branche structurée (`ensure_structured_session`) qui ne peut jamais aboutir // (le launcher ne crée plus de session structurée) ⇒ erreur systématique // « aucune session structurée vivante après lancement ». On laisse donc // `self.structured = None` pour que `ask_agent` retombe sur le chemin PTY+MCP // fonctionnel. `drain_with_readiness` (readiness/heartbeat lot 1) reste dormant // tant que la voie structurée n'est pas réactivée au composition root. ); // --- Windows (L10) --- let move_tab = Arc::new(MoveTabToNewWindow::new( Arc::clone(&store_port), Arc::clone(&ids) as Arc, )); Self { health, create_project, open_project, close_project, close_tab, list_projects, read_project_context, update_project_context, open_terminal, write_terminal, resize_terminal, close_terminal, load_layout, mutate_layout, list_layouts, create_layout, rename_layout, delete_layout, set_active_layout, snapshot_running_agents, reconcile_layouts, detect_profiles, list_profiles, save_profile, delete_profile, configure_profiles, reference_profiles, first_run_state, pty_port, terminal_sessions, event_bus, pty_bridge, structured_sessions, chat_bridge, create_agent, list_agents, read_agent_context, update_agent_context, delete_agent, launch_agent, change_agent_profile, list_resumable_agents, inspect_conversation, project_store, get_project_permissions, update_project_permissions, update_agent_permissions, resolve_agent_permissions, create_template, update_template, list_templates, delete_template, create_agent_from_template, detect_agent_drift, sync_agent_with_template, git_status, git_stage, git_unstage, git_commit, git_branches, git_checkout, git_log, git_init, git_graph, create_skill, update_skill, list_skills, delete_skill, assign_skill, unassign_skill, create_memory, update_memory, list_memories, get_memory, delete_memory, read_memory_index, resolve_memory_links, recall_memory, list_embedder_profiles, save_embedder_profile, delete_embedder_profile, describe_embedder_engines, dismiss_embedder_suggestion, orchestrator_service, orchestrator_watchers: Mutex::new(HashMap::new()), mcp_servers: Mutex::new(HashMap::new()), session_limit_service, resume_contexts, move_tab, } } /// Starts a [`FsOrchestratorWatcher`] for `project` if one is not already /// running for it (idempotent). The watcher tails the project's /// `.ideai/requests/` tree and dispatches each request through the shared /// [`OrchestratorService`]; processed-request events are republished on the /// domain event bus so the relay surfaces them to the frontend. /// /// Called from the `open_project` / `create_project` commands. Must run inside /// the Tokio runtime (the watcher spawns a background task) — Tauri async /// commands satisfy this. pub fn ensure_orchestrator_watch(&self, project: &Project) { let mut watchers = self .orchestrator_watchers .lock() .expect("orchestrator watcher registry poisoned"); if watchers.contains_key(&project.id) { return; } let bus = Arc::clone(&self.event_bus); let events: Arc = Arc::new(move |event| bus.publish(event)); let handle = FsOrchestratorWatcher::start( project.clone(), Arc::clone(&self.orchestrator_service), events, ); watchers.insert(project.id, handle); drop(watchers); // Start the MCP server for this project, beside (and in parallel with) the // file watcher — both are entry doors onto the *same* OrchestratorService // (Décision 4). Idempotent like the watcher above. self.ensure_mcp_server(project); } /// Starts an [`McpServer`] for `project` if one is not already registered /// (idempotent), the **twin** of [`ensure_orchestrator_watch`]. Registers its /// lifecycle handle in [`mcp_servers`](Self::mcp_servers); dropping/stopping the /// handle tears down the supervision task. /// /// ## Transport decision (S-MCP, cadrage v5 §1) & endpoint lifecycle (M5a) /// /// At project-open time there is **no CLI connected yet**: a peer only appears /// once an MCP-capable agent is launched with the injected MCP config (M5d) and /// its spawned `idea mcp-server` bridge dials this project's loopback endpoint. /// We therefore do **not** run a blocking `McpServer::serve`/accept loop here /// (that must never figer the open/close of a project). M5a's job is narrower: /// **bind the project's loopback endpoint** ([`mcp_endpoint`], the single source /// of truth) so it is ready when a bridge connects, and **hold** the listener in /// the handle. The actual accept-and-serve-per-peer is M5b/M5c. /// /// Binding is **non-blocking** (create the listener, then park on the stop /// signal) and **idempotent** (one endpoint per project: a second open returns /// early without rebinding). On close the handle is dropped, which on Unix /// unlinks the socket file (interprocess reclaim guard) — no leak. fn ensure_mcp_server(&self, project: &Project) { let mut servers = self .mcp_servers .lock() .expect("mcp server registry poisoned"); if servers.contains_key(&project.id) { return; } let bus = Arc::clone(&self.event_bus); let events: Arc = Arc::new(move |event| bus.publish(event)); let endpoint = mcp_endpoint(&project.id); let listener = bind_endpoint(&endpoint); // The project-id string the handshake guard compares against: the same // hyphen-free hex form the endpoint encodes, which M5d's `--project` reuses. let project_id = project.id.as_uuid().simple().to_string(); // Readiness de démarrage : quand le pont MCP d'un agent se connecte (initialize), // libère son éventuel 1er tour différé (fix race cold-launch via signal MCP). // L'id arrive en hex (handshake `requester`) ⇒ on le parse en AgentId ici (la // composition root est la seule à connaître la frontière infra↔domaine). let service_for_ready = Arc::clone(&self.orchestrator_service); let ready_sink: Arc = Arc::new(move |requester: &str| { if let Ok(uuid) = Uuid::parse_str(requester) { service_for_ready.release_agent_cold_start(AgentId::from_uuid(uuid)); } }); let handle = McpServerHandle::start( McpServer::new(Arc::clone(&self.orchestrator_service), project.clone()) .with_events(events) .with_ready_sink(ready_sink), endpoint, listener, project_id, ); servers.insert(project.id, handle); } /// Returns the ids of every currently-open project. /// /// Derived from the orchestrator watcher registry, which holds exactly one /// entry per open project (started on open/create, dropped on close). Used by /// the shutdown hook to snapshot running agents across all open projects /// before the global PTY kill. #[must_use] pub fn open_project_ids(&self) -> Vec { self.orchestrator_watchers .lock() .map(|w| w.keys().copied().collect()) .unwrap_or_default() } /// Repairs the persisted Claude run-dir artefacts of `project` so a freshly /// launched AppImage does not inherit stale MCP/settings state from older runs. /// /// Best-effort and local-only: remote SSH/WSL projects are skipped because this /// migration rewrites the local run-dir files the AppImage owns. Launch-time /// repair still remains available on the normal activation path. pub async fn reconcile_claude_run_dirs(&self, project: &Project) { self.migrate_claude_run_dirs(project).await; } /// Stops and removes the orchestrator watcher for `project_id`, if any. /// Called from `close_project` so a closed project stops consuming requests. /// Symmetrically stops the project's MCP server (its twin) so both entry doors /// are torn down together. pub fn stop_orchestrator_watch(&self, project_id: &ProjectId) { if let Some(handle) = self .orchestrator_watchers .lock() .expect("orchestrator watcher registry poisoned") .remove(project_id) { handle.stop(); } if let Some(handle) = self .mcp_servers .lock() .expect("mcp server registry poisoned") .remove(project_id) { handle.stop(); } } /// Best-effort migration of persisted Claude run dirs at project-open time: /// repairs stale `.mcp.json` declarations and missing /// `enabledMcpjsonServers` entries in `.claude/settings.local.json`. pub async fn migrate_claude_run_dirs(&self, project: &Project) { if project.remote.kind() != RemoteKind::Local { return; } let agents = match self .list_agents .execute(ListAgentsInput { project: project.clone(), }) .await { Ok(output) => output.agents, Err(_) => return, }; let profiles = match self.list_profiles.execute().await { Ok(output) => output.profiles, Err(_) => return, }; let profile_by_id: HashMap<_, _> = profiles.into_iter().map(|p| (p.id, p)).collect(); for agent in agents { let Some(profile) = profile_by_id.get(&agent.profile_id) else { continue; }; // Dispatch par profil : Claude (`.mcp.json` + settings) ou Codex // (`config.toml` isolé via `CODEX_HOME`). Tout autre profil ⇒ rien à migrer. if is_claude_mcp_profile(profile) { let _ = migrate_claude_run_dir(project, &agent.id, profile).await; } else if is_codex_mcp_profile(profile) { let _ = migrate_codex_run_dir(project, &agent.id, profile).await; } } } } async fn migrate_claude_run_dir( project: &Project, agent_id: &AgentId, profile: &AgentProfile, ) -> Result<(), std::io::Error> { let run_dir = project .root .as_str() .trim_end_matches(['/', '\\']) .to_owned() + &format!("/.ideai/run/{agent_id}"); let run_dir_path = Path::new(&run_dir); if tokio::fs::metadata(run_dir_path).await.is_err() { return Ok(()); } migrate_claude_settings(run_dir_path, project.root.as_str()).await?; let runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { exe, endpoint: crate::mcp_endpoint::mcp_endpoint(&project.id) .as_cli_arg() .to_owned(), project_id: project.id.as_uuid().simple().to_string(), requester: agent_id.to_string(), }); migrate_claude_mcp_config(run_dir_path, profile, runtime.as_ref()).await?; Ok(()) } async fn migrate_claude_settings(run_dir: &Path, project_root: &str) -> Result<(), std::io::Error> { let claude_dir = run_dir.join(".claude"); let settings_path = claude_dir.join("settings.local.json"); let next = match tokio::fs::read_to_string(&settings_path).await { Ok(existing) => merge_claude_settings_json(&existing, project_root), Err(err) if err.kind() == std::io::ErrorKind::NotFound => { Some(claude_settings_seed_value(project_root)) } Err(err) => return Err(err), }; let Some(next) = next else { return Ok(()); }; tokio::fs::create_dir_all(&claude_dir).await?; let mut body = serde_json::to_string_pretty(&next).map_err(std::io::Error::other)?; body.push('\n'); write_atomically(&settings_path, &body) } async fn migrate_claude_mcp_config( run_dir: &Path, profile: &AgentProfile, runtime: Option<&McpRuntime>, ) -> Result<(), std::io::Error> { let Some(target) = claude_mcp_config_target(profile) else { return Ok(()); }; let Some(runtime) = runtime else { return Ok(()); }; let mcp_path = run_dir.join(target); let desired_server = mcp_server_entry(profile, Some(runtime)); let next = match tokio::fs::read_to_string(&mcp_path).await { Ok(existing) => merge_mcp_json(&existing, desired_server), Err(err) if err.kind() == std::io::ErrorKind::NotFound => { Some(wrap_idea_mcp_server(desired_server)) } Err(err) => return Err(err), }; let Some(next) = next else { return Ok(()); }; if let Some(parent) = mcp_path.parent() { tokio::fs::create_dir_all(parent).await?; } let mut body = serde_json::to_string_pretty(&next).map_err(std::io::Error::other)?; body.push('\n'); write_atomically(&mcp_path, &body) } fn is_claude_mcp_profile(profile: &AgentProfile) -> bool { let is_claude = profile.structured_adapter == Some(StructuredAdapter::Claude) || matches!( &profile.context_injection, ContextInjection::ConventionFile { target } if target .rsplit(['/', '\\']) .next() .unwrap_or(target) .eq_ignore_ascii_case("CLAUDE.md") ); is_claude && claude_mcp_config_target(profile).is_some() } fn claude_mcp_config_target(profile: &AgentProfile) -> Option<&str> { match profile.mcp.as_ref().map(|mcp| &mcp.config) { Some(McpConfigStrategy::ConfigFile { target }) => Some(target.as_str()), _ => None, } } /// Pendant Codex de [`migrate_claude_run_dir`] : répare le `config.toml` MCP isolé /// du run dir (table `[mcp_servers.idea]`) pour que Codex, qui lit ses serveurs MCP /// dans `$CODEX_HOME/config.toml`, voie les outils `idea_*` après un redémarrage. /// `CODEX_HOME` est isolé au run dir au lancement (jamais le `~/.codex` global), donc /// ce fichier appartient entièrement à IdeA : il est régénéré (clobber) dès qu'un /// runtime réel est disponible. Best-effort, idempotent. async fn migrate_codex_run_dir( project: &Project, agent_id: &AgentId, profile: &AgentProfile, ) -> Result<(), std::io::Error> { let run_dir = project .root .as_str() .trim_end_matches(['/', '\\']) .to_owned() + &format!("/.ideai/run/{agent_id}"); let run_dir_path = Path::new(&run_dir); if tokio::fs::metadata(run_dir_path).await.is_err() { return Ok(()); } let runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { exe, endpoint: crate::mcp_endpoint::mcp_endpoint(&project.id) .as_cli_arg() .to_owned(), project_id: project.id.as_uuid().simple().to_string(), requester: agent_id.to_string(), }); migrate_codex_mcp_config(run_dir_path, profile, runtime.as_ref()).await } async fn migrate_codex_mcp_config( run_dir: &Path, profile: &AgentProfile, runtime: Option<&McpRuntime>, ) -> Result<(), std::io::Error> { let Some((target, _home_env)) = codex_mcp_config_target(profile) else { return Ok(()); }; // Sans runtime réel, seule une déclaration minimale est disponible : on ne // régénère pas (mirror de `migrate_claude_mcp_config`). let Some(runtime) = runtime else { return Ok(()); }; let toml_path = run_dir.join(target); let desired = mcp_server_entry_toml(profile, Some(runtime)); // `config.toml` isolé = entièrement géré par IdeA ⇒ régénération (clobber) ; // idempotent (no-op si le contenu est déjà à jour). match tokio::fs::read_to_string(&toml_path).await { Ok(existing) if existing == desired => return Ok(()), Ok(_) => {} Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} Err(err) => return Err(err), } if let Some(parent) = toml_path.parent() { tokio::fs::create_dir_all(parent).await?; } write_atomically(&toml_path, &desired) } fn is_codex_mcp_profile(profile: &AgentProfile) -> bool { let is_codex = profile.structured_adapter == Some(StructuredAdapter::Codex) || matches!( &profile.context_injection, ContextInjection::ConventionFile { target } if target .rsplit(['/', '\\']) .next() .unwrap_or(target) .eq_ignore_ascii_case("AGENTS.md") ); is_codex && codex_mcp_config_target(profile).is_some() } fn codex_mcp_config_target(profile: &AgentProfile) -> Option<(&str, &str)> { match profile.mcp.as_ref().map(|mcp| &mcp.config) { Some(McpConfigStrategy::TomlConfigHome { target, home_env }) => { Some((target.as_str(), home_env.as_str())) } _ => None, } } /// Rend le contenu du `config.toml` Codex (table `[mcp_servers.idea]`) en réutilisant /// l'encodeur TOML partagé [`domain::McpServerWiring::to_config_toml`] (D2) — même /// source de wiring que la déclaration `.mcp.json` Claude, donc zéro dérive. fn mcp_server_entry_toml(profile: &AgentProfile, runtime: Option<&McpRuntime>) -> String { let transport = profile .mcp .as_ref() .map_or(McpTransport::Stdio, |m| m.transport); let (command, args) = match runtime { Some(rt) => ( rt.exe.clone(), vec![ "mcp-server".to_owned(), "--endpoint".to_owned(), rt.endpoint.clone(), "--project".to_owned(), rt.project_id.clone(), "--requester".to_owned(), rt.requester.clone(), ], ), None => ("idea".to_owned(), vec!["mcp-server".to_owned()]), }; domain::McpServerWiring::new(command, args, transport).to_config_toml() } fn merge_claude_settings_json(existing: &str, project_root: &str) -> Option { let mut doc = match serde_json::from_str::(existing) { Ok(Value::Object(map)) => Value::Object(map), Ok(_) | Err(_) => return Some(claude_settings_seed_value(project_root)), }; let before = doc.clone(); let root = doc.as_object_mut().expect("object preserved above"); let permissions = root .entry("permissions".to_owned()) .or_insert_with(|| Value::Object(Map::new())); if !permissions.is_object() { *permissions = Value::Object(Map::new()); } let permissions = permissions .as_object_mut() .expect("permissions forced to object"); permissions.insert( "defaultMode".to_owned(), Value::String("bypassPermissions".to_owned()), ); permissions.insert( "additionalDirectories".to_owned(), Value::Array(merge_string_array( permissions.get("additionalDirectories"), [project_root], )), ); permissions.insert( "allow".to_owned(), Value::Array(merge_string_array( permissions.get("allow"), ["Read", "Edit", "Write", "Bash"], )), ); permissions.insert( "deny".to_owned(), Value::Array(merge_string_array( permissions.get("deny"), [ "Bash(sudo *)", "Bash(rm -rf /)", "Bash(rm -rf /*)", "Bash(rm -rf ~)", "Bash(rm -rf ~/)", "Bash(rm -rf ~/*)", "Bash(rm -rf $HOME*)", "Bash(mkfs*)", "Bash(dd if=*)", "Bash(shutdown*)", "Bash(reboot*)", ], )), ); root.insert( "skipDangerousModePermissionPrompt".to_owned(), Value::Bool(true), ); root.insert( "enabledMcpjsonServers".to_owned(), Value::Array(merge_string_array( root.get("enabledMcpjsonServers"), ["idea"], )), ); let sandbox = root .entry("sandbox".to_owned()) .or_insert_with(|| Value::Object(Map::new())); if !sandbox.is_object() { *sandbox = Value::Object(Map::new()); } let sandbox = sandbox.as_object_mut().expect("sandbox forced to object"); sandbox.insert("enabled".to_owned(), Value::Bool(false)); if doc != before { Some(doc) } else { None } } fn merge_mcp_json(existing: &str, desired_server: Value) -> Option { let mut doc = match serde_json::from_str::(existing) { Ok(Value::Object(map)) => map, Ok(_) | Err(_) => return Some(wrap_idea_mcp_server(desired_server)), }; let mcp_servers = doc .entry("mcpServers".to_owned()) .or_insert_with(|| Value::Object(Map::new())); if !mcp_servers.is_object() { *mcp_servers = Value::Object(Map::new()); } let changed = mcp_servers.get("idea") != Some(&desired_server); if !changed { return None; } match mcp_servers { Value::Object(servers) => { servers.insert("idea".to_owned(), desired_server); Some(Value::Object(doc)) } _ => None, } } fn wrap_idea_mcp_server(server: Value) -> Value { let mut servers = Map::new(); servers.insert("idea".to_owned(), server); let mut root = Map::new(); root.insert("mcpServers".to_owned(), Value::Object(servers)); Value::Object(root) } fn mcp_server_entry(profile: &AgentProfile, runtime: Option<&McpRuntime>) -> Value { let transport = match profile.mcp.as_ref().map(|mcp| mcp.transport) { Some(McpTransport::Socket) => "socket", Some(McpTransport::Stdio) | None => "stdio", }; let (command, args) = match runtime { Some(rt) => ( rt.exe.clone(), vec![ Value::String("mcp-server".to_owned()), Value::String("--endpoint".to_owned()), Value::String(rt.endpoint.clone()), Value::String("--project".to_owned()), Value::String(rt.project_id.clone()), Value::String("--requester".to_owned()), Value::String(rt.requester.clone()), ], ), None => ( "idea".to_owned(), vec![Value::String("mcp-server".to_owned())], ), }; let mut server = Map::new(); server.insert("command".to_owned(), Value::String(command)); server.insert("args".to_owned(), Value::Array(args)); server.insert("transport".to_owned(), Value::String(transport.to_owned())); Value::Object(server) } fn claude_settings_seed_value(project_root: &str) -> Value { json!({ "permissions": { "defaultMode": "bypassPermissions", "additionalDirectories": [project_root], "allow": ["Read", "Edit", "Write", "Bash"], "deny": [ "Bash(sudo *)", "Bash(rm -rf /)", "Bash(rm -rf /*)", "Bash(rm -rf ~)", "Bash(rm -rf ~/)", "Bash(rm -rf ~/*)", "Bash(rm -rf $HOME*)", "Bash(mkfs*)", "Bash(dd if=*)", "Bash(shutdown*)", "Bash(reboot*)" ] }, "skipDangerousModePermissionPrompt": true, "enabledMcpjsonServers": ["idea"], "sandbox": { "enabled": false } }) } fn merge_string_array<'a>( existing: Option<&Value>, required: impl IntoIterator, ) -> Vec { let mut seen = HashSet::::new(); let mut out = Vec::new(); if let Some(existing) = existing.and_then(Value::as_array) { for item in existing.iter().filter_map(Value::as_str) { if seen.insert(item.to_owned()) { out.push(Value::String(item.to_owned())); } } } for item in required { if seen.insert(item.to_owned()) { out.push(Value::String(item.to_owned())); } } out } fn write_atomically(path: &Path, content: &str) -> std::io::Result<()> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } let file_name = path .file_name() .map(OsString::from) .unwrap_or_else(|| OsString::from("tmp")); let tmp_name = format!(".{}.{}.tmp", file_name.to_string_lossy(), Uuid::new_v4()); let tmp_path = path.with_file_name(tmp_name); std::fs::write(&tmp_path, content.as_bytes())?; #[cfg(windows)] if path.exists() { let _ = std::fs::remove_file(path); } std::fs::rename(&tmp_path, path)?; Ok(()) } #[cfg(test)] mod run_dir_migration_tests { use super::{ claude_settings_seed_value, is_claude_mcp_profile, mcp_server_entry, merge_claude_settings_json, merge_mcp_json, migrate_claude_run_dir, }; use application::McpRuntimeProvider; use domain::ids::{AgentId, ProfileId, ProjectId}; use domain::profile::{ AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport, StructuredAdapter, }; use serde_json::json; use uuid::Uuid; fn claude_profile() -> AgentProfile { AgentProfile::new( ProfileId::from_uuid(Uuid::from_u128(9)), "Claude Code", "claude", Vec::new(), ContextInjection::convention_file("CLAUDE.md").unwrap(), None, "{agentRunDir}", None, ) .unwrap() .with_structured_adapter(StructuredAdapter::Claude) .with_mcp(McpCapability::new( McpConfigStrategy::config_file(".mcp.json").unwrap(), McpTransport::Stdio, )) } fn runtime(agent_id: AgentId) -> application::McpRuntime { application::McpRuntime { exe: "/opt/IdeA.AppImage".to_owned(), endpoint: "/run/user/1000/idea-mcp/proj.sock".to_owned(), project_id: ProjectId::from_uuid(Uuid::from_u128(1234)) .as_uuid() .simple() .to_string(), requester: agent_id.to_string(), } } #[test] fn merge_claude_settings_adds_idea_and_preserves_existing_entries() { let merged = merge_claude_settings_json( r#"{ "permissions": { "additionalDirectories": ["/tmp/custom"], "allow": ["Read"], "deny": ["Bash(custom)"] }, "enabledMcpjsonServers": ["other"], "extra": true }"#, "/home/me/proj", ) .unwrap(); let parsed = merged; assert_eq!(parsed["extra"], json!(true)); assert_eq!( parsed["permissions"]["defaultMode"], json!("bypassPermissions") ); assert!(parsed["permissions"]["additionalDirectories"] .as_array() .unwrap() .contains(&json!("/tmp/custom"))); assert!(parsed["permissions"]["additionalDirectories"] .as_array() .unwrap() .contains(&json!("/home/me/proj"))); assert!(parsed["enabledMcpjsonServers"] .as_array() .unwrap() .contains(&json!("other"))); assert!(parsed["enabledMcpjsonServers"] .as_array() .unwrap() .contains(&json!("idea"))); } #[test] fn merge_idea_mcp_json_rewrites_idea_and_preserves_other_servers() { let agent_id = AgentId::from_uuid(Uuid::from_u128(77)); let desired = mcp_server_entry(&claude_profile(), Some(&runtime(agent_id))); let merged = merge_mcp_json( r#"{ "mcpServers": { "idea": { "command": "idea", "args": ["mcp-server"], "transport": "stdio" }, "other": { "command": "keep-me", "args": [] } } }"#, desired.clone(), ) .unwrap(); let parsed = merged; assert_eq!(parsed["mcpServers"]["other"]["command"], json!("keep-me")); assert_eq!(parsed["mcpServers"]["idea"], desired); } #[test] fn reconcile_claude_run_dir_repairs_legacy_files_on_disk() { let agent_id = AgentId::from_uuid(Uuid::from_u128(88)); let temp = std::env::temp_dir().join(format!("idea-run-migrate-{}", Uuid::new_v4())); let project_root = temp.join("project"); let run_dir = project_root.join(".ideai/run").join(agent_id.to_string()); std::fs::create_dir_all(run_dir.join(".claude")).unwrap(); std::fs::write( run_dir.join(".claude/settings.local.json"), r#"{"permissions":{"additionalDirectories":["/tmp/old"]}}"#, ) .unwrap(); std::fs::write( run_dir.join(".mcp.json"), r#"{"mcpServers":{"idea":{"command":"idea","args":["mcp-server"],"transport":"stdio"}}}"#, ) .unwrap(); let project = domain::Project::new( ProjectId::from_uuid(Uuid::from_u128(1234)), "demo", domain::project::ProjectPath::new(project_root.to_string_lossy().into_owned()).unwrap(), domain::remote::RemoteRef::local(), 1, ) .unwrap(); let profile = claude_profile(); tokio::runtime::Builder::new_current_thread() .enable_all() .build() .unwrap() .block_on(async { migrate_claude_run_dir(&project, &agent_id, &profile) .await .unwrap(); }); let settings: serde_json::Value = serde_json::from_str( &std::fs::read_to_string(run_dir.join(".claude/settings.local.json")).unwrap(), ) .unwrap(); assert!(settings["enabledMcpjsonServers"] .as_array() .unwrap() .contains(&json!("idea"))); assert!(settings["permissions"]["additionalDirectories"] .as_array() .unwrap() .contains(&json!(project.root.as_str()))); let mcp: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(run_dir.join(".mcp.json")).unwrap()) .unwrap(); let expected_runtime = crate::mcp_endpoint::AppMcpRuntimeProvider .runtime_for(&project, agent_id) .unwrap(); assert_eq!( mcp["mcpServers"]["idea"], mcp_server_entry(&claude_profile(), Some(&expected_runtime)) ); let _ = std::fs::remove_dir_all(temp); } #[test] fn claude_profile_guard_matches_only_claude_mcp_profiles() { assert!(is_claude_mcp_profile(&claude_profile())); } #[test] fn invalid_settings_fall_back_to_seed() { let merged = merge_claude_settings_json("{not-json", "/home/me/proj").unwrap(); assert_eq!(merged, claude_settings_seed_value("/home/me/proj")); } #[test] fn malformed_typed_settings_are_repaired_without_panicking() { let merged = merge_claude_settings_json( r#"{ "permissions": [], "sandbox": false, "enabledMcpjsonServers": "idea" }"#, "/home/me/proj", ) .unwrap(); assert_eq!( merged["permissions"]["defaultMode"], json!("bypassPermissions") ); assert_eq!(merged["sandbox"]["enabled"], json!(false)); assert_eq!(merged["enabledMcpjsonServers"], json!(["idea"])); } } /// Binds the project's loopback endpoint listener (M5a). Best-effort: returns the /// bound [`LocalSocketListener`] or `None` if the bind fails (e.g. a stale socket /// file from a crashed run). A `None` never figes open/close — the registry entry /// is still created so the lifecycle stays idempotent; the real accept/serve (M5c) /// will surface a hard failure if it actually needs the listener. /// /// On Unix this binds a filesystem-path UDS under the per-user runtime dir; the /// parent directory is created if missing, and a corpse socket from a previous run /// is replaced (`reclaim_name` so the file is unlinked on drop). On Windows it binds /// the named pipe — no filesystem entry to manage. #[must_use] fn bind_endpoint(endpoint: &McpEndpoint) -> Option { // Ensure the runtime dir exists (Unix path sockets need their parent dir). if let Some(path) = endpoint.socket_path() { if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); } // D1 — reclaim the **corpse** socket left by a SIGKILL'd run BEFORE binding. // `reclaim_name(true)` (below) only unlinks the socket on *drop*; in // `interprocess` 2.4 it does **not** clear a pre-existing inode at bind time, // so a stale socket file makes the bind fail with `EADDRINUSE` (the D1 test // proves this). We therefore unlink it ourselves first — but only when the // path is actually a **socket** (never clobber a real file we don't own). #[cfg(unix)] { use std::os::unix::fs::FileTypeExt; if let Ok(meta) = std::fs::symlink_metadata(&path) { if meta.file_type().is_socket() { let _ = std::fs::remove_file(&path); } } } } let name = endpoint.as_cli_arg().to_fs_name::().ok()?; ListenerOptions::new() .name(name) // Reclaim (unlink) on drop so a clean close leaves no socket behind. .reclaim_name(true) .create_tokio() .ok() } /// Drives one accepted loopback peer (= one `idea mcp-server` bridge = one agent): /// reads its **handshake line**, then serves the rest of the stream as JSON-RPC. /// /// Sequence (cadrage v5 §1.4, M5b handshake format): /// 1. Split the duplex connection so reads and writes are independent. /// 2. Read **one** newline-terminated handshake line /// (`{"project":"…","requester":"…"}`) off a `BufReader` over the read half. /// 3. If the handshake's `project` is present and **mismatches** this server's /// project, close the connection (return) without serving — a defensive guard, it /// logs nothing and never crashes the accept loop. /// 4. Wrap the **same** `BufReader` (carrying any bytes already buffered past the /// handshake) + the write half in a [`StdioTransport`] and hand it to /// [`McpServer::serve_as`] tagged with the handshake's `requester` — so /// `OrchestratorRequestProcessed.requester_id` becomes the real agent id. /// /// Generic over the stream type so it never names the `interprocess` connection /// type and stays unit-testable over an in-memory duplex. async fn serve_peer(server: Arc, expected_project: &str, conn: S) where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static, { let (read_half, write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); // 1 handshake line. A read error or immediate EOF ⇒ no peer to serve. let mut handshake = String::new(); match reader.read_line(&mut handshake).await { Ok(0) | Err(_) => return, Ok(_) => {} } let (handshake_project, requester) = parse_handshake(&handshake); // Defensive: a bridge dialed the wrong project's endpoint. Drop it cleanly. if !handshake_project.is_empty() && !expected_project.is_empty() && handshake_project != expected_project { return; } // The same BufReader keeps any bytes already buffered past the handshake, so no // JSON-RPC line is lost between the handshake and the serve loop. let mut transport = StdioTransport::from_buffered(reader, write_half); server.serve_as(requester, &mut transport).await; } /// Parses the M5b handshake line into `(project, requester)`. Tolerant: a malformed /// or partial line yields empty strings (⇒ legacy `"mcp"` requester, no project /// guard), never an error — the serve loop must never crash on a bad handshake. fn parse_handshake(line: &str) -> (String, String) { serde_json::from_str::(line.trim()) .ok() .map(|v| { let project = v .get("project") .and_then(serde_json::Value::as_str) .unwrap_or_default() .to_owned(); let requester = v .get("requester") .and_then(serde_json::Value::as_str) .unwrap_or_default() .to_owned(); (project, requester) }) .unwrap_or_default() } /// Lifecycle handle for a per-project [`McpServer`] supervision task — the **twin** /// of [`OrchestratorWatchHandle`](infrastructure::OrchestratorWatchHandle). /// /// It carries the **same** stop mechanism as the watcher (a one-shot /// `mpsc::Sender<()>`): [`stop`](Self::stop) signals the task to exit, and dropping /// the handle stops it too (the channel closes). /// /// ## Accept-and-serve-per-peer (M5c) /// /// The supervision task runs an **`accept` loop** on the bound loopback `listener`, /// arbitrated against the stop signal by `tokio::select!`: /// - **`accept`** is async and parks on the absence of a peer, so the loop never /// figes the open/close of a project — at project-open time no CLI is connected /// yet (a peer only appears once an MCP-capable agent is launched, M5d). /// - **each accepted connection** = one `idea mcp-server` bridge = one agent. The /// task reads the **handshake line** (`{"project","requester"}`, cadrage v5 §1.4), /// then spawns an isolated task running [`McpServer::serve_as`] over the rest of /// the stream, so one peer's disconnection/error never affects the others or the /// accept loop. /// - **stop** breaks the loop, **aborts** the in-flight serve tasks (`JoinSet`), /// and drops the listener — which on Unix unlinks the socket file via /// interprocess' reclaim guard, so closing a project leaves no socket behind. pub struct McpServerHandle { stop: tokio::sync::mpsc::Sender<()>, /// The loopback address this server listens on — the single source of truth /// ([`mcp_endpoint`]) shared with the CLI-declaration writer (M5d). endpoint: McpEndpoint, } impl McpServerHandle { /// Spawns the supervision task that owns `server` and the bound `listener`, runs /// the accept-and-serve-per-peer loop, and stops cleanly on signal. Must run /// inside the ambient Tokio runtime (Tauri async commands satisfy this), exactly /// like [`FsOrchestratorWatcher::start`](infrastructure::FsOrchestratorWatcher). #[must_use] fn start( server: McpServer, endpoint: McpEndpoint, listener: Option, project_id: String, ) -> Self { let (stop_tx, mut stop_rx) = tokio::sync::mpsc::channel::<()>(1); // The base server is shared (Arc) so each accepted peer derives its own // requester-tagged clone (McpServer::serve_as) without contending. let server = Arc::new(server); tokio::spawn(async move { // No listener (bind failed, e.g. stale socket): nothing to accept. Park // on stop so the lifecycle stays idempotent and non-blocking. let Some(listener) = listener else { let _ = stop_rx.recv().await; return; }; // In-flight per-peer serve tasks; aborted en masse on stop so no serve // outlives the project's close. let mut serves: tokio::task::JoinSet<()> = tokio::task::JoinSet::new(); loop { tokio::select! { // Stop requested (or the handle dropped, closing the channel). _ = stop_rx.recv() => break, accepted = listener.accept() => { match accepted { Ok(conn) => { let server = Arc::clone(&server); let expected_project = project_id.clone(); serves.spawn(async move { serve_peer(server, &expected_project, conn).await; }); } // A transient accept error must not kill the loop; the // listener stays bound and keeps accepting the next peer. Err(_) => continue, } } } } // Stop: terminate every in-flight serve, then drop the listener (unlinks // the Unix socket file). Pending peers die with their owning CLI anyway. serves.abort_all(); drop(serves); drop(listener); }); Self { stop: stop_tx, endpoint, } } /// The loopback endpoint this project's server is bound to (M5a source of truth). #[must_use] pub fn endpoint(&self) -> &McpEndpoint { &self.endpoint } /// Signals the supervision task to stop (best-effort; dropping the handle also /// stops it). Mirrors [`OrchestratorWatchHandle::stop`](infrastructure::OrchestratorWatchHandle::stop). pub fn stop(&self) { let _ = self.stop.try_send(()); } } /// Build the project memory recall port from an embedder profile. /// /// This is the only place that knows the concrete recall adapters (DIP): callers /// only ever see `Arc`. With the default `none` profile, /// `embedder_from_profile` returns `None` and we hand back the plain /// `NaiveMemoryRecall` — strictly identical, dependency-free behaviour. When an /// embedder is configured, we wrap naïve + vector recall in an /// `AdaptiveMemoryRecall` that switches stages live per the profile strategy. pub(crate) fn build_memory_recall( fs: Arc, store: Arc, profile: &EmbedderProfile, onnx_cache_dir: &std::path::Path, ) -> Arc { let naive = Arc::new(NaiveMemoryRecall::new(Arc::clone(&store))) as Arc; match embedder_from_profile(profile, onnx_cache_dir) { None => naive, Some(embedder) => { let embedder: Arc = Arc::from(embedder); let vector: Arc = Arc::new(VectorMemoryRecall::new( embedder, Arc::clone(&store), Arc::clone(&fs), )); Arc::new(AdaptiveMemoryRecall::new( naive, vector, Arc::clone(&store), profile.strategy, )) } } } #[cfg(test)] mod mcp_serve_peer_tests { //! M5c — server side of the bind transport: [`serve_peer`] + [`parse_handshake`]. //! //! These drive the **private** `serve_peer` free function over an in-memory //! `tokio::io::duplex` (no socket, no child process), the test seam the prod //! code was made generic for (`serve_peer` over any `AsyncRead+AsyncWrite`). //! The `OrchestratorService` is wired over the **same** in-memory fakes the //! infrastructure MCP tests use (`infrastructure/tests/mcp_server.rs`), so MCP //! behaviour is asserted against a real service with zero I/O. //! //! GARDE-FOU : every `await` that could block on a peer that never speaks is //! bounded by `tokio::time::timeout`; a hung accept/serve fails fast instead of //! hanging the suite. use std::collections::HashMap; use std::sync::{Arc, Mutex}; use std::time::Duration; use async_trait::async_trait; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, OrchestratorService, TerminalSessions, UpdateAgentContext, }; use domain::agent::{AgentManifest, ManifestEntry}; use domain::events::{DomainEvent, OrchestrationSource}; use domain::ids::{AgentId, ProfileId, ProjectId, SkillId}; 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::{PtySize, SessionId}; use serde_json::{json, Value}; use uuid::Uuid; use super::serve_peer; use infrastructure::McpServer; /// Test timeout for any single peer interaction. Generous but finite: a correct /// duplex round-trip is sub-millisecond, so this only ever fires on a real hang. const TIMEOUT: Duration = Duration::from_secs(5); // ----------------------------------------------------------------------- // Fakes — mirrored from `infrastructure/tests/mcp_server.rs` (the established // MCP harness), trimmed to exactly what `OrchestratorService::new` needs. // ----------------------------------------------------------------------- #[derive(Default)] struct ContextsInner { manifest: AgentManifest, contents: HashMap, } #[derive(Clone)] struct FakeContexts(Arc>); impl FakeContexts { fn new() -> Self { Self(Arc::new(Mutex::new(ContextsInner { manifest: AgentManifest { version: 1, entries: Vec::new(), }, contents: HashMap::new(), }))) } 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 { 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 { 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 { 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>); #[async_trait] impl ProfileStore for FakeProfiles { async fn list(&self) -> Result, 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 { 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, StoreError> { Ok(Vec::new()) } async fn get( &self, _scope: SkillScope, _root: &ProjectPath, _id: SkillId, ) -> Result { 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, domain::ports::MemoryError> { Ok(Vec::new()) } } struct FakeRuntime; impl AgentRuntime for FakeRuntime { fn detect(&self, _p: &AgentProfile) -> Result { Ok(true) } fn prepare_invocation( &self, profile: &AgentProfile, _ctx: &PreparedContext, cwd: &ProjectPath, _session: &SessionPlan, ) -> Result { 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, 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 { Ok(false) } async fn create_dir_all(&self, _p: &RemotePath) -> Result<(), FsError> { Ok(()) } async fn list(&self, _p: &RemotePath) -> Result, 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 { 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 { Ok(Box::new(std::iter::empty())) } fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } async fn kill(&self, _h: &PtyHandle) -> Result { 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); 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() } /// The hyphen-free hex project-id the handshake guard compares against — the /// exact form `ensure_mcp_server` derives and M5d's `--project` reuses. fn project_id_arg(p: &Project) -> String { p.id.as_uuid().simple().to_string() } /// A capturing event sink (the MCP twin of the file watcher's publish closure): /// records every [`DomainEvent`] so a test can assert `requester_id`. fn capturing_events() -> ( Arc, Arc>>, ) { let captured = Arc::new(Mutex::new(Vec::new())); let sink = captured.clone(); let publish: Arc = Arc::new(move |e: DomainEvent| sink.lock().unwrap().push(e)); (publish, captured) } /// Builds an `OrchestratorService` over the in-memory fakes (no structured /// registry), mirroring `infrastructure/tests/mcp_server.rs::build_service`. fn build_service(contexts: FakeContexts) -> Arc { // Profil Claude **complet** (adaptateur structuré + capacité MCP `.mcp.json`) : // seul profil que la garde F2 (`guard_mcp_bridge_supported`) laisse passer pour // `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, 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, Arc::new(SeqIds(Mutex::new(1))), )); Arc::new(OrchestratorService::new( create, launch, list, close, update, create_skill, Arc::clone(&profiles) as Arc, Arc::clone(&sessions), )) } // --- duplex client helpers --------------------------------------------- /// Frames a `tools/call` request line (newline-terminated, as the transport /// expects per `StdioTransport::recv`). fn tools_call_line(id: i64, tool: &str, arguments: Value) -> String { let mut s = serde_json::to_string(&json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": tool, "arguments": arguments } })) .unwrap(); s.push('\n'); s } /// A `tools/list` request line. fn tools_list_line(id: i64) -> String { let mut s = serde_json::to_string(&json!({ "jsonrpc": "2.0", "id": id, "method": "tools/list" })) .unwrap(); s.push('\n'); s } /// A handshake line `{"project":..,"requester":..}` followed by `\n`. fn handshake_line(project: &str, requester: &str) -> String { format!("{{\"project\":\"{project}\",\"requester\":\"{requester}\"}}\n") } /// Reads exactly one newline-delimited JSON-RPC response off the client side of /// the duplex, bounded by [`TIMEOUT`]. Returns `None` on EOF/timeout (the peer /// closed without replying — used by the project-guard test). async fn read_one_response(client: &mut R) -> Option where R: tokio::io::AsyncRead + Unpin, { let mut buf = Vec::new(); let mut byte = [0u8; 1]; loop { match tokio::time::timeout(TIMEOUT, client.read(&mut byte)).await { Ok(Ok(0)) => return None, // EOF before a full line Ok(Ok(_)) => { if byte[0] == b'\n' { break; } buf.push(byte[0]); } Ok(Err(_)) => return None, Err(_) => return None, // GARDE-FOU: timed out waiting for a reply } } serde_json::from_slice(&buf).ok() } /// Spawns `serve_peer` over the server half of a fresh duplex and returns the /// client half plus the join handle. The peer is bounded by the test's reads. fn spawn_peer( server: Arc, expected_project: String, ) -> (tokio::io::DuplexStream, tokio::task::JoinHandle<()>) { let (client, server_side) = tokio::io::duplex(64 * 1024); let handle = tokio::spawn(async move { serve_peer(server, &expected_project, server_side).await; }); (client, handle) } // ----------------------------------------------------------------------- // 1. Handshake + tools/list end-to-end over the duplex. // ----------------------------------------------------------------------- #[tokio::test] async fn handshake_then_tools_list_round_trips_over_duplex() { let proj = project(); let service = build_service(FakeContexts::new()); let server = Arc::new(McpServer::new(service, proj.clone())); let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); // Write the handshake line, then a tools/list request. client .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) .await .unwrap(); client .write_all(tools_list_line(1).as_bytes()) .await .unwrap(); client.flush().await.unwrap(); let resp = tokio::time::timeout(TIMEOUT, read_one_response(&mut client)) .await .expect("GARDE-FOU: tools/list timed out") .expect("a tools/list response line"); assert_eq!(resp["id"], json!(1)); let tools = resp["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:?}" ); drop(client); // EOF ⇒ serve loop ends tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer task did not finish") .unwrap(); } // ----------------------------------------------------------------------- // 2. The handshake's requester is propagated to the processed event. // ----------------------------------------------------------------------- #[tokio::test] async fn handshake_requester_propagates_to_processed_event() { let proj = project(); let contexts = FakeContexts::new(); contexts.seed_agent("architect"); let service = build_service(contexts); let (publish, captured) = capturing_events(); let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); client .write_all(handshake_line(&project_id_arg(&proj), "agent-42").as_bytes()) .await .unwrap(); client .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) .await .unwrap(); client.flush().await.unwrap(); let resp = read_one_response(&mut client) .await .expect("a tools/call response"); assert_eq!(resp["result"]["isError"], json!(false), "got {resp}"); drop(client); tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer did not finish") .unwrap(); let events = captured.lock().unwrap(); let processed: Vec<&DomainEvent> = events .iter() .filter(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })) .collect(); assert_eq!( processed.len(), 1, "exactly one processed event; got {events:?}" ); match processed[0] { DomainEvent::OrchestratorRequestProcessed { requester_id, action, source, .. } => { assert_eq!( requester_id, "agent-42", "the real handshake requester must be propagated (not 'mcp')" ); assert_eq!(action, "idea_list_agents"); assert_eq!(*source, OrchestrationSource::Mcp); } other => panic!("expected OrchestratorRequestProcessed, got {other:?}"), } } // ----------------------------------------------------------------------- // 3. Empty requester in the handshake ⇒ legacy "mcp" label (back-compat). // ----------------------------------------------------------------------- #[tokio::test] async fn empty_requester_handshake_falls_back_to_legacy_mcp_label() { let proj = project(); let contexts = FakeContexts::new(); contexts.seed_agent("architect"); let service = build_service(contexts); let (publish, captured) = capturing_events(); let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); // Empty requester ("") in the handshake. let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); client .write_all(handshake_line(&project_id_arg(&proj), "").as_bytes()) .await .unwrap(); client .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) .await .unwrap(); client.flush().await.unwrap(); let _ = read_one_response(&mut client).await.expect("a response"); drop(client); tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer did not finish") .unwrap(); let events = captured.lock().unwrap(); let processed = events .iter() .find_map(|e| match e { DomainEvent::OrchestratorRequestProcessed { requester_id, .. } => { Some(requester_id.clone()) } _ => None, }) .expect("a processed event"); assert_eq!( processed, "mcp", "empty handshake requester must keep the legacy 'mcp' label" ); } // ----------------------------------------------------------------------- // 4. JSON-RPC bytes glued onto the handshake's write are not lost // (proves `StdioTransport::from_buffered` preserves buffered bytes). // ----------------------------------------------------------------------- #[tokio::test] async fn jsonrpc_request_glued_to_handshake_is_served() { let proj = project(); let service = build_service(FakeContexts::new()); let server = Arc::new(McpServer::new(service, proj.clone())); let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); // ONE write carrying the handshake line AND the tools/list line, back to back. let mut glued = handshake_line(&project_id_arg(&proj), "agent-1"); glued.push_str(&tools_list_line(7)); client.write_all(glued.as_bytes()).await.unwrap(); client.flush().await.unwrap(); let resp = read_one_response(&mut client) .await .expect("the glued tools/list must still be served"); assert_eq!( resp["id"], json!(7), "the buffered request id must come through" ); assert!( resp["result"]["tools"].is_array(), "buffered request produced a real tools/list result; got {resp}" ); drop(client); tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer did not finish") .unwrap(); } // ----------------------------------------------------------------------- // 5a. Project guard: a mismatching handshake project ⇒ closed without serving. // ----------------------------------------------------------------------- #[tokio::test] async fn mismatched_project_handshake_is_closed_without_serving() { let proj = project(); let contexts = FakeContexts::new(); contexts.seed_agent("architect"); let service = build_service(contexts); let (publish, captured) = capturing_events(); let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); // serve_peer is told to expect this project's id... let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); // ...but the handshake claims a DIFFERENT project. client .write_all(handshake_line("ffffffffffffffffffffffffffffffff", "intruder").as_bytes()) .await .unwrap(); client .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) .await .unwrap(); client.flush().await.unwrap(); // No reply must come: the peer returned before serving. read returns EOF. let resp = read_one_response(&mut client).await; assert!( resp.is_none(), "mismatched project must be closed WITHOUT a response; got {resp:?}" ); tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer did not finish") .unwrap(); // And nothing was dispatched ⇒ no processed event. let events = captured.lock().unwrap(); assert!( !events .iter() .any(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })), "a rejected peer must not dispatch anything; got {events:?}" ); } // ----------------------------------------------------------------------- // 5b. Empty handshake project ⇒ served normally (guard does not reject). // ----------------------------------------------------------------------- #[tokio::test] async fn empty_handshake_project_is_served_normally() { let proj = project(); let service = build_service(FakeContexts::new()); let server = Arc::new(McpServer::new(service, proj.clone())); let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); // Empty project in the handshake — the guard must NOT reject it. client .write_all(handshake_line("", "agent-1").as_bytes()) .await .unwrap(); client .write_all(tools_list_line(1).as_bytes()) .await .unwrap(); client.flush().await.unwrap(); let resp = read_one_response(&mut client) .await .expect("empty-project handshake must still be served"); assert!(resp["result"]["tools"].is_array(), "got {resp}"); drop(client); tokio::time::timeout(TIMEOUT, peer) .await .expect("GARDE-FOU: peer did not finish") .unwrap(); } // ----------------------------------------------------------------------- // 6. Peer isolation: one peer closing/erroring does not stop another. // ----------------------------------------------------------------------- #[tokio::test] async fn one_peer_failure_does_not_affect_a_concurrent_peer() { let proj = project(); let service = build_service(FakeContexts::new()); let server = Arc::new(McpServer::new(service, proj.clone())); // Peer A: a broken peer that immediately closes after a partial handshake // (no newline) — its serve_peer must end without crashing the runtime. let (mut client_a, peer_a) = spawn_peer(Arc::clone(&server), project_id_arg(&proj)); client_a.write_all(b"{\"project\":").await.unwrap(); // partial, no newline client_a.flush().await.unwrap(); drop(client_a); // abrupt close mid-handshake // Peer B: a healthy peer, served concurrently, must still get its reply. let (mut client_b, peer_b) = spawn_peer(Arc::clone(&server), project_id_arg(&proj)); client_b .write_all(handshake_line(&project_id_arg(&proj), "agent-b").as_bytes()) .await .unwrap(); client_b .write_all(tools_list_line(2).as_bytes()) .await .unwrap(); client_b.flush().await.unwrap(); let resp = read_one_response(&mut client_b) .await .expect("healthy peer B must be served despite peer A failing"); assert_eq!(resp["id"], json!(2)); assert!(resp["result"]["tools"].is_array(), "got {resp}"); drop(client_b); // Both peer tasks must terminate cleanly within the bound. tokio::time::timeout(TIMEOUT, peer_a) .await .expect("GARDE-FOU: peer A did not finish") .unwrap(); tokio::time::timeout(TIMEOUT, peer_b) .await .expect("GARDE-FOU: peer B did not finish") .unwrap(); } // ----------------------------------------------------------------------- // Bonus — parse_handshake unit behaviour (tolerant parsing contract). // ----------------------------------------------------------------------- #[test] fn parse_handshake_is_tolerant() { use super::parse_handshake; // Well-formed. assert_eq!( parse_handshake("{\"project\":\"p1\",\"requester\":\"a1\"}\n"), ("p1".to_owned(), "a1".to_owned()) ); // Malformed JSON ⇒ empty strings, never a panic. assert_eq!(parse_handshake("{not json"), (String::new(), String::new())); // Missing fields ⇒ empty strings. assert_eq!(parse_handshake("{}"), (String::new(), String::new())); } } #[cfg(test)] mod tests { //! Wiring contract for [`build_memory_recall`] (Pièce 1, §14.5.5). //! //! The *behaviour* of `NaiveMemoryRecall` / `VectorMemoryRecall` / //! `AdaptiveMemoryRecall` is covered exhaustively in //! `infrastructure/tests/vector_recall.rs`. Here we only pin the **composition //! root contract**: with the default `EmbedderProfile::none()` profile (the //! dependency-free default), `build_memory_recall` must hand back a recall whose //! observable behaviour is *identical* to a bare `NaiveMemoryRecall` over the //! same store — same index ordering, same budget truncation. //! //! `build_memory_recall` is `pub(crate)`, so this lives in `state.rs` (it is not //! reachable from the `tests/` integration crate). Everything is in-memory and //! deterministic. use std::collections::HashMap; use std::sync::Mutex; use super::*; use async_trait::async_trait; use domain::markdown::MarkdownDoc; use domain::memory::{Memory, MemoryFrontmatter, MemorySlug, MemoryType}; use domain::ports::{DirEntry, FsError, MemoryQuery, RemotePath}; use domain::project::ProjectPath; // In-memory FileSystem (same minimal shape as the infrastructure test fixtures). #[derive(Default)] struct MemFs { files: Mutex>>, } #[async_trait] impl FileSystem for MemFs { async fn read(&self, path: &RemotePath) -> Result, FsError> { self.files .lock() .unwrap() .get(path.as_str()) .cloned() .ok_or_else(|| FsError::NotFound(path.as_str().to_string())) } async fn write(&self, path: &RemotePath, data: &[u8]) -> Result<(), FsError> { self.files .lock() .unwrap() .insert(path.as_str().to_string(), data.to_vec()); Ok(()) } async fn exists(&self, path: &RemotePath) -> Result { Ok(self.files.lock().unwrap().contains_key(path.as_str())) } async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> { Ok(()) } async fn list(&self, _path: &RemotePath) -> Result, FsError> { Ok(Vec::new()) } async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> { Ok(()) } } fn root() -> ProjectPath { ProjectPath::new("/proj").unwrap() } fn note(slug: &str, hook: &str) -> Memory { Memory::new( MemoryFrontmatter { name: MemorySlug::new(slug).unwrap(), description: hook.to_string(), r#type: MemoryType::Project, }, MarkdownDoc::new("# body"), ) .unwrap() } fn query(text: &str, budget: usize) -> MemoryQuery { MemoryQuery { text: text.to_string(), token_budget: budget, } } async fn seed_store() -> (Arc, Arc) { let fs: Arc = Arc::new(MemFs::default()); let store_concrete = Arc::new(FsMemoryStore::new(Arc::clone(&fs))); for n in [ note("alpha", "apple orange grape"), note("beta", "kiwi mango papaya"), note("gamma", "carrot potato onion"), ] { store_concrete.save(&root(), &n).await.unwrap(); } let store: Arc = store_concrete; (fs, store) } /// With the default `none` profile, `build_memory_recall` is observationally a /// plain `NaiveMemoryRecall`: same index ordering and same budget truncation, /// entry-for-entry, across a representative spread of budgets. #[tokio::test] async fn build_memory_recall_none_profile_matches_naive_recall() { let (fs, store) = seed_store().await; let wired = build_memory_recall( Arc::clone(&fs), Arc::clone(&store), &EmbedderProfile::none(), std::path::Path::new("/unused-onnx-cache"), ); let naive = NaiveMemoryRecall::new(Arc::clone(&store)); // 0 ⇒ empty; mid budgets ⇒ partial truncation; huge ⇒ full index order. for budget in [0usize, 3, 6, 100, 100_000] { let q = query("kiwi mango papaya", budget); let expected = naive.recall(&root(), &q).await.unwrap(); let got = wired.recall(&root(), &q).await.unwrap(); assert_eq!( got, expected, "none profile must equal bare NaiveMemoryRecall at budget {budget}" ); } } } // =========================================================================== // M5e — Smoke end-to-end over the REAL loopback (no real CLI, no network). // =========================================================================== #[cfg(test)] mod mcp_e2e_loopback_tests { //! M5e — **smoke end-to-end** of the bind transport over a **real** //! `interprocess` loopback (cadrage v5 §6 row M5e, §2 contract). //! //! ## Chosen e2e level — the **real accept loop**, justified //! //! Unlike M5c (which drove the *private* `serve_peer` over an in-memory //! `tokio::io::duplex`), M5e proves the chain across an **actual** local socket. //! Two options were on the table (cadrage M5e): //! //! 1. the **full `McpServerHandle::start` accept loop** on a real //! `mcp_endpoint`, dialed by a real `interprocess` client, or //! 2. a single `serve_peer` over one accepted real socket. //! //! We pick **(1) the real accept loop**. It is the *most faithful* slice — it //! exercises exactly the production path a launched CLI's `idea mcp-server` //! bridge takes: `bind_endpoint` → `McpServerHandle::start` → `listener.accept()` //! → `serve_peer` (handshake + project guard) → `StdioTransport` → //! `McpServer::serve_as` → the **real `dispatch`** (over the same in-memory fakes //! as M2/M5c). And it is **safely bounded**: the accept loop is already async and //! parks on the absence of a peer, while the *client* side is a plain //! request/response exchange we wrap in `tokio::time::timeout`. Option (2) would //! skip the bind/accept/lifecycle wiring for no extra safety, since the risk //! (a peer that never speaks) is bounded identically by the client-side timeout. //! //! No production test seam was added: `McpServerHandle::start`, `bind_endpoint` //! and `mcp_endpoint` are all reachable from this in-crate module as-is. The //! client uses the same `interprocess` `Stream` the real bridge uses. //! //! GARDE-FOU : the listener bind, every client connect, and every response read //! is wrapped in `tokio::time::timeout`; a hung accept/serve/handshake fails the //! test fast instead of hanging the suite. use std::collections::HashMap; use std::sync::{Arc, Mutex}; use std::time::Duration; use async_trait::async_trait; use interprocess::local_socket::tokio::Stream as LocalSocketStream; use interprocess::local_socket::traits::tokio::Stream as _; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use uuid::Uuid; use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, OrchestratorService, TerminalSessions, UpdateAgentContext, }; use domain::agent::{AgentManifest, ManifestEntry}; use domain::events::{DomainEvent, OrchestrationSource}; use domain::ids::{AgentId, NodeId, ProfileId, ProjectId, SkillId}; 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 serde_json::{json, Value}; use super::{bind_endpoint, mcp_endpoint, McpServerHandle}; use crate::mcp_endpoint::McpEndpoint; use infrastructure::{ InMemoryConversationRegistry, InMemoryMailbox, McpServer, MediatedInbox, SystemMillisClock, }; /// Test timeout for any single loopback interaction. Generous but finite: a /// correct round-trip is sub-millisecond, so this only ever fires on a real hang. const TIMEOUT: Duration = Duration::from_secs(5); // ----------------------------------------------------------------------- // Fakes — mirrored from `infrastructure/tests/mcp_server.rs` (the established // MCP harness). Includes a `FakeSession` so `idea_ask_agent` resolves inline. // ----------------------------------------------------------------------- #[derive(Default)] struct ContextsInner { manifest: AgentManifest, contents: HashMap, } #[derive(Clone)] struct FakeContexts(Arc>); impl FakeContexts { fn new() -> Self { Self(Arc::new(Mutex::new(ContextsInner { manifest: AgentManifest { version: 1, entries: Vec::new(), }, contents: HashMap::new(), }))) } 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 { 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 { 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 { 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>); #[async_trait] impl ProfileStore for FakeProfiles { async fn list(&self) -> Result, 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 { 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, StoreError> { Ok(Vec::new()) } async fn get( &self, _scope: SkillScope, _root: &ProjectPath, _id: SkillId, ) -> Result { 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, domain::ports::MemoryError> { Ok(Vec::new()) } } struct FakeRuntime; impl AgentRuntime for FakeRuntime { fn detect(&self, _p: &AgentProfile) -> Result { Ok(true) } fn prepare_invocation( &self, profile: &AgentProfile, _ctx: &PreparedContext, cwd: &ProjectPath, _session: &SessionPlan, ) -> Result { 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, 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 { Ok(false) } async fn create_dir_all(&self, _p: &RemotePath) -> Result<(), FsError> { Ok(()) } async fn list(&self, _p: &RemotePath) -> Result, 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 { 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 { Ok(Box::new(std::iter::empty())) } fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { Ok(Vec::new()) } async fn kill(&self, _h: &PtyHandle) -> Result { 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); 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 { // A fresh project id per call ⇒ a distinct endpoint per test (no socket-file // collision when tests run in parallel). Project::new( ProjectId::from_uuid(Uuid::new_v4()), "demo", ProjectPath::new("/home/me/proj").unwrap(), RemoteRef::local(), 1_700_000_000_000, ) .unwrap() } fn project_id_arg(p: &Project) -> String { p.id.as_uuid().simple().to_string() } fn capturing_events() -> ( Arc, Arc>>, ) { let captured = Arc::new(Mutex::new(Vec::new())); let sink = captured.clone(); let publish: Arc = Arc::new(move |e: DomainEvent| sink.lock().unwrap().push(e)); (publish, captured) } /// Builds an `OrchestratorService` over the in-memory fakes, returning the /// shared `TerminalSessions` (so a test can pre-bind a live PTY target for an /// `idea_ask_agent`) and the `InMemoryMailbox` (so a test can observe pending /// tickets). Mirrors the established harness; no use case is re-invented. fn build_service( contexts: FakeContexts, ) -> ( Arc, Arc, Arc, ) { // Profil Claude **complet** (adaptateur structuré + capacité MCP `.mcp.json`) : // seul profil que la garde F2 (`guard_mcp_bridge_supported`) laisse passer pour // `idea_ask_agent` — le round-trip e2e testé ici suppose une cible éligible. 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 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, 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, Arc::new(SeqIds(Mutex::new(1))), )); let input = Arc::new(MediatedInbox::with_pty( Arc::clone(&mailbox), Arc::new(SystemMillisClock), Arc::new(FakePty) as Arc, )) as Arc; let conversations = Arc::new(InMemoryConversationRegistry::new()) as Arc; let service = OrchestratorService::new( create, launch, list, close, update, create_skill, Arc::clone(&profiles) as Arc, Arc::clone(&sessions), ) .with_input_mediator( input, Arc::clone(&mailbox) as Arc, ) .with_conversations(conversations); (Arc::new(service), sessions, mailbox) } /// Codex twin of [`build_service`]: identical wiring, but the single profile /// targets the **Codex** structured adapter with a `TomlConfigHome` MCP strategy /// (`$CODEX_HOME/config.toml`). This is the couple the bridge guard /// (`materializes_idea_bridge`) must now let through — proving Codex is eligible /// for `idea_ask_agent`, exactly like Claude. No real `codex` binary is ever /// spawned: the runtime/PTY are the same in-memory fakes. fn build_service_codex( contexts: FakeContexts, ) -> ( Arc, Arc, Arc, ) { let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new( ProfileId::from_uuid(Uuid::from_u128(9)), "Codex CLI", "codex", Vec::new(), ContextInjection::stdin(), None, "{agentRunDir}", None, ) .unwrap() .with_structured_adapter(StructuredAdapter::Codex) .with_mcp(McpCapability::new( McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME").unwrap(), McpTransport::Stdio, ))]))); let sessions = Arc::new(TerminalSessions::new()); 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, 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, Arc::new(SeqIds(Mutex::new(1))), )); let input = Arc::new(MediatedInbox::with_pty( Arc::clone(&mailbox), Arc::new(SystemMillisClock), Arc::new(FakePty) as Arc, )) as Arc; let conversations = Arc::new(InMemoryConversationRegistry::new()) as Arc; let service = OrchestratorService::new( create, launch, list, close, update, create_skill, Arc::clone(&profiles) as Arc, Arc::clone(&sessions), ) .with_input_mediator( input, Arc::clone(&mailbox) as Arc, ) .with_conversations(conversations); (Arc::new(service), sessions, mailbox) } /// 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 }, ), ); } // --- real loopback harness --------------------------------------------- /// Stands up the **real** production accept-and-serve path on a real /// `interprocess` loopback for `project`: binds the endpoint via the production /// `bind_endpoint`, starts `McpServerHandle::start` (the M5c accept loop), and /// returns the live handle plus the endpoint to dial. /// /// Asserts the bind actually succeeded (a `None` listener would make the e2e /// vacuous) — if the per-user runtime dir is unwritable the test fails loudly /// rather than silently testing nothing. fn start_real_server( service: Arc, project: &Project, events: Option>, ) -> (McpServerHandle, McpEndpoint) { let endpoint = mcp_endpoint(&project.id); let listener = bind_endpoint(&endpoint); assert!( listener.is_some(), "M5e needs a real bound listener; bind_endpoint returned None for {:?}", endpoint.as_cli_arg() ); let mut server = McpServer::new(service, project.clone()); if let Some(publish) = events { server = server.with_events(publish); } let handle = McpServerHandle::start(server, endpoint.clone(), listener, project_id_arg(project)); (handle, endpoint) } /// Connects a **real** `interprocess` client to `endpoint`, bounded by [`TIMEOUT`] /// so an endpoint that never accepts fails fast (GARDE-FOU). async fn connect_client(endpoint: &McpEndpoint) -> LocalSocketStream { use interprocess::local_socket::{GenericFilePath, ToFsName as _}; let name = endpoint .as_cli_arg() .to_fs_name::() .expect("valid endpoint name"); tokio::time::timeout(TIMEOUT, LocalSocketStream::connect(name)) .await .expect("GARDE-FOU: connect to endpoint timed out") .expect("connect to a live endpoint") } fn handshake_line(project: &str, requester: &str) -> String { format!("{{\"project\":\"{project}\",\"requester\":\"{requester}\"}}\n") } fn tools_call_line(id: i64, tool: &str, arguments: Value) -> String { let mut s = serde_json::to_string(&json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": tool, "arguments": arguments } })) .unwrap(); s.push('\n'); s } /// Reads exactly one newline-delimited JSON-RPC response off the client side of /// the real socket, bounded by [`TIMEOUT`]. `None` on EOF/timeout. async fn read_one_response(reader: &mut R) -> Option where R: AsyncBufReadExt + Unpin, { let mut line = String::new(); match tokio::time::timeout(TIMEOUT, reader.read_line(&mut line)).await { Ok(Ok(0)) => None, // EOF before a full line Ok(Ok(_)) => serde_json::from_str(line.trim_end()).ok(), Ok(Err(_)) => None, Err(_) => None, // GARDE-FOU: timed out waiting for a reply } } // ----------------------------------------------------------------------- // 1. idea_list_agents e2e over the real loopback ⇒ JSON array of agents. // ----------------------------------------------------------------------- #[tokio::test] async fn list_agents_round_trips_over_real_loopback() { let contexts = FakeContexts::new(); contexts.seed_agent("architect"); contexts.seed_agent("dev-backend"); let (service, _sessions, _mailbox) = build_service(contexts); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); let conn = connect_client(&endpoint).await; let (read_half, mut write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); // Handshake (real project id) + a tools/call idea_list_agents. write_half .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) .await .unwrap(); write_half .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) .await .unwrap(); write_half.flush().await.unwrap(); let resp = read_one_response(&mut reader) .await .expect("a list_agents response line"); assert_eq!(resp["id"], json!(1)); assert!(resp["error"].is_null(), "transport error: {resp}"); let result = &resp["result"]; assert_eq!(result["isError"], json!(false), "got {result}"); // The inline text is the JSON array of the project's agents. let text = result["content"][0]["text"].as_str().expect("text block"); 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"), "got {names:?}"); assert!(names.contains(&"dev-backend"), "got {names:?}"); drop(write_half); // EOF ⇒ serve loop ends cleanly handle.stop(); } // ----------------------------------------------------------------------- // 2. idea_ask_agent → idea_reply e2e over the real loopback (Option 1, B-3/B-4): // the asker's `idea_ask_agent` blocks on the mailbox; the target's `idea_reply` // (on a second connection, with its agent id as the handshake requester) // resolves it and the asker gets the result INLINE. // ----------------------------------------------------------------------- #[tokio::test] async fn ask_then_reply_round_trips_inline_over_real_loopback() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("architect"); let (service, sessions, mailbox) = build_service(contexts); // Target already live in the PTY registry, so the ask reuses its terminal. seed_live_pty( &sessions, agent_id, SessionId::from_uuid(Uuid::from_u128(4242)), ); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); // Connection A: the asker. let conn_a = connect_client(&endpoint).await; let (read_a, mut write_a) = tokio::io::split(conn_a); let mut reader_a = BufReader::new(read_a); write_a .write_all(handshake_line(&project_id_arg(&proj), "agent-asker").as_bytes()) .await .unwrap(); write_a .write_all( tools_call_line( 7, "idea_ask_agent", json!({ "target": "architect", "task": "What is the answer?" }), ) .as_bytes(), ) .await .unwrap(); write_a.flush().await.unwrap(); // The ask is now blocked awaiting the reply — observe the pending ticket. tokio::time::timeout(TIMEOUT, async { while mailbox.pending(&agent_id) == 0 { tokio::task::yield_now().await; } }) .await .expect("ask must enqueue a ticket"); // Connection B: the target rendering its result. Its handshake requester is // the target agent's id, which the server injects as the Reply `from`. let conn_b = connect_client(&endpoint).await; let (read_b, mut write_b) = tokio::io::split(conn_b); let mut reader_b = BufReader::new(read_b); write_b .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) .await .unwrap(); write_b .write_all( tools_call_line(8, "idea_reply", json!({ "result": "the answer is 42" })) .as_bytes(), ) .await .unwrap(); write_b.flush().await.unwrap(); let reply_resp = read_one_response(&mut reader_b) .await .expect("a reply ack line"); assert_eq!( reply_resp["result"]["isError"], json!(false), "got {reply_resp}" ); // The asker now receives its inline result. let resp = read_one_response(&mut reader_a) .await .expect("an ask response line"); assert_eq!(resp["id"], json!(7)); assert!(resp["error"].is_null(), "transport error: {resp}"); let result = &resp["result"]; assert_eq!(result["isError"], json!(false), "got {result}"); assert_eq!( result["content"][0]["text"].as_str().expect("text block"), "the answer is 42", "ask reply must be returned inline over the real loopback; got {result}" ); drop(write_a); drop(write_b); handle.stop(); } // ----------------------------------------------------------------------- // 2b. Codex twin of the round-trip above: the *only* difference is the target // profile (Codex + TomlConfigHome MCP). It proves the bridge guard now lets // a Codex target through AND that the inline ask→reply loop still completes. // ----------------------------------------------------------------------- #[tokio::test] async fn ask_then_reply_round_trips_inline_over_real_loopback_codex() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("architect"); let (service, sessions, mailbox) = build_service_codex(contexts); // Target already live in the PTY registry, so the ask reuses its terminal. seed_live_pty( &sessions, agent_id, SessionId::from_uuid(Uuid::from_u128(4242)), ); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); // Connection A: the asker. let conn_a = connect_client(&endpoint).await; let (read_a, mut write_a) = tokio::io::split(conn_a); let mut reader_a = BufReader::new(read_a); write_a .write_all(handshake_line(&project_id_arg(&proj), "agent-asker").as_bytes()) .await .unwrap(); write_a .write_all( tools_call_line( 7, "idea_ask_agent", json!({ "target": "architect", "task": "What is the answer?" }), ) .as_bytes(), ) .await .unwrap(); write_a.flush().await.unwrap(); // The ask is now blocked awaiting the reply — observe the pending ticket. tokio::time::timeout(TIMEOUT, async { while mailbox.pending(&agent_id) == 0 { tokio::task::yield_now().await; } }) .await .expect("ask must enqueue a ticket"); // Connection B: the target rendering its result. Its handshake requester is // the target agent's id, which the server injects as the Reply `from`. let conn_b = connect_client(&endpoint).await; let (read_b, mut write_b) = tokio::io::split(conn_b); let mut reader_b = BufReader::new(read_b); write_b .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) .await .unwrap(); write_b .write_all( tools_call_line(8, "idea_reply", json!({ "result": "the answer is 42" })) .as_bytes(), ) .await .unwrap(); write_b.flush().await.unwrap(); let reply_resp = read_one_response(&mut reader_b) .await .expect("a reply ack line"); assert_eq!( reply_resp["result"]["isError"], json!(false), "got {reply_resp}" ); // The asker now receives its inline result. let resp = read_one_response(&mut reader_a) .await .expect("an ask response line"); assert_eq!(resp["id"], json!(7)); assert!(resp["error"].is_null(), "transport error: {resp}"); let result = &resp["result"]; assert_eq!(result["isError"], json!(false), "got {result}"); assert_eq!( result["content"][0]["text"].as_str().expect("text block"), "the answer is 42", "ask reply must be returned inline over the real loopback (Codex target); got {result}" ); drop(write_a); drop(write_b); handle.stop(); } // ----------------------------------------------------------------------- // 3. idea_reply with no in-flight ask ⇒ typed tool error (isError:true), no // panic, connection stays healthy for a follow-up call. // ----------------------------------------------------------------------- #[tokio::test] async fn orphan_reply_is_typed_error_over_real_loopback() { let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("dev"); let (service, _sessions, _mailbox) = build_service(contexts); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); let conn = connect_client(&endpoint).await; let (read_half, mut write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); // The peer identifies as `agent_id`; an idea_reply with no matching ask in // flight ⇒ the IdeA command fails ⇒ tool result isError:true (not transport). write_half .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) .await .unwrap(); write_half .write_all(tools_call_line(3, "idea_reply", json!({ "result": "orphan" })).as_bytes()) .await .unwrap(); write_half.flush().await.unwrap(); let resp = read_one_response(&mut reader) .await .expect("a reply response line"); assert_eq!(resp["id"], json!(3)); assert!( resp["error"].is_null(), "an orphan reply must be a tool error, not a transport error: {resp}" ); let result = &resp["result"]; assert_eq!(result["isError"], json!(true), "got {result}"); assert!( !result["content"][0]["text"] .as_str() .unwrap_or_default() .is_empty(), "error text should describe the failure; got {result}" ); // The connection is still healthy: a follow-up tools/list still answers. write_half .write_all( serde_json::to_string(&json!({ "jsonrpc": "2.0", "id": 4, "method": "tools/list" })) .map(|mut s| { s.push('\n'); s }) .unwrap() .as_bytes(), ) .await .unwrap(); write_half.flush().await.unwrap(); let again = read_one_response(&mut reader) .await .expect("server still serves after a tool error"); assert_eq!(again["id"], json!(4)); assert!(again["result"]["tools"].is_array(), "got {again}"); drop(write_half); handle.stop(); } // ----------------------------------------------------------------------- // 4. Malformed JSON-RPC after the handshake ⇒ JSON-RPC error, never a panic, // the server stays alive for a subsequent valid call. // ----------------------------------------------------------------------- #[tokio::test] async fn malformed_jsonrpc_after_handshake_errors_and_server_survives() { let contexts = FakeContexts::new(); contexts.seed_agent("architect"); let (service, _sessions, _mailbox) = build_service(contexts); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, None); let conn = connect_client(&endpoint).await; let (read_half, mut write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); write_half .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) .await .unwrap(); // A malformed JSON-RPC line (not valid JSON) — must NOT crash the serve loop. write_half.write_all(b"{ this is not json\n").await.unwrap(); write_half.flush().await.unwrap(); let resp = read_one_response(&mut reader) .await .expect("a parse error still owes a response"); let error = &resp["error"]; assert!( !error.is_null(), "malformed input must yield a JSON-RPC error: {resp}" ); // JSON-RPC mandates a null id when the request could not be correlated. assert_eq!(resp["id"], Value::Null, "got {resp}"); assert!( resp["result"].is_null(), "no result on parse error; got {resp}" ); // The server survived: a subsequent valid call still answers. write_half .write_all( serde_json::to_string(&json!({ "jsonrpc": "2.0", "id": 2, "method": "tools/list" })) .map(|mut s| { s.push('\n'); s }) .unwrap() .as_bytes(), ) .await .unwrap(); write_half.flush().await.unwrap(); let again = read_one_response(&mut reader) .await .expect("server must survive malformed input"); assert_eq!(again["id"], json!(2)); assert!(again["result"]["tools"].is_array(), "got {again}"); drop(write_half); handle.stop(); } // ----------------------------------------------------------------------- // 5. Bonus — the handshake requester crosses the REAL loopback and tags the // OrchestratorRequestProcessed event with the real agent id (not "mcp"). // ----------------------------------------------------------------------- #[tokio::test] async fn handshake_requester_propagates_over_real_loopback() { let contexts = FakeContexts::new(); let (service, _sessions, _mailbox) = build_service(contexts); let (publish, captured) = capturing_events(); let proj = project(); let (handle, endpoint) = start_real_server(service, &proj, Some(publish)); let conn = connect_client(&endpoint).await; let (read_half, mut write_half) = tokio::io::split(conn); let mut reader = BufReader::new(read_half); // Handshake carries the real requesting agent id. write_half .write_all(handshake_line(&project_id_arg(&proj), "agent-42").as_bytes()) .await .unwrap(); // A successful launch (creates + launches an agent from scratch) ⇒ a // processed beacon is emitted, tagged with the handshake requester. write_half .write_all( tools_call_line( 1, "idea_launch_agent", json!({ "target": "dev-backend", "profile": "claude-code" }), ) .as_bytes(), ) .await .unwrap(); write_half.flush().await.unwrap(); let resp = read_one_response(&mut reader) .await .expect("a launch response line"); assert!(resp["error"].is_null(), "transport error: {resp}"); assert_eq!(resp["result"]["isError"], json!(false), "got {resp}"); // Drain the connection so the serve task has surely published the event. drop(write_half); // Give the spawned serve task a beat to flush its publish (bounded). let _ = tokio::time::timeout(TIMEOUT, async { loop { if captured .lock() .unwrap() .iter() .any(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })) { break; } tokio::task::yield_now().await; } }) .await; 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"); assert!(*ok, "the launch succeeded"); assert_eq!( requester_id, "agent-42", "the real handshake requester must cross the loopback (not 'mcp')" ); } other => panic!("expected OrchestratorRequestProcessed, got {other:?}"), } drop(events); handle.stop(); } } #[cfg(test)] mod bind_endpoint_d1_tests { //! D1 — non-regression lock on [`bind_endpoint`]'s corpse-socket reclaim //! (cadrage §7 row D1, §5.2 "verrouille `reclaim_name(true)`"). //! //! A crashed run (SIGKILL) leaves the Unix socket **file** behind: a plain //! `bind` would then fail with `EADDRINUSE`. The production path passes //! `reclaim_name(true)`, which **unlinks the corpse** before binding. These tests //! pin that behaviour so a future refactor cannot silently drop the flag and //! resurrect the "address already in use" failure on restart. //! //! Unix-only: on Windows the endpoint is a named pipe with no filesystem corpse to //! reclaim, so there is nothing to assert. #![cfg(unix)] use super::{bind_endpoint, mcp_endpoint}; use domain::ProjectId; use uuid::Uuid; /// Rebinding after a **corpse** socket (file left in place, as after a SIGKILL) /// succeeds — no `EADDRINUSE` — because `reclaim_name(true)` unlinks it first. /// /// The corpse is reproduced **faithfully**: a `std::os::unix::net::UnixListener` /// binds the path then is dropped — std does **not** unlink on drop, so the socket /// inode is left on the filesystem with **no live listener**, exactly the state a /// SIGKILL'd run leaves behind (the fd is gone, the inode lingers). A plain `bind` /// over that inode would return `EADDRINUSE`; `bind_endpoint`'s `reclaim_name(true)` /// must unlink it first. #[tokio::test] async fn rebind_after_corpse_socket_succeeds() { use std::os::unix::net::UnixListener; let ep = mcp_endpoint(&ProjectId::from_uuid(Uuid::from_u128(0xD1_0001))); let path = ep .socket_path() .expect("unix endpoint exposes a socket path"); // Clean any leftover from a previous run of this very test. let _ = std::fs::remove_file(&path); if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); } // 1) Lay down a CORPSE: bind with std (which leaves the inode on drop) and // drop it ⇒ socket file remains with no live listener (SIGKILL aftermath). { let corpse = UnixListener::bind(&path).expect("lay corpse socket"); drop(corpse); } assert!( path.exists(), "corpse socket inode remains (no live listener)" ); // 2) Rebind over the corpse: must succeed (reclaim unlinks then binds), NOT // fail with EADDRINUSE. let first = bind_endpoint(&ep); assert!( first.is_some(), "rebind over the corpse socket must succeed (reclaim_name unlinks it)" ); assert!(path.exists(), "a fresh live socket now sits at the path"); // 3) Idempotent over a live socket: a second bind also reclaims + succeeds. let second = bind_endpoint(&ep); assert!(second.is_some(), "bind is idempotent across a live socket"); // 4) No file leak after a clean close: dropping the live listener(s) unlinks // the socket (interprocess reclaim guard on drop). drop(first); drop(second); assert!( !path.exists(), "socket file unlinked on clean close (no leak)" ); } }