runner_manager_agent/

reconcile.rs

// owner: e1-reconciliation-capacity

//! The loop that turns GitHub demand into a decision to start runners — and
//! that refuses to start them when it should not.
//!
//! Every ceiling in this product is enforced from here, so the module is
//! organised around the four things that can go wrong silently:
//!
//! * [`PollSchedule`] — the budget-aware interval. Demand shares one 5,000
//!   requests/hour ceiling with inventory and workflow counts, so this loop
//!   polls an *active* target on a bounded interval (default 60 s, hard floor
//!   30 s per target) and *increases* the delay under a rate-limit signal,
//!   never decreases it to catch up. A target with nothing happening is polled
//!   at the shorter *idle* interval (default 10 s), because its polls are
//!   conditional requests answered `304`, which GitHub does not charge against
//!   the hourly ceiling — and because work usually arrives after idle. When
//!   the account's remaining quota runs low, the schedule stretches.
//! * [`RepositoryCache`] — the per-organization repository list, refreshed on
//!   an interval materially slower than the demand poll. Re-listing an
//!   organization at demand-poll frequency is what exhausts the shared budget
//!   the paragraph above exists to protect.
//! * [`Reconciler::reconcile`] — the allocation pass. It re-reads the attempt
//!   set **under the host-wide allocation lock, once per runtime created**, so
//!   two policies reconciling concurrently cannot both spend the same headroom.
//! * [`LifecycleEvent`] — what `g2` and the local log sink see. Every field is
//!   an identifier, a count, an enumerated state or a duration; nothing free
//!   text, and nothing that came off the wire.
//!
//! # There is no acquisition step, and none may be added
//!
//! The scale-set model called `AcquireJobs` to reserve an assignment before
//! scaling. The REST path has no equivalent (`01-current-architecture.md`, edge
//! case 6), so demand is **advisory**. Two consequences are load-bearing here
//! and neither is a defect:
//!
//! 1. **A surplus runner is an accepted outcome.** Another host serving the
//!    same labels may take the job first; this host's runner then finds no work
//!    and exits on its idle timeout, having cost one capacity slot and one cold
//!    start. That terminal outcome is
//!    [`AttemptOutcome::ExitedIdleWithoutWork`], is cleaned like any other, and
//!    is counted apart from a failure — see [`ReconcileReport::idle_exits`].
//! 2. **The same job is still `queued` on the next poll** while its runner
//!    starts. Subtracting only unassigned supply in
//!    [`HostAllocator::allocate`] is what stops that from starting a second
//!    runner, and then a third. This module's only job in that arithmetic is to
//!    hand the allocator the attempt set the host actually holds — which is why
//!    [`RunnerLauncher`] supplies both the attempts and the launch, from one
//!    supply point, for the reason `b1` gives at
//!    [`HostAllocator::from_attempts`].
//!
//! `tests::nothing_in_this_module_reserves_or_claims_a_job` is a tripwire on the
//! obvious shape of a reservation being added back.
//!
//! # Demand is measured in JOBS, filtered by this policy's routing labels
//!
//! `02-target-architecture.md` writes the formula as *"queued jobs whose
//! `runs-on` matches this policy's routing labels"*, and that is now exactly
//! what this module clamps. It was not always: an earlier owner decision priced
//! the per-run job listing out and left this module clamping a count of
//! **runs**, unfiltered. `crates/github/src/demand.rs` records that decision,
//! why it was reversed, and what the reversal costs in requests.
//!
//! What the reversal means here is two changes to one line:
//!
//! * **A run of eight jobs is now eight units of demand, not one.** Under the
//!   run count a matrix filled one runner per poll while the rest of the matrix
//!   waited, so a host configured for ten concurrent runners served an
//!   eight-job matrix nearly serially. That was the defect that forced the
//!   decision back.
//! * **A job this host cannot serve is no longer demand.** A repository whose
//!   jobs target `ubuntu-latest`, or another host's `rm-<host>-…` label, used to
//!   drive its policy toward `max_capacity` and start runners that idled until
//!   they timed out. The gateway now returns each queued job's `runs-on`, so
//!   `b1`'s predicate finally has its input.
//!
//! **The predicate is still `b1`'s and the input is still `c4`'s.** This module
//! calls [`runner_manager_domain::policy::RoutingLabels`]'s `tally` and
//! implements no label comparison of its own;
//! `tests::the_label_predicate_is_b1s_and_this_module_only_applies_it` scans
//! this file's own source and fails if a second implementation grows here, which
//! is the same tripwire `c4` carries one layer down.
//!
//! # The filtering happens here rather than in the gateway, on purpose
//!
//! One target can be watched by more than one policy, each with its own routing
//! labels, and [`Reconciler`]'s `poll_targets` deliberately polls a target **once**
//! for all of them. A gateway that filtered would have to be told whose labels to
//! filter by, which would make the poll per-policy and multiply its request cost
//! by the number of policies sharing the target — the budget model prices a
//! target, not a policy. So the gateway returns the jobs and each policy tallies
//! them against its own labels.
//!
//! # What is still approximate
//!
//! The surplus-runner path above is narrowed by this change and not closed. A
//! `runs-on: ${{ matrix.runner }}` cannot be resolved without evaluating the
//! workflow, so `b1` reports it as unresolvable: never counted as demand, never
//! silently dropped, and surfaced through
//! [`LifecycleEvent::DemandObserved::unresolvable`] so that an operator can see
//! a workflow this host will never serve sitting in the queue. And demand
//! remains advisory — another host may still take a job this one started a
//! runner for — which is what the two ceilings bound.
//!
//! # What is testable without a network, a filesystem, or a process
//!
//! All of it. [`DemandSource`], [`RunnerLauncher`], [`AllocationLock`],
//! [`RepositoryDirectory`], [`Jitter`] and [`EventSink`] are ports;
//! [`GatewayDemand`], [`FileAllocationLock`], [`RandomJitter`] and
//! [`TracingEvents`] are the production adapters, and every one of them is a
//! thin shell over a decision made in this file.

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::num::NonZeroU16;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use runner_manager_domain::attempt::{
    AttemptOutcome, AttemptState, FailureReason, IsolationProviderFailure, RunnerAttempt,
};
use runner_manager_domain::capacity::{Allocation, HostAllocator, LimitingFactor};
use runner_manager_domain::memory::{MemoryFigures, MemoryVerdict};
use runner_manager_domain::model::{
    AttemptId, Clock, Host, IdlePollInterval, Org, OwnerRepo, PolicyId, RefreshInterval,
    ScaleTarget, Timestamp,
};
use runner_manager_domain::policy::{DemandTally, ScalePolicy};
use runner_manager_github::demand::{DemandGateway, QueuedDemand, demand_requests_per_poll};
use runner_manager_github::rest::{
    ActivityScope, CancelToken, InventoryError, RateLimitKind, RefreshState,
};
use runner_manager_github::traffic::{PrimaryPressure, RateLimitSnapshot};

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// How much slower than the demand poll the per-organization repository list is
/// refreshed.
///
/// There is no organization-wide workflow-runs endpoint, so an organization
/// target costs one demand request **per repository the App is installed on**
/// (`crates/github/src/demand.rs`). Discovering that repository list costs
/// requests of its own, and it is the one input to a demand poll that changes on
/// a human timescale: repositories are added to an installation by hand, not by
/// a workflow starting.
///
/// Thirty polls is 30 minutes at the 60-second default and 15 at the 30-second
/// floor — slow enough that the list is a rounding error against the demand
/// requests it scopes, and fast enough that a repository added to the
/// installation starts being served within one coffee break rather than at the
/// next restart.
pub const REPOSITORY_LIST_REFRESH_MULTIPLE: u32 = 30;

/// The longest the *unjittered* offline back-off may grow to.
///
/// A back-off is a safety mechanism, and an unclamped one is an outage with
/// extra steps. Fifteen minutes matches
/// [`runner_manager_github::rest::MAX_RATE_LIMIT_BACKOFF`], which is the other
/// place in this product where a delay is allowed to grow, and it is far inside
/// the 24-hour bound at which GitHub cancels the queued jobs this loop exists to
/// serve.
pub const MAX_OFFLINE_BACKOFF: Duration = Duration::from_secs(15 * 60);

/// The most the offline back-off is doubled, before the cap applies.
///
/// At the 60-second default this reaches [`MAX_OFFLINE_BACKOFF`] on the sixth
/// consecutive failure, which is roughly half an hour of outage. Past that the
/// cap holds it flat.
const MAX_BACKOFF_DOUBLINGS: u32 = 5;

/// How much of the computed back-off is jitter.
///
/// Jitter is **added** rather than subtracted, so a back-off never comes out
/// shorter than the delay it was computed from. Subtractive jitter would let the
/// first offline poll retry sooner than the nominal interval, which is the
/// opposite of backing off; it is spelled out because "add jitter" reads as
/// symmetric and is not.
const JITTER_RATIO: f64 = 0.5;

/// How much of a rate-limit wait is jitter: added, never subtracted.
///
/// Every target loop on a host shares one rate-limit quiet period, so without
/// it they all wake at the same instant the period ends, and a burst is the
/// shape most likely to meet the secondary limit again. A tenth spreads them
/// over a few seconds to a minute and a half without overshooting a
/// 15-minute wait by much.
const RATE_LIMIT_JITTER_RATIO: f64 = 0.1;

/// The most demand requests a minute one target's idle polling may cost.
///
/// A third of GitHub's 900-a-minute secondary limit. A target of one or a few
/// repositories never reaches it at the 5-second floor, but an organization of
/// a hundred repositories costs about four hundred requests a pass, and at a
/// 10-second idle interval that alone would exceed the whole account's
/// secondary limit. Its idle interval stretches instead.
pub const MAX_IDLE_REQUESTS_PER_MINUTE_PER_TARGET: u32 = 300;

/// GitHub cancels a queued job after this long.
///
/// `01-current-architecture.md` records the measurement; `03-control-flows.md`
/// flow 3.3 requires that the offline state **states** it, because an agent
/// offline for longer than this has lost queued work and the operator cannot
/// infer that from "offline". [`OfflineState`] is where it is said.
pub const GITHUB_CANCELS_QUEUED_JOBS_AFTER: Duration = Duration::from_secs(24 * 60 * 60);

/// How long [`FileAllocationLock`] waits for the host-wide allocation lock
/// before reporting contention.
///
/// Contention here is expected rather than exceptional — it is two of this
/// host's own policies creating runtimes at the same moment — and each hold
/// lasts only as long as one runtime creation. Waiting a few seconds turns the
/// common case into a short pause instead of a skipped runner.
///
/// # How many of these a poll actually costs
///
/// One per runtime created, none for a policy that is granted nothing, and at
/// most one further hold per policy — the case where the pre-check proposed a
/// grant and the under-lock re-read found the host had filled up underneath it,
/// so that hold creates nothing and ends the loop. `(3..=5)` in
/// `two_policies_reconciling_concurrently_never_exceed_host_capacity` is that
/// bound with two policies and three runtimes; the deterministic single-policy
/// case is pinned at exactly one per runtime.
///
/// That is worth stating because it did not used to be true and the
/// difference only shows up here: the budget was checked *after* the lock had
/// been taken and the attempt set re-read, so a policy granted N runners took
/// N+1 holds, and `start_runners` ran for every readable autoscale policy
/// including the zero-demand ones — so an idle host with P policies took P
/// host-wide locks per poll for nothing. Free under
/// [`InProcessAllocationLock`]; under [`FileAllocationLock`] each one is a
/// `spawn_blocking` plus a filesystem lock, with this wait behind it.
///
/// [`Reconciler::start_runners`] now pre-checks lock-free and stops as soon as
/// the budget is spent. The under-lock re-read still decides.
pub const ALLOCATION_LOCK_WAIT: Duration = Duration::from_secs(5);

// ---------------------------------------------------------------------------
// What one demand poll produced
// ---------------------------------------------------------------------------

/// One target's demand poll, as a value this module can decide from.
///
/// The failure half is `c3`'s [`RefreshState`] rather than an
/// [`InventoryError`], for the reason `c3` gives: `InventoryError` owns a
/// `reqwest::Error` and a `serde_json::Error`, so it is neither `Clone` nor
/// `PartialEq` and cannot be stored, compared, or rendered. Summarising at the
/// gateway boundary — exactly once, in [`GatewayDemand`] — is what lets the
/// whole schedule below be a pure function of values a test can construct.
///
/// [`RefreshState::Ready`] never appears in [`PollOutcome::Failed`]:
/// [`RefreshState::from_error`] cannot produce it, and a demand poll returns a
/// [`QueuedDemand`] rather than the runner inventory that variant carries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PollOutcome {
    /// GitHub answered. The count may still be a floor — see
    /// [`QueuedDemand::is_complete`].
    Ready(QueuedDemand),
    /// GitHub did not answer, or answered something this loop must slow down
    /// for.
    Failed(RefreshState),
}

impl PollOutcome {
    /// The demand reading, when there is one.
    #[must_use]
    pub const fn reading(&self) -> Option<&QueuedDemand> {
        match self {
            Self::Ready(demand) => Some(demand),
            Self::Failed(_) => None,
        }
    }

    /// The failure, when there is one.
    #[must_use]
    pub const fn failure(&self) -> Option<&RefreshState> {
        match self {
            Self::Failed(state) => Some(state),
            Self::Ready(_) => None,
        }
    }

    /// Whether GitHub could not be reached at all, as opposed to answering
    /// something unwelcome.
    ///
    /// The whole of flow 3.3 turns on this distinction: an outage retains
    /// running runners and backs off, while a rejection is a configuration
    /// problem that waiting does not fix.
    #[must_use]
    pub fn is_offline(&self) -> bool {
        matches!(self, Self::Failed(RefreshState::Offline))
    }
}

/// Where this loop gets its demand from.
///
/// A port rather than a direct [`DemandGateway`] dependency, because the two
/// failures this loop must handle differently — unreachable and rate-limited —
/// are distinguished by [`RefreshState`], and a test that wants to drive the
/// offline path should not have to manufacture a `reqwest::Error` to do it.
/// [`GatewayDemand`] is the one adapter that talks to `c4`.
#[async_trait::async_trait]
pub trait DemandSource: fmt::Debug + Send + Sync {
    /// Queued runs across `scope`, or why there are none to report.
    async fn poll(&self, scope: &ActivityScope) -> PollOutcome;
}

/// [`DemandSource`] over `c4`'s [`DemandGateway`].
///
/// Holds the [`CancelToken`] so that a shutting-down daemon can withdraw a poll
/// that is already blocked on a socket; `f3` keeps a clone and cancels it.
#[derive(Debug)]
pub struct GatewayDemand<G> {
    gateway: G,
    cancel: CancelToken,
}

impl<G: DemandGateway> GatewayDemand<G> {
    #[must_use]
    pub const fn new(gateway: G, cancel: CancelToken) -> Self {
        Self { gateway, cancel }
    }

    #[must_use]
    pub const fn gateway(&self) -> &G {
        &self.gateway
    }
}

#[async_trait::async_trait]
impl<G: DemandGateway + 'static> DemandSource for GatewayDemand<G> {
    async fn poll(&self, scope: &ActivityScope) -> PollOutcome {
        match self.gateway.queued_demand(scope, &self.cancel).await {
            Ok(demand) => PollOutcome::Ready(demand),
            // The one place an `InventoryError` is summarised. `c3` owns the
            // mapping — including transport-to-`Offline`, which is what flow
            // 3.3 branches on — so this loop never re-decides it.
            Err(error) => PollOutcome::Failed(RefreshState::from_error(&error)),
        }
    }
}

// ---------------------------------------------------------------------------
// The repository list, cached
// ---------------------------------------------------------------------------

/// Which repositories an organization installation reaches.
///
/// `f1` already holds this, from
/// [`runner_manager_github::AuthenticatedClient::discover_installations`]. It is
/// a port here so that [`RepositoryCache`] can be tested for the property that
/// matters — how *often* it asks — without a network.
#[async_trait::async_trait]
pub trait RepositoryDirectory: fmt::Debug + Send + Sync {
    /// The repositories this credential reaches in `org`.
    ///
    /// # Errors
    /// Anything the underlying gateway reports.
    async fn repositories(&self, org: &Org) -> Result<Vec<OwnerRepo>, InventoryError>;
}

#[derive(Debug, Clone)]
struct CachedRepositories {
    repositories: Vec<OwnerRepo>,
    fetched_at: Timestamp,
}

/// The per-organization repository list, refreshed far more slowly than demand.
///
/// # Why this is not just "call the directory each poll"
///
/// An organization demand poll already costs one request per repository. Adding
/// the installation listing to every poll makes the *scoping* of a poll cost
/// requests on the same schedule as the poll itself, which is how a
/// ten-repository organization at the 30-second floor stops fitting inside the
/// half-of-5,000 allowance `f2` admits targets against. The repository list is
/// also the one input that changes on a human timescale, so refreshing it
/// [`REPOSITORY_LIST_REFRESH_MULTIPLE`] times more slowly costs nothing real.
///
/// # A repository target never consults the directory at all
///
/// Its scope is itself. That is not an optimisation; asking an installation
/// listing which repositories a single named repository covers would be asking a
/// question whose answer is already in the target.
#[derive(Debug)]
pub struct RepositoryCache {
    directory: Arc<dyn RepositoryDirectory>,
    clock: Arc<dyn Clock>,
    ttl: Duration,
    entries: Mutex<BTreeMap<Org, CachedRepositories>>,
    lookups: AtomicU64,
}

impl RepositoryCache {
    /// Build a cache whose refresh interval is `poll` slowed by
    /// [`REPOSITORY_LIST_REFRESH_MULTIPLE`].
    #[must_use]
    pub fn new(
        directory: Arc<dyn RepositoryDirectory>,
        clock: Arc<dyn Clock>,
        poll: RefreshInterval,
    ) -> Self {
        let ttl = Duration::from_secs(u64::from(poll.as_secs()))
            .saturating_mul(REPOSITORY_LIST_REFRESH_MULTIPLE);
        Self {
            directory,
            clock,
            ttl,
            entries: Mutex::new(BTreeMap::new()),
            lookups: AtomicU64::new(0),
        }
    }

    /// How long a cached repository list is reused for.
    #[must_use]
    pub const fn ttl(&self) -> Duration {
        self.ttl
    }

    /// How many times the underlying directory was actually asked.
    ///
    /// Measured rather than assumed, for the reason `c4` measures its own
    /// request count: a budget nothing counts is a table in a document.
    #[must_use]
    pub fn lookups(&self) -> u64 {
        self.lookups.load(Ordering::SeqCst)
    }

    /// The scope one demand poll of `target` covers.
    ///
    /// # Errors
    /// Whatever the directory reported, for an organization target whose list is
    /// stale or absent. A repository target cannot fail.
    pub async fn scope_for(&self, target: &ScaleTarget) -> Result<ActivityScope, InventoryError> {
        match target {
            ScaleTarget::Repository(repository) => {
                Ok(ActivityScope::repository(repository.clone()))
            }
            ScaleTarget::Organization(org) => {
                let repositories = self.repositories_of(org).await?;
                Ok(ActivityScope::organization(org.clone(), repositories))
            }
        }
    }

    async fn repositories_of(&self, org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
        let now = self.clock.now();
        if let Some(fresh) = self.fresh_entry(org, now) {
            return Ok(fresh);
        }

        // The directory call is deliberately made with no lock held. Two
        // concurrent misses can therefore both ask, which costs one extra
        // listing on the poll that follows a restart; holding a `std::sync`
        // mutex across an `await` would cost a blocked executor thread and, on
        // a current-thread runtime, a deadlock. The cheaper mistake is the one
        // that spends a request.
        let repositories = self.directory.repositories(org).await?;
        self.lookups.fetch_add(1, Ordering::SeqCst);
        self.store(org.clone(), repositories.clone(), now);
        Ok(repositories)
    }

    fn fresh_entry(&self, org: &Org, now: Timestamp) -> Option<Vec<OwnerRepo>> {
        let entries = self.entries.lock().ok()?;
        let entry = entries.get(org)?;
        let age = now.signed_duration_since(entry.fetched_at).to_std().ok()?;
        (age < self.ttl).then(|| entry.repositories.clone())
    }

    fn store(&self, org: Org, repositories: Vec<OwnerRepo>, fetched_at: Timestamp) {
        if let Ok(mut entries) = self.entries.lock() {
            entries.insert(
                org,
                CachedRepositories {
                    repositories,
                    fetched_at,
                },
            );
        }
    }
}

// ---------------------------------------------------------------------------
// Jitter
// ---------------------------------------------------------------------------

/// The randomness in the offline back-off, as a port.
///
/// Flow 3.3 requires jittered back-off, and a jittered delay is by construction
/// not reproducible — so the source of the randomness is a port, and every test
/// below asserts the *bounds* of the delay against a fixed fraction rather than
/// asserting a number it could only have got by running the generator.
pub trait Jitter: fmt::Debug + Send + Sync {
    /// A fraction in `[0.0, 1.0)`. Values outside that range are clamped by the
    /// caller, so an implementation cannot lengthen a back-off without bound.
    fn fraction(&self) -> f64;
}

/// The production source.
#[derive(Debug, Clone, Copy, Default)]
pub struct RandomJitter;

impl Jitter for RandomJitter {
    fn fraction(&self) -> f64 {
        rand::random::<f64>()
    }
}

/// A fixed fraction, for tests and for the acceptance suite.
#[derive(Debug, Clone, Copy)]
pub struct FixedJitter(pub f64);

impl Jitter for FixedJitter {
    fn fraction(&self) -> f64 {
        self.0
    }
}

/// No jitter at all: the back-off is exactly what the schedule computed.
#[derive(Debug, Clone, Copy, Default)]
pub struct NoJitter;

impl Jitter for NoJitter {
    fn fraction(&self) -> f64 {
        0.0
    }
}

// ---------------------------------------------------------------------------
// The schedule
// ---------------------------------------------------------------------------

/// Why the next poll is when it is.
///
/// Reported rather than inferred, because
/// `04-subsystem-contracts.md` requires that rate limiting be *"displayed, never
/// hidden"* — and a delay that grew for a reason the caller cannot name is
/// hidden however visible the number is.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PollPace {
    /// The configured (active) interval. Nothing is throttling this loop.
    Nominal,
    /// The idle interval: nothing is happening on this target, so its polls
    /// are `304`s and cost nothing against the hourly quota. Not a slowdown.
    Idle,
    /// The account's remaining hourly quota is low, so this loop waits a
    /// multiple of its active interval. GitHub's quota is per user, shared by
    /// every host signed in as that user, so this can be another machine's
    /// doing.
    Stretched { pressure: PrimaryPressure },
    /// GitHub's rate limit is exhausted. Resolves by waiting.
    RateLimited { kind: RateLimitKind },
    /// GitHub's temporary authentication lockout. The credential is fine.
    LockedOut,
    /// GitHub could not be reached. `consecutive` counts the unbroken run of
    /// failures the back-off was computed from.
    Offline { consecutive: u32 },
    /// GitHub answered something no amount of waiting fixes — a rejected
    /// credential, a permissions refusal, or an error status. The loop keeps
    /// polling at its nominal interval so that a fix is noticed, and says that
    /// it is blocked rather than pretending the poll succeeded.
    Blocked,
}

impl PollPace {
    /// Whether this pace is a slowdown the operator should be told about.
    #[must_use]
    pub const fn is_throttled(&self) -> bool {
        !matches!(self, Self::Nominal | Self::Idle)
    }

    /// A fixed, credential-free name for the log sink and for `g2`.
    #[must_use]
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::Nominal => "nominal",
            Self::Idle => "idle",
            Self::Stretched {
                pressure: PrimaryPressure::Normal,
            } => "nominal",
            Self::Stretched {
                pressure: PrimaryPressure::Low,
            } => "stretched_low_budget",
            Self::Stretched {
                pressure: PrimaryPressure::Critical,
            } => "stretched_critical_budget",
            Self::RateLimited {
                kind: RateLimitKind::Primary,
            } => "rate_limited_primary",
            Self::RateLimited {
                kind: RateLimitKind::Secondary,
            } => "rate_limited_secondary",
            Self::LockedOut => "locked_out",
            Self::Offline { .. } => "offline",
            Self::Blocked => "blocked",
        }
    }
}

impl fmt::Display for PollPace {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// When to poll next, and why then.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NextPoll {
    pub delay: Duration,
    pub pace: PollPace,
}

/// The bounded, budget-aware poll interval.
///
/// # The floor is a rate-budget constraint, not a preference
///
/// [`RefreshInterval`] refuses anything under 30 seconds at construction, and
/// every delay this type produces is at least that — including the ones it
/// computes from a remote header. A rate limit may only ever make this loop
/// *slower*.
///
/// # `retry_delay` is an absolute floor, not an addend
///
/// `c3` documents [`RefreshState::retry_delay`] as *"the earliest time a retry
/// may occur"*: the scheduling rule is `next_attempt_at = now + retry_delay`,
/// and **not** the ordinary interval plus it. Adding the two compounds on every
/// successive retry — each new answer carries the remaining window, so an
/// addend ratchets outward — and the symptom is a dashboard that stays dark
/// long after GitHub said it could come back, which reads as a hang rather than
/// as a rate limit. So the two are combined with `max`, which is what makes the
/// floor a floor.
#[derive(Debug, Clone)]
pub struct PollSchedule {
    interval: RefreshInterval,
    /// The idle interval. Equal to the active one until [`Self::with_idle`]
    /// says otherwise, so a caller that never heard of it keeps its schedule.
    idle: Duration,
    consecutive_offline: u32,
    offline_since: Option<Timestamp>,
}

/// What a successful pass found, beyond "GitHub answered".
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PollConditions {
    /// Nothing is happening: no live runner, no run whose listings changed.
    /// See `QueuedDemand::is_quiet`.
    pub quiet: bool,
    /// The account's hourly quota as GitHub last reported it.
    pub quota: Option<RateLimitSnapshot>,
    /// What one poll of this target costs, in demand requests. Stretches the
    /// idle interval of a target too large to poll that often; see
    /// [`MAX_IDLE_REQUESTS_PER_MINUTE_PER_TARGET`].
    pub requests_per_pass: u32,
}

impl PollSchedule {
    #[must_use]
    pub const fn new(interval: RefreshInterval) -> Self {
        Self {
            interval,
            idle: Duration::from_secs(interval.as_secs() as u64),
            consecutive_offline: 0,
            offline_since: None,
        }
    }

    /// Poll a quiet target every `idle` instead of every active interval.
    ///
    /// Never longer than the active interval: an idle interval above it would
    /// make a quiet target *slower* to notice work than a busy one.
    #[must_use]
    pub fn with_idle(mut self, idle: Duration) -> Self {
        self.set_intervals(self.interval, idle);
        self
    }

    /// Change both intervals while running.
    pub fn set_intervals(&mut self, interval: RefreshInterval, idle: Duration) {
        self.interval = interval;
        self.idle = idle.min(self.nominal()).max(Self::idle_floor());
    }

    #[must_use]
    pub const fn interval(&self) -> RefreshInterval {
        self.interval
    }

    /// The interval a quiet target is polled at.
    #[must_use]
    pub const fn idle(&self) -> Duration {
        self.idle
    }

    /// The shortest delay this schedule ever produces: the idle interval's
    /// floor. See `IdlePollInterval` for why it is lower than [`Self::floor`].
    #[must_use]
    pub const fn idle_floor() -> Duration {
        Duration::from_secs(IdlePollInterval::MIN_SECS as u64)
    }

    /// The nominal interval as a [`Duration`].
    #[must_use]
    pub const fn nominal(&self) -> Duration {
        Duration::from_secs(self.interval.as_secs() as u64)
    }

    /// The unbroken run of offline polls this schedule has seen.
    #[must_use]
    pub const fn consecutive_offline(&self) -> u32 {
        self.consecutive_offline
    }

    /// How long GitHub has been unreachable, or `None` when it is not.
    ///
    /// Measured from the first poll of the current run rather than inferred
    /// from [`Self::consecutive_offline`] times the interval. The two diverge
    /// as soon as the back-off starts doubling, and this is the number the
    /// 24-hour queue-cancellation warning is compared against — an estimate
    /// would make that warning fire early or late, and it is the one thing the
    /// offline state exists to say.
    #[must_use]
    pub fn offline_for(&self, now: Timestamp) -> Option<Duration> {
        let since = self.offline_since?;
        now.signed_duration_since(since).to_std().ok()
    }

    /// The hard floor no computed delay may go below.
    #[must_use]
    pub const fn floor() -> Duration {
        Duration::from_secs(RefreshInterval::MIN_SECS as u64)
    }

    /// Decide when to poll next, given how this pass ended.
    ///
    /// `failure` is the most severe failure across the targets polled this pass,
    /// or `None` when every target answered. A pass that answered resets the
    /// offline run, which is the whole of "recovery needs no bookkeeping":
    /// demand is recomputed from the current queued-run set on every poll, so
    /// there is nothing else to unwind.
    pub fn next_poll(
        &mut self,
        failure: Option<&RefreshState>,
        now: Timestamp,
        jitter: &dyn Jitter,
    ) -> NextPoll {
        self.next_poll_with(failure, PollConditions::default(), now, jitter)
    }

    /// [`Self::next_poll`], with what a successful pass found.
    ///
    /// # On success, three paces, in this order
    ///
    /// 1. **Stretched**, when the account's remaining quota is under
    ///    `LOW_PRIMARY_REMAINING_PERCENT`: the active interval times
    ///    [`PrimaryPressure::stretch_factor`] (2, or 4 under
    ///    `CRITICAL_PRIMARY_REMAINING_PERCENT`), but never past the quota's
    ///    reset — waiting longer than that buys nothing, because the bucket is
    ///    full again. This wins over idle: a `304` is free, but the poll that
    ///    finds new work is not, and every host on the account draws from the
    ///    same bucket.
    /// 2. **Idle**, when the pass was quiet.
    /// 3. **Nominal**, the active interval, otherwise.
    pub fn next_poll_with(
        &mut self,
        failure: Option<&RefreshState>,
        conditions: PollConditions,
        now: Timestamp,
        jitter: &dyn Jitter,
    ) -> NextPoll {
        let nominal = self.nominal();

        let next = match failure {
            None => {
                self.recovered();
                let pressure = conditions
                    .quota
                    .map_or(PrimaryPressure::Normal, |quota| quota.pressure(now));
                match (pressure, conditions.quota) {
                    (PrimaryPressure::Low | PrimaryPressure::Critical, Some(quota)) => NextPoll {
                        delay: nominal
                            .saturating_mul(pressure.stretch_factor())
                            .min(quota.reset_in(now).max(nominal)),
                        pace: PollPace::Stretched { pressure },
                    },
                    // An idle interval no shorter than the active one is no
                    // idle interval at all, and naming it would be noise.
                    _ if conditions.quiet && self.idle < nominal => NextPoll {
                        delay: self.idle.max(idle_floor_for(conditions.requests_per_pass)),
                        pace: PollPace::Idle,
                    },
                    _ => NextPoll {
                        delay: nominal,
                        pace: PollPace::Nominal,
                    },
                }
            }
            Some(RefreshState::Offline) => {
                self.consecutive_offline = self.consecutive_offline.saturating_add(1);
                // The instant the *run* began, not the instant of this poll.
                self.offline_since.get_or_insert(now);
                NextPoll {
                    delay: self.offline_delay(nominal, jitter),
                    pace: PollPace::Offline {
                        consecutive: self.consecutive_offline,
                    },
                }
            }
            Some(state @ RefreshState::RateLimited(limit)) => {
                self.recovered();
                // `max`, never `+`. See the type documentation. The jitter is
                // added to the result, never to the floor GitHub named.
                let wait = retry_floor(state, now).max(nominal);
                NextPoll {
                    delay: wait.saturating_add(
                        wait.mul_f64(RATE_LIMIT_JITTER_RATIO * jitter.fraction().clamp(0.0, 1.0)),
                    ),
                    pace: PollPace::RateLimited { kind: limit.kind },
                }
            }
            Some(state @ RefreshState::LockedOut { .. }) => {
                self.recovered();
                NextPoll {
                    delay: retry_floor(state, now).max(nominal),
                    pace: PollPace::LockedOut,
                }
            }
            // Unauthorized, Forbidden, Failed, Cancelled. `retry_delay` is
            // `None` for all of them, and deliberately: no wait fixes a revoked
            // credential or a missing grant. Polling stops being useful but
            // does not stop, because the poll is also how a re-authentication
            // is noticed.
            Some(_) => {
                // GitHub answered, so it is reachable: whatever is wrong, it is
                // not an outage, and an outage run that was open must close.
                self.recovered();
                NextPoll {
                    delay: nominal,
                    pace: PollPace::Blocked,
                }
            }
        };

        debug_assert!(
            next.delay >= Self::floor() || next.pace == PollPace::Idle,
            "the 30-second floor is a rate-budget constraint and no charged branch may go below it"
        );
        debug_assert!(
            next.delay >= Self::idle_floor(),
            "the idle floor is a secondary-limit constraint and no branch may go below it"
        );
        next
    }

    /// GitHub answered something. Whatever it was, the outage run is over.
    fn recovered(&mut self) {
        self.consecutive_offline = 0;
        self.offline_since = None;
    }

    fn offline_delay(&self, nominal: Duration, jitter: &dyn Jitter) -> Duration {
        let doublings = self
            .consecutive_offline
            .saturating_sub(1)
            .min(MAX_BACKOFF_DOUBLINGS);
        let grown = nominal.saturating_mul(1_u32 << doublings);
        let capped = grown.min(MAX_OFFLINE_BACKOFF);
        // Additive, never subtractive: see `JITTER_RATIO`. The result may exceed
        // `MAX_OFFLINE_BACKOFF` by up to the jitter ratio, which is the price of
        // keeping a fleet of agents from retrying in lockstep at the plateau —
        // a cap applied *after* jitter would collapse every agent onto the same
        // instant precisely when the outage is longest.
        let spread = capped.mul_f64(JITTER_RATIO * jitter.fraction().clamp(0.0, 1.0));
        capped.saturating_add(spread).max(Self::floor())
    }
}

/// Whether a pass found nothing happening, so the next one may come at the
/// idle interval.
///
/// All three are needed. Every target answered, and quietly
/// (`QueuedDemand::is_quiet`); this pass started no runner; and no policy
/// here holds a live one. A live runner means a job is running or about to,
/// and its follow-on work (`needs:`, the next matrix wave) is exactly what an
/// active target is polled for. A pass that polled nothing is not quiet: there
/// is nothing to be quiet about, and the active interval is what it had
/// before.
fn is_quiet(readings: &BTreeMap<ScaleTarget, PollOutcome>, report: &ReconcileReport) -> bool {
    !readings.is_empty()
        && readings
            .values()
            .all(|outcome| outcome.reading().is_some_and(QueuedDemand::is_quiet))
        && report.started == 0
        // A missing allocation after a journal failure is unknown activity,
        // not evidence that the host is idle.
        && report.host_capacity_unreadable == 0
        && report.attempts_unreadable == 0
        // Demand this host matched and did not serve is not quiet either: a
        // launch that keeps failing (a refused JIT registration) would
        // otherwise be retried, and charged, every few seconds.
        && report
            .allocations
            .iter()
            .all(|allocation| allocation.active_owned == 0 && allocation.demand == 0)
}

/// The shortest idle delay a poll of `requests_per_pass` requests may have.
fn idle_floor_for(requests_per_pass: u32) -> Duration {
    Duration::from_secs(
        u64::from(requests_per_pass) * 60 / u64::from(MAX_IDLE_REQUESTS_PER_MINUTE_PER_TARGET),
    )
}

/// `c3`'s retry floor, with the one fallback this loop needs.
///
/// [`RefreshState::retry_delay`] answers `None` for the states no wait fixes,
/// and those never reach here — the caller matches them into
/// [`PollPace::Blocked`] first. The fallback exists so that a future
/// `RefreshState` variant added to the two arms above cannot silently schedule a
/// zero-second retry against an endpoint that asked for quiet.
fn retry_floor(state: &RefreshState, now: Timestamp) -> Duration {
    state.retry_delay(now).unwrap_or(PollSchedule::floor())
}

// ---------------------------------------------------------------------------
// Offline
// ---------------------------------------------------------------------------

/// What an operator is told while GitHub is unreachable.
///
/// Flow 3.3 requires four things of an outage — start no new runner, retain
/// existing runner processes, report `offline`, back off with jitter — and one
/// thing of the *state*: that it says GitHub cancels queued jobs after 24 hours,
/// so a prolonged outage loses queued work. That bound is stated here rather
/// than left for a reader to infer, because an operator who does not know it has
/// no reason to treat a long outage as urgent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OfflineState {
    /// The unbroken run of failed polls.
    pub consecutive: u32,
    /// How long until the next attempt.
    pub retry_in: Duration,
    /// How long this loop has been unable to reach GitHub, when it is known.
    pub offline_for: Option<Duration>,
}

impl OfflineState {
    #[must_use]
    pub const fn new(consecutive: u32, retry_in: Duration) -> Self {
        Self {
            consecutive,
            retry_in,
            offline_for: None,
        }
    }

    #[must_use]
    pub const fn since(mut self, offline_for: Duration) -> Self {
        self.offline_for = Some(offline_for);
        self
    }

    /// Whether the outage has already outlasted GitHub's queue.
    ///
    /// `false` when the duration is unknown: this reports a fact, and "we cannot
    /// tell" is not the same fact as "not yet".
    #[must_use]
    pub fn has_outlasted_the_queue(&self) -> bool {
        self.offline_for
            .is_some_and(|elapsed| elapsed >= GITHUB_CANCELS_QUEUED_JOBS_AFTER)
    }
}

impl fmt::Display for OfflineState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "GitHub is unreachable; no new runners are being started and running \
             runners are left alone. Retrying in {}s",
            self.retry_in.as_secs()
        )?;
        if self.has_outlasted_the_queue() {
            f.write_str(
                ". This outage has lasted more than 24 hours, and GitHub cancels a queued \
                 job after 24 hours, so queued work has been lost",
            )
        } else {
            f.write_str(
                ". GitHub cancels a queued job after 24 hours, so an outage longer than \
                 that loses queued work",
            )
        }
    }
}

// ---------------------------------------------------------------------------
// The launcher port
// ---------------------------------------------------------------------------

/// What this loop asks `e3` to create.
#[derive(Debug, Clone, Copy)]
pub struct LaunchRequest<'a> {
    pub host: &'a Host,
    pub policy: &'a ScalePolicy,
    /// Proof that e1 owns the host allocation lock while e3 reserves the
    /// runner and prunes what is obsolete.
    // Crate-visible so only this allocator can mint the request that reaches
    // package pruning. A caller holding an unrelated public AllocationLock can
    // no longer assemble a LaunchRequest and present that guard as authority.
    pub(crate) allocation_guard: &'a AllocationGuard,
}

/// Why one runner could not be started.
///
/// Carries `b1`'s [`FailureReason`] rather than a taxonomy of this module's own:
/// the reasons a runner fails to start are `e3`'s to know and `b1`'s to name,
/// and a third vocabulary here would be a third answer to a question the
/// operator asks once.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("the runner could not be started: {reason}")]
pub struct LaunchFailure {
    pub reason: FailureReason,
}

/// A lifecycle conclusion that must return through ordinary demand and
/// capacity allocation before another runner may start.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReplacementIntent {
    pub policy: PolicyId,
    pub previous_attempt: AttemptId,
    pub operation: &'static str,
}

impl LaunchFailure {
    #[must_use]
    pub const fn new(reason: FailureReason) -> Self {
        Self { reason }
    }
}

/// Why the persisted host ceiling could not be read. Closed, non-secret reasons
/// remain distinguishable in reports and structured logs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HostCapacityError {
    MissingHost,
    JournalUnreadable,
}

impl HostCapacityError {
    #[must_use]
    pub const fn reason(self) -> &'static str {
        match self {
            Self::MissingHost => "host_missing",
            Self::JournalUnreadable => "host_journal_unreadable",
        }
    }
}

#[derive(Debug)]
enum AllocationFailure {
    HostCapacity(HostCapacityError),
    Attempts(LaunchFailure),
}

/// The seam between the decision to start a runner and the act of starting one.
///
/// `e3` implements this; every test in this file fakes it, which is what makes
/// the whole allocator path decidable with no process, no filesystem and no
/// network.
///
/// # Why the attempt set comes through the same port as the launch
///
/// `b1` makes this argument at [`HostAllocator::from_attempts`] and it applies
/// one layer up: the host-wide total (D9) and every per-policy count (D7) are
/// two questions asked of **one** set, and a design that let the caller supply
/// the set separately from the thing that creates its members is a design in
/// which the two can disagree. Worse, it makes `&[]` expressible — and an empty
/// attempt set is exactly the shape that drops the `- active_owned_runners`
/// term, starts a second runner for a job already being served, and reports no
/// error while doing it.
///
/// So the launcher is asked, under the allocation lock, immediately before each
/// runtime is created. There is no second supply point and no cached copy.
///
/// # The two ways an implementer can say "I hold no attempts"
///
/// The argument above closes the hole for a *caller*. It stayed open one level
/// down for the **implementer**, in two shapes that both oversubscribe the
/// machine and neither of which reports anything:
///
/// * **By failing.** `attempts()` used to be infallible, which left `e3` — which
///   reads a journal off a disk — a choice between panicking and answering
///   `vec![]` on an I/O error. An empty set is indistinguishable from an idle
///   host, so a transient read failure reads as "nothing is running" and the
///   next pass allocates the whole machine for jobs already being served. It is
///   fallible now, and [`Reconciler`] treats a failure the way it treats a lock
///   it could not take: start nothing, say so, try again next pass.
/// * **By lagging.** [`Self::launch`] returns the attempt it created rather than
///   its identifier, so the caller can carry it. See that method for the
///   measurement that made this necessary.
#[async_trait::async_trait]
pub trait RunnerLauncher: fmt::Debug + Send + Sync {
    /// Read this host's current ceiling from the journal, including edits made
    /// after daemon startup. An unreadable or missing host is an error, never
    /// permission to fall back to the startup ceiling.
    async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError>;

    /// Reconcile this policy's existing processes before demand is read and
    /// capacity is recomputed. A concluded pre-acceptance attempt thereby
    /// becomes an ordinary allocation candidate in this same pass; replacement
    /// never bypasses the allocator.
    async fn supervise(
        &self,
        _policy: &ScalePolicy,
    ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
        Ok(Vec::new())
    }

    /// Every attempt this host holds, across every policy, terminal ones
    /// included.
    ///
    /// Terminal attempts are included rather than filtered out because the
    /// caller needs both answers from one set:
    /// [`AttemptState::counts_against_capacity`] decides the ceiling, and the
    /// terminal ones are what [`RunnerLauncher::clean`] is for.
    ///
    /// # Errors
    /// [`LaunchFailure`] when the set could not be read. **Never answer `Ok`
    /// with an empty vector to signal a failure** — the caller cannot tell that
    /// from an idle host, and the two lead to opposite actions.
    async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure>;

    /// Create exactly one runtime and start one runner, and return the attempt
    /// that now exists.
    ///
    /// Called once per grant, with the host-wide allocation lock held.
    ///
    /// # The attempt is returned, not just its identifier
    ///
    /// The host ceiling is enforced against a host-wide total, and that total is
    /// recomputed from [`Self::attempts`] on every hold. If a launch is not yet
    /// visible there when the *next* policy is allocated for — a journal write
    /// that has not landed, an asynchronous store, a cache — then that policy's
    /// grant is computed from a set missing the previous policy's runners, and
    /// it is too large.
    ///
    /// That is measured, not hypothetical. With a launcher whose attempts never
    /// became visible, two policies on a host of **three** started **six**
    /// runners, with the allocation lock held correctly throughout:
    /// `host_capacity=3, started=6, launches=6`. Serialisation was never the
    /// problem; the arithmetic under it was reading a stale set.
    ///
    /// So an implementer *should* make the new attempt visible to
    /// [`Self::attempts`] before returning — and the caller does not depend on
    /// it. [`Reconciler`] carries what this pass created and merges it, by
    /// [`RunnerAttempt::id`], with whatever the launcher reports. A launcher
    /// that honours the contract is not double-counted, and one that lags cannot
    /// oversubscribe the host.
    ///
    /// # Every call must return a **fresh** [`RunnerAttempt::id`]
    ///
    /// This is a requirement, not a convention, because the merge above is what
    /// carries the host ceiling and the merge is keyed on the identifier. Two
    /// calls that answer with the same id are two runtimes that the host-wide
    /// total counts once, and the machine is then allocated past
    /// `host_capacity`: probed at `host_capacity = 3` with one slot already
    /// busy and a launcher answering with a duplicate id, the pass started
    /// **four** runners for five occupied slots.
    ///
    /// That is a narrower defect than the lagging launcher above — that one
    /// needed no bug at all, this one needs a broken id generator — but `e3` is
    /// the implementor and cannot honour a requirement nobody states.
    /// [`Reconciler::host_attempts`] carries a `debug_assert` that fires on a
    /// collision, so a development build finds it at the first duplicate rather
    /// than through an oversubscribed host.
    ///
    /// # Errors
    /// [`LaunchFailure`], carrying the [`FailureReason`] `e3` recorded.
    ///
    /// Called under the allocation lock, so it stops at what must be decided
    /// there: the attempt, journalled, holding its slot. The runner itself is
    /// [`Self::finish_launch`], outside the lock.
    async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure>;

    /// Bring up a runner [`Self::launch`] reserved, after the host allocation
    /// lock was released: its package, its registration and its process, none
    /// of which decides a ceiling, and all of which may take minutes.
    ///
    /// The reservation is what holds the slot, and it is journalled, so another
    /// allocator counts it from the moment the lock is let go, and a crash at
    /// any point here leaves a journalled attempt that startup recovery
    /// concludes. A launcher whose `launch` does everything keeps the default.
    ///
    /// # Errors
    /// [`LaunchFailure`], carrying the [`FailureReason`] recorded.
    async fn finish_launch(
        &self,
        _policy: &ScalePolicy,
        attempt: RunnerAttempt,
        _waits: ReservationWaits,
    ) -> Result<RunnerAttempt, LaunchFailure> {
        Ok(attempt)
    }

    /// Remove a terminal attempt's runtime and mark it `cleaned`.
    ///
    /// Never called for a non-terminal attempt: capacity is reclaimed when an
    /// attempt reaches a terminal state and at no other time.
    ///
    /// # Errors
    /// [`LaunchFailure`], carrying the [`FailureReason`] `e3` recorded.
    async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure>;
}

/// A [`RunnerLauncher`] that refuses every launch with
/// [`FailureReason::HostUnfit`] while `unfit` is set, and passes everything
/// else straight through.
///
/// The daemon sets the flag while a required host check fails (`host
/// doctor`). The refusal comes before the inner launcher runs, so nothing is
/// registered with GitHub, and the reconciler reports it as a
/// `runner_start_failed` event with reason `host_unfit` -- visible, rather than
/// a host that silently stopped serving. Supervision, the attempt set and
/// cleanup are never gated: what already runs is still looked after.
#[derive(Debug)]
pub struct FitnessGatedLauncher {
    inner: Arc<dyn RunnerLauncher>,
    unfit: Arc<AtomicBool>,
}

impl FitnessGatedLauncher {
    #[must_use]
    pub const fn new(inner: Arc<dyn RunnerLauncher>, unfit: Arc<AtomicBool>) -> Self {
        Self { inner, unfit }
    }
}

#[async_trait::async_trait]
impl RunnerLauncher for FitnessGatedLauncher {
    async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
        self.inner.host_capacity(host).await
    }

    async fn supervise(
        &self,
        policy: &ScalePolicy,
    ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
        self.inner.supervise(policy).await
    }

    async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
        self.inner.attempts().await
    }

    async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
        if self.unfit.load(Ordering::Acquire) {
            return Err(LaunchFailure::new(FailureReason::HostUnfit));
        }
        self.inner.launch(request).await
    }

    async fn finish_launch(
        &self,
        policy: &ScalePolicy,
        attempt: RunnerAttempt,
        waits: ReservationWaits,
    ) -> Result<RunnerAttempt, LaunchFailure> {
        self.inner.finish_launch(policy, attempt, waits).await
    }

    async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
        self.inner.clean(attempt).await
    }
}

// ---------------------------------------------------------------------------
// The host-wide allocation lock
// ---------------------------------------------------------------------------

/// The lock is held for as long as this value lives.
///
/// Opaque on purpose: what is being held differs between the in-process and the
/// file-backed implementation, and a caller that could see which one it has
/// would eventually branch on it.
pub struct AllocationGuard {
    _held: Box<dyn std::any::Any + Send + Sync>,
    launch_hold: LaunchHold,
}

impl AllocationGuard {
    /// Wrap whatever the implementation holds. Dropping the guard drops it.
    #[must_use]
    fn new<T: Send + Sync + 'static>(held: T) -> Self {
        Self {
            _held: Box::new(held),
            launch_hold: LaunchHold::default(),
        }
    }

    /// What a launch keeps after the lock is released; see [`LaunchHold`].
    fn launch_hold(&self) -> LaunchHold {
        self.launch_hold.clone()
    }
}

/// What must stay held while a runner reserved under the allocation lock is
/// brought up outside it. Empty, except where a boundary other than the lock
/// must not be crossed by a launch in progress: see
/// [`WslRecoveryAllocationLock`].
#[derive(Clone, Default)]
struct LaunchHold {
    _held: Option<Arc<dyn std::any::Any + Send + Sync>>,
}

impl fmt::Debug for AllocationGuard {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("AllocationGuard")
    }
}

/// The host-wide allocation lock could not be taken.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("the host-wide allocation lock is held by another allocator; no runtime was created")]
pub struct AllocationLockBusy;

/// Flow 2.4's *"takes the host-wide allocation lock before creating each local
/// runtime"*, as a port.
///
/// # Why a lock is needed at all, given the allocator already exists
///
/// [`HostAllocator`] enforces D9 across the policies of **one** pass. It cannot
/// enforce anything across two passes running at once, and `f3` runs one
/// demand-polling loop per target: without serialisation, two loops read the
/// same headroom, each finds it sufficient, and the host ends up with the sum of
/// two grants it only ever had room for one of. The lock is what makes the
/// read-decide-create sequence atomic, and it is taken once per runtime rather
/// than once per pass so that a slow package download in one policy does not
/// hold the whole host still.
#[async_trait::async_trait]
pub trait AllocationLock: fmt::Debug + Send + Sync {
    /// Take the lock, waiting briefly for it.
    ///
    /// # Errors
    /// [`AllocationLockBusy`] when it could not be taken. A refused grant is
    /// always safe: the next pass re-reads the headroom and tries again.
    async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy>;
}

/// The lock every task inside one agent process contends for.
///
/// This is the implementation that matters in practice, because the
/// single-instance lock (`d1`) already guarantees one agent per host: the
/// concurrency the allocation lock actually has to serialise is `f3`'s
/// per-target loops inside that one process. A `tokio` mutex rather than a
/// `std` one because it is held across the `await` that creates the runtime.
#[derive(Debug, Default)]
pub struct InProcessAllocationLock {
    mutex: Arc<tokio::sync::Mutex<()>>,
}

impl InProcessAllocationLock {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait::async_trait]
impl AllocationLock for InProcessAllocationLock {
    async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
        let mutex = Arc::clone(&self.mutex);
        let guard = mutex.lock_owned().await;
        Ok(AllocationGuard::new(guard))
    }
}

/// `d1`'s file lock, which is host-wide across processes as well as across
/// tasks.
///
/// Defence in depth behind [`InProcessAllocationLock`], for the configuration
/// `d1` documents as the one where two agents can genuinely coexist: the
/// platform state directory is per-account, so a service-account daemon and an
/// interactive `daemon run` resolve different paths and do not contend for the
/// single-instance lock. They do contend here if they share a state directory.
///
/// [`runner_manager_platform::lock::HostLock::acquire`] blocks the calling
/// thread and its own documentation names this caller: *"Async callers must wrap
/// it in [`tokio::task::spawn_blocking`]"*. That is what this does, and the
/// returned `HostLock` lives inside the guard, because dropping it is the
/// release.
#[derive(Debug, Clone)]
pub struct FileAllocationLock {
    paths: Arc<runner_manager_platform::paths::AppPaths>,
    wait: Duration,
}

impl FileAllocationLock {
    #[must_use]
    pub const fn new(paths: Arc<runner_manager_platform::paths::AppPaths>) -> Self {
        Self {
            paths,
            wait: ALLOCATION_LOCK_WAIT,
        }
    }

    #[must_use]
    pub const fn with_wait(mut self, wait: Duration) -> Self {
        self.wait = wait;
        self
    }
}

#[async_trait::async_trait]
impl AllocationLock for FileAllocationLock {
    async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
        use runner_manager_platform::lock::{HostLock, LockKind};

        let paths = Arc::clone(&self.paths);
        let wait = self.wait;
        let held = tokio::task::spawn_blocking(move || {
            HostLock::acquire(&paths, LockKind::Allocation, wait)
        })
        .await;

        match held {
            Ok(Ok(lock)) => Ok(AllocationGuard::new(lock)),
            // A refused lock and a panicked blocking task are the same outcome
            // to this caller: no runtime was created and the next pass will
            // re-read the headroom. Neither is allowed to look like a grant.
            Ok(Err(_)) | Err(_) => Err(AllocationLockBusy),
        }
    }
}

/// Adds the Windows/WSL recovery fence to an ordinary allocation lock, and
/// says on disk when allocations keep being refused.
///
/// A Windows watchdog writes its drain request before attempting the same
/// atomic directory claim. Therefore either this process owns the directory and
/// finishes the launches already in progress, or Windows owns it and no new
/// launch can pass. The claim is shared by every launch of this process, from
/// its reservation until its runner is up ([`LaunchHold`]), and released with
/// the last: a drain request refuses new reservations, so the launches in
/// progress run out and Windows gets the directory, as when there was one.
/// A malformed configured fence fails closed. A guest claim
/// whose owner is provably gone is reclaimed first; see
/// [`runner_manager_platform::wsl::fence`].
///
/// # Refusals that persist are recorded
///
/// A refusal is always safe for one pass, and a refusal on every pass is a host
/// that starts nothing while it looks healthy -- 13 days of it, once. So after
/// [`launch_health::BLOCKED_AFTER_DEFERRALS`] consecutive refusals spanning at
/// least [`launch_health::BLOCKED_AFTER`], with no launch of this process in
/// progress, the lock writes the `state/` record `status`, `service status`,
/// `wsl status` and `host doctor` read. The next granted allocation removes it.
/// Refusals while this process is itself launching are its own contention and
/// count neither way.
///
/// [`launch_health::BLOCKED_AFTER_DEFERRALS`]: runner_manager_platform::launch_health::BLOCKED_AFTER_DEFERRALS
/// [`launch_health::BLOCKED_AFTER`]: runner_manager_platform::launch_health::BLOCKED_AFTER
#[derive(Debug)]
pub struct WslRecoveryAllocationLock {
    paths: Arc<runner_manager_platform::paths::AppPaths>,
    inner: Arc<dyn AllocationLock>,
    blocked_after_refusals: u32,
    blocked_after: Duration,
    /// A quiet spell this long between refusals starts a new run.
    streak_gap: Duration,
    in_flight: Arc<std::sync::atomic::AtomicUsize>,
    /// The claim the launches in progress share; taking it is serialised.
    claim: tokio::sync::Mutex<std::sync::Weak<LaunchFenceGuard>>,
    refusals: Mutex<Refusals>,
}

/// The refusals since the last grant.
#[derive(Debug, Default)]
struct Refusals {
    /// The current run of refusals: when it began, the latest one, and how
    /// many it holds.
    streak: Option<Streak>,
    recorded: bool,
}

#[derive(Debug, Clone, Copy)]
struct Streak {
    since: chrono::DateTime<chrono::Utc>,
    latest: chrono::DateTime<chrono::Utc>,
    count: u32,
}

/// Counts a launch of this process as in progress until dropped.
struct InFlight(Arc<std::sync::atomic::AtomicUsize>);

impl InFlight {
    fn enter(count: &Arc<std::sync::atomic::AtomicUsize>) -> Self {
        count.fetch_add(1, Ordering::SeqCst);
        Self(Arc::clone(count))
    }
}

impl Drop for InFlight {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::SeqCst);
    }
}

/// Releases the guest launch claim when the allocation ends, and says so when
/// it cannot: this once failed silently while a distribution started nothing
/// for 13 days.
struct LaunchFenceGuard(Option<runner_manager_platform::wsl::fence::GuestLaunchClaim>);

impl Drop for LaunchFenceGuard {
    fn drop(&mut self) {
        if let Some(claim) = self.0.take()
            && let Err(error) = claim.release()
        {
            tracing::warn!(
                %error,
                "the WSL launch fence could not be released; the next launch reclaims it"
            );
        }
    }
}

impl WslRecoveryAllocationLock {
    #[must_use]
    pub fn new(
        paths: Arc<runner_manager_platform::paths::AppPaths>,
        inner: Arc<dyn AllocationLock>,
    ) -> Self {
        use runner_manager_platform::launch_health::{BLOCKED_AFTER, BLOCKED_AFTER_DEFERRALS};
        Self {
            paths,
            inner,
            blocked_after_refusals: BLOCKED_AFTER_DEFERRALS,
            blocked_after: BLOCKED_AFTER,
            streak_gap: BLOCKED_AFTER,
            in_flight: Arc::default(),
            claim: tokio::sync::Mutex::default(),
            refusals: Mutex::default(),
        }
    }

    /// Record blocked launches after `refusals` consecutive refusals spanning
    /// at least `after`, instead of the defaults.
    #[cfg(test)]
    const fn with_blocked_after(mut self, refusals: u32, after: Duration) -> Self {
        self.blocked_after_refusals = refusals;
        self.blocked_after = after;
        self
    }

    /// Start a new run of refusals after a quiet spell of `gap`.
    #[cfg(test)]
    const fn with_streak_gap(mut self, gap: Duration) -> Self {
        self.streak_gap = gap;
        self
    }

    fn refusals(&self) -> std::sync::MutexGuard<'_, Refusals> {
        self.refusals
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn note_granted(&self) {
        let mut refusals = self.refusals();
        refusals.streak = None;
        if refusals.recorded {
            match runner_manager_platform::launch_health::clear_launches_blocked(&self.paths) {
                Ok(()) => refusals.recorded = false,
                // Kept, so the next grant tries again.
                Err(error) => tracing::warn!(%error, "the blocked-launch record cannot be cleared"),
            }
        }
    }

    fn note_refused(&self, cause: &runner_manager_platform::launch_health::BlockCause) {
        use runner_manager_platform::launch_health::{Viewpoint, record_launches_blocked};
        use runner_manager_platform::wsl::fence::elapsed_at_least;

        if self.in_flight.load(Ordering::SeqCst) > 0 {
            return;
        }
        let now = chrono::Utc::now();
        let mut refusals = self.refusals();
        // A refusal long after the previous one starts a new run: two brief
        // contentions days apart are not five minutes of nothing but refusals.
        let streak = match refusals.streak {
            Some(streak) if !elapsed_at_least(streak.latest, now, self.streak_gap) => Streak {
                latest: now,
                count: streak.count.saturating_add(1),
                ..streak
            },
            _ => Streak {
                since: now,
                latest: now,
                count: 1,
            },
        };
        refusals.streak = Some(streak);
        if streak.count < self.blocked_after_refusals
            || !elapsed_at_least(streak.since, now, self.blocked_after)
        {
            return;
        }
        let blocked = cause.blocked_since(streak.since, Viewpoint::Guest);
        if !refusals.recorded {
            tracing::warn!(
                refusals = streak.count,
                "runner launches are blocked {blocked}: {}",
                blocked.remedy
            );
        }
        // Rewritten on every refusal, so readers can tell a record nobody
        // refreshes -- no demand any more -- from one still being refused.
        match record_launches_blocked(&self.paths, &blocked, now) {
            Ok(()) => refusals.recorded = true,
            Err(error) => tracing::warn!(%error, "the blocked-launch record cannot be written"),
        }
    }
}

/// Claim the guest launch fence under `root`, or say what holds it.
fn claim_guest_fence(
    root: &std::path::Path,
) -> Result<Arc<LaunchFenceGuard>, runner_manager_platform::launch_health::BlockCause> {
    use runner_manager_platform::launch_health::BlockCause;
    use runner_manager_platform::wsl::fence::{
        FENCE_DIRECTORY, FenceClaim, GuestClaimAttempt, try_claim_guest_launch,
    };

    match try_claim_guest_launch(root) {
        Ok(GuestClaimAttempt::Claimed { claim, reclaimed }) => {
            if let Some(reclaimed) = reclaimed {
                tracing::warn!(
                    shared_root = %root.display(),
                    %reclaimed,
                    "reclaimed a WSL launch fence claim whose owner is gone"
                );
            }
            // Guarded at once, so a claim let go on the drain path is released
            // -- and logged when it cannot be -- like any other.
            Ok(Arc::new(LaunchFenceGuard(Some(claim))))
        }
        Ok(GuestClaimAttempt::Busy) => {
            let directory = root.join(FENCE_DIRECTORY);
            Err(match FenceClaim::owner(root) {
                Ok(owner) => BlockCause::FenceHeld { directory, owner },
                Err(error) => BlockCause::FenceOwnerUnreadable {
                    directory,
                    detail: error.to_string(),
                },
            })
        }
        Err(error) => {
            // An owner record nobody can decode stops the reclaim the same way
            // on every pass; name it rather than the config.
            Err(match FenceClaim::owner(root) {
                Err(owner_error) => BlockCause::FenceOwnerUnreadable {
                    directory: root.join(FENCE_DIRECTORY),
                    detail: owner_error.to_string(),
                },
                Ok(_) => BlockCause::FenceUnusable {
                    detail: error.to_string(),
                },
            })
        }
    }
}

#[async_trait::async_trait]
impl AllocationLock for WslRecoveryAllocationLock {
    async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
        use runner_manager_platform::launch_health::BlockCause;
        use runner_manager_platform::wsl::fence::{DrainRequest, GuestRecoveryConfig};

        let paths = Arc::clone(&self.paths);
        let mut shared = self.claim.lock().await;
        let held = shared.upgrade();
        let claim = tokio::task::spawn_blocking(move || {
            let unusable = |error: &dyn fmt::Display| BlockCause::FenceUnusable {
                detail: error.to_string(),
            };
            let config = match GuestRecoveryConfig::read(&paths) {
                Ok(Some(config)) => config,
                Ok(None) => return Ok(Arc::new(LaunchFenceGuard(None))),
                Err(error) => return Err(unusable(&error)),
            };
            let root = config.shared_root;
            let claim = match held {
                // A launch of this process already owns the directory.
                Some(held) => {
                    if let Some(Err(error)) = held.0.as_ref().map(|claim| claim.renew()) {
                        tracing::warn!(%error, "the WSL launch fence's age could not be renewed");
                    }
                    held
                }
                None => claim_guest_fence(&root)?,
            };
            match DrainRequest::read(&root) {
                Ok(None) => Ok(claim),
                Ok(Some(_)) => Err(BlockCause::DrainRequested { root }),
                Err(error) => Err(unusable(&error)),
            }
        })
        .await
        .map_err(|_| AllocationLockBusy)?;

        let claim = match claim {
            Ok(claim) => claim,
            Err(cause) => {
                drop(shared);
                self.note_refused(&cause);
                return Err(AllocationLockBusy);
            }
        };
        if claim.0.is_some() {
            *shared = Arc::downgrade(&claim);
        }
        drop(shared);
        let Ok(inner) = self.inner.acquire().await else {
            self.note_refused(&BlockCause::AllocationLockBusy {
                path: self
                    .paths
                    .state_dir()
                    .join(runner_manager_platform::lock::LockKind::Allocation.file_name()),
            });
            return Err(AllocationLockBusy);
        };
        self.note_granted();
        let in_flight = InFlight::enter(&self.in_flight);
        let mut guard = AllocationGuard::new(inner);
        guard.launch_hold = LaunchHold {
            _held: Some(Arc::new((claim, in_flight))),
        };
        Ok(guard)
    }
}

// ---------------------------------------------------------------------------
// Lifecycle events
// ---------------------------------------------------------------------------

/// Which terminal thing happened, as a closed vocabulary.
///
/// The distinction `g2` renders: an idle exit is the accepted surplus case and
/// **not** a failure, and showing it as one sends an operator hunting a fault
/// that does not exist.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OutcomeKind {
    CompletedJob,
    IdleExit,
    Failed,
    Orphaned,
}

impl OutcomeKind {
    #[must_use]
    pub const fn of(outcome: &AttemptOutcome) -> Self {
        match outcome {
            AttemptOutcome::CompletedJob => Self::CompletedJob,
            AttemptOutcome::ExitedIdleWithoutWork => Self::IdleExit,
            AttemptOutcome::Failed { .. } => Self::Failed,
            AttemptOutcome::Orphaned => Self::Orphaned,
        }
    }

    #[must_use]
    pub const fn is_failure(&self) -> bool {
        matches!(self, Self::Failed | Self::Orphaned)
    }

    #[must_use]
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::CompletedJob => "completed_job",
            Self::IdleExit => "exited_idle_without_work",
            Self::Failed => "failed",
            Self::Orphaned => "orphaned",
        }
    }
}

impl fmt::Display for OutcomeKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// A [`FailureReason`]'s variant name, with no detail.
///
/// [`FailureReason::Other`] carries a `String` that `e3` fills in, and an event
/// is not the place for it: `07-security.md`'s log scan runs over everything
/// this loop emits, and free text is the one shape that can carry a credential
/// past a field allow-list. The operator-facing detail reaches the journal
/// through `b2` and the screen through `g2`; what reaches an *event* is the
/// variant.
#[must_use]
pub const fn failure_reason_kind(reason: &FailureReason) -> &'static str {
    match reason {
        FailureReason::JitRequestFailed => "jit_request_failed",
        FailureReason::JitExpired => "jit_expired",
        FailureReason::RunnerPackageUnverified => "runner_package_unverified",
        FailureReason::RunnerVersionRejected => "runner_version_rejected",
        FailureReason::ProcessStartFailed => "process_start_failed",
        FailureReason::ProcessExitedUnexpectedly => "process_exited_unexpectedly",
        FailureReason::RegistrationTimedOut => "registration_timed_out",
        FailureReason::TerminatedAfterRegistrationTimeout => {
            "terminated_after_registration_timeout"
        }
        FailureReason::IsolationProvider(category) => match category {
            IsolationProviderFailure::Unsupported => "isolation_provider_unsupported",
            IsolationProviderFailure::NotInstalled => "isolation_provider_not_installed",
            IsolationProviderFailure::PermissionDenied => "isolation_provider_permission_denied",
            IsolationProviderFailure::ImageUnavailableOrIncompatible => {
                "isolation_image_unavailable"
            }
            IsolationProviderFailure::DiskQuotaUnavailable => "isolation_disk_quota_unavailable",
            IsolationProviderFailure::Degraded => "isolation_provider_degraded",
            IsolationProviderFailure::RuntimeOperationFailed => {
                "isolation_runtime_operation_failed"
            }
            IsolationProviderFailure::OwnershipMismatch => "isolation_ownership_mismatch",
            IsolationProviderFailure::UnsafeRuntimePath => "isolation_unsafe_runtime_path",
            IsolationProviderFailure::JitHandoffRejected => "isolation_jit_handoff_rejected",
        },
        FailureReason::WorkspaceCleanupDeferred => "ephemeral_workspace_could_not_be_removed",
        FailureReason::HostUnfit => "host_unfit",
        FailureReason::Other(_) => "other",
    }
}

/// Fixed, credential-free reasons a queued job cannot route to one profile.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum RoutingRefusal {
    OverlappingProfiles,
}

impl RoutingRefusal {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::OverlappingProfiles => "overlapping_profile_matches",
        }
    }
}

impl fmt::Display for RoutingRefusal {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::OverlappingProfiles => {
                "repair overlapping profile labels or require a unique profile selector"
            }
        })
    }
}

/// What `g2`'s activity view and the local log sink see.
///
/// **Every field is an identifier, a count, a duration, or a `&'static str`
/// drawn from a closed set.** There is no `String` anywhere in this enum, which
/// is what makes "no emitted event contains a token, a JIT blob, or a credential
/// header" a property of the type rather than a discipline each call site has to
/// keep. `tests::no_emitted_event_can_carry_a_credential` renders every variant
/// through `d1`'s scrubber and asserts nothing changes, with a positive control
/// so the assertion cannot pass vacuously.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LifecycleEvent {
    /// A demand poll answered for one target.
    DemandObserved {
        policy: PolicyId,
        /// Queued jobs this policy's routing labels match. The number clamped.
        demand: u32,
        /// Queued jobs whose required labels this policy does not carry.
        ///
        /// Never demand. Reported because the difference between this and
        /// `demand` is the whole value of the label filtering, and an operator
        /// wondering why a busy repository started no runners is owed it.
        not_matched: u32,
        /// Queued jobs whose `runs-on` could not be resolved statically.
        ///
        /// `b1` requires these be "reported as unresolvable rather than silently
        /// counted or silently dropped": counting one would start a runner for a
        /// job that may not be ours, and dropping it would hide a workflow this
        /// host can never serve. A count rather than the reasons themselves
        /// because this type is `Copy`, and `c4` logs the reasons where it
        /// builds them.
        unresolvable: u32,
        /// `false` when the count is a floor rather than a total.
        complete: bool,
    },
    /// A queued job cannot be assigned unambiguously to this target's profiles.
    RoutingRefused {
        policy: PolicyId,
        reason: RoutingRefusal,
        count: u32,
    },
    /// A target could not be polled, so its policies start nothing this pass.
    TargetUnreadable {
        policy: PolicyId,
        reason: &'static str,
    },
    /// One policy's share of the pass.
    Allocated {
        policy: PolicyId,
        demand: u32,
        desired: u16,
        active_owned: u16,
        headroom: u16,
        to_start: u16,
        limiting: LimitingFactor,
    },
    /// A monitor-only policy was skipped entirely, before any demand request
    /// was issued for it (D19).
    MonitorOnlySkipped { policy: PolicyId },
    /// One runtime was created and one runner started.
    RunnerStarted {
        policy: PolicyId,
        attempt: AttemptId,
    },
    /// One runner could not be started.
    RunnerStartFailed {
        policy: PolicyId,
        reason: &'static str,
    },
    /// The allocation lock was not free, so `count` runners this policy was
    /// granted were not created this pass.
    AllocationDeferred { policy: PolicyId, count: u16 },
    /// The host lacks the memory for one more runner of this policy: free
    /// memory less what its running attempts may still grow into is under
    /// the estimate plus the margin. Not `started`: the demand stays queued
    /// and a later pass tries again. `started`: nothing runs that could free
    /// memory, so the runner was started anyway.
    MemoryShort {
        policy: PolicyId,
        figures: MemoryFigures,
        started: bool,
    },
    /// The host's attempt set could not be read at all.
    ///
    /// Distinct from an empty set on purpose, and the whole reason
    /// [`RunnerLauncher::attempts`] is fallible: the two produce the same
    /// *number* and demand opposite actions.
    AttemptsUnreadable { reason: &'static str },
    /// The ceiling is unknown; this does not imply that the attempt set is unreadable.
    HostCapacityUnreadable { reason: HostCapacityError },
    /// A terminal attempt's runtime was removed.
    AttemptCleaned {
        policy: PolicyId,
        attempt: AttemptId,
        outcome: OutcomeKind,
    },
    /// A terminal attempt's runtime could not be removed. It will be retried on
    /// the next pass, and this is what keeps that retry from being silent.
    AttemptCleanFailed {
        policy: PolicyId,
        attempt: AttemptId,
        reason: &'static str,
    },
    /// Scale-down declined to remove a runner that is executing a job.
    ScaleDownRefused {
        policy: PolicyId,
        attempt: AttemptId,
    },
    /// When the next poll is, and why then.
    PollScheduled { retry_in_ms: u64, pace: PollPace },
}

impl LifecycleEvent {
    /// A fixed name, for the `event` field `d1`'s sink allows verbatim.
    #[must_use]
    pub const fn name(&self) -> &'static str {
        match self {
            Self::DemandObserved { .. } => "demand_observed",
            Self::RoutingRefused { .. } => "routing_refused",
            Self::TargetUnreadable { .. } => "target_unreadable",
            Self::Allocated { .. } => "allocated",
            Self::MonitorOnlySkipped { .. } => "monitor_only_skipped",
            Self::RunnerStarted { .. } => "runner_started",
            Self::RunnerStartFailed { .. } => "runner_start_failed",
            Self::AllocationDeferred { .. } => "allocation_deferred",
            Self::MemoryShort { started: false, .. } => "memory_deferred",
            Self::MemoryShort { started: true, .. } => "memory_admitted_alone",
            Self::AttemptsUnreadable { .. } => "attempts_unreadable",
            Self::HostCapacityUnreadable { .. } => "host_capacity_unreadable",
            Self::AttemptCleaned { .. } => "attempt_cleaned",
            Self::AttemptCleanFailed { .. } => "attempt_clean_failed",
            Self::ScaleDownRefused { .. } => "scale_down_refused",
            Self::PollScheduled { .. } => "poll_scheduled",
        }
    }

    /// Which policy this event is about.
    #[must_use]
    pub const fn policy(&self) -> Option<PolicyId> {
        match self {
            Self::DemandObserved { policy, .. }
            | Self::RoutingRefused { policy, .. }
            | Self::TargetUnreadable { policy, .. }
            | Self::Allocated { policy, .. }
            | Self::MonitorOnlySkipped { policy }
            | Self::RunnerStarted { policy, .. }
            | Self::RunnerStartFailed { policy, .. }
            | Self::AllocationDeferred { policy, .. }
            | Self::MemoryShort { policy, .. }
            | Self::AttemptCleaned { policy, .. }
            | Self::AttemptCleanFailed { policy, .. }
            | Self::ScaleDownRefused { policy, .. } => Some(*policy),
            Self::PollScheduled { .. }
            | Self::AttemptsUnreadable { .. }
            | Self::HostCapacityUnreadable { .. } => None,
        }
    }
}

impl fmt::Display for LifecycleEvent {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DemandObserved {
                policy,
                demand,
                not_matched,
                unresolvable,
                complete,
            } => write!(
                f,
                "policy {policy}: {demand} queued jobs for this host{}{}{}",
                if *not_matched == 0 {
                    String::new()
                } else {
                    format!(", {not_matched} for other labels")
                },
                if *unresolvable == 0 {
                    String::new()
                } else {
                    format!(", {unresolvable} with an unresolvable `runs-on`")
                },
                if *complete {
                    ""
                } else {
                    " (a floor, not a total)"
                }
            ),
            Self::TargetUnreadable { policy, reason } => {
                write!(f, "policy {policy}: target unreadable ({reason})")
            }
            Self::RoutingRefused {
                policy,
                reason,
                count,
            } => write!(f, "policy {policy}: {count} queued jobs blocked: {reason}"),
            Self::Allocated {
                policy,
                demand,
                desired,
                active_owned,
                headroom,
                to_start,
                limiting,
            } => write!(
                f,
                "policy {policy}: demand {demand}, desired {desired}, {active_owned} in \
                 flight, {headroom} free on this host, starting {to_start} ({limiting})"
            ),
            Self::MonitorOnlySkipped { policy } => {
                write!(f, "policy {policy}: monitor-only, skipped")
            }
            Self::RunnerStarted { policy, attempt } => {
                write!(f, "policy {policy}: started attempt {attempt}")
            }
            Self::RunnerStartFailed { policy, reason } => {
                write!(f, "policy {policy}: could not start a runner ({reason})")
            }
            Self::AllocationDeferred { policy, count } => write!(
                f,
                "policy {policy}: the allocation lock was held; {count} granted runners \
                 were not created"
            ),
            Self::MemoryShort {
                policy,
                figures,
                started: false,
            } => write!(f, "policy {policy}: waiting for memory: {figures}"),
            Self::MemoryShort {
                policy,
                figures,
                started: true,
            } => write!(
                f,
                "policy {policy}: started although the host lacks the memory, because \
                 nothing else runs here: {figures}"
            ),
            Self::AttemptsUnreadable { reason } => write!(
                f,
                "the host's attempt set could not be read ({reason}); nothing was started, \
                 and this is not the same as the host being idle"
            ),
            Self::HostCapacityUnreadable { reason } => write!(
                f,
                "the host capacity could not be read ({}); new launches wait for readable host settings",
                reason.reason()
            ),
            Self::AttemptCleaned {
                policy,
                attempt,
                outcome,
            } => write!(f, "policy {policy}: cleaned attempt {attempt} ({outcome})"),
            Self::AttemptCleanFailed {
                policy,
                attempt,
                reason,
            } => write!(
                f,
                "policy {policy}: attempt {attempt} could not be cleaned ({reason}); it \
                 will be retried"
            ),
            Self::ScaleDownRefused { policy, attempt } => write!(
                f,
                "policy {policy}: attempt {attempt} is executing a job and was not removed"
            ),
            Self::PollScheduled { retry_in_ms, pace } => {
                write!(f, "next poll in {retry_in_ms}ms ({pace})")
            }
        }
    }
}

/// Where lifecycle events go.
pub trait EventSink: fmt::Debug + Send + Sync {
    fn emit(&self, event: LifecycleEvent);
}

/// Discards everything. For callers that only want the report.
#[derive(Debug, Clone, Copy, Default)]
pub struct NoEvents;

impl EventSink for NoEvents {
    fn emit(&self, _event: LifecycleEvent) {}
}

/// The local log sink, through `d1`'s redacting layer.
///
/// Every field name below is on
/// [`runner_manager_platform::logging::ALLOWED_FIELDS`]; anything else would be
/// replaced with `[redacted]` and the line would lose its meaning rather than
/// its safety. `tests::every_field_name_this_sink_emits_is_one_d1_allows` keeps
/// that true.
#[derive(Debug, Clone, Copy, Default)]
pub struct TracingEvents;

impl EventSink for TracingEvents {
    fn emit(&self, event: LifecycleEvent) {
        let name = event.name();
        match event {
            LifecycleEvent::DemandObserved {
                policy,
                demand,
                not_matched,
                unresolvable,
                complete,
            } => {
                // An empty reading is the idle steady state, and the idle
                // interval repeats it every few seconds per target: at `info`
                // it would be most of the log. Anything worth reading is still
                // `info`.
                if demand == 0 && not_matched == 0 && unresolvable == 0 && complete {
                    tracing::debug!(
                        event = name,
                        policy_id = %policy,
                        demand,
                        not_matched,
                        unresolvable,
                        count = u64::from(complete),
                    );
                } else {
                    tracing::info!(
                        event = name,
                        policy_id = %policy,
                        demand,
                        not_matched,
                        unresolvable,
                        count = u64::from(complete),
                    );
                }
                // There is deliberately no `warn!` here for the "demand is zero
                // but jobs were not matched" shape, though it is the one this
                // change introduced: before demand was filtered, a repository
                // with work in it always produced some, and now a policy whose
                // labels do not cover its jobs produces none.
                //
                // The reason is that the shape is indistinguishable from a
                // healthy one. A repository served by a Windows host and a macOS
                // host has the other host's jobs queued in it constantly, so
                // each agent would warn on every poll about work that is being
                // served correctly by the other machine. Telling the two apart
                // needs to know whether this policy has *ever* matched anything,
                // which is state across polls that this loop does not keep.
                //
                // What an operator gets instead is the `not_matched` count, on
                // this event and in its `Display`, which `g2` renders. "0 queued
                // jobs for this host, 5 for other labels" is the diagnosis; a
                // warning that fired on every healthy minute would be the kind
                // nobody reads.
            }
            LifecycleEvent::TargetUnreadable { policy, reason } => {
                tracing::warn!(event = name, policy_id = %policy, reason);
            }
            LifecycleEvent::RoutingRefused {
                policy,
                reason,
                count,
            } => tracing::warn!(
                event = name,
                policy_id = %policy,
                reason = reason.as_str(),
                count,
                "{reason}"
            ),
            LifecycleEvent::Allocated {
                policy,
                demand,
                desired,
                active_owned,
                headroom,
                to_start,
                limiting,
            } if demand == 0 && to_start == 0 && active_owned == 0 => tracing::debug!(
                // Nothing wanted, nothing held, nothing started: the idle
                // steady state, see `DemandObserved` above.
                event = name,
                policy_id = %policy,
                demand,
                desired,
                capacity = active_owned,
                headroom,
                count = to_start,
                reason = %limiting,
            ),
            LifecycleEvent::Allocated {
                policy,
                demand,
                desired,
                active_owned,
                headroom,
                to_start,
                limiting,
            } => tracing::info!(
                event = name,
                policy_id = %policy,
                demand,
                desired,
                capacity = active_owned,
                headroom,
                count = to_start,
                reason = %limiting,
            ),
            LifecycleEvent::MonitorOnlySkipped { policy } => {
                tracing::debug!(event = name, policy_id = %policy, mode = "monitor_only");
            }
            LifecycleEvent::RunnerStarted { policy, attempt } => {
                tracing::info!(event = name, policy_id = %policy, attempt_id = %attempt);
            }
            LifecycleEvent::RunnerStartFailed { policy, reason } => {
                tracing::warn!(event = name, policy_id = %policy, reason);
            }
            LifecycleEvent::AllocationDeferred { policy, count } => {
                // Contention is normally brief, but a configured WSL recovery
                // fence whose shared root is unwritable reaches the same
                // fail-closed result forever. Demand was already observed and
                // capacity was already granted, so hiding this at DEBUG makes
                // a host claim healthy while it starts nothing.
                tracing::warn!(
                    event = name,
                    policy_id = %policy,
                    reason = "allocation_lock_unavailable",
                    count
                );
            }
            LifecycleEvent::MemoryShort {
                policy,
                figures,
                started,
            } => {
                let reason = if started {
                    "larger_than_free_memory_nothing_else_running"
                } else {
                    "insufficient_memory"
                };
                tracing::warn!(
                    event = name,
                    policy_id = %policy,
                    reason,
                    free_bytes = figures.free,
                    reserved_bytes = figures.reserved,
                    estimate_bytes = figures.estimate.bytes,
                    estimate_source = figures.estimate.source.as_str(),
                    margin_bytes = figures.margin,
                );
            }
            LifecycleEvent::AttemptsUnreadable { reason } => {
                tracing::warn!(event = name, reason);
            }
            LifecycleEvent::HostCapacityUnreadable { reason } => {
                tracing::warn!(event = name, reason = reason.reason());
            }
            LifecycleEvent::AttemptCleaned {
                policy,
                attempt,
                outcome,
            } => tracing::info!(
                event = name,
                policy_id = %policy,
                attempt_id = %attempt,
                outcome = outcome.as_str(),
            ),
            LifecycleEvent::AttemptCleanFailed {
                policy,
                attempt,
                reason,
            } => tracing::warn!(
                event = name,
                policy_id = %policy,
                attempt_id = %attempt,
                reason,
            ),
            LifecycleEvent::ScaleDownRefused { policy, attempt } => tracing::info!(
                event = name,
                policy_id = %policy,
                attempt_id = %attempt,
                attempt_state = "busy",
            ),
            LifecycleEvent::PollScheduled { retry_in_ms, pace } => {
                if pace == PollPace::Idle {
                    tracing::debug!(event = name, retry_in_ms, state = pace.as_str());
                } else {
                    tracing::info!(event = name, retry_in_ms, state = pace.as_str());
                }
            }
        }
    }
}

/// Keeps every event, in order.
///
/// `g2`'s activity view is a reader of this, and so is every test below.
#[derive(Debug, Default)]
pub struct EventLog {
    events: Mutex<Vec<LifecycleEvent>>,
}

impl EventLog {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    #[must_use]
    pub fn events(&self) -> Vec<LifecycleEvent> {
        self.events.lock().map(|e| e.clone()).unwrap_or_default()
    }

    /// How many events of one name were emitted.
    #[must_use]
    pub fn count_of(&self, name: &str) -> usize {
        self.events()
            .iter()
            .filter(|event| event.name() == name)
            .count()
    }
}

impl EventSink for EventLog {
    fn emit(&self, event: LifecycleEvent) {
        if let Ok(mut events) = self.events.lock() {
            events.push(event);
        }
    }
}

/// Both sinks at once: the log sink for the operator's file, the buffer for
/// `g2`'s screen.
#[derive(Debug)]
pub struct TeeEvents(pub Arc<dyn EventSink>, pub Arc<dyn EventSink>);

impl EventSink for TeeEvents {
    fn emit(&self, event: LifecycleEvent) {
        self.0.emit(event);
        self.1.emit(event);
    }
}

// ---------------------------------------------------------------------------
// The reconciler
// ---------------------------------------------------------------------------

/// Where a policy's demand comes from when the optional coordinator places
/// its runners (coordinator stage 3).
///
/// A profile carrying a shared pool label (`rm-pool-linux`, ...) on a target
/// the coordinator covers starts runners only for the launches the
/// coordinator placed on this host and the host accepted — never from its own
/// reading of the queue, which every host in the pool sees the same way. This
/// is a *count*, not a job: a runner still takes whatever matching job GitHub
/// hands it, nothing here reserves one, and the number still goes through
/// [`HostAllocator`] and every ceiling, exactly as a polled count does.
pub trait DemandDirector: fmt::Debug + Send + Sync {
    /// `Some(n)`: `policy`'s demand this pass is `n`, not its reading.
    /// `None`: its own reading, as without a coordinator.
    fn directed_demand(&self, policy: &ScalePolicy) -> Option<u32>;

    /// A runner was started for `policy` (as soon as it exists, whether or
    /// not its demand was directed).
    fn launched(&self, policy: &ScalePolicy, attempt: AttemptId);

    /// No runner could be started for `policy`, for a real reason: not a
    /// runner still being brought up. `attempt` is the one reported through
    /// [`Self::launched`] whose bring-up failed, if it got that far: what it was
    /// reported for is waiting again, to be retried rather than placed anew.
    fn start_failed(&self, _policy: &ScalePolicy, _attempt: Option<AttemptId>) {}

    /// Where `policy`'s oldest launch without a runner stands among this host's
    /// launches: lower goes first. `None` when it has none.
    fn next_waiting(&self, _policy: &ScalePolicy) -> Option<u64> {
        None
    }

    /// How many launches of other targets of this host are older than
    /// `policy`'s oldest without a runner and may yet reserve: `policy`'s
    /// launch lets them go first.
    fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
        0
    }

    /// Those older launches did not reserve in [`OLDER_LAUNCH_TURN`], so nothing
    /// waits for them any more, unless they start waiting afresh.
    fn pass_over_older(&self, _policy: &ScalePolicy) {}
}

/// How long a launch waits for older launches of other targets while none of
/// them reserves. Every target's loop reserves as soon as it is told of a launch
/// (see [`Reconciler::serve_launches`]), so this is not a turn anyone is expected
/// to use up: it is for a loop that cannot reserve at all, a target added a moment
/// ago that has no loop yet, or a host-wide lock refused. It starts over whenever
/// one of them reserves, and costs one launch that one wait, not one per launch.
const OLDER_LAUNCH_TURN: Duration = Duration::from_secs(2);
const OLDER_LAUNCH_POLL: Duration = Duration::from_millis(10);

/// How soon an accepted launch that did not start is tried again.
const LAUNCH_RETRY: Duration = Duration::from_secs(30);

/// Where a reservation's wait went, for its launch's timings.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ReservationWaits {
    /// Letting older launches of this host reserve first.
    pub order: Duration,
    /// For the allocation lock, and from the reservation to its bring-up.
    pub lock: Duration,
}

/// What one round of serving launches has reserved so far.
#[derive(Default)]
struct ReservationRound {
    /// Per policy, how many it reserved and the most launches seen open: a lease
    /// closing during the round must not cut short the launches after it.
    reserved: BTreeMap<PolicyId, (u32, u32)>,
    exhausted: BTreeSet<PolicyId>,
}

/// One step of a round: what it reserved, the round, and the policies it read.
type Reserving<'a> = std::pin::Pin<
    Box<
        dyn std::future::Future<Output = (Option<Reserved>, ReservationRound, Vec<ScalePolicy>)>
            + Send
            + 'a,
    >,
>;

/// A runner reserved under the allocation lock, to bring up outside it.
struct Reserved {
    policy: ScalePolicy,
    attempt: RunnerAttempt,
    waits: ReservationWaits,
    reserved_at: std::time::Instant,
    /// Let go when this runner is up or has failed.
    _hold: LaunchHold,
}

/// Whether the host has the memory for one more runner: asked under the
/// allocation lock before every start, after every ceiling granted it, so a
/// directed launch and a host's own demand meet the same check. See
/// [`runner_manager_domain::memory`].
pub trait MemoryGate: fmt::Debug + Send + Sync {
    /// Whether one more runner of `policy` may start, given every attempt on
    /// the host (those this pass reserved included).
    fn admit(&self, policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict;

    /// `policy` was granted a runner and memory did not stop it: a waiting
    /// state recorded for it is over.
    fn admitted(&self, _policy: &ScalePolicy) {}
}

/// No memory admission: every start the ceilings grant goes ahead.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoMemoryGate;

impl MemoryGate for NoMemoryGate {
    fn admit(&self, _policy: &ScalePolicy, _attempts: &[RunnerAttempt]) -> MemoryVerdict {
        MemoryVerdict::Unchecked
    }
}

/// Every policy's demand is its own reading: no coordinator places runners.
#[derive(Debug, Default, Clone, Copy)]
pub struct OwnDemand;

impl DemandDirector for OwnDemand {
    fn directed_demand(&self, _policy: &ScalePolicy) -> Option<u32> {
        None
    }

    fn launched(&self, _policy: &ScalePolicy, _attempt: AttemptId) {}
}

/// Everything one reconciler needs, written down at the call site.
///
/// A struct rather than seven positional arguments, for the reason `b1` gives at
/// `PersistedAttempt`: several of these are `Arc<dyn …>` and transposing two of
/// them type-checks. Construct it with a struct literal so every port is named.
#[derive(Clone)]
pub struct ReconcilerPorts {
    pub demand: Arc<dyn DemandSource>,
    pub launcher: Arc<dyn RunnerLauncher>,
    pub lock: Arc<dyn AllocationLock>,
    pub directory: Arc<dyn RepositoryDirectory>,
    pub clock: Arc<dyn Clock>,
    pub jitter: Arc<dyn Jitter>,
    pub events: Arc<dyn EventSink>,
}

impl fmt::Debug for ReconcilerPorts {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ReconcilerPorts").finish_non_exhaustive()
    }
}

/// What one reconciliation pass did.
///
/// `started` and the allocations are reported separately on purpose: an
/// allocation is what the pass *decided* under the lock, and `started` is what
/// actually came up. They differ when a launch fails or when the lock was held,
/// and collapsing them would hide both.
#[derive(Debug, Clone, Default)]
pub struct ReconcileReport {
    /// One entry per policy that got as far as being allocated for.
    pub allocations: Vec<Allocation>,
    /// Policies skipped because they are monitor-only (D19).
    pub monitor_only: Vec<PolicyId>,
    /// Policies whose target could not be polled this pass.
    pub unreadable: Vec<PolicyId>,
    /// Policies whose target GitHub actually answered for this pass.
    ///
    /// The counterpart to [`Self::unreadable`], and the only honest evidence
    /// that this host reached GitHub at all. [`Self::allocations`] is not: a
    /// policy this host does not own is allocated for with no demand and
    /// without any target being polled, so a pass where every poll failed can
    /// still end with allocations in it.
    pub targets_read: u16,
    /// Runners actually started.
    pub started: u16,
    /// Pre-acceptance attempts routed back through this pass's ordinary
    /// demand/capacity decision.
    pub replacement_intents: u16,
    /// Terminal attempts whose runtime was removed.
    pub cleaned: u16,
    /// Of those, the surplus case: registered, got no job, exited on its idle
    /// timeout. **Not** a failure.
    pub idle_exits: u16,
    /// Of those, the ones an operator should look at.
    pub failures: u16,
    /// Runners this pass was granted but did not start because the allocation
    /// lock was held.
    ///
    /// **Grants, not policies.** It used to be incremented once per
    /// `start_runners` call that met a held lock, so a policy that launched two
    /// of five and then lost the lock reported `1` while three runners went
    /// unstarted -- a number that agreed with neither its own name nor its
    /// documentation.
    pub deferred: u16,
    /// Runners this pass was granted but did not start because the host
    /// lacked the memory for them. Their demand stays queued.
    pub memory_deferred: u16,
    /// Times the host's attempt set could not be read this pass.
    ///
    /// Non-zero means the pass decided less than it looks like it decided: a
    /// policy whose attempt set was unreadable started nothing and is *not* in
    /// [`Self::allocations`], because there was no set to compute an allocation
    /// from. It is not the same as the host being idle, which is the whole
    /// reason [`RunnerLauncher::attempts`] is fallible.
    ///
    /// **A count, where [`Self::unreadable`] is a `Vec<PolicyId>`, and that
    /// asymmetry is deliberate.** An unreadable *target* is a fact about one
    /// policy's GitHub target; an unreadable *attempt set* is a fact about this
    /// host's journal, which no policy owns — two of the three paths that reach
    /// it (`clean_terminal_attempts` and `scale_down`) have no policy in hand at
    /// all. Naming policies here would mean either inventing an owner for a
    /// host-wide failure or reporting a partial list, and both read as more
    /// precision than there is. The pass is distinguishable from an idle one,
    /// which is what the field exists for; the per-policy attribution is not
    /// available, and is recorded as missing rather than faked.
    pub attempts_unreadable: u16,
    /// Allocation checks blocked by a missing or unreadable host ceiling.
    pub host_capacity_unreadable: u16,
    /// Terminal attempts whose runtime could not be removed. Retried next pass.
    pub clean_failures: u16,
    /// The most severe failure across the targets polled, when there was one.
    pub failure: Option<RefreshState>,
    /// What to display while GitHub is unreachable, including how long the
    /// outage has run and therefore whether queued work has already been lost.
    pub offline: Option<OfflineState>,
    /// When to poll next, and why then.
    pub next_poll: NextPoll,
    /// Demand requests this pass projected against the shared hourly ceiling.
    pub demand_requests: u32,
}

impl ReconcileReport {
    /// Whether this pass actually reached GitHub, which is the only thing that
    /// entitles it to write a `last GitHub contact`.
    ///
    /// # Positive evidence, because the absence of a failure is not evidence
    ///
    /// The record used to be written whenever [`Self::failure`] was `None`, on
    /// the belief that an unauthorized target lands in [`Self::unreadable`]
    /// rather than in `failure`. **That belief is wrong.** `unreadable` is
    /// pushed only from the `PollOutcome::Failed` arm, `failure` is the maximum
    /// over every `Failed` reading, and `RefreshState::Unauthorized` scores 2 —
    /// so a non-empty `unreadable` always implies `failure.is_some()`, and
    /// guarding on both would have changed nothing at all.
    ///
    /// The path that really writes a contact record without touching GitHub is
    /// a pass that polls **nothing**: every policy draining, owned by another
    /// host, or monitor-only. `pollable` is then empty, no reading exists, no
    /// failure is computed, and the old guard passed. That is how
    /// `service status` can answer `healthy` on a host doing nothing at all.
    ///
    /// So this asks for evidence rather than for the absence of a complaint. A
    /// pass with nothing to ask reaches nobody and records nothing, which is
    /// what `never` in `service status` is for.
    ///
    /// Conservative on purpose: `repositories.scope_for` is a real request that
    /// can succeed before a demand poll fails, and it is not counted. Contact
    /// that cannot be proven is not claimed.
    #[must_use]
    pub const fn reached_github(&self) -> bool {
        self.targets_read > 0
    }
}

impl Default for NextPoll {
    fn default() -> Self {
        Self {
            delay: PollSchedule::floor(),
            pace: PollPace::Nominal,
        }
    }
}

impl ReconcileReport {
    /// Whether GitHub was unreachable this pass.
    #[must_use]
    pub fn is_offline(&self) -> bool {
        matches!(self.failure, Some(RefreshState::Offline))
    }

    /// The offline state to display, when this pass was one.
    #[must_use]
    pub const fn offline_state(&self) -> Option<&OfflineState> {
        self.offline.as_ref()
    }

    /// Attempts this pass created. The idle-host assertion reads this.
    #[must_use]
    pub const fn starts_nothing(&self) -> bool {
        self.started == 0
    }
}

/// What one scale-down request did.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScaleDownReport {
    /// Terminal attempts whose runtime was removed.
    pub removed: u16,
    /// Attempts executing a job. **Removed nothing, left `busy`.**
    pub refused_busy: u16,
    /// Terminal attempts whose runtime could not be removed.
    pub clean_failures: u16,
    /// Live attempts that are not yet busy. Also removed nothing: capacity is
    /// reclaimed only when an attempt reaches a terminal state.
    pub retained: u16,
    /// The host's attempt set could not be read, so **every other field here is
    /// meaningless** rather than zero.
    ///
    /// This is the same distinction [`ReconcileReport::attempts_unreadable`]
    /// draws, and it is here for the same reason: a default
    /// [`ScaleDownReport`] and a scale-down that could not see the machine are
    /// both all-zeros, and they mean opposite things — "there was nothing to
    /// reclaim" against "we do not know what there was". Check
    /// [`Self::is_conclusive`] before reading a zero as an answer.
    pub attempts_unreadable: bool,
}

impl ScaleDownReport {
    /// Whether the counts here describe the machine at all.
    ///
    /// `false` means the attempt set could not be read, so every zero is
    /// "unknown" rather than "none".
    #[must_use]
    pub const fn is_conclusive(&self) -> bool {
        !self.attempts_unreadable
    }
}

/// How long a failed cleanup of one attempt waits before it is tried again,
/// doubling per consecutive failure up to [`CLEAN_RETRY_MAXIMUM`].
///
/// A runtime that cannot be removed used to be retried on every pass, about
/// every six seconds, and each failure logs a warning: one stuck Windows
/// workspace wrote some 15,000 of them a day. The retry is still real; it is
/// just no longer a log flood.
const CLEAN_RETRY_INITIAL: Duration = Duration::from_secs(30);

/// The longest a failing cleanup waits between tries.
const CLEAN_RETRY_MAXIMUM: Duration = Duration::from_secs(30 * 60);

/// One attempt's consecutive cleanup failures, and when it may be tried again.
#[derive(Debug, Clone, Copy)]
struct CleanRetry {
    failures: u32,
    not_before: Timestamp,
}

/// Per-attempt cleanup backoff. One each in the [`Reconciler`]'s sweep of
/// concluded attempts and in the lifecycle launcher's own supervision pass,
/// the two places a failed cleanup is retried.
#[derive(Debug, Default)]
pub(crate) struct CleanBackoff {
    retries: BTreeMap<AttemptId, CleanRetry>,
}

impl CleanBackoff {
    /// The doubling-with-a-ceiling schedule; the attempt bound is unused.
    const SCHEDULE: crate::lifecycle::RetryPolicy =
        crate::lifecycle::RetryPolicy::bounded(u32::MAX, CLEAN_RETRY_INITIAL, CLEAN_RETRY_MAXIMUM);

    /// Whether `attempt`'s cleanup should be tried at `now`.
    pub(crate) fn is_due(&self, attempt: AttemptId, now: Timestamp) -> bool {
        self.retries
            .get(&attempt)
            .is_none_or(|retry| now >= retry.not_before)
    }

    /// Records a failed cleanup and schedules the next try.
    pub(crate) fn failed(&mut self, attempt: AttemptId, now: Timestamp) {
        let failures = self
            .retries
            .get(&attempt)
            .map_or(1, |retry| retry.failures.saturating_add(1));
        let delay = Self::SCHEDULE.delay(failures);
        self.retries.insert(
            attempt,
            CleanRetry {
                failures,
                // At most `CLEAN_RETRY_MAXIMUM`, so always representable.
                not_before: now + chrono::Duration::from_std(delay).unwrap_or_default(),
            },
        );
    }

    /// Forgets `attempt`: it was cleaned.
    pub(crate) fn succeeded(&mut self, attempt: AttemptId) {
        self.retries.remove(&attempt);
    }

    /// Forgets every attempt not in `present`, so the map cannot outgrow the
    /// journal.
    fn retain(&mut self, present: &BTreeSet<AttemptId>) {
        self.retries.retain(|attempt, _| present.contains(attempt));
    }
}

/// The reconciliation loop.
///
/// One per target, as `f3` runs them; they share a [`RunnerLauncher`] and an
/// [`AllocationLock`], which is what keeps the host ceiling true across all of
/// them.
#[derive(Debug)]
pub struct Reconciler {
    host: Host,
    demand: Arc<dyn DemandSource>,
    launcher: Arc<dyn RunnerLauncher>,
    lock: Arc<dyn AllocationLock>,
    repositories: RepositoryCache,
    clock: Arc<dyn Clock>,
    jitter: Arc<dyn Jitter>,
    events: Arc<dyn EventSink>,
    schedule: PollSchedule,
    clean_backoff: CleanBackoff,
    director: Arc<dyn DemandDirector>,
    /// What serves this target's launches beside its passes, when something
    /// does: see [`Self::with_launch_side`].
    launch_side: Option<Arc<tokio::sync::Notify>>,
    memory: Arc<dyn MemoryGate>,
}

impl Reconciler {
    /// Build a reconciler polling at the host's configured interval.
    #[must_use]
    pub fn new(host: Host, ports: ReconcilerPorts) -> Self {
        let interval = host.refresh_interval;
        let repositories = RepositoryCache::new(
            Arc::clone(&ports.directory),
            Arc::clone(&ports.clock),
            interval,
        );
        Self {
            host,
            demand: ports.demand,
            launcher: ports.launcher,
            lock: ports.lock,
            repositories,
            clock: ports.clock,
            jitter: ports.jitter,
            events: ports.events,
            schedule: PollSchedule::new(interval),
            clean_backoff: CleanBackoff::default(),
            director: Arc::new(OwnDemand),
            launch_side: None,
            memory: Arc::new(NoMemoryGate),
        }
    }

    /// Checks every start against the host's memory. See [`MemoryGate`].
    #[must_use]
    pub fn with_memory_gate(mut self, memory: Arc<dyn MemoryGate>) -> Self {
        self.memory = memory;
        self
    }

    /// Leaves this reconciler's directed launches to whatever `signal` wakes
    /// ([`Self::serve_launches`]). A pass then only reports on them. A pass that
    /// brought one up itself would be held for as long as its package takes, and
    /// a pass is what concludes a runner that exited and frees its slot.
    #[must_use]
    pub fn with_launch_side(mut self, signal: Arc<tokio::sync::Notify>) -> Self {
        self.launch_side = Some(signal);
        self
    }

    /// Lets `director` replace a policy's polled demand with the launches the
    /// coordinator placed here. See [`DemandDirector`].
    #[must_use]
    pub fn with_director(mut self, director: Arc<dyn DemandDirector>) -> Self {
        self.director = director;
        self
    }

    #[must_use]
    pub const fn host(&self) -> &Host {
        &self.host
    }

    #[must_use]
    pub const fn schedule(&self) -> &PollSchedule {
        &self.schedule
    }

    /// Adopt intervals an operator changed while the loop runs.
    pub fn set_intervals(&mut self, active: RefreshInterval, idle: Duration) {
        self.host.refresh_interval = active;
        self.schedule.set_intervals(active, idle);
    }

    /// The repository-list cache, so `f1` can report what it has spent.
    #[must_use]
    pub const fn repositories(&self) -> &RepositoryCache {
        &self.repositories
    }

    /// Starts the runners of `policies`' coordinator launches, and nothing
    /// else: no GitHub poll, no supervision. Every launch open at that moment is
    /// served by one call. See [`Self::serve_launches`].
    #[cfg(test)]
    pub async fn start_directed(&self, policies: &[ScalePolicy]) -> ReconcileReport {
        let mut report = ReconcileReport::default();
        let mut round = ReservationRound::default();
        let mut reserved = Vec::new();
        while let Some(next) = self
            .reserve_next_launch(policies, &mut round, &mut report)
            .await
        {
            reserved.push(next);
        }
        self.finish_reserved(reserved, &mut report).await;
        report
    }

    /// Serves a target's coordinator launches until `stop`: on `signal`, and
    /// [`LAUNCH_RETRY`] after one that did not start, it reserves a runner for
    /// every launch open at that moment, one at a time, and brings each up as soon
    /// as it is reserved. A launch the coordinator placed here needs no reading of
    /// the queue, so this runs beside the target's passes: a launch accepted while
    /// one is polling GitHub, or while runners already reserved are coming up, is
    /// reserved and brought up at once rather than after them.
    ///
    /// It goes through the same allocator, ceilings and host-wide lock as
    /// [`Self::reconcile`]. Neither needs to know of the other's runners: each
    /// reads the journal under the lock, which holds every reservation. What it
    /// reserves is the launcher's to keep apart from supervision while it comes
    /// up (`LifecycleLauncher::supervise`), and the next full pass supervises it.
    /// After `stop` it reserves nothing more and returns once what it reserved is
    /// up or has failed.
    pub async fn serve_launches(
        &self,
        policies: &(dyn Fn() -> Vec<ScalePolicy> + Send + Sync),
        signal: &tokio::sync::Notify,
        mut stop: tokio::sync::watch::Receiver<bool>,
    ) {
        use futures::StreamExt as _;

        let mut bringing_up = futures::stream::FuturesUnordered::new();
        let mut reserving: Option<Reserving<'_>> = None;
        let mut retry_at: Option<tokio::time::Instant> = None;
        let mut known: Vec<ScalePolicy> = Vec::new();
        let mut stopping = *stop.borrow();
        loop {
            if stopping && reserving.is_none() && bringing_up.is_empty() {
                return;
            }
            tokio::select! {
                () = signal.notified(), if !stopping && reserving.is_none() => {
                    reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
                }
                () = async {
                    match retry_at {
                        Some(at) => tokio::time::sleep_until(at).await,
                        None => std::future::pending().await,
                    }
                }, if !stopping && reserving.is_none() => {
                    retry_at = None;
                    reserving = Some(self.reserve_step(policies(), ReservationRound::default()));
                }
                (reserved, round, read) = async {
                    match reserving.as_mut() {
                        Some(reserving) => reserving.await,
                        None => std::future::pending().await,
                    }
                } => {
                    reserving = None;
                    match reserved {
                        Some(reserved) => {
                            bringing_up.push(self.bring_up(reserved));
                            known = read;
                            if !stopping {
                                reserving = Some(self.reserve_step(policies(), round));
                            }
                        }
                        None => known = read,
                    }
                }
                Some(_) = bringing_up.next(), if !bringing_up.is_empty() => {}
                changed = stop.changed(), if !stopping => {
                    stopping = changed.is_err() || *stop.borrow();
                }
            }
            // One that did not start, a registration that failed, is tried again
            // well inside its lease's lifetime.
            if reserving.is_none()
                && retry_at.is_none()
                && known
                    .iter()
                    .any(|policy| self.director.next_waiting(policy).is_some())
            {
                retry_at = Some(tokio::time::Instant::now() + LAUNCH_RETRY);
            }
        }
    }

    /// [`Self::reserve_next_launch`], as a future that owns what it needs.
    fn reserve_step(
        &self,
        policies: Vec<ScalePolicy>,
        mut round: ReservationRound,
    ) -> Reserving<'_> {
        Box::pin(async move {
            let mut report = ReconcileReport::default();
            let reserved = self
                .reserve_next_launch(&policies, &mut round, &mut report)
                .await;
            (reserved, round, policies)
        })
    }

    /// Reserves one runner for the oldest of `policies`' open launches: a target's
    /// own profiles keep the order too. Slow work comes after this, in
    /// [`Self::bring_up`], so no launch's package holds up another's reservation.
    /// `None` when `round` has served all it can.
    async fn reserve_next_launch(
        &self,
        policies: &[ScalePolicy],
        round: &mut ReservationRound,
        report: &mut ReconcileReport,
    ) -> Option<Reserved> {
        loop {
            let policy = policies
                .iter()
                .filter(|policy| {
                    policy.owns_runners()
                        && policy.is_owned_by(self.host.id)
                        && policy.may_start_runners()
                        && !round.exhausted.contains(&policy.id)
                })
                .min_by_key(|policy| self.director.next_waiting(policy).unwrap_or(u64::MAX))?;
            let (served, most) = round.reserved.entry(policy.id).or_default();
            // Bounded by the launches open, however late the journal shows these.
            if let Some(directed) = self.director.directed_demand(policy)
                && *served < directed.max(*most)
            {
                *most = directed.max(*most);
                let mut reserved = Vec::new();
                self.start_runners(policy, directed, 1, report, &mut reserved)
                    .await;
                if let Some(next) = reserved.pop() {
                    *served += 1;
                    return Some(next);
                }
            }
            round.exhausted.insert(policy.id);
        }
    }

    /// One reconciliation pass over `policies`.
    ///
    /// The order of operations is `03-control-flows.md` flow 2, and the two
    /// steps most worth naming are the ones that are silent when they are wrong:
    ///
    /// * **Monitor-only policies are removed before the demand poll**, not
    ///   after. D19 says such a policy "is skipped entirely by reconciliation",
    ///   and a poll issued on its behalf would spend requests from the shared
    ///   ceiling for a policy that can never act on the answer. This is asserted
    ///   on [`ScalePolicy::owns_runners`] rather than deduced from
    ///   `max_capacity` being absent.
    /// * **The attempt set is re-read under the lock, once per runtime.** See
    ///   [`RunnerLauncher`] for why it comes from there and nowhere else.
    pub async fn reconcile(&mut self, policies: &[ScalePolicy]) -> ReconcileReport {
        let mut report = ReconcileReport::default();
        // Everything this pass has created, carried across policies so that the
        // host-wide total cannot be computed from a set that is missing it. See
        // `RunnerLauncher::launch`.
        let mut reserved: Vec<Reserved> = Vec::new();

        // --- Flow 2.1-2.2: who is even asking, and what did GitHub say -------
        let mut pollable: Vec<&ScalePolicy> = Vec::new();
        let mut supervision_failed = BTreeSet::new();
        for policy in policies {
            if !policy.owns_runners() {
                report.monitor_only.push(policy.id);
                self.events
                    .emit(LifecycleEvent::MonitorOnlySkipped { policy: policy.id });
                continue;
            }
            if !policy.is_owned_by(self.host.id) {
                // Ownership rule 2 and precedence rule 4. The allocator reports
                // both by name below; polling on their behalf would spend
                // requests for an answer that cannot be acted on.
                continue;
            }
            match self.launcher.supervise(policy).await {
                Ok(intents) => {
                    report.replacement_intents = report
                        .replacement_intents
                        .saturating_add(u16::try_from(intents.len()).unwrap_or(u16::MAX));
                }
                Err(failure) => {
                    supervision_failed.insert(policy.id);
                    self.report_unreadable_attempts(&mut report, &failure);
                    continue;
                }
            }
            if !policy.may_start_runners() {
                continue;
            }
            pollable.push(policy);
        }

        let readings = self.poll_targets(&pollable, &mut report).await;
        report.targets_read = u16::try_from(
            readings
                .values()
                .filter(|outcome| matches!(outcome, PollOutcome::Ready(_)))
                .count(),
        )
        .unwrap_or(u16::MAX);
        let mut routed = BTreeMap::new();
        for (target, outcome) in &readings {
            let PollOutcome::Ready(reading) = outcome else {
                continue;
            };
            let profiles: Vec<&ScalePolicy> = pollable
                .iter()
                .copied()
                .filter(|policy| &policy.target == target)
                .collect();
            let (tallies, refusals) = route_demand(policies, &profiles, reading);
            if let Some(first) = profiles.first() {
                for (reason, count) in refusals {
                    self.events.emit(LifecycleEvent::RoutingRefused {
                        policy: first.id,
                        reason,
                        count,
                    });
                }
            }
            routed.insert(target.clone(), tallies);
        }

        // --- Flow 2.8: terminal attempts, whatever else this pass does -------
        //
        // Run before the allocation phase so that a report's `cleaned` count
        // describes the same instant its allocations do. It does not change the
        // arithmetic: a terminal attempt already stopped counting against
        // capacity when it became terminal, which is `b1`'s
        // `counts_against_capacity`. It touches no live process, so it is also
        // safe during an outage — flow 3.3 requires that running runners be
        // retained, and nothing here can reach one.
        self.clean_terminal_attempts(&mut report).await;

        // --- Flow 2.3-2.6: the allocation -----------------------------------
        //
        // The predicates are re-tested here rather than the reading being looked
        // up by target, and that is not redundancy. **Targets are shared.** A
        // monitor-only policy watching `acme/app` alongside an autoscale policy
        // on the *same* repository finds a reading in the map that the other
        // policy paid for, and a lookup-driven loop then serves it: it emits a
        // demand observation on its behalf and clamps a number it has no
        // business seeing.
        //
        // Nothing downstream goes wrong when that happens — `may_start_runners`
        // is false for a monitor-only policy, so `HostAllocator` refuses it and
        // `to_start` is zero. It simply is not *skipped*, and D19's word is
        // "entirely".
        for policy in policies {
            if !policy.owns_runners() {
                // Already recorded and reported above, before any demand request
                // was issued. It owns no routing labels, takes no part in
                // demand, and can never be the reason a runner starts. Asserted
                // on the mode rather than deduced from `max_capacity` being
                // absent, which is what the specification requires.
                continue;
            }
            if !policy.is_owned_by(self.host.id) || !policy.may_start_runners() {
                // Ownership rule 2 and precedence rule 4. Allocated for with no
                // demand, so the refusal is reported by name rather than by
                // absence.
                match self.allocate_only(policy, 0, &reserved).await {
                    Ok(allocation) => {
                        self.emit_allocation(&allocation);
                        report.allocations.push(allocation);
                    }
                    Err(failure) => self.report_allocation_failure(&mut report, &failure),
                }
                continue;
            }
            if supervision_failed.contains(&policy.id) {
                continue;
            }
            // Coordinated placement: the launches the coordinator placed
            // here are the demand, whatever this host's own reading says (or
            // whether it could read at all).
            if let Some(directed) = self.director.directed_demand(policy) {
                let Some(launch_side) = &self.launch_side else {
                    self.start_runners(policy, directed, u16::MAX, &mut report, &mut reserved)
                        .await;
                    continue;
                };
                match self.allocate_only(policy, directed, &reserved).await {
                    Ok(allocation) => {
                        if !allocation.starts_nothing() {
                            launch_side.notify_one();
                        }
                        report.allocations.push(allocation);
                    }
                    Err(failure) => self.report_allocation_failure(&mut report, &failure),
                }
                continue;
            }
            let Some(reading) = readings.get(&policy.target) else {
                // Unreachable: every policy reaching here was in `pollable`, and
                // `poll_targets` inserts an outcome for each of their targets.
                debug_assert!(false, "a pollable policy's target has no reading");
                continue;
            };
            match reading {
                PollOutcome::Failed(state) => {
                    report.unreadable.push(policy.id);
                    self.events.emit(LifecycleEvent::TargetUnreadable {
                        policy: policy.id,
                        reason: unreadable_reason(state),
                    });
                }
                PollOutcome::Ready(demand) => {
                    let tally = routed
                        .get(&policy.target)
                        .and_then(|tallies| tallies.get(&policy.id))
                        .expect("every pollable profile has a routed tally");
                    let count = tally.demand();
                    self.events.emit(LifecycleEvent::DemandObserved {
                        policy: policy.id,
                        demand: count,
                        not_matched: tally.not_matched,
                        unresolvable: u32::try_from(tally.unresolvable.len()).unwrap_or(u32::MAX),
                        complete: demand.is_complete(),
                    });
                    self.start_runners(policy, count, u16::MAX, &mut report, &mut reserved)
                        .await;
                }
            }
        }
        self.finish_reserved(reserved, &mut report).await;

        // --- Flow 2.1 / 3.3: when to come back ------------------------------
        let failure = readings
            .values()
            .filter_map(PollOutcome::failure)
            .max_by_key(|state| severity(state))
            .cloned();
        let now = self.clock.now();
        let conditions = PollConditions {
            quiet: is_quiet(&readings, &report),
            requests_per_pass: report.demand_requests,
            quota: readings
                .values()
                .filter_map(PollOutcome::reading)
                .filter_map(QueuedDemand::rate_limit)
                .max_by_key(|quota| quota.observed_at)
                .copied(),
        };
        report.next_poll =
            self.schedule
                .next_poll_with(failure.as_ref(), conditions, now, self.jitter.as_ref());
        report.failure = failure;
        // Flow 3.3's fourth obligation: the offline state carries the 24-hour
        // bound, and it can only say whether that bound has passed if it is
        // given the real elapsed time rather than an estimate from the interval.
        if let PollPace::Offline { consecutive } = report.next_poll.pace {
            let state = OfflineState::new(consecutive, report.next_poll.delay);
            report.offline = Some(match self.schedule.offline_for(now) {
                Some(elapsed) => state.since(elapsed),
                None => state,
            });
        }
        self.events.emit(LifecycleEvent::PollScheduled {
            retry_in_ms: u64::try_from(report.next_poll.delay.as_millis()).unwrap_or(u64::MAX),
            pace: report.next_poll.pace,
        });

        report
    }

    /// Poll each distinct target once, however many policies share it.
    ///
    /// Two policies on one repository are one demand request, not two. That is
    /// not a micro-optimisation: the budget model in
    /// `04-subsystem-contracts.md` prices a *target*, and a loop that spent per
    /// policy would quietly exceed the projection `f2` admitted the
    /// configuration against.
    async fn poll_targets(
        &self,
        pollable: &[&ScalePolicy],
        report: &mut ReconcileReport,
    ) -> BTreeMap<ScaleTarget, PollOutcome> {
        let targets: BTreeSet<ScaleTarget> = pollable.iter().map(|p| p.target.clone()).collect();

        let mut readings = BTreeMap::new();
        for target in targets {
            let outcome = match self.repositories.scope_for(&target).await {
                Ok(scope) => {
                    report.demand_requests = report
                        .demand_requests
                        .saturating_add(demand_requests_per_poll(&scope));
                    self.demand.poll(&scope).await
                }
                // The repository list could not be refreshed, so the scope of
                // the poll is unknown. Polling a stale or empty scope would
                // report a demand number for a set of repositories nobody
                // chose, which is worse than reporting that the target could
                // not be read.
                Err(error) => PollOutcome::Failed(RefreshState::from_error(&error)),
            };
            readings.insert(target, outcome);
        }
        readings
    }

    /// Compute one policy's allocation without creating anything.
    ///
    /// # Why this is safe without the lock
    ///
    /// **Not** because it cannot grant — it can, and
    /// [`Reconciler::start_runners`] uses it as a pre-check precisely for the
    /// number it returns. That was the original reason and this function
    /// outgrew it; the reason now is that it *decides* nothing. Nothing is
    /// created here, the headroom it read is re-read under the lock before any
    /// runtime exists, and the under-lock allocation may only lower what this
    /// one proposed. So there is no read-decide-create sequence here to make
    /// atomic, and the worst this can be is optimistic — which the lock then
    /// corrects.
    ///
    /// # Errors
    /// A host-ceiling or attempt-journal read failure. Neither permits an
    /// allocation, and they remain distinct in diagnostics.
    async fn allocate_only(
        &self,
        policy: &ScalePolicy,
        demand: u32,
        reserved: &[Reserved],
    ) -> Result<Allocation, AllocationFailure> {
        self.allocate_over(policy, demand, reserved)
            .await
            .map(|(allocation, _)| allocation)
    }

    /// [`Self::allocate_only`], and the attempt set it was decided over.
    async fn allocate_over(
        &self,
        policy: &ScalePolicy,
        demand: u32,
        reserved: &[Reserved],
    ) -> Result<(Allocation, Vec<RunnerAttempt>), AllocationFailure> {
        // Read again under the shared launch lock, not just at each target's
        // poll boundary: another target must not use an old, higher ceiling.
        let mut host = self.host.clone();
        host.host_capacity = self
            .launcher
            .host_capacity(&self.host)
            .await
            .map_err(AllocationFailure::HostCapacity)?;
        let attempts = self
            .host_attempts(reserved)
            .await
            .map_err(AllocationFailure::Attempts)?;
        let mut allocator = HostAllocator::from_attempts(&host, &attempts);
        let allocation = allocator.allocate(policy, demand);
        Ok((allocation, attempts))
    }

    /// Flow 2.4-2.6: start runners for one policy, one lock hold per runtime.
    ///
    /// # Two stopping conditions, and both are needed
    ///
    /// The loop re-reads the attempt set under every hold, so the obvious stop
    /// is "the allocator granted nothing". That condition **alone does not
    /// terminate**, and the failure is not hypothetical — it was measured.
    /// Handing the allocator a set that does not include the runners this loop
    /// just started (an empty one, a stale one, or a launcher whose journal
    /// write has not landed yet) makes every grant look like the first, and the
    /// pass starts runners until something outside it intervenes. With the set
    /// dropped entirely, the three-consecutive-polls test below does not report
    /// three attempts; it *never returns*.
    ///
    /// So the grant decided on the first hold is also a **budget**. A later hold
    /// may lower it — the host may have filled up meanwhile — and can never
    /// raise it, which bounds the pass at the number this policy was actually
    /// allocated. That is `c2`'s reasoning for `MAX_PAGES` one layer down: the
    /// reconciliation loop is the one place in this product that must not be
    /// able to wedge, so the bound is structural rather than a consequence of
    /// every input being well behaved.
    ///
    /// At most `limit` runners: one, when the caller decides who is next.
    async fn start_runners(
        &self,
        policy: &ScalePolicy,
        demand: u32,
        limit: u16,
        report: &mut ReconcileReport,
        reserved: &mut Vec<Reserved>,
    ) {
        // A lock-free pre-check, for one reason only: the host-wide lock should
        // not be taken by a policy that is going to be granted nothing. On an
        // idle host with P policies that was P lock acquisitions per poll --
        // free under `InProcessAllocationLock`, a `spawn_blocking` and a
        // filesystem lock apiece under `FileAllocationLock`.
        //
        // It is safe because it can only be optimistic. Anything it grants is
        // re-decided under the lock below and may be lowered there; the only
        // thing it can get wrong in the other direction is refusing a grant that
        // headroom freed a moment later would have allowed, which the next poll
        // picks up.
        let intent = match self.allocate_only(policy, demand, reserved).await {
            Ok(intent) => intent,
            Err(failure) => {
                self.report_allocation_failure(report, &failure);
                self.director.start_failed(policy, None);
                return;
            }
        };

        // The allocation that is *reported* is the one taken under the lock when
        // a lock was taken, because that is the one that decided anything. The
        // pre-check stands in only when no hold was ever obtained.
        let mut decided: Option<Allocation> = None;
        let mut budget = intent.to_start.min(limit);

        while budget > 0 {
            let order_wait = self.let_older_launches_reserve(policy).await;
            // A launch may have been cancelled while this one waited.
            let demand = if order_wait.is_zero() {
                demand
            } else {
                self.director.directed_demand(policy).unwrap_or(demand)
            };
            let waiting_since = std::time::Instant::now();
            let guard = match self.lock.acquire().await {
                Ok(guard) => guard,
                Err(_) => {
                    // Grants, not policies: this is what the policy was owed and
                    // did not get.
                    report.deferred = report.deferred.saturating_add(budget);
                    self.events.emit(LifecycleEvent::AllocationDeferred {
                        policy: policy.id,
                        count: budget,
                    });
                    // The lock is held briefly now, so this is not a turn: a
                    // Windows drain, the fence, or another process.
                    self.director.start_failed(policy, None);
                    break;
                }
            };
            let lock_wait = waiting_since.elapsed();

            // The read and the decision are both inside the hold, and so is the
            // journalled reservation below. Two concurrent passes therefore serialise on the
            // whole sequence rather than on the decision alone -- reading the
            // headroom outside the lock is the shape in which two policies both
            // find room for the last slot.
            let (allocation, attempts) = match self.allocate_over(policy, demand, reserved).await {
                Ok(decided) => decided,
                Err(failure) => {
                    drop(guard);
                    self.report_allocation_failure(report, &failure);
                    self.director.start_failed(policy, None);
                    break;
                }
            };

            if decided.is_none() {
                // The under-lock decision may be smaller than the pre-check, and
                // never larger: `min` rather than assignment, so a later hold
                // cannot raise the bound either.
                budget = budget.min(allocation.to_start);
                decided = Some(allocation.clone());
            }

            // Either stop is sufficient on its own in the well-behaved case;
            // neither is sufficient when the launcher lags. See the doc comment.
            if allocation.starts_nothing() || budget == 0 {
                drop(guard);
                break;
            }

            // Memory, last: every ceiling granted this runner. Read under the
            // hold with the set the grant was decided over, so a runner this
            // pass reserved a moment ago holds its estimate here already.
            match self.memory.admit(policy, &attempts) {
                MemoryVerdict::Defer(figures) => {
                    drop(guard);
                    report.memory_deferred = report.memory_deferred.saturating_add(budget);
                    self.events.emit(LifecycleEvent::MemoryShort {
                        policy: policy.id,
                        figures,
                        started: false,
                    });
                    // A directed launch waits like one that could not start:
                    // past `START_LIMIT` it is dropped, and the coordinator
                    // places its job on another host.
                    self.director.start_failed(policy, None);
                    break;
                }
                MemoryVerdict::AdmitAlone(figures) => {
                    self.events.emit(LifecycleEvent::MemoryShort {
                        policy: policy.id,
                        figures,
                        started: true,
                    });
                    self.memory.admitted(policy);
                }
                MemoryVerdict::Admit(_) | MemoryVerdict::Unchecked => self.memory.admitted(policy),
            }

            let created = self
                .launcher
                .launch(LaunchRequest {
                    host: &self.host,
                    policy,
                    allocation_guard: &guard,
                })
                .await;
            let hold = guard.launch_hold();
            drop(guard);

            match created {
                Ok(attempt) => {
                    let id = attempt.id;
                    // At once, not once it is up: the launch has its runner, and
                    // the runner's report names the launch it was started for.
                    self.director.launched(policy, id);
                    budget -= 1;
                    // Carried across policies for the rest of this pass, so the
                    // host-wide total cannot be computed from a set that is
                    // missing it. See `RunnerLauncher::launch`.
                    reserved.push(Reserved {
                        policy: policy.clone(),
                        attempt,
                        waits: ReservationWaits {
                            order: order_wait,
                            lock: lock_wait,
                        },
                        reserved_at: std::time::Instant::now(),
                        _hold: hold,
                    });
                }
                Err(failure) => {
                    self.note_start_failed(policy, None, &failure);
                    break;
                }
            }
        }

        let allocation = decided.unwrap_or(intent);
        self.emit_allocation(&allocation);
        report.allocations.push(allocation);
    }

    /// Oldest first across targets: `policy`'s launch reserves after the older
    /// launches of this host's other targets, whose loops are told of them as
    /// soon as they are accepted. A reservation is a lock hold and a journal
    /// write, so this is short, and only a loop that cannot reserve at all makes
    /// it long: see [`OLDER_LAUNCH_TURN`]. Answers how long it waited.
    async fn let_older_launches_reserve(&self, policy: &ScalePolicy) -> Duration {
        let began = tokio::time::Instant::now();
        let mut older = self.director.older_waiting(policy);
        let mut limit = began + OLDER_LAUNCH_TURN;
        while older > 0 {
            if tokio::time::Instant::now() >= limit {
                tracing::warn!(
                    event = "older_launch_passed_over",
                    policy_id = %policy.id,
                    count = older,
                );
                self.director.pass_over_older(policy);
                break;
            }
            tokio::time::sleep(OLDER_LAUNCH_POLL).await;
            let still = self.director.older_waiting(policy);
            if still < older {
                limit = tokio::time::Instant::now() + OLDER_LAUNCH_TURN;
            }
            older = still;
        }
        began.elapsed()
    }

    /// Bring up what this pass reserved, together and outside the allocation
    /// lock, so one slow package holds up neither another runner nor another
    /// target. How many prepare at once is the launcher's to bound.
    ///
    /// Awaited within the pass, because supervision is what would otherwise see
    /// these attempts half made: one target's attempts are only ever supervised
    /// by its own loop, which is here.
    async fn finish_reserved(&self, reserved: Vec<Reserved>, report: &mut ReconcileReport) {
        let finished =
            futures::future::join_all(reserved.into_iter().map(|reserved| self.bring_up(reserved)))
                .await;
        let started = finished.into_iter().filter(|started| *started).count();
        report.started = report
            .started
            .saturating_add(u16::try_from(started).unwrap_or(u16::MAX));
    }

    /// One reserved runner, up or failed: whether it is up.
    async fn bring_up(&self, reserved: Reserved) -> bool {
        let id = reserved.attempt.id;
        let mut waits = reserved.waits;
        waits.lock += reserved.reserved_at.elapsed();
        let result = self
            .launcher
            .finish_launch(&reserved.policy, reserved.attempt, waits)
            .await;
        // Told here and not after the others are up: the journal already
        // says this one failed, and the attempt reports read it.
        match &result {
            Ok(_) => self.events.emit(LifecycleEvent::RunnerStarted {
                policy: reserved.policy.id,
                attempt: id,
            }),
            Err(failure) => self.note_start_failed(&reserved.policy, Some(id), failure),
        }
        result.is_ok()
    }

    fn note_start_failed(
        &self,
        policy: &ScalePolicy,
        attempt: Option<AttemptId>,
        failure: &LaunchFailure,
    ) {
        self.events.emit(LifecycleEvent::RunnerStartFailed {
            policy: policy.id,
            reason: failure_reason_kind(&failure.reason),
        });
        self.director.start_failed(policy, attempt);
    }

    /// The attempt set the host holds, plus everything this pass has already
    /// created.
    ///
    /// The merge is by [`RunnerAttempt::id`], so a launcher that makes its
    /// launches visible before returning -- which
    /// [`RunnerLauncher::launch`] asks for -- contributes each attempt once, and
    /// one that lags still cannot hide a runner from the host-wide total. The
    /// ceiling therefore holds on the strength of this function rather than on
    /// the strength of an implementer honouring a comment.
    ///
    /// # Errors
    /// Whatever [`RunnerLauncher::attempts`] reported.
    async fn host_attempts(
        &self,
        reserved: &[Reserved],
    ) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
        let mut attempts = self.launcher.attempts().await?;

        // Each `launch` creates one runtime, so each must answer with an
        // identifier no other attempt has. Two entries sharing one here are two
        // runtimes the host-wide total below counts once, which is the ceiling
        // failing silently -- so a development build stops at the first
        // duplicate instead. `RunnerLauncher::launch` states the requirement;
        // this is what makes it findable.
        debug_assert!(
            reserved
                .iter()
                .map(|reserved| reserved.attempt.id)
                .collect::<BTreeSet<AttemptId>>()
                .len()
                == reserved.len(),
            "`RunnerLauncher::launch` returned an AttemptId this pass had already seen; \
             the host ceiling is enforced against a set keyed on that identifier, so a \
             duplicate is two runtimes counted as one"
        );

        let known: BTreeSet<AttemptId> = attempts.iter().map(|attempt| attempt.id).collect();
        attempts.extend(
            reserved
                .iter()
                .filter(|reserved| !known.contains(&reserved.attempt.id))
                .map(|reserved| reserved.attempt.clone()),
        );
        Ok(attempts)
    }

    fn report_allocation_failure(&self, report: &mut ReconcileReport, failure: &AllocationFailure) {
        match failure {
            AllocationFailure::HostCapacity(reason) => {
                report.host_capacity_unreadable = report.host_capacity_unreadable.saturating_add(1);
                self.events
                    .emit(LifecycleEvent::HostCapacityUnreadable { reason: *reason });
            }
            AllocationFailure::Attempts(failure) => {
                self.report_unreadable_attempts(report, failure)
            }
        }
    }

    /// The attempt set could not be read, so nothing may be decided from it.
    ///
    /// Counted rather than swallowed for the reason the module documentation
    /// gives: an unreadable set and an idle host produce the same *number* and
    /// demand opposite actions, so the difference has to survive into the
    /// report.
    fn report_unreadable_attempts(&self, report: &mut ReconcileReport, failure: &LaunchFailure) {
        report.attempts_unreadable = report.attempts_unreadable.saturating_add(1);
        // The variant, never a literal and never the detail. A hand-written
        // `"attempts_unreadable"` said only what the event's own name already
        // said, and threw away the one thing the field is for -- *which* failure
        // it was. `FailureReason::Other` carries free text that must not reach
        // an event, which is what `failure_reason_kind` is for and what
        // `a_cleanup_that_cannot_succeed_...` pins for the sibling path.
        self.events.emit(LifecycleEvent::AttemptsUnreadable {
            reason: failure_reason_kind(&failure.reason),
        });
    }

    /// Remove the runtimes of attempts that have already concluded.
    ///
    /// `is_concluded` and not `is_terminal`: `cleaned` is terminal and already
    /// done, and `busy` is not terminal at all. That is what makes it impossible
    /// for this path to reach a runner executing a job.
    async fn clean_terminal_attempts(&mut self, report: &mut ReconcileReport) {
        let attempts = match self.launcher.attempts().await {
            Ok(attempts) => attempts,
            Err(failure) => {
                self.report_unreadable_attempts(report, &failure);
                return;
            }
        };
        self.clean_backoff
            .retain(&attempts.iter().map(|attempt| attempt.id).collect());
        let now = self.clock.now();
        for attempt in attempts {
            if !attempt.state().is_concluded() {
                continue;
            }
            let Some(outcome) = attempt.outcome() else {
                continue;
            };
            // Backed off after failing: not tried, so neither cleaned nor a
            // fresh failure to report.
            if !self.clean_backoff.is_due(attempt.id, now) {
                continue;
            }
            let kind = OutcomeKind::of(outcome);
            match self.launcher.clean(attempt.id).await {
                Ok(()) => {
                    self.clean_backoff.succeeded(attempt.id);
                    report.cleaned = report.cleaned.saturating_add(1);
                    if kind.is_failure() {
                        report.failures = report.failures.saturating_add(1);
                    } else if kind == OutcomeKind::IdleExit {
                        // The surplus case. Counted apart from a failure because
                        // `g2` renders it apart, and because an operator told
                        // that a normal surplus exit is an error goes hunting a
                        // fault that does not exist.
                        report.idle_exits = report.idle_exits.saturating_add(1);
                    }
                    self.events.emit(LifecycleEvent::AttemptCleaned {
                        policy: attempt.policy_id,
                        attempt: attempt.id,
                        outcome: kind,
                    });
                }
                // A runtime directory that cannot be removed is retried on every
                // poll. Silently, before this arm existed: no event, no counter,
                // no report field, so a cleanup that can never succeed was an
                // invisible permanent loop. It wedges no capacity -- a terminal
                // attempt already stopped counting -- but this module's
                // organising principle is the things that go wrong silently, and
                // `clean` returns a `Result` precisely so the caller can say
                // something. It is retried with per-attempt backoff
                // ([`CLEAN_RETRY_INITIAL`]) so the saying is not a flood.
                Err(failure) => {
                    self.clean_backoff.failed(attempt.id, now);
                    report.clean_failures = report.clean_failures.saturating_add(1);
                    self.events.emit(LifecycleEvent::AttemptCleanFailed {
                        policy: attempt.policy_id,
                        attempt: attempt.id,
                        reason: failure_reason_kind(&failure.reason),
                    });
                }
            }
        }
    }

    /// Reclaim what can be reclaimed for one policy, and nothing else.
    ///
    /// **A busy attempt is never removed.** `04-subsystem-contracts.md`:
    /// *"`busy` cannot transition to cleanup due to a scale-down request"*.
    /// Capacity comes back when an attempt reaches a terminal state and at no
    /// other time, so a scale-down against a host full of busy runners removes
    /// nothing, changes nothing, and says so.
    pub async fn scale_down(&self, policy: &ScalePolicy) -> ScaleDownReport {
        let mut report = ScaleDownReport::default();
        let attempts = match self.launcher.attempts().await {
            Ok(attempts) => attempts,
            Err(failure) => {
                // The same rule as everywhere else, and this was the one place
                // it was still broken: an unreadable set is not an empty one,
                // and a bare `default()` here reported all zeros -- byte for
                // byte an idle host with nothing to reclaim.
                self.events.emit(LifecycleEvent::AttemptsUnreadable {
                    reason: failure_reason_kind(&failure.reason),
                });
                report.attempts_unreadable = true;
                return report;
            }
        };
        for attempt in attempts {
            if attempt.policy_id != policy.id {
                continue;
            }
            match attempt.state() {
                AttemptState::Busy => {
                    report.refused_busy = report.refused_busy.saturating_add(1);
                    self.events.emit(LifecycleEvent::ScaleDownRefused {
                        policy: policy.id,
                        attempt: attempt.id,
                    });
                }
                state if state.is_concluded() => {
                    let kind = attempt
                        .outcome()
                        .map_or(OutcomeKind::Failed, OutcomeKind::of);
                    match self.launcher.clean(attempt.id).await {
                        Ok(()) => {
                            report.removed = report.removed.saturating_add(1);
                            self.events.emit(LifecycleEvent::AttemptCleaned {
                                policy: policy.id,
                                attempt: attempt.id,
                                outcome: kind,
                            });
                        }
                        Err(failure) => {
                            report.clean_failures = report.clean_failures.saturating_add(1);
                            self.events.emit(LifecycleEvent::AttemptCleanFailed {
                                policy: policy.id,
                                attempt: attempt.id,
                                reason: failure_reason_kind(&failure.reason),
                            });
                        }
                    }
                }
                AttemptState::Cleaned => {}
                // `allocated`, `jit_received`, `starting`, `idle`: live, holding
                // a slot, and not this function's to end.
                _ => report.retained = report.retained.saturating_add(1),
            }
        }
        report
    }

    fn emit_allocation(&self, allocation: &Allocation) {
        self.events.emit(LifecycleEvent::Allocated {
            policy: allocation.policy_id,
            demand: allocation.demand,
            desired: allocation.desired,
            active_owned: allocation.active_owned,
            headroom: allocation.headroom_before,
            to_start: allocation.to_start,
            limiting: allocation.limiting_factor,
        });
    }
}

/// One policy's demand, from the reading its target answered with.
///
/// A repository target tallies its own repository's queued jobs; an organization
/// target tallies every repository its scope covered, because one policy watching
/// an organization serves any repository in it.
///
/// # Why this takes a whole policy rather than a target and a label set
///
/// Because both halves have to come from the same policy, and a signature that
/// took them separately made it possible for them not to. The predecessor took a
/// `&ScaleTarget` alone and could not filter at all; the obvious repair was to
/// add a `&RoutingLabels` beside it, and at three call sites — two of them in
/// tests — nothing would have caught passing one policy's target with another
/// policy's labels. It compiles, it runs, and it silently serves the wrong
/// repository's queue.
///
/// # A monitor-only policy has no labels, and cannot reach here
///
/// [`Reconciler::reconcile`] filters on [`ScalePolicy::owns_runners`] before any
/// demand request is issued (D19), so the `None` arm is unreachable rather than
/// merely unlikely. It returns an empty tally instead of unwrapping, because a
/// panic in the reconciliation loop would take the daemon down over a policy
/// that was only ever going to start nothing.
fn demand_for(policy: &ScalePolicy, reading: &QueuedDemand) -> DemandTally {
    let Some(labels) = policy.routing_labels() else {
        debug_assert!(
            false,
            "a monitor-only policy is skipped before the demand poll (D19)"
        );
        return DemandTally::default();
    };

    match &policy.target {
        ScaleTarget::Repository(repository) => labels.tally(reading.jobs_for(repository)),
        ScaleTarget::Organization(_) => labels.tally(reading.jobs()),
    }
}

/// Tally one shared target reading using GitHub's required-label subset rule.
/// Refuse ambiguous ownership, including inactive local profiles: disabling an
/// isolated profile must not redirect its broad-label demand to a native sibling.
fn route_demand(
    all_policies: &[ScalePolicy],
    profiles: &[&ScalePolicy],
    reading: &QueuedDemand,
) -> (
    BTreeMap<PolicyId, DemandTally>,
    BTreeMap<RoutingRefusal, u32>,
) {
    let mut tallies: BTreeMap<PolicyId, DemandTally> = profiles
        .iter()
        .map(|policy| (policy.id, demand_for(policy, reading)))
        .collect();
    let mut refusals: BTreeMap<RoutingRefusal, u32> = BTreeMap::new();
    let Some(first) = profiles.first() else {
        return (tallies, refusals);
    };
    let owners: Vec<_> = all_policies
        .iter()
        .filter(|policy| policy.target == first.target && policy.host_id == first.host_id)
        .filter(|policy| policy.owns_runners())
        .collect();
    let jobs: Vec<&runner_manager_domain::policy::RunsOn> = match &first.target {
        ScaleTarget::Repository(repository) => reading.jobs_for(repository).iter().collect(),
        ScaleTarget::Organization(_) => reading.jobs().collect(),
    };
    for job in jobs {
        let Ok(_) = job.required_labels() else {
            // `demand_for` already preserves this as unresolvable for each profile.
            continue;
        };
        let matched: Vec<PolicyId> = owners
            .iter()
            .filter(|policy| {
                policy
                    .routing_labels()
                    .is_some_and(|labels| labels.matches(job).is_match())
            })
            .map(|policy| policy.id)
            .collect();
        if matched.len() > 1 {
            *refusals
                .entry(RoutingRefusal::OverlappingProfiles)
                .or_default() += 1;
            for id in matched {
                if let Some(tally) = tallies.get_mut(&id) {
                    tally.matched -= 1;
                    tally.not_matched += 1;
                }
            }
        }
    }
    (tallies, refusals)
}

/// How urgently one failure should slow the loop down.
///
/// Ordering matters only for picking the worst of several targets: an outage
/// outranks a rate limit because backing off a socket that is not answering is
/// the safer error, and both outrank a per-target rejection that says nothing
/// about the credential as a whole.
const fn severity(state: &RefreshState) -> u8 {
    match state {
        RefreshState::Offline => 5,
        RefreshState::RateLimited(_) => 4,
        RefreshState::LockedOut { .. } => 3,
        RefreshState::Unauthorized => 2,
        RefreshState::Forbidden { .. } | RefreshState::Failed { .. } => 1,
        RefreshState::Cancelled | RefreshState::Ready(_) => 0,
    }
}

/// Why one target could not be read, as a fixed, credential-free name.
///
/// Deliberately not a [`PollPace`]: a pace describes the *schedule*, which is a
/// property of the whole pass, and stamping one onto a single target would have
/// meant inventing a `consecutive` count for a target that has none. What an
/// event needs here is the reason, and `c3`'s [`RefreshState`] already names it.
///
/// `RefreshState::Failed` carries GitHub's own message and
/// `RefreshState::Forbidden` may carry one too. Neither reaches the event: this
/// returns the variant, for the reason [`failure_reason_kind`] states.
const fn unreadable_reason(state: &RefreshState) -> &'static str {
    match state {
        RefreshState::Ready(_) => "ready",
        RefreshState::Offline => "offline",
        RefreshState::RateLimited(_) => "rate_limited",
        RefreshState::LockedOut { .. } => "locked_out",
        RefreshState::Unauthorized => "unauthorized",
        RefreshState::Forbidden { .. } => "forbidden",
        RefreshState::Failed { .. } => "failed",
        RefreshState::Cancelled => "cancelled",
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The reading that said `healthy` for 28 hours while nothing worked.
    ///
    /// A daemon every one of whose targets answered `401` kept writing a fresh
    /// `last GitHub contact`, because an unauthorized target is `unreadable`
    /// rather than a `failure`. Both that record and the `service status` built
    /// on it were used as evidence during the investigation, and both were
    /// wrong; see `docs/spikes/token-expiry-and-renewal.md`.
    #[test]
    fn a_pass_that_reached_no_target_does_not_claim_it_reached_github() {
        let mut report = ReconcileReport::default();
        assert!(
            !report.reached_github(),
            "a pass that polled nothing -- every policy draining, owned elsewhere, or \
             monitor-only -- reached nobody. This is the case the old guard let through, and \
             the only one it ever let through."
        );

        report.unreadable.push(PolicyId::from_u128(1));
        assert!(
            !report.reached_github(),
            "every target this pass tried was unreadable, so there is no contact to record"
        );

        report.targets_read = 1;
        assert!(
            report.reached_github(),
            "one target answering is contact, whatever else failed alongside it"
        );

        // `allocations` deliberately does not count: a policy this host does
        // not own is allocated for with no demand and without polling anything,
        // so a pass where every poll failed can still carry allocations.
        let mut unowned = ReconcileReport::default();
        unowned.unreadable.push(PolicyId::from_u128(2));
        unowned.allocations.push(Allocation {
            policy_id: PolicyId::from_u128(2),
            demand: 0,
            desired: 0,
            active_owned: 0,
            headroom_before: 0,
            to_start: 0,
            limiting_factor: LimitingFactor::Demand,
        });
        assert!(
            !unowned.reached_github(),
            "an allocation is not evidence that GitHub answered"
        );
    }

    /// The claim the old guard rested on, checked rather than assumed.
    ///
    /// `report.failure.is_none()` was believed to be compatible with an
    /// all-unauthorized pass. It is not: `unreadable` is pushed only from the
    /// `Failed` arm and `failure` is the maximum over every `Failed` reading,
    /// so guarding on `failure.is_none() && reached_github()` would have been
    /// `failure.is_none()` with extra words. This pins the severity that makes
    /// it so, because a future `severity(Unauthorized) == 0` would quietly
    /// restore the belief.
    #[test]
    fn an_unauthorized_target_is_a_failure_and_not_merely_unreadable() {
        assert!(
            severity(&RefreshState::Unauthorized) > 0,
            "an unauthorized reading must survive `max_by_key(severity)` into `report.failure`, \
             or a pass where every target was refused would report no failure at all"
        );
    }

    use std::sync::atomic::AtomicUsize;

    use std::num::NonZeroU16;

    use runner_manager_domain::attempt::PersistedAttempt;
    use runner_manager_domain::model::{CachePolicy, HostId};
    use runner_manager_domain::policy::PolicyMode;
    use runner_manager_domain::workspace::WorkspaceKind;
    use runner_manager_github::rest::RateLimited;
    use runner_manager_testkit::clock::FakeClock;
    use runner_manager_testkit::fixtures;
    use runner_manager_testkit::github::FakeGithub;

    // =======================================================================
    // Fakes
    // =======================================================================

    fn host_with(capacity: u16) -> Host {
        fixtures::host().capacity(capacity).build()
    }

    fn repo(raw: &str) -> OwnerRepo {
        OwnerRepo::parse(raw).expect("a valid OWNER/REPO")
    }

    /// An `active`, enabled autoscale policy on the fixture host.
    use runner_manager_domain::policy::RunsOn;

    fn policy(id: u128, target: &str, max: u16) -> ScalePolicy {
        fixtures::policy()
            .id(PolicyId::from_u128(id))
            .repository(target)
            .autoscale("home", max)
            .active()
            .build()
    }

    /// The host label every policy in these tests carries.
    ///
    /// `policy` above builds through `fixtures::policy().autoscale("home", …)`,
    /// which derives `rm-home-win-x64`. A job fixture that did not carry it
    /// would be filtered out as another host's work, so the two are tied
    /// together here rather than repeated as a literal at each call site.
    const HOST_LABEL: &str = "rm-home-win-x64";

    /// `n` queued jobs this host's policies match.
    ///
    /// The ordinary demand fixture. Since the reversal of the run-counting
    /// decision the unit `e1` clamps is a job, so a test wanting demand `n` asks
    /// for `n` jobs rather than for `n` runs.
    fn jobs(n: usize) -> Vec<RunsOn> {
        fixtures::queued_jobs(&[HOST_LABEL], n)
    }

    /// `e3`, faked: an attempt table and a launch counter, no process anywhere.
    #[derive(Debug, Default)]
    struct FakeLauncher {
        capacity: Mutex<Option<NonZeroU16>>,
        capacity_unreadable: AtomicBool,
        capacity_after_launch: Mutex<Option<NonZeroU16>>,
        attempts: Mutex<Vec<RunnerAttempt>>,
        next_id: AtomicU64,
        launches: AtomicUsize,
        cleaned: Mutex<Vec<AttemptId>>,
        /// Yields this many times between reading the attempt set and recording
        /// a new one, so an unserialised allocator has a window to be wrong in.
        yields_before_recording: usize,
        /// Reports success without the attempt ever becoming visible, which is
        /// the shape a slow journal write has. Every grant then looks like the
        /// first.
        forgetful: bool,
        fail_next: Mutex<Option<FailureReason>>,
        /// Reports that the attempt set cannot be read at all, which is the one
        /// answer a caller must never confuse with an idle host.
        attempts_fail: Mutex<bool>,
        /// Refuses every cleanup, so the silent-retry path has something to be
        /// loud about.
        clean_fails: bool,
        replacements: Mutex<Vec<ReplacementIntent>>,
    }

    impl FakeLauncher {
        fn new() -> Self {
            Self::default()
        }

        fn with_yields(mut self, yields: usize) -> Self {
            self.yields_before_recording = yields;
            self
        }

        fn forgetful() -> Self {
            Self {
                forgetful: true,
                ..Self::default()
            }
        }

        fn seeded(self, attempts: Vec<RunnerAttempt>) -> Self {
            *self.attempts.lock().unwrap() = attempts;
            self
        }

        fn launches(&self) -> usize {
            self.launches.load(Ordering::SeqCst)
        }

        fn snapshot(&self) -> Vec<RunnerAttempt> {
            self.attempts.lock().unwrap().clone()
        }

        fn live_count(&self) -> usize {
            self.snapshot()
                .iter()
                .filter(|a| a.counts_against_capacity())
                .count()
        }

        fn fail_next(&self, reason: FailureReason) {
            *self.fail_next.lock().unwrap() = Some(reason);
        }

        fn fail_attempts(&self, failing: bool) {
            *self.attempts_fail.lock().unwrap() = failing;
        }

        fn refusing_cleanup(attempts: Vec<RunnerAttempt>) -> Self {
            Self {
                clean_fails: true,
                ..Self::default()
            }
            .seeded(attempts)
        }

        fn cleaned(&self) -> Vec<AttemptId> {
            self.cleaned.lock().unwrap().clone()
        }

        fn replacing(self, intent: ReplacementIntent) -> Self {
            self.replacements.lock().unwrap().push(intent);
            self
        }
    }

    #[async_trait::async_trait]
    impl RunnerLauncher for FakeLauncher {
        async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
            if self.capacity_unreadable.load(Ordering::SeqCst) {
                return Err(HostCapacityError::JournalUnreadable);
            }
            Ok(self.capacity.lock().unwrap().unwrap_or(host.host_capacity))
        }

        async fn supervise(
            &self,
            policy: &ScalePolicy,
        ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
            let mut replacements = self.replacements.lock().unwrap();
            let selected: Vec<_> = replacements
                .extract_if(.., |intent| intent.policy == policy.id)
                .collect();
            if !selected.is_empty() {
                let retired: BTreeSet<_> = selected
                    .iter()
                    .map(|intent| intent.previous_attempt)
                    .collect();
                self.attempts
                    .lock()
                    .unwrap()
                    .retain(|attempt| !retired.contains(&attempt.id));
            }
            Ok(selected)
        }

        async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
            if *self.attempts_fail.lock().unwrap() {
                return Err(LaunchFailure::new(FailureReason::Other(
                    "the journal could not be read".into(),
                )));
            }
            Ok(self.snapshot())
        }

        async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
            if let Some(reason) = self.fail_next.lock().unwrap().take() {
                return Err(LaunchFailure::new(reason));
            }
            // The window an unserialised caller would lose the race in.
            for _ in 0..self.yields_before_recording {
                tokio::task::yield_now().await;
            }
            let id =
                AttemptId::from_u128(u128::from(self.next_id.fetch_add(1, Ordering::SeqCst) + 1));
            let created = RunnerAttempt::allocate(
                id,
                request.policy.id,
                "runtime/p/a",
                request.host.created_at,
            );
            self.launches.fetch_add(1, Ordering::SeqCst);
            if let Some(capacity) = self.capacity_after_launch.lock().unwrap().take() {
                *self.capacity.lock().unwrap() = Some(capacity);
            }
            if !self.forgetful {
                self.attempts.lock().unwrap().push(created.clone());
            }
            Ok(created)
        }

        async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
            if self.clean_fails {
                return Err(LaunchFailure::new(FailureReason::Other(
                    "the runtime directory is locked".into(),
                )));
            }
            self.cleaned.lock().unwrap().push(attempt);
            let mut attempts = self.attempts.lock().unwrap();
            attempts.retain(|a| a.id != attempt);
            Ok(())
        }
    }

    /// A demand source a test programs directly, with no gateway underneath.
    #[derive(Debug, Default)]
    struct FakeDemand {
        outcome: Mutex<Option<PollOutcome>>,
        /// Answers programmed for one target, which beat the blanket one.
        per_target: Mutex<BTreeMap<ScaleTarget, PollOutcome>>,
        scopes: Mutex<Vec<ActivityScope>>,
    }

    impl FakeDemand {
        fn ready(count: u32, repository: &OwnerRepo) -> Self {
            let fake = Self::default();
            fake.set(PollOutcome::Ready(QueuedDemand::of(
                repository.clone(),
                jobs(count as usize),
            )));
            fake
        }

        fn failing(state: RefreshState) -> Self {
            let fake = Self::default();
            fake.set(PollOutcome::Failed(state));
            fake
        }

        fn set(&self, outcome: PollOutcome) {
            *self.outcome.lock().unwrap() = Some(outcome);
        }

        /// Program one target's answer, overriding the blanket one.
        fn set_for(&self, target: &ScaleTarget, outcome: PollOutcome) {
            self.per_target
                .lock()
                .unwrap()
                .insert(target.clone(), outcome);
        }

        fn polls(&self) -> Vec<ActivityScope> {
            self.scopes.lock().unwrap().clone()
        }
    }

    #[async_trait::async_trait]
    impl DemandSource for FakeDemand {
        async fn poll(&self, scope: &ActivityScope) -> PollOutcome {
            self.scopes.lock().unwrap().push(scope.clone());
            if let Some(outcome) = self.per_target.lock().unwrap().get(scope.target()) {
                return outcome.clone();
            }
            self.outcome
                .lock()
                .unwrap()
                .clone()
                .unwrap_or(PollOutcome::Ready(QueuedDemand::default()))
        }
    }

    #[derive(Debug, Default)]
    struct FakeDirectory {
        repositories: Vec<OwnerRepo>,
        calls: AtomicUsize,
    }

    impl FakeDirectory {
        fn of(repositories: Vec<OwnerRepo>) -> Self {
            Self {
                repositories,
                calls: AtomicUsize::new(0),
            }
        }

        fn calls(&self) -> usize {
            self.calls.load(Ordering::SeqCst)
        }
    }

    #[async_trait::async_trait]
    impl RepositoryDirectory for FakeDirectory {
        async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(self.repositories.clone())
        }
    }

    /// A lock that grants everything and counts how many holders it had at once.
    ///
    /// The counter is the assertion: "under simulated lock contention" is only
    /// meaningful if something measures that the contention was actually
    /// serialised.
    #[derive(Debug)]
    struct CountingLock {
        inner: InProcessAllocationLock,
        concurrent: Arc<AtomicUsize>,
        peak: Arc<AtomicUsize>,
        acquisitions: Arc<AtomicUsize>,
    }

    impl CountingLock {
        fn new() -> Self {
            Self {
                inner: InProcessAllocationLock::new(),
                concurrent: Arc::new(AtomicUsize::new(0)),
                peak: Arc::new(AtomicUsize::new(0)),
                acquisitions: Arc::new(AtomicUsize::new(0)),
            }
        }

        fn peak(&self) -> usize {
            self.peak.load(Ordering::SeqCst)
        }

        fn acquisitions(&self) -> usize {
            self.acquisitions.load(Ordering::SeqCst)
        }
    }

    #[derive(Debug)]
    struct CountingGuard {
        _inner: AllocationGuard,
        concurrent: Arc<AtomicUsize>,
    }

    impl Drop for CountingGuard {
        fn drop(&mut self) {
            self.concurrent.fetch_sub(1, Ordering::SeqCst);
        }
    }

    #[async_trait::async_trait]
    impl AllocationLock for CountingLock {
        async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
            let inner = self.inner.acquire().await?;
            self.acquisitions.fetch_add(1, Ordering::SeqCst);
            let now = self.concurrent.fetch_add(1, Ordering::SeqCst) + 1;
            self.peak.fetch_max(now, Ordering::SeqCst);
            Ok(AllocationGuard::new(CountingGuard {
                _inner: inner,
                concurrent: Arc::clone(&self.concurrent),
            }))
        }
    }

    /// The lock that is not one: what the host looks like with the serialisation
    /// removed. Used only by the control half of the contention test.
    #[derive(Debug, Default)]
    struct NoLock;

    #[async_trait::async_trait]
    impl AllocationLock for NoLock {
        async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
            Ok(AllocationGuard::new(()))
        }
    }

    #[tokio::test]
    async fn a_wsl_drain_request_fences_the_exact_pre_launch_boundary() {
        use runner_manager_platform::wsl::fence::DrainRequest;

        let (_local, shared, paths) = guest_paths();
        DrainRequest::new(4, chrono::Utc::now())
            .write(shared.path())
            .unwrap();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));

        assert!(lock.acquire().await.is_err());
    }

    #[tokio::test]
    async fn a_guest_launch_claim_is_released_with_its_allocation_guard() {
        use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
        let guard = lock.acquire().await.unwrap();
        assert!(shared.path().join(FENCE_DIRECTORY).exists());
        drop(guard);
        assert!(!shared.path().join(FENCE_DIRECTORY).exists());
    }

    /// Captures what `tracing` writes on this thread.
    #[derive(Clone, Default)]
    struct Logs(Arc<Mutex<Vec<u8>>>);

    impl std::io::Write for Logs {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    /// The release that once failed with no trace while a distribution
    /// started nothing for 13 days is now in the log, with its path.
    #[tokio::test]
    async fn a_launch_fence_that_cannot_be_released_is_logged_with_its_path() {
        use runner_manager_platform::wsl::fence::FENCE_DIRECTORY;

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
        let guard = lock.acquire().await.unwrap();
        // Something the removal cannot take away: a file where the claim
        // directory was.
        let fence = shared.path().join(FENCE_DIRECTORY);
        std::fs::remove_dir_all(&fence).unwrap();
        std::fs::write(&fence, b"not a directory").unwrap();

        let logs = Logs::default();
        let writer = logs.clone();
        let subscriber = tracing_subscriber::fmt()
            .with_writer(move || writer.clone())
            .with_ansi(false)
            .finish();
        tracing::subscriber::with_default(subscriber, || drop(guard));

        let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
        assert!(text.contains("WARN"), "{text}");
        assert!(text.contains("could not be released"), "{text}");
        assert!(text.contains(FENCE_DIRECTORY), "the path is named: {text}");
    }

    fn guest_paths() -> (
        tempfile::TempDir,
        tempfile::TempDir,
        Arc<runner_manager_platform::paths::AppPaths>,
    ) {
        use runner_manager_platform::paths::AppPaths;
        use runner_manager_platform::wsl::fence::GuestRecoveryConfig;

        let local = tempfile::tempdir().unwrap();
        let shared = tempfile::tempdir().unwrap();
        let paths = Arc::new(AppPaths::rooted_at(local.path()));
        paths.create_all().unwrap();
        GuestRecoveryConfig::new(shared.path().to_path_buf())
            .write(&paths)
            .unwrap();
        (local, shared, paths)
    }

    /// The incident end to end: a guest claim left by a daemon of an earlier
    /// WSL boot no longer stops a launch.
    #[tokio::test]
    async fn a_stale_guest_claim_no_longer_blocks_the_next_launch() {
        use runner_manager_platform::wsl::fence::{FENCE_DIRECTORY, OWNER_FILE};

        let (_local, shared, paths) = guest_paths();
        let fence = shared.path().join(FENCE_DIRECTORY);
        std::fs::create_dir_all(&fence).unwrap();
        let pid = std::process::id();
        std::fs::write(
            fence.join(OWNER_FILE),
            format!(
                r#"{{"schema_version":1,"kind":"guest_launch","generation":null,"process_id":{pid},"acquired_at":"2026-09-24T00:47:18.724Z","identity":{{"pid":{pid},"start_token":"linux:earlier-boot:1"}}}}"#
            ),
        )
        .unwrap();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));

        let guard = lock.acquire().await.expect("the stale claim was reclaimed");
        drop(guard);
        assert!(!fence.exists());
    }

    #[tokio::test]
    async fn refusals_that_persist_are_recorded_and_a_grant_clears_them() {
        use runner_manager_platform::launch_health::recorded_launches_blocked;
        use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
            .with_blocked_after(3, Duration::ZERO);
        // Windows owns the fence, so every guest launch is refused.
        let windows =
            FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
                .unwrap()
                .unwrap();

        for _ in 0..2 {
            assert!(lock.acquire().await.is_err());
        }
        assert_eq!(
            recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
            None,
            "not yet"
        );
        assert!(lock.acquire().await.is_err());
        let blocked = recorded_launches_blocked(&paths, chrono::Utc::now())
            .unwrap()
            .expect("three refusals in a row are recorded");
        assert!(
            blocked.reason.contains("Windows WSL recovery"),
            "{blocked:?}"
        );

        windows.release().unwrap();
        let guard = lock.acquire().await.unwrap();
        assert_eq!(
            recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
            None
        );
        drop(guard);
    }

    #[tokio::test]
    async fn refusals_while_this_process_is_launching_are_its_own_contention() {
        use runner_manager_platform::launch_health::recorded_launches_blocked;

        use runner_manager_platform::wsl::fence::DrainRequest;

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
            .with_blocked_after(1, Duration::ZERO);
        let launching = lock.acquire().await.unwrap();
        DrainRequest::new(1, chrono::Utc::now())
            .write(shared.path())
            .unwrap();
        for _ in 0..3 {
            assert!(lock.acquire().await.is_err());
        }
        assert_eq!(
            recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
            None
        );
        drop(launching);
    }

    /// The launches of this process share the guest claim from reservation to
    /// runner: the other target's launch is not refused, and Windows gets the
    /// directory only when the last has ended.
    #[tokio::test]
    async fn launches_in_progress_share_the_fence_until_the_last_ends() {
        use runner_manager_platform::wsl::fence::{DrainRequest, FenceClaim, FenceOwnerKind};

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
        let claimed = |root: &std::path::Path| FenceClaim::owner(root).unwrap().is_some();

        let first = lock.acquire().await.unwrap();
        let first_runner = first.launch_hold();
        drop(first);
        assert!(claimed(shared.path()), "held while its runner comes up");

        let claimed_at = FenceClaim::owner(shared.path())
            .unwrap()
            .unwrap()
            .acquired_at;
        tokio::time::sleep(Duration::from_millis(5)).await;
        let second = lock.acquire().await.expect("this process owns the fence");
        assert!(
            FenceClaim::owner(shared.path())
                .unwrap()
                .unwrap()
                .acquired_at
                > claimed_at,
            "a claim launches keep sharing is not one stuck for that long"
        );
        let second_runner = second.launch_hold();
        drop(second);
        drop(first_runner);
        assert!(claimed(shared.path()), "the other launch is still running");

        // A drain stops new launches and lets the one in progress finish.
        DrainRequest::new(1, chrono::Utc::now())
            .write(shared.path())
            .unwrap();
        assert!(lock.acquire().await.is_err());
        assert!(claimed(shared.path()));
        assert!(
            FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
                .unwrap()
                .is_none()
        );
        drop(second_runner);
        assert!(!claimed(shared.path()), "the last one lets it go");
        assert!(
            FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
                .unwrap()
                .is_some()
        );
    }

    /// Refusals separated by quiet spells -- brief contention days apart --
    /// never add up to a blocked host.
    #[tokio::test]
    async fn refusals_with_quiet_spells_between_them_never_add_up() {
        use runner_manager_platform::launch_health::recorded_launches_blocked;
        use runner_manager_platform::wsl::fence::{FenceClaim, FenceOwnerKind};

        let (_local, shared, paths) = guest_paths();
        let lock = WslRecoveryAllocationLock::new(Arc::clone(&paths), Arc::new(NoLock))
            .with_blocked_after(3, Duration::ZERO)
            .with_streak_gap(Duration::ZERO);
        let windows =
            FenceClaim::try_claim(shared.path(), FenceOwnerKind::WindowsRecovery, Some(1))
                .unwrap()
                .unwrap();
        for _ in 0..5 {
            assert!(lock.acquire().await.is_err());
        }
        assert_eq!(
            recorded_launches_blocked(&paths, chrono::Utc::now()).unwrap(),
            None
        );
        windows.release().unwrap();
    }

    /// A drain request refuses the launch after the claim was made; that
    /// claim is released rather than left to block the next pass.
    #[tokio::test]
    async fn a_claim_refused_by_a_drain_request_is_released() {
        use runner_manager_platform::wsl::fence::{DrainRequest, FENCE_DIRECTORY};

        let (_local, shared, paths) = guest_paths();
        DrainRequest::new(4, chrono::Utc::now())
            .write(shared.path())
            .unwrap();
        let lock = WslRecoveryAllocationLock::new(paths, Arc::new(NoLock));
        assert!(lock.acquire().await.is_err());
        assert!(!shared.path().join(FENCE_DIRECTORY).exists());
    }

    /// A lock nobody can take.
    #[derive(Debug, Default)]
    struct HeldLock;

    #[async_trait::async_trait]
    impl AllocationLock for HeldLock {
        async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
            Err(AllocationLockBusy)
        }
    }

    /// Everything one test needs, wired together.
    struct Harness {
        launcher: Arc<FakeLauncher>,
        demand: Arc<FakeDemand>,
        events: Arc<EventLog>,
        clock: Arc<FakeClock>,
        reconciler: Reconciler,
    }

    impl Harness {
        fn build(
            host: Host,
            launcher: Arc<FakeLauncher>,
            demand: Arc<FakeDemand>,
            lock: Arc<dyn AllocationLock>,
        ) -> Self {
            let events = Arc::new(EventLog::new());
            let clock = Arc::new(FakeClock::default());
            let reconciler = Reconciler::new(
                host,
                ReconcilerPorts {
                    demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
                    launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                    lock,
                    directory: Arc::new(FakeDirectory::default()),
                    clock: Arc::clone(&clock) as Arc<dyn Clock>,
                    jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
                    events: Arc::clone(&events) as Arc<dyn EventSink>,
                },
            );
            Self {
                launcher,
                demand,
                events,
                clock,
                reconciler,
            }
        }

        fn simple(capacity: u16, demand_count: u32, target: &str) -> Self {
            let launcher = Arc::new(FakeLauncher::new());
            let demand = Arc::new(FakeDemand::ready(demand_count, &repo(target)));
            Self::build(
                host_with(capacity),
                launcher,
                demand,
                Arc::new(InProcessAllocationLock::new()),
            )
        }
    }

    fn attempt_in(state: AttemptState, id: u128, policy: u128) -> RunnerAttempt {
        let outcome = state.is_terminal().then(|| match state {
            AttemptState::Failed => {
                AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
            }
            AttemptState::Orphaned => AttemptOutcome::Orphaned,
            _ => AttemptOutcome::CompletedJob,
        });
        RunnerAttempt::from_persisted(PersistedAttempt {
            id: AttemptId::from_u128(id),
            policy_id: PolicyId::from_u128(policy),
            github_runner_id: None,
            state,
            outcome,
            process_id: None,
            execution: runner_manager_domain::execution::AttemptExecution::Native {
                process_id: None,
            },
            runtime_path: "runtime/p/a".into(),
            workspace_kind: WorkspaceKind::Ephemeral,
            workspace_slot: None,
            created_at: fixtures::created_at(),
            terminal_at: state.is_terminal().then(fixtures::created_at),
            last_state_change_at: fixtures::created_at(),
        })
        .expect("a state/outcome pair the domain accepts")
    }

    /// A concluded attempt carrying a specific outcome.
    fn concluded(id: u128, policy: u128, outcome: AttemptOutcome) -> RunnerAttempt {
        RunnerAttempt::from_persisted(PersistedAttempt {
            id: AttemptId::from_u128(id),
            policy_id: PolicyId::from_u128(policy),
            github_runner_id: None,
            state: outcome.terminal_state(),
            outcome: Some(outcome),
            process_id: None,
            execution: runner_manager_domain::execution::AttemptExecution::Native {
                process_id: None,
            },
            runtime_path: "runtime/p/a".into(),
            workspace_kind: WorkspaceKind::Ephemeral,
            workspace_slot: None,
            created_at: fixtures::created_at(),
            terminal_at: Some(fixtures::created_at()),
            last_state_change_at: fixtures::created_at(),
        })
        .expect("a state/outcome pair the domain accepts")
    }

    // =======================================================================
    // The in-flight term: the single most likely way this task goes wrong
    // =======================================================================

    /// `e1`'s Definition of Done, verbatim: *"A job that remains `queued` across
    /// three consecutive polls while its attempt is `starting` yields exactly
    /// one attempt — the test fails if the in-flight term is dropped from the
    /// formula."*
    ///
    /// `b1` tests the arithmetic underneath this
    /// (`capacity::tests::the_same_queued_job_on_two_polls_yields_one_attempt_
    /// not_two`). What *this* test covers is the only way `e1` can drop the
    /// term without touching `b1` at all: handing the allocator an attempt set
    /// that is not the one the host holds.
    ///
    /// # This was measured, not assumed, and the first measurement was worse
    /// # than the failure it was looking for
    ///
    /// Replacing `self.launcher.attempts().await` in
    /// [`Reconciler::start_runners`] with `Vec::new()` compiles and runs. Before
    /// that function carried a budget, this test did not go red — it **never
    /// returned**: every grant looked like the first, so the pass started
    /// runners forever inside poll 1. That is the runaway-runner failure exactly
    /// as an operator would meet it, and it is why the budget exists.
    ///
    /// With the budget in place the same injection fails cleanly and says what
    /// happened: `poll 2 … left: 2, right: 1`. Both measurements were run
    /// before this assertion was written.
    #[tokio::test]
    async fn three_polls_of_one_still_queued_run_yield_exactly_one_attempt() {
        let mut harness = Harness::simple(4, 1, "acme/app");
        let policy = policy(1, "acme/app", 4);

        for poll in 1..=3 {
            let report = harness
                .reconciler
                .reconcile(std::slice::from_ref(&policy))
                .await;
            assert_eq!(
                harness.launcher.launches(),
                1,
                "poll {poll} started another runner for a job already being served; the \
                 `- active_owned_runners` term reached `HostAllocator` as a set this host \
                 does not hold"
            );
            assert_eq!(report.allocations.len(), 1);
            let allocation = &report.allocations[0];
            assert_eq!(allocation.demand, 1, "poll {poll}: still queued at GitHub");
            if poll == 1 {
                assert_eq!(allocation.to_start, 1);
                assert_eq!(report.started, 1);
            } else {
                assert_eq!(allocation.active_owned, 1, "poll {poll}");
                assert_eq!(allocation.to_start, 0, "poll {poll}");
                assert_eq!(report.started, 0, "poll {poll}");
            }
        }
        assert_eq!(harness.launcher.live_count(), 1);
    }

    #[tokio::test]
    async fn draining_parallel_workflow_starts_runners_for_the_remaining_queue() {
        let mut harness = Harness::simple(10, 8, "acme/app");
        let policy = policy(1, "acme/app", 10);
        let first = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;
        assert_eq!(first.started, 8);

        // Five jobs finish while three longer jobs keep their runners busy.
        let attempts = harness.launcher.snapshot();
        *harness.launcher.attempts.lock().unwrap() = attempts
            .iter()
            .take(3)
            .map(|attempt| attempt_in(AttemptState::Busy, attempt.id.as_uuid().as_u128(), 1))
            .collect();
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("acme/app"),
            jobs(2),
        )));

        for poll in 0..3 {
            let report = harness
                .reconciler
                .reconcile(std::slice::from_ref(&policy))
                .await;
            assert_eq!(report.started, if poll == 0 { 2 } else { 0 });
            assert_eq!(harness.launcher.live_count(), 5);
        }
    }

    /// The other half of the measurement above: the loop must terminate even
    /// when the attempt set never catches up with it.
    ///
    /// Dropping the in-flight term made
    /// `three_polls_of_one_still_queued_run_yield_exactly_one_attempt` hang
    /// rather than fail — the loop had one stopping condition and it was the one
    /// the bug removed. A launcher whose journal write has not landed presents
    /// exactly the same shape without any bug at all, so the budget in
    /// [`Reconciler::start_runners`] bounds the pass structurally. This is what
    /// asserts the bound is really there.
    #[tokio::test]
    async fn a_launcher_whose_attempts_never_appear_cannot_wedge_the_pass() {
        let launcher = Arc::new(FakeLauncher::forgetful());
        let mut harness = Harness::build(
            host_with(64),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 8)])
            .await;

        assert_eq!(
            report.started, 3,
            "the pass is bounded by the grant it was given, not by the attempt set catching \
             up with it"
        );
        assert_eq!(launcher.launches(), 3);
        assert!(
            launcher.snapshot().is_empty(),
            "the launcher never recorded anything, which is the whole point of the fixture"
        );
    }

    /// The host-wide ceiling must hold across policies even when the launcher
    /// lags, and the per-policy budget alone does not reach that case.
    ///
    /// Review found this, with this file's own `forgetful` fixture and one more
    /// policy: the budget bounds *each policy's* loop to its own first grant,
    /// but policy B's first grant is computed from a set that does not yet
    /// contain policy A's launches, so B's bound is itself too large. Two
    /// policies on a host of three started **six** runners --
    /// `host_capacity=3, started=6, launches=6` -- with the lock held correctly
    /// throughout. Serialisation was never the problem; the arithmetic under it
    /// was reading a stale set.
    #[tokio::test]
    async fn two_policies_cannot_exceed_host_capacity_even_when_the_launcher_lags() {
        let launcher = Arc::new(FakeLauncher::forgetful());
        let demand = Arc::new(FakeDemand::default());
        demand.set_for(
            &ScaleTarget::repository("acme/left").unwrap(),
            PollOutcome::Ready(QueuedDemand::of(repo("acme/left"), jobs(3))),
        );
        demand.set_for(
            &ScaleTarget::repository("acme/right").unwrap(),
            PollOutcome::Ready(QueuedDemand::of(repo("acme/right"), jobs(3))),
        );
        let mut harness = Harness::build(
            host_with(3),
            Arc::clone(&launcher),
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/left", 3), policy(2, "acme/right", 3)])
            .await;

        assert_eq!(
            report.started, 3,
            "host_capacity is 3 and two policies each allowed 3 started {} runners \
             between them; the second policy's grant was computed from a set that did \
             not yet contain the first policy's launches",
            report.started
        );
        assert_eq!(launcher.launches(), 3);
    }

    /// Finding 1: an attempt set that cannot be read is not an empty one.
    ///
    /// `attempts()` used to be infallible, which left `e3` — reading a journal
    /// off a disk — a choice between panicking and answering `vec![]`. The
    /// second is silent and catastrophic: an empty set is indistinguishable from
    /// an idle host, so a transient read failure reads as "nothing is running"
    /// and the pass allocates the whole machine for jobs already being served.
    ///
    /// The contrast is the assertion. Identical host, identical demand,
    /// identical policy; the only difference is whether the launcher can answer.
    #[tokio::test]
    async fn an_unreadable_attempt_set_starts_nothing_and_is_not_read_as_an_idle_host() {
        let launcher = Arc::new(FakeLauncher::new());
        let mut harness = Harness::build(
            host_with(8),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(4, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );
        let policy = policy(1, "acme/app", 8);

        launcher.fail_attempts(true);
        let unreadable = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(
            unreadable.started, 0,
            "nothing may be decided from a set that was not read"
        );
        assert_eq!(launcher.launches(), 0);
        assert!(unreadable.attempts_unreadable > 0, "and the pass says so");
        assert!(
            unreadable.allocations.is_empty(),
            "no allocation is reported either: there was no set to compute one from, and \
             an allocation of zero would claim a decision nobody made"
        );
        assert!(harness.events.count_of("attempts_unreadable") > 0);

        // The same everything, with a launcher that can answer.
        launcher.fail_attempts(false);
        let readable = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;
        assert_eq!(
            readable.started, 4,
            "the difference between the two passes is only whether the set could be read"
        );
        assert_eq!(readable.attempts_unreadable, 0);
    }

    /// Finding 3: a cleanup that can never succeed was an invisible permanent
    /// loop.
    ///
    /// `if …clean(…).await.is_ok()` had no `else`, so a runtime directory that
    /// could not be removed was retried on every poll with no event, no counter
    /// and no report field. It wedges no capacity — a terminal attempt already
    /// stopped counting — but `clean` returns a `Result` precisely so the caller
    /// can say something, and this module's organising principle is the things
    /// that go wrong silently.
    #[tokio::test]
    async fn a_cleanup_that_cannot_succeed_is_reported_rather_than_retried_in_silence() {
        let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
            1,
            1,
            AttemptOutcome::ExitedIdleWithoutWork,
        )]));
        let mut harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4)])
            .await;

        assert_eq!(report.cleaned, 0);
        assert_eq!(report.clean_failures, 1);
        assert_eq!(harness.events.count_of("attempt_clean_failed"), 1);
        assert_eq!(
            harness.events.count_of("attempt_cleaned"),
            0,
            "and it is not reported as cleaned"
        );
        assert_eq!(
            launcher.snapshot().len(),
            1,
            "the attempt is still there, so the retry is real -- what changed is that it \
             is no longer silent"
        );

        // The reason is the variant, never the detail: the fixture's failure
        // carries free text and none of it reaches the event.
        let reasons: Vec<&'static str> = harness
            .events
            .events()
            .into_iter()
            .filter_map(|event| match event {
                LifecycleEvent::AttemptCleanFailed { reason, .. } => Some(reason),
                _ => None,
            })
            .collect();
        assert_eq!(reasons, vec!["other"]);
    }

    /// A cleanup that keeps failing is retried with per-attempt exponential
    /// backoff rather than on every pass, and a success forgets the backoff.
    #[tokio::test]
    async fn a_failing_cleanup_backs_off_exponentially_per_attempt() {
        let launcher = Arc::new(FakeLauncher::refusing_cleanup(vec![concluded(
            1,
            1,
            AttemptOutcome::ExitedIdleWithoutWork,
        )]));
        let mut harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );
        let policies = [policy(1, "acme/app", 4)];
        let start = harness.clock.now();
        let at = |offset: Duration| start + chrono::Duration::from_std(offset).unwrap();
        let mut failures_at = async |offset: Duration| {
            harness.clock.set(at(offset));
            harness.reconciler.reconcile(&policies).await.clean_failures
        };

        assert_eq!(failures_at(Duration::ZERO).await, 1);
        // Inside the first delay: not tried, so nothing new to report.
        assert_eq!(
            failures_at(CLEAN_RETRY_INITIAL - Duration::from_secs(1)).await,
            0,
            "a failed cleanup was retried on the very next pass"
        );
        assert_eq!(failures_at(CLEAN_RETRY_INITIAL).await, 1);
        // The second delay is twice the first.
        let second = CLEAN_RETRY_INITIAL + CLEAN_RETRY_INITIAL * 2;
        assert_eq!(
            failures_at(second - Duration::from_secs(1)).await,
            0,
            "the delay did not double after a second failure"
        );
        assert_eq!(failures_at(second).await, 1);
        assert_eq!(harness.events.count_of("attempt_clean_failed"), 3);

        // The delay never exceeds the ceiling, however many failures accrue.
        let mut backoff = CleanBackoff::default();
        let id = AttemptId::from_u128(9);
        for _ in 0..40 {
            backoff.failed(id, start);
        }
        assert!(!backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM - Duration::from_secs(1))));
        assert!(backoff.is_due(id, at(CLEAN_RETRY_MAXIMUM)));
        backoff.succeeded(id);
        assert!(
            backoff.is_due(id, start),
            "a cleaned attempt kept its backoff"
        );
    }

    /// N2: an unreadable attempt set makes a scale-down inconclusive, not empty.
    ///
    /// Making `attempts()` fallible closed this everywhere the allocation path
    /// touches, and left it open in the one place that returns a different type:
    /// `scale_down` answered `ScaleDownReport::default()`, which is all zeros
    /// and byte-for-byte identical to an idle host with nothing to reclaim. The
    /// two mean opposite things — "there was nothing to remove" against "we
    /// cannot see what there was".
    ///
    /// Measured as the sibling test measures it: identical host, identical
    /// attempts, identical policy, and the only difference is whether the
    /// launcher can answer.
    #[tokio::test]
    async fn an_unreadable_attempt_set_makes_scale_down_inconclusive_rather_than_empty() {
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![
            attempt_in(AttemptState::Busy, 1, 1),
            concluded(2, 1, AttemptOutcome::CompletedJob),
        ]));
        let harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );
        let policy = policy(1, "acme/app", 4);

        launcher.fail_attempts(true);
        let blind = harness.reconciler.scale_down(&policy).await;

        assert!(!blind.is_conclusive(), "the machine was never read");
        assert_ne!(
            blind,
            ScaleDownReport::default(),
            "a scale-down that could not see the host must not be equal to one that saw \
             an idle host; that equality is the whole finding"
        );
        assert_eq!(blind.removed, 0);
        assert_eq!(
            blind.refused_busy, 0,
            "and this zero means `unknown`, not `none`"
        );
        assert_eq!(harness.events.count_of("attempts_unreadable"), 1);

        // The same everything, with a launcher that can answer.
        launcher.fail_attempts(false);
        let seeing = harness.reconciler.scale_down(&policy).await;

        assert!(seeing.is_conclusive());
        assert_eq!(seeing.removed, 1, "the concluded attempt was reclaimed");
        assert_eq!(seeing.refused_busy, 1, "and the busy one was left alone");
        assert_ne!(
            seeing, blind,
            "the difference between the two is only whether the set could be read"
        );
    }

    /// Finding 7: the lower arm of the clamp, driven through the reconciler.
    ///
    /// `demand_below_min_capacity_starts_nothing_in_v1` runs `demand = 0`
    /// against `min_capacity = 0`, which is *at* the floor and never raises
    /// `desired` — the assertion held for a reason unrelated to the boundary it
    /// named. D7 fixes `min` at 0 for v1, but `AutoscaleConfig::new` accepts
    /// `min > 0` today, so the path is representable and was undriven.
    #[tokio::test]
    async fn demand_below_min_capacity_is_raised_to_min_capacity() {
        let mut warm = ScalePolicy::new(
            PolicyId::from_u128(1),
            ScaleTarget::repository("acme/app").unwrap(),
            1,
            fixtures::HOST_ID,
            PolicyMode::autoscale(
                fixtures::routing_labels("home"),
                2,
                NonZeroU16::new(5).expect("non-zero"),
            )
            .expect("min <= max"),
            CachePolicy::default(),
        );
        warm.activate().expect("pending -> active");

        let mut harness = Harness::simple(8, 0, "acme/app");
        let report = harness.reconciler.reconcile(&[warm]).await;

        assert_eq!(
            report.allocations[0].demand, 0,
            "GitHub reported no queued runs"
        );
        assert_eq!(
            report.allocations[0].desired, 2,
            "min_capacity raised the target above demand"
        );
        assert_eq!(
            report.allocations[0].limiting_factor,
            LimitingFactor::MinCapacity
        );
        assert_eq!(report.started, 2, "and two runners were actually started");
        assert_eq!(harness.launcher.live_count(), 2);
    }

    // =======================================================================
    // Memory admission
    // =======================================================================

    /// The domain rule over a fixed free-memory reading: every attempt on the
    /// host is estimated at `estimate` and none has been sampled yet.
    #[derive(Debug)]
    struct FixedMemory {
        free: u64,
        estimate: u64,
        admitted: std::sync::atomic::AtomicU32,
    }

    impl MemoryGate for FixedMemory {
        fn admit(&self, _policy: &ScalePolicy, attempts: &[RunnerAttempt]) -> MemoryVerdict {
            let running: Vec<runner_manager_domain::memory::Running> = attempts
                .iter()
                .filter(|attempt| attempt.counts_against_capacity())
                .map(|_| runner_manager_domain::memory::Running {
                    estimate: self.estimate,
                    current_rss: None,
                })
                .collect();
            runner_manager_domain::memory::admit(
                Some(self.free),
                &running,
                runner_manager_domain::memory::Estimate::choose(None, Some(self.estimate)),
                GIB,
            )
        }

        fn admitted(&self, _policy: &ScalePolicy) {
            self.admitted
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }

    const GIB: u64 = 1 << 30;

    fn with_memory(harness: Harness, free: u64, estimate: u64) -> (Harness, Arc<FixedMemory>) {
        let gate = Arc::new(FixedMemory {
            free,
            estimate,
            admitted: std::sync::atomic::AtomicU32::new(0),
        });
        let Harness {
            launcher,
            demand,
            events,
            clock,
            reconciler,
        } = harness;
        let reconciler = reconciler.with_memory_gate(Arc::clone(&gate) as Arc<dyn MemoryGate>);
        (
            Harness {
                launcher,
                demand,
                events,
                clock,
                reconciler,
            },
            gate,
        )
    }

    #[tokio::test]
    async fn a_start_the_memory_cannot_hold_is_deferred_and_the_demand_kept() {
        // 10 GiB free, 4 GiB per runner, a 1 GiB margin: the first fits whole,
        // the second fits beside the first's 4 GiB still to come, the third
        // does not. Every ceiling would have granted five.
        let (mut harness, gate) = with_memory(Harness::simple(8, 5, "acme/app"), 10 * GIB, 4 * GIB);
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;

        assert_eq!(
            report.allocations[0].to_start, 5,
            "the ceilings granted five"
        );
        assert_eq!(report.started, 2);
        assert_eq!(report.memory_deferred, 3, "deferred, not failed");
        assert_eq!(report.failures, 0);
        assert_eq!(gate.admitted.load(std::sync::atomic::Ordering::Relaxed), 2);
        let deferred: Vec<MemoryFigures> = harness
            .events
            .events()
            .into_iter()
            .filter_map(|event| match event {
                LifecycleEvent::MemoryShort {
                    figures,
                    started: false,
                    ..
                } => Some(figures),
                _ => None,
            })
            .collect();
        assert_eq!(deferred.len(), 1);
        assert_eq!(deferred[0].free, 10 * GIB);
        assert_eq!(deferred[0].reserved, 8 * GIB);
        assert_eq!(deferred[0].needed(), 5 * GIB);

        // The demand is still queued: the next pass asks again, and is
        // refused again while nothing has ended.
        let again = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;
        assert_eq!(again.started, 0);
        assert!(again.memory_deferred > 0);
        assert_eq!(harness.launcher.live_count(), 2);
    }

    #[tokio::test]
    async fn a_runner_bigger_than_the_free_memory_starts_when_nothing_else_runs() {
        let (mut harness, _) = with_memory(Harness::simple(8, 2, "acme/app"), GIB, 4 * GIB);
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;

        assert_eq!(report.started, 1, "no starvation, but only one");
        assert_eq!(report.memory_deferred, 1);
        let names: Vec<&str> = harness
            .events
            .events()
            .iter()
            .map(LifecycleEvent::name)
            .filter(|name| name.starts_with("memory_"))
            .collect();
        assert_eq!(names, ["memory_admitted_alone", "memory_deferred"]);
    }

    #[tokio::test]
    async fn without_a_memory_gate_every_granted_runner_starts() {
        let mut harness = Harness::simple(8, 5, "acme/app");
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;
        assert_eq!(report.started, 5);
        assert_eq!(report.memory_deferred, 0);
    }

    // =======================================================================
    // Capacity, at the boundaries
    // =======================================================================

    #[tokio::test]
    async fn demand_above_max_capacity_is_clamped_to_max_capacity() {
        let mut harness = Harness::simple(100, 10, "acme/app");
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 3)])
            .await;

        assert_eq!(report.allocations[0].demand, 10);
        assert_eq!(
            report.allocations[0].desired, 3,
            "max_capacity beats demand"
        );
        assert_eq!(report.started, 3);
        assert_eq!(
            report.allocations[0].limiting_factor,
            LimitingFactor::MaxCapacity
        );
    }

    #[tokio::test]
    async fn demand_below_min_capacity_starts_nothing_in_v1() {
        // D7 fixes `min_capacity` at 0, so "below the floor" is "no demand", and
        // the product requirement it satisfies is "no idle runners when unused".
        let mut harness = Harness::simple(8, 0, "acme/app");
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4)])
            .await;

        assert_eq!(report.allocations[0].desired, 0);
        assert_eq!(report.started, 0);
        assert!(report.starts_nothing());
    }

    #[tokio::test]
    async fn lifecycle_replacement_intent_is_consumed_by_the_ordinary_allocator() {
        let policy = policy(1, "octo/repo", 1);
        let previous = attempt_in(AttemptState::Starting, 41, 1);
        let intent = ReplacementIntent {
            policy: policy.id,
            previous_attempt: previous.id,
            operation: "exit_before_acceptance_replacement",
        };
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![previous]).replacing(intent));
        let demand = Arc::new(FakeDemand::ready(1, &repo("octo/repo")));
        let mut harness = Harness::build(
            host_with(1),
            Arc::clone(&launcher),
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness.reconciler.reconcile(&[policy]).await;

        assert_eq!(report.replacement_intents, 1);
        assert_eq!(report.started, 1);
        assert_eq!(launcher.launches(), 1);
        assert_eq!(launcher.live_count(), 1);
    }

    #[tokio::test]
    async fn zero_host_headroom_starts_nothing_at_maximum_demand() {
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![
            attempt_in(AttemptState::Busy, 1, 1),
            attempt_in(AttemptState::Busy, 2, 1),
        ]));
        let demand = Arc::new(FakeDemand::ready(u32::from(u16::MAX), &repo("acme/app")));
        let mut harness = Harness::build(
            host_with(2),
            launcher,
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 2)])
            .await;
        assert_eq!(report.started, 0);
        assert_eq!(report.allocations[0].headroom_before, 0);
        assert_eq!(harness.launcher.launches(), 0);
    }

    #[tokio::test]
    async fn headroom_smaller_than_the_per_policy_allowance_wins() {
        // Four slots held by *another* policy on a host of six: this policy is
        // allowed five and gets two.
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![
            attempt_in(AttemptState::Busy, 1, 99),
            attempt_in(AttemptState::Busy, 2, 99),
            attempt_in(AttemptState::Idle, 3, 99),
            attempt_in(AttemptState::Starting, 4, 99),
        ]));
        let demand = Arc::new(FakeDemand::ready(5, &repo("acme/app")));
        let mut harness = Harness::build(
            host_with(6),
            launcher,
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;
        assert_eq!(
            report.allocations[0].desired, 5,
            "its own ceiling allows five"
        );
        assert_eq!(report.started, 2, "the host has two slots free");
        assert_eq!(
            report.allocations[0].limiting_factor,
            LimitingFactor::HostCapacity
        );
        assert_eq!(harness.launcher.live_count(), 6);
    }

    #[tokio::test]
    async fn the_idle_host_assertion_holds() {
        // "No demand means zero runner processes and zero attempts out of
        // terminal state."
        let mut harness = Harness::simple(8, 0, "acme/app");
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/app", 4)])
            .await;

        assert_eq!(report.started, 0);
        assert_eq!(harness.launcher.launches(), 0);
        assert!(harness.launcher.snapshot().is_empty());
        assert_eq!(
            harness
                .launcher
                .snapshot()
                .iter()
                .filter(|a| !a.is_terminal())
                .count(),
            0
        );
    }

    // =======================================================================
    // Live host capacity, including changes while a pass is creating runners.
    // =======================================================================

    #[tokio::test]
    async fn live_capacity_increase_is_used_without_rebuilding_the_reconciler() {
        let mut harness = Harness::simple(1, 5, "acme/app");
        let policies = [policy(1, "acme/app", 5)];
        assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);

        *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(4);
        assert_eq!(harness.reconciler.reconcile(&policies).await.started, 3);
        assert_eq!(harness.launcher.live_count(), 4);
    }

    #[tokio::test]
    async fn live_capacity_decrease_keeps_busy_runners_and_waits_for_headroom() {
        let mut harness = Harness::simple(5, 5, "acme/app");
        let busy: Vec<_> = (1..=3)
            .map(|id| attempt_in(AttemptState::Busy, id, 1))
            .collect();
        *harness.launcher.attempts.lock().unwrap() = busy.clone();
        *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
        let policies = [policy(1, "acme/app", 5)];

        assert_eq!(harness.reconciler.reconcile(&policies).await.started, 0);
        assert_eq!(harness.launcher.snapshot(), busy);
        assert!(harness.launcher.cleaned().is_empty());

        // Two jobs finish; one busy job still owns a slot.
        harness.launcher.attempts.lock().unwrap().truncate(1);
        assert_eq!(harness.reconciler.reconcile(&policies).await.started, 1);
        assert_eq!(harness.launcher.live_count(), 2);
    }

    #[tokio::test]
    async fn live_capacity_is_rechecked_between_launches_in_the_same_pass() {
        let mut harness = Harness::simple(5, 5, "acme/app");
        *harness.launcher.capacity_after_launch.lock().unwrap() = NonZeroU16::new(1);

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 5)])
            .await;
        assert_eq!(report.started, 1);
        assert_eq!(harness.launcher.live_count(), 1);
    }

    #[tokio::test]
    async fn live_capacity_unreadable_blocks_launches_until_the_journal_recovers() {
        let mut harness = Harness::simple(5, 5, "acme/app");
        harness
            .launcher
            .capacity_unreadable
            .store(true, Ordering::SeqCst);
        let policies = [policy(1, "acme/app", 5)];
        let report = harness.reconciler.reconcile(&policies).await;
        assert_eq!(report.started, 0);
        assert_eq!(report.host_capacity_unreadable, 1);
        assert_eq!(report.attempts_unreadable, 0);
        assert_eq!(harness.events.count_of("attempts_unreadable"), 0);
        assert!(
            harness
                .events
                .events()
                .contains(&LifecycleEvent::HostCapacityUnreadable {
                    reason: HostCapacityError::JournalUnreadable,
                })
        );

        harness
            .launcher
            .capacity_unreadable
            .store(false, Ordering::SeqCst);
        *harness.launcher.capacity.lock().unwrap() = NonZeroU16::new(2);
        assert_eq!(harness.reconciler.reconcile(&policies).await.started, 2);
    }

    #[tokio::test]
    async fn unreadable_local_state_does_not_treat_a_busy_host_as_idle() {
        for host_failure in [true, false] {
            let mut harness = Harness::simple(5, 0, "acme/app");
            harness
                .reconciler
                .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
            *harness.launcher.attempts.lock().unwrap() = vec![attempt_in(AttemptState::Busy, 1, 1)];
            harness
                .launcher
                .capacity_unreadable
                .store(host_failure, Ordering::SeqCst);
            harness.launcher.fail_attempts(!host_failure);
            let report = harness
                .reconciler
                .reconcile(&[policy(1, "acme/app", 5)])
                .await;
            assert_eq!(report.host_capacity_unreadable, u16::from(host_failure));
            assert_eq!(report.next_poll.pace, PollPace::Nominal);
            assert_eq!(harness.launcher.live_count(), 1);
        }
    }

    #[derive(Debug)]
    struct HostEditOnLock {
        inner: InProcessAllocationLock,
        launcher: Arc<FakeLauncher>,
        unreadable: bool,
    }

    #[async_trait::async_trait]
    impl AllocationLock for HostEditOnLock {
        async fn acquire(&self) -> Result<AllocationGuard, AllocationLockBusy> {
            let guard = self.inner.acquire().await?;
            *self.launcher.capacity.lock().unwrap() = NonZeroU16::new(1);
            self.launcher
                .capacity_unreadable
                .store(self.unreadable, Ordering::SeqCst);
            Ok(guard)
        }
    }

    #[tokio::test]
    async fn live_capacity_edits_and_read_failures_while_waiting_for_the_lock_block_launches() {
        for unreadable in [false, true] {
            let launcher =
                Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
            let lock = Arc::new(HostEditOnLock {
                inner: InProcessAllocationLock::new(),
                launcher: Arc::clone(&launcher),
                unreadable,
            });
            let mut harness = Harness::build(
                host_with(5),
                Arc::clone(&launcher),
                Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
                lock,
            );
            let report = harness
                .reconciler
                .reconcile(&[policy(1, "acme/app", 5)])
                .await;
            assert_eq!(report.started, 0);
            assert_eq!(launcher.live_count(), 1);
            assert!(launcher.cleaned().is_empty());
            assert_eq!(report.host_capacity_unreadable, u16::from(unreadable));
            assert_eq!(report.attempts_unreadable, 0);
        }
    }

    // =======================================================================
    // D9 under concurrency: the other silent failure
    // =======================================================================

    /// `e1`'s Definition of Done: *"Two policies on one host with
    /// `host_capacity` smaller than the sum of their `max_capacity` values never
    /// exceed `host_capacity` under concurrent reconciliation — asserted under
    /// simulated lock contention, with no duplicate runners."*
    ///
    /// The contention is simulated by [`FakeLauncher::with_yields`], which puts
    /// executor yield points *between* the launcher reading the attempt set and
    /// recording the new one. Without serialisation both tasks read a headroom
    /// of three and both spend it.
    ///
    /// # Watched failing before it was made to pass
    ///
    /// Granting from this lock without taking the inner mutex — leaving every
    /// counter and every yield point exactly as they are — fails this assertion
    /// with `left: 4, right: 3`: four runners on a host of three, from two
    /// policies each individually inside their own `max_capacity`. The control
    /// test below keeps that measurement standing permanently by running the
    /// same body against [`NoLock`].
    #[tokio::test(flavor = "current_thread")]
    async fn two_policies_reconciling_concurrently_never_exceed_host_capacity() {
        let lock = Arc::new(CountingLock::new());
        let (launches, live) =
            two_policies_concurrently(Arc::clone(&lock) as Arc<dyn AllocationLock>, None).await;

        assert_eq!(
            launches, 3,
            "the sum across policies must never exceed host_capacity, and each policy is \
             individually within its own max_capacity of 3"
        );
        assert_eq!(live, 3, "and no duplicate runner survived the race");
        assert_eq!(
            lock.peak(),
            1,
            "the allocation lock had one holder at a time; without that the read of the \
             headroom and the creation of the runtime are not atomic"
        );
        assert!(
            (3..=5).contains(&lock.acquisitions()),
            "the lock is taken before *each* runtime, not once per pass: three runtimes \
             means at least three holds, and at most one further hold per policy to \
             discover the host filled up underneath it. It was taken {} times",
            lock.acquisitions()
        );
    }

    /// The control for the test above: the same body with the lock removed.
    ///
    /// It exists so that the assertion above cannot pass vacuously. If a future
    /// change makes the unserialised path safe by accident — a launcher that
    /// records synchronously, say — this test goes red and says so, rather than
    /// the other one silently proving nothing.
    #[tokio::test(flavor = "current_thread")]
    async fn without_the_allocation_lock_two_policies_oversubscribe_the_host() {
        let (launches, _) =
            two_policies_concurrently(Arc::new(NoLock) as Arc<dyn AllocationLock>, None).await;

        assert!(
            launches > 3,
            "with no serialisation both policies must be able to spend the same headroom; \
             they started {launches} runners on a host of 3. If this is ever 3, the \
             contention window closed and `two_policies_reconciling_concurrently_never_\
             exceed_host_capacity` has stopped proving anything"
        );
    }

    /// Two policies, one host of three, each allowed three, reconciled at once.
    ///
    /// Returns `(launches, live attempts)`.
    async fn two_policies_concurrently(
        lock: Arc<dyn AllocationLock>,
        live_capacity: Option<NonZeroU16>,
    ) -> (usize, usize) {
        let launcher = Arc::new(FakeLauncher::new().with_yields(4));
        *launcher.capacity.lock().unwrap() = live_capacity;
        let host = host_with(3);

        let mut left = Harness::build(
            host.clone(),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(3, &repo("acme/left"))),
            Arc::clone(&lock),
        )
        .reconciler;
        let mut right = Harness::build(
            host,
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(3, &repo("acme/right"))),
            Arc::clone(&lock),
        )
        .reconciler;

        let a = policy(1, "acme/left", 3);
        let b = policy(2, "acme/right", 3);

        let left = tokio::spawn(async move { left.reconcile(&[a]).await });
        let right = tokio::spawn(async move { right.reconcile(&[b]).await });
        let (_, _) = (left.await.unwrap(), right.await.unwrap());

        (launcher.launches(), launcher.live_count())
    }

    #[tokio::test(flavor = "current_thread")]
    async fn live_capacity_is_shared_by_concurrent_targets_with_stale_startup_hosts() {
        let (launches, live) =
            two_policies_concurrently(Arc::new(InProcessAllocationLock::new()), NonZeroU16::new(2))
                .await;
        assert_eq!(launches, 2);
        assert_eq!(live, 2);
    }

    // =======================================================================
    // D19: monitor-only
    // =======================================================================

    /// `e1`'s Definition of Done: *"A `MonitorOnly` policy under maximum demand
    /// starts zero runners and issues no demand request."*
    ///
    /// Driven through `c4`'s real gateway fake so that "issued no demand
    /// request" is asserted against the thing that would have issued it, rather
    /// than against this module's own bookkeeping. `FakeGithub` records every
    /// call it is asked to make.
    #[tokio::test]
    async fn a_monitor_only_policy_under_maximum_demand_starts_nothing_and_polls_nothing() {
        let gateway = FakeGithub::new().with_queued_jobs(repo("acme/app"), jobs(10_000));
        let gateway = Arc::new(GatewayDemand::new(gateway, CancelToken::new()));
        let launcher = Arc::new(FakeLauncher::new());
        let events = Arc::new(EventLog::new());

        let mut reconciler = Reconciler::new(
            host_with(10),
            ReconcilerPorts {
                demand: Arc::clone(&gateway) as Arc<dyn DemandSource>,
                launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::clone(&events) as Arc<dyn EventSink>,
            },
        );

        let monitor = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .repository("acme/app")
            .monitor_only()
            .active()
            .build();

        let report = reconciler.reconcile(&[monitor]).await;

        assert_eq!(report.started, 0);
        assert_eq!(launcher.launches(), 0);
        assert_eq!(report.monitor_only, vec![PolicyId::from_u128(1)]);
        assert_eq!(
            report.demand_requests, 0,
            "a monitor-only policy spends nothing from the shared hourly ceiling"
        );
        assert!(
            gateway.gateway().calls().is_empty(),
            "a monitor-only policy issued a demand request: {:?}",
            gateway.gateway().calls()
        );
        assert_eq!(events.count_of("monitor_only_skipped"), 1);
        assert_eq!(
            events.count_of("demand_observed"),
            0,
            "and it contributed no demand"
        );
    }

    /// D19 says a monitor-only policy is *"skipped entirely by
    /// reconciliation"*, and "entirely" is the load-bearing word once two
    /// policies share a target.
    ///
    /// This defect was found by review rather than by the test above, which
    /// cannot see it: there, the monitor-only policy is the *only* policy, so
    /// nobody polls its target and the lookup finds nothing. Give it a
    /// repository an autoscale policy already polls and the lookup succeeds —
    /// and the monitor-only policy was then allocated for and had a demand
    /// observation emitted on its behalf. It still started nothing, because
    /// `may_start_runners` is false for it and `HostAllocator` refuses it by
    /// name, so no ceiling was ever at risk. It simply was not skipped.
    ///
    /// Removing the `owns_runners` guard from the allocation loop was watched
    /// failing this test before it was restored:
    /// `a monitor-only policy was allocated for: [… limiting_factor:
    /// MonitorOnly]`.
    #[tokio::test]
    async fn a_monitor_only_policy_sharing_a_target_is_still_skipped_entirely() {
        let lock = Arc::new(CountingLock::new());
        let launcher = Arc::new(FakeLauncher::new());
        let events = Arc::new(EventLog::new());
        let mut reconciler = Reconciler::new(
            host_with(4),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
                launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                lock: Arc::clone(&lock) as Arc<dyn AllocationLock>,
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::clone(&events) as Arc<dyn EventSink>,
            },
        );

        let watcher = fixtures::policy()
            .id(PolicyId::from_u128(2))
            .repository("acme/app")
            .monitor_only()
            .active()
            .build();

        let report = reconciler
            .reconcile(&[policy(1, "acme/app", 4), watcher])
            .await;

        assert_eq!(report.started, 2, "the autoscale policy is served normally");
        assert_eq!(report.monitor_only, vec![PolicyId::from_u128(2)]);
        assert_eq!(
            events.count_of("demand_observed"),
            1,
            "the demand observation belongs to the autoscale policy alone"
        );
        assert!(
            report
                .allocations
                .iter()
                .all(|a| a.policy_id == PolicyId::from_u128(1)),
            "a monitor-only policy was allocated for: {:?}",
            report.allocations
        );
        assert_eq!(
            lock.acquisitions(),
            2,
            "one hold per runtime created, and none on behalf of the monitor-only policy. \
             It was three before the budget was checked at the top of the loop rather than \
             after the re-read, which cost every policy a surplus hold to discover there \
             was nothing left to grant"
        );
    }

    #[tokio::test]
    async fn the_monitor_only_refusal_is_asserted_on_the_mode_not_on_a_missing_ceiling() {
        // The specification requires this to be asserted rather than deduced
        // from `max_capacity` being absent. `HostAllocator` reports it by name,
        // and this loop reaches that arm through `owns_runners`, which is a
        // question about the mode.
        let monitor = fixtures::monitor_only_policy();
        assert!(!monitor.owns_runners());
        assert_eq!(monitor.max_capacity(), None);

        let host = host_with(10);
        let attempts: Vec<RunnerAttempt> = Vec::new();
        let mut allocator = HostAllocator::from_attempts(&host, &attempts);
        let allocation = allocator.allocate(&monitor, 10_000);
        assert_eq!(allocation.limiting_factor, LimitingFactor::MonitorOnly);
        assert_eq!(allocation.to_start, 0);
        assert_eq!(
            allocator.headroom(),
            10,
            "and it consumes no headroom, so an autoscale policy on the same host is \
             unaffected"
        );
    }

    // =======================================================================
    // The surplus runner, and busy protection
    // =======================================================================

    /// `e1`'s Definition of Done: *"A surplus attempt that receives no job
    /// reaches a terminal state recorded as an idle exit, is cleaned, and is not
    /// reported as a failure."*
    #[tokio::test]
    async fn a_surplus_attempt_is_cleaned_as_an_idle_exit_and_not_as_a_failure() {
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![
            concluded(1, 1, AttemptOutcome::ExitedIdleWithoutWork),
            concluded(
                2,
                1,
                AttemptOutcome::failed(FailureReason::JitRequestFailed),
            ),
            concluded(3, 1, AttemptOutcome::CompletedJob),
        ]));
        let mut harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4)])
            .await;

        assert_eq!(report.cleaned, 3);
        assert_eq!(report.idle_exits, 1, "the surplus case, counted apart");
        assert_eq!(
            report.failures, 1,
            "only the failed attempt is a failure; the idle exit and the completed job are \
             not"
        );
        assert_eq!(launcher.cleaned().len(), 3);
        assert!(launcher.snapshot().is_empty());

        let cleaned: Vec<OutcomeKind> = harness
            .events
            .events()
            .into_iter()
            .filter_map(|event| match event {
                LifecycleEvent::AttemptCleaned { outcome, .. } => Some(outcome),
                _ => None,
            })
            .collect();
        assert!(cleaned.contains(&OutcomeKind::IdleExit));
        assert!(
            !OutcomeKind::IdleExit.is_failure(),
            "an idle exit rendered as a failure sends an operator hunting a fault that does \
             not exist"
        );
    }

    /// `e1`'s Definition of Done: *"A scale-down request with a busy attempt
    /// removes nothing and leaves the attempt `busy`."*
    #[tokio::test]
    async fn scale_down_removes_nothing_from_a_busy_attempt() {
        let busy = attempt_in(AttemptState::Busy, 1, 1);
        let launcher = Arc::new(FakeLauncher::new().seeded(vec![
            busy.clone(),
            attempt_in(AttemptState::Starting, 2, 1),
            concluded(3, 1, AttemptOutcome::CompletedJob),
        ]));
        let harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .scale_down(&policy(1, "acme/app", 4))
            .await;

        assert_eq!(report.refused_busy, 1);
        assert_eq!(
            report.retained, 1,
            "the `starting` attempt is not ended either"
        );
        assert_eq!(report.removed, 1, "only the concluded attempt is reclaimed");

        let after = launcher.snapshot();
        let still_busy = after
            .iter()
            .find(|a| a.id == AttemptId::from_u128(1))
            .expect("the busy attempt is still there");
        assert_eq!(
            still_busy.state(),
            AttemptState::Busy,
            "scale-down removed nothing from a runner that is executing a job, and left it \
             busy"
        );
        assert!(!launcher.cleaned().contains(&AttemptId::from_u128(1)));

        // And the domain refuses it from the other side too, by name, so a
        // future caller that tried anyway would not get a generic transition
        // error.
        let mut busy = busy;
        assert!(matches!(
            busy.clean(fixtures::created_at()),
            Err(runner_manager_domain::attempt::AttemptError::BusyCannotBeCleaned)
        ));
        assert_eq!(harness.events.count_of("scale_down_refused"), 1);
    }

    // =======================================================================
    // The schedule
    // =======================================================================

    #[test]
    fn the_default_interval_is_sixty_seconds_and_the_floor_is_thirty() {
        assert_eq!(RefreshInterval::DEFAULT_SECS, 60);
        assert_eq!(RefreshInterval::MIN_SECS, 30);
        assert_eq!(PollSchedule::floor(), Duration::from_secs(30));
        assert!(
            RefreshInterval::from_secs(29).is_err(),
            "the floor is a rate-budget constraint, and a caller must not be able to write \
             a shorter interval at all"
        );

        let mut schedule = PollSchedule::new(RefreshInterval::default());
        let next = schedule.next_poll(None, fixtures::created_at(), &NoJitter);
        assert_eq!(next.delay, Duration::from_secs(60));
        assert_eq!(next.pace, PollPace::Nominal);

        let mut floored = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
        assert_eq!(
            floored
                .next_poll(None, fixtures::created_at(), &NoJitter)
                .delay,
            Duration::from_secs(30)
        );
    }

    /// `e1`'s Definition of Done: *"The poll interval … increases under a
    /// rate-limit signal, and the increase is visible in emitted state rather
    /// than silent."*
    #[test]
    fn a_rate_limit_increases_the_delay_and_names_itself() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());

        let limited = RefreshState::RateLimited(RateLimited {
            kind: RateLimitKind::Secondary,
            retry_after: Some(Duration::from_secs(300)),
            remaining: None,
            reset_unix_secs: None,
        });
        let next = schedule.next_poll(Some(&limited), now, &NoJitter);

        assert_eq!(next.delay, Duration::from_secs(300));
        assert_eq!(
            next.pace,
            PollPace::RateLimited {
                kind: RateLimitKind::Secondary
            },
            "the increase is reported, never hidden"
        );
        assert!(next.pace.is_throttled());
        assert_eq!(next.pace.as_str(), "rate_limited_secondary");
    }

    /// Constraint on this task: *"Read `RefreshState::retry_delay` as an
    /// absolute floor, not an addend."*
    #[test]
    fn the_retry_delay_is_an_absolute_floor_and_never_an_addend() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());

        let limited = RefreshState::RateLimited(RateLimited {
            kind: RateLimitKind::Primary,
            retry_after: Some(Duration::from_secs(300)),
            remaining: Some(0),
            reset_unix_secs: None,
        });

        // Five successive answers, each carrying the window that is *left*.
        // An addend would compound: 360, 660, 960 … and look like a hang.
        for _ in 0..5 {
            let next = schedule.next_poll(Some(&limited), now, &NoJitter);
            assert_eq!(
                next.delay,
                Duration::from_secs(300),
                "the delay is `max(interval, retry_delay)`; `interval + retry_delay` would \
                 have compounded on every successive retry"
            );
        }

        // And when GitHub asks for less than the interval, the interval wins:
        // the floor is never crossed to catch up.
        let brief = RefreshState::RateLimited(RateLimited {
            kind: RateLimitKind::Secondary,
            retry_after: Some(Duration::from_secs(5)),
            remaining: None,
            reset_unix_secs: None,
        });
        let next = schedule.next_poll(Some(&brief), now, &NoJitter);
        assert_eq!(
            next.delay,
            Duration::from_secs(60),
            "a short `retry-after` may not drop the loop below its own interval"
        );
        assert!(next.delay >= PollSchedule::floor());
    }

    #[test]
    fn no_branch_of_the_schedule_can_go_below_the_thirty_second_floor() {
        let now = fixtures::created_at();
        let states = [
            None,
            Some(RefreshState::Offline),
            Some(RefreshState::RateLimited(RateLimited {
                kind: RateLimitKind::Secondary,
                retry_after: Some(Duration::from_secs(1)),
                remaining: None,
                reset_unix_secs: None,
            })),
            Some(RefreshState::LockedOut {
                retry_after: Duration::from_secs(1),
            }),
            Some(RefreshState::Unauthorized),
            Some(RefreshState::Forbidden { message: None }),
            Some(RefreshState::Failed {
                status: Some(500),
                message: "server error".into(),
            }),
            Some(RefreshState::Cancelled),
        ];

        for state in &states {
            let mut schedule = PollSchedule::new(RefreshInterval::from_secs(30).unwrap());
            let next = schedule.next_poll(state.as_ref(), now, &NoJitter);
            assert!(
                next.delay >= PollSchedule::floor(),
                "{state:?} scheduled a poll {}ms away, under the 30-second floor",
                next.delay.as_millis()
            );
        }
    }

    #[test]
    fn an_offline_run_backs_off_with_jitter_and_a_recovery_resets_it() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());

        // Doubling, from the nominal interval.
        let mut previous = Duration::ZERO;
        for consecutive in 1..=6_u32 {
            let next = schedule.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
            assert_eq!(next.pace, PollPace::Offline { consecutive });
            assert!(
                next.delay >= previous,
                "the back-off must not shrink while the outage continues"
            );
            assert!(next.delay >= Duration::from_secs(60));
            previous = next.delay;
        }
        assert!(previous <= MAX_OFFLINE_BACKOFF, "and it is capped");

        // Jitter widens the delay rather than narrowing it, so a fleet of
        // agents does not retry in lockstep.
        let mut jittered = PollSchedule::new(RefreshInterval::default());
        let none = jittered.next_poll(Some(&RefreshState::Offline), now, &NoJitter);
        let mut jittered = PollSchedule::new(RefreshInterval::default());
        let full = jittered.next_poll(Some(&RefreshState::Offline), now, &FixedJitter(0.999));
        assert!(full.delay > none.delay);
        assert!(full.delay <= none.delay.mul_f64(1.0 + JITTER_RATIO));

        // Recovery resets the run with no bookkeeping of its own.
        assert_eq!(schedule.consecutive_offline(), 6);
        let recovered = schedule.next_poll(None, now, &NoJitter);
        assert_eq!(recovered.pace, PollPace::Nominal);
        assert_eq!(recovered.delay, Duration::from_secs(60));
        assert_eq!(schedule.consecutive_offline(), 0);
    }

    #[test]
    fn the_offline_state_states_the_twenty_four_hour_bound() {
        assert_eq!(
            GITHUB_CANCELS_QUEUED_JOBS_AFTER,
            Duration::from_secs(24 * 60 * 60)
        );

        let brief = OfflineState::new(1, Duration::from_secs(120));
        let rendered = brief.to_string();
        assert!(rendered.contains("24 hours"), "{rendered}");
        assert!(rendered.contains("Retrying in 120s"), "{rendered}");
        assert!(!brief.has_outlasted_the_queue());

        let long = brief.since(GITHUB_CANCELS_QUEUED_JOBS_AFTER + Duration::from_secs(1));
        assert!(long.has_outlasted_the_queue());
        assert!(
            long.to_string().contains("queued work has been lost"),
            "{long}"
        );

        // "We cannot tell" is not "not yet".
        assert!(!OfflineState::new(9, Duration::from_secs(60)).has_outlasted_the_queue());
    }

    // =======================================================================
    // Idle polling and the shared quota
    // =======================================================================

    fn quota(remaining: u64, reset_in_secs: i64, now: Timestamp) -> RateLimitSnapshot {
        RateLimitSnapshot {
            limit: 5_000,
            remaining,
            used: Some(5_000 - remaining),
            reset_unix_secs: u64::try_from(now.timestamp() + reset_in_secs).expect("positive"),
            observed_at: now,
        }
    }

    fn quiet() -> PollConditions {
        PollConditions {
            quiet: true,
            quota: None,
            requests_per_pass: 0,
        }
    }

    #[test]
    fn a_quiet_pass_polls_at_the_idle_interval_and_a_busy_one_at_the_active_interval() {
        let now = fixtures::created_at();
        let mut schedule =
            PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));

        let idle = schedule.next_poll_with(None, quiet(), now, &NoJitter);
        assert_eq!(idle.delay, Duration::from_secs(10));
        assert_eq!(idle.pace, PollPace::Idle);
        assert!(!idle.pace.is_throttled(), "idle is faster, not a slowdown");

        let busy = schedule.next_poll_with(None, PollConditions::default(), now, &NoJitter);
        assert_eq!(busy.delay, Duration::from_secs(60));
        assert_eq!(busy.pace, PollPace::Nominal);
    }

    #[test]
    fn a_schedule_never_told_an_idle_interval_keeps_its_active_one() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());
        assert_eq!(
            schedule.next_poll_with(None, quiet(), now, &NoJitter).delay,
            Duration::from_secs(60)
        );
    }

    #[test]
    fn an_idle_interval_is_clamped_between_its_floor_and_the_active_interval() {
        let schedule = PollSchedule::new(RefreshInterval::from_secs(30).expect("the floor"))
            .with_idle(Duration::from_secs(120));
        assert_eq!(
            schedule.idle(),
            Duration::from_secs(30),
            "an idle interval above the active one would make quiet targets slower to react"
        );
        let schedule =
            PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(1));
        assert_eq!(schedule.idle(), PollSchedule::idle_floor());
    }

    /// The owner's fourth requirement: when the shared quota runs low, stretch
    /// the interval automatically, and say so.
    #[test]
    fn a_low_quota_stretches_polling_and_wins_over_idle() {
        let now = fixtures::created_at();
        let mut schedule =
            PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));

        let healthy = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: true,
                quota: Some(quota(1_000, 3_000, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(healthy.pace, PollPace::Idle, "a fifth left is still normal");

        let low = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: true,
                quota: Some(quota(999, 3_000, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(low.delay, Duration::from_secs(120));
        assert_eq!(
            low.pace,
            PollPace::Stretched {
                pressure: PrimaryPressure::Low
            }
        );
        assert!(low.pace.is_throttled());
        assert_eq!(low.pace.as_str(), "stretched_low_budget");

        let critical = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: false,
                quota: Some(quota(100, 3_000, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(critical.delay, Duration::from_secs(240));
        assert_eq!(critical.pace.as_str(), "stretched_critical_budget");
    }

    #[test]
    fn a_stretch_never_outlasts_the_quotas_reset_nor_undercuts_the_active_interval() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());

        let soon = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: false,
                quota: Some(quota(10, 90, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(
            soon.delay,
            Duration::from_secs(90),
            "the bucket refills in 90 s; waiting 240 s buys nothing"
        );

        let imminent = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: false,
                quota: Some(quota(10, 5, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(imminent.delay, Duration::from_secs(60));

        let expired = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: true,
                quota: Some(quota(10, -5, now)),
                requests_per_pass: 0,
            },
            now,
            &NoJitter,
        );
        assert_eq!(
            expired.pace,
            PollPace::Nominal,
            "a reading from a window that has reset says nothing; the schedule here had no \
             idle interval, so it is the active one"
        );
    }

    #[test]
    fn an_idle_target_too_large_to_poll_that_often_stretches_its_idle_interval() {
        let now = fixtures::created_at();
        let mut schedule =
            PollSchedule::new(RefreshInterval::default()).with_idle(Duration::from_secs(10));
        let small = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: true,
                quota: None,
                requests_per_pass: 40,
            },
            now,
            &NoJitter,
        );
        assert_eq!(small.delay, Duration::from_secs(10), "ten repositories fit");
        let organization = schedule.next_poll_with(
            None,
            PollConditions {
                quiet: true,
                quota: None,
                requests_per_pass: 400,
            },
            now,
            &NoJitter,
        );
        assert_eq!(
            organization.delay,
            Duration::from_secs(80),
            "400 requests a pass at 300 a minute is one pass every 80 s"
        );
        assert_eq!(organization.pace, PollPace::Idle);
    }

    #[test]
    fn a_rate_limit_wait_is_jittered_upward_only() {
        let now = fixtures::created_at();
        let mut schedule = PollSchedule::new(RefreshInterval::default());
        let limited = RefreshState::RateLimited(RateLimited {
            kind: RateLimitKind::Secondary,
            retry_after: Some(Duration::from_secs(300)),
            remaining: None,
            reset_unix_secs: None,
        });
        assert_eq!(
            schedule
                .next_poll(Some(&limited), now, &FixedJitter(0.5))
                .delay,
            Duration::from_secs(315),
            "every loop shares one quiet period; without jitter they all wake together"
        );
        assert_eq!(
            schedule.next_poll(Some(&limited), now, &NoJitter).delay,
            Duration::from_secs(300),
            "never shorter than GitHub asked"
        );
    }

    #[tokio::test]
    async fn demand_this_host_did_not_serve_keeps_the_target_off_the_idle_pace() {
        // Capacity zero would refuse the policy outright; a host already full
        // is the ordinary way matched demand goes unserved for a pass.
        let launcher =
            Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 9, 9)]));
        let demand = Arc::new(FakeDemand::ready(2, &repo("acme/app")));
        let mut harness = Harness::build(
            host_with(1),
            launcher,
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(report.started, 0, "the host is full");
        assert_eq!(
            report.next_poll.pace,
            PollPace::Nominal,
            "matched demand it could not serve is work waiting, not idle"
        );
    }

    #[tokio::test]
    async fn a_target_with_nothing_to_do_is_polled_at_the_idle_interval() {
        let mut harness = Harness::simple(4, 0, "acme/app");
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(report.next_poll.pace, PollPace::Idle);
        assert_eq!(report.next_poll.delay, Duration::from_secs(10));
    }

    #[tokio::test]
    async fn a_target_whose_runs_are_changing_is_polled_at_the_active_interval() {
        let mut harness = Harness::simple(4, 0, "acme/app");
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        harness.demand.set(PollOutcome::Ready(
            QueuedDemand::of(repo("acme/app"), Vec::new()).with_busy(repo("acme/app")),
        ));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(report.next_poll.pace, PollPace::Nominal);
        assert_eq!(report.next_poll.delay, Duration::from_secs(60));
    }

    #[tokio::test]
    async fn a_target_with_a_live_runner_is_polled_at_the_active_interval() {
        let launcher =
            Arc::new(FakeLauncher::new().seeded(vec![attempt_in(AttemptState::Busy, 1, 1)]));
        let demand = Arc::new(FakeDemand::ready(0, &repo("acme/app")));
        let mut harness = Harness::build(
            host_with(4),
            launcher,
            demand,
            Arc::new(InProcessAllocationLock::new()),
        );
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(
            report.next_poll.pace,
            PollPace::Nominal,
            "a job is running: its follow-on work is what the active interval watches for"
        );
    }

    #[tokio::test]
    async fn a_pass_that_starts_a_runner_is_not_idle() {
        let mut harness = Harness::simple(4, 2, "acme/app");
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert!(report.started > 0);
        assert_eq!(report.next_poll.pace, PollPace::Nominal);
    }

    #[tokio::test]
    async fn the_quota_reported_with_a_reading_stretches_the_next_poll() {
        let mut harness = Harness::simple(4, 0, "acme/app");
        harness
            .reconciler
            .set_intervals(RefreshInterval::default(), Duration::from_secs(10));
        let now = harness.clock.now();
        harness.demand.set(PollOutcome::Ready(
            QueuedDemand::of(repo("acme/app"), Vec::new())
                .with_rate_limit(Some(quota(200, 1_800, now))),
        ));
        let policy = policy(1, "acme/app", 4);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert_eq!(
            report.next_poll.pace,
            PollPace::Stretched {
                pressure: PrimaryPressure::Critical
            }
        );
        assert_eq!(report.next_poll.delay, Duration::from_secs(240));
    }

    // =======================================================================
    // Offline, end to end
    // =======================================================================

    /// `e1`'s Definition of Done: *"An unreachable GitHub yields `offline`, zero
    /// new runners, retained existing processes, and jittered backoff; recovery
    /// resumes polling and does not double-count a job that was already being
    /// served."*
    #[tokio::test]
    async fn an_unreachable_github_starts_nothing_retains_everything_and_backs_off() {
        let live = vec![
            attempt_in(AttemptState::Busy, 1, 1),
            attempt_in(AttemptState::Starting, 2, 1),
        ];
        let launcher = Arc::new(FakeLauncher::new().seeded(live.clone()));
        let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
        let mut harness = Harness::build(
            host_with(8),
            Arc::clone(&launcher),
            Arc::clone(&demand),
            Arc::new(InProcessAllocationLock::new()),
        );
        let policy = policy(1, "acme/app", 8);

        let report = harness
            .reconciler
            .reconcile(std::slice::from_ref(&policy))
            .await;

        assert!(report.is_offline());
        assert_eq!(report.started, 0, "no new runner during an outage");
        assert_eq!(launcher.launches(), 0);
        assert_eq!(
            launcher.snapshot(),
            live,
            "existing runner processes are retained, untouched"
        );
        assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
        assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
        assert!(report.next_poll.delay >= Duration::from_secs(60));
        let offline = report.offline_state().expect("an offline state to display");
        assert!(offline.to_string().contains("24 hours"));

        // Recovery: two jobs are queued, but only the starting runner can
        // cover them. The busy runner consumes capacity and serves other work.
        demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("acme/app"),
            jobs(2),
        )));
        let recovered = harness.reconciler.reconcile(&[policy]).await;

        assert!(!recovered.is_offline());
        assert_eq!(recovered.next_poll.pace, PollPace::Nominal);
        assert_eq!(
            recovered.started, 1,
            "two queued jobs and one starting runner leave one job uncovered"
        );
        assert_eq!(recovered.allocations[0].active_owned, 2);
        assert_eq!(launcher.live_count(), 3);
    }

    /// One unreachable target must not idle a whole host.
    ///
    /// The failure that decides the *schedule* is the most severe across every
    /// target polled — backing the whole loop off during an outage is the safe
    /// error, and `f3` runs one reconciler per target anyway, so in production
    /// the two are usually the same thing. What must not follow from that is
    /// refusing to serve a policy whose own target answered perfectly well, and
    /// the two are easy to conflate because the offline reading is sitting in
    /// the same map.
    #[tokio::test]
    async fn one_offline_target_does_not_stop_a_reachable_one() {
        let mut harness = Harness::simple(8, 0, "acme/app");
        harness.demand.set_for(
            &ScaleTarget::repository("acme/app").unwrap(),
            PollOutcome::Ready(QueuedDemand::of(repo("acme/app"), jobs(2))),
        );
        harness.demand.set_for(
            &ScaleTarget::repository("acme/broken").unwrap(),
            PollOutcome::Failed(RefreshState::Offline),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4), policy(2, "acme/broken", 4)])
            .await;

        assert_eq!(
            report.started, 2,
            "the reachable target was served; an unreachable sibling repository must not \
             idle the host"
        );
        assert_eq!(report.unreadable, vec![PolicyId::from_u128(2)]);
        assert_eq!(harness.demand.polls().len(), 2, "both targets were polled");

        // And the schedule takes the worse of the two.
        assert!(report.is_offline());
        assert_eq!(report.next_poll.pace, PollPace::Offline { consecutive: 1 });
    }

    /// The 24-hour bound has to be reachable in production, not only in a unit
    /// test of [`OfflineState`].
    ///
    /// This was a real gap: the reconciler built its offline state from the
    /// back-off count alone, so `offline_for` was always `None` and
    /// [`OfflineState::has_outlasted_the_queue`] could never be true outside a
    /// test that constructed the value by hand. An operator whose agent had been
    /// offline for two days would have been told that an outage longer than 24
    /// hours *would* lose queued work, in the future tense, having already lost
    /// it.
    ///
    /// The elapsed time is measured from the first poll of the run rather than
    /// derived from the interval, because the back-off doubles and the two
    /// diverge immediately.
    #[tokio::test]
    async fn a_day_long_outage_says_that_queued_work_has_already_been_lost() {
        let clock = Arc::new(FakeClock::default());
        let launcher = Arc::new(FakeLauncher::new());
        let demand = Arc::new(FakeDemand::failing(RefreshState::Offline));
        let mut reconciler = Reconciler::new(
            host_with(4),
            ReconcilerPorts {
                demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
                launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::clone(&clock) as Arc<dyn Clock>,
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        );
        let policy = policy(1, "acme/app", 4);

        // The outage begins.
        let first = reconciler.reconcile(std::slice::from_ref(&policy)).await;
        let state = first.offline_state().expect("an offline state");
        assert!(!state.has_outlasted_the_queue());
        assert!(
            state
                .to_string()
                .contains("an outage longer than that loses"),
            "{state}"
        );

        // A day and a minute later, still unreachable.
        clock.advance_secs(24 * 60 * 60 + 60);
        let later = reconciler.reconcile(std::slice::from_ref(&policy)).await;
        let state = later.offline_state().expect("an offline state");
        assert!(state.has_outlasted_the_queue());
        assert!(
            state.to_string().contains("queued work has been lost"),
            "{state}"
        );
        assert_eq!(launcher.launches(), 0, "and still nothing was started");

        // Recovery closes the run, so a *later* outage measures from itself
        // rather than from the first one.
        demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("acme/app"),
            jobs(0),
        )));
        let recovered = reconciler.reconcile(std::slice::from_ref(&policy)).await;
        assert!(recovered.offline_state().is_none());
        assert_eq!(reconciler.schedule().offline_for(clock.now()), None);

        demand.set(PollOutcome::Failed(RefreshState::Offline));
        let again = reconciler.reconcile(std::slice::from_ref(&policy)).await;
        assert!(
            !again
                .offline_state()
                .expect("an offline state")
                .has_outlasted_the_queue(),
            "a new outage must not inherit the age of the one before it"
        );
    }

    /// Finding 5: the adapter, not the lock underneath it.
    ///
    /// `d1` covers `LockKind::Allocation` including a contended `acquire_at`
    /// with a wait. What that does not reach is this adapter: the
    /// `spawn_blocking` wrapper, the collapse of both a refused lock and a
    /// panicked blocking task into `AllocationLockBusy`, and — the one that
    /// would be silent — whether [`AllocationGuard`] really holds the
    /// `HostLock`, since dropping it is the only release there is. A guard that
    /// dropped the lock on the way out would make every acquisition succeed and
    /// the ceiling would hold by luck.
    ///
    /// The original disclosure said this needed a real filesystem and was
    /// therefore expensive. `AppPaths::rooted_at` plus `tempfile` — already a
    /// non-dev dependency of this crate — makes it about fifteen lines, so the
    /// reason was weaker than stated.
    #[tokio::test]
    async fn the_file_allocation_lock_excludes_a_second_holder_and_releases_on_drop() {
        let root = tempfile::tempdir().expect("a temporary directory");
        let paths = Arc::new(runner_manager_platform::paths::AppPaths::rooted_at(
            root.path(),
        ));
        let lock = FileAllocationLock::new(paths).with_wait(Duration::from_millis(50));

        let held = lock.acquire().await.expect("an uncontended lock is free");
        assert!(
            matches!(lock.acquire().await, Err(AllocationLockBusy)),
            "a second holder was admitted; on Unix the lock is per open file description \
             and on Windows the share mode denies write, so this must be refused even \
             from inside the same process"
        );

        drop(held);
        let regained = lock.acquire().await;
        assert!(
            regained.is_ok(),
            "dropping the guard is the only release there is, so a guard that does not \
             hold the `HostLock` leaves it held forever"
        );
    }

    #[test]
    fn tee_events_reaches_both_sinks() {
        // `f3` wires the log sink and `g2`'s buffer at once, and an event that
        // reached only one of them would be an activity view missing lines the
        // log file has, or the reverse.
        let left = Arc::new(EventLog::new());
        let right = Arc::new(EventLog::new());
        let tee = TeeEvents(
            Arc::clone(&left) as Arc<dyn EventSink>,
            Arc::clone(&right) as Arc<dyn EventSink>,
        );

        tee.emit(LifecycleEvent::MonitorOnlySkipped {
            policy: PolicyId::from_u128(1),
        });

        assert_eq!(left.count_of("monitor_only_skipped"), 1);
        assert_eq!(right.count_of("monitor_only_skipped"), 1);
    }

    // =======================================================================
    // Budget: the repository list, and the per-target poll
    // =======================================================================

    #[tokio::test]
    async fn the_repository_list_refreshes_far_more_slowly_than_the_demand_poll() {
        let clock = Arc::new(FakeClock::default());
        let directory = Arc::new(FakeDirectory::of(vec![repo("acme/one"), repo("acme/two")]));
        let cache = RepositoryCache::new(
            Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
            Arc::clone(&clock) as Arc<dyn Clock>,
            RefreshInterval::default(),
        );
        let target = ScaleTarget::organization("acme").unwrap();

        assert_eq!(
            cache.ttl(),
            Duration::from_secs(60 * u64::from(REPOSITORY_LIST_REFRESH_MULTIPLE))
        );

        // Every poll inside the window reuses the list.
        for _ in 0..REPOSITORY_LIST_REFRESH_MULTIPLE {
            let scope = cache.scope_for(&target).await.unwrap();
            assert_eq!(scope.repositories().len(), 2);
            clock.advance_secs(60);
        }
        assert_eq!(
            directory.calls(),
            1,
            "re-listing an organization at demand-poll frequency is what exhausts the \
             shared request budget"
        );
        assert_eq!(cache.lookups(), 1);

        // Past it, exactly one more.
        cache.scope_for(&target).await.unwrap();
        assert_eq!(directory.calls(), 2);
    }

    #[tokio::test]
    async fn a_repository_target_never_consults_the_directory() {
        let directory = Arc::new(FakeDirectory::of(vec![repo("acme/other")]));
        let cache = RepositoryCache::new(
            Arc::clone(&directory) as Arc<dyn RepositoryDirectory>,
            Arc::new(FakeClock::default()) as Arc<dyn Clock>,
            RefreshInterval::default(),
        );
        let target = ScaleTarget::repository("acme/app").unwrap();

        let scope = cache.scope_for(&target).await.unwrap();
        assert_eq!(scope.repositories(), &[repo("acme/app")]);
        assert_eq!(directory.calls(), 0);
    }

    #[tokio::test]
    async fn two_policies_on_one_target_cost_one_demand_poll_not_two() {
        // `04-subsystem-contracts.md` prices a *target*. A loop that spent per
        // policy would exceed the projection `f2` admitted the configuration
        // against, silently.
        let mut harness = Harness::simple(8, 4, "o/r");
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [
                fixtures::queued_job(&[HOST_LABEL]),
                fixtures::queued_job(&[HOST_LABEL]),
                fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
                fixtures::queued_job(&["rm-home-win-x64-py-isolated"]),
            ],
        )));
        let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        isolated.activate().unwrap();
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "o/r", 2), isolated])
            .await;

        assert_eq!(harness.demand.polls().len(), 1);
        assert_eq!(report.targets_read, 1, "profiles share one target reading");
        assert_eq!(
            report.demand_requests,
            runner_manager_github::demand::DEMAND_REQUESTS_PER_REPOSITORY_PER_POLL,
            "one repository's worth of demand requests, not two policies' worth. Read              from the constant rather than written as a literal so that repricing the              poll cannot silently turn this into an assertion about the wrong thing"
        );
        assert_eq!(
            report.started, 4,
            "native and isolated profiles each get two"
        );
    }

    #[tokio::test]
    async fn subset_labels_without_selector_scale_only_the_compatible_local_profile() {
        use runner_manager_domain::model::{Arch, HostLabel, Os};
        use runner_manager_domain::policy::RoutingLabels;
        let mut harness = Harness::build(
            fixtures::host()
                .os(Os::MacOs)
                .architecture(Arch::Arm64)
                .capacity(10)
                .build(),
            Arc::new(FakeLauncher::new()),
            Arc::new(FakeDemand::ready(0, &repo("o/r"))),
            Arc::new(InProcessAllocationLock::new()),
        );
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [
                fixtures::queued_job(&["self-hosted", "macOS", "ARM64"]),
                fixtures::queued_job(&["SELF-HOSTED", "MACOS", "arm64"]),
                fixtures::queued_job(&["self-hosted", "macOS", "X64"]),
                fixtures::queued_job(&["self-hosted", "Windows", "ARM64"]),
                fixtures::queued_job(&["self-hosted", "macOS", "ARM64", "gpu"]),
                fixtures::queued_job(&["rm-other-osx-arm64", "macOS", "ARM64"]),
            ],
        )));
        let mut local = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .repository("o/r")
            .autoscale("macmini", 10)
            .routing_labels(RoutingLabels::derive(
                &HostLabel::new("macmini").unwrap(),
                Os::MacOs,
                Arch::Arm64,
            ))
            .active()
            .build();
        for label in ["self-hosted", "macos", "arm64", "extra"] {
            local.add_routing_label(fixtures::label(label)).unwrap();
        }
        let mut foreign = local.to_persisted();
        foreign.id = PolicyId::from_u128(2);
        foreign.host_id = HostId::from_u128(0xdead);
        let foreign = ScalePolicy::from_persisted(foreign).unwrap();
        let report = harness.reconciler.reconcile(&[local, foreign]).await;
        assert_eq!(report.started, 2);
        assert_eq!(report.targets_read, 1);
        assert!(harness.events.events().iter().any(|event| matches!(
            event,
            LifecycleEvent::DemandObserved {
                demand: 2,
                not_matched: 4,
                ..
            }
        )));
        assert!(
            !harness
                .events
                .events()
                .iter()
                .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
        );
    }

    #[tokio::test]
    async fn isolated_overlap_never_falls_back_to_native_in_either_state_or_order() {
        use runner_manager_domain::execution::{
            Backend, ExecutionPolicy, ImageReference, ResourceLimits,
        };
        let mut native = policy(1, "o/r", 2);
        native
            .add_routing_label(fixtures::label("self-hosted"))
            .unwrap();
        let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        isolated
            .add_routing_label(fixtures::label("self-hosted"))
            .unwrap();
        isolated
            .set_execution_policy(ExecutionPolicy::Isolated {
                backend: Backend::Oci,
                image: ImageReference::new(format!(
                    "registry.example/runner@sha256:{}",
                    "a".repeat(64)
                ))
                .unwrap(),
                resources: ResourceLimits {
                    cpu_millis: 1000,
                    memory_mib: 512,
                    disk_mib: 1024,
                },
            })
            .unwrap();
        assert!(!isolated.may_start_runners());
        let mut active_isolated = isolated.clone();
        active_isolated.activate().unwrap();
        for profiles in [
            [native.clone(), isolated.clone()],
            [isolated.clone(), native.clone()],
            [native.clone(), active_isolated.clone()],
            [active_isolated, native.clone()],
        ] {
            let mut harness = Harness::simple(4, 0, "o/r");
            harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
                repo("o/r"),
                [fixtures::queued_job(&["SELF-HOSTED"])],
            )));
            let report = harness.reconciler.reconcile(&profiles).await;
            assert_eq!(report.started, 0);
            assert!(harness.events.events().iter().any(|event| matches!(
                event,
                LifecycleEvent::RoutingRefused {
                    reason: RoutingRefusal::OverlappingProfiles,
                    count: 1,
                    ..
                }
            )));
        }
        // A job explicitly requiring the unavailable isolated selector cannot
        // be served by the native sibling either.
        let mut harness = Harness::simple(4, 0, "o/r");
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [fixtures::queued_job(&["rm-home-win-x64-py-isolated"])],
        )));
        assert_eq!(
            harness
                .reconciler
                .reconcile(&[native, isolated])
                .await
                .started,
            0
        );
    }

    #[tokio::test]
    async fn shared_optional_labels_do_not_duplicate_native_and_isolated_demand() {
        let mut harness = Harness::simple(2, 0, "o/r");
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [
                fixtures::queued_job(&[HOST_LABEL, "gpu"]),
                fixtures::queued_job(&["rm-home-win-x64-py-isolated", "gpu"]),
                fixtures::queued_job(&["gpu"]),
                fixtures::unresolvable_job(),
            ],
        )));
        let mut native = policy(1, "o/r", 2);
        native.add_routing_label(fixtures::label("gpu")).unwrap();
        let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        isolated.add_routing_label(fixtures::label("gpu")).unwrap();
        isolated.activate().unwrap();

        let report = harness.reconciler.reconcile(&[native, isolated]).await;
        assert_eq!(harness.demand.polls().len(), 1);
        assert_eq!(report.targets_read, 1);
        assert_eq!(report.started, 2);
        assert_eq!(report.allocations.len(), 2);
        assert!(
            report
                .allocations
                .iter()
                .all(|allocation| allocation.to_start == 1)
        );
        assert!(harness.events.events().iter().any(|event| matches!(
            event,
            LifecycleEvent::RoutingRefused {
                reason: RoutingRefusal::OverlappingProfiles,
                count: 1,
                ..
            }
        )));
        assert_eq!(
            harness
                .events
                .events()
                .iter()
                .filter(|event| matches!(
                    event,
                    LifecycleEvent::DemandObserved {
                        unresolvable: 1,
                        ..
                    }
                ))
                .count(),
            2,
            "an expression stays unresolvable for both profiles"
        );
    }

    #[tokio::test]
    async fn missing_required_selector_labels_do_not_match_and_overlaps_show_remedies() {
        let mut harness = Harness::simple(4, 0, "o/r");
        harness.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [fixtures::queued_job(&[
                HOST_LABEL,
                "rm-home-win-x64-py-isolated",
            ])],
        )));
        let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        isolated.activate().unwrap();
        let report = harness
            .reconciler
            .reconcile(&[policy(1, "o/r", 2), isolated])
            .await;
        assert_eq!(report.started, 0);
        assert_eq!(
            harness
                .events
                .events()
                .iter()
                .filter(|event| matches!(
                    event,
                    LifecycleEvent::DemandObserved {
                        demand: 0,
                        not_matched: 1,
                        ..
                    }
                ))
                .count(),
            2
        );

        // Domain constructors permit extra selector-shaped labels. Routing
        // depends on all required labels and unique ownership, not their shape.
        // The store separately rejects sibling selector reuse on write/load.
        let mut superset = policy(1, "o/r", 2);
        assert!(
            superset
                .add_routing_label(fixtures::label("rm-home-win-x64-py-isolated"))
                .unwrap()
        );
        let mut sibling = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        sibling.activate().unwrap();
        let mut unique = Harness::simple(4, 0, "o/r");
        unique.demand.set(PollOutcome::Ready(QueuedDemand::of(
            repo("o/r"),
            [fixtures::queued_job(&[
                HOST_LABEL,
                "rm-home-win-x64-py-isolated",
            ])],
        )));
        let report = unique.reconciler.reconcile(&[superset, sibling]).await;
        assert_eq!(report.started, 1);
        assert_eq!(
            report
                .allocations
                .iter()
                .find(|allocation| allocation.to_start == 1)
                .unwrap()
                .policy_id,
            PolicyId::from_u128(1)
        );
        assert!(
            !unique
                .events
                .events()
                .iter()
                .any(|event| matches!(event, LifecycleEvent::RoutingRefused { .. }))
        );

        let mut overlap = Harness::simple(4, 1, "o/r");
        let report = overlap
            .reconciler
            .reconcile(&[policy(1, "o/r", 2), policy(2, "o/r", 2)])
            .await;
        assert_eq!(report.started, 0);
        assert!(overlap.events.events().iter().any(|event| matches!(
            event,
            LifecycleEvent::RoutingRefused {
                reason: RoutingRefusal::OverlappingProfiles,
                count: 1,
                ..
            }
        )));
        assert!(
            RoutingRefusal::OverlappingProfiles
                .to_string()
                .contains("repair")
        );
    }

    // =======================================================================
    // Failure paths
    // =======================================================================

    #[tokio::test]
    async fn a_failed_launch_stops_the_run_and_is_reported_without_free_text() {
        let launcher = Arc::new(FakeLauncher::new());
        launcher.fail_next(FailureReason::Other("token ghp_0123456789abcdef".into()));
        let mut harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
            Arc::new(InProcessAllocationLock::new()),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4)])
            .await;
        assert_eq!(report.started, 0);
        assert_eq!(report.allocations[0].to_start, 3, "the decision stands");

        let failures: Vec<&'static str> = harness
            .events
            .events()
            .into_iter()
            .filter_map(|event| match event {
                LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
                _ => None,
            })
            .collect();
        assert_eq!(failures, vec!["other"]);
        assert!(
            !failures[0].contains("ghp_"),
            "an event carried a `FailureReason::Other` detail verbatim"
        );
    }

    /// The host-fitness preflight: while a required host check fails, the pass
    /// still decides what it would start, but nothing reaches the inner
    /// launcher -- so nothing is registered with GitHub -- and the event names
    /// `host_unfit`. Clearing the flag lets the next pass start runners.
    #[tokio::test]
    async fn an_unfit_host_registers_no_runner_and_reports_host_unfit() {
        let launcher = Arc::new(FakeLauncher::new());
        let unfit = Arc::new(AtomicBool::new(true));
        let events = Arc::new(EventLog::new());
        let mut reconciler = Reconciler::new(
            host_with(4),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::ready(2, &repo("acme/app"))),
                launcher: Arc::new(FitnessGatedLauncher::new(
                    Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                    Arc::clone(&unfit),
                )),
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()) as Arc<dyn Clock>,
                jitter: Arc::new(NoJitter) as Arc<dyn Jitter>,
                events: Arc::clone(&events) as Arc<dyn EventSink>,
            },
        );
        let policies = [policy(1, "acme/app", 4)];

        let report = reconciler.reconcile(&policies).await;
        assert_eq!(report.started, 0);
        assert_eq!(launcher.launches(), 0, "the inner launcher must not run");
        let reasons: Vec<&'static str> = events
            .events()
            .into_iter()
            .filter_map(|event| match event {
                LifecycleEvent::RunnerStartFailed { reason, .. } => Some(reason),
                _ => None,
            })
            .collect();
        assert_eq!(reasons, ["host_unfit"]);

        unfit.store(false, Ordering::Release);
        let report = reconciler.reconcile(&policies).await;
        assert_eq!(report.started, 2);
        assert_eq!(launcher.launches(), 2);
    }

    #[tokio::test]
    async fn a_held_allocation_lock_starts_nothing_and_says_so() {
        let launcher = Arc::new(FakeLauncher::new());
        let mut harness = Harness::build(
            host_with(4),
            Arc::clone(&launcher),
            Arc::new(FakeDemand::ready(3, &repo("acme/app"))),
            Arc::new(HeldLock),
        );

        let report = harness
            .reconciler
            .reconcile(&[policy(1, "acme/app", 4)])
            .await;
        assert_eq!(report.started, 0);
        assert_eq!(
            report.deferred, 3,
            "three runners were granted and none was created; `deferred` counts grants, \
             not policies -- it reported `1` when a policy that launched two of five and \
             then lost the lock had left three unstarted"
        );
        assert_eq!(launcher.launches(), 0);
        assert_eq!(harness.events.count_of("allocation_deferred"), 1);
        assert!(
            harness
                .events
                .events()
                .iter()
                .any(|event| matches!(event, LifecycleEvent::AllocationDeferred { count: 3, .. })),
            "the event carries the same number the report does"
        );
        assert_eq!(
            report.allocations.len(),
            1,
            "the intent is still reported, so an operator staring at a queue sees why \
             nothing started"
        );
    }

    #[tokio::test]
    async fn a_foreign_or_draining_policy_is_reported_by_name_and_polls_nothing() {
        let mut harness = Harness::simple(8, 5, "acme/app");

        let foreign = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .repository("acme/app")
            .host(HostId::from_u128(0xdead))
            .autoscale("office", 4)
            .active()
            .build();
        let mut draining = policy(2, "acme/app", 4);
        draining.request_disable().unwrap();

        let report = harness.reconciler.reconcile(&[foreign, draining]).await;

        assert_eq!(report.started, 0);
        assert_eq!(
            harness.demand.polls().len(),
            0,
            "neither can act on an answer"
        );
        let factors: Vec<LimitingFactor> = report
            .allocations
            .iter()
            .map(|a| a.limiting_factor)
            .collect();
        assert!(factors.contains(&LimitingFactor::ForeignHost));
        assert!(factors.contains(&LimitingFactor::NotReconciling));
    }

    #[tokio::test]
    async fn an_unreadable_repository_list_makes_the_target_unreadable_not_empty() {
        // Polling a scope nobody chose would report a demand number for the
        // wrong set of repositories, which is worse than reporting nothing.
        #[derive(Debug)]
        struct BrokenDirectory;

        #[async_trait::async_trait]
        impl RepositoryDirectory for BrokenDirectory {
            async fn repositories(&self, _org: &Org) -> Result<Vec<OwnerRepo>, InventoryError> {
                Err(InventoryError::Cancelled)
            }
        }

        let launcher = Arc::new(FakeLauncher::new());
        let events = Arc::new(EventLog::new());
        let mut reconciler = Reconciler::new(
            host_with(4),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::default()),
                launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(BrokenDirectory),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::clone(&events) as Arc<dyn EventSink>,
            },
        );

        let org_policy = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .organization("acme")
            .autoscale("home", 4)
            .active()
            .build();

        let report = reconciler.reconcile(&[org_policy]).await;
        assert_eq!(report.started, 0);
        assert_eq!(report.unreadable, vec![PolicyId::from_u128(1)]);
        assert_eq!(events.count_of("target_unreadable"), 1);
    }

    // =======================================================================
    // What the events may carry
    // =======================================================================

    /// One value of every [`LifecycleEvent`] variant.
    ///
    /// Hand-written, and what keeps it honest is the wildcard-free `match` in
    /// [`LifecycleEvent::name`]: adding a variant stops that compiling and puts
    /// the author here. The same residual `b1` records for `FailureReason::ALL`
    /// applies — an author who writes the `name` arm and forgets this list gets
    /// a green suite with the variant unscanned.
    fn every_event() -> Vec<LifecycleEvent> {
        let policy = PolicyId::from_u128(0xabcd_ef01);
        let attempt = AttemptId::from_u128(0x1234_5678);
        vec![
            LifecycleEvent::DemandObserved {
                policy,
                demand: u32::MAX,
                not_matched: u32::MAX,
                unresolvable: u32::MAX,
                complete: false,
            },
            LifecycleEvent::TargetUnreadable {
                policy,
                reason: unreadable_reason(&RefreshState::Failed {
                    status: Some(500),
                    message: "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz"
                        .into(),
                }),
            },
            LifecycleEvent::Allocated {
                policy,
                demand: u32::MAX,
                desired: u16::MAX,
                active_owned: 7,
                headroom: 9,
                to_start: 2,
                limiting: LimitingFactor::HostCapacity,
            },
            LifecycleEvent::MonitorOnlySkipped { policy },
            LifecycleEvent::RunnerStarted { policy, attempt },
            LifecycleEvent::RunnerStartFailed {
                policy,
                reason: failure_reason_kind(&FailureReason::Other(
                    "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
                )),
            },
            LifecycleEvent::AllocationDeferred { policy, count: 4 },
            LifecycleEvent::AttemptsUnreadable {
                reason: failure_reason_kind(&FailureReason::Other(
                    "x-api-key: ghp_0123456789abcdefghijklmnopqrstuvwxyz".into(),
                )),
            },
            LifecycleEvent::HostCapacityUnreadable {
                reason: HostCapacityError::MissingHost,
            },
            LifecycleEvent::HostCapacityUnreadable {
                reason: HostCapacityError::JournalUnreadable,
            },
            LifecycleEvent::AttemptCleanFailed {
                policy,
                attempt,
                reason: failure_reason_kind(&FailureReason::ProcessExitedUnexpectedly),
            },
            LifecycleEvent::AttemptCleaned {
                policy,
                attempt,
                outcome: OutcomeKind::IdleExit,
            },
            LifecycleEvent::ScaleDownRefused { policy, attempt },
            LifecycleEvent::PollScheduled {
                retry_in_ms: 900_000,
                pace: PollPace::RateLimited {
                    kind: RateLimitKind::Primary,
                },
            },
        ]
    }

    /// `e1`'s Definition of Done: *"No emitted event contains a token, a JIT
    /// blob, or a credential header."*
    ///
    /// Asserted by rendering every variant and putting the result through `d1`'s
    /// own scrubber: if any of it looked like a credential to the redactor that
    /// guards the log file, the round trip would not be the identity. The
    /// positive control at the bottom is what stops that assertion passing
    /// because the scrubber is asleep.
    #[test]
    fn no_emitted_event_can_carry_a_credential() {
        use runner_manager_platform::logging::redact;

        for event in every_event() {
            let displayed = event.to_string();
            assert_eq!(
                redact(&displayed),
                displayed,
                "`{}` renders something `d1`'s sink would have to redact",
                event.name()
            );

            let debugged = format!("{event:?}");
            assert_eq!(
                redact(&debugged),
                debugged,
                "`{}`'s Debug renders something `d1`'s sink would have to redact",
                event.name()
            );
        }

        // The control: the scrubber is awake, and would have caught a credential
        // had one been there.
        let secret = "Authorization: Bearer ghp_0123456789abcdefghijklmnopqrstuvwxyz";
        assert_ne!(
            redact(secret),
            secret,
            "the scan above proves nothing if `redact` no longer recognises a credential"
        );
    }

    #[test]
    fn every_field_name_this_sink_emits_is_one_d1_allows() {
        use runner_manager_platform::logging::is_field_allowed;

        // The names `TracingEvents` writes. Kept beside the sink rather than
        // derived from it, because a derived list would move with the code and
        // assert nothing.
        for field in [
            "event",
            "policy_id",
            "attempt_id",
            "attempt_state",
            "demand",
            "desired",
            "capacity",
            "headroom",
            "count",
            "reason",
            "outcome",
            "mode",
            "lock",
            "retry_in_ms",
            "state",
        ] {
            assert!(
                is_field_allowed(field),
                "`{field}` is not on `d1`'s allow-list, so this sink would emit \
                 `[redacted]` in its place and the line would lose its meaning"
            );
        }
    }

    #[test]
    fn every_failure_reason_has_a_credential_free_kind() {
        for reason in FailureReason::ALL {
            let kind = failure_reason_kind(&reason);
            assert!(!kind.is_empty());
            assert!(
                kind.chars().all(|c| c.is_ascii_lowercase() || c == '_'),
                "`{kind}` is not a fixed identifier"
            );
        }
        assert_eq!(
            failure_reason_kind(&FailureReason::Other("ghp_secret".into())),
            "other",
            "the detail of an `Other` reason never reaches an event"
        );
        let categories = [
            IsolationProviderFailure::NotInstalled,
            IsolationProviderFailure::PermissionDenied,
            IsolationProviderFailure::ImageUnavailableOrIncompatible,
            IsolationProviderFailure::DiskQuotaUnavailable,
            IsolationProviderFailure::RuntimeOperationFailed,
        ];
        let mut kinds = std::collections::BTreeSet::new();
        for category in categories {
            let kind = failure_reason_kind(&FailureReason::IsolationProvider(category));
            assert_ne!(kind, "other");
            assert!(!kind.contains("secret"));
            kinds.insert(kind);
        }
        assert_eq!(kinds.len(), categories.len());
    }

    // =======================================================================
    // The two tripwires
    // =======================================================================

    /// One source file's production half, with comment lines dropped.
    ///
    /// Both exclusions are `c4`'s, and load-bearing for the same reasons. The
    /// **test module** goes because the tests in it legitimately name the shapes
    /// they forbid — this module's own positive control is a literal
    /// `async fn acquire_jobs`, which would accuse the file of the thing it is
    /// proving it does not do. The **comments** go because this module's
    /// documentation explains the seam at length and has to name what does not
    /// exist in order to say why; a scan that forbade the explanation is a scan
    /// that gets the explanation deleted.
    fn production_half_of(source: &str) -> String {
        let production = source
            .split_once("\n#[cfg(test)]")
            .map_or(source, |(production, _)| production);
        production
            .lines()
            .filter(|line| !line.trim_start().starts_with("//"))
            .collect::<Vec<_>>()
            .join("\n")
    }

    /// This file's own production half.
    fn this_file_above_its_tests_without_prose() -> String {
        production_half_of(include_str!("reconcile.rs"))
    }

    /// The one normalisation both halves of the scan use.
    ///
    /// # This is a second copy of `crates/github/src/demand.rs`, deliberately
    ///
    /// `production_half_of`, this function, [`FORBIDDEN`] and
    /// `forbidden_shape_in` together duplicate `demand.rs:1530-1619`. Sharing
    /// them would mean putting them in `crates/testkit`, which `e1` does not
    /// own, so the copy was the only option available to this task.
    ///
    /// **It is worth consolidating later, and here is the specific hazard.**
    /// The last defect in `c4`'s copy was two spellings of "the same"
    /// normalisation drifting apart — the haystack lower-cased and the needle
    /// not — which made three of its seven assertions vacuously true from the
    /// day they were written. Two copies is the same hazard one level up. The
    /// mitigation inside *this* copy is that one function serves both the scan
    /// and its positive control, so a normaliser that stops matching fails the
    /// control loudly rather than passing the scan silently; what that cannot
    /// catch is this copy and `c4`'s diverging from each other.
    fn normalise_for_scan(text: &str) -> String {
        text.to_ascii_lowercase().replace(['_', ' '], "")
    }

    /// The Actions-service call this design has no equivalent of, plus the
    /// shapes an implementer would invent in its place.
    ///
    /// Spelled in halves so that no needle ever appears whole in the text being
    /// scanned, and keyed to `fn`/`struct` so that the prose above may keep
    /// explaining why there is no reservation. `c4` records both trades at
    /// length; this list is its counterpart one layer up. Note that the
    /// allocation lock's own `fn acquire` is deliberately *not* matched: the
    /// needle is `acquire`-a-**job**, and a lock is not one.
    const FORBIDDEN: &[&str] = &[
        concat!("fn ", "acquire", "_job"),
        concat!("fn ", "claim", "_job"),
        concat!("fn ", "lease", "_job"),
        concat!("fn ", "reserve", "_job"),
        concat!("fn ", "ack", "nowledge"),
        concat!("struct ", "Job", "Lease"),
        concat!("struct ", "Job", "Claim"),
        concat!("struct ", "Job", "Reservation"),
    ];

    fn forbidden_shape_in(source: &str) -> Option<&'static str> {
        let haystack = normalise_for_scan(source);
        FORBIDDEN
            .iter()
            .copied()
            .find(|forbidden| haystack.contains(&normalise_for_scan(forbidden)))
    }

    /// `e1`'s Definition of Done: *"No reservation, claim, lease, or acquisition
    /// call exists in the crate; a test or review note records that this is
    /// deliberate rather than missing."*
    ///
    /// **Deliberate, not missing.** The scale-set model let a listener call
    /// `AcquireJobs` to claim an assignment before scaling; the REST path has no
    /// equivalent, so demand is advisory and two hosts serving the same labels
    /// can both start a runner for one queued run. Adding a local reservation
    /// table would not remove that — the other host cannot see it — it would
    /// only hide the surplus case from the tests that measure it. The three
    /// controls that actually bound it are host-scoped routing labels,
    /// `max_capacity`, and `host_capacity`, and the last two are enforced in
    /// this file.
    ///
    /// The scan is a tripwire on the obvious shape rather than a proof: a
    /// reservation reached through a trait method or a differently-named helper
    /// would walk past it. Review is the primary control, exactly as `c4` states
    /// for its own copy.
    ///
    /// # It scans the crate, because the bullet says "in the crate"
    ///
    /// It used to scan this file alone while quoting a crate-wide claim, which
    /// left `lifecycle.rs` — `e3`, the launcher, and by far the likeliest place
    /// for someone to "fix" the surplus-runner case with a local lease — covered
    /// by nothing. Reading another owner's file is not editing it, so ownership
    /// was never the obstacle.
    ///
    /// The walk below is `c4`'s, and it **recurses** for the reason `c4`
    /// records: a module directory (`src/reconcile/mod.rs`) arrives as an entry
    /// that does not end in `.rs`, so a flat filter drops it and takes every
    /// file underneath with it, leaving the scan passing over files it covers
    /// by nothing at all. The listed-versus-on-disk assertion is what stops
    /// `SOURCES` going stale the moment `e2` or `e3` adds a module.
    #[test]
    fn nothing_in_this_crate_reserves_or_claims_a_job() {
        const SOURCES: &[(&str, &str)] = &[
            ("dependency_caches.rs", include_str!("dependency_caches.rs")),
            ("docker_cleanup.rs", include_str!("docker_cleanup.rs")),
            (
                "docker_cleanup/http.rs",
                include_str!("docker_cleanup/http.rs"),
            ),
            (
                "docker_cleanup/proxy.rs",
                include_str!("docker_cleanup/proxy.rs"),
            ),
            (
                "docker_cleanup/tests.rs",
                include_str!("docker_cleanup/tests.rs"),
            ),
            ("lib.rs", include_str!("lib.rs")),
            ("lifecycle.rs", include_str!("lifecycle.rs")),
            (
                "lifecycle/windows_hyperv.rs",
                include_str!("lifecycle/windows_hyperv.rs"),
            ),
            ("oci.rs", include_str!("oci.rs")),
            ("package.rs", include_str!("package.rs")),
            ("reconcile.rs", include_str!("reconcile.rs")),
        ];

        fn walk(directory: &std::path::Path, prefix: &str, found: &mut Vec<String>) {
            for entry in std::fs::read_dir(directory).expect("the crate's own src/ is readable") {
                let entry = entry.expect("a readable directory entry");
                let name = entry.file_name().to_string_lossy().into_owned();
                // `/`-joined, which is what `include_str!` takes on every
                // platform, so the two sides compare directly.
                let joined = if prefix.is_empty() {
                    name.clone()
                } else {
                    format!("{prefix}/{name}")
                };
                if entry.path().is_dir() {
                    walk(&entry.path(), &joined, found);
                } else if name.ends_with(".rs") {
                    found.push(joined);
                }
            }
        }

        let mut listed: Vec<&str> = SOURCES.iter().map(|(name, _)| *name).collect();
        listed.sort_unstable();
        let mut on_disk = Vec::new();
        walk(
            std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/src")),
            "",
            &mut on_disk,
        );
        on_disk.sort_unstable();
        assert_eq!(
            listed, on_disk,
            "a source file was added or removed; this scan claims to cover the whole crate \
             and a stale list makes that claim false"
        );

        for (name, source) in SOURCES {
            assert_eq!(
                forbidden_shape_in(&production_half_of(source)),
                None,
                "{name} names a forbidden shape: there is no `AcquireJobs` equivalent over \
                 REST, and a local lease coordinates this host with itself and with nothing \
                 else. If an owner decision restored one, that decision belongs in this \
                 module's documentation and in this test before it belongs in the code"
            );
        }

        // The control: the scan can see a shape when there is one, through the
        // same matcher the loop above uses.
        assert!(
            forbidden_shape_in("async fn acquire_jobs(&self) -> Vec<Job> { todo!() }").is_some(),
            "the scan above proves nothing if the needles no longer match"
        );
    }

    /// This module **applies** `b1`'s label predicate and implements none of it.
    ///
    /// The counterpart to `c4`'s scan over `crates/github/src/demand.rs`, and it
    /// checks the opposite thing, because the two modules sit on opposite sides
    /// of the same seam. `c4` builds a `RunsOn` per queued job and must name no
    /// `RoutingLabels`; this module holds the policy whose labels decide, so it
    /// must call `RoutingLabels::tally` and must not re-derive what that call
    /// answers.
    ///
    /// So the scan is in two halves:
    ///
    /// * **Present.** `DemandTally` has to appear, because [`demand_for`]
    ///   returns one. A production half that named it nowhere would mean the
    ///   filtering had been dropped and every queued job in a watched repository
    ///   was driving this policy toward `max_capacity` again.
    /// * **Absent.** The vocabulary of a *second* implementation. `b1` names the
    ///   three outcomes of matching one job; this module consumes the aggregate
    ///   and never a single job's verdict, so naming `RunsOnMatch` or
    ///   `UnresolvableRunsOn` here means a `match` on an outcome that
    ///   `RoutingLabels::tally` has already decided — which is how two copies of
    ///   a predicate start.
    ///
    /// Like the needles in `nothing_in_this_module_reserves_or_claims_a_job`,
    /// this is a tripwire on the obvious shape rather than a proof: a hand-rolled
    /// comparison of raw label strings that never names a `policy` type would
    /// walk past it. Stated rather than implied, for the same reason it is
    /// stated there.
    #[test]
    fn the_label_predicate_is_b1s_and_this_module_only_applies_it() {
        let production = this_file_above_its_tests_without_prose();

        assert!(
            production.contains("DemandTally"),
            "the reconciliation loop must tally queued jobs against this policy's routing \
             labels. A production half that named `DemandTally` nowhere would mean the \
             label filtering had been removed, and a repository whose jobs target \
             `ubuntu-latest` would drive its policy toward `max_capacity` again"
        );

        for second_implementation in ["RunsOnMatch", "UnresolvableRunsOn"] {
            assert!(
                !production.contains(second_implementation),
                "the reconciliation loop names `{second_implementation}`, which is the \
                 vocabulary of deciding one job's `runs-on` -- and `RoutingLabels::tally` \
                 has already decided it. This module applies the predicate and does not \
                 re-implement it; if an owner decision changed that, it belongs in this \
                 module's documentation and in this test before it belongs in the code"
            );
        }
    }

    /// The demand this module clamps is the *matched* count, and a job this host
    /// cannot serve is not demand.
    ///
    /// The behaviour the whole reversal was for, asserted end to end through
    /// `demand_for` rather than through `b1`'s predicate in isolation: a policy
    /// carrying this host's labels, a reading holding some of its jobs and some
    /// of somebody else's, and the three counts kept apart.
    #[test]
    fn demand_is_the_queued_jobs_this_policy_can_actually_serve() {
        let policy = policy(1, "acme/app", 10);
        let reading = QueuedDemand::of(
            repo("acme/app"),
            [
                fixtures::queued_job(&[HOST_LABEL]),
                fixtures::queued_job(&[HOST_LABEL]),
                fixtures::queued_job(&["ubuntu-latest"]),
                fixtures::unresolvable_job(),
            ],
        );

        let tally = demand_for(&policy, &reading);

        assert_eq!(
            tally.demand(),
            2,
            "only the jobs whose required labels this policy carries are demand"
        );
        assert_eq!(
            tally.not_matched, 1,
            "a `ubuntu-latest` job is somebody else's work; counting it would start a \
             runner that idles until it times out"
        );
        assert_eq!(
            tally.unresolvable.len(),
            1,
            "an unresolvable `runs-on` is never demand and never discarded"
        );
    }

    /// A repository target reads its own repository; an organization target
    /// reads the whole scope.
    #[test]
    fn an_organization_policy_tallies_every_repository_its_scope_covers() {
        let mut per_repository = BTreeMap::new();
        per_repository.insert(repo("acme/left"), fixtures::queued_jobs(&[HOST_LABEL], 3));
        per_repository.insert(repo("acme/right"), fixtures::queued_jobs(&[HOST_LABEL], 4));
        let reading = QueuedDemand::new(per_repository);

        let repository_policy = policy(1, "acme/left", 10);
        assert_eq!(
            demand_for(&repository_policy, &reading).demand(),
            3,
            "a repository target reads its own repository's queue and not the aggregate"
        );

        let org_policy = fixtures::policy()
            .id(PolicyId::from_u128(2))
            .organization("acme")
            .autoscale("home", 10)
            .active()
            .build();
        assert_eq!(
            demand_for(&org_policy, &reading).demand(),
            7,
            "an organization policy serves any repository in its scope, so its demand is \
             the whole aggregate's"
        );
    }

    #[test]
    fn the_accepted_over_count_is_bounded_by_the_two_ceilings_and_nothing_else() {
        // The owner decision accepts that a repository whose jobs only target
        // `ubuntu-latest` still drives its policy toward `max_capacity`. What
        // stops that being unbounded is exactly what stops any other demand
        // being unbounded, which is asserted here rather than assumed.
        let host = host_with(2);
        let policy = policy(1, "acme/app", 5);
        let attempts: Vec<RunnerAttempt> = Vec::new();
        let mut allocator = HostAllocator::from_attempts(&host, &attempts);

        let allocation = allocator.allocate(&policy, u32::MAX);
        assert_eq!(allocation.desired, 5, "max_capacity beats reported demand");
        assert_eq!(allocation.to_start, 2, "host_capacity beats max_capacity");
        assert_eq!(allocation.limiting_factor, LimitingFactor::HostCapacity);
    }

    /// A director standing in for the coordinator's launches: a fixed count
    /// per policy, and every attempt it was told about.
    #[derive(Debug, Default)]
    struct FixedDirector {
        directed: Mutex<BTreeMap<PolicyId, u32>>,
        started: Mutex<Vec<(PolicyId, AttemptId)>>,
        failed: Mutex<Vec<(PolicyId, Option<AttemptId>)>>,
        /// What each look at the older launches answers; the last one repeats.
        older: Mutex<std::collections::VecDeque<usize>>,
        passed_over: AtomicUsize,
        /// Each policy's place among the launches.
        orders: Mutex<BTreeMap<PolicyId, u64>>,
    }

    impl DemandDirector for FixedDirector {
        fn directed_demand(&self, policy: &ScalePolicy) -> Option<u32> {
            self.directed.lock().unwrap().get(&policy.id).copied()
        }

        fn launched(&self, policy: &ScalePolicy, attempt: AttemptId) {
            self.started.lock().unwrap().push((policy.id, attempt));
        }

        fn start_failed(&self, policy: &ScalePolicy, attempt: Option<AttemptId>) {
            self.failed.lock().unwrap().push((policy.id, attempt));
        }

        fn next_waiting(&self, policy: &ScalePolicy) -> Option<u64> {
            self.orders.lock().unwrap().get(&policy.id).copied()
        }

        fn older_waiting(&self, _policy: &ScalePolicy) -> usize {
            let mut older = self.older.lock().unwrap();
            if older.len() > 1 {
                older.pop_front().unwrap_or(0)
            } else {
                older.front().copied().unwrap_or(0)
            }
        }

        fn pass_over_older(&self, _policy: &ScalePolicy) {
            self.passed_over.fetch_add(1, Ordering::SeqCst);
            self.older.lock().unwrap().clear();
        }
    }

    /// A launcher that holds every runner in its bring-up until as many others
    /// (or the test) are at the barrier.
    #[derive(Debug)]
    struct HeldBringUp {
        inner: Arc<FakeLauncher>,
        barrier: Arc<tokio::sync::Barrier>,
        fail: Option<FailureReason>,
        waits: Mutex<Vec<ReservationWaits>>,
    }

    #[async_trait::async_trait]
    impl RunnerLauncher for HeldBringUp {
        async fn host_capacity(&self, host: &Host) -> Result<NonZeroU16, HostCapacityError> {
            self.inner.host_capacity(host).await
        }

        async fn supervise(
            &self,
            policy: &ScalePolicy,
        ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
            self.inner.supervise(policy).await
        }

        async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
            self.inner.attempts().await
        }

        async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
            self.inner.launch(request).await
        }

        async fn finish_launch(
            &self,
            _policy: &ScalePolicy,
            attempt: RunnerAttempt,
            waits: ReservationWaits,
        ) -> Result<RunnerAttempt, LaunchFailure> {
            self.waits.lock().unwrap().push(waits);
            self.barrier.wait().await;
            match &self.fail {
                Some(reason) => Err(LaunchFailure::new(reason.clone())),
                None => Ok(attempt),
            }
        }

        async fn clean(&self, attempt: AttemptId) -> Result<(), LaunchFailure> {
            self.inner.clean(attempt).await
        }
    }

    fn directed_reconciler(
        capacity: u16,
        launcher: Arc<dyn RunnerLauncher>,
        lock: Arc<dyn AllocationLock>,
        director: Arc<FixedDirector>,
    ) -> Reconciler {
        Reconciler::new(
            host_with(capacity),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::ready(0, &repo("acme/app"))),
                launcher,
                lock,
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        )
        .with_director(director as Arc<dyn DemandDirector>)
    }

    /// Two targets' runners are being brought up at the same time: each is at
    /// the barrier with the allocation lock released, or neither gets there.
    #[tokio::test]
    async fn two_targets_bring_up_their_runners_at_once() {
        let fake = Arc::new(FakeLauncher::new());
        let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
        let barrier = Arc::new(tokio::sync::Barrier::new(2));
        let held = Arc::new(HeldBringUp {
            inner: Arc::clone(&fake),
            barrier,
            fail: None,
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
        director.directed.lock().unwrap().insert(one.id, 1);
        director.directed.lock().unwrap().insert(two.id, 1);
        let first = directed_reconciler(
            2,
            Arc::clone(&held) as _,
            Arc::clone(&lock),
            Arc::clone(&director),
        );
        let second = directed_reconciler(2, held as _, lock, director);

        let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
            tokio::join!(
                first.start_directed(std::slice::from_ref(&one)),
                second.start_directed(std::slice::from_ref(&two)),
            )
        })
        .await
        .expect("neither waits for the other's bring-up");
        assert_eq!((a.started, b.started), (1, 1));
    }

    /// No oversubscription: the slot is held from the reservation, while the
    /// runner is still being brought up and the lock is free.
    #[tokio::test]
    async fn a_runner_coming_up_holds_its_slot_against_another_target() {
        let fake = Arc::new(FakeLauncher::new());
        let lock: Arc<dyn AllocationLock> = Arc::new(InProcessAllocationLock::new());
        let barrier = Arc::new(tokio::sync::Barrier::new(2));
        let held = Arc::new(HeldBringUp {
            inner: Arc::clone(&fake),
            barrier: Arc::clone(&barrier),
            fail: None,
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let (one, two) = (policy(1, "acme/app", 2), policy(2, "acme/other", 2));
        director.directed.lock().unwrap().insert(one.id, 1);
        director.directed.lock().unwrap().insert(two.id, 1);
        let first = directed_reconciler(
            1,
            Arc::clone(&held) as _,
            Arc::clone(&lock),
            Arc::clone(&director),
        );
        let second = directed_reconciler(1, held as _, lock, director);

        let (a, b) = tokio::time::timeout(Duration::from_secs(10), async {
            tokio::join!(first.start_directed(std::slice::from_ref(&one)), async {
                while fake.live_count() == 0 {
                    tokio::task::yield_now().await;
                }
                let report = second.start_directed(std::slice::from_ref(&two)).await;
                barrier.wait().await;
                report
            })
        })
        .await
        .expect("the first is let go");
        assert_eq!((a.started, b.started), (1, 0));
        assert_eq!(fake.live_count(), 1);
    }

    /// The launch was given its runner at the reservation, and a bring-up that
    /// fails says so, rather than counting as started.
    #[tokio::test]
    async fn a_failed_bring_up_is_a_failed_start() {
        let fake = Arc::new(FakeLauncher::new());
        let held = Arc::new(HeldBringUp {
            inner: fake,
            barrier: Arc::new(tokio::sync::Barrier::new(1)),
            fail: Some(FailureReason::JitRequestFailed),
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 2);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            2,
            held as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );

        let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;

        assert_eq!(report.started, 0);
        let started = director.started.lock().unwrap()[0].1;
        assert_eq!(
            *director.failed.lock().unwrap(),
            vec![(pool.id, Some(started))]
        );
    }

    fn scripted_older(director: &FixedDirector, looks: &[(usize, usize)]) {
        *director.older.lock().unwrap() = looks
            .iter()
            .flat_map(|(value, count)| std::iter::repeat_n(*value, *count))
            .collect();
    }

    /// Oldest first across targets: a younger launch reserves after the older
    /// ones, however long that takes while they are reserving, and says so in its
    /// timings.
    #[tokio::test(start_paused = true)]
    async fn a_younger_launch_reserves_after_older_ones_while_they_make_progress() {
        let held = Arc::new(HeldBringUp {
            inner: Arc::new(FakeLauncher::new()),
            barrier: Arc::new(tokio::sync::Barrier::new(1)),
            fail: None,
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            5,
            Arc::clone(&held) as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );

        // Three older launches, each reserving just within the turn: four and a
        // half seconds in all, well over one turn, and nothing is passed over.
        let looks =
            (OLDER_LAUNCH_TURN.as_millis() * 3 / 4 / OLDER_LAUNCH_POLL.as_millis()) as usize;
        scripted_older(&director, &[(3, looks), (2, looks), (1, looks), (0, 1)]);
        let began = tokio::time::Instant::now();
        let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 1);
        assert_eq!(
            began.elapsed(),
            OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap()
        );
        assert_eq!(director.passed_over.load(Ordering::SeqCst), 0);
        assert_eq!(
            held.waits.lock().unwrap()[0].order,
            OLDER_LAUNCH_POLL * u32::try_from(looks * 3).unwrap(),
            "the wait is in the launch's timings"
        );
    }

    /// A pass concludes runners, so it is not held by a launch's bring-up: with a
    /// launch side, it reports on the launch and hands it over.
    #[tokio::test]
    async fn a_pass_hands_a_directed_launch_to_the_launch_side() {
        let fake = Arc::new(FakeLauncher::new());
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let signal = Arc::new(tokio::sync::Notify::new());
        let mut reconciler = directed_reconciler(
            5,
            Arc::clone(&fake) as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        )
        .with_launch_side(Arc::clone(&signal));

        let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;

        assert_eq!(report.started, 0);
        assert_eq!(fake.live_count(), 0);
        assert_eq!(report.allocations[0].to_start, 1);
        tokio::time::timeout(Duration::from_secs(1), signal.notified())
            .await
            .expect("the launch side is told");

        // Nothing to start: nothing to tell it.
        director.directed.lock().unwrap().insert(pool.id, 0);
        reconciler.reconcile(std::slice::from_ref(&pool)).await;
        assert!(
            tokio::time::timeout(Duration::from_millis(50), signal.notified())
                .await
                .is_err()
        );
    }

    /// One that never reserves costs one wait, then nothing: a loop that cannot
    /// reserve holds no launch up for long, and not every later one again.
    #[tokio::test(start_paused = true)]
    async fn an_older_launch_that_cannot_reserve_is_passed_over_once() {
        let fake = Arc::new(FakeLauncher::new());
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            5,
            Arc::clone(&fake) as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );

        scripted_older(&director, &[(1, 1)]);
        let began = tokio::time::Instant::now();
        let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 1, "started anyway");
        assert!(began.elapsed() >= OLDER_LAUNCH_TURN);
        assert!(began.elapsed() < OLDER_LAUNCH_TURN + Duration::from_secs(1));
        assert_eq!(director.passed_over.load(Ordering::SeqCst), 1);

        director.directed.lock().unwrap().insert(pool.id, 2);
        let began = tokio::time::Instant::now();
        let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 1);
        assert_eq!(began.elapsed(), Duration::ZERO);
    }

    /// A target's own profiles keep the order too: the oldest launch reserves
    /// first, one runner at a time.
    #[tokio::test]
    async fn a_targets_profiles_reserve_oldest_launch_first() {
        let director = Arc::new(FixedDirector::default());
        let (younger, older) = (policy(1, "acme/app", 3), policy(2, "acme/app", 3));
        director.directed.lock().unwrap().insert(younger.id, 1);
        director.directed.lock().unwrap().insert(older.id, 1);
        director.orders.lock().unwrap().insert(younger.id, 9);
        director.orders.lock().unwrap().insert(older.id, 4);
        let reconciler = directed_reconciler(
            5,
            Arc::new(FakeLauncher::new()) as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );

        let report = reconciler
            .start_directed(&[younger.clone(), older.clone()])
            .await;

        assert_eq!(report.started, 2);
        let order: Vec<PolicyId> = director
            .started
            .lock()
            .unwrap()
            .iter()
            .map(|(policy, _)| *policy)
            .collect();
        assert_eq!(order, vec![older.id, younger.id]);
    }

    /// Serves `pool` until told to stop, with `signal` and `stop` for the test.
    fn serving<'a>(
        reconciler: &'a Reconciler,
        pool: &ScalePolicy,
        signal: &'a tokio::sync::Notify,
        stop: tokio::sync::watch::Receiver<bool>,
    ) -> impl std::future::Future<Output = ()> + 'a {
        let pool = pool.clone();
        async move {
            let policies = move || vec![pool.clone()];
            reconciler.serve_launches(&policies, signal, stop).await;
        }
    }

    /// The batch does not hold a launch back. Each bring-up waits at the barrier
    /// for the other, so the second must reach it while the first is coming up, or
    /// neither ever gets past.
    #[tokio::test]
    async fn a_launch_accepted_during_a_batch_starts_without_waiting_for_it() {
        let fake = Arc::new(FakeLauncher::new());
        let held = Arc::new(HeldBringUp {
            inner: Arc::clone(&fake),
            barrier: Arc::new(tokio::sync::Barrier::new(2)),
            fail: None,
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            5,
            held as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );
        let signal = tokio::sync::Notify::new();
        let (stop, stopped) = tokio::sync::watch::channel(false);

        tokio::time::timeout(Duration::from_secs(10), async {
            tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
                signal.notify_one();
                while director.started.lock().unwrap().is_empty() {
                    tokio::task::yield_now().await;
                }
                director.directed.lock().unwrap().insert(pool.id, 2);
                signal.notify_one();
                while director.started.lock().unwrap().len() < 2 {
                    tokio::task::yield_now().await;
                }
                stop.send_replace(true);
            })
        })
        .await
        .expect("the second came up beside the first, and stopping ends it");
        assert_eq!(fake.live_count(), 2);
    }

    /// Serving launches beside a batch changes no ceiling: the runner still
    /// coming up holds its slot.
    #[tokio::test]
    async fn a_launch_during_a_batch_takes_no_slot_that_batch_holds() {
        let fake = Arc::new(FakeLauncher::new());
        let held = Arc::new(HeldBringUp {
            inner: Arc::clone(&fake),
            barrier: Arc::new(tokio::sync::Barrier::new(2)),
            fail: None,
            waits: Mutex::default(),
        });
        let barrier = Arc::clone(&held.barrier);
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            1,
            held as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );
        let signal = tokio::sync::Notify::new();
        let (stop, stopped) = tokio::sync::watch::channel(false);

        tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
            signal.notify_one();
            while director.started.lock().unwrap().is_empty() {
                tokio::task::yield_now().await;
            }
            director.directed.lock().unwrap().insert(pool.id, 2);
            signal.notify_one();
            for _ in 0..100 {
                tokio::task::yield_now().await;
            }
            assert_eq!(
                director.started.lock().unwrap().len(),
                1,
                "the host is full"
            );
            assert_eq!(fake.live_count(), 1);
            stop.send_replace(true);
            barrier.wait().await;
        });
    }

    /// Nothing reserved waits for the rest of the round: here the next reservation
    /// cannot start until the first runner's bring-up has run, because an older
    /// launch elsewhere holds it for a whole turn.
    #[tokio::test(start_paused = true)]
    async fn a_reserved_runner_comes_up_before_the_next_reservation_ends() {
        let held = Arc::new(HeldBringUp {
            inner: Arc::new(FakeLauncher::new()),
            barrier: Arc::new(tokio::sync::Barrier::new(1)),
            fail: None,
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 2);
        let reconciler = directed_reconciler(
            5,
            Arc::clone(&held) as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );
        let signal = tokio::sync::Notify::new();
        let (stop, stopped) = tokio::sync::watch::channel(false);

        // Nothing older at the first look, and one that never reserves at the next.
        scripted_older(&director, &[(0, 1), (1, 1)]);
        tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
            signal.notify_one();
            tokio::time::sleep(Duration::from_millis(1)).await;
            assert_eq!(
                held.waits.lock().unwrap().len(),
                1,
                "the first is coming up"
            );
            tokio::time::sleep(OLDER_LAUNCH_TURN + Duration::from_secs(1)).await;
            assert_eq!(held.waits.lock().unwrap().len(), 2);
            stop.send_replace(true);
        });
    }

    /// A launch whose runner failed coming up is tried again, without a signal
    /// and without a pass. One signalled after the stop starts nothing.
    #[tokio::test(start_paused = true)]
    async fn a_launch_that_did_not_start_is_retried_and_a_stop_ends_it() {
        let held = Arc::new(HeldBringUp {
            inner: Arc::new(FakeLauncher::new()),
            barrier: Arc::new(tokio::sync::Barrier::new(1)),
            fail: Some(FailureReason::JitRequestFailed),
            waits: Mutex::default(),
        });
        let director = Arc::new(FixedDirector::default());
        let pool = policy(1, "acme/app", 3);
        director.directed.lock().unwrap().insert(pool.id, 1);
        director.orders.lock().unwrap().insert(pool.id, 1);
        let reconciler = directed_reconciler(
            5,
            held as _,
            Arc::new(InProcessAllocationLock::new()),
            Arc::clone(&director),
        );
        let signal = tokio::sync::Notify::new();
        let (stop, stopped) = tokio::sync::watch::channel(false);

        tokio::join!(serving(&reconciler, &pool, &signal, stopped), async {
            signal.notify_one();
            tokio::time::sleep(Duration::from_millis(1)).await;
            assert_eq!(director.failed.lock().unwrap().len(), 1);
            // The fake keeps the failed runner: one more launch needs one more.
            director.directed.lock().unwrap().insert(pool.id, 2);
            tokio::time::sleep(LAUNCH_RETRY - Duration::from_secs(1)).await;
            assert_eq!(director.failed.lock().unwrap().len(), 1, "not yet");
            tokio::time::sleep(Duration::from_secs(2)).await;
            assert_eq!(director.failed.lock().unwrap().len(), 2, "retried");

            stop.send_replace(true);
            signal.notify_one();
        });
        assert_eq!(director.failed.lock().unwrap().len(), 2);
    }

    /// Coordinated placement: a directed policy starts runners for the
    /// coordinator's launches only, through every ceiling, whatever its own
    /// reading says; an undirected one keeps its own demand.
    #[tokio::test]
    async fn a_directed_policy_starts_only_what_it_was_directed_to_within_its_ceilings() {
        let demand = Arc::new(FakeDemand::ready(5, &repo("acme/app")));
        let launcher = Arc::new(FakeLauncher::new());
        let director = Arc::new(FixedDirector::default());
        let mut reconciler = Reconciler::new(
            host_with(10),
            ReconcilerPorts {
                demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
                launcher: Arc::clone(&launcher) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        )
        .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
        let pool = policy(1, "acme/app", 2);

        // Five queued jobs on the poll, but no launch placed here: nothing.
        director.directed.lock().unwrap().insert(pool.id, 0);
        let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 0);

        // One launch: one runner, reported back to the director.
        director.directed.lock().unwrap().insert(pool.id, 1);
        let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 1);
        assert_eq!(director.started.lock().unwrap().len(), 1);

        // The same launch again is covered by the runner already starting.
        let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 0);

        // Three launches against max_capacity 2: the ceiling holds.
        director.directed.lock().unwrap().insert(pool.id, 3);
        let report = reconciler.reconcile(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 1);
        assert_eq!(launcher.live_count(), 2);

        // A launch-only start: every open launch at once, and no poll.
        let polls = demand.polls().len();
        director.directed.lock().unwrap().insert(pool.id, 0);
        let report = reconciler.start_directed(std::slice::from_ref(&pool)).await;
        assert_eq!(report.started, 0);
        let fresh = Arc::new(FakeLauncher::new());
        let starter = Reconciler::new(
            host_with(10),
            ReconcilerPorts {
                demand: Arc::clone(&demand) as Arc<dyn DemandSource>,
                launcher: Arc::clone(&fresh) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        )
        .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
        let wide = policy(2, "acme/app", 6);
        director.directed.lock().unwrap().insert(wide.id, 6);
        let report = starter.start_directed(std::slice::from_ref(&wide)).await;
        assert_eq!(report.started, 6, "six launches, one call");
        assert_eq!(demand.polls().len(), polls, "GitHub was not polled");
        // An undirected policy starts nothing this way.
        assert_eq!(
            starter
                .start_directed(&[policy(3, "acme/app", 2)])
                .await
                .started,
            0
        );

        // Its own reading failed: a directed policy still starts its launch.
        let failing = Arc::new(FakeLauncher::new());
        let mut reconciler = Reconciler::new(
            host_with(10),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::failing(RefreshState::Offline)),
                launcher: Arc::clone(&failing) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        )
        .with_director(Arc::clone(&director) as Arc<dyn DemandDirector>);
        director.directed.lock().unwrap().insert(pool.id, 1);
        assert_eq!(reconciler.reconcile(&[pool]).await.started, 1);

        // Undirected (fallback): its own five jobs, clamped to max 2.
        let own = Arc::new(FakeLauncher::new());
        let mut reconciler = Reconciler::new(
            host_with(10),
            ReconcilerPorts {
                demand: Arc::new(FakeDemand::ready(5, &repo("acme/app"))),
                launcher: Arc::clone(&own) as Arc<dyn RunnerLauncher>,
                lock: Arc::new(InProcessAllocationLock::new()),
                directory: Arc::new(FakeDirectory::default()),
                clock: Arc::new(FakeClock::default()),
                jitter: Arc::new(NoJitter),
                events: Arc::new(NoEvents),
            },
        )
        .with_director(Arc::new(OwnDemand));
        assert_eq!(
            reconciler
                .reconcile(&[policy(1, "acme/app", 2)])
                .await
                .started,
            2
        );
    }
}