IdeA/crates/application/tests/orchestrator_service.rs

//! Integration tests for [`OrchestratorService`] (ARCHITECTURE §14.3).
//!
//! The service is wired over the *real* agent/terminal use cases, themselves
//! backed by in-memory fakes (the same fake patterns as `agent_lifecycle.rs`).
//! This proves the dispatch contract end-to-end without real I/O:
//!
//! - `spawn_agent` on an **unknown** agent → create + launch (manifest grows, PTY
//!   spawns, `AgentLaunched` published),
//! - `spawn_agent` on a **known** agent → launch only (no second manifest entry),
//! - `stop_agent` → the agent's live session is killed and de-registered,
//! - `update_agent_context` → the agent `.md` is overwritten,
//! - unknown profile / unknown agent → `NotFound`, no spawn.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use domain::agent::{Agent, AgentManifest, AgentOrigin, ManifestEntry};
use domain::events::DomainEvent;
use domain::ids::SkillId;
use domain::ids::{AgentId, NodeId, ProfileId, ProjectId};
use domain::markdown::MarkdownDoc;
use domain::ports::{
    AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, AgentSessionFactory,
    ContextInjectionPlan, DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError,
    IdGenerator, OutputStream, PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort,
    RemotePath, ReplyEvent, ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec,
    StoreError,
};
use domain::profile::{AgentProfile, ContextInjection, StructuredAdapter};
use domain::project::{Project, ProjectPath};
use domain::remote::RemoteRef;
use domain::skill::{Skill, SkillScope};
use domain::terminal::{SessionKind, TerminalSession};
use domain::{OrchestratorCommand, OrchestratorRequest, PtySize, SessionId};
use uuid::Uuid;

use application::{
    CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
    OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext,
};

// ---------------------------------------------------------------------------
// Fakes (mirror agent_lifecycle.rs)
// ---------------------------------------------------------------------------

#[derive(Default)]
struct ContextsInner {
    manifest: AgentManifest,
    contents: HashMap<String, String>,
}

#[derive(Clone)]
struct FakeContexts(Arc<Mutex<ContextsInner>>);

impl FakeContexts {
    fn new() -> Self {
        Self(Arc::new(Mutex::new(ContextsInner {
            manifest: AgentManifest {
                version: 1,
                entries: Vec::new(),
            },
            contents: HashMap::new(),
        })))
    }
    fn with_agent(agent: &Agent, content: &str) -> Self {
        let me = Self::new();
        {
            let mut inner = me.0.lock().unwrap();
            inner
                .manifest
                .entries
                .push(ManifestEntry::from_agent(agent));
            inner
                .contents
                .insert(agent.context_path.clone(), content.to_owned());
        }
        me
    }
    fn manifest(&self) -> AgentManifest {
        self.0.lock().unwrap().manifest.clone()
    }
    fn content(&self, md_path: &str) -> Option<String> {
        self.0.lock().unwrap().contents.get(md_path).cloned()
    }
    fn md_path_of(&self, agent: &AgentId) -> Option<String> {
        self.0
            .lock()
            .unwrap()
            .manifest
            .entries
            .iter()
            .find(|e| &e.agent_id == agent)
            .map(|e| e.md_path.clone())
    }
}

#[async_trait]
impl AgentContextStore for FakeContexts {
    async fn read_context(
        &self,
        _project: &Project,
        agent: &AgentId,
    ) -> Result<MarkdownDoc, StoreError> {
        let md_path = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
        self.content(&md_path)
            .map(MarkdownDoc::new)
            .ok_or(StoreError::NotFound)
    }
    async fn write_context(
        &self,
        _project: &Project,
        agent: &AgentId,
        md: &MarkdownDoc,
    ) -> Result<(), StoreError> {
        let md_path = self.md_path_of(agent).ok_or(StoreError::NotFound)?;
        self.0
            .lock()
            .unwrap()
            .contents
            .insert(md_path, md.as_str().to_owned());
        Ok(())
    }
    async fn load_manifest(&self, _project: &Project) -> Result<AgentManifest, StoreError> {
        Ok(self.manifest())
    }
    async fn save_manifest(
        &self,
        _project: &Project,
        manifest: &AgentManifest,
    ) -> Result<(), StoreError> {
        self.0.lock().unwrap().manifest = manifest.clone();
        Ok(())
    }
}

#[derive(Clone)]
struct FakeProfiles(Arc<Vec<AgentProfile>>);
impl FakeProfiles {
    fn new(profiles: Vec<AgentProfile>) -> Self {
        Self(Arc::new(profiles))
    }
}
#[async_trait]
impl ProfileStore for FakeProfiles {
    async fn list(&self) -> Result<Vec<AgentProfile>, StoreError> {
        Ok((*self.0).clone())
    }
    async fn save(&self, _profile: &AgentProfile) -> Result<(), StoreError> {
        Ok(())
    }
    async fn delete(&self, _id: ProfileId) -> Result<(), StoreError> {
        Ok(())
    }
    async fn is_configured(&self) -> Result<bool, StoreError> {
        Ok(true)
    }
    async fn mark_configured(&self) -> Result<(), StoreError> {
        Ok(())
    }
}

// Empty skill store: the orchestrator tests spawn agents with no assigned skills.
#[derive(Default)]
struct FakeSkills;
#[async_trait]
impl SkillStore for FakeSkills {
    async fn list(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
    ) -> Result<Vec<Skill>, StoreError> {
        Ok(Vec::new())
    }
    async fn get(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
        _id: SkillId,
    ) -> Result<Skill, StoreError> {
        Err(StoreError::NotFound)
    }
    async fn save(&self, _skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> {
        Ok(())
    }
    async fn delete(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
        _id: SkillId,
    ) -> Result<(), StoreError> {
        Ok(())
    }
}

/// An empty [`MemoryRecall`]: no project memory ⇒ no memory section injected,
/// leaving the launch behaviour unchanged for the service-level tests.
#[derive(Default)]
struct FakeRecall;
#[async_trait]
impl domain::ports::MemoryRecall for FakeRecall {
    async fn recall(
        &self,
        _root: &ProjectPath,
        _query: &domain::ports::MemoryQuery,
    ) -> Result<Vec<domain::MemoryIndexEntry>, domain::ports::MemoryError> {
        Ok(Vec::new())
    }
}

/// A [`SkillStore`] that records every `save` so a `create_skill` dispatch can be
/// asserted against the persisted skill (name + scope).
#[derive(Clone, Default)]
struct RecordingSkills(Arc<Mutex<Vec<Skill>>>);

#[async_trait]
impl SkillStore for RecordingSkills {
    async fn list(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
    ) -> Result<Vec<Skill>, StoreError> {
        Ok(self.0.lock().unwrap().clone())
    }
    async fn get(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
        id: SkillId,
    ) -> Result<Skill, StoreError> {
        self.0
            .lock()
            .unwrap()
            .iter()
            .find(|s| s.id == id)
            .cloned()
            .ok_or(StoreError::NotFound)
    }
    async fn save(&self, skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> {
        self.0.lock().unwrap().push(skill.clone());
        Ok(())
    }
    async fn delete(
        &self,
        _scope: SkillScope,
        _root: &ProjectPath,
        _id: SkillId,
    ) -> Result<(), StoreError> {
        Ok(())
    }
}

struct FakeRuntime;
impl AgentRuntime for FakeRuntime {
    fn detect(&self, _profile: &AgentProfile) -> Result<bool, RuntimeError> {
        Ok(true)
    }
    fn prepare_invocation(
        &self,
        profile: &AgentProfile,
        _ctx: &PreparedContext,
        cwd: &ProjectPath,
        _session: &SessionPlan,
    ) -> Result<SpawnSpec, RuntimeError> {
        Ok(SpawnSpec {
            command: profile.command.clone(),
            args: profile.args.clone(),
            cwd: cwd.clone(),
            env: Vec::new(),
            context_plan: Some(ContextInjectionPlan::Stdin),
        })
    }
}

#[derive(Clone, Default)]
struct FakeFs;
#[async_trait]
impl FileSystem for FakeFs {
    async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
        Err(FsError::NotFound(path.as_str().to_owned()))
    }
    async fn write(&self, _path: &RemotePath, _data: &[u8]) -> Result<(), FsError> {
        Ok(())
    }
    async fn exists(&self, _path: &RemotePath) -> Result<bool, FsError> {
        Ok(false)
    }
    async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
        Ok(())
    }
    async fn list(&self, _path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
        Ok(Vec::new())
    }
    async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
        Ok(())
    }
}

#[derive(Clone)]
struct FakePty {
    next_id: SessionId,
    kills: Arc<Mutex<Vec<SessionId>>>,
    spawns: Arc<Mutex<Vec<SessionId>>>,
}
impl FakePty {
    fn new(next_id: SessionId) -> Self {
        Self {
            next_id,
            kills: Arc::new(Mutex::new(Vec::new())),
            spawns: Arc::new(Mutex::new(Vec::new())),
        }
    }
    fn spawns(&self) -> Vec<SessionId> {
        self.spawns.lock().unwrap().clone()
    }
    fn kills(&self) -> Vec<SessionId> {
        self.kills.lock().unwrap().clone()
    }
}
#[async_trait]
impl PtyPort for FakePty {
    async fn spawn(&self, _spec: SpawnSpec, _size: PtySize) -> Result<PtyHandle, PtyError> {
        self.spawns.lock().unwrap().push(self.next_id);
        Ok(PtyHandle {
            session_id: self.next_id,
        })
    }
    fn write(&self, _handle: &PtyHandle, _data: &[u8]) -> Result<(), PtyError> {
        Ok(())
    }
    fn resize(&self, _handle: &PtyHandle, _size: PtySize) -> Result<(), PtyError> {
        Ok(())
    }
    fn subscribe_output(&self, _handle: &PtyHandle) -> Result<OutputStream, PtyError> {
        Ok(Box::new(std::iter::empty()))
    }
    fn scrollback(&self, _handle: &PtyHandle) -> Result<Vec<u8>, PtyError> {
        Ok(Vec::new())
    }
    async fn kill(&self, handle: &PtyHandle) -> Result<ExitStatus, PtyError> {
        self.kills.lock().unwrap().push(handle.session_id);
        Ok(ExitStatus { code: Some(0) })
    }
}

#[derive(Default, Clone)]
struct SpyBus(Arc<Mutex<Vec<DomainEvent>>>);
impl SpyBus {
    fn events(&self) -> Vec<DomainEvent> {
        self.0.lock().unwrap().clone()
    }
}
impl EventBus for SpyBus {
    fn publish(&self, event: DomainEvent) {
        self.0.lock().unwrap().push(event);
    }
    fn subscribe(&self) -> EventStream {
        Box::new(std::iter::empty())
    }
}

// ---------------------------------------------------------------------------
// D6 fakes: structured session + factory (inter-agent messaging)
// ---------------------------------------------------------------------------

/// What the scripted session does on its next `send`.
enum SendScript {
    /// Emit this event stream verbatim (drained to its `Final`).
    Stream(Vec<ReplyEvent>),
    /// Sleep `delay` *before* emitting the stream — forces a `send_blocking` timeout.
    Delayed {
        delay: std::time::Duration,
        events: Vec<ReplyEvent>,
    },
}

/// A fake [`AgentSession`] that records every `send` prompt and every `shutdown`,
/// so a test can assert the rendezvous happened (send) and the session was *not*
/// killed (zero shutdowns) on timeout.
struct FakeSession {
    id: SessionId,
    script: Mutex<Option<SendScript>>,
    sends: Arc<Mutex<Vec<String>>>,
    shutdowns: Arc<Mutex<usize>>,
}
impl FakeSession {
    fn new(id: SessionId, script: SendScript) -> Arc<Self> {
        Arc::new(Self {
            id,
            script: Mutex::new(Some(script)),
            sends: Arc::new(Mutex::new(Vec::new())),
            shutdowns: Arc::new(Mutex::new(0)),
        })
    }
}
#[async_trait]
impl AgentSession for FakeSession {
    fn id(&self) -> SessionId {
        self.id
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        self.sends.lock().unwrap().push(prompt.to_owned());
        let script = self
            .script
            .lock()
            .unwrap()
            .take()
            .expect("send scripted exactly once");
        match script {
            SendScript::Stream(events) => Ok(Box::new(events.into_iter())),
            SendScript::Delayed { delay, events } => {
                tokio::time::sleep(delay).await;
                Ok(Box::new(events.into_iter()))
            }
        }
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        *self.shutdowns.lock().unwrap() += 1;
        Ok(())
    }
}

/// A fake [`AgentSessionFactory`] that hands out a pre-built [`FakeSession`] on
/// `start`, so launching a *dead* target in structured mode registers a session
/// the orchestrator can then drive. `supports` is true for any profile bearing a
/// `structured_adapter` (mirrors the real factory's gate).
struct FakeFactory {
    session: Arc<FakeSession>,
    starts: Arc<Mutex<usize>>,
}
impl FakeFactory {
    fn new(session: Arc<FakeSession>) -> Self {
        Self {
            session,
            starts: Arc::new(Mutex::new(0)),
        }
    }
}
#[async_trait]
impl AgentSessionFactory for FakeFactory {
    fn supports(&self, profile: &AgentProfile) -> bool {
        profile.structured_adapter.is_some()
    }
    async fn start(
        &self,
        _profile: &AgentProfile,
        _ctx: &PreparedContext,
        _cwd: &ProjectPath,
        _session: &SessionPlan,
    ) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
        *self.starts.lock().unwrap() += 1;
        Ok(Arc::clone(&self.session) as Arc<dyn AgentSession>)
    }
}

fn final_(c: &str) -> ReplyEvent {
    ReplyEvent::Final {
        content: c.to_owned(),
    }
}

struct SeqIds(Mutex<u128>);
impl SeqIds {
    fn new() -> Self {
        Self(Mutex::new(1))
    }
}
impl IdGenerator for SeqIds {
    fn new_uuid(&self) -> Uuid {
        let mut n = self.0.lock().unwrap();
        let id = Uuid::from_u128(*n);
        *n += 1;
        id
    }
}

// ---------------------------------------------------------------------------
// Builders
// ---------------------------------------------------------------------------

fn pid(n: u128) -> ProfileId {
    ProfileId::from_uuid(Uuid::from_u128(n))
}
fn aid(n: u128) -> AgentId {
    AgentId::from_uuid(Uuid::from_u128(n))
}
fn sid(n: u128) -> SessionId {
    SessionId::from_uuid(Uuid::from_u128(n))
}
fn nid(n: u128) -> NodeId {
    NodeId::from_uuid(Uuid::from_u128(n))
}

fn project() -> Project {
    Project::new(
        ProjectId::from_uuid(Uuid::from_u128(1000)),
        "demo",
        ProjectPath::new("/home/me/proj").unwrap(),
        RemoteRef::local(),
        1_700_000_000_000,
    )
    .unwrap()
}

fn claude_profile() -> AgentProfile {
    AgentProfile::new(
        pid(9),
        "Claude Code",
        "claude",
        Vec::new(),
        ContextInjection::stdin(),
        Some("claude --version".to_owned()),
        "{agentRunDir}",
        None,
    )
    .unwrap()
}

fn scratch_agent(id: AgentId, name: &str, md: &str) -> Agent {
    Agent::new(id, name, md, pid(9), AgentOrigin::Scratch, false).unwrap()
}

/// Same id (`pid(9)`) as [`claude_profile`] so agents created with it resolve to a
/// structured-capable profile, but bearing a `structured_adapter` so `LaunchAgent`
/// routes through the (fake) [`AgentSessionFactory`] instead of the PTY.
fn structured_profile() -> AgentProfile {
    AgentProfile::new(
        pid(9),
        "Claude Code",
        "claude",
        Vec::new(),
        ContextInjection::stdin(),
        Some("claude --version".to_owned()),
        "{agentRunDir}",
        None,
    )
    .unwrap()
    .with_structured_adapter(StructuredAdapter::Claude)
}

/// Everything wired for a dispatch test.
struct Fixture {
    service: OrchestratorService,
    contexts: FakeContexts,
    pty: FakePty,
    bus: SpyBus,
    sessions: Arc<TerminalSessions>,
    skills: RecordingSkills,
}

fn fixture(contexts: FakeContexts) -> Fixture {
    let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
    let sessions = Arc::new(TerminalSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    let launch = Arc::new(LaunchAgent::new(
        Arc::new(contexts.clone()),
        Arc::clone(&profiles) as Arc<dyn ProfileStore>,
        Arc::new(FakeRuntime),
        Arc::new(FakeFs),
        Arc::new(pty.clone()),
        Arc::new(FakeSkills),
        Arc::clone(&sessions),
        Arc::new(bus.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(FakeRecall),
        None,
    ));
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let skills = RecordingSkills::default();
    let create_skill = Arc::new(CreateSkill::new(
        Arc::new(skills.clone()) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));

    let service = OrchestratorService::new(
        create,
        launch,
        list,
        close,
        update,
        create_skill,
        Arc::clone(&profiles) as Arc<dyn ProfileStore>,
        Arc::clone(&sessions),
    );

    Fixture {
        service,
        contexts,
        pty,
        bus,
        sessions,
        skills,
    }
}

fn cmd(json: &str) -> OrchestratorCommand {
    serde_json::from_str::<OrchestratorRequest>(json)
        .unwrap()
        .validate()
        .unwrap()
}

// ---------------------------------------------------------------------------
// D6 fixture: orchestrator wired for `agent.message` (inter-agent rendezvous)
// ---------------------------------------------------------------------------

/// Everything wired for an `agent.message` dispatch test (§17.4).
struct AskFixture {
    service: OrchestratorService,
    structured: Arc<StructuredSessions>,
    /// PTY (terminal) registry the service holds — lets a test register a raw
    /// terminal session for an agent to exercise the PTY-only-live branch.
    sessions: Arc<TerminalSessions>,
    bus: SpyBus,
    /// PTY spawn/kill spy of the underlying `LaunchAgent` — used to prove a
    /// dead-target launch did *not* fall back to a raw PTY (zero spawns).
    pty: FakePty,
    /// `start` counter of the fake factory (dead-target launch path).
    factory_starts: Arc<Mutex<usize>>,
}

/// Builds an orchestrator with the structured registry + event bus wired, and a
/// `LaunchAgent` whose structured routing uses `factory` over `profiles`.
///
/// `pty_only` controls the launched profile: `false` ⇒ structured-capable profile
/// (factory hands out a session); `true` ⇒ plain PTY profile (no
/// `structured_adapter`) so `launched.structured` stays `None`.
fn ask_fixture(
    contexts: FakeContexts,
    factory_session: Arc<FakeSession>,
    pty_only: bool,
) -> AskFixture {
    let profile = if pty_only {
        claude_profile()
    } else {
        structured_profile()
    };
    let profiles = Arc::new(FakeProfiles::new(vec![profile]));
    let sessions = Arc::new(TerminalSessions::new());
    let structured = Arc::new(StructuredSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();
    let factory = Arc::new(FakeFactory::new(factory_session));
    let factory_starts = Arc::clone(&factory.starts);

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    let launch = Arc::new(
        LaunchAgent::new(
            Arc::new(contexts.clone()),
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::new(FakeRuntime),
            Arc::new(FakeFs),
            Arc::new(pty.clone()),
            Arc::new(FakeSkills),
            Arc::clone(&sessions),
            Arc::new(bus.clone()),
            Arc::new(SeqIds::new()),
            Arc::new(FakeRecall),
            None,
        )
        .with_structured(factory as Arc<dyn AgentSessionFactory>, Arc::clone(&structured)),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let skills = RecordingSkills::default();
    let create_skill = Arc::new(CreateSkill::new(
        Arc::new(skills) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));

    let service = OrchestratorService::new(
        create,
        launch,
        list,
        close,
        update,
        create_skill,
        Arc::clone(&profiles) as Arc<dyn ProfileStore>,
        Arc::clone(&sessions),
    )
    .with_structured(Arc::clone(&structured))
    .with_events(Arc::new(bus.clone()));

    AskFixture {
        service,
        structured,
        sessions,
        bus,
        pty,
        factory_starts,
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn spawn_unknown_agent_creates_then_launches() {
    let fx = fixture(FakeContexts::new());

    let out = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude-code" }"#),
        )
        .await
        .expect("dispatch ok");
    assert!(out.detail.contains("dev-backend"));

    // The agent was created (manifest grew to one entry) and launched (session
    // registered as an agent, AgentLaunched published).
    let manifest = fx.contexts.manifest();
    assert_eq!(manifest.entries.len(), 1);
    assert_eq!(manifest.entries[0].name, "dev-backend");

    assert!(fx.sessions.session(&sid(777)).is_some());
    let launched = fx.bus.events().into_iter().any(
        |e| matches!(e, DomainEvent::AgentLaunched { session_id, .. } if session_id == sid(777)),
    );
    assert!(launched, "AgentLaunched must be published");
}

#[tokio::test]
async fn spawn_known_agent_launches_without_recreating() {
    let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));

    fx.service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude-code" }"#),
        )
        .await
        .expect("dispatch ok");

    // No second manifest entry — the existing agent was reused, just launched.
    assert_eq!(fx.contexts.manifest().entries.len(), 1);
    assert_eq!(fx.contexts.manifest().entries[0].agent_id, agent.id);
    assert!(fx.sessions.session(&sid(777)).is_some());
}

#[tokio::test]
async fn agent_run_existing_agent_does_not_require_profile() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));

    fx.service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.run", "targetAgent":"architect", "task":"Analyse" }"#),
        )
        .await
        .expect("dispatch ok");

    assert_eq!(fx.contexts.manifest().entries.len(), 1);
    assert!(fx.sessions.session(&sid(777)).is_some());
}

/// **Réattache même cellule (R0a, orchestrateur)** : un `spawn`/`agent.run`
/// `Visible` qui re-cible la **même** cellule hôte d'un agent déjà vivant est un
/// rebind de vue ⇒ pas d'erreur, pas de respawn.
#[tokio::test]
async fn agent_run_visible_same_cell_rebinds_existing_session() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let cell = nid(10);

    fx.service
        .dispatch(
            &project(),
            cmd(&format!(
                r#"{{ "action":"spawn_agent", "name":"architect", "profile":"claude", "visibility":"visible", "nodeId":"{cell}" }}"#
            )),
        )
        .await
        .expect("first launch ok");

    let out = fx
        .service
        .dispatch(
            &project(),
            cmd(&format!(
                r#"{{ "type":"agent.run", "targetAgent":"architect", "visibility":"visible", "attachToCell":"{cell}" }}"#
            )),
        )
        .await
        .expect("same-cell rebind ok");

    assert!(out.detail.contains("attached agent architect"));
    let session = fx.sessions.session(&sid(777)).expect("live session");
    assert_eq!(session.node_id, cell);
    assert_eq!(fx.pty.spawns().len(), 1, "rebind must not respawn");
}

/// **Second lancement ailleurs refusé (R0a, orchestrateur)** : un `spawn`/`agent.run`
/// `Visible` qui vise une **autre** cellule d'un agent singleton déjà vivant est un
/// second lancement ⇒ refus `AgentAlreadyRunning` (node hôte rapporté), pas de
/// respawn, session non déplacée.
#[tokio::test]
async fn agent_run_visible_other_cell_refuses_when_live_elsewhere() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let first_cell = nid(10);
    let next_cell = nid(20);

    fx.service
        .dispatch(
            &project(),
            cmd(&format!(
                r#"{{ "action":"spawn_agent", "name":"architect", "profile":"claude", "visibility":"visible", "nodeId":"{first_cell}" }}"#
            )),
        )
        .await
        .expect("first launch ok");

    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(&format!(
                r#"{{ "type":"agent.run", "targetAgent":"architect", "visibility":"visible", "attachToCell":"{next_cell}" }}"#
            )),
        )
        .await
        .expect_err("fresh second launch elsewhere is refused");

    assert_eq!(err.code(), "AGENT_ALREADY_RUNNING", "got {err:?}");
    match err {
        application::AppError::AgentAlreadyRunning { node_id, .. } => {
            assert_eq!(node_id, first_cell, "reports the live host cell, not target");
        }
        other => panic!("expected AgentAlreadyRunning, got {other:?}"),
    }

    // Session not moved, no respawn.
    let session = fx.sessions.session(&sid(777)).expect("live session");
    assert_eq!(session.node_id, first_cell, "session stays on its host cell");
    assert_eq!(fx.pty.spawns().len(), 1, "refused launch must not respawn");
}

#[tokio::test]
async fn stop_agent_kills_the_right_session() {
    let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));

    // Launch it first so a session is registered for the agent.
    fx.service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude" }"#),
        )
        .await
        .unwrap();
    assert!(fx.sessions.session(&sid(777)).is_some());

    // Now stop it.
    fx.service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#),
        )
        .await
        .expect("stop ok");

    // The PTY for that session was killed and the session de-registered.
    assert_eq!(fx.pty.kills(), vec![sid(777)]);
    assert!(fx.sessions.session(&sid(777)).is_none());
}

#[tokio::test]
async fn stop_agent_without_live_session_is_not_found() {
    let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));

    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#),
        )
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
}

#[tokio::test]
async fn update_agent_context_overwrites_md() {
    let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# old"));

    fx.service
        .dispatch(
            &project(),
            cmd(
                r##"{ "action":"update_agent_context", "name":"dev-backend", "context":"# new body" }"##,
            ),
        )
        .await
        .expect("update ok");

    assert_eq!(
        fx.contexts.content("agents/dev-backend.md").as_deref(),
        Some("# new body")
    );
}

#[tokio::test]
async fn spawn_with_unknown_profile_is_not_found_and_does_not_create() {
    let fx = fixture(FakeContexts::new());

    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"x", "profile":"does-not-exist" }"#),
        )
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");

    // No agent was created when the profile could not be resolved.
    assert!(fx.contexts.manifest().entries.is_empty());
    assert!(fx.sessions.session(&sid(777)).is_none());
}

#[tokio::test]
async fn create_skill_persists_a_project_scoped_skill_by_default() {
    let fx = fixture(FakeContexts::new());

    let out = fx
        .service
        .dispatch(
            &project(),
            cmd(r##"{ "action":"create_skill", "name":"deploy", "context":"# Deploy" }"##),
        )
        .await
        .expect("create_skill ok");

    assert!(out.detail.contains("deploy"));

    let saved = fx.skills.0.lock().unwrap();
    assert_eq!(saved.len(), 1);
    assert_eq!(saved[0].name, "deploy");
    assert_eq!(saved[0].scope, SkillScope::Project);
}

#[tokio::test]
async fn create_skill_honours_global_scope() {
    let fx = fixture(FakeContexts::new());

    fx.service
        .dispatch(
            &project(),
            cmd(
                r##"{ "action":"create_skill", "name":"shared", "context":"# Shared", "scope":"global" }"##,
            ),
        )
        .await
        .expect("create_skill ok");

    let saved = fx.skills.0.lock().unwrap();
    assert_eq!(saved.len(), 1);
    assert_eq!(saved[0].scope, SkillScope::Global);
}

// ---------------------------------------------------------------------------
// D6 — agent.message (synchronous inter-agent rendezvous, §17.4)
// ---------------------------------------------------------------------------

const ASK_JSON: &str =
    r#"{ "type":"agent.message", "requestedBy":"Main", "targetAgent":"architect", "task":"Analyse §17" }"#;

/// Zone 1 — live structured target ⇒ `send_blocking` is driven and the outcome's
/// `reply` carries the turn's `Final` content. Anti-always-green: the assertion is
/// on the *exact* content, so it fails for `None` or any other string.
#[tokio::test]
async fn ask_live_structured_target_returns_final_content() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    // A live session already registered for the agent in the *structured* registry.
    let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_("the §17 answer")]));
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));

    let out = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("ask ok");

    // The exact Final content is returned (would fail on None or any other text).
    assert_eq!(out.reply.as_deref(), Some("the §17 answer"));
    // The prompt actually reached the live session.
    assert_eq!(session.sends.lock().unwrap().as_slice(), &["Analyse §17"]);
    // Live session reused: factory never started a new one, no PTY spawned.
    assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
    assert!(fx.pty.spawns().is_empty(), "must not spawn a PTY");
    // Session not killed.
    assert_eq!(*session.shutdowns.lock().unwrap(), 0);
}

/// Anti-always-green guard, explicit negative: the same fixture but the session
/// returns a *different* Final ⇒ the strict equality from the test above must NOT
/// hold. Proves the assertion has teeth.
#[tokio::test]
async fn ask_reply_assertion_is_content_sensitive() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_("SOMETHING ELSE")]));
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(session as Arc<dyn AgentSession>, aid(1), nid(1));

    let out = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("ask ok");
    assert_ne!(out.reply.as_deref(), Some("the §17 answer"));
    assert_eq!(out.reply.as_deref(), Some("SOMETHING ELSE"));
}

/// Zone 4 — on success, `DomainEvent::AgentReplied` is published with the right
/// agent id and `reply_len` (byte length of the content).
#[tokio::test]
async fn ask_success_publishes_agent_replied_event() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let content = "réponse"; // multibyte: len() in bytes, not chars
    let session = FakeSession::new(sid(500), SendScript::Stream(vec![final_(content)]));
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(session as Arc<dyn AgentSession>, aid(1), nid(1));

    fx.service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("ask ok");

    let replied = fx
        .bus
        .events()
        .into_iter()
        .find_map(|e| match e {
            DomainEvent::AgentReplied {
                agent_id,
                reply_len,
            } => Some((agent_id, reply_len)),
            _ => None,
        })
        .expect("AgentReplied must be published");
    assert_eq!(replied.0, aid(1));
    assert_eq!(replied.1, content.len());
}

/// Zone 2 (orchestrator seam) — a failing turn (stream ends without a `Final`)
/// surfaces a typed `PROCESS` error WITHOUT the orchestrator killing the session,
/// and the session stays registered (retry possible). The *genuine* timeout
/// variant of this invariant is asserted in
/// [`timeout_invariant_send_blocking_does_not_kill_session`] (the 300s prod bound
/// cannot be awaited here, so we drive the same "never shutdown on ask" branch via
/// the no-`Final` error and the bounded `send_blocking` directly).
#[tokio::test]
async fn ask_failed_turn_does_not_kill_session_and_errors() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let session = FakeSession::new(sid(500), SendScript::Stream(vec![/* no Final */]));
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));

    let err = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .unwrap_err();

    // Typed PROCESS error (send_blocking's AgentSessionError → AppError::Process).
    assert_eq!(err.code(), "PROCESS", "got {err:?}");
    // Session NOT killed and STILL registered (retry possible).
    assert_eq!(
        *session.shutdowns.lock().unwrap(),
        0,
        "ask must never shutdown the target session"
    );
    assert!(
        fx.structured.session_for_agent(&aid(1)).is_some(),
        "session must remain in the registry after a failed turn"
    );
}

/// Zone 2 (real timeout) — uses a *short-bounded* `send_blocking` directly against
/// the orchestrator's session is impossible (bound is internal/300s); instead we
/// prove the genuine timeout-does-not-kill invariant at the `send_blocking` seam
/// with a real delayed session, mirroring how the orchestrator awaits it.
#[tokio::test]
async fn timeout_invariant_send_blocking_does_not_kill_session() {
    use application::send_blocking;
    let session = FakeSession::new(
        sid(500),
        SendScript::Delayed {
            delay: std::time::Duration::from_secs(1),
            events: vec![final_("too late")],
        },
    );
    let out = send_blocking(
        session.as_ref(),
        "Analyse §17",
        Some(std::time::Duration::from_millis(20)),
    )
    .await;
    assert_eq!(out, Err(AgentSessionError::Timeout));
    assert_eq!(
        *session.shutdowns.lock().unwrap(),
        0,
        "timeout must not kill the session"
    );
    // Negative control: had we waited (no bound) the call would succeed — proves the
    // timeout above is the cause of the error, not a broken session.
    let session2 = FakeSession::new(
        sid(501),
        SendScript::Delayed {
            delay: std::time::Duration::from_millis(5),
            events: vec![final_("eventually")],
        },
    );
    let ok = send_blocking(session2.as_ref(), "x", None).await;
    assert_eq!(ok, Ok("eventually".to_owned()));
}

/// Zone 3 — dead target ⇒ `LaunchAgent` is invoked in structured mode (factory
/// `start` called, structured session registered) *then* `send_blocking`; the
/// reply is returned and no raw PTY is spawned.
#[tokio::test]
async fn ask_dead_target_launches_structured_then_sends() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    // The session the fake factory will hand out on launch.
    let session = FakeSession::new(sid(700), SendScript::Stream(vec![final_("launched reply")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        Arc::clone(&session),
        false,
    );
    // No live session up front.
    assert!(fx.structured.session_for_agent(&aid(1)).is_none());

    let out = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("ask ok");

    assert_eq!(out.reply.as_deref(), Some("launched reply"));
    // Factory was started exactly once (structured launch), the prompt was sent.
    assert_eq!(*fx.factory_starts.lock().unwrap(), 1);
    assert_eq!(session.sends.lock().unwrap().as_slice(), &["Analyse §17"]);
    // Structured-mode launch ⇒ NO raw PTY spawn.
    assert!(fx.pty.spawns().is_empty(), "structured launch must not spawn a PTY");
    // The session is now registered (1 session/agent).
    assert!(fx.structured.session_for_agent(&aid(1)).is_some());
}

/// Zone 5a — target already live in the **PTY** registry (raw terminal, no
/// structured channel) ⇒ typed `INVALID`, never an ACK, never a launch.
#[tokio::test]
async fn ask_pty_live_target_is_invalid_no_ack() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    // Register the agent as live in the **PTY** (terminal) registry only — a raw
    // terminal with no structured reply channel.
    let pty_session = TerminalSession::starting(
        sid(800),
        nid(2),
        project().root.clone(),
        SessionKind::Agent { agent_id: aid(1) },
        PtySize::new(24, 80).unwrap(),
    );
    fx.sessions
        .insert(PtyHandle { session_id: sid(800) }, pty_session);
    assert!(fx.structured.session_for_agent(&aid(1)).is_none());

    let err = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .unwrap_err();

    assert_eq!(
        err.code(),
        "INVALID",
        "PTY-live target is not addressable by ask: {err:?}"
    );
    // Never launched, never sent: factory untouched, no PTY spawned.
    assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
    assert!(fx.pty.spawns().is_empty());
    // No structured session conjured.
    assert!(fx.structured.session_for_agent(&aid(1)).is_none());
}

/// Zone 5b — after launching, the target turns out PTY-only (profile without a
/// `structured_adapter` ⇒ `launched.structured` is `None`) ⇒ typed `INVALID`, no
/// ACK, and the orchestrator does not pretend a reply.
#[tokio::test]
async fn ask_pty_only_profile_after_launch_is_invalid() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    // pty_only = true ⇒ launched profile has no structured_adapter ⇒ structured None.
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        true,
    );
    assert!(fx.structured.session_for_agent(&aid(1)).is_none());

    let err = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .unwrap_err();

    assert_eq!(err.code(), "INVALID", "PTY-only target must error typed: {err:?}");
    // No structured session was registered, no reply pretended.
    assert!(fx.structured.session_for_agent(&aid(1)).is_none());
}

/// Zone 6/7 — structured registry not wired ⇒ `ask` is INVALID (legacy service).
#[tokio::test]
async fn ask_without_structured_wired_is_invalid() {
    // The default `fixture` builds the service via `OrchestratorService::new`
    // WITHOUT `.with_structured(...)`, so AskAgent cannot be served.
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let err = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .unwrap_err();
    assert_eq!(err.code(), "INVALID", "got {err:?}");
}

/// Zone 6/7 — unknown target agent ⇒ typed `NOT_FOUND`, no launch, no reply.
#[tokio::test]
async fn ask_unknown_target_is_not_found() {
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(FakeContexts::new(), throwaway, false);
    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"ghost", "task":"hi" }"#),
        )
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
    assert_eq!(*fx.factory_starts.lock().unwrap(), 0);
    assert!(fx.pty.spawns().is_empty());
}

/// Zone 7 — non-regression: `agent.run` (fire-and-forget) still yields `reply:
/// None` even through a structured-wired orchestrator.
#[tokio::test]
async fn non_regression_agent_run_reply_is_none() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let session = FakeSession::new(sid(700), SendScript::Stream(vec![final_("x")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        session,
        false,
    );
    let out = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.run", "targetAgent":"architect", "task":"go" }"#),
        )
        .await
        .expect("run ok");
    assert_eq!(out.reply, None, "agent.run must not carry a reply");
}

// ---------------------------------------------------------------------------
// A0 — sérialisation FIFO des tours par agent (cadrage v5 §4)
//
// Le verrou `ask_locks` de l'orchestrateur doit garantir : (1) deux `ask`
// concurrents vers la **même** cible sont sérialisés FIFO (le 2ᵉ `send` ne
// démarre PAS tant que le 1ᵉ tour n'a pas rendu son `Final`) ; (2) deux `ask`
// vers des agents **différents** s'exécutent en parallèle ; (3) le verrou est
// relâché même si le tour se solde par une erreur.
//
// Pour observer le *timing*, on remplace le `FakeSession` scripté one-shot par
// un `GatedSession` **pilotable** : chaque `send` enregistre son démarrage
// (compteur + ordre) puis **bloque** sur un permis de `Semaphore` que le test
// délivre à la demande, avant de retourner son `Final`. Cela matérialise un tour
// dont on contrôle la fin (« retient le Final jusqu'à ce que le test le libère »).
// ---------------------------------------------------------------------------

use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::sync::Semaphore;
use tokio::time::{timeout, Duration};

/// Borne de sûreté : aucun `await` de test ne doit jamais dépasser ça (garde-fou
/// anti-hang du lot de concurrence). Largement au-dessus du temps réel attendu.
const TEST_GUARD: Duration = Duration::from_secs(10);

/// Une session **pilotable** : chaque `send` signale son démarrage (incrémente
/// `started`, pousse le prompt dans `order`) puis **attend un permis** du
/// sémaphore `gate` avant de retourner son `Final` (contenu = prompt reçu, pour
/// vérifier l'ordre de complétion). Le test délivre les permis un par un
/// (`release_one`) ⇒ il contrôle exactement quand chaque tour se termine.
///
/// `mode` choisit l'issue du tour : `Final` (succès) ou `NoFinal` (flux vide ⇒
/// `send_blocking` renvoie une erreur, pour tester la libération du verrou sur
/// erreur).
struct GatedSession {
    id: SessionId,
    /// Nombre de `send` **démarrés** (avant déblocage). Observé par le test pour
    /// prouver qu'un 2ᵉ tour n'a PAS démarré pendant que le 1ᵉ est retenu.
    started: AtomicUsize,
    /// Nombre de `send` **terminés** (Final rendu / flux retourné).
    finished: AtomicUsize,
    /// Ordre des prompts reçus (FIFO attendu).
    order: Arc<Mutex<Vec<String>>>,
    /// Permis débloquant un tour à la fois.
    gate: Semaphore,
    /// Nombre de `shutdown` (doit rester 0 : l'orchestrateur ne tue jamais la
    /// session de la cible).
    shutdowns: AtomicUsize,
    mode: GateMode,
}

#[derive(Clone, Copy)]
enum GateMode {
    /// Le tour rend un `Final` dont le contenu reprend le prompt.
    FinalEchoesPrompt,
    /// Le flux est vide (pas de `Final`) ⇒ tour interrompu, erreur typée.
    NoFinal,
}

impl GatedSession {
    fn new(id: SessionId, mode: GateMode) -> Arc<Self> {
        Arc::new(Self {
            id,
            started: AtomicUsize::new(0),
            finished: AtomicUsize::new(0),
            order: Arc::new(Mutex::new(Vec::new())),
            gate: Semaphore::new(0),
            shutdowns: AtomicUsize::new(0),
            mode,
        })
    }
    fn started(&self) -> usize {
        self.started.load(Ordering::SeqCst)
    }
    fn finished(&self) -> usize {
        self.finished.load(Ordering::SeqCst)
    }
    fn order(&self) -> Vec<String> {
        self.order.lock().unwrap().clone()
    }
    /// Délivre un permis ⇒ débloque exactement UN tour en attente.
    fn release_one(&self) {
        self.gate.add_permits(1);
    }
}

#[async_trait]
impl AgentSession for GatedSession {
    fn id(&self) -> SessionId {
        self.id
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        self.started.fetch_add(1, Ordering::SeqCst);
        self.order.lock().unwrap().push(prompt.to_owned());
        // Bloque jusqu'à ce que le test délivre un permis (retient le Final).
        let permit = self.gate.acquire().await.expect("gate not closed");
        permit.forget();
        self.finished.fetch_add(1, Ordering::SeqCst);
        match self.mode {
            GateMode::FinalEchoesPrompt => Ok(Box::new(std::iter::once(final_(prompt)))),
            GateMode::NoFinal => Ok(Box::new(std::iter::empty())),
        }
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        self.shutdowns.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

/// Attend (borné) qu'une condition observée sur le fake devienne vraie, en cédant
/// la main au runtime entre deux lectures. Évite tout `sleep` fixe fragile.
async fn await_until<F: Fn() -> bool>(cond: F) {
    timeout(TEST_GUARD, async {
        while !cond() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("condition never reached within guard (possible hang/deadlock)");
}

const ASK_ARCHITECT: &str =
    r#"{ "type":"agent.message", "requestedBy":"Main", "targetAgent":"architect", "task":"T1" }"#;
const ASK_ARCHITECT_2: &str =
    r#"{ "type":"agent.message", "requestedBy":"Main", "targetAgent":"architect", "task":"T2" }"#;

/// **A0 #1 — FIFO même cible (cœur du lot).** Deux `ask` concurrents vers le
/// **même** agent, avec une session pilotable qui retient son `Final`. On prouve
/// que le 2ᵉ tour (`send`) ne **démarre pas** tant que le 1ᵉ n'a pas rendu son
/// `Final`, puis qu'en libérant, les deux complètent **dans l'ordre FIFO** (T1
/// avant T2), sans entrelacement.
#[tokio::test]
async fn ask_same_target_serialises_turns_fifo() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let session = GatedSession::new(sid(500), GateMode::FinalEchoesPrompt);
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));

    let proj = project();
    let svc = &fx.service;

    timeout(TEST_GUARD, async {
        // Tour 1 : on le pilote jusqu'à ce que son `send` soit en vol (verrou
        // acquis + bloqué sur le gate) AVANT de lancer le tour 2 ⇒ ordre d'entrée
        // en file déterministe (T1 devant T2). `biased` : on tente d'abord r1.
        let r1 = svc.dispatch(&proj, cmd(ASK_ARCHITECT));
        tokio::pin!(r1);
        loop {
            tokio::select! {
                biased;
                _ = &mut r1 => panic!("turn 1 should stay blocked on its Final"),
                () = await_until(|| session.started() >= 1) => break,
            }
        }

        // À ce point : tour 1 démarré et BLOQUÉ sur le Final.
        assert_eq!(session.started(), 1, "exactly the first turn started");
        assert_eq!(session.finished(), 0, "first turn is still blocked on Final");

        // Lance le tour 2 concurremment. Il doit rester EN FILE (verrou de la même
        // cible tenu par le tour 1) : son `send` ne doit PAS démarrer. On poll les
        // DEUX futures pendant qu'on laisse au runtime des occasions de mal faire.
        let r2 = svc.dispatch(&proj, cmd(ASK_ARCHITECT_2));
        tokio::pin!(r2);
        for _ in 0..50 {
            tokio::select! {
                biased;
                _ = &mut r1 => panic!("turn 1 still blocked, must not complete yet"),
                _ = &mut r2 => panic!("turn 2 must not run before turn 1 releases the lock"),
                () = tokio::task::yield_now() => {}
            }
        }
        assert_eq!(
            session.started(),
            1,
            "second turn MUST NOT start while the first holds the per-agent lock (no interleaving)"
        );

        // Libère le tour 1 ⇒ il rend son Final, relâche le verrou, le tour 2 démarre.
        session.release_one();
        let out1 = (&mut r1).await.expect("turn 1 ok");
        assert_eq!(out1.reply.as_deref(), Some("T1"));

        // Le tour 2 peut maintenant démarrer (verrou libre) ; il bloque sur SON
        // gate. On le pilote jusqu'à son démarrage, puis on le libère.
        loop {
            tokio::select! {
                biased;
                res = &mut r2 => {
                    // Démarré ET libéré dans la même boucle : accepte la complétion.
                    let out2 = res.expect("turn 2 ok");
                    assert_eq!(out2.reply.as_deref(), Some("T2"));
                    break;
                }
                () = await_until(|| session.started() >= 2) => {
                    session.release_one();
                }
            }
        }
    })
    .await
    .expect("FIFO scenario hung (deadlock?)");

    // Complétion FIFO stricte : T1 puis T2, sans entrelacement.
    assert_eq!(session.order(), vec!["T1".to_owned(), "T2".to_owned()]);
    assert_eq!(session.finished(), 2);
    assert_eq!(
        session.shutdowns.load(Ordering::SeqCst),
        0,
        "ask must never shutdown the target session"
    );
}

/// **A0 #2 — agents différents en parallèle.** Un `ask` vers A (retenu sur son
/// `Final`) ne doit PAS bloquer un `ask` vers B : le tour de B démarre et
/// **complète** pendant que A est encore en cours. Verrou **par agent_id** ⇒ pas
/// de contention croisée. Si A bloquait B, l'attente de complétion de B
/// dépasserait le garde-fou et le test échouerait.
#[tokio::test]
async fn ask_different_targets_run_in_parallel() {
    // Deux agents distincts dans le manifeste.
    let contexts = FakeContexts::new();
    {
        let agent_a = scratch_agent(aid(1), "alpha", "agents/alpha.md");
        let agent_b = scratch_agent(aid(2), "beta", "agents/beta.md");
        let mut inner = contexts.0.lock().unwrap();
        inner
            .manifest
            .entries
            .push(ManifestEntry::from_agent(&agent_a));
        inner
            .manifest
            .entries
            .push(ManifestEntry::from_agent(&agent_b));
        inner
            .contents
            .insert(agent_a.context_path.clone(), "# a".to_owned());
        inner
            .contents
            .insert(agent_b.context_path.clone(), "# b".to_owned());
    }

    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(contexts, throwaway, false);

    // A : retenu indéfiniment (jamais libéré pendant le test). B : libérable.
    let sess_a = GatedSession::new(sid(500), GateMode::FinalEchoesPrompt);
    let sess_b = GatedSession::new(sid(600), GateMode::FinalEchoesPrompt);
    fx.structured
        .insert(Arc::clone(&sess_a) as Arc<dyn AgentSession>, aid(1), nid(1));
    fx.structured
        .insert(Arc::clone(&sess_b) as Arc<dyn AgentSession>, aid(2), nid(2));

    let proj = project();
    let svc = &fx.service;

    let ask_a = r#"{ "type":"agent.message", "targetAgent":"alpha", "task":"A1" }"#;
    let ask_b = r#"{ "type":"agent.message", "targetAgent":"beta", "task":"B1" }"#;

    timeout(TEST_GUARD, async {
        // Lance A et attends qu'il soit en vol (bloqué sur son Final), sans le libérer.
        let a_fut = svc.dispatch(&proj, cmd(ask_a));
        tokio::pin!(a_fut);
        // Pilote a_fut jusqu'à ce que A soit en vol.
        let drive_a_until_inflight = async {
            // a_fut bloque dans send → on le poll une fois via select avec la condition.
            loop {
                tokio::select! {
                    biased;
                    _ = &mut a_fut => panic!("A should stay blocked on its Final"),
                    () = await_until(|| sess_a.started() >= 1) => break,
                }
            }
        };
        drive_a_until_inflight.await;
        assert_eq!(sess_a.started(), 1);
        assert_eq!(sess_a.finished(), 0, "A is held on its Final");

        // Pendant que A est retenu, B doit pouvoir démarrer ET compléter.
        let b_out = async {
            // B bloque aussi sur son gate : on attend qu'il démarre, on le libère.
            let b_fut = svc.dispatch(&proj, cmd(ask_b));
            tokio::pin!(b_fut);
            loop {
                tokio::select! {
                    biased;
                    res = &mut b_fut => break res,
                    () = await_until(|| sess_b.started() >= 1) => {
                        sess_b.release_one();
                    }
                }
            }
        }
        .await
        .expect("B completes while A is still in flight");

        assert_eq!(b_out.reply.as_deref(), Some("B1"));
        // A est toujours en vol, non terminé : la parallélisme est prouvée.
        assert_eq!(sess_a.finished(), 0, "A must still be in flight (not blocked by B, nor B by A)");
        assert_eq!(sess_b.finished(), 1);
    })
    .await
    .expect("parallel scenario hung — A likely blocked B (cross-agent contention)");
}

/// **A0 #3 — libération du verrou sur erreur.** Un 1ᵉ tour qui se solde par une
/// **erreur** (flux sans `Final` ⇒ `PROCESS`) doit néanmoins **relâcher** le
/// verrou de la cible : un `ask` suivant vers le **même** agent peut alors
/// acquérir le verrou et démarrer (pas de verrou fuité / pas de deadlock).
#[tokio::test]
async fn ask_releases_lock_after_errored_turn() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    // 1ᵉ tour : flux vide ⇒ send_blocking renvoie une erreur (PROCESS).
    let session = GatedSession::new(sid(500), GateMode::NoFinal);
    let throwaway = FakeSession::new(sid(999), SendScript::Stream(vec![final_("unused")]));
    let fx = ask_fixture(
        FakeContexts::with_agent(&agent, "# persona"),
        throwaway,
        false,
    );
    fx.structured
        .insert(Arc::clone(&session) as Arc<dyn AgentSession>, aid(1), nid(1));

    let proj = project();
    let svc = &fx.service;

    timeout(TEST_GUARD, async {
        // 1ᵉ tour : on le débloque immédiatement, il échoue (pas de Final).
        session.release_one();
        let err = svc
            .dispatch(&proj, cmd(ASK_ARCHITECT))
            .await
            .expect_err("first turn errors (no Final)");
        assert_eq!(err.code(), "PROCESS", "errored turn surfaces typed PROCESS: {err:?}");
        assert_eq!(session.started(), 1);

        // 2ᵉ tour vers la MÊME cible : si le verrou avait fuité sur l'erreur, ce
        // `dispatch` resterait bloqué sur `lock_owned()` et le garde-fou sauterait.
        // On débloque le 2ᵉ tour et on attend qu'il RENDE LA MAIN — preuve que le
        // verrou était libre. (Cette session est `NoFinal` ⇒ le 2ᵉ tour échoue lui
        // aussi ; ce qui compte ici est qu'il a pu *acquérir le verrou et démarrer*,
        // pas l'issue du tour — l'issue est couverte par le test FIFO.)
        session.release_one();
        let err2 = svc
            .dispatch(&proj, cmd(ASK_ARCHITECT_2))
            .await
            .expect_err("NoFinal session ⇒ second turn also errors typed");
        assert_eq!(err2.code(), "PROCESS", "got {err2:?}");
        assert_eq!(
            session.started(),
            2,
            "second turn actually started — the lock was freed after the errored first turn"
        );
    })
    .await
    .expect("lock appears leaked after an errored turn (second ask never acquired it)");
}

// A0 #4 — Plafond d'attente en file (`ASK_QUEUE_WAIT_CAP = 600 s`). NON testé en
// unitaire : il n'est ni injectable ni réductible depuis l'extérieur (constante
// privée), et un vrai test exigerait d'attendre 600 s — exclu (durée). La
// sémantique nominale (attente puis acquisition du verrou) est couverte par
// `ask_same_target_serialises_turns_fifo` (le 2ᵉ tour attend en file PUIS
// acquiert). Le chemin d'expiration (timeout d'attente ⇒ `AppError::Process`,
// même type que le timeout de tour) reste non couvert ici — signalé au Dev :
// rendre le cap injectable (ex. champ optionnel surchargeable en test) permettrait
// un test déterministe court.