cirun-agent 0.6.0

//! HTTP client + provisioning orchestrator. Polls the cirun api for runners
//! to provision/delete, dispatches via the executor registry, tracks retries
//! and per-runner executor binding.

use crate::api::{AgentInfo, ApiResponse, RunnerToProvision};
use crate::provision::{provision_single_runner, ProvisionResult};
use log::{debug, error, info, warn};
use reqwest::{Client, Error};
use serde_json::json;
use std::collections::HashMap;
use std::env;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use uuid::Uuid;

// Client for interacting with the CiRun API
pub struct CirunClient {
    client: Client,
    base_url: String,
    api_token: String,
    agent: AgentInfo,
    /// Per-runner attempt counter. Mutexed for the same reason `in_flight`
    /// is: the `ProvisionReporter` impl takes `&self`, so internal state
    /// it mutates needs interior mutability. Lock contention is minimal —
    /// only one task touches the map at a time in practice (sequential
    /// `try_join_next` drain).
    pub retry_tracker: std::sync::Mutex<HashMap<String, u32>>,
    /// None means no limit, Some(n) means max n concurrent VMs
    max_runners: Option<u32>,
    /// Per-runner executor binding, learned at provision time. Cleanup/delete
    /// paths consult this map instead of guessing from env vars.
    /// Mutex (not just inner map) because lookups happen behind `&self` and
    /// inserts behind `&mut self` from the main loop.
    pub runner_executors: std::sync::Mutex<HashMap<String, crate::executor::ExecutorKind>>,
    /// Names of runners currently being provisioned by this agent. Used to
    /// suppress racing delete requests: if the api sends a
    /// `runners_to_delete` for a name still in `in_flight`, the agent ignores
    /// it instead of killing the half-built VM mid-spawn.
    pub in_flight: std::sync::Mutex<std::collections::HashSet<String>>,
    /// Executors available on this host, probed once at startup. `Arc` so it
    /// can be cheaply cloned into per-runner provision tasks.
    pub registry: Arc<crate::executor::registry::Registry>,
}

impl CirunClient {
    pub fn new(
        base_url: &str,
        api_token: &str,
        agent: AgentInfo,
        max_runners: Option<u32>,
        executor_filter: &crate::executor::ExecutorFilter,
    ) -> Self {
        let client = Client::builder()
            .timeout(Duration::from_secs(15))
            .connect_timeout(Duration::from_secs(10))
            .build()
            .expect("Failed to build HTTP client");

        CirunClient {
            client,
            base_url: base_url.to_string(),
            api_token: api_token.to_string(),
            agent,
            retry_tracker: std::sync::Mutex::new(HashMap::new()),
            max_runners,
            runner_executors: std::sync::Mutex::new(HashMap::new()),
            in_flight: std::sync::Mutex::new(std::collections::HashSet::new()),
            registry: Arc::new(crate::executor::registry::Registry::probe_filtered(
                executor_filter,
            )),
        }
    }

    /// Whether this agent can take a given runner. Three conditions must
    /// hold: the runner's executor must resolve, the agent must actually
    /// have that executor registered, and the executor must produce the
    /// requested `runner.os`.
    ///
    /// The old gate was `runner.os == agent.os`, which silently dropped
    /// linux jobs from macOS agents running Docker Desktop (issue #14).
    /// The new gate ignores the host OS and asks the executor what OS it
    /// can serve — Docker on macOS can serve linux runners, meda is
    /// linux-only, lume is macOS-only.
    ///
    /// Why the gate still matters: when the api fans the same runner out
    /// by executor capability alone, a mismatched dispatch would attempt
    /// provision, fail (image arch mismatch, CPU bounds, etc.), and the
    /// racing failure notification would orphan-delete the agent that did
    /// accept the work.
    fn is_runner_dispatchable(&self, r: &RunnerToProvision) -> bool {
        let kind = match crate::executor::resolve_executor_kind(
            r.executor.as_deref(),
            r.extra_config.as_ref(),
            &r.os,
        ) {
            Ok(k) => k,
            // Let the provision flow surface the misconfig with full context.
            Err(_) => return true,
        };
        if self.registry.get(kind).is_err() {
            debug!(
                "Skipping runner '{}' — executor {:?} not available on this host",
                r.name, kind
            );
            return false;
        }
        if !crate::executor::executor_serves_os(kind, &r.os) {
            debug!(
                "Skipping runner '{}' — executor {:?} cannot serve runner.os={}",
                r.name, kind, r.os
            );
            return false;
        }
        true
    }

    /// Best-effort lookup of which executor owns a runner. Falls back to the
    /// OS-default when the map has no entry (e.g. agent restarted between
    /// provision and cleanup, or runner was created by an older agent build).
    /// `seed_runner_executors_from_registry` should be called at startup to
    /// reduce the chance of the fallback firing on a fresh process.
    fn executor_for_runner(&self, runner_name: &str) -> crate::executor::ExecutorKind {
        if let Ok(map) = self.runner_executors.lock() {
            if let Some(k) = map.get(runner_name) {
                return *k;
            }
        }
        // Log the fallback so an operator can spot it in journald — silent
        // mis-routing was a real risk per the round-1 review.
        warn!(
            "runner '{}' has no executor binding; falling back to OS default",
            runner_name
        );
        match env::consts::OS {
            "macos" => crate::executor::ExecutorKind::Lume,
            _ => crate::executor::ExecutorKind::Meda,
        }
    }

    /// Populate `runner_executors` from the registry's view of the world.
    /// Call at startup so an agent restart with in-flight runners doesn't
    /// silently mis-route deletes to the wrong executor. Best-effort: per-
    /// executor list failures are logged and skipped.
    pub async fn seed_runner_executors_from_registry(&mut self) {
        let runners = self.registry.list_all().await;
        let mut count = 0usize;
        if let Ok(mut map) = self.runner_executors.lock() {
            for (kind, runner) in runners {
                if runner.name.starts_with("cirun-") {
                    map.insert(runner.name, kind);
                    count += 1;
                }
            }
        }
        info!(
            "Seeded runner_executors map with {} entries from live executors",
            count
        );
    }

    // Helper method to create a request builder with common headers
    fn create_request(&self, method: reqwest::Method, url: &str) -> reqwest::RequestBuilder {
        let request_id = Uuid::new_v4().to_string();
        info!("Creating request with ID: {}", request_id);

        self.client
            .request(method, url)
            .header("Authorization", format!("Bearer {}", self.api_token))
            .header("X-Request-ID", request_id)
            .header("X-Agent-ID", &self.agent.id)
    }

    async fn handle_orphaned_runners(&self, response: reqwest::Response) {
        // Parse response for runners_to_delete (orphaned VMs)
        match response.json::<ApiResponse>().await {
            Ok(api_response) => {
                if !api_response.runners_to_delete.is_empty() {
                    info!(
                        "API returned {} orphaned runners to delete from POST",
                        api_response.runners_to_delete.len()
                    );
                    let in_flight = self.in_flight_snapshot();
                    for runner in &api_response.runners_to_delete {
                        // Same race-protection as the GET-path delete loop:
                        // skip orphan-cleanup for runners we are still
                        // building. cirun api may class a runner as orphan
                        // before its provision flow has finished here, and
                        // killing the half-built VM mid-spawn produces the
                        // "disappeared during settle" error.
                        if in_flight.contains(&runner.name) {
                            warn!(
                                "ignoring SaaS orphan-delete for '{}' — provision in flight on this agent",
                                runner.name
                            );
                            continue;
                        }
                        match self.delete_runner(&runner.name).await {
                            Ok(_) => {
                                info!("[OK] Successfully deleted orphaned runner: {}", runner.name);
                            }
                            Err(e) => {
                                error!("✘ Failed to delete orphaned runner {}: {}", runner.name, e)
                            }
                        }
                    }
                }
            }
            Err(e) => {
                info!(
                    "No runners_to_delete in POST response or parse error: {}",
                    e
                );
            }
        }
    }

    pub async fn report_running_vms(&self) {
        use crate::executor::ExecutorKind;
        info!("Reporting running VMs to API");
        let all = self.registry.list_all().await;
        let vms: Vec<_> = all
            .into_iter()
            .filter(|(_, r)| r.name.starts_with("cirun-"))
            .map(|(k, r)| {
                let os = match k {
                    ExecutorKind::Docker | ExecutorKind::Meda => "linux",
                    ExecutorKind::Lume => "macos",
                };
                json!({
                    "name": r.name,
                    "os": os,
                    "cpu": 0,
                    "memory": 0,
                    "disk_size": 0,
                })
            })
            .collect();

        let url = format!("{}/agent", self.base_url);
        match self
            .create_request(reqwest::Method::POST, &url)
            .json(&json!({ "agent": self.agent, "vms": vms }))
            .send()
            .await
        {
            Ok(response) => {
                info!("API response status: {}", response.status());
                self.handle_orphaned_runners(response).await;
            }
            Err(e) => error!("Failed to send running VMs: {}", e),
        }
    }

    async fn delete_runner(&self, runner_name: &str) -> Result<(), Box<dyn std::error::Error>> {
        let kind = self.executor_for_runner(runner_name);
        info!(
            "Attempting to delete runner '{}' via {:?} executor",
            runner_name, kind
        );
        // Binding is intentionally NOT removed on success. The cirun api may
        // re-request a delete for the same runner across consecutive polling
        // cycles (POST orphan-list then GET runners_to_delete); dropping the
        // binding would force the second request through the OS-default
        // fallback (Lume on macOS / Meda on Linux) which then fails with
        // "not found" and burns the retry budget. Every executor's `kill` is
        // idempotent on "already gone", so keeping the binding is safe.
        self.registry
            .get(kind)
            .map_err(|e| -> Box<dyn std::error::Error> {
                Box::new(std::io::Error::other(e.to_string()))
            })?
            .kill(runner_name)
            .await
            .map_err(|e| -> Box<dyn std::error::Error> {
                Box::new(std::io::Error::other(e.to_string()))
            })
    }

    /// Get the current retry count for a runner. Returns 0 on a poisoned
    /// mutex — better to under-report and let the runner be re-tried
    /// than to crash the agent.
    fn get_retry_count(&self, runner_name: &str) -> u32 {
        self.retry_tracker
            .lock()
            .map(|m| m.get(runner_name).copied().unwrap_or(0))
            .unwrap_or(0)
    }

    /// Increment the retry count for a runner and return the new count.
    /// Called from the `ProvisionReporter::report` impl when a runner
    /// hits a real (non-admission) failure.
    pub(crate) fn increment_retry(&self, runner_name: &str) -> u32 {
        match self.retry_tracker.lock() {
            Ok(mut m) => {
                let count = m.entry(runner_name.to_string()).or_insert(0);
                *count += 1;
                *count
            }
            Err(_) => 1, // mutex poisoned; treat as "first attempt"
        }
    }

    /// Clear the retry count for a runner (success path).
    pub(crate) fn clear_retry(&self, runner_name: &str) {
        if let Ok(mut m) = self.retry_tracker.lock() {
            m.remove(runner_name);
        }
    }

    /// Check if a runner should be retried based on max_retries
    fn should_retry(&self, runner_name: &str, max_retries: u32) -> bool {
        self.get_retry_count(runner_name) < max_retries
    }

    /// POST a generic `AgentEvent` to the cirun-go backend. ONE entry
    /// point for all agent → backend observability — cirun-go reads
    /// `event.kind` and dispatches per its own action table (update
    /// check run, bump DB retry counter, log, …). Adding a new event
    /// kind on the agent side means: add an `EventKind` variant in
    /// `src/reporting.rs`, add a `to_agent_event` arm, mirror the kind
    /// string on cirun-go. No new HTTP route. No new payload schema.
    ///
    /// `pub(crate)` because the public entry point is the
    /// `ProvisionReporter::report` trait method; direct callers would
    /// bypass the retry-counter policy that's coupled to certain
    /// event kinds.
    pub(crate) async fn notify_event(&self, event: &crate::reporting::AgentEvent) {
        let url = format!("{}/agent", self.base_url);

        info!(
            "Emitting agent event runner={} kind={:?} severity={:?}: {}",
            event.runner_name, event.kind, event.severity, event.title
        );

        let request_data = json!({
            "agent": self.agent,
            "event": event,
        });

        match self
            .create_request(reqwest::Method::POST, &url)
            .json(&request_data)
            .send()
            .await
        {
            Ok(response) => {
                if response.status().is_success() {
                    debug!("Successfully posted agent event");
                } else {
                    warn!(
                        "API returned non-success status for agent event: {}",
                        response.status()
                    );
                }
            }
            Err(e) => {
                warn!("Failed to post agent event: {}", e);
            }
        }
    }

    /// Snapshot of the in-flight provision name set. Held briefly under the
    /// mutex; callers that need to hold the lock longer should grab the guard
    /// directly. Returns an empty set if the mutex is poisoned (best-effort).
    fn in_flight_snapshot(&self) -> std::collections::HashSet<String> {
        self.in_flight.lock().map(|s| s.clone()).unwrap_or_default()
    }

    pub async fn manage_runner_lifecycle(
        &mut self,
        provision_set: &mut JoinSet<ProvisionResult>,
        in_flight: &mut std::collections::HashSet<String>,
    ) -> Result<ApiResponse, Error> {
        let url = format!("{}/agent", self.base_url);
        info!("Fetching runner provision/deletion data from: {}", url);

        // `executors` advertises which runtimes this agent can serve, so SaaS
        // can route by capability instead of host OS. Lets a single Mac host
        // pick up both linux/docker and macos/lume jobs.
        let request_data = json!({
            "agent": self.agent,
            "executors": self.registry.kind_names(),
        });

        let response = self
            .create_request(reqwest::Method::GET, &url)
            .json(&request_data)
            .send()
            .await?;

        info!("Response status: {}", response.status());
        let json: ApiResponse = response.json().await?;

        // Handle any runners that need deletion
        if !json.runners_to_delete.is_empty() {
            info!(
                "Received {} runners to delete",
                json.runners_to_delete.len()
            );

            for runner in &json.runners_to_delete {
                // Skip delete if this runner is currently being provisioned
                // by THIS agent. The cirun api occasionally orphan-cleans a
                // runner before the agent has finished its create flow
                // (observed 2026-05-15 with meda: VM was created, deleted
                // mid-spawn, then `inspect` returned 500 → "disappeared
                // during settle"). The provision flow itself emits
                // `notify_provision_failure` on real errors, so SaaS will
                // re-evaluate; ignoring the racing delete keeps the
                // half-built VM alive long enough to either finish or fail
                // for a real reason.
                if in_flight.contains(&runner.name) {
                    warn!(
                        "ignoring SaaS delete request for '{}' — provision in flight on this agent",
                        runner.name
                    );
                    continue;
                }
                match self.delete_runner(&runner.name).await {
                    Ok(_) => {
                        info!("[OK] Successfully deleted runner: {}", runner.name);
                        self.report_running_vms().await;
                    }

                    Err(e) => error!("✘ Failed to delete runner {}: {}", runner.name, e),
                }
            }
        }

        // Handle runners that need provisioning
        if !json.runners_to_provision.is_empty() {
            info!(
                "Received {} runners to provision",
                json.runners_to_provision.len()
            );

            // First, handle retry-exhausted runners (notify API, skip them).
            // Same agent-event funnel used elsewhere: emit a
            // ProvisionEvent::Failed and let the reporter impl build the
            // wire payload so cirun-go gets the consistent shape.
            for runner in &json.runners_to_provision {
                let current_attempts = self.get_retry_count(&runner.name);
                if !self.should_retry(&runner.name, runner.max_retries) {
                    warn!(
                        "Runner '{}' has exceeded max retries ({}/{}). Skipping provisioning.",
                        runner.name, current_attempts, runner.max_retries
                    );
                    use crate::reporting::ProvisionReporter;
                    self.report(crate::reporting::ProvisionEvent::Failed {
                        runner_name: runner.name.clone(),
                        error: format!("Exceeded max retries ({})", runner.max_retries),
                        diagnostics: serde_json::Map::new(),
                    })
                    .await;
                }
            }

            // Collect eligible runners (not retry-exhausted, not already
            // in-flight, AND served by an executor we actually have).
            // The capability filter is the important one: when the api
            // fans the same runner out to multiple agents, an agent that
            // can't serve it must drop it silently. Returning a
            // ProvisionFailure for a capability mismatch causes the api
            // to mark the runner failed and then issue an orphan-delete
            // — which can race with the WINNING agent's still-running VM
            // and tear it down mid-job. Filtering here means the api only
            // ever hears from agents that can actually do the work.
            let eligible_runners: Vec<RunnerToProvision> = json
                .runners_to_provision
                .iter()
                .filter(|r| self.should_retry(&r.name, r.max_retries))
                .filter(|r| {
                    if in_flight.contains(&r.name) {
                        info!("Skipping runner '{}' — already in-flight", r.name);
                        false
                    } else {
                        true
                    }
                })
                .filter(|r| self.is_runner_dispatchable(r))
                .cloned()
                .collect();

            if !eligible_runners.is_empty() {
                // Calculate available slots based on runner capacity
                let available_slots = if let Some(max_runners) = self.max_runners {
                    let running_count = self.registry.total_count_running().await;
                    let slots = (max_runners as usize).saturating_sub(running_count);
                    info!(
                        "Runner capacity: {}/{} running, {} slots available, {} requested",
                        running_count,
                        max_runners,
                        slots,
                        eligible_runners.len()
                    );
                    if slots == 0 {
                        info!("No runner slots available. Runners will be picked up on next poll.");
                    }
                    slots
                } else {
                    eligible_runners.len()
                };

                if available_slots > 0 {
                    // Cap runners to available slots
                    let runners_to_spawn: Vec<RunnerToProvision> =
                        eligible_runners.into_iter().take(available_slots).collect();

                    info!(
                        "Spawning {} runners in parallel (max concurrency: {})",
                        runners_to_spawn.len(),
                        available_slots
                    );

                    let semaphore = Arc::new(Semaphore::new(available_slots));

                    for runner in runners_to_spawn {
                        in_flight.insert(runner.name.clone());
                        // Mirror into self.in_flight so the orphan-delete
                        // path (handle_orphaned_runners, called from the
                        // POST report-running-vms response) can also see
                        // it without taking the local arg.
                        if let Ok(mut s) = self.in_flight.lock() {
                            s.insert(runner.name.clone());
                        }
                        let sem = semaphore.clone();
                        let reg = Arc::clone(&self.registry);
                        provision_set.spawn(provision_single_runner(runner, sem, reg));
                    }

                    info!(
                        "Spawned provisioning tasks. Total in-flight: {}",
                        provision_set.len()
                    );
                }
            }
        }

        Ok(json)
    }
}

/// `ProvisionReporter` is the agent's single funnel for provision-outcome
/// observability (see `src/reporting.rs`). Every per-event policy
/// decision (which HTTP payload to send, whether to touch the retry
/// counter) lives in this impl — main.rs just calls `report(event)`.
///
/// Adding a new event type means: add a variant in `ProvisionEvent`,
/// add a match arm here. main.rs does not change.
#[async_trait::async_trait]
impl crate::reporting::ProvisionReporter for CirunClient {
    async fn report(&self, event: crate::reporting::ProvisionEvent) {
        use crate::reporting::{to_agent_event, ProvisionEvent};

        // Per-event POLICY: retry-counter mutation. AtCapacity does NOT
        // increment because the runner was never spawned (admission
        // denial); the runner stays in the SaaS's `requested` pool to
        // be re-fanned on the next poll. Succeeded clears any prior
        // retry state. Failed bumps the counter for max_retries
        // accounting and the attempt number rides on the wire event.
        let attempt = match &event {
            ProvisionEvent::Succeeded { runner_name } => {
                self.clear_retry(runner_name);
                0
            }
            ProvisionEvent::Failed { runner_name, .. } => self.increment_retry(runner_name),
            ProvisionEvent::AtCapacity { .. } => 0,
        };

        // WIRE: build the generic AgentEvent and post it through the
        // single observability funnel. Succeeded returns None today —
        // the running-VMs heartbeat already tells SaaS the runner is up,
        // so we skip the wire emit to avoid duplicate check-run noise.
        if let Some(agent_event) = to_agent_event(&event, attempt) {
            self.notify_event(&agent_event).await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::executor::{
        registry::Registry, Executor, ExecutorKind, OwnedRunner, ProvisionError, RunnerSpec,
        RunnerState,
    };
    use async_trait::async_trait;
    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;

    /// Fake executor that records every kill() invocation and always reports
    /// success. Used to drive `delete_runner` without touching docker/lume.
    struct RecordingExecutor {
        killed: StdMutex<Vec<String>>,
    }

    impl RecordingExecutor {
        fn new() -> Self {
            Self {
                killed: StdMutex::new(Vec::new()),
            }
        }
    }

    #[async_trait]
    impl Executor for RecordingExecutor {
        fn settle_timeout(&self) -> Duration {
            Duration::from_secs(1)
        }
        fn validate(&self, _spec: &RunnerSpec) -> Result<(), ProvisionError> {
            Ok(())
        }
        async fn inspect(&self, _name: &str) -> Result<RunnerState, ProvisionError> {
            Ok(RunnerState::Absent)
        }
        async fn spawn(&self, _spec: &RunnerSpec) -> Result<(), ProvisionError> {
            Ok(())
        }
        async fn kill(&self, name: &str) -> Result<(), ProvisionError> {
            self.killed.lock().unwrap().push(name.to_string());
            Ok(())
        }
        async fn list_owned(&self) -> Result<Vec<OwnedRunner>, ProvisionError> {
            Ok(Vec::new())
        }
    }

    fn test_client(registry: Arc<Registry>) -> CirunClient {
        CirunClient {
            client: Client::builder().build().unwrap(),
            base_url: "https://example.invalid".to_string(),
            api_token: "tok".to_string(),
            agent: AgentInfo {
                id: "agent-x".into(),
                hostname: "h".into(),
                os: "linux".into(),
                arch: "x86_64".into(),
            },
            retry_tracker: std::sync::Mutex::new(HashMap::new()),
            max_runners: None,
            runner_executors: std::sync::Mutex::new(HashMap::new()),
            in_flight: std::sync::Mutex::new(std::collections::HashSet::new()),
            registry,
        }
    }

    /// Regression test for the binding-loss bug observed in prod 2026-05-15:
    /// a POST-orphan-delete succeeded via Docker, but the binding was removed
    /// from `runner_executors`. The very next GET cycle re-requested the same
    /// delete; with no binding the routing fell back to the OS-default (Lume
    /// on macOS), which then hit "VM not found" and burned the retry budget.
    ///
    /// Successful deletes must preserve the binding so that idempotent re-
    /// requests from the api route to the same executor — where `kill()` is
    /// already a no-op when the runner is already gone.
    #[tokio::test]
    async fn delete_runner_preserves_binding_for_idempotent_retries() {
        let mut execs: HashMap<ExecutorKind, Arc<dyn Executor>> = HashMap::new();
        execs.insert(ExecutorKind::Docker, Arc::new(RecordingExecutor::new()));
        let registry = Arc::new(Registry::from_executors(execs));
        let client = test_client(registry);

        client
            .runner_executors
            .lock()
            .unwrap()
            .insert("cirun-r1".into(), ExecutorKind::Docker);

        client.delete_runner("cirun-r1").await.expect("delete ok");

        let map = client.runner_executors.lock().unwrap();
        assert_eq!(
            map.get("cirun-r1"),
            Some(&ExecutorKind::Docker),
            "binding must persist after a successful delete so SaaS retries do not fall back to the OS-default executor"
        );
    }

    fn make_runner(name: &str, runner_os: &str, executor: Option<&str>) -> RunnerToProvision {
        RunnerToProvision {
            name: name.into(),
            provision_script: String::new(),
            image: "ubuntu:24.04".into(),
            os: runner_os.into(),
            cpu: 2,
            memory: 4,
            disk: 20,
            login: crate::api::RunnerLogin {
                username: "runner".into(),
                password: "p".into(),
            },
            max_retries: 3,
            executor: executor.map(|s| s.to_string()),
            gpu: None,
            extra_config: None,
        }
    }

    fn macos_test_client(registry: Arc<Registry>) -> CirunClient {
        let mut c = test_client(registry);
        c.agent.os = "macos".into();
        c
    }

    /// Regression for issue #14 — a macOS agent that has Docker registered
    /// must accept a `runner.os=linux` job dispatched with `executor=docker`.
    /// Old gate compared runner.os to agent.os and dropped these silently.
    #[tokio::test]
    async fn macos_agent_with_docker_dispatches_linux_runner() {
        let mut execs: HashMap<ExecutorKind, Arc<dyn Executor>> = HashMap::new();
        execs.insert(ExecutorKind::Docker, Arc::new(RecordingExecutor::new()));
        let registry = Arc::new(Registry::from_executors(execs));
        let client = macos_test_client(registry);

        let runner = make_runner("cirun-r1", "linux", Some("docker"));
        assert!(
            client.is_runner_dispatchable(&runner),
            "Docker on macOS must serve linux runners (issue #14)"
        );
    }

    /// Lume on macOS cannot run linux containers; an explicit lume+linux
    /// dispatch must be rejected.
    #[tokio::test]
    async fn macos_agent_with_only_lume_drops_linux_runner() {
        let mut execs: HashMap<ExecutorKind, Arc<dyn Executor>> = HashMap::new();
        execs.insert(ExecutorKind::Lume, Arc::new(RecordingExecutor::new()));
        let registry = Arc::new(Registry::from_executors(execs));
        let client = macos_test_client(registry);

        let runner = make_runner("cirun-r1", "linux", Some("lume"));
        assert!(
            !client.is_runner_dispatchable(&runner),
            "lume cannot serve linux runners regardless of host"
        );
    }

    /// Capability gate: a linux runner asking for docker on an agent that
    /// has no docker registered must be dropped.
    #[tokio::test]
    async fn agent_without_requested_executor_drops_runner() {
        let mut execs: HashMap<ExecutorKind, Arc<dyn Executor>> = HashMap::new();
        execs.insert(ExecutorKind::Meda, Arc::new(RecordingExecutor::new()));
        let registry = Arc::new(Registry::from_executors(execs));
        let client = test_client(registry); // agent.os = linux

        let runner = make_runner("cirun-r1", "linux", Some("docker"));
        assert!(
            !client.is_runner_dispatchable(&runner),
            "agent without docker must not accept a docker-tagged runner"
        );
    }

    /// Sanity: looking up a runner that was never bound falls back to the
    /// OS-default and emits the fallback warning. Locks in the historical
    /// behaviour the regression test above depends on.
    #[test]
    fn unbound_runner_routes_to_os_default() {
        let registry = Arc::new(Registry::from_executors(HashMap::new()));
        let client = test_client(registry);
        let kind = client.executor_for_runner("nope");
        let expected = match env::consts::OS {
            "macos" => ExecutorKind::Lume,
            _ => ExecutorKind::Meda,
        };
        assert_eq!(kind, expected);
    }
}