//! 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, 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::{OrchestratorCommand, OrchestratorRequest, PtySize, SessionId};
use uuid::Uuid;

use application::{
    CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents,
    OrchestratorService, 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>>>,
    /// Records `(session_id, text)` of every `write` so the inter-agent tests can
    /// assert the delegated task was written into the target's terminal.
    writes: Arc<Mutex<Vec<(SessionId, String)>>>,
}
impl FakePty {
    fn new(next_id: SessionId) -> Self {
        Self {
            next_id,
            kills: Arc::new(Mutex::new(Vec::new())),
            spawns: Arc::new(Mutex::new(Vec::new())),
            writes: Arc::new(Mutex::new(Vec::new())),
        }
    }
    fn spawns(&self) -> Vec<SessionId> {
        self.spawns.lock().unwrap().clone()
    }
    #[allow(dead_code)]
    fn kills(&self) -> Vec<SessionId> {
        self.kills.lock().unwrap().clone()
    }
    /// The texts written to a given session (lossy UTF-8), in order.
    fn writes_for(&self, session_id: SessionId) -> Vec<String> {
        self.writes
            .lock()
            .unwrap()
            .iter()
            .filter(|(s, _)| *s == session_id)
            .map(|(_, t)| t.clone())
            .collect()
    }
}
#[async_trait]
impl PtyPort for FakePty {
    async fn spawn(&self, _spec: SpawnSpec, _size: PtySize) -> Result<PtyHandle, PtyError> {
        self.spawns.lock().unwrap().push(self.next_id);
        Ok(PtyHandle {
            session_id: self.next_id,
        })
    }
    fn write(&self, handle: &PtyHandle, data: &[u8]) -> Result<(), PtyError> {
        self.writes.lock().unwrap().push((
            handle.session_id,
            String::from_utf8_lossy(data).into_owned(),
        ));
        Ok(())
    }
    fn resize(&self, _handle: &PtyHandle, _size: PtySize) -> Result<(), PtyError> {
        Ok(())
    }
    fn subscribe_output(&self, _handle: &PtyHandle) -> Result<OutputStream, PtyError> {
        Ok(Box::new(std::iter::empty()))
    }
    fn scrollback(&self, _handle: &PtyHandle) -> Result<Vec<u8>, PtyError> {
        Ok(Vec::new())
    }
    async fn 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())
    }
}

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

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

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

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

fn claude_profile() -> AgentProfile {
    // Profil Claude **complet** : adaptateur structuré + capacité MCP `.mcp.json`.
    // C'est la condition que la garde F2 (`guard_mcp_bridge_supported`) exige pour
    // accepter une cible d'`idea_ask_agent` (seul Claude consomme le pont `.mcp.json`).
    AgentProfile::new(
        pid(9),
        "Claude Code",
        "claude",
        Vec::new(),
        ContextInjection::stdin(),
        Some("claude --version".to_owned()),
        "{agentRunDir}",
        None,
    )
    .unwrap()
    .with_structured_adapter(StructuredAdapter::Claude)
    .with_mcp(McpCapability::new(
        McpConfigStrategy::config_file(".mcp.json").unwrap(),
        McpTransport::Stdio,
    ))
}

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

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

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

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

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

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

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

// ---------------------------------------------------------------------------
// Basic dispatch coverage (spawn / stop / update / skill) — non-ask commands.
// ---------------------------------------------------------------------------

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

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

    let manifest = fx.contexts.manifest();
    assert_eq!(manifest.entries.len(), 1);
    assert_eq!(manifest.entries[0].name, "dev-backend");
    assert!(fx.sessions.session(&sid(777)).is_some());
    let launched = fx.bus.events().into_iter().any(
        |e| matches!(e, DomainEvent::AgentLaunched { session_id, .. } if session_id == sid(777)),
    );
    assert!(launched, "AgentLaunched must be published");
}

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

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

    assert_eq!(fx.contexts.manifest().entries.len(), 1);
    assert_eq!(fx.contexts.manifest().entries[0].agent_id, agent.id);
    assert!(fx.sessions.session(&sid(777)).is_some());
}

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

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

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

/// R0a (orchestrator): a `Visible` spawn re-targeting the same host cell of a live
/// agent is a view rebind ⇒ no error, no respawn.
#[tokio::test]
async fn agent_run_visible_same_cell_rebinds_existing_session() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let cell = nid(10);

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

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

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

/// R0a (orchestrator): a `Visible` spawn targeting a *different* cell of a live
/// singleton ⇒ refused `AgentAlreadyRunning` (host cell reported), no respawn.
#[tokio::test]
async fn agent_run_visible_other_cell_refuses_when_live_elsewhere() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let first_cell = nid(10);
    let next_cell = nid(20);

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

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

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

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

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

    fx.service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"dev-backend", "profile":"claude" }"#),
        )
        .await
        .unwrap();
    assert!(fx.sessions.session(&sid(777)).is_some());

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

    assert_eq!(fx.pty.kills(), vec![sid(777)]);
    assert!(fx.sessions.session(&sid(777)).is_none());
}

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

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

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

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

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

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

    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "action":"spawn_agent", "name":"x", "profile":"does-not-exist" }"#),
        )
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
    assert!(fx.contexts.manifest().entries.is_empty());
    assert!(fx.sessions.session(&sid(777)).is_none());
}

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

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

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

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

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

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

// ---------------------------------------------------------------------------
// Option 1 « Terminal + MCP » — idea_ask_agent / idea_reply (B-3 / B-4)
//
// The target's view is a raw native terminal (PTY). Delegation flows through the
// mailbox: `ask_agent` ensures the target is live in the PTY registry, writes the
// prefixed task into its terminal, and AWAITS a `PendingReply`; the target's later
// `idea_reply` resolves the head ticket of its queue, waking the awaiting ask.
//
// To exercise the blocking rendezvous deterministically, a test spawns the ask as a
// task, waits until the mailbox holds a pending ticket, then dispatches the matching
// `Reply` (or cancels via timeout) to unblock it. A real `InMemoryMailbox` is used
// (the production adapter), wired through `with_mailbox`.
// ---------------------------------------------------------------------------

use std::collections::VecDeque;

use domain::conversation::{
    Conversation, ConversationId, ConversationParty, ConversationRegistry, ConversationSession,
    SessionRef,
};
use domain::input::{AgentBusyState, InputMediator};
use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId};
use tokio::time::{timeout, Duration};

/// Safety guard: no awaited rendezvous in these tests should ever exceed this.
const TEST_GUARD: Duration = Duration::from_secs(10);

/// Minimal in-test [`AgentMailbox`] mirroring the production `InMemoryMailbox`
/// (which lives in `infrastructure` and cannot be a dev-dep of `application` — that
/// would be a dependency cycle). Per-agent FIFO of `(Ticket, oneshot::Sender)`.
#[derive(Default)]
struct TestMailbox {
    queues: Mutex<HashMap<AgentId, VecDeque<(Ticket, tokio::sync::oneshot::Sender<String>)>>>,
}
impl TestMailbox {
    fn new() -> Self {
        Self {
            queues: Mutex::new(HashMap::new()),
        }
    }
    fn pending(&self, agent: &AgentId) -> usize {
        self.queues
            .lock()
            .unwrap()
            .get(agent)
            .map_or(0, VecDeque::len)
    }
    /// Ids of the tickets queued for `agent`, in FIFO order (test inspection).
    fn ticket_ids(&self, agent: &AgentId) -> Vec<TicketId> {
        self.queues
            .lock()
            .unwrap()
            .get(agent)
            .map(|q| q.iter().map(|(t, _)| t.id).collect())
            .unwrap_or_default()
    }
}
impl AgentMailbox for TestMailbox {
    fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
        let (tx, rx) = tokio::sync::oneshot::channel::<String>();
        self.queues
            .lock()
            .unwrap()
            .entry(agent)
            .or_default()
            .push_back((ticket, tx));
        PendingReply::new(Box::pin(async move {
            rx.await.map_err(|_| MailboxError::Cancelled)
        }))
    }
    fn resolve(&self, agent: AgentId, result: String) -> Result<(), MailboxError> {
        let slot = {
            let mut q = self.queues.lock().unwrap();
            let queue = q
                .get_mut(&agent)
                .filter(|q| !q.is_empty())
                .ok_or(MailboxError::NoPendingRequest(agent))?;
            queue.pop_front().expect("non-empty")
        };
        let _ = slot.1.send(result);
        Ok(())
    }
    fn resolve_ticket(
        &self,
        agent: AgentId,
        ticket_id: TicketId,
        result: String,
    ) -> Result<(), MailboxError> {
        let slot = {
            let mut q = self.queues.lock().unwrap();
            let queue = q
                .get_mut(&agent)
                .filter(|q| !q.is_empty())
                .ok_or(MailboxError::NoPendingRequest(agent))?;
            let pos = queue
                .iter()
                .position(|(t, _)| t.id == ticket_id)
                .ok_or(MailboxError::NoPendingRequest(agent))?;
            queue.remove(pos).expect("found position")
        };
        let _ = slot.1.send(result);
        Ok(())
    }
    fn cancel_head(&self, agent: AgentId, ticket_id: TicketId) {
        let mut q = self.queues.lock().unwrap();
        if let Some(queue) = q.get_mut(&agent) {
            if queue.front().map(|(t, _)| t.id) == Some(ticket_id) {
                queue.pop_front();
            }
        }
    }
}

/// A delivering [`InputMediator`] for the application tests: it composes the
/// [`TestMailbox`] (correlation engine) and writes the prefixed turn into the agent's
/// bound PTY handle — the test mirror of `infrastructure::MediatedInbox::with_pty`
/// (which `application` cannot depend on without a dependency cycle).
struct TestMediator {
    mailbox: Arc<TestMailbox>,
    pty: Arc<dyn PtyPort>,
    handles: Mutex<HashMap<AgentId, PtyHandle>>,
    busy: Mutex<HashMap<AgentId, AgentBusyState>>,
    /// Records every agent passed to `preempt`, so the interrupt test can assert the
    /// Interrompre path calls preempt (and only preempt — no enqueue, no resolve).
    preempts: Arc<Mutex<Vec<AgentId>>>,
}
impl TestMediator {
    fn new(mailbox: Arc<TestMailbox>, pty: Arc<dyn PtyPort>) -> Self {
        Self {
            mailbox,
            pty,
            handles: Mutex::new(HashMap::new()),
            busy: Mutex::new(HashMap::new()),
            preempts: Arc::new(Mutex::new(Vec::new())),
        }
    }
    fn preempts(&self) -> Vec<AgentId> {
        self.preempts.lock().unwrap().clone()
    }
}
impl InputMediator for TestMediator {
    fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
        let ticket_id = ticket.id;
        {
            let mut b = self.busy.lock().unwrap();
            let st = b.entry(agent).or_insert(AgentBusyState::Idle);
            if !st.is_busy() {
                *st = AgentBusyState::Busy {
                    ticket: ticket_id,
                    since_ms: 0,
                };
            }
        }
        if let Some(handle) = self.handles.lock().unwrap().get(&agent).cloned() {
            let line = format!(
                "[IdeA · tâche de {} · ticket {}] {}\n",
                ticket.requester, ticket_id, ticket.task
            );
            let _ = self.pty.write(&handle, line.as_bytes());
        }
        self.mailbox.enqueue(agent, ticket)
    }
    fn bind_handle(&self, agent: AgentId, handle: PtyHandle) {
        self.handles.lock().unwrap().insert(agent, handle);
    }
    fn delivers_turn(&self, agent: AgentId) -> bool {
        self.handles.lock().unwrap().contains_key(&agent)
    }
    fn preempt(&self, agent: AgentId) {
        self.preempts.lock().unwrap().push(agent);
    }
    fn mark_idle(&self, agent: AgentId) {
        self.busy
            .lock()
            .unwrap()
            .insert(agent, AgentBusyState::Idle);
    }
    fn busy_state(&self, agent: AgentId) -> AgentBusyState {
        self.busy
            .lock()
            .unwrap()
            .get(&agent)
            .copied()
            .unwrap_or(AgentBusyState::Idle)
    }
}

/// Minimal in-test [`ConversationRegistry`] (pair-keyed lazy get-or-create), mirroring
/// `infrastructure::InMemoryConversationRegistry`.
#[derive(Default)]
struct TestConversations {
    by_pair: Mutex<HashMap<(String, String), ConversationId>>,
    by_id: Mutex<HashMap<ConversationId, Conversation>>,
}
impl TestConversations {
    fn new() -> Self {
        Self::default()
    }
    fn key(a: ConversationParty, b: ConversationParty) -> (String, String) {
        let s = |p: ConversationParty| match p {
            ConversationParty::User => "user".to_owned(),
            ConversationParty::Agent { agent_id } => agent_id.to_string(),
        };
        let (ka, kb) = (s(a), s(b));
        if ka <= kb {
            (ka, kb)
        } else {
            (kb, ka)
        }
    }
}
impl ConversationRegistry for TestConversations {
    fn resolve(&self, a: ConversationParty, b: ConversationParty) -> Conversation {
        let key = Self::key(a, b);
        let mut pairs = self.by_pair.lock().unwrap();
        if let Some(id) = pairs.get(&key).copied() {
            return self.by_id.lock().unwrap().get(&id).cloned().unwrap();
        }
        let id = ConversationId::new_random();
        let conv = Conversation::try_new(id, a, b).expect("valid pair");
        pairs.insert(key, id);
        self.by_id.lock().unwrap().insert(id, conv.clone());
        conv
    }
    fn bind_session(&self, id: ConversationId, session: SessionRef) {
        if let Some(c) = self.by_id.lock().unwrap().get_mut(&id) {
            c.session = ConversationSession::Live {
                handle_ref: session,
            };
        }
    }
    fn suspend(&self, id: ConversationId, resumable_id: Option<String>) {
        if let Some(c) = self.by_id.lock().unwrap().get_mut(&id) {
            c.session = ConversationSession::Dormant;
            c.resumable_id = resumable_id;
        }
    }
    fn get(&self, id: ConversationId) -> Option<Conversation> {
        self.by_id.lock().unwrap().get(&id).cloned()
    }
}

/// Everything wired for an `idea_ask_agent` / `idea_reply` dispatch test.
struct AskFixture {
    service: Arc<OrchestratorService>,
    /// The real in-memory mailbox the service holds — lets a test observe pending
    /// tickets and (in production it is the orchestrator that resolves them).
    mailbox: Arc<TestMailbox>,
    /// PTY (terminal) registry — the channel the target's task is written into.
    sessions: Arc<TerminalSessions>,
    bus: SpyBus,
    /// PTY spy: records spawns (a dead target is launched as a PTY) and writes.
    pty: FakePty,
    /// The mediated inbox the service holds — lets a test observe `busy_state`
    /// transitions (e.g. an `idea_reply` flips the emitter back to Idle — lot C5).
    mediator: Arc<TestMediator>,
}

/// Builds an orchestrator wired with a real [`InMemoryMailbox`] + PTY (B-3) and a
/// plain (non-structured) [`LaunchAgent`] — so a dead target is launched as a PTY,
/// exactly like production after B-2.
fn ask_fixture(contexts: FakeContexts) -> AskFixture {
    let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
    let sessions = Arc::new(TerminalSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();
    let mailbox = Arc::new(TestMailbox::new());

    let 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 create_skill = Arc::new(CreateSkill::new(
        Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));

    let mediator = Arc::new(TestMediator::new(
        Arc::clone(&mailbox),
        Arc::new(pty.clone()) as Arc<dyn PtyPort>,
    ));
    let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
    let service = Arc::new(
        OrchestratorService::new(
            create,
            launch,
            list,
            close,
            update,
            create_skill,
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::clone(&sessions),
        )
        .with_input_mediator(
            Arc::clone(&mediator) as Arc<dyn InputMediator>,
            Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
        )
        .with_conversations(conversations)
        .with_events(Arc::new(bus.clone())),
    );

    AskFixture {
        service,
        mailbox,
        sessions,
        bus,
        pty,
        mediator,
    }
}

/// Seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it instead
/// of launching one.
fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) {
    let session = TerminalSession::starting(
        session_id,
        nid(1),
        ProjectPath::new("/home/me/proj").unwrap(),
        SessionKind::Agent { agent_id },
        PtySize::new(24, 80).unwrap(),
    );
    sessions.insert(PtyHandle { session_id }, session);
}

/// Waits (bounded) for a condition to hold, yielding between polls.
async fn await_until<F: Fn() -> bool>(cond: F) {
    timeout(TEST_GUARD, async {
        while !cond() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("condition never reached within guard (possible hang/deadlock)");
}

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

/// Builds an `agent.reply` command for `from` with `result` (the handshake-injected
/// path, exactly what the MCP server produces from a peer's `idea_reply`).
fn reply_cmd(from: AgentId, result: &str) -> OrchestratorCommand {
    OrchestratorCommand::Reply {
        from,
        ticket: None,
        result: result.to_owned(),
    }
}

/// A reply correlated **by ticket** (multi-thread correlation).
fn reply_cmd_ticket(from: AgentId, ticket: TicketId, result: &str) -> OrchestratorCommand {
    OrchestratorCommand::Reply {
        from,
        ticket: Some(ticket),
        result: result.to_owned(),
    }
}

// --- B-3: ask blocks on the mailbox, reply unblocks it ---------------------

/// A live-PTY target: `ask` writes the prefixed task into its terminal and AWAITS;
/// a matching `idea_reply` resolves the head ticket and the ask returns its content.
#[tokio::test]
async fn ask_live_pty_target_writes_task_and_returns_reply() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    // Target already live in the PTY registry (session 800).
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });

    // The ask enqueues a ticket and blocks awaiting the reply.
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    // The task was written into the target's terminal, prefixed for idea_reply.
    let writes = fx.pty.writes_for(sid(800));
    assert_eq!(
        writes.len(),
        1,
        "exactly one task write to the live terminal"
    );
    assert!(writes[0].contains("Analyse §17"), "task body written");
    assert!(
        writes[0].contains("[IdeA · tâche"),
        "delegated-task prefix present"
    );
    assert!(
        writes[0].contains("idea_reply") || writes[0].contains("ticket"),
        "carries ticket/idea_reply cue"
    );
    // No PTY spawned: the live terminal was reused.
    assert!(
        fx.pty.spawns().is_empty(),
        "live target reused, not respawned"
    );

    // The target renders its result via idea_reply ⇒ resolves the head ticket.
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "the §17 answer"))
        .await
        .expect("reply ok");

    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes once replied")
        .expect("join ok")
        .expect("ask ok");
    assert_eq!(out.reply.as_deref(), Some("the §17 answer"));
    assert_eq!(fx.mailbox.pending(&aid(1)), 0, "ticket drained on reply");
}

/// Lot C5 — explicit OR signal: an `idea_reply` from the emitting agent flips it back
/// to `Idle` (its FIFO can advance), independently of any prompt-ready pattern. Before
/// the reply the agent is `Busy` on the delegated turn; after, it is `Idle`.
#[tokio::test]
async fn idea_reply_marks_emitter_idle() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    // The delegated turn started ⇒ the target is Busy.
    assert!(
        fx.mediator.busy_state(aid(1)).is_busy(),
        "target Busy while processing the delegated turn"
    );

    // The target renders its result via idea_reply ⇒ explicit end-of-turn signal.
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "the §17 answer"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");

    // C5: the reply marked the emitter Idle (FIFO can advance) — no prompt pattern needed.
    assert_eq!(
        fx.mediator.busy_state(aid(1)),
        AgentBusyState::Idle,
        "idea_reply is the explicit signal that frees the turn"
    );
}

/// A dead target is launched in the background (PTY) before the task is written.
#[tokio::test]
async fn ask_dead_target_launches_pty_then_writes_and_replies() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    assert!(fx.sessions.session_for_agent(&aid(1)).is_none());

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });

    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    // The dead target was launched as a PTY (spawn recorded, session registered).
    assert_eq!(
        fx.pty.spawns(),
        vec![sid(777)],
        "dead target launched as PTY"
    );
    assert_eq!(fx.sessions.session_for_agent(&aid(1)), Some(sid(777)));
    // The delegated task was written into the freshly-launched terminal (the Stdin
    // context injection may also write the persona, so we look for the task prefix).
    assert!(
        fx.pty
            .writes_for(sid(777))
            .iter()
            .any(|w| w.contains("[IdeA · tâche") && w.contains("Analyse §17")),
        "delegated task written into the launched terminal"
    );

    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "launched reply"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");
    assert_eq!(out.reply.as_deref(), Some("launched reply"));
}

/// On a successful reply, `DomainEvent::AgentReplied` is published with the right
/// agent id and byte length.
#[tokio::test]
async fn ask_success_publishes_agent_replied_event() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));
    let content = "réponse"; // multibyte: len() in bytes

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), content))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();

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

// --- B-3: errors / not-found / not-wired -----------------------------------

/// Unknown target ⇒ typed `NOT_FOUND`, no launch, no ticket.
#[tokio::test]
async fn ask_unknown_target_is_not_found() {
    let fx = ask_fixture(FakeContexts::new());
    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"ghost", "task":"hi" }"#),
        )
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "got {err:?}");
    assert!(fx.pty.spawns().is_empty());
}

/// Mailbox/PTY not wired (legacy `OrchestratorService::new` without `with_mailbox`)
/// ⇒ `ask` is INVALID.
#[tokio::test]
async fn ask_without_mailbox_wired_is_invalid() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    // The plain `fixture` builds the service WITHOUT `.with_mailbox(...)`.
    let fx = fixture(FakeContexts::with_agent(&agent, "# persona"));
    let err = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .unwrap_err();
    assert_eq!(err.code(), "INVALID", "got {err:?}");
}

/// `agent.run` (fire-and-forget) still yields `reply: None` through a mailbox-wired
/// orchestrator (non-regression).
#[tokio::test]
async fn non_regression_agent_run_reply_is_none() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    let out = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.run", "targetAgent":"architect", "task":"go" }"#),
        )
        .await
        .expect("run ok");
    assert_eq!(out.reply, None, "agent.run must not carry a reply");
}

// --- B-4: idea_reply (Reply command) ---------------------------------------

/// A `Reply` with no in-flight ask for that agent ⇒ typed `INVALID` (no panic).
#[tokio::test]
async fn reply_without_pending_ask_is_invalid() {
    let fx = ask_fixture(FakeContexts::new());
    let err = fx
        .service
        .dispatch(&project(), reply_cmd(aid(7), "orphan result"))
        .await
        .unwrap_err();
    assert_eq!(
        err.code(),
        "INVALID",
        "orphan reply is typed INVALID: {err:?}"
    );
}

/// `Reply` resolves the HEAD ticket of the emitting agent's queue (positional
/// correlation): two asks to the same target, two replies, FIFO order.
#[tokio::test]
async fn reply_resolves_head_of_queue_fifo() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    // Two concurrent asks to the SAME target. The per-agent turn lock serialises
    // them: only the first holds the lock and enqueues; the second waits on the lock.
    let svc1 = Arc::clone(&fx.service);
    let ask1 = tokio::spawn(async move {
        svc1.dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"architect", "task":"T1" }"#),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;

    // First reply resolves T1; ask1 returns it. Then ask2 may proceed.
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "R1"))
        .await
        .expect("reply 1 ok");
    let out1 = timeout(TEST_GUARD, ask1).await.unwrap().unwrap().unwrap();
    assert_eq!(out1.reply.as_deref(), Some("R1"));

    let svc2 = Arc::clone(&fx.service);
    let ask2 = tokio::spawn(async move {
        svc2.dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"architect", "task":"T2" }"#),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "R2"))
        .await
        .expect("reply 2 ok");
    let out2 = timeout(TEST_GUARD, ask2).await.unwrap().unwrap().unwrap();
    assert_eq!(out2.reply.as_deref(), Some("R2"));
}

/// Two asks to DIFFERENT targets do not block each other (per-agent lock + per-agent
/// queue): A is held (never replied) while B completes.
#[tokio::test]
async fn ask_different_targets_run_in_parallel() {
    let contexts = FakeContexts::new();
    {
        let a = scratch_agent(aid(1), "alpha", "agents/alpha.md");
        let b = scratch_agent(aid(2), "beta", "agents/beta.md");
        let mut inner = contexts.0.lock().unwrap();
        inner.manifest.entries.push(ManifestEntry::from_agent(&a));
        inner.manifest.entries.push(ManifestEntry::from_agent(&b));
        inner
            .contents
            .insert(a.context_path.clone(), "# a".to_owned());
        inner
            .contents
            .insert(b.context_path.clone(), "# b".to_owned());
    }
    let fx = ask_fixture(contexts);
    seed_live_pty(&fx.sessions, aid(1), sid(801));
    seed_live_pty(&fx.sessions, aid(2), sid(802));

    // A: launched and held (never replied during the test).
    let svc_a = Arc::clone(&fx.service);
    let _ask_a = tokio::spawn(async move {
        svc_a
            .dispatch(
                &project(),
                cmd(r#"{ "type":"agent.message", "targetAgent":"alpha", "task":"A1" }"#),
            )
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;

    // B completes while A is still in flight (no cross-agent contention).
    let svc_b = Arc::clone(&fx.service);
    let ask_b = tokio::spawn(async move {
        svc_b
            .dispatch(
                &project(),
                cmd(r#"{ "type":"agent.message", "targetAgent":"beta", "task":"B1" }"#),
            )
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(2), "B-done"))
        .await
        .expect("reply B ok");
    let out_b = timeout(TEST_GUARD, ask_b)
        .await
        .expect("B completes while A is held")
        .unwrap()
        .unwrap();
    assert_eq!(out_b.reply.as_deref(), Some("B-done"));
    // A is still pending: it was not unblocked by B.
    assert_eq!(fx.mailbox.pending(&aid(1)), 1, "A still in flight");
}

// ---------------------------------------------------------------------------
// F1 — McpRuntimeProvider câblé (preuve que le runtime atteint LaunchAgentInput)
// + F2 — garde guard_mcp_bridge_supported
//
// F1 était le bug : sur le chemin `ask`, `ensure_live_pty` passait
// `mcp_runtime: None` ⇒ `.mcp.json` minimal ⇒ pont MCP jamais spawné ⇒ timeout.
// Le fix injecte un `McpRuntimeProvider`. On le prouve **de bout en bout** : la
// déclaration `.mcp.json` réellement écrite par `LaunchAgent::apply_mcp_config`
// (via le FileSystem) doit porter l'endpoint/exe/requester du runtime fourni —
// ce qui n'est possible QUE si le runtime a traversé `LaunchAgentInput.mcp_runtime`.
// ---------------------------------------------------------------------------

/// FileSystem enregistrant chaque `write(path, content)` pour pouvoir inspecter la
/// déclaration `.mcp.json` matérialisée par le lancement. `exists → false` pour que
/// `apply_mcp_config` écrive bien le fichier (chemin non-clobbering).
#[derive(Clone, Default)]
struct CapturingFs(Arc<Mutex<Vec<(String, String)>>>);
impl CapturingFs {
    fn writes(&self) -> Vec<(String, String)> {
        self.0.lock().unwrap().clone()
    }
    /// Contenu écrit dont le chemin se termine par `suffix` (ex. `.mcp.json`).
    fn content_ending_with(&self, suffix: &str) -> Option<String> {
        self.0
            .lock()
            .unwrap()
            .iter()
            .find(|(p, _)| p.ends_with(suffix))
            .map(|(_, c)| c.clone())
    }
}
#[async_trait]
impl FileSystem for CapturingFs {
    async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
        Err(FsError::NotFound(path.as_str().to_owned()))
    }
    async fn write(&self, path: &RemotePath, data: &[u8]) -> Result<(), FsError> {
        self.0.lock().unwrap().push((
            path.as_str().to_owned(),
            String::from_utf8_lossy(data).into_owned(),
        ));
        Ok(())
    }
    async fn exists(&self, _path: &RemotePath) -> Result<bool, FsError> {
        Ok(false)
    }
    async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
        Ok(())
    }
    async fn list(&self, _path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
        Ok(Vec::new())
    }
    async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
        Ok(())
    }
}

/// Provider de runtime MCP de test : renvoie un [`McpRuntime`] connu et **enregistre**
/// l'`(project_id, agent_id)` reçus, prouvant que l'orchestrateur l'interroge avec la
/// cible relancée.
#[derive(Clone, Default)]
struct FakeMcpRuntimeProvider {
    calls: Arc<Mutex<Vec<(ProjectId, AgentId)>>>,
}
impl FakeMcpRuntimeProvider {
    fn calls(&self) -> Vec<(ProjectId, AgentId)> {
        self.calls.lock().unwrap().clone()
    }
    /// Le runtime connu que ce provider injecte (valeurs sentinelles repérables dans
    /// le `.mcp.json` écrit).
    fn known_runtime(agent_id: AgentId) -> application::McpRuntime {
        application::McpRuntime {
            exe: "/opt/idea/idea.AppImage".to_owned(),
            endpoint: "/run/user/1000/idea-mcp/SENTINEL-endpoint.sock".to_owned(),
            project_id: "deadbeefdeadbeefdeadbeefdeadbeef".to_owned(),
            requester: agent_id.to_string(),
        }
    }
}
impl application::McpRuntimeProvider for FakeMcpRuntimeProvider {
    fn runtime_for(&self, project: &Project, agent_id: AgentId) -> Option<application::McpRuntime> {
        self.calls.lock().unwrap().push((project.id, agent_id));
        Some(Self::known_runtime(agent_id))
    }
}

/// Codex : profil structuré NON supporté par le pont `.mcp.json` (la garde F2 doit
/// le refuser, même s'il déclare une capacité MCP `ConfigFile(.mcp.json)`).
fn codex_profile() -> AgentProfile {
    AgentProfile::new(
        pid(11),
        "Codex CLI",
        "codex",
        Vec::new(),
        ContextInjection::stdin(),
        Some("codex --version".to_owned()),
        "{agentRunDir}",
        None,
    )
    .unwrap()
    .with_structured_adapter(StructuredAdapter::Codex)
    .with_mcp(McpCapability::new(
        McpConfigStrategy::config_file(".mcp.json").unwrap(),
        McpTransport::Stdio,
    ))
}

/// Fixture `ask` paramétrable : choix du FileSystem (capturant), des profils, et d'un
/// éventuel [`McpRuntimeProvider`]. Reproduit `ask_fixture` mais expose les leviers
/// nécessaires aux preuves F1/F2.
struct AskFixtureEx {
    service: Arc<OrchestratorService>,
    mailbox: Arc<TestMailbox>,
    sessions: Arc<TerminalSessions>,
    pty: FakePty,
    fs: CapturingFs,
}

fn ask_fixture_ex(
    contexts: FakeContexts,
    profiles: Vec<AgentProfile>,
    provider: Option<Arc<FakeMcpRuntimeProvider>>,
) -> AskFixtureEx {
    let profiles = Arc::new(FakeProfiles::new(profiles));
    let sessions = Arc::new(TerminalSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();
    let mailbox = Arc::new(TestMailbox::new());
    let fs = CapturingFs::default();

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

    let mediator = Arc::new(TestMediator::new(
        Arc::clone(&mailbox),
        Arc::new(pty.clone()) as Arc<dyn PtyPort>,
    )) as Arc<dyn InputMediator>;
    let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
    let mut service = OrchestratorService::new(
        create,
        launch,
        list,
        close,
        update,
        create_skill,
        Arc::clone(&profiles) as Arc<dyn ProfileStore>,
        Arc::clone(&sessions),
    )
    .with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
    .with_conversations(conversations)
    .with_events(Arc::new(bus.clone()));

    if let Some(p) = provider {
        service = service.with_mcp_runtime_provider(p as Arc<dyn application::McpRuntimeProvider>);
    }

    AskFixtureEx {
        service: Arc::new(service),
        mailbox,
        sessions,
        pty,
        fs,
    }
}

/// F1 — PREUVE DIRECTE. Un `McpRuntimeProvider` câblé renvoie un runtime connu ; un
/// `ask` vers une cible **morte** passe par `ensure_live_pty` → `launch_agent`, et la
/// déclaration `.mcp.json` réellement matérialisée porte l'endpoint/exe/requester du
/// runtime fourni. C'est la preuve que `mcp_runtime: Some(runtime)` a traversé
/// `LaunchAgentInput` (avant le fix il valait `None` ⇒ déclaration minimale).
#[tokio::test]
async fn f1_ask_dead_target_injects_provider_runtime_into_mcp_json() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let provider = Arc::new(FakeMcpRuntimeProvider::default());
    let fx = ask_fixture_ex(
        FakeContexts::with_agent(&agent, "# persona"),
        vec![claude_profile()],
        Some(Arc::clone(&provider)),
    );
    assert!(fx.sessions.session_for_agent(&aid(1)).is_none());

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;

    // Le provider a été interrogé avec le projet et la cible relancée.
    let calls = provider.calls();
    assert_eq!(calls.len(), 1, "provider interrogé exactement une fois");
    assert_eq!(calls[0].0, project().id, "interrogé pour le bon projet");
    assert_eq!(
        calls[0].1,
        aid(1),
        "interrogé pour la cible relancée (= --requester)"
    );

    // La déclaration `.mcp.json` écrite porte les valeurs sentinelles du runtime.
    let decl = fx
        .fs
        .content_ending_with(".mcp.json")
        .expect("un .mcp.json doit être écrit au lancement de la cible morte");
    assert!(
        decl.contains("/opt/idea/idea.AppImage"),
        "exe du runtime injecté présent dans le .mcp.json: {decl}"
    );
    assert!(
        decl.contains("SENTINEL-endpoint.sock"),
        "endpoint du runtime injecté présent: {decl}"
    );
    assert!(
        decl.contains("deadbeefdeadbeefdeadbeefdeadbeef"),
        "project_id du runtime injecté présent: {decl}"
    );
    assert!(
        decl.contains(&aid(1).to_string()),
        "requester (= cible) présent: {decl}"
    );
    // Preuve de discrimination avec le mode dégradé : le `command` est l'exe injecté,
    // PAS la valeur minimale `"command": "idea"` qu'on aurait sans runtime.
    assert!(
        !decl.contains("\"command\": \"idea\""),
        "command NE doit PAS être la valeur minimale (runtime injecté): {decl}"
    );

    // Débloque l'ask pour ne pas laisser la tâche pendante.
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "done"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}

/// F1 — NON-RÉGRESSION. Sans provider câblé (`OrchestratorService::new` legacy), le
/// lancement issu d'`ask` passe `mcp_runtime: None` ⇒ déclaration **minimale**
/// (`command = "idea"`, pas d'endpoint). Comportement legacy intact.
#[tokio::test]
async fn f1_ask_without_provider_writes_minimal_mcp_json() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture_ex(
        FakeContexts::with_agent(&agent, "# persona"),
        vec![claude_profile()],
        None, // pas de provider câblé
    );

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;

    let decl = fx
        .fs
        .content_ending_with(".mcp.json")
        .expect("un .mcp.json minimal doit être écrit");
    assert!(
        decl.contains("\"command\": \"idea\""),
        "command minimal attendu sans provider: {decl}"
    );
    assert!(
        !decl.contains("--endpoint"),
        "aucune déclaration d'endpoint sans runtime injecté: {decl}"
    );

    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "done"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}

/// F2 — GARDE (rejet). Une cible au profil Codex (structuré mais non consommateur du
/// pont `.mcp.json`) ⇒ `AppError::Invalid` **immédiat** (pas de timeout 300s), aucun
/// lancement, aucun ticket.
#[tokio::test]
async fn f2_ask_codex_target_is_invalid_no_launch() {
    let agent = Agent::new(
        aid(1),
        "coder",
        "agents/coder.md",
        pid(11), // profil Codex
        AgentOrigin::Scratch,
        false,
    )
    .unwrap();
    let fx = ask_fixture_ex(
        FakeContexts::with_agent(&agent, "# persona"),
        vec![codex_profile()],
        Some(Arc::new(FakeMcpRuntimeProvider::default())),
    );

    let err = fx
        .service
        .dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#),
        )
        .await
        .unwrap_err();

    assert_eq!(
        err.code(),
        "INVALID",
        "garde F2 = erreur typée Invalid: {err:?}"
    );
    let msg = err.to_string();
    assert!(
        msg.contains("idea_*") || msg.contains("pont") || msg.contains(".mcp.json"),
        "message explicite sur le pont non supporté: {msg}"
    );
    assert!(
        fx.pty.spawns().is_empty(),
        "aucun lancement sur cible refusée"
    );
    assert_eq!(fx.mailbox.pending(&aid(1)), 0, "aucun ticket enfilé");
}

/// F2 — GARDE (passant). Une cible au profil Claude conforme passe la garde et
/// l'`ask` se déroule normalement jusqu'à la réponse (preuve que la garde ne bloque
/// pas les cibles légitimes).
#[tokio::test]
async fn f2_ask_claude_target_passes_guard() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture_ex(
        FakeContexts::with_agent(&agent, "# persona"),
        vec![claude_profile()],
        Some(Arc::new(FakeMcpRuntimeProvider::default())),
    );
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "ok claude"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
    assert_eq!(out.reply.as_deref(), Some("ok claude"));
}

// ---------------------------------------------------------------------------
// C3 — conversation par paire : routage de fil, détection de cycle, corrélation
// par ticket, fallback tête, timeout libère la file.
// ---------------------------------------------------------------------------

/// Builds an ask-wired service exposing the **shared** [`TestConversations`] registry
/// (so a test can assert the resolved thread is A↔B, never User↔B). Mirrors
/// `ask_fixture` but returns the registry handle.
struct AskFixtureC3 {
    service: Arc<OrchestratorService>,
    mailbox: Arc<TestMailbox>,
    sessions: Arc<TerminalSessions>,
    conversations: Arc<TestConversations>,
}

fn ask_fixture_c3(contexts: FakeContexts) -> AskFixtureC3 {
    let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
    let sessions = Arc::new(TerminalSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();
    let mailbox = Arc::new(TestMailbox::new());
    let conversations = Arc::new(TestConversations::new());

    let 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 create_skill = Arc::new(CreateSkill::new(
        Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));
    let mediator = Arc::new(TestMediator::new(
        Arc::clone(&mailbox),
        Arc::new(pty.clone()) as Arc<dyn PtyPort>,
    )) as Arc<dyn InputMediator>;

    let service = Arc::new(
        OrchestratorService::new(
            create,
            launch,
            list,
            close,
            update,
            create_skill,
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::clone(&sessions),
        )
        .with_input_mediator(mediator, Arc::clone(&mailbox) as Arc<dyn AgentMailbox>)
        .with_conversations(Arc::clone(&conversations) as Arc<dyn ConversationRegistry>)
        .with_events(Arc::new(bus.clone())),
    );

    AskFixtureC3 {
        service,
        mailbox,
        sessions,
        conversations,
    }
}

/// Seeds two agents A (id 1) and B (id 2) in the manifest.
fn seed_two_agents(contexts: &FakeContexts) {
    let a = scratch_agent(aid(1), "alpha", "agents/alpha.md");
    let b = scratch_agent(aid(2), "beta", "agents/beta.md");
    let mut inner = contexts.0.lock().unwrap();
    inner.manifest.entries.push(ManifestEntry::from_agent(&a));
    inner.manifest.entries.push(ManifestEntry::from_agent(&b));
    inner
        .contents
        .insert(a.context_path.clone(), "# a".to_owned());
    inner
        .contents
        .insert(b.context_path.clone(), "# b".to_owned());
}

/// An `agent.message` from agent A (uuid requester) to a named target.
fn ask_from_agent_json(requester: AgentId, target: &str, task: &str) -> String {
    format!(
        r#"{{ "type":"agent.message", "requestedBy":"{requester}", "targetAgent":"{target}", "task":"{task}" }}"#
    )
}

/// C3 — an ask A→B routes into the **A↔B** thread, never User↔B. The resolved
/// conversation relates the two agents, and is distinct from the User↔B thread.
#[tokio::test]
async fn ask_agent_routes_into_a_to_b_conversation_not_user_b() {
    let contexts = FakeContexts::new();
    seed_two_agents(&contexts);
    let fx = ask_fixture_c3(contexts);
    seed_live_pty(&fx.sessions, aid(2), sid(802)); // B live

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move {
        svc.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(1), "beta", "do B")),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;

    // The registry now holds the A↔B thread (agent 1 ↔ agent 2), not User↔B.
    let a_to_b = fx.conversations.resolve(
        ConversationParty::agent(aid(1)),
        ConversationParty::agent(aid(2)),
    );
    let user_b = fx
        .conversations
        .resolve(ConversationParty::User, ConversationParty::agent(aid(2)));
    assert_ne!(a_to_b.id, user_b.id, "A↔B is a distinct thread from User↔B");
    assert!(
        a_to_b.same_pair(
            ConversationParty::agent(aid(1)),
            ConversationParty::agent(aid(2))
        ),
        "ask A→B resolved the A↔B pair"
    );
    // The live B session is bound to the A↔B thread (session keyed by conversation).
    assert!(
        a_to_b.session.is_live(),
        "A↔B thread bound to a live session"
    );

    fx.service
        .dispatch(&project(), reply_cmd(aid(2), "B done"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
    assert_eq!(out.reply.as_deref(), Some("B done"));
}

/// C3 — a re-entrant delegation A→B→A is refused with a typed error **before** any
/// deadlock: while A→B is in flight (edge A→B posted), B→A must be rejected.
#[tokio::test]
async fn cycle_a_to_b_to_a_is_refused_before_deadlock() {
    let contexts = FakeContexts::new();
    seed_two_agents(&contexts);
    let fx = ask_fixture_c3(contexts);
    seed_live_pty(&fx.sessions, aid(1), sid(801)); // A live
    seed_live_pty(&fx.sessions, aid(2), sid(802)); // B live

    // A→B in flight (held — B never replies during the test): edge A→B is posted.
    let svc = Arc::clone(&fx.service);
    let _ask_ab = tokio::spawn(async move {
        svc.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(1), "beta", "to B")),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;

    // Now B→A would close the cycle ⇒ typed INVALID, no enqueue on A, returns fast.
    let err = timeout(
        TEST_GUARD,
        fx.service.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(2), "alpha", "back to A")),
        ),
    )
    .await
    .expect("cycle ask must return fast, never deadlock")
    .unwrap_err();
    assert_eq!(
        err.code(),
        "INVALID",
        "re-entrant cycle is typed INVALID: {err:?}"
    );
    assert!(
        err.to_string().contains("cycle") || err.to_string().contains("ré-entrante"),
        "explicit cycle message: {err}"
    );
    assert_eq!(fx.mailbox.pending(&aid(1)), 0, "no ticket enqueued on A");
}

/// C3 — once an A→B turn completes, the edge is cleared, so a later B→A is allowed
/// (no phantom edge): proves the wait-for edge is RAII-scoped to the turn.
#[tokio::test]
async fn edge_cleared_after_turn_allows_later_reverse_delegation() {
    let contexts = FakeContexts::new();
    seed_two_agents(&contexts);
    let fx = ask_fixture_c3(contexts);
    seed_live_pty(&fx.sessions, aid(1), sid(801));
    seed_live_pty(&fx.sessions, aid(2), sid(802));

    // A→B completes.
    let svc = Arc::clone(&fx.service);
    let ask_ab = tokio::spawn(async move {
        svc.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(1), "beta", "to B")),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(2), "ok"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask_ab).await.unwrap().unwrap().unwrap();

    // Edge A→B is now gone ⇒ B→A is allowed (would only cycle if A→B still pending).
    let svc2 = Arc::clone(&fx.service);
    let ask_ba = tokio::spawn(async move {
        svc2.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(2), "alpha", "now to A")),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "A ok"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask_ba).await.unwrap().unwrap().unwrap();
    assert_eq!(
        out.reply.as_deref(),
        Some("A ok"),
        "reverse delegation allowed"
    );
}

/// C3 — reply correlates **by ticket** (deterministic id-keyed resolution). The agent
/// echoes the ticket id from the `[IdeA · … · ticket <id>]` prefix; replying by that
/// id resolves exactly that request. (Per «1 agent = 1 employee» the turn lock keeps
/// at most one turn live per agent, so resolution is sequential here — by-ticket is
/// what makes it deterministic regardless of which requester it serves.)
#[tokio::test]
async fn reply_correlates_by_ticket() {
    let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
    let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
    seed_live_pty(&fx.sessions, aid(2), sid(802));

    // Requester A1 asks B; capture B's queued ticket id and reply it by id.
    let svc1 = Arc::clone(&fx.service);
    let ask1 = tokio::spawn(async move {
        svc1.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T1")))
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    let t1 = fx.mailbox.ticket_ids(&aid(2))[0];
    fx.service
        .dispatch(&project(), reply_cmd_ticket(aid(2), t1, "R1"))
        .await
        .expect("reply T1 by ticket ok");
    let out1 = timeout(TEST_GUARD, ask1).await.unwrap().unwrap().unwrap();
    assert_eq!(out1.reply.as_deref(), Some("R1"));

    // A different requester A2 asks B; replying a STALE/unknown ticket id is a typed
    // error (no panic), while the correct id resolves it.
    let svc2 = Arc::clone(&fx.service);
    let ask2 = tokio::spawn(async move {
        svc2.dispatch(&project(), cmd(&ask_from_agent_json(aid(11), "beta", "T2")))
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    let bogus = TicketId::new_random();
    let err = fx
        .service
        .dispatch(&project(), reply_cmd_ticket(aid(2), bogus, "wrong"))
        .await
        .unwrap_err();
    assert_eq!(
        err.code(),
        "INVALID",
        "unknown ticket id ⇒ typed INVALID: {err:?}"
    );
    let t2 = fx.mailbox.ticket_ids(&aid(2))[0];
    fx.service
        .dispatch(&project(), reply_cmd_ticket(aid(2), t2, "R2"))
        .await
        .expect("reply T2 by ticket ok");
    let out2 = timeout(TEST_GUARD, ask2).await.unwrap().unwrap().unwrap();
    assert_eq!(out2.reply.as_deref(), Some("R2"));
}

/// C3 — a reply **without** a ticket falls back to the head of the queue (compat
/// mono-thread agents).
#[tokio::test]
async fn reply_without_ticket_falls_back_to_head() {
    let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
    let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
    seed_live_pty(&fx.sessions, aid(2), sid(802));

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move {
        svc.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T")))
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    // No ticket id ⇒ head-of-queue fallback.
    fx.service
        .dispatch(&project(), reply_cmd(aid(2), "head reply"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
    assert_eq!(out.reply.as_deref(), Some("head reply"));
}

/// C3 — when the rendezvous resolves to a closed reply channel (the same retirement
/// path the bounded turn timeout takes), the ticket is **retired** from the queue and
/// the target session stays **alive** (ready for the next turn). We trigger it by
/// cancelling the head ticket (drops the sender) while the ask awaits — `cancel_head`
/// + the `Cancelled` arm mirror the timeout's `cancel_head` exactly.
#[tokio::test]
async fn timeout_path_frees_queue_and_keeps_target_alive() {
    let agent = scratch_agent(aid(2), "beta", "agents/beta.md");
    let fx = ask_fixture_c3(FakeContexts::with_agent(&agent, "# b"));
    seed_live_pty(&fx.sessions, aid(2), sid(802));

    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move {
        svc.dispatch(&project(), cmd(&ask_from_agent_json(aid(10), "beta", "T")))
            .await
    });
    await_until(|| fx.mailbox.pending(&aid(2)) == 1).await;
    let t = fx.mailbox.ticket_ids(&aid(2))[0];

    // Retire the head ticket (drops its sender) ⇒ the awaiting ask sees a closed
    // channel and returns a typed error, the SAME cleanup the turn timeout performs.
    fx.mailbox.cancel_head(aid(2), t);
    let err = timeout(TEST_GUARD, ask)
        .await
        .expect("ask returns promptly once the channel closes")
        .unwrap()
        .unwrap_err();
    assert!(
        matches!(err.code(), "PROCESS" | "INVALID"),
        "closed-channel ask is a typed error: {err:?}"
    );
    // Queue freed, and the target B is still live (never killed by the failed ask).
    assert_eq!(
        fx.mailbox.pending(&aid(2)),
        0,
        "queue freed after retirement"
    );
    assert_eq!(
        fx.sessions.session_for_agent(&aid(2)),
        Some(sid(802)),
        "target stays alive for the next turn"
    );
}

// ---------------------------------------------------------------------------
// C4 — SubmitHumanInput (Envoyer) + interrupt (Interrompre)
// ---------------------------------------------------------------------------

/// Fixture for the C4 human-input / interrupt tests. Identical wiring to
/// [`ask_fixture`] but keeps the **concrete** [`TestMediator`] so a test can assert
/// the preempt path (no enqueue, no resolved ticket) and the shared FIFO.
struct C4Fixture {
    service: Arc<OrchestratorService>,
    mailbox: Arc<TestMailbox>,
    sessions: Arc<TerminalSessions>,
    mediator: Arc<TestMediator>,
    pty: FakePty,
}

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

    let 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 create_skill = Arc::new(CreateSkill::new(
        Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));

    let mediator = Arc::new(TestMediator::new(
        Arc::clone(&mailbox),
        Arc::new(pty.clone()) as Arc<dyn PtyPort>,
    ));
    let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
    let service = Arc::new(
        OrchestratorService::new(
            create,
            launch,
            list,
            close,
            update,
            create_skill,
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::clone(&sessions),
        )
        .with_input_mediator(
            Arc::clone(&mediator) as Arc<dyn InputMediator>,
            Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
        )
        .with_conversations(conversations)
        .with_events(Arc::new(bus.clone())),
    );

    C4Fixture {
        service,
        mailbox,
        sessions,
        mediator,
        pty,
    }
}

/// A human `submit` enqueues a `from_human` ticket in the SAME FIFO the delegations
/// use: an `ask` A→B and a `submit` Human→B serialise on B's single queue.
#[tokio::test]
async fn submit_and_ask_share_one_fifo_per_agent() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = c4_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    // A delegation A→B blocks in B's FIFO (it awaits a reply).
    let svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move {
        svc.dispatch(
            &project(),
            cmd(&ask_from_agent_json(aid(10), "architect", "deleg")),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;

    // A human submit to the SAME agent lands behind it in the SAME FIFO.
    fx.service
        .submit_human_input(&project(), aid(1), "human task".to_owned())
        .await
        .expect("submit succeeds");

    assert_eq!(
        fx.mailbox.pending(&aid(1)),
        2,
        "human submit serialises behind the delegation in one FIFO"
    );
    // The human ticket's task text was written into the target's live terminal.
    let writes = fx.pty.writes_for(sid(800));
    assert!(
        writes.iter().any(|w| w.contains("human task")),
        "human turn delivered to the PTY: {writes:?}"
    );

    // Unblock the delegation so the spawned task ends cleanly.
    fx.mailbox
        .cancel_head(aid(1), fx.mailbox.ticket_ids(&aid(1))[0]);
    let _ = timeout(TEST_GUARD, ask).await;
}

/// `interrupt` = `preempt`: it calls the mediator's preempt for the agent and does
/// NOT enqueue anything nor resolve any ticket.
#[tokio::test]
async fn interrupt_preempts_without_enqueue_or_resolve() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = c4_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    assert_eq!(fx.mailbox.pending(&aid(1)), 0);
    fx.service
        .interrupt_agent(&project(), aid(1))
        .await
        .expect("interrupt succeeds");

    assert_eq!(fx.mediator.preempts(), vec![aid(1)], "preempt called once");
    assert_eq!(fx.mailbox.pending(&aid(1)), 0, "interrupt enqueues nothing");
}

/// A human submit to an unknown agent id is a typed NotFound (never a panic).
#[tokio::test]
async fn submit_unknown_agent_is_not_found() {
    let fx = c4_fixture(FakeContexts::new());
    let err = fx
        .service
        .submit_human_input(&project(), aid(999), "x".to_owned())
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "unknown agent submit: {err:?}");
}

/// An interrupt to an unknown agent id is a typed NotFound and never calls preempt.
#[tokio::test]
async fn interrupt_unknown_agent_is_not_found() {
    let fx = c4_fixture(FakeContexts::new());
    let err = fx
        .service
        .interrupt_agent(&project(), aid(999))
        .await
        .unwrap_err();
    assert_eq!(err.code(), "NOT_FOUND", "unknown agent interrupt: {err:?}");
    assert!(
        fx.mediator.preempts().is_empty(),
        "no preempt on unknown agent"
    );
}

// ---------------------------------------------------------------------------
// P6b — live wiring of `RecordTurn` in `ask_agent` (conversation persistence).
//
// On a successful inter-agent delegation, `ask_agent` persists the **pair**
// best-effort: a `Prompt` turn at enqueue (text = task, source = requester) and a
// `Response` turn on the `Ok(Ok(result))` branch (text = result, source = target),
// both on the SAME resolved `conversation_id`. Persistence is best-effort: a missing
// provider/clock, a provider returning `None`, or a failing append must NEVER turn a
// successful ask into an error.
//
// These tests reuse the full `ask`/`reply` rendezvous harness above (real mailbox +
// mediated PTY delivery + conversation registry) and add a recording in-memory
// `ConversationLog` / `RecordTurnProvider` / fixed `Clock`, mirroring the P6a fakes
// in `conversation_record.rs`.
// ---------------------------------------------------------------------------

use application::{OrchestratorOutcome, RecordTurn, RecordTurnProvider};
use domain::ports::Clock;
use domain::InputSource;
use domain::{
    ConversationLog, ConversationTurn as DomainTurn, Handoff, HandoffStore, HandoffSummarizer,
    TurnId, TurnRole,
};

/// A deterministic [`Clock`]: every persisted turn is stamped with this constant.
struct FixedClock(i64);
impl Clock for FixedClock {
    fn now_millis(&self) -> i64 {
        self.0
    }
}

/// In-memory [`ConversationLog`] that **records the order and content** of every
/// append so a test can assert the exact `[Prompt, Response]` sequence and inspect
/// each turn's `text` / `role` / `source` / `conversation`. Optionally fails on
/// `append` (best-effort path).
#[derive(Default)]
struct RecordingLog {
    appends: Mutex<Vec<DomainTurn>>,
    fail_append: bool,
}
impl RecordingLog {
    fn failing() -> Self {
        Self {
            appends: Mutex::new(Vec::new()),
            fail_append: true,
        }
    }
    fn appends(&self) -> Vec<DomainTurn> {
        self.appends.lock().unwrap().clone()
    }
}
#[async_trait]
impl ConversationLog for RecordingLog {
    async fn append(
        &self,
        _conversation: ConversationId,
        turn: DomainTurn,
    ) -> Result<(), StoreError> {
        if self.fail_append {
            return Err(StoreError::Io(
                "recording log: forced append failure".to_owned(),
            ));
        }
        self.appends.lock().unwrap().push(turn);
        Ok(())
    }
    async fn read(
        &self,
        _conversation: ConversationId,
        _since: Option<TurnId>,
    ) -> Result<Vec<DomainTurn>, StoreError> {
        Ok(self.appends())
    }
    async fn last(
        &self,
        _conversation: ConversationId,
        n: usize,
    ) -> Result<Vec<DomainTurn>, StoreError> {
        let all = self.appends();
        let start = all.len().saturating_sub(n);
        Ok(all[start..].to_vec())
    }
}

/// Trivial in-memory [`HandoffStore`] — the P6b tests only assert on the log; the
/// handoff is exercised end-to-end by P6a. Kept minimal.
#[derive(Default)]
struct NoopHandoffStore(Mutex<HashMap<ConversationId, Handoff>>);
#[async_trait]
impl HandoffStore for NoopHandoffStore {
    async fn load(&self, conversation: ConversationId) -> Result<Option<Handoff>, StoreError> {
        Ok(self.0.lock().unwrap().get(&conversation).cloned())
    }
    async fn save(&self, conversation: ConversationId, handoff: Handoff) -> Result<(), StoreError> {
        self.0.lock().unwrap().insert(conversation, handoff);
        Ok(())
    }
}

/// Deterministic summarizer: folds the new turns' text onto the previous summary.
#[derive(Default)]
struct ConcatSummarizer;
#[async_trait]
impl HandoffSummarizer for ConcatSummarizer {
    async fn fold(&self, prev: Option<Handoff>, new_turns: &[DomainTurn]) -> Handoff {
        let mut summary = prev.map(|h| h.summary_md).unwrap_or_default();
        for t in new_turns {
            summary.push_str(&t.text);
        }
        let up_to = new_turns
            .last()
            .map(|t| t.id)
            .expect("at least one new turn");
        Handoff::new(summary, up_to, None)
    }
}

/// A [`RecordTurnProvider`] that always materialises the **same** shared
/// [`RecordTurn`] (its log is observable by the test), ignoring the root — the
/// fixture pins a single project, so root-routing is out of scope here.
struct SharedRecordProvider(Arc<RecordTurn>);
impl RecordTurnProvider for SharedRecordProvider {
    fn record_turn_for(&self, _root: &ProjectPath) -> Option<Arc<RecordTurn>> {
        Some(Arc::clone(&self.0))
    }
}

/// A [`RecordTurnProvider`] that always declines (no `RecordTurn` for this root) —
/// drives the "provider returns None ⇒ silent no-op" branch.
struct NoneRecordProvider;
impl RecordTurnProvider for NoneRecordProvider {
    fn record_turn_for(&self, _root: &ProjectPath) -> Option<Arc<RecordTurn>> {
        None
    }
}

/// Builds the same `ask` harness as [`ask_fixture`] but additionally wires
/// `with_record_turn(provider, clock)`. Returns the [`AskFixture`] plus the shared
/// recording log (when one was provided) for assertion.
fn ask_fixture_with_record(
    contexts: FakeContexts,
    provider: Arc<dyn RecordTurnProvider>,
    clock_ms: i64,
) -> AskFixture {
    let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()]));
    let sessions = Arc::new(TerminalSessions::new());
    let pty = FakePty::new(sid(777));
    let bus = SpyBus::default();
    let mailbox = Arc::new(TestMailbox::new());

    let 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 create_skill = Arc::new(CreateSkill::new(
        Arc::new(RecordingSkills::default()) as Arc<dyn SkillStore>,
        Arc::new(SeqIds::new()),
    ));

    let mediator = Arc::new(TestMediator::new(
        Arc::clone(&mailbox),
        Arc::new(pty.clone()) as Arc<dyn PtyPort>,
    ));
    let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
    let service = Arc::new(
        OrchestratorService::new(
            create,
            launch,
            list,
            close,
            update,
            create_skill,
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::clone(&sessions),
        )
        .with_input_mediator(
            Arc::clone(&mediator) as Arc<dyn InputMediator>,
            Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
        )
        .with_conversations(conversations)
        .with_events(Arc::new(bus.clone()))
        .with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
    );

    AskFixture {
        service,
        mailbox,
        sessions,
        bus,
        pty,
        mediator,
    }
}

/// Drives a full `ask` round-trip (requester `from`, target architect) on the given
/// service, replying `result`. Returns the dispatch outcome.
async fn run_ask_roundtrip(
    fx: &AskFixture,
    ask_json: &str,
    reply_from: AgentId,
    result: &str,
) -> OrchestratorOutcome {
    let svc = Arc::clone(&fx.service);
    let json = ask_json.to_owned();
    let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(&json)).await });
    await_until(|| fx.mailbox.pending(&reply_from) == 1).await;
    fx.service
        .dispatch(&project(), reply_cmd(reply_from, result))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes once replied")
        .expect("join ok")
        .expect("ask ok")
}

/// Round-trip from the **User** (requester `None`) writes exactly two turns on the
/// SAME conversation, in order Prompt then Response, with the right text/role/source.
#[tokio::test]
async fn p6b_user_ask_records_prompt_then_response_pair() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let log = Arc::new(RecordingLog::default());
    let record = Arc::new(RecordTurn::new(
        Arc::clone(&log) as Arc<dyn ConversationLog>,
        Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
        Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
    ));
    let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
    let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 123);
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    // ASK_JSON uses `requestedBy:"Main"` but no agent requester id ⇒ User-sourced.
    let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "the §17 answer").await;
    assert_eq!(out.reply.as_deref(), Some("the §17 answer"));

    let appends = log.appends();
    assert_eq!(
        appends.len(),
        2,
        "exactly two turns persisted (Prompt + Response)"
    );

    let prompt = &appends[0];
    let response = &appends[1];
    // Same conversation for both turns.
    assert_eq!(
        prompt.conversation, response.conversation,
        "both turns share the resolved conversation id"
    );
    // Order + role.
    assert_eq!(prompt.role, TurnRole::Prompt, "first turn is the Prompt");
    assert_eq!(
        response.role,
        TurnRole::Response,
        "second turn is the Response"
    );
    // Text: task then result.
    assert_eq!(
        prompt.text, "Analyse §17",
        "prompt text is the delegated task"
    );
    assert_eq!(
        response.text, "the §17 answer",
        "response text is the reply result"
    );
    // Source: Human prompt (no agent requester), target-sourced response.
    assert_eq!(
        prompt.source,
        InputSource::Human,
        "User ask ⇒ Human prompt source"
    );
    assert_eq!(
        response.source,
        InputSource::agent(aid(1)),
        "response is sourced from the target (architect)"
    );
    // Clock stamp came from the injected Clock.
    assert_eq!(prompt.at_ms, 123);
    assert_eq!(response.at_ms, 123);
}

/// Round-trip with an **agent** requester (A) resolves the `A↔B` conversation and the
/// Prompt source is `agent(A)`; the Response source remains the target B.
#[tokio::test]
async fn p6b_agent_requester_records_pair_on_a_b_thread() {
    let architect = scratch_agent(aid(1), "architect", "agents/architect.md");
    let log = Arc::new(RecordingLog::default());
    let record = Arc::new(RecordTurn::new(
        Arc::clone(&log) as Arc<dyn ConversationLog>,
        Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
        Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
    ));
    let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
    let fx = ask_fixture_with_record(
        FakeContexts::with_agent(&architect, "# persona"),
        provider,
        0,
    );
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    // Requester is agent A (id 7). The orchestrator threads `requester` from the
    // `requestedBy` field of `agent.message` (parsed as a uuid when well-formed).
    let a = aid(7);
    let a_uuid = uuid::Uuid::from_u128(7);
    let ask_json = format!(
        r#"{{ "type":"agent.message", "requestedBy":"{a_uuid}", "targetAgent":"architect", "task":"Analyse §17" }}"#
    );
    let out = run_ask_roundtrip(&fx, &ask_json, aid(1), "ack").await;
    assert_eq!(out.reply.as_deref(), Some("ack"));

    let appends = log.appends();
    assert_eq!(appends.len(), 2, "two turns persisted");
    let (prompt, response) = (&appends[0], &appends[1]);

    // Independently resolve the A↔B conversation the SAME way the service does and
    // assert both turns landed on it.
    let convs = TestConversations::new();
    let expected = convs
        .resolve(
            ConversationParty::agent(a),
            ConversationParty::agent(aid(1)),
        )
        .id;
    // (Resolution is deterministic per pair within one registry; we instead assert the
    // turns share a thread and the prompt source carries A — the load-bearing facts.)
    let _ = expected;
    assert_eq!(prompt.conversation, response.conversation, "shared thread");
    assert_eq!(
        prompt.source,
        InputSource::agent(a),
        "agent requester ⇒ Prompt sourced from A"
    );
    assert_eq!(prompt.role, TurnRole::Prompt);
    assert_eq!(response.role, TurnRole::Response);
    assert_eq!(
        response.source,
        InputSource::agent(aid(1)),
        "Response sourced from the target B"
    );
}

/// No provider wired (plain `ask_fixture`) ⇒ the round-trip still succeeds and nothing
/// is persisted (no panic, no error).
#[tokio::test]
async fn p6b_no_provider_is_silent_no_op() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    // `ask_fixture` does NOT call `with_record_turn`.
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await;
    assert_eq!(
        out.reply.as_deref(),
        Some("ok"),
        "ask succeeds without persistence"
    );
}

/// Provider wired but returns `None` for the root ⇒ ask still succeeds (best-effort
/// skip).
#[tokio::test]
async fn p6b_provider_returns_none_is_silent_no_op() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let provider = Arc::new(NoneRecordProvider) as Arc<dyn RecordTurnProvider>;
    let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0);
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await;
    assert_eq!(
        out.reply.as_deref(),
        Some("ok"),
        "ask succeeds when provider declines"
    );
}

/// `RecordTurn` built on a log whose `append` always fails ⇒ persistence errors are
/// swallowed; the delegation result is NOT degraded (ask returns `Ok`).
#[tokio::test]
async fn p6b_failing_record_does_not_degrade_ask() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let failing_log = Arc::new(RecordingLog::failing());
    let record = Arc::new(RecordTurn::new(
        Arc::clone(&failing_log) as Arc<dyn ConversationLog>,
        Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
        Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
    ));
    let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
    let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0);
    seed_live_pty(&fx.sessions, aid(1), sid(800));

    let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "still ok").await;
    assert_eq!(
        out.reply.as_deref(),
        Some("still ok"),
        "a failing append must not turn a successful delegation into an error"
    );
    // The failing log recorded nothing (every append errored).
    assert!(
        failing_log.appends().is_empty(),
        "no turns stored when append fails"
    );
}