//! 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, BackgroundCompletionStream, BackgroundTaskHandle,
    BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore,
    ContextInjectionPlan, DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError,
    IdGenerator, MemoryError, MemoryStore, 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::{
    BackgroundTask, BackgroundTaskState, BackgroundTaskWakePolicy, OrchestratorCommand,
    OrchestratorRequest, PtySize, SessionId, TaskId,
};
use domain::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug};
use uuid::Uuid;

use application::{
    CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, HarvestMemoryFromTurn,
    LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, OrchestratorService,
    SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext, UpdateLiveState,
};
use domain::live_state::{LiveEntry, LiveState, WorkStatus};
use domain::ports::LiveStateStore;

// ---------------------------------------------------------------------------
// 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(),
                orchestrator: None,
            },
            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;
#[async_trait]
impl AgentRuntime for FakeRuntime {
    async 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),
            sandbox: None,
        })
    }
}

#[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(Clone, Default)]
struct RecordingBackgroundStore {
    tasks: Arc<Mutex<HashMap<TaskId, BackgroundTask>>>,
}
impl RecordingBackgroundStore {
    fn task(&self, id: TaskId) -> Option<BackgroundTask> {
        self.tasks.lock().unwrap().get(&id).cloned()
    }
}
#[async_trait]
impl BackgroundTaskStore for RecordingBackgroundStore {
    async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
        let mut tasks = self.tasks.lock().unwrap();
        if tasks.contains_key(&task.id) {
            return Err(BackgroundTaskPortError::AlreadyExists);
        }
        tasks.insert(task.id, task.clone());
        Ok(())
    }

    async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
        Ok(self.task(id))
    }

    async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
        self.tasks.lock().unwrap().insert(task.id, task.clone());
        Ok(())
    }

    async fn list_open_for_agent(
        &self,
        agent_id: AgentId,
    ) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
        Ok(self
            .tasks
            .lock()
            .unwrap()
            .values()
            .filter(|task| task.owner_agent_id == agent_id && !task.is_terminal())
            .cloned()
            .collect())
    }

    async fn list_undelivered_completions(
        &self,
    ) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
        Ok(self
            .tasks
            .lock()
            .unwrap()
            .values()
            .filter(|task| task.has_pending_completion_delivery())
            .cloned()
            .collect())
    }

    async fn mark_completion_delivered(
        &self,
        task_id: TaskId,
    ) -> Result<(), BackgroundTaskPortError> {
        let task = self
            .task(task_id)
            .ok_or(BackgroundTaskPortError::NotFound)?
            .mark_completion_delivered()
            .map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?;
        self.save(&task).await
    }
}

#[derive(Clone, Default)]
struct RecordingBackgroundRunner {
    specs: Arc<Mutex<Vec<BackgroundTaskSpec>>>,
}
impl RecordingBackgroundRunner {
    fn specs(&self) -> Vec<BackgroundTaskSpec> {
        self.specs.lock().unwrap().clone()
    }
}
#[async_trait]
impl BackgroundTaskRunner for RecordingBackgroundRunner {
    async fn spawn(
        &self,
        spec: BackgroundTaskSpec,
    ) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
        let task_id = spec.task_id;
        self.specs.lock().unwrap().push(spec);
        Ok(BackgroundTaskHandle { task_id })
    }

    async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
        Ok(())
    }

    fn subscribe_completions(&self) -> BackgroundCompletionStream {
        Box::new(std::iter::empty())
    }
}

#[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())
    }
}

/// In-memory [`MemoryStore`] for the auto-memory harvest hook (Lot E1): records the
/// saved slugs, with an optional forced `save` failure to prove harvest errors stay
/// best-effort (never break the reply).
#[derive(Default)]
struct HarvestMemories {
    saved: Mutex<Vec<MemorySlug>>,
    fail_save: bool,
}
impl HarvestMemories {
    fn slugs(&self) -> Vec<String> {
        self.saved
            .lock()
            .unwrap()
            .iter()
            .map(|s| s.as_str().to_owned())
            .collect()
    }
}
#[async_trait]
impl MemoryStore for HarvestMemories {
    async fn list(&self, _root: &ProjectPath) -> Result<Vec<Memory>, MemoryError> {
        Ok(Vec::new())
    }
    async fn get(&self, _root: &ProjectPath, _slug: &MemorySlug) -> Result<Memory, MemoryError> {
        Err(MemoryError::NotFound)
    }
    async fn save(&self, _root: &ProjectPath, memory: &Memory) -> Result<(), MemoryError> {
        if self.fail_save {
            return Err(MemoryError::Frontmatter("forced failure".to_owned()));
        }
        self.saved.lock().unwrap().push(memory.slug().clone());
        Ok(())
    }
    async fn delete(&self, _root: &ProjectPath, _slug: &MemorySlug) -> Result<(), MemoryError> {
        Ok(())
    }
    async fn read_index(&self, _root: &ProjectPath) -> Result<Vec<MemoryIndexEntry>, MemoryError> {
        Ok(Vec::new())
    }
    async fn resolve_links(
        &self,
        _root: &ProjectPath,
        _slug: &MemorySlug,
    ) -> Result<Vec<MemoryLink>, MemoryError> {
        Ok(Vec::new())
    }
}

/// In-memory [`LiveStateStore`] for the live-state auto-update hook (lot LS3): keeps a
/// keyed last-writer-wins [`LiveState`] and counts upserts (to prove no write-storm),
/// with an optional forced failure to prove the auto-update stays best-effort.
#[derive(Default)]
struct RecordingLiveState {
    state: Mutex<LiveState>,
    upserts: Mutex<usize>,
    fail: bool,
}
impl RecordingLiveState {
    fn entry_for(&self, agent: AgentId) -> Option<LiveEntry> {
        self.state
            .lock()
            .unwrap()
            .entries
            .iter()
            .find(|e| e.agent_id == agent)
            .cloned()
    }
    fn upsert_count(&self) -> usize {
        *self.upserts.lock().unwrap()
    }
}
#[async_trait]
impl LiveStateStore for RecordingLiveState {
    async fn load(&self) -> Result<LiveState, StoreError> {
        if self.fail {
            return Err(StoreError::Io("forced failure".to_owned()));
        }
        Ok(self.state.lock().unwrap().clone())
    }
    async fn upsert(&self, entry: LiveEntry) -> Result<(), StoreError> {
        if self.fail {
            return Err(StoreError::Io("forced failure".to_owned()));
        }
        *self.upserts.lock().unwrap() += 1;
        self.state.lock().unwrap().upsert(entry);
        Ok(())
    }
    async fn prune(&self, now_ms: u64, ttl_ms: u64, max_n: usize) -> Result<(), StoreError> {
        if self.fail {
            return Err(StoreError::Io("forced failure".to_owned()));
        }
        self.state.lock().unwrap().prune(now_ms, ttl_ms, max_n);
        Ok(())
    }
}

/// A [`LiveStateProvider`] that always yields the **same** [`UpdateLiveState`] over a
/// shared recording store (root-agnostic for the test).
struct TestLiveStateProvider {
    update: Arc<UpdateLiveState>,
}
impl LiveStateProvider for TestLiveStateProvider {
    fn live_state_for(&self, _root: &ProjectPath) -> Option<Arc<UpdateLiveState>> {
        Some(Arc::clone(&self.update))
    }
}

/// A [`LiveStateReadProvider`] (lot LS4, `idea_workstate_read`) yielding the same
/// [`GetLiveStateLean`] over a shared recording store (root-agnostic for the test).
struct TestLiveStateReadProvider {
    getter: Arc<GetLiveStateLean>,
}
impl LiveStateReadProvider for TestLiveStateReadProvider {
    fn live_state_lean_for(&self, _root: &ProjectPath) -> Option<Arc<GetLiveStateLean>> {
        Some(Arc::clone(&self.getter))
    }
}

/// Fixed millis clock for deterministic `updated_at_ms`.
struct FixedMillisClock(i64);
impl Clock for FixedMillisClock {
    fn now_millis(&self) -> i64 {
        self.0
    }
}

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 run_in_background_dispatch_creates_running_command_for_requester() {
    let fx = fixture(FakeContexts::new());
    let store = Arc::new(RecordingBackgroundStore::default());
    let runner = Arc::new(RecordingBackgroundRunner::default());
    let spawn = Arc::new(SpawnBackgroundCommand::new(
        Arc::clone(&store) as Arc<dyn BackgroundTaskStore>,
        Arc::clone(&runner) as Arc<dyn BackgroundTaskRunner>,
        Arc::new(FixedMillisClock(42)) as Arc<dyn domain::ports::Clock>,
        Arc::new(SeqIds::new()) as Arc<dyn IdGenerator>,
    ));
    let service = fx.service.with_spawn_background_command(spawn);
    let owner = aid(77);

    let out = service
        .dispatch(
            &project(),
            OrchestratorCommand::RunInBackground {
                owner,
                label: "B8 T1".to_owned(),
                command: "bash".to_owned(),
                args: vec![
                    "-lc".to_owned(),
                    "sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
                ],
                cwd: Some("crates".to_owned()),
                deadline_ms: None,
            },
        )
        .await
        .expect("dispatch ok");

    let payload: serde_json::Value = serde_json::from_str(&out.detail).unwrap();
    assert_eq!(payload["state"], "Running");
    let task_id =
        TaskId::from_uuid(uuid::Uuid::parse_str(payload["taskId"].as_str().unwrap()).unwrap());
    let task = store.task(task_id).expect("task persisted");
    assert_eq!(task.project_id, project().id);
    assert_eq!(task.owner_agent_id, owner);
    assert_eq!(task.state, BackgroundTaskState::Running);
    assert_eq!(task.wake_policy, BackgroundTaskWakePolicy::WakeOwner);
    assert_eq!(
        task.kind,
        domain::BackgroundTaskKind::Command {
            label: "B8 T1".to_owned(),
        }
    );

    let specs = runner.specs();
    assert_eq!(specs.len(), 1);
    assert_eq!(specs[0].task_id, task_id);
    assert_eq!(specs[0].owner_agent_id, owner);
    assert_eq!(specs[0].wake_policy, BackgroundTaskWakePolicy::WakeOwner);
    let command = specs[0].command.as_ref().expect("command spec");
    assert_eq!(command.command, "bash");
    assert_eq!(
        command.args,
        vec![
            "-lc".to_owned(),
            "sleep 45 && echo DONE_MARKER_DEVBACKEND_B8_T1".to_owned(),
        ]
    );
    assert_eq!(command.cwd.as_str(), "/home/me/proj/crates");
}

#[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, TurnResolution};
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<TurnResolution>)>>>,
    completions: Arc<CompletionBus>,
}
impl TestMailbox {
    fn new() -> Self {
        Self {
            queues: Mutex::new(HashMap::new()),
            completions: Arc::new(CompletionBus::default()),
        }
    }
    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()
    }
    fn completions(&self) -> Arc<CompletionBus> {
        Arc::clone(&self.completions)
    }
}

#[derive(Clone)]
enum TestCompletion {
    Replied(String),
    NoReply,
    Cancelled,
}

#[derive(Default)]
struct CompletionBus {
    by_agent: Mutex<HashMap<AgentId, VecDeque<TestCompletion>>>,
    notify: tokio::sync::Notify,
}

impl CompletionBus {
    fn push(&self, agent: AgentId, completion: TestCompletion) {
        self.by_agent
            .lock()
            .unwrap()
            .entry(agent)
            .or_default()
            .push_back(completion);
        self.notify.notify_waiters();
    }

    async fn next(&self, agent: AgentId) -> TestCompletion {
        loop {
            if let Some(item) = self
                .by_agent
                .lock()
                .unwrap()
                .get_mut(&agent)
                .and_then(VecDeque::pop_front)
            {
                return item;
            }
            self.notify.notified().await;
        }
    }
}

impl AgentMailbox for TestMailbox {
    fn enqueue(&self, agent: AgentId, ticket: Ticket) -> PendingReply {
        let (tx, rx) = tokio::sync::oneshot::channel::<TurnResolution>();
        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")
        };
        self.completions
            .push(agent, TestCompletion::Replied(result.clone()));
        let _ = slot.1.send(TurnResolution::Replied(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")
        };
        self.completions
            .push(agent, TestCompletion::Replied(result.clone()));
        let _ = slot.1.send(TurnResolution::Replied(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();
                self.completions.push(agent, TestCompletion::Cancelled);
            }
        }
    }
    fn complete_without_reply(&self, agent: AgentId, ticket_id: TicketId) {
        // Mirror the production adapter: head-only, idempotent, fire-and-forget-safe.
        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) {
                return;
            }
            if queue.front().is_some_and(|(_, tx)| tx.is_closed()) {
                return; // receiver gone (human submit / timed-out caller): preserve head.
            }
            let (_, tx) = queue.pop_front().expect("head just matched");
            self.completions.push(agent, TestCompletion::NoReply);
            let _ = tx.send(TurnResolution::ReturnedToPromptNoReply);
        }
    }
}

/// 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>,
    /// The auto-memory store wired into the harvest hook (Lot E1): lets a test
    /// assert which notes a successful reply persisted.
    memories: Arc<HarvestMemories>,
    /// The live-state store wired into the auto-update hook (lot LS3): lets a test
    /// assert the target's Working/Done transitions and the upsert count.
    live: Arc<RecordingLiveState>,
    structured: Arc<StructuredSessions>,
}

struct CompletionSession {
    id: SessionId,
    agent: AgentId,
    completions: Arc<CompletionBus>,
}

#[async_trait]
impl AgentSession for CompletionSession {
    fn id(&self) -> SessionId {
        self.id
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        match self.completions.next(self.agent).await {
            TestCompletion::Replied(content) => {
                Ok(Box::new(vec![ReplyEvent::Final { content }].into_iter()))
            }
            TestCompletion::NoReply => Err(AgentSessionError::Io(
                "target returned without a structured final".to_owned(),
            )),
            TestCompletion::Cancelled => Err(AgentSessionError::Io(
                "structured test turn cancelled".to_owned(),
            )),
        }
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        Ok(())
    }
}

#[derive(Clone)]
struct CompletionFactory {
    completions: Arc<CompletionBus>,
    agents_by_path: HashMap<String, AgentId>,
    next_id: Arc<Mutex<u128>>,
}

impl CompletionFactory {
    fn new(completions: Arc<CompletionBus>, agents_by_path: HashMap<String, AgentId>) -> Self {
        Self {
            completions,
            agents_by_path,
            next_id: Arc::new(Mutex::new(7000)),
        }
    }
}

#[async_trait]
impl AgentSessionFactory for CompletionFactory {
    fn supports(&self, profile: &AgentProfile) -> bool {
        profile.structured_adapter.is_some()
    }
    async fn start(
        &self,
        _profile: &AgentProfile,
        ctx: &PreparedContext,
        _cwd: &ProjectPath,
        _session: &SessionPlan,
        _sandbox: Option<&domain::sandbox::SandboxPlan>,
    ) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
        let id = {
            let mut n = self.next_id.lock().unwrap();
            let id = SessionId::from_uuid(Uuid::from_u128(*n));
            *n += 1;
            id
        };
        let agent = self
            .agents_by_path
            .get(&ctx.relative_path)
            .copied()
            .ok_or_else(|| {
                AgentSessionError::Start(format!(
                    "test completion session cannot resolve agent for {}",
                    ctx.relative_path
                ))
            })?;
        Ok(Arc::new(CompletionSession {
            id,
            agent,
            completions: Arc::clone(&self.completions),
        }))
    }
}

fn completion_factory_for_contexts(
    contexts: &FakeContexts,
    completions: Arc<CompletionBus>,
) -> CompletionFactory {
    let agents_by_path: HashMap<String, AgentId> = contexts
        .manifest()
        .entries
        .iter()
        .map(|e| (e.md_path.clone(), e.agent_id))
        .collect();
    CompletionFactory::new(completions, agents_by_path)
}

fn seed_structured_sessions(
    structured: &StructuredSessions,
    contexts: &FakeContexts,
    completions: Arc<CompletionBus>,
) {
    for (idx, entry) in contexts.manifest().entries.iter().enumerate() {
        let session = Arc::new(CompletionSession {
            id: SessionId::from_uuid(Uuid::from_u128(8_000 + idx as u128)),
            agent: entry.agent_id,
            completions: Arc::clone(&completions),
        }) as Arc<dyn AgentSession>;
        structured.insert(session, entry.agent_id, nid(8_000 + idx as u128));
    }
}

/// 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 {
    ask_fixture_full(contexts, false, false, true)
}

/// Like [`ask_fixture`] but lets a test force the auto-memory store to fail its
/// `save`, proving the harvest stays best-effort (the reply still succeeds).
fn ask_fixture_with_memory(contexts: FakeContexts, fail_memory: bool) -> AskFixture {
    ask_fixture_full(contexts, fail_memory, false, true)
}

/// Like [`ask_fixture`] but lets a test force the live-state store to fail, proving
/// the auto-update stays best-effort (the ask/reply still succeeds).
fn ask_fixture_with_live(contexts: FakeContexts, fail_live: bool) -> AskFixture {
    ask_fixture_full(contexts, false, fail_live, true)
}

/// Like [`ask_fixture`] but leaves the live-state auto-update **unwired** (`None`),
/// to prove no write happens when the hook is absent (zéro régression).
fn ask_fixture_without_live(contexts: FakeContexts) -> AskFixture {
    ask_fixture_full(contexts, false, false, false)
}

/// Full builder: wires the auto-memory harvest (Lot E1) and, when `wire_live`, the
/// live-state auto-update (lot LS3), each with an optional forced failure.
fn ask_fixture_full(
    contexts: FakeContexts,
    fail_memory: bool,
    fail_live: bool,
    wire_live: bool,
) -> 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 structured = Arc::new(StructuredSessions::new());
    let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
    seed_structured_sessions(&structured, &contexts, mailbox.completions());
    let memories = Arc::new(HarvestMemories {
        saved: Mutex::new(Vec::new()),
        fail_save: fail_memory,
    });
    let live = Arc::new(RecordingLiveState {
        fail: fail_live,
        ..Default::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,
        )
        .with_structured(
            Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let 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 mut builder = 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_structured(Arc::clone(&structured))
    .with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
        Arc::clone(&memories) as Arc<dyn MemoryStore>,
        Arc::new(bus.clone()),
    )));
    if wire_live {
        builder = builder.with_live_state(Arc::new(TestLiveStateProvider {
            update: Arc::new(UpdateLiveState::new(
                Arc::clone(&live) as Arc<dyn LiveStateStore>,
                Arc::new(FixedMillisClock(1_234)) as Arc<dyn Clock>,
            )),
        }) as Arc<dyn LiveStateProvider>);
        // LS4 read side over the SAME store, so `idea_workstate_set` then
        // `idea_workstate_read` round-trips (no separate write provider would).
        builder = builder.with_live_state_read(Arc::new(TestLiveStateReadProvider {
            getter: Arc::new(GetLiveStateLean::new(
                Arc::clone(&live) as Arc<dyn LiveStateStore>,
                Arc::new(FixedMillisClock(1_234)) as Arc<dyn Clock>,
            )),
        }) as Arc<dyn LiveStateReadProvider>);
    }
    let service = Arc::new(builder);

    AskFixture {
        service,
        mailbox,
        sessions,
        bus,
        pty,
        mediator,
        memories,
        live,
        structured,
    }
}

/// 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 structured target: `ask` enqueues a bookkeeping ticket, sends the turn to the
/// headless session, and returns the session's `Final`. The PTY is not used.
#[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 PTY is no longer part of the inter-agent conversation path.
    let writes = fx.pty.writes_for(sid(800));
    assert_eq!(
        writes.len(),
        0,
        "inter-agent conversation must not write to the PTY"
    );
    // 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");
}

// --- LS3: live-state auto-update from the delegation cycle ------------------

/// A successful `ask` flips the **target** to `Working` with the distilled intent and
/// the current ticket — derived from the existing delegation transition (zéro token).
#[tokio::test]
async fn ask_sets_target_working_with_intent_and_ticket() {
    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 });

    // The Working transition is posted right after the enqueue, before any reply.
    await_until(|| fx.live.entry_for(aid(1)).is_some()).await;
    let entry = fx
        .live
        .entry_for(aid(1))
        .expect("target live entry present");
    assert_eq!(entry.status, WorkStatus::Working);
    assert_eq!(
        entry.intent, "Analyse §17",
        "intent distilled from the task"
    );
    assert!(entry.ticket.is_some(), "current ticket recorded");
    assert!(
        entry.last_delegation.is_none(),
        "no delegation resolved yet"
    );
    // The live-state path does not touch the memory store (canonical effects separate).
    assert!(
        fx.memories.slugs().is_empty(),
        "no memory write on this path"
    );

    // Drain the ask so the spawned task completes cleanly.
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "done"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");
}

/// A successful `reply` flips the target to `Done` and records `last_delegation` = the
/// resolved ticket.
#[tokio::test]
async fn reply_sets_target_done_with_last_delegation() {
    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;

    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "the answer"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");

    let entry = fx
        .live
        .entry_for(aid(1))
        .expect("target live entry present");
    assert_eq!(
        entry.status,
        WorkStatus::Done,
        "target marked Done on reply"
    );
    assert!(
        entry.last_delegation.is_some(),
        "last_delegation set to the resolved ticket"
    );
    // Working then Done ⇒ exactly two upserts for one delegation (no write-storm).
    assert_eq!(
        fx.live.upsert_count(),
        2,
        "one Working + one Done write per delegation"
    );
}

/// A live-state store that fails must NOT fail the `ask`/`reply`: the delegation
/// still returns its reply (best-effort hook, exactly like the memory harvest).
#[tokio::test]
async fn live_state_failure_does_not_break_delegation() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture_with_live(FakeContexts::with_agent(&agent, "# persona"), true);
    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), "still works"))
        .await
        .expect("reply ok despite failing live-state");
    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok despite failing live-state");
    assert_eq!(out.reply.as_deref(), Some("still works"));
    // The failing store recorded nothing.
    assert!(fx.live.entry_for(aid(1)).is_none());
}

/// When the live-state hook is **unwired** (`None`), the delegation behaves exactly as
/// before — no live-state write at all (zéro régression).
#[tokio::test]
async fn no_live_state_wired_means_no_write() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture_without_live(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;
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "ok"))
        .await
        .expect("reply ok");
    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");

    assert_eq!(fx.live.upsert_count(), 0, "no write when hook is None");
    assert!(fx.live.entry_for(aid(1)).is_none());
}

// ---------------------------------------------------------------------------
// LS4 — `idea_workstate_set` / `idea_workstate_read` service round-trip.
// ---------------------------------------------------------------------------

/// `idea_workstate_set` writes the current agent's row, then `idea_workstate_read`
/// returns it enriched with the **resolved display name** (via `ListAgents`), with the
/// declared status/intent — and **never** the `progress` field.
#[tokio::test]
async fn workstate_set_then_read_round_trips_with_resolved_name_and_no_progress() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    // set — keyed on the handshake agent id; carries a progress note (must not leak).
    let ack = fx
        .service
        .dispatch(
            &project(),
            OrchestratorCommand::SetWorkState {
                agent: aid(1),
                status: WorkStatus::Working,
                intent: Some("ship LS4".to_owned()),
                progress: Some("secret internal note".to_owned()),
                ticket: None,
                last_delegation: None,
            },
        )
        .await
        .expect("set ok");
    // ACK only — no inline reply on the write path.
    assert!(ack.reply.is_none(), "set is ACK only, no reply: {ack:?}");

    // read — the JSON array carries one row for the architect.
    let out = fx
        .service
        .dispatch(
            &project(),
            OrchestratorCommand::ReadWorkState {
                requester: ConversationParty::User,
            },
        )
        .await
        .expect("read ok");
    let json: serde_json::Value =
        serde_json::from_str(out.reply.as_deref().expect("read carries a reply")).unwrap();
    let rows = json.as_array().expect("array");
    assert_eq!(rows.len(), 1, "one live row expected: {json}");
    let row = &rows[0];
    // Name resolved from the manifest (not the bare UUID).
    assert_eq!(row["agent"], serde_json::json!("architect"));
    assert_eq!(row["status"], serde_json::json!("working"));
    assert_eq!(row["intent"], serde_json::json!("ship LS4"));
    // progress is NEVER exposed on read (the lean DTO has no such field).
    assert!(
        row.get("progress").is_none(),
        "progress must never surface on read: {row}"
    );
}

/// `idea_workstate_set` is last-writer-wins: a second set for the same agent replaces
/// the row in place (never a duplicate), and the read reflects the latest write.
#[tokio::test]
async fn workstate_set_is_last_writer_wins() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    for (status, intent) in [
        (WorkStatus::Working, "first"),
        (WorkStatus::Blocked, "second"),
    ] {
        fx.service
            .dispatch(
                &project(),
                OrchestratorCommand::SetWorkState {
                    agent: aid(1),
                    status,
                    intent: Some(intent.to_owned()),
                    progress: None,
                    ticket: None,
                    last_delegation: None,
                },
            )
            .await
            .expect("set ok");
    }

    let out = fx
        .service
        .dispatch(
            &project(),
            OrchestratorCommand::ReadWorkState {
                requester: ConversationParty::User,
            },
        )
        .await
        .expect("read ok");
    let json: serde_json::Value = serde_json::from_str(out.reply.as_deref().unwrap()).unwrap();
    let rows = json.as_array().unwrap();
    assert_eq!(rows.len(), 1, "LWW ⇒ exactly one row, no duplicate: {json}");
    assert_eq!(rows[0]["status"], serde_json::json!("blocked"));
    assert_eq!(rows[0]["intent"], serde_json::json!("second"));
}

/// `idea_workstate_read` drops rows whose agent left (or never was in) the manifest —
/// the manifest boundary is authoritative.
#[tokio::test]
async fn workstate_read_drops_off_manifest_rows() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    // aid(1) is in the manifest; aid(2) is NOT.
    for who in [aid(1), aid(2)] {
        fx.service
            .dispatch(
                &project(),
                OrchestratorCommand::SetWorkState {
                    agent: who,
                    status: WorkStatus::Working,
                    intent: Some("x".to_owned()),
                    progress: None,
                    ticket: None,
                    last_delegation: None,
                },
            )
            .await
            .expect("set ok");
    }

    let out = fx
        .service
        .dispatch(
            &project(),
            OrchestratorCommand::ReadWorkState {
                requester: ConversationParty::User,
            },
        )
        .await
        .expect("read ok");
    let json: serde_json::Value = serde_json::from_str(out.reply.as_deref().unwrap()).unwrap();
    let rows = json.as_array().unwrap();
    assert_eq!(rows.len(), 1, "the off-manifest row is dropped: {json}");
    assert_eq!(rows[0]["agent"], serde_json::json!("architect"));
}

/// An oversize `intent` (above the domain hard byte threshold) is **rejected** by the
/// write path (the domain bound surfaces as `AppError::Invalid`), never silently stored.
#[tokio::test]
async fn workstate_set_oversize_intent_is_rejected() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    let huge = "x".repeat(domain::live_state::FIELD_MAX_BYTES + 1);
    let err = fx
        .service
        .dispatch(
            &project(),
            OrchestratorCommand::SetWorkState {
                agent: aid(1),
                status: WorkStatus::Working,
                intent: Some(huge),
                progress: None,
                ticket: None,
                last_delegation: None,
            },
        )
        .await
        .expect_err("oversize intent must be rejected");
    assert!(
        matches!(err, application::AppError::Invalid(_)),
        "oversize ⇒ AppError::Invalid, got {err:?}"
    );
}

/// The target returned to its prompt **without** `idea_reply`: the mediator's grace
/// window expired and completed the turn "no reply". The awaiting ask must error out
/// promptly with the typed, retryable [`AppError::TargetReturnedNoReply`] — never hang
/// until the long safety timeout. (The grace **timer** lives in the infrastructure
/// mediator; here we drive the resulting mailbox completion directly.)
#[tokio::test]
async fn ask_target_returns_to_prompt_without_reply_is_a_typed_error() {
    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;

    // Simulate the grace-window completion (target back at prompt, no idea_reply).
    let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
    fx.mailbox.complete_without_reply(aid(1), ticket);

    let err = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes promptly, not after the long timeout")
        .expect("join ok")
        .expect_err("a no-reply turn is a typed error");
    assert_eq!(err.code(), "TARGET_RETURNED_NO_REPLY", "got {err:?}");
    assert_eq!(fx.mailbox.pending(&aid(1)), 0, "head retired by completion");
}

/// An `idea_reply` that lands **first** still wins over a (later) no-reply completion:
/// the completion then finds the head gone and is a no-op (idempotent race).
#[tokio::test]
async fn ask_reply_wins_then_late_completion_is_noop() {
    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;

    let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
    // idea_reply resolves the head first…
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "the real answer"))
        .await
        .expect("reply ok");
    // …then a late grace completion for the same ticket is a no-op.
    fx.mailbox.complete_without_reply(aid(1), ticket);

    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");
    assert_eq!(out.reply.as_deref(), Some("the real answer"));
}

// --- Lot E1: auto-memory harvest on a successful ask reply -----------------

/// A reply carrying an `idea-memory` block is harvested **after** the reply
/// resolves: the note is persisted via the store and a `MemorySaved` event fires.
#[tokio::test]
async fn ask_reply_with_directive_harvests_memory_after_reply() {
    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;

    let reply = "Here is the answer.\n\n```idea-memory\nslug: section-17-rule\ntitle: §17 rule\ntype: project\ndescription: How §17 routing works\n---\nRoute structured asks straight to the session.\n```";
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), reply))
        .await
        .expect("reply ok");

    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");
    // The reply is returned unchanged …
    assert_eq!(out.reply.as_deref(), Some(reply));
    // … and the embedded note was harvested into the store + announced.
    assert_eq!(fx.memories.slugs(), vec!["section-17-rule"]);
    let saved: Vec<_> = fx
        .bus
        .events()
        .into_iter()
        .filter_map(|e| match e {
            DomainEvent::MemorySaved { slug } => Some(slug.as_str().to_owned()),
            _ => None,
        })
        .collect();
    assert!(
        saved.contains(&"section-17-rule".to_owned()),
        "MemorySaved announced, got {saved:?}"
    );
}

/// A store failure during harvest must not break the reply: the ask still returns
/// its content, and nothing is persisted.
#[tokio::test]
async fn ask_reply_harvest_store_failure_does_not_break_reply() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = ask_fixture_with_memory(FakeContexts::with_agent(&agent, "# persona"), true);
    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;

    let reply = "Reply.\n\n```idea-memory\nslug: doomed-note\ntitle: T\ntype: project\ndescription: hook\n---\nBody.\n```";
    fx.service
        .dispatch(&project(), reply_cmd(aid(1), reply))
        .await
        .expect("reply ok despite failing memory store");

    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask still ok despite harvest store failure");
    assert_eq!(
        out.reply.as_deref(),
        Some(reply),
        "reply unaffected by harvest"
    );
    assert!(
        fx.memories.slugs().is_empty(),
        "failed save persisted nothing"
    );
}

/// A reply with no directive harvests nothing (and the reply is unchanged) — the
/// common case must not touch the memory store.
#[tokio::test]
async fn ask_reply_without_directive_harvests_nothing() {
    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;

    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "Just a plain answer."))
        .await
        .expect("reply ok");

    timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok")
        .expect("ask ok");
    assert!(fx.memories.slugs().is_empty());
}

/// A cancelled turn (head ticket dropped before any reply) yields no reply, so the
/// harvest never runs — nothing is persisted even with a would-be directive pending.
#[tokio::test]
async fn ask_cancelled_turn_does_not_harvest() {
    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;

    // Cancel the head ticket (drops the reply sender) ⇒ the ask resolves as a
    // channel-closed error: no Response turn, hence no harvest.
    let ticket = fx.mailbox.ticket_ids(&aid(1))[0];
    fx.mailbox.cancel_head(aid(1), ticket);

    let out = timeout(TEST_GUARD, ask)
        .await
        .expect("ask completes")
        .expect("join ok");
    assert!(out.is_err(), "cancelled turn ⇒ ask errors");
    assert!(fx.memories.slugs().is_empty(), "no harvest on cancel");
}

/// 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"
    );
}

/// Régression (garde RAII de fin de tour) — LE test décisif : un `ask_agent` dont le
/// **futur est abandonné (drop)** alors qu'il attendait la réponse doit laisser la cible
/// `Idle`, **pas** `Busy` à vie. Avant le fix, aucune branche du `match`/`select!` ne
/// s'exécutait sur un drop ⇒ l'agent restait `Busy` et toute délégation suivante était
/// mise en file derrière ce tour fantôme sans jamais être livrée.
#[tokio::test]
async fn dropped_ask_future_frees_busy_target() {
    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 });
    // Le tour a démarré : la cible est Busy, un ticket est en file.
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    assert!(
        fx.mediator.busy_state(aid(1)).is_busy(),
        "cible Busy pendant le tour délégué"
    );

    // Le demandeur interrompt son appel ⇒ le futur `ask_agent` est dropped.
    ask.abort();
    let _ = ask.await; // récolte la JoinError(Cancelled), ignorée.

    // Le garde RAII a ramené la cible Idle ET retiré le ticket fantôme de la FIFO.
    await_until(|| !fx.mediator.busy_state(aid(1)).is_busy()).await;
    assert_eq!(
        fx.mediator.busy_state(aid(1)),
        AgentBusyState::Idle,
        "futur dropped ⇒ la cible est ramenée Idle par le garde (fix cause racine)"
    );
    assert_eq!(
        fx.mailbox.pending(&aid(1)),
        0,
        "ticket fantôme retiré de la FIFO au Drop du garde"
    );
}

/// Régression (garde RAII) — une **2e délégation** vers une cible dont un 1er `ask` a
/// été abandonné est bien livrée : l'agent n'est pas coincé `Busy`, son tour suivant
/// démarre et peut être résolu par `idea_reply`.
#[tokio::test]
async fn second_delegation_delivered_after_dropped_ask() {
    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));

    // 1er ask : démarre puis est abandonné (drop).
    let svc = Arc::clone(&fx.service);
    let ask1 = tokio::spawn(async move { svc.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    ask1.abort();
    let _ = ask1.await;
    await_until(|| fx.mailbox.pending(&aid(1)) == 0).await;
    let _ = fx.mailbox.completions().next(aid(1)).await;

    // 2e ask vers la MÊME cible : doit démarrer un nouveau tour (pas coincé derrière un
    // tour fantôme) et se laisser résoudre.
    let svc2 = Arc::clone(&fx.service);
    let ask2 = tokio::spawn(async move { svc2.dispatch(&project(), cmd(ASK_JSON)).await });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    assert!(
        fx.mediator.busy_state(aid(1)).is_busy(),
        "le 2e tour démarre bien (cible Busy) — preuve qu'elle n'était pas coincée"
    );

    fx.service
        .dispatch(&project(), reply_cmd(aid(1), "réponse au 2e tour"))
        .await
        .expect("reply ok");
    let out = timeout(TEST_GUARD, ask2)
        .await
        .expect("le 2e ask se termine")
        .expect("join ok")
        .expect("ask ok");
    assert_eq!(out.reply.as_deref(), Some("réponse au 2e tour"));
    assert_eq!(
        fx.mediator.busy_state(aid(1)),
        AgentBusyState::Idle,
        "cible Idle après résolution du 2e tour"
    );
}

/// Régression (garde RAII) — la **branche erreur/timeout** (canal fermé) ramène aussi la
/// cible `Idle`. Avant le fix, ces branches faisaient `cancel_head` mais jamais
/// `mark_idle` ⇒ l'agent restait `Busy`. On déclenche la fermeture du canal en retirant
/// le ticket de tête (même nettoyage que le timeout de tour).
#[tokio::test]
async fn cancelled_ask_marks_target_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;
    assert!(fx.mediator.busy_state(aid(1)).is_busy());

    // Retire le ticket de tête (drop du sender) ⇒ l'ask voit un canal fermé et part en
    // erreur typée (PROCESS), le MÊME nettoyage que le timeout de tour.
    let t = fx.mailbox.ticket_ids(&aid(1))[0];
    fx.mailbox.cancel_head(aid(1), t);
    let err = timeout(TEST_GUARD, ask)
        .await
        .expect("ask retourne vite sur canal fermé")
        .expect("join ok")
        .unwrap_err();
    assert!(
        matches!(err.code(), "PROCESS" | "INVALID"),
        "ask sur canal fermé ⇒ erreur typée : {err:?}"
    );
    // Le garde a ramené la cible Idle (la FIFO peut avancer).
    assert_eq!(
        fx.mediator.busy_state(aid(1)),
        AgentBusyState::Idle,
        "branche erreur ⇒ cible Idle (garde RAII)"
    );
}

/// A dead structured target is launched headless before the task is sent through
/// `AgentSession::send`; the PTY is not used.
#[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());
    assert!(fx.structured.session_for_agent(&aid(1)).is_some());

    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;
    assert!(fx.pty.spawns().is_empty(), "AskAgent does not spawn a PTY");
    assert!(
        fx.pty.writes_for(sid(777)).is_empty(),
        "AskAgent does not write into a PTY"
    );

    fx.mailbox
        .resolve(aid(1), "launched reply".to_owned())
        .expect("structured Final");
    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 {
    /// 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 structured = Arc::new(StructuredSessions::new());
    let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());

    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,
        )
        .with_structured(
            Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let 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()))
    .with_structured(Arc::clone(&structured));

    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 via le Final structured.
    fx.mailbox
        .resolve(aid(1), "done".to_owned())
        .expect("structured completion 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.mailbox
        .resolve(aid(1), "done".to_owned())
        .expect("structured completion ok");
    timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
}

/// F2 — GARDE (passant en mode structured). Une cible Codex passe si l'adaptateur
/// structured est câblé par la fixture : pas de fallback PTY, réponse par `Final`.
#[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 svc = Arc::clone(&fx.service);
    let ask = tokio::spawn(async move {
        svc.dispatch(
            &project(),
            cmd(r#"{ "type":"agent.message", "targetAgent":"coder", "task":"refactor" }"#),
        )
        .await
    });
    await_until(|| fx.mailbox.pending(&aid(1)) == 1).await;
    fx.mailbox
        .resolve(aid(1), "codex ok".to_owned())
        .expect("structured completion ok");
    let out = timeout(TEST_GUARD, ask).await.unwrap().unwrap().unwrap();
    assert_eq!(out.reply.as_deref(), Some("codex ok"));
    assert!(fx.pty.spawns().is_empty(), "no PTY fallback");
}

/// 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.mailbox
        .resolve(aid(1), "ok claude".to_owned())
        .expect("structured completion 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 structured = Arc::new(StructuredSessions::new());
    let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    let launch = Arc::new(
        LaunchAgent::new(
            Arc::new(contexts.clone()),
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::new(FakeRuntime),
            Arc::new(FakeFs),
            Arc::new(pty.clone()),
            Arc::new(FakeSkills),
            Arc::clone(&sessions),
            Arc::new(bus.clone()),
            Arc::new(SeqIds::new()),
            Arc::new(FakeRecall),
            None,
        )
        .with_structured(
            Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let 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()))
        .with_structured(Arc::clone(&structured)),
    );

    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 structured = Arc::new(StructuredSessions::new());
    let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
    seed_structured_sessions(&structured, &contexts, mailbox.completions());

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    let launch = Arc::new(
        LaunchAgent::new(
            Arc::new(contexts.clone()),
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::new(FakeRuntime),
            Arc::new(FakeFs),
            Arc::new(pty.clone()),
            Arc::new(FakeSkills),
            Arc::clone(&sessions),
            Arc::new(bus.clone()),
            Arc::new(SeqIds::new()),
            Arc::new(FakeRecall),
            None,
        )
        .with_structured(
            Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let 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_structured(Arc::clone(&structured)),
    );

    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 structured = Arc::new(StructuredSessions::new());
    let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
    seed_structured_sessions(&structured, &contexts, mailbox.completions());

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    let launch = Arc::new(
        LaunchAgent::new(
            Arc::new(contexts.clone()),
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::new(FakeRuntime),
            Arc::new(FakeFs),
            Arc::new(pty.clone()),
            Arc::new(FakeSkills),
            Arc::clone(&sessions),
            Arc::new(bus.clone()),
            Arc::new(SeqIds::new()),
            Arc::new(FakeRecall),
            None,
        )
        .with_structured(
            Arc::new(completion_factory) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(pty.clone()),
        Arc::clone(&sessions),
    ));
    let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone())));
    let 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_structured(Arc::clone(&structured))
        .with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
    );

    AskFixture {
        service,
        mailbox,
        sessions,
        bus,
        pty,
        mediator,
        // Harvest is intentionally not wired here: these fixtures exercise the
        // RecordTurn checkpoint in isolation (proving it stays unchanged).
        memories: Arc::new(HarvestMemories::default()),
        // Live-state likewise unwired here (RecordTurn-focused fixtures).
        live: Arc::new(RecordingLiveState::default()),
        structured,
    }
}

/// 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.mailbox
        .resolve(reply_from, result.to_owned())
        .expect("structured completion 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"
    );
}

// ===========================================================================
// Lot 1b — auto-lancement d'une cible **froide** structurée sur le chemin `ask`
// ===========================================================================
//
// Avant le Lot 1b, `ask_agent` ne routait vers `ask_structured` que si la cible avait
// **déjà** une session structurée vivante ; une cible structurée froide tombait dans
// `ensure_live_pty` (chemin PTY) — or une cible structurée n'a pas de PTY, donc le tour
// restait bloqué `Busy` (débloquable seulement par un `idea_reply` explicite). Le Lot 1b
// démarre la session structurée via le **même** launcher que le chemin PTY (qui route
// §17.4 vers `launch_structured` et l'insère dans le registre **partagé**), puis route
// vers `ask_structured` : le `Final` déterministe de la session débloque le tour et
// marque l'agent `Idle` (`mark_idle`), **sans** `idea_reply` ni PTY.

use std::sync::atomic::{AtomicUsize, Ordering};

use application::StructuredSessions;
use domain::ports::{
    AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream,
};

/// Session structurée fake : `send()` rend un unique [`ReplyEvent::Final`] qui **écho**
/// le prompt reçu — c'est ce `Final` que `drain_with_readiness` transforme en réponse
/// du tour **et** en `mark_idle`. Aucun `idea_reply` n'est nécessaire.
struct EchoSession {
    id: SessionId,
}
#[async_trait]
impl AgentSession for EchoSession {
    fn id(&self) -> SessionId {
        self.id
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        let stream: ReplyStream = Box::new(
            vec![ReplyEvent::Final {
                content: format!("structured: {prompt}"),
            }]
            .into_iter(),
        );
        Ok(stream)
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        Ok(())
    }
}

/// Fabrique fake : `supports` = profil structuré, et `start` **compte** les démarrages
/// (preuve qu'une session froide a bien été auto-lancée) en rendant une [`EchoSession`].
#[derive(Clone, Default)]
struct CountingFactory {
    starts: Arc<AtomicUsize>,
    next_id: Arc<Mutex<u128>>,
}
impl CountingFactory {
    fn new() -> Self {
        Self {
            starts: Arc::new(AtomicUsize::new(0)),
            next_id: Arc::new(Mutex::new(9000)),
        }
    }
    fn start_count(&self) -> usize {
        self.starts.load(Ordering::SeqCst)
    }
}
#[async_trait]
impl AgentSessionFactory for CountingFactory {
    fn supports(&self, profile: &AgentProfile) -> bool {
        profile.structured_adapter.is_some()
    }
    async fn start(
        &self,
        _profile: &AgentProfile,
        _ctx: &PreparedContext,
        _cwd: &ProjectPath,
        _session: &SessionPlan,
        _sandbox: Option<&domain::sandbox::SandboxPlan>,
    ) -> Result<Arc<dyn AgentSession>, AgentSessionError> {
        self.starts.fetch_add(1, Ordering::SeqCst);
        let id = {
            let mut n = self.next_id.lock().unwrap();
            let id = SessionId::from_uuid(Uuid::from_u128(*n));
            *n += 1;
            id
        };
        Ok(Arc::new(EchoSession { id }))
    }
}

/// Tout le câblage `ask` **structuré** : launcher + service voient le **même** registre
/// `StructuredSessions`, et le launcher porte la fabrique (routage §17.4).
struct StructuredAskFixture {
    service: Arc<OrchestratorService>,
    structured: Arc<StructuredSessions>,
    sessions: Arc<TerminalSessions>,
    factory: CountingFactory,
    pty: FakePty,
}

fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture {
    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 structured = Arc::new(StructuredSessions::new());
    let factory = CountingFactory::new();

    let create = Arc::new(CreateAgentFromScratch::new(
        Arc::new(contexts.clone()),
        Arc::new(SeqIds::new()),
        Arc::new(bus.clone()),
    ));
    // Launcher câblé **structuré** : pour un profil à `structured_adapter`, `execute`
    // route §17.4 → `launch_structured`, démarre la session via la fabrique et l'insère
    // dans CE registre partagé (le même `Arc` que le service ci-dessous).
    let launch = Arc::new(
        LaunchAgent::new(
            Arc::new(contexts.clone()),
            Arc::clone(&profiles) as Arc<dyn ProfileStore>,
            Arc::new(FakeRuntime),
            Arc::new(FakeFs),
            Arc::new(pty.clone()),
            Arc::new(FakeSkills),
            Arc::clone(&sessions),
            Arc::new(bus.clone()),
            Arc::new(SeqIds::new()),
            Arc::new(FakeRecall),
            None,
        )
        .with_structured(
            Arc::new(factory.clone()) as Arc<dyn AgentSessionFactory>,
            Arc::clone(&structured),
        ),
    );
    let list = Arc::new(ListAgents::new(Arc::new(contexts.clone())));
    let close = Arc::new(CloseTerminal::new(
        Arc::new(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()))
        // Même registre que le launcher : `ask_agent` y relit la session fraîchement
        // insérée par `launch_structured`.
        .with_structured(Arc::clone(&structured)),
    );

    StructuredAskFixture {
        service,
        structured,
        sessions,
        factory,
        pty,
    }
}

/// Cible **froide** structurée (adaptateur Claude, aucune session vivante) : `ask_agent`
/// auto-lance sa session structurée via le launcher (factory.start appelé **une** fois),
/// puis la draine — le `Final` débloque le round-trip **sans** `idea_reply` ni PTY.
#[tokio::test]
async fn ask_cold_structured_target_autolaunches_session_and_final_unblocks() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    // Pré-condition : aucune session structurée vivante pour la cible (elle est froide).
    assert!(
        fx.structured.session_for_agent(&aid(1)).is_none(),
        "cible structurée froide : aucune session avant l'ask"
    );

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

    // Le `Final` de la session a rendu le contenu — sans aucun `idea_reply` dispatché.
    assert_eq!(
        out.reply.as_deref(),
        Some("structured: Analyse §17"),
        "le Final de la session auto-lancée débloque le tour"
    );
    // Exactement **un** démarrage de session structurée (la cible froide a été lancée).
    assert_eq!(
        fx.factory.start_count(),
        1,
        "une session structurée a été auto-lancée pour la cible froide"
    );
    // La session est désormais vivante dans le registre partagé (insérée par le launcher).
    assert!(
        fx.structured.session_for_agent(&aid(1)).is_some(),
        "la session auto-lancée est enregistrée dans StructuredSessions"
    );
    // Chemin structuré ⇒ **aucun** PTY spawné (la cible n'a pas de terminal).
    assert!(
        fx.pty.spawns().is_empty(),
        "cible structurée : aucun PTY spawné (chemin chat, pas terminal)"
    );
}

/// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude)
/// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve
/// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin).
#[tokio::test]
async fn ask_warm_structured_target_reuses_session_without_relaunch() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));

    // 1er ask : auto-lance (start_count == 1).
    let _ = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("first ask ok");
    assert_eq!(fx.factory.start_count(), 1, "1er ask auto-lance la session");

    // 2e ask : cible chaude ⇒ aucune relance.
    let out = fx
        .service
        .dispatch(&project(), cmd(ASK_JSON))
        .await
        .expect("second ask ok");
    assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
    assert_eq!(
        fx.factory.start_count(),
        1,
        "cible chaude : la session est réutilisée, aucune relance"
    );
}

/// Régression live : une cible structurée peut déjà être vivante en PTY parce qu'elle
/// a été ouverte depuis la surface humaine/menu historique. Le chemin `ask` headless
/// doit alors créer une session structurée **sans fermer** ce PTY visible : les deux
/// canaux d'entrée coexistent, et la délégation passe par `AgentSession::send`.
#[tokio::test]
async fn ask_structured_target_live_as_pty_keeps_visible_terminal() {
    let agent = scratch_agent(aid(1), "architect", "agents/architect.md");
    let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona"));
    seed_live_pty(&fx.sessions, aid(1), sid(800));

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

    assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17"));
    assert_eq!(fx.factory.start_count(), 1, "headless session started");
    assert!(fx.pty.kills().is_empty(), "visible PTY must not be stopped");
    assert!(
        fx.structured.session_for_agent(&aid(1)).is_some(),
        "target now has a structured session"
    );
    assert!(
        fx.sessions.session_for_agent(&aid(1)).is_some(),
        "target still has its visible PTY session"
    );
}