runner-manager 0.4.42

// owner: f3-cli-daemon-service

//! The foreground agent and its graceful drain boundary.

use std::collections::BTreeSet;
use std::future::Future;
use std::io::Write;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

#[cfg(test)]
use runner_manager_agent::lifecycle::NativeProcesses;
use runner_manager_agent::lifecycle::{
    CachedRuntimePackages, LifecycleGithub, LifecycleGithubObservation, LifecycleLauncher,
    LifecyclePorts, PersistentDemand, PlatformExecutionProvider, RetryPolicy, TokioRetryDelay,
    TracingAttemptEvents,
};
use runner_manager_agent::package::{
    CachePorts, ExponentialBackoff, GatewayCatalog, HttpFetcher, PackageCache,
};
use runner_manager_agent::reconcile::{
    AllocationLock, FileAllocationLock, FitnessGatedLauncher, GatewayDemand, RandomJitter,
    ReconcileReport, Reconciler, ReconcilerPorts, RepositoryDirectory, TeeEvents, TracingEvents,
    WslRecoveryAllocationLock,
};
use runner_manager_domain::attempt::{AttemptState, FailureReason, active_count, active_count_for};
use runner_manager_domain::model::{AttemptId, Clock, Org, OwnerRepo, ScaleTarget, StartMode};
use runner_manager_domain::policy::{PolicyState, ScalePolicy};
use runner_manager_domain::store::Store;
use runner_manager_github::demand::RestDemand;
use runner_manager_github::device_flow::DeviceFlow;
use runner_manager_github::jit::{JitError, JitGateway, JitRunnerRequest, RestJit};
use runner_manager_github::rest::{
    CancelToken, InventoryError, InventoryGateway, RefreshState, RestInventory,
};
use runner_manager_github::{
    AppRegistration, AuthenticatedClient, CredentialRenewal, GithubError, UserAccessToken,
};
use runner_manager_platform::lock::{HostLock, LockError, LockKind};
use runner_manager_platform::service::{
    InstallRecord, clear_credential_unreadable, clear_github_credential_rejection,
    record_credential_unreadable, record_github_contact, record_github_credential_rejected,
    sign_in_instruction,
};
use runner_manager_platform::wsl::fence::{
    DrainRequest, GuestHeartbeat, GuestRecoveryConfig,
    SCHEMA_VERSION as WSL_RECOVERY_SCHEMA_VERSION, unmanaged_runner_service_count,
};

use runner_manager_platform::secrets::SecretStore;
use secrecy::{ExposeSecret as _, SecretString};

use super::{CliError, Context, DaemonCommand, Failure, write_failed};

pub fn dispatch(
    context: &Context,
    command: &DaemonCommand,
    out: &mut dyn Write,
    service_shutdown: Option<runner_manager_platform::service::ServiceShutdown>,
) -> Result<(), CliError> {
    match command {
        DaemonCommand::Run(args) => {
            #[cfg(windows)]
            if std::env::var_os(super::service::SUPERVISED_ENVIRONMENT).is_none() {
                match super::service::migrate_legacy_windows_login_registration(context, out) {
                    Ok(true) => return Ok(()),
                    Ok(false) => {}
                    Err(error) => tracing::warn!(
                        %error,
                        "the legacy Windows login task could not start its restart supervisor; continuing under Task Scheduler"
                    ),
                }
            }
            let runtime = super::runtime()?;
            runtime.block_on(run(
                context,
                out,
                service_shutdown,
                args.windows_service_host,
            ))
        }
        DaemonCommand::AdoptCredential(args) => {
            super::auth::adopt_credential(context, args.start_at, out)
        }
    }
}

async fn run(
    context: &Context,
    out: &mut dyn Write,
    service_shutdown: Option<runner_manager_platform::service::ServiceShutdown>,
    windows_service_host: bool,
) -> Result<(), CliError> {
    loop {
        #[cfg(target_os = "macos")]
        if converge_launchd_definition(context, out, service_shutdown.clone()).await? {
            return Ok(());
        }
        match run_generation(context, out, service_shutdown.clone(), windows_service_host).await? {
            DaemonOutcome::Stopped => return Ok(()),
            DaemonOutcome::Reload => {}
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DaemonOutcome {
    Stopped,
    Reload,
}

async fn run_generation(
    context: &Context,
    out: &mut dyn Write,
    service_shutdown: Option<runner_manager_platform::service::ServiceShutdown>,
    windows_service_host: bool,
) -> Result<DaemonOutcome, CliError> {
    let instance = acquire_instance(context)?;
    // From here on this generation's loop beats, so a blocked one reads as
    // stalled in `status` and `service status` rather than as healthy.
    let _heartbeat = Heartbeat::start(context.paths().clone());
    reset_launch_state(context);
    let store = Arc::new(context.store()?);
    let host = super::host::local_host_or_create(context, store.as_ref())?;
    let targets = active_autoscale_targets(store.policies().map_err(local_store_failure)?);
    let failed = write_failed("the daemon state");

    // The service runs a private copy so package managers can replace their
    // source while it is running. Resolve that source before the idle branch:
    // a host with every policy disabled still has to take upgrades.
    let own_binary = InstallRecord::read(context.paths())
        .ok()
        .flatten()
        .and_then(|record| record.source_binary);

    writeln!(out, "daemon running (pid {})", std::process::id()).map_err(failed)?;

    // A host with no policies owns no runner work, but it may still own a
    // renewable credential. Keep that pair rotating: otherwise an installed,
    // healthy service can sit quietly until the six-month refresh half expires
    // and require a person merely because no TUI happened to be open.
    if targets.is_empty() {
        let heartbeat_store = Arc::clone(&store) as Arc<dyn Store>;
        let heartbeat_paths = context.paths().clone();
        tokio::select! {
            signal = wait_for_shutdown(service_shutdown) => {
                signal.map_err(signal_failure)?;
                writeln!(out, "daemon stopped; no runner was terminated").map_err(failed)?;
                return Ok(DaemonOutcome::Stopped);
            }
            version = async {
                match own_binary.clone() {
                    Some(path) => wait_for_upgrade(path).await,
                    None => std::future::pending().await,
                }
            } => {
                return stop_for_upgrade(context, host.service_start_mode, own_binary.as_deref(), &version, out)
                    .map(|()| DaemonOutcome::Stopped);
            }
            () = wait_for_policy_set_change(Arc::clone(&store) as Arc<dyn Store>, BTreeSet::new()) => {
                writeln!(out, "a repository policy was added; reloading").map_err(failed)?;
                return Ok(DaemonOutcome::Reload);
            }
            () = maintain_idle_credential(context, host.service_start_mode) => {
                unreachable!("credential maintenance runs until the daemon is stopped")
            },
            () = answer_github_requests(context, host.service_start_mode) => {
                unreachable!("GitHub requests are answered until the daemon is stopped")
            },
            () = maintain_diagnostics(context.paths().logs_dir().to_path_buf()) => {
                unreachable!("diagnostics maintenance runs until the daemon is stopped")
            },
            () = maintain_docker_cleanup(
                runner_manager_agent::docker_cleanup::DockerCleanup::new(host.id, context.paths().config_dir(), context.paths().state_dir()),
                Arc::clone(&store) as Arc<dyn Store>,
            ) => {
                unreachable!("Docker cleanup maintenance runs until the daemon is stopped")
            },
            // No runner grows the caches on an idle host, so nothing prunes on
            // a schedule; an operator's `host cache prune` is still honoured.
            () = maintain_dependency_caches(
                super::cache::daemon_caches(context, Arc::clone(&store) as Arc<dyn Store>, host.id),
                context.paths().state_dir().to_path_buf(),
                false,
            ) => {
                unreachable!("dependency cache maintenance runs until the daemon is stopped")
            },
            () = maintain_wsl_guest_heartbeat(heartbeat_paths, heartbeat_store) => {
                unreachable!("WSL heartbeat maintenance runs until the daemon is stopped")
            },
            () = maintain_wsl_recovery_without_local_policies(
                context,
                host.service_start_mode,
                windows_service_host,
            ) => {
                unreachable!("WSL recovery watchdog runs until the daemon is stopped")
            },
        }
    }

    let mode = host.service_start_mode;
    let secrets = context.secret_store(mode)?;
    let loaded = secrets.load();
    // Said on disk either way, because this process is about to exit and be
    // restarted by the service manager, and `service status` cannot see why
    // otherwise.
    note_credential_readability(context, loaded.is_ok());
    let secret = loaded
        .map_err(|source| {
            CliError::with_remedy(
                Failure::SecretStore,
                format!("cannot read the stored GitHub credential: {source}"),
                sign_in_instruction(mode),
            )
        })?
        .ok_or_else(|| {
            CliError::with_remedy(
                Failure::NotAuthenticated,
                "no GitHub credential is stored for this daemon's start mode",
                sign_in_instruction(mode),
            )
        })?;
    let app = context.app_registration()?;
    // ------------------------------------------------------------------
    // THE DAEMON IS THE REASON RENEWAL EXISTS.
    // ------------------------------------------------------------------
    // An access token lives eight hours and this process is meant to run for
    // months, so without renewal it would need an interactive sign-in every
    // working day. With it, `auth login` happens once per machine -- and,
    // because each host renews its own pair rather than re-authorising, two
    // machines stop evicting each other's credential.
    //
    // A credential with no refresh half -- which is every one issued while the
    // App has expiration switched off -- simply never renews, and this costs it
    // nothing.
    let renewal: Arc<dyn CredentialRenewal> = Arc::new(super::auth::StoringRenewal::new(
        DeviceFlow::new(app.clone(), context.endpoints().clone()).map_err(|source| {
            CliError::new(
                Failure::GithubUnavailable,
                format!("cannot prepare credential renewal: {source}"),
            )
        })?,
        Arc::clone(&secrets),
        context
            .paths()
            .state_dir()
            .join(LockKind::CredentialRenewal.file_name()),
    ));
    let client = Arc::new(
        AuthenticatedClient::new(
            context.endpoints().clone(),
            UserAccessToken::from_stored(secret),
            context.clock(),
        )
        .map_err(github_failure)?
        .with_renewal(renewal)
        // ------------------------------------------------------------------
        // AND A WAY TO NOTICE A SIGN-IN THAT ALREADY HAPPENED.
        // ------------------------------------------------------------------
        // The credential above was read once, just now. Renewal keeps it
        // current for as long as it is renewable -- but a daemon that starts
        // holding a credential already past saving has no refresh half to
        // spend, and without this it never reads the store again, so the
        // `auth login` an operator runs to fix it changes nothing until
        // somebody restarts the service. That cost this project 28 hours on
        // one host; see `docs/spikes/token-expiry-and-renewal.md`.
        .with_credential_source(Arc::new(super::auth::StoredCredential::new(Arc::clone(
            &secrets,
        )))),
    );
    let clock = context.clock();
    let inventory = Arc::new(RestInventory::new(Arc::clone(&client), Arc::clone(&clock)));
    let wsl_inventory = Arc::clone(&inventory) as Arc<dyn InventoryGateway>;
    let jit = Arc::new(RestJit::new(Arc::clone(&client)));
    let lifecycle_github = Arc::new(GithubLifecycle {
        jit,
        inventory: Arc::clone(&inventory),
        clock: Arc::clone(&clock),
    });
    let directory = Arc::new(GithubDirectory {
        client: Arc::clone(&client),
        app,
    });

    let paths = Arc::new(context.paths().clone());
    let events: Arc<dyn runner_manager_agent::reconcile::EventSink> = Arc::new(TeeEvents(
        Arc::new(TracingEvents),
        Arc::new(runner_manager_agent::reconcile::EventLog::new()),
    ));
    let base_lock: Arc<dyn AllocationLock> = Arc::new(FileAllocationLock::new(Arc::clone(&paths)));
    let shared_lock = Arc::new(WslRecoveryAllocationLock::new(
        Arc::clone(&paths),
        base_lock,
    ));
    // ------------------------------------------------------------------
    // NO RUNNER STARTS ON A HOST THAT FAILS A REQUIRED CHECK.
    // ------------------------------------------------------------------
    // Evaluated before the first pass, and again every few minutes by
    // `maintain_host_fitness`. While it fails, every launch is refused with
    // `host_unfit` before a runner is registered with GitHub; supervision,
    // cleanup and recovery of what already runs carry on.
    let host_unfit = Arc::new(AtomicBool::new(false));
    observe_host_fitness(context, &host_unfit, true).await;
    // One handle for every target: the settings file and the cache root are
    // the host's, and the namespace comes from each launch's policy.
    let dependency_caches =
        super::cache::daemon_caches(context, Arc::clone(&store) as Arc<dyn Store>, host.id);
    // ------------------------------------------------------------------
    // ONE DEMAND GATEWAY FOR EVERY TARGET.
    // ------------------------------------------------------------------
    // Shared, so a repository covered by two targets is read once per idle
    // interval, every listing's `ETag` is kept in one place, and a rate limit
    // one loop meets silences every loop. See `RestDemand`.
    let intervals = match super::polling::configured(context, store.as_ref()) {
        Ok((intervals, None)) => intervals,
        Ok((intervals, Some(reason))) => {
            tracing::warn!(%reason, "the idle poll interval could not be read; using the default");
            intervals
        }
        Err(error) => {
            tracing::warn!(%error, "the poll intervals could not be read; using the defaults");
            super::polling::Intervals::with_default_idle(host.refresh_interval)
        }
    };
    let demand_gateway = Arc::new(RestDemand::new(Arc::clone(&client), Arc::clone(&clock)));
    let polling = Arc::new(super::polling::PollingMonitor::new(
        context.paths().clone(),
        Arc::clone(&store) as Arc<dyn Store>,
        host.id,
        Arc::clone(&demand_gateway),
        Arc::clone(&clock),
        intervals,
    ));
    let mut managed_targets = Vec::with_capacity(targets.len());
    for policies in targets {
        let package_target = policies[0].target.clone();
        let catalog = Arc::new(GatewayCatalog::new(
            RestInventory::new(Arc::clone(&client), Arc::clone(&clock)),
            package_target.clone(),
        ));
        let cache = Arc::new(PackageCache::new(
            context.paths(),
            host.os,
            host.architecture,
            CachePorts {
                catalog,
                fetcher: Arc::new(HttpFetcher::default()),
                backoff: Arc::new(ExponentialBackoff::default()),
                clock: Arc::clone(&clock),
            },
        ));
        let lifecycle_store = Arc::new(TargetRecoveryStore::new(
            Arc::clone(&store) as Arc<dyn Store>,
            &policies,
        ));
        let launcher = Arc::new(LifecycleLauncher::new(
            host.id,
            context.paths().clone(),
            context.paths().logs_dir(),
            1,
            runner_manager_domain::attempt::RecoveryTimeouts::provisional(),
            RetryPolicy::bounded(3, Duration::from_secs(2), Duration::from_secs(30)),
            LifecyclePorts {
                store: Arc::clone(&lifecycle_store) as Arc<dyn Store>,
                github: Arc::clone(&lifecycle_github) as Arc<dyn LifecycleGithub>,
                packages: Arc::new(CachedRuntimePackages::new(cache)),
                processes: Arc::new(
                    PlatformExecutionProvider::new(host.id)
                        .with_runner_env_file(runner_manager_platform::runner_env::path_in(
                            context.paths().config_dir(),
                        ))
                        .with_dependency_caches(dependency_caches.clone())
                        .with_docker_cleanup(
                            runner_manager_agent::docker_cleanup::DockerCleanup::new(
                                host.id,
                                context.paths().config_dir(),
                                context.paths().state_dir(),
                            ),
                        ),
                ),
                clock: Arc::clone(&clock),
                demand: Arc::new(PersistentDemand),
                delay: Arc::new(TokioRetryDelay),
                events: Arc::new(TracingAttemptEvents),
                reconcile_events: Arc::clone(&events),
            },
        ));
        launcher
            .recover_startup(&policies)
            .await
            .map_err(|source| {
                CliError::new(
                    Failure::LocalState,
                    format!(
                        "startup recovery for {} did not complete: {source}",
                        package_target
                    ),
                )
            })?;
        lifecycle_store.finish_recovery();
        let cancel = CancelToken::new();
        let demand = Arc::new(GatewayDemand::new(
            Arc::clone(&demand_gateway),
            cancel.clone(),
        ));
        managed_targets.push(ManagedTarget {
            policies,
            store: Arc::clone(&store) as Arc<dyn Store>,
            reconciler: Reconciler::new(
                host.clone(),
                ReconcilerPorts {
                    demand,
                    launcher: Arc::new(FitnessGatedLauncher::new(
                        launcher,
                        Arc::clone(&host_unfit),
                    )),
                    lock: Arc::clone(&shared_lock) as Arc<_>,
                    directory: Arc::clone(&directory) as Arc<_>,
                    clock: Arc::clone(&clock),
                    jitter: Arc::new(RandomJitter),
                    events: Arc::clone(&events),
                },
            ),
            cancel,
            polling: Arc::clone(&polling),
        });
    }

    let (shutdown, _) = tokio::sync::watch::channel(false);
    let (upgrade, _) = tokio::sync::watch::channel(false);
    let mut loops = tokio::task::JoinSet::new();
    let contacts: Arc<dyn ContactRecorder> = Arc::new(FileContactRecorder {
        paths: context.paths().clone(),
        clock: Arc::clone(&clock),
        write: Mutex::new(None),
    });
    // Captured before the targets are moved into their loops: this is the set a
    // restart is measured against.
    let served: BTreeSet<String> = managed_targets
        .iter()
        .filter_map(|target| target.policies.first())
        .map(|policy| policy.target.to_string())
        .collect();
    for target in managed_targets {
        loops.spawn(run_target_loop(
            target,
            shutdown.subscribe(),
            upgrade.subscribe(),
            Arc::clone(&contacts),
        ));
    }

    // The *source*, not `current_exe`. A service registered by `service
    // install` runs a copy this product owns, precisely so that the package
    // manager's own file stays replaceable while the service runs -- so the
    // file that changes on an upgrade is the source, and the copy never
    // changes on its own. `None` for a daemon started by hand, or for a
    // registration made before copies existed, and then nothing is watched.
    let mut upgraded_to = None;
    let mut restart_reason: Option<&'static str> = None;
    let heartbeat_store = Arc::clone(&store) as Arc<dyn Store>;
    let heartbeat_paths = context.paths().clone();
    let watchdog_paths = context.paths().clone();
    let early = tokio::select! {
        signal = wait_for_shutdown(service_shutdown.clone()) => {
            signal.map_err(signal_failure)?;
            None
        }
        version = async {
            match own_binary.clone() {
                Some(path) => wait_for_upgrade(path).await,
                // Nothing to watch, so never resolve; the other arms decide.
                None => std::future::pending().await,
            }
        } => {
            writeln!(
                out,
                "a newer runner-manager ({version}) was installed; finishing every running                  job before handing over"
            )
            .map_err(failed)?;
            tracing::info!(
                version = %version,
                "a newer binary was installed; draining before restart"
            );
            upgraded_to = Some(version);
            None
        }
        () = wait_for_policy_set_change(Arc::clone(&store) as Arc<dyn Store>, served) => {
            writeln!(
                out,
                "the set of repositories this host serves changed; finishing every running job                  before reloading"
            )
            .map_err(failed)?;
            tracing::info!("the policy set changed; draining before restart");
            restart_reason = Some("the set of repositories this host serves changed");
            None
        }
        () = maintain_credential(Arc::clone(&client)) => {
            unreachable!("credential maintenance runs until the daemon is stopped")
        }
        () = maintain_host_fitness(context, Arc::clone(&host_unfit)) => {
            unreachable!("host fitness checks run until the daemon is stopped")
        }
        () = maintain_wsl_guest_heartbeat(heartbeat_paths, heartbeat_store) => {
            unreachable!("WSL heartbeat maintenance runs until the daemon is stopped")
        }
        () = maintain_dependency_caches(
            dependency_caches.clone(),
            context.paths().state_dir().to_path_buf(),
            true,
        ) => {
            unreachable!("dependency cache pruning runs until the daemon is stopped")
        }
        () = answer_github_requests(context, mode) => {
            unreachable!("GitHub requests are answered until the daemon is stopped")
        }
        () = maintain_diagnostics(context.paths().logs_dir().to_path_buf()) => {
            unreachable!("diagnostics maintenance runs until the daemon is stopped")
        }
        () = maintain_docker_cleanup(
            runner_manager_agent::docker_cleanup::DockerCleanup::new(host.id, context.paths().config_dir(), context.paths().state_dir()),
            Arc::clone(&store) as Arc<dyn Store>,
        ) => {
            unreachable!("Docker cleanup maintenance runs until the daemon is stopped")
        }
        () = maintain_wsl_recovery(watchdog_paths, wsl_inventory, windows_service_host) => {
            unreachable!("WSL recovery watchdog runs until the daemon is stopped")
        }
        result = loops.join_next() => result,
    };
    if upgraded_to.is_some() || restart_reason.is_some() {
        let _ = upgrade.send(true);
        // Not `shutdown`: that one is bounded, and an upgrade must outlast any
        // job rather than any deadline.
        while let Some(result) = loops.join_next().await {
            result.map_err(|source| {
                CliError::new(
                    Failure::LocalState,
                    format!("a daemon target loop failed: {source}"),
                )
            })??;
        }
        // Only an upgrade replaces the binary. A policy-set reload restarts the
        // same one, and putting a file in place for it would be a write nobody
        // asked for.
        if let Some(version) = upgraded_to {
            return stop_for_upgrade(context, mode, own_binary.as_deref(), &version, out)
                .map(|()| DaemonOutcome::Stopped);
        }
        let reason = restart_reason.unwrap_or("this daemon was asked to reload");
        writeln!(out, "every runner finished; reloading").map_err(failed)?;
        tracing::info!(%reason, "reloading the daemon in-process after the drain");
        // A policy-set change is configuration reload, not a binary upgrade.
        // Exiting with UpgradePending made a login task depend on Task
        // Scheduler's bounded failure retries and could leave it stopped
        // forever. Release the host lock and rebuild the target loops in this
        // process instead. The outer loop drops this generation's host lock,
        // store and clients before rebuilding them, so repeated edits consume
        // constant stack and memory. The SCM receiver is cloned so every
        // generation remains stoppable.
        drop(instance);
        return Ok(DaemonOutcome::Reload);
    }
    let _ = shutdown.send(true);
    if let Some(result) = early {
        let outcome = result.map_err(|source| {
            CliError::new(
                Failure::LocalState,
                format!("a daemon target loop failed: {source}"),
            )
        })?;
        outcome?;
        return Err(CliError::new(
            Failure::LocalState,
            "a daemon target loop stopped before shutdown",
        ));
    }
    while let Some(result) = loops.join_next().await {
        result.map_err(|source| {
            CliError::new(
                Failure::LocalState,
                format!("a daemon target loop failed: {source}"),
            )
        })??;
    }
    writeln!(out, "daemon stopped; no busy runner was terminated").map_err(failed)?;
    Ok(DaemonOutcome::Stopped)
}

async fn maintain_wsl_recovery_without_local_policies(
    context: &Context,
    mode: StartMode,
    windows_service_host: bool,
) {
    if windows_service_host {
        super::wsl_watchdog::maintain_lifecycle_tasks(context.paths().clone()).await;
        return;
    }
    if !wsl_recovery_is_available(windows_service_host) {
        std::future::pending::<()>().await;
        return;
    }
    loop {
        let inventory = context
            .secret_store(mode)
            .ok()
            .and_then(|store| store.load().ok().flatten())
            .and_then(|secret| {
                AuthenticatedClient::new(
                    context.endpoints().clone(),
                    UserAccessToken::from_stored(secret),
                    context.clock(),
                )
                .ok()
            })
            .map(|client| {
                Arc::new(RestInventory::new(Arc::new(client), context.clock()))
                    as Arc<dyn InventoryGateway>
            });
        if let Some(inventory) = inventory {
            super::wsl_watchdog::maintain(context.paths().clone(), inventory).await;
        }
        tokio::time::sleep(Duration::from_secs(60)).await;
    }
}

async fn maintain_wsl_recovery(
    paths: runner_manager_platform::paths::AppPaths,
    inventory: Arc<dyn InventoryGateway>,
    windows_service_host: bool,
) {
    if windows_service_host {
        super::wsl_watchdog::maintain_lifecycle_tasks(paths).await;
        return;
    }
    if !wsl_recovery_is_available(windows_service_host) {
        std::future::pending::<()>().await;
        return;
    }
    super::wsl_watchdog::maintain(paths, inventory).await;
}

/// WSL distributions belong to an interactive Windows account. An SCM service
/// runs as LocalSystem and cannot truthfully probe those distributions: every
/// attempt looks like a guest failure and used to overwrite a healthy user's
/// recovery state with `degraded` every ten seconds. Login tasks and manually
/// started daemons retain the watchdog because they run in the owning account.
const fn wsl_recovery_is_available(windows_service_host: bool) -> bool {
    cfg!(windows) && !windows_service_host
}

const WSL_GUEST_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10);

/// Reconcile owned Docker resources off the executor in active and idle modes.
async fn maintain_docker_cleanup(
    cleanup: runner_manager_agent::docker_cleanup::DockerCleanup,
    store: Arc<dyn Store>,
) {
    if !runner_manager_agent::docker_cleanup::DockerCleanup::supported() {
        std::future::pending::<()>().await;
    }
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    loop {
        interval.tick().await;
        let cleanup = cleanup.clone();
        let store = Arc::clone(&store);
        let result = tokio::task::spawn_blocking(move || {
            cleanup.sweep_with_lookup(
                &|id| store.attempt(id).map_err(std::io::Error::other),
                false,
            )
        })
        .await;
        match result {
            Ok(Ok(report))
                if report.candidates > 0 || report.pending > 0 || report.protected > 0 =>
            {
                tracing::info!(
                    event = "docker_cleanup",
                    count = report.removed,
                    reason = "attempt_scoped",
                    "candidates={} pending={} protected={}",
                    report.candidates,
                    report.pending,
                    report.protected
                );
            }
            Ok(Ok(_)) => {}
            _ => tracing::warn!(
                event = "docker_cleanup_failed",
                "Docker cleanup deferred; no unverified resources were selected"
            ),
        }
    }
}

/// Bound completed-attempt diagnostics even while every policy is disabled.
/// The filesystem walk runs off the async executor, once at start and hourly.
async fn maintain_diagnostics(directory: std::path::PathBuf) {
    let mut interval = tokio::time::interval(Duration::from_secs(60 * 60));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    loop {
        interval.tick().await;
        let directory = directory.clone();
        match tokio::task::spawn_blocking(move || {
            runner_manager_platform::log_retention::prune_attempt_logs(
                &directory,
                std::time::SystemTime::now(),
            )
        })
        .await
        {
            Ok(Ok(report)) if report.failed == 0 => {
                if report.removed > 0 {
                    tracing::info!(event = "diagnostics_pruned", count = report.removed);
                }
            }
            _ => tracing::warn!(
                event = "diagnostics_prune_failed",
                "some old attempt diagnostics could not be removed; retrying next hour"
            ),
        }
    }
}

/// Measures the dependency caches and prunes them to the cap, at start and
/// then every [`PRUNE_INTERVAL`](runner_manager_platform::dependency_cache::PRUNE_INTERVAL)
/// when `scheduled`, and whenever `host cache prune` asks through a
/// [`PRUNE_REQUEST_FILE`](runner_manager_platform::dependency_cache::PRUNE_REQUEST_FILE)
/// in `state_dir` -- which is how an operator whose account cannot write the
/// service's cache root still gets a prune. The walk runs on a blocking
/// thread: a pnpm store holds hundreds of thousands of files.
///
/// Requests are taken while a pass is running too -- a walk of a full cache
/// can outlast the time the asking command waits to hear that somebody is
/// listening -- and carried out once that pass ends, one pass at a time.
async fn maintain_dependency_caches(
    caches: runner_manager_agent::dependency_caches::DependencyCaches,
    state_dir: std::path::PathBuf,
    scheduled: bool,
) {
    use runner_manager_platform::dependency_cache::{
        PRUNE_INTERVAL, PRUNE_REQUEST_POLL, prune_request_path, take_prune_request,
    };
    let request_file = prune_request_path(&state_dir);
    let mut next_scheduled = scheduled.then(std::time::Instant::now);
    let mut pending: Option<String> = None;
    let mut running: Option<tokio::task::JoinHandle<()>> = None;
    loop {
        // Almost every look finds nothing, and a look is one `stat`.
        if pending.is_none() && request_file.exists() {
            pending = take_prune_request(&state_dir);
        }
        if running
            .as_ref()
            .is_some_and(tokio::task::JoinHandle::is_finished)
        {
            running = None;
        }
        if running.is_none() {
            let pass = caches.clone();
            if let Some(request) = pending.take() {
                let state = state_dir.clone();
                running = Some(tokio::task::spawn_blocking(move || {
                    pass.answer_prune_request(&state, request);
                }));
            } else if next_scheduled.is_some_and(|at| std::time::Instant::now() >= at) {
                running = Some(tokio::task::spawn_blocking(move || pass.prune_scheduled()));
                next_scheduled = Some(std::time::Instant::now() + PRUNE_INTERVAL);
            }
        }
        tokio::time::sleep(PRUNE_REQUEST_POLL).await;
    }
}

/// Answers what the stored credential reaches, for a command that cannot read
/// it: `repo add`, `org add` or `auth status` in an SSH session on a login-mode
/// Mac, whose login keychain that session cannot unlock. This process reads it
/// in the session it runs in. See
/// [`GITHUB_DISCOVERY`](runner_manager_platform::service_request::GITHUB_DISCOVERY).
async fn answer_github_requests(context: &Context, mode: StartMode) {
    use runner_manager_platform::service_request::{GITHUB_DISCOVERY, POLL};
    let state_dir = context.paths().state_dir().to_path_buf();
    // One discovery answers every request that arrives within this long, so
    // whatever can write the state directory cannot spend the credential's
    // GitHub budget faster than this.
    const REUSE: Duration = Duration::from_secs(30);
    let mut last: Option<(
        std::time::Instant,
        Result<super::auth::CredentialState, String>,
    )> = None;
    loop {
        let requests = GITHUB_DISCOVERY.take(&state_dir);
        if !requests.is_empty() {
            let outcome = match &last {
                Some((at, outcome)) if at.elapsed() < REUSE => outcome.clone(),
                _ => {
                    let outcome = super::auth::credential_state_for_service(context, mode).await;
                    last = Some((std::time::Instant::now(), outcome.clone()));
                    outcome
                }
            };
            for request in requests {
                if let Err(error) = GITHUB_DISCOVERY.answer(&state_dir, &request, outcome.clone()) {
                    tracing::warn!(
                        %error,
                        reason = "github_request_unanswered",
                        "could not answer a command waiting for what the credential reaches"
                    );
                }
            }
        }
        tokio::time::sleep(POLL).await;
    }
}

/// Publish only local, non-secret evidence. Windows owns the destructive
/// decision; the guest owns the authoritative attempt journal and the audit
/// for persistent runner services Windows cannot safely infer.
async fn maintain_wsl_guest_heartbeat(
    paths: runner_manager_platform::paths::AppPaths,
    store: Arc<dyn Store>,
) {
    loop {
        match GuestRecoveryConfig::read(&paths) {
            Ok(Some(config)) => {
                let request = DrainRequest::read(&config.shared_root);
                let attempts = store.attempts();
                let policies = store.policies();
                let heartbeat = GuestHeartbeat {
                    schema_version: WSL_RECOVERY_SCHEMA_VERSION,
                    observed_at: chrono::Utc::now(),
                    acknowledged_generation: request
                        .as_ref()
                        .ok()
                        .and_then(|request| request.as_ref().map(|request| request.generation)),
                    local_active_attempts: attempts
                        .as_ref()
                        .ok()
                        .map(|attempts| u32::from(active_count(attempts))),
                    local_busy_attempts: attempts.as_ref().ok().and_then(|attempts| {
                        u32::try_from(
                            attempts
                                .iter()
                                .filter(|attempt| attempt.state() == AttemptState::Busy)
                                .count(),
                        )
                        .ok()
                    }),
                    managed_targets: policies
                        .map(|policies| {
                            policies
                                .into_iter()
                                .map(|policy| policy.target)
                                .collect::<BTreeSet<_>>()
                                .into_iter()
                                .collect()
                        })
                        .unwrap_or_default(),
                    unmanaged_runner_services: unmanaged_runner_service_count(),
                };
                if let Err(error) = request {
                    tracing::warn!(
                        reason = "wsl_recovery_state_unreadable",
                        %error,
                        "WSL recovery drain request is unavailable"
                    );
                }
                if let Err(error) = heartbeat.write(&config.shared_root) {
                    tracing::warn!(
                        reason = "wsl_recovery_state_unwritable",
                        %error,
                        "WSL recovery heartbeat cannot be published"
                    );
                }
            }
            Ok(None) => {}
            Err(_) => tracing::warn!(
                reason = "wsl_recovery_config_unreadable",
                "WSL recovery configuration is unavailable"
            ),
        }
        tokio::time::sleep(WSL_GUEST_HEARTBEAT_INTERVAL).await;
    }
}

/// How often an otherwise idle daemon checks the credential timestamps.
///
/// A short interval is only a local mutex read until the renewal window opens;
/// it does not poll GitHub.
const CREDENTIAL_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(15 * 60);
/// Renew ahead of the eight-hour boundary so a transient network failure gets
/// more than one attempt before the access half stops working.
const CREDENTIAL_RENEWAL_WINDOW: Duration = Duration::from_secs(30 * 60);

/// Evaluates the required host checks and updates `unfit`, logging only when
/// the verdict changes so a host that stays unfit does not fill the log. The
/// first evaluation of a generation also names the missing recommended
/// preparation, once.
async fn observe_host_fitness(context: &Context, unfit: &AtomicBool, first: bool) {
    let verdict = bounded_preflight(
        super::doctor::daemon_preflight(context),
        PREFLIGHT_DEADLINE,
        runner_manager_platform::host_fitness::probes_in_flight,
        context,
    )
    .await;
    match verdict {
        Ok(verdict) => {
            let now_unfit = !verdict.required.is_empty();
            let was_unfit = unfit.swap(now_unfit, Ordering::AcqRel);
            if now_unfit && (first || !was_unfit) {
                tracing::warn!(
                    event = "host_unfit",
                    reason = %verdict.required.join(","),
                    count = verdict.required.len(),
                    "required host checks fail; no runner is started until `runner-manager host doctor` passes"
                );
            } else if was_unfit && !now_unfit {
                tracing::warn!(
                    event = "host_fit",
                    "required host checks pass again; runners may start"
                );
            }
            if first && !verdict.recommended.is_empty() {
                tracing::warn!(
                    event = "host_doctor",
                    reason = %verdict.recommended.join(","),
                    count = verdict.recommended.len(),
                    "recommended host preparation is missing; run `runner-manager host doctor`"
                );
            }
        }
        Err(error) => {
            // A check that could not run is not evidence the host is unfit.
            unfit.store(false, Ordering::Release);
            tracing::warn!(
                event = "host_doctor_unavailable",
                "host checks could not run, so no runner is held back: {error}"
            );
        }
    }
}

/// How long the host checks may take before the daemon treats them as blocked.
/// Each runner-root probe gives up after five seconds on its own; this bounds
/// everything else a check might wait on.
const PREFLIGHT_DEADLINE: Duration = Duration::from_secs(120);

use super::doctor::BLOCKED_CHECKS_ID;

/// Runs `preflight` for at most `deadline`, and says what checks that never
/// finish mean.
///
/// While a directory probe is still blocked (`stuck` names the paths), a
/// runner placed on this host would block there too -- a hung volume, or a
/// runner root behind a macOS question nobody has answered -- so the host
/// counts as unfit until it answers. A check that is merely slow elsewhere
/// holds no runner back, as a check that could not run never does.
async fn bounded_preflight(
    preflight: impl Future<Output = Result<super::doctor::DaemonVerdict, String>>,
    deadline: Duration,
    stuck: impl FnOnce() -> Vec<std::path::PathBuf>,
    context: &Context,
) -> Result<super::doctor::DaemonVerdict, String> {
    match tokio::time::timeout(deadline, preflight).await {
        Ok(Err(error)) if error == super::doctor::PREFLIGHT_STILL_RUNNING => {}
        Ok(verdict) => return verdict,
        Err(_) => {}
    }
    let stuck = stuck();
    if stuck.is_empty() {
        return Err(format!(
            "the host checks did not finish within {} seconds",
            deadline.as_secs()
        ));
    }
    let stuck = stuck
        .iter()
        .map(|path| path.display().to_string())
        .collect::<Vec<_>>()
        .join(", ");
    let finding = runner_manager_platform::host_fitness::UnfitFinding {
        id: BLOCKED_CHECKS_ID.to_owned(),
        detail: format!(
            "the host checks did not finish within {} seconds: {stuck} does not answer",
            deadline.as_secs()
        ),
        remedy: Some(
            if cfg!(target_os = "macos") {
                "look for a dialog on this Mac's desktop asking whether runner-manager may \
                 access files on a volume, and click Allow; otherwise check the volume holding \
                 the runner root"
            } else {
                "check the volume holding the runner root (a stalled external disk or network \
                 mount); `runner-manager host doctor` names it"
            }
            .to_owned(),
        ),
    };
    if let Err(error) = runner_manager_platform::host_fitness::record_host_unfit(
        context.paths(),
        std::slice::from_ref(&finding),
        context.clock().now(),
    ) {
        tracing::warn!(
            event = "host_unfit_unrecorded",
            "the host-fitness record could not be written: {error}"
        );
    }
    Ok(super::doctor::DaemonVerdict {
        required: vec![finding.id],
        recommended: Vec::new(),
    })
}

/// This generation's heartbeat task. It runs on the daemon's one runtime
/// thread, so it says the loop is not blocked, not that every task advances.
/// Dropped when the generation returns,
/// which stops it and removes the record, so a daemon that exits cleanly
/// never reads as stalled. See [`runner_manager_platform::daemon_heartbeat`].
struct Heartbeat {
    task: tokio::task::JoinHandle<()>,
    paths: runner_manager_platform::paths::AppPaths,
}

impl Heartbeat {
    fn start(paths: runner_manager_platform::paths::AppPaths) -> Self {
        let beating = paths.clone();
        let task = tokio::spawn(async move {
            use runner_manager_platform::daemon_heartbeat::{HEARTBEAT_INTERVAL, beat};
            let mut failing = false;
            loop {
                match beat(&beating, chrono::Utc::now()) {
                    // Said once: a record that stops being written reads as a
                    // stalled daemon, and this is the line that says why.
                    Err(error) if !failing => {
                        failing = true;
                        tracing::warn!(
                            %error,
                            reason = "heartbeat_unwritten",
                            "the daemon heartbeat could not be written, so `status` will call \
                             this daemon stalled"
                        );
                    }
                    Err(_) => {}
                    Ok(()) => failing = false,
                }
                tokio::time::sleep(HEARTBEAT_INTERVAL).await;
            }
        });
        Self { task, paths }
    }
}

impl Drop for Heartbeat {
    fn drop(&mut self) {
        self.task.abort();
        let _ = runner_manager_platform::daemon_heartbeat::stop(&self.paths);
    }
}

/// Re-evaluates the required host checks every few minutes, so a fix made with
/// `host prepare` is picked up without restarting the daemon.
async fn maintain_host_fitness(context: &Context, unfit: Arc<AtomicBool>) {
    loop {
        tokio::time::sleep(super::doctor::DAEMON_RECHECK).await;
        observe_host_fitness(context, &unfit, false).await;
    }
}

/// Keep the daemon's credential alive even when no runner request uses it.
async fn maintain_credential(client: Arc<AuthenticatedClient>) {
    loop {
        client
            .renew_if_expiring_within(CREDENTIAL_RENEWAL_WINDOW)
            .await;
        tokio::time::sleep(CREDENTIAL_MAINTENANCE_INTERVAL).await;
    }
}

/// The idle variant also notices a credential stored after the service starts.
/// This matters when installation precedes `auth login` and no policy change is
/// available to make the service manager restart the daemon.
async fn maintain_idle_credential(context: &Context, mode: StartMode) {
    let mut client = None;
    loop {
        if client.is_none() {
            client = idle_credential_client(context, mode);
        }
        if let Some(client) = &client {
            client
                .renew_if_expiring_within(CREDENTIAL_RENEWAL_WINDOW)
                .await;
        }
        tokio::time::sleep(CREDENTIAL_MAINTENANCE_INTERVAL).await;
    }
}

/// Prepare credential maintenance for a host that currently owns no policies.
///
/// Idle mode historically required neither a sign-in nor even a readable
/// secret store, and installing a service before onboarding is valid. Preserve
/// that property: maintenance is best-effort until a policy-set change restarts
/// the daemon through the strict, authenticated path above.
fn idle_credential_client(context: &Context, mode: StartMode) -> Option<Arc<AuthenticatedClient>> {
    let secrets = match context.secret_store(mode) {
        Ok(secrets) => secrets,
        Err(error) => {
            tracing::debug!(%error, "idle credential maintenance could not reach the store");
            return None;
        }
    };
    let loaded = secrets.load();
    // Only ever cleared here, never recorded: idle mode does not need a
    // readable store, but a record left by an earlier, policy-serving start
    // must not outlive the sign-in that ended it.
    if loaded.is_ok() {
        note_credential_readability(context, true);
    }
    let secret = match loaded {
        Ok(Some(secret)) => secret,
        Ok(None) => return None,
        Err(error) => {
            tracing::debug!(%error, "idle credential maintenance could not read the store");
            return None;
        }
    };
    let app = match context.app_registration() {
        Ok(app) => app,
        Err(error) => {
            tracing::debug!(%error, "idle credential maintenance could not load the App registration");
            return None;
        }
    };
    let flow = match DeviceFlow::new(app, context.endpoints().clone()) {
        Ok(flow) => flow,
        Err(error) => {
            tracing::debug!(%error, "idle credential maintenance could not prepare renewal");
            return None;
        }
    };
    let renewal: Arc<dyn CredentialRenewal> = Arc::new(super::auth::StoringRenewal::new(
        flow,
        Arc::clone(&secrets),
        context
            .paths()
            .state_dir()
            .join(LockKind::CredentialRenewal.file_name()),
    ));
    let client = match AuthenticatedClient::new(
        context.endpoints().clone(),
        UserAccessToken::from_stored(secret),
        context.clock(),
    ) {
        Ok(client) => client,
        Err(error) => {
            tracing::debug!(%error, "idle credential maintenance could not prepare the GitHub client");
            return None;
        }
    };
    Some(Arc::new(
        client
            .with_renewal(renewal)
            .with_credential_source(Arc::new(super::auth::StoredCredential::new(secrets))),
    ))
}

fn stop_for_upgrade(
    context: &Context,
    mode: StartMode,
    source: Option<&std::path::Path>,
    version: &str,
    out: &mut dyn Write,
) -> Result<(), CliError> {
    match source.map(replace_own_binary) {
        Some(Ok(())) if hands_over_credential(context, mode) => {
            hand_over_credential_to_new_binary(context, mode);
        }
        Some(Err(error)) => {
            tracing::warn!(
                %error,
                "the new binary could not be put in place; the service manager will restart the version already there"
            );
            writeln!(out, "warning: {error}").map_err(write_failed("the daemon state"))?;
        }
        _ => {}
    }
    writeln!(
        out,
        "every runner finished; stopping so {version} can take over"
    )
    .map_err(write_failed("the daemon state"))?;
    Err(CliError::with_remedy(
        Failure::UpgradePending,
        format!(
            "a newer runner-manager ({version}) is installed and every runner this daemon held has finished; stopping so the service manager starts the new one"
        ),
        "runner-manager service status",
    ))
}

/// Says on disk whether this daemon could read its stored credential, for
/// `service status` to report. See `record_credential_unreadable`.
fn note_credential_readability(context: &Context, readable: bool) {
    let marked = if readable {
        clear_credential_unreadable(context.paths())
    } else {
        record_credential_unreadable(context.paths(), context.clock().now())
    };
    if let Err(error) = marked {
        tracing::warn!(%error, "cannot record whether the stored credential is readable");
    }
}

// ---------------------------------------------------------------------------
// Handing the credential to the new binary
// ---------------------------------------------------------------------------

/// Whether an upgrade hands the stored credential to the new binary.
///
/// Only the standard macOS **login** keychain binds an item to the build that
/// wrote it. The System Keychain (boot mode) and a keychain under `--data-dir`
/// grant their item to every application, and DPAPI and a `0600` file are
/// readable by any program the account runs, so a rewrite anywhere else would
/// change nothing -- except to delete and re-add a credential that works.
fn hands_over_credential(context: &Context, mode: StartMode) -> bool {
    cfg!(target_os = "macos") && mode == StartMode::Login && context.data_root.is_none()
}

/// How long the handover waits for a renewal in another process to finish.
const HANDOVER_LOCK_WAIT: Duration = Duration::from_secs(30);

/// How long the handover waits for the new binary to store the credential.
///
/// Every runner has already drained by then, so a child that never finishes
/// would hold the host with no runner and no restart; past this it is stopped
/// and the restart happens anyway.
const HANDOVER_CHILD_DEADLINE: Duration = Duration::from_secs(60);

/// What the handover did.
#[derive(Debug, PartialEq, Eq)]
enum Handover {
    /// There was no credential to hand over.
    NothingStored,
    /// This daemon could not read its own credential, so it had nothing to
    /// hand over either. The new daemon will report the same thing.
    Unreadable,
    /// The new binary stored the credential, so it is now the writer.
    HandedOver,
    /// The new binary did not store it. `restored` says whether this daemon
    /// had to put the credential back because the failed attempt had already
    /// removed it.
    Failed { restored: bool },
}

/// Has the binary that just replaced this one store this daemon's credential,
/// so that the daemon it becomes after the restart can read it.
///
/// # Why the new binary has to be the writer
///
/// The login keychain records, with every item, the *partition* of the code
/// that created it, and for an ad-hoc signed binary that partition is
/// `cdhash:<hash of this exact build>`. A caller from another partition is
/// refused with `errSecAuthFailed` (`-25293`) without a prompt, however wide
/// the item's application list is -- measured on macOS 27 with two ad-hoc
/// builds of one program: an item granted to *any application*, and one whose
/// list named the second build explicitly, both refused it. Every release is a
/// new build, so every upgrade locked the new daemon out of the item its
/// predecessor wrote, and somebody had to sign in again at the Mac.
///
/// The one program that can still read the item at this moment is this one,
/// and the one that has to own it next is the binary now at
/// [`std::env::current_exe`]. So this reads the credential and pipes it into
/// `daemon adopt-credential` run from the new binary, whose ordinary store
/// replaces the item with one partitioned to the new build. The command-line
/// copy the package manager installed is byte-identical to the service's copy,
/// so `runner-manager auth status` reads it too.
///
/// Best effort: every failure is logged and leaves the restart to happen
/// anyway, which is the state the operator was in before this existed.
fn hand_over_credential_to_new_binary(context: &Context, mode: StartMode) {
    let binary = match std::env::current_exe() {
        Ok(binary) => binary,
        Err(error) => {
            tracing::warn!(%error, "the credential cannot be handed over: this binary's own path is unknown");
            return;
        }
    };
    let secrets = match context.secret_store(mode) {
        Ok(secrets) => secrets,
        Err(error) => {
            tracing::warn!(error = %error.message(), "the credential cannot be handed over: the store is unreachable");
            return;
        }
    };
    // Held so that a renewal in another process cannot rewrite the item between
    // the read and the new binary's write. A lock that cannot be had is not a
    // reason to skip the handover: without it the new daemon is locked out for
    // certain.
    let _renewal = HostLock::acquire(
        context.paths(),
        LockKind::CredentialRenewal,
        HANDOVER_LOCK_WAIT,
    )
    .map_err(|error| tracing::warn!(%error, "handing the credential over without the renewal lock"))
    .ok();
    let outcome = hand_over(secrets.as_ref(), |document| {
        adopt_through(&binary, context, mode, document)
    });
    match outcome {
        Handover::HandedOver => tracing::info!("the credential was handed over to the new binary"),
        Handover::NothingStored => tracing::info!("no credential is stored; nothing to hand over"),
        Handover::Unreadable | Handover::Failed { .. } => tracing::warn!(
            ?outcome,
            remedy = %sign_in_instruction(mode),
            "the credential was not handed over; the new daemon may need a sign-in"
        ),
    }
}

/// The handover, with the new binary behind `adopt` so it can be tested
/// without one.
fn hand_over(
    secrets: &dyn SecretStore,
    adopt: impl FnOnce(&SecretString) -> Result<(), String>,
) -> Handover {
    let document = match secrets.load() {
        // Normalised through the credential type, so that a bare token written
        // by an old version reaches the new binary as the envelope it accepts.
        Ok(Some(secret)) => UserAccessToken::from_stored(secret).to_stored_document(),
        Ok(None) => return Handover::NothingStored,
        Err(error) => {
            tracing::warn!(%error, "this daemon cannot read its own credential to hand it over");
            return Handover::Unreadable;
        }
    };
    let Err(reason) = adopt(&document) else {
        return Handover::HandedOver;
    };
    tracing::warn!(%reason, "the new binary did not store the credential");
    // Replacing an item this program may not read starts by deleting it, so a
    // write that failed after that would leave nothing at all. An item still
    // there, readable or not, is left alone: it is either the original or the
    // new binary's.
    let restored = matches!(secrets.load(), Ok(None)) && secrets.store(&document).is_ok();
    Handover::Failed { restored }
}

/// Runs `daemon adopt-credential` from `binary` with `document` on its stdin,
/// against the same directories and store this daemon uses.
///
/// Only reached without `--data-dir` (see [`hands_over_credential`]), so the
/// store is the platform-standard one and the directories are this daemon's.
fn adopt_through(
    binary: &std::path::Path,
    context: &Context,
    mode: StartMode,
    document: &SecretString,
) -> Result<(), String> {
    use std::io::{Read as _, Write as _};
    use std::process::{Command, Stdio};

    let mut child = Command::new(binary)
        .args([
            "daemon",
            "adopt-credential",
            "--start-at",
            &mode.to_string(),
        ])
        .args(super::service::service_directory_arguments(context.paths()))
        .env_remove(super::DATA_DIR_VARIABLE)
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .map_err(|error| format!("{} could not be started: {error}", binary.display()))?;
    // Drained on its own thread so that a chatty child can never block on a
    // full pipe while this waits for it to exit.
    let mut stderr = child
        .stderr
        .take()
        .expect("stderr was configured as a pipe");
    let stderr = std::thread::spawn(move || {
        let mut text = Vec::new();
        let _ = stderr.read_to_end(&mut text);
        text
    });
    // Dropped at the end of the statement, which closes the pipe before the wait.
    let written = child
        .stdin
        .take()
        .expect("stdin was configured as a pipe")
        .write_all(document.expose_secret().as_bytes())
        .map_err(|error| format!("the credential could not be written to it: {error}"));
    // Waited for even when the write failed, so the child is never left behind,
    // and never for longer than the deadline: every runner has drained already.
    let deadline = std::time::Instant::now() + HANDOVER_CHILD_DEADLINE;
    let status = loop {
        match child.try_wait() {
            Ok(Some(status)) => break status,
            Ok(None) if std::time::Instant::now() < deadline => {
                std::thread::sleep(Duration::from_millis(50));
            }
            Ok(None) => {
                let _ = child.kill();
                let _ = child.wait();
                return Err(format!(
                    "it did not finish within {} seconds and was stopped",
                    HANDOVER_CHILD_DEADLINE.as_secs()
                ));
            }
            Err(error) => {
                let _ = child.kill();
                return Err(format!("the new binary could not be waited for: {error}"));
            }
        }
    };
    written?;
    if status.success() {
        return Ok(());
    }
    let stderr = stderr.join().unwrap_or_default();
    Err(format!(
        "it exited with {status}: {}",
        String::from_utf8_lossy(&stderr).trim()
    ))
}

// ---------------------------------------------------------------------------
// A launchd plist written by an older build
// ---------------------------------------------------------------------------

/// How long the daemon waits for launchd to stop it after asking for a reload.
/// The helper boots the job out about a second after it starts.
#[cfg(target_os = "macos")]
const LAUNCHD_RELOAD_GRACE: Duration = Duration::from_secs(90);

/// Puts the current plist in place of one an older build wrote with the wrong
/// `ProcessType`, and has launchd load the job again. Returns `true` when this
/// daemon has stopped for that.
///
/// # Why the daemon does it
///
/// `runner-manager update` replaces the binary and nothing else: the old daemon
/// drains, swaps its private copy and exits, and launchd restarts the new
/// binary from the job it already holds, old `ProcessType` included. So the
/// first daemon of a fixed version is the earliest code that can repair the
/// plist, and straight after an upgrade drain it holds no runner, which is the
/// one moment a reload interrupts nothing. With a runner held it defers to a
/// later generation rather than stop under a job.
///
/// Only the launchd job itself does this: `XPC_SERVICE_NAME` is the job's
/// label, which launchd sets, and the parent is launchd. A runner the daemon
/// started inherits the variable but not the parent, so a `daemon run` inside a
/// job never reloads the service. Every failure leaves the loaded job as it
/// was and is logged; none stops the daemon.
#[cfg(target_os = "macos")]
async fn converge_launchd_definition(
    context: &Context,
    out: &mut dyn Write,
    service_shutdown: Option<runner_manager_platform::service::ServiceShutdown>,
) -> Result<bool, CliError> {
    let operations = super::service::operations(context);
    let label = super::service::identity().launchd_label();
    let launchd_job = std::env::var("XPC_SERVICE_NAME").is_ok_and(|name| name == label)
        && std::os::unix::process::parent_id() == 1;
    if !launchd_job {
        return Ok(false);
    }
    let drift = match operations.definition_drift() {
        Ok(drift) => drift.filter(|drift| {
            drift.manager == runner_manager_platform::service::DefinitionKind::LaunchdPlist
        }),
        Err(error) => {
            tracing::warn!(%error, "cannot compare the installed launchd plist");
            return Ok(false);
        }
    };
    let Some(drift) = drift else {
        // The file can be current while the loaded job is not: a rewrite whose
        // reload failed, or a hand edit nobody reloaded. Reported, never acted
        // on: a reload that did not help would repeat on every start.
        // `kickstart` would not help: it restarts from the job launchd holds.
        if runner_manager_platform::process::runs_at_background_priority()
            && let Ok(Some(record)) = InstallRecord::read(context.paths())
            && let Some(plist) = record.definition_path
        {
            let (sudo, domain) = match record.start_mode {
                StartMode::Boot => ("sudo ", "system".to_string()),
                StartMode::Login => ("", "gui/$(id -u)".to_string()),
            };
            tracing::warn!(
                reason = "launchd_job_outdated",
                "launchd runs this daemon at background priority although its plist is current, \
                 so the plist was never reloaded and every runner is throttled; reload it while no \
                 job is running: {sudo}launchctl bootout {domain}/{label} && {sudo}launchctl \
                 bootstrap {domain} '{}'",
                plist.display()
            );
        }
        return Ok(false);
    };
    let active = match context
        .store()
        .and_then(|store| store.attempts().map_err(local_store_failure))
    {
        Ok(attempts) => active_count(&attempts),
        Err(error) => {
            tracing::warn!(
                %error,
                "cannot count this host's runners, so the outdated launchd plist is left for now"
            );
            return Ok(false);
        }
    };
    if active > 0 {
        tracing::warn!(
            path = %drift.path.display(),
            active_runners = active,
            reason = "launchd_plist_outdated",
            "the launchd plist throttles every runner; it is rewritten when the daemon next starts with no runner, or restarts with none after a policy change"
        );
        return Ok(false);
    }
    if let Err(error) = operations.reload_launchd_definition(&drift) {
        tracing::warn!(%error, "the outdated launchd plist could not be replaced");
        return Ok(false);
    }
    tracing::info!(
        path = %drift.path.display(),
        process_type = runner_manager_platform::service::LAUNCHD_PROCESS_TYPE,
        "rewrote the launchd plist written by {}; launchd is loading the job again",
        drift.installed_by_version
    );
    tokio::select! {
        signal = wait_for_shutdown(service_shutdown) => {
            signal.map_err(signal_failure)?;
            writeln!(out, "daemon stopped so launchd can load the rewritten plist; no runner was held")
                .map_err(write_failed("the daemon state"))?;
            Ok(true)
        }
        () = tokio::time::sleep(LAUNCHD_RELOAD_GRACE) => {
            tracing::warn!(
                "launchd did not reload the job after the plist was rewritten; continuing under the old ProcessType"
            );
            Ok(false)
        }
    }
}

// ---------------------------------------------------------------------------
// Upgrade detection
// ---------------------------------------------------------------------------

/// How often the daemon looks at its own binary.
///
/// Deliberately unrelated to the poll interval: this is two `stat` calls
/// against a local path and costs nothing GitHub can see, so it is not on the
/// REST budget and does not need to be.
const UPGRADE_CHECK_INTERVAL: Duration = Duration::from_secs(30);

/// What a file looked like, in the two fields that change when it is replaced.
///
/// Not a hash. Reading 13 MB every thirty seconds to notice a change that a
/// `stat` already reports would be paying a lot for a stronger answer than the
/// question needs — and the answer is confirmed by executing the binary anyway
/// (see [`upgraded_version`]), which is a stronger check than any digest.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct BinaryStamp {
    len: u64,
    modified: Option<std::time::SystemTime>,
}

impl BinaryStamp {
    fn of(path: &std::path::Path) -> Option<Self> {
        let meta = std::fs::metadata(path).ok()?;
        Some(Self {
            len: meta.len(),
            modified: meta.modified().ok(),
        })
    }
}

/// The version a replaced binary reports, when it is genuinely a different one.
///
/// # Why the file is executed rather than trusted
///
/// A changed `stat` says the bytes moved, not that they are complete. A package
/// manager writing 13 MB is momentarily a file of the right name and the wrong
/// length, and restarting into that leaves the machine in a restart loop with
/// no daemon — the failure this whole feature exists to avoid, caused by the
/// feature itself.
///
/// Running `--version` settles both halves at once: it exits non-zero or not at
/// all if the file is partial, and it prints the version if it is not. That is
/// also the only way to learn the *new* version, since this process can only
/// ever report the one it was compiled as.
///
/// `None` means "nothing to do": unreadable, unrunnable, or the same version
/// this daemon already is. A same-version rewrite — a reinstall of what is
/// already there — is deliberately not an upgrade, because restarting for it
/// would interrupt runners to change nothing.
fn upgraded_version(path: &std::path::Path) -> Option<String> {
    let output = std::process::Command::new(path)
        .arg("--version")
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    changed_build_version(&String::from_utf8(output.stdout).ok()?)
}

fn changed_build_version(reported: &str) -> Option<String> {
    // `--version` prints `runner-manager X.Y.Z`; the last token is the version.
    let reported = reported.split_whitespace().last()?.to_string();
    (reported != env!("RUNNER_MANAGER_BUILD_VERSION")).then_some(reported)
}

/// Resolves when a different, runnable version has replaced this daemon's own
/// binary. Never resolves otherwise.
async fn wait_for_upgrade(path: std::path::PathBuf) -> String {
    // The stamp this daemon started from. A binary replaced *before* the first
    // look is still caught, because the version comparison below is against
    // what this process was compiled as rather than against the file.
    if let Some(version) = upgraded_version(&path) {
        return version;
    }
    let mut known = BinaryStamp::of(&path);
    loop {
        tokio::time::sleep(UPGRADE_CHECK_INTERVAL).await;
        let current = BinaryStamp::of(&path);
        if current == known {
            continue;
        }
        // Record the new stamp before validating, so a partial write is not
        // re-examined every thirty seconds until it happens to finish.
        known = current;
        if current.is_none() {
            continue;
        }
        if let Some(version) = upgraded_version(&path) {
            return version;
        }
    }
}

/// Puts the source binary in place of the one this process is running.
///
/// The running file is renamed aside rather than deleted: Windows refuses to
/// unlink an executable that is running -- which this one is, right now -- and
/// permits renaming it. The leftover is removed by the next install or upgrade,
/// whichever comes first, and is harmless until then because nothing names it.
fn replace_own_binary(source: &std::path::Path) -> Result<(), String> {
    let own = std::env::current_exe().map_err(|error| format!("own path unknown: {error}"))?;
    let aside = own.with_extension("old");
    let _ = std::fs::remove_file(&aside);
    std::fs::rename(&own, &aside)
        .map_err(|error| format!("the running binary could not be moved aside: {error}"))?;
    if let Err(error) = std::fs::copy(source, &own) {
        // Put back what was working. A daemon that restarts into nothing is a
        // machine with no runner manager at all.
        let _ = std::fs::rename(&aside, &own);
        return Err(format!(
            "the new binary could not be copied into place: {error}"
        ));
    }
    Ok(())
}

/// Resolves when the *set* of targets this daemon should serve has changed.
///
/// # Why this is separate from re-reading a policy
///
/// [`TargetReconciler::refresh_policies`] keeps a running loop honest about its
/// own policy — armed or drained, what capacity, which labels. It cannot create
/// a loop that does not exist, and a loop owns a package cache, a launcher and
/// a startup recovery that ran once. So a repository *added* while the daemon
/// runs has no loop to refresh, and one *removed* leaves a loop with nothing to
/// serve.
///
/// Both are handled the way an operator handles them today, minus the operator:
/// finish what is running and let the service manager start a daemon that reads
/// the whole set afresh. Startup recovery then adopts every runner still up,
/// which is exactly what it exists for.
async fn wait_for_policy_set_change(store: Arc<dyn Store>, initial: BTreeSet<String>) {
    loop {
        tokio::time::sleep(POLICY_SET_CHECK_INTERVAL).await;
        let Ok(policies) = store.policies() else {
            // Unreadable is not changed. Restarting on a transient read error
            // would turn a blip into a drain.
            continue;
        };
        let current: BTreeSet<String> = active_autoscale_targets(policies)
            .iter()
            .map(|group| group[0].target.to_string())
            .collect();
        if current != initial {
            return;
        }
    }
}

/// How often the daemon looks for a repository added or removed under it.
///
/// Slower than a poll and faster than an operator notices: this is one local
/// SQLite read, and it costs GitHub nothing.
const POLICY_SET_CHECK_INTERVAL: Duration = Duration::from_secs(30);

async fn wait_for_shutdown(
    service_shutdown: Option<runner_manager_platform::service::ServiceShutdown>,
) -> std::io::Result<()> {
    match service_shutdown {
        Some(shutdown) => {
            shutdown.wait().await;
            Ok(())
        }
        None => shutdown_signal().await,
    }
}

/// Gives one lifecycle launcher a target-scoped journal only while it performs
/// startup recovery. Ordinary reconciliation sees the complete host attempt
/// set again, which preserves the host-wide capacity invariant.
#[derive(Debug)]
struct TargetRecoveryStore {
    inner: Arc<dyn Store>,
    policies: BTreeSet<runner_manager_domain::model::PolicyId>,
    recovering: AtomicBool,
}

impl TargetRecoveryStore {
    fn new(inner: Arc<dyn Store>, policies: &[ScalePolicy]) -> Self {
        Self {
            inner,
            policies: policies.iter().map(|policy| policy.id).collect(),
            recovering: AtomicBool::new(true),
        }
    }

    fn finish_recovery(&self) {
        self.recovering.store(false, Ordering::Release);
    }
}

impl Store for TargetRecoveryStore {
    fn put_host(
        &self,
        host: &runner_manager_domain::model::Host,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.put_host(host)
    }

    fn host(
        &self,
        id: runner_manager_domain::model::HostId,
    ) -> Result<Option<runner_manager_domain::model::Host>, runner_manager_domain::store::StoreError>
    {
        self.inner.host(id)
    }

    fn hosts(
        &self,
    ) -> Result<Vec<runner_manager_domain::model::Host>, runner_manager_domain::store::StoreError>
    {
        self.inner.hosts()
    }

    fn set_runner_root_override(
        &self,
        id: runner_manager_domain::model::HostId,
        expected: Option<&runner_manager_domain::path::LocalAbsolutePath>,
        new_root: Option<&runner_manager_domain::path::LocalAbsolutePath>,
        expected_uncleaned: u16,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner
            .set_runner_root_override(id, expected, new_root, expected_uncleaned)
    }

    fn insert_policy(
        &self,
        policy: &ScalePolicy,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.insert_policy(policy)
    }

    fn update_policy(
        &self,
        policy: &ScalePolicy,
        expected_revision: u64,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.update_policy(policy, expected_revision)
    }

    fn update_policy_confirming_active_count(
        &self,
        policy: &ScalePolicy,
        expected_revision: u64,
        expected_active: u16,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner
            .update_policy_confirming_active_count(policy, expected_revision, expected_active)
    }

    fn update_policy_confirming_uncleaned_count(
        &self,
        policy: &ScalePolicy,
        expected_revision: u64,
        expected_uncleaned: u16,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.update_policy_confirming_uncleaned_count(
            policy,
            expected_revision,
            expected_uncleaned,
        )
    }

    fn remove_policy(
        &self,
        id: runner_manager_domain::model::PolicyId,
        expected_revision: u64,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.remove_policy(id, expected_revision)
    }

    fn policy(
        &self,
        id: runner_manager_domain::model::PolicyId,
    ) -> Result<Option<ScalePolicy>, runner_manager_domain::store::StoreError> {
        self.inner.policy(id)
    }

    fn policies(&self) -> Result<Vec<ScalePolicy>, runner_manager_domain::store::StoreError> {
        self.inner.policies()
    }

    fn record_attempt(
        &self,
        attempt: &runner_manager_domain::attempt::RunnerAttempt,
    ) -> Result<(), runner_manager_domain::store::StoreError> {
        self.inner.record_attempt(attempt)
    }

    fn attempt(
        &self,
        id: AttemptId,
    ) -> Result<
        Option<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.attempt(id)
    }

    fn attempts(
        &self,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        let mut attempts = self.inner.attempts()?;
        if self.recovering.load(Ordering::Acquire) {
            attempts.retain(|attempt| self.policies.contains(&attempt.policy_id));
        }
        Ok(attempts)
    }

    fn attempts_for_policy(
        &self,
        policy_id: runner_manager_domain::model::PolicyId,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.attempts_for_policy(policy_id)
    }

    // The four workspace reads below are per-policy or host-wide *facts*, not
    // the host-wide capacity view `attempts` narrows during recovery, so they
    // delegate unchanged. Narrowing a slot-lease read would be actively wrong:
    // a lease held by a policy this launcher is not recovering still excludes
    // its slot, and hiding it would let a second attempt take the same `sN`.
    fn active_attempts_for_policy(
        &self,
        policy_id: runner_manager_domain::model::PolicyId,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.active_attempts_for_policy(policy_id)
    }

    fn uncleaned_attempts_for_policy(
        &self,
        policy_id: runner_manager_domain::model::PolicyId,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.uncleaned_attempts_for_policy(policy_id)
    }

    fn slot_leases_for_policy(
        &self,
        policy_id: runner_manager_domain::model::PolicyId,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.slot_leases_for_policy(policy_id)
    }

    fn uncleaned_ephemeral_attempts(
        &self,
    ) -> Result<
        Vec<runner_manager_domain::attempt::RunnerAttempt>,
        runner_manager_domain::store::StoreError,
    > {
        self.inner.uncleaned_ephemeral_attempts()
    }

    fn remove_attempt(
        &self,
        id: AttemptId,
    ) -> Result<bool, runner_manager_domain::store::StoreError> {
        self.inner.remove_attempt(id)
    }
}

/// The policies this daemon supervises, grouped by target.
///
/// # Why a draining policy is here, when it may start nothing
///
/// `may_start_runners` alone was the filter, and it deadlocked the one state
/// that exists to be left. `draining` means *a runner this host owns is still
/// finishing*: the policy stops admitting new ones and waits for the last to
/// end. Filtering it out left nothing supervising that runner — so it was never
/// reaped, the active count never reached zero, the drain never completed, and
/// the policy could not be re-enabled either, because `active` is not a legal
/// transition from `draining`.
///
/// Disabling a policy that held a runner therefore lost it permanently: an
/// online registration, a live listener process and a runtime directory, with
/// nothing left that would ever clean any of them up. Observed on a real host.
///
/// A draining policy loaded here starts nothing — `Reconciler` checks
/// `may_start_runners` itself and allocates zero — so what this admits is
/// supervision, which is exactly what the state is waiting for.
fn active_autoscale_targets(mut policies: Vec<ScalePolicy>) -> Vec<Vec<ScalePolicy>> {
    policies.retain(|policy| policy.may_start_runners() || policy.state() == PolicyState::Draining);
    policies.sort_by(|left, right| left.target.cmp(&right.target));
    let mut targets: Vec<Vec<ScalePolicy>> = Vec::new();
    for policy in policies {
        match targets.last_mut() {
            Some(group) if group[0].target == policy.target => group.push(policy),
            _ => targets.push(vec![policy]),
        }
    }
    targets
}

trait TargetReconciler: Send + 'static {
    fn policies(&self) -> &[ScalePolicy];
    fn begin_drain(&mut self);
    fn reconcile(&mut self) -> Pin<Box<dyn Future<Output = ReconcileReport> + Send + '_>>;

    /// Adopt poll intervals an operator changed while the daemon runs.
    fn refresh_intervals(&mut self) {}

    /// Report one pass's pace, for `status` and `host show`.
    fn observe(&self, _report: &ReconcileReport) {}

    fn active_owned(&self, report: &ReconcileReport) -> Option<u16> {
        active_owned(report, self.policies())
    }

    /// Re-read this target's policies from the journal.
    ///
    /// # Why a running daemon has to be told
    ///
    /// The policy set was read once, at startup, and never again — so every
    /// `set-scale`, `set-capacity` and `add-label` reported success to the
    /// operator and changed nothing until somebody restarted the service. It
    /// was watched happening: a policy drained by the CLI went on starting
    /// runners for three more cycles, because the daemon was still holding the
    /// copy it had loaded minutes earlier.
    ///
    /// A read failure leaves the loaded policies alone. Reconciling against a
    /// set that could not be read would be worse than reconciling against a
    /// slightly stale one, and the next pass tries again.
    fn refresh_policies(&mut self);

    /// Runners this host still holds for this target, counted from the **local
    /// journal** rather than from GitHub.
    ///
    /// # Why a second count exists beside [`Self::active_owned`]
    ///
    /// They answer different questions and fail differently, and an upgrade
    /// needs the one that cannot fail. `active_owned` reads the reconcile
    /// report, and a target GitHub could not be polled contributes no
    /// allocation to it — so its answer is `None`, meaning "unknown", for as
    /// long as the credential or the network is bad.
    ///
    /// Waiting for `Some(0)` from that is what made shutdown hang forever, and
    /// an upgrade drain that waits without a deadline would inherit exactly
    /// that: a revoked token would mean the upgrade never happens, on the one
    /// machine whose daemon most needs replacing.
    ///
    /// Whether *this host* has a runner process alive is a local fact. The
    /// journal holds it, `d1` keeps it true across a crash, and no network is
    /// involved. So the upgrade drain waits on this instead, and can then
    /// afford to wait as long as a job takes.
    fn local_active(&self) -> Option<u16>;
}

struct ManagedTarget {
    policies: Vec<ScalePolicy>,
    reconciler: Reconciler,
    cancel: CancelToken,
    store: Arc<dyn Store>,
    polling: Arc<super::polling::PollingMonitor>,
}

impl TargetReconciler for ManagedTarget {
    fn policies(&self) -> &[ScalePolicy] {
        &self.policies
    }

    fn begin_drain(&mut self) {
        self.cancel.cancel();
        begin_drain(&mut self.policies);
    }

    fn reconcile(&mut self) -> Pin<Box<dyn Future<Output = ReconcileReport> + Send + '_>> {
        Box::pin(self.reconciler.reconcile(&self.policies))
    }

    fn refresh_intervals(&mut self) {
        let intervals = self.polling.intervals();
        self.reconciler
            .set_intervals(intervals.active, intervals.effective_idle());
    }

    fn observe(&self, report: &ReconcileReport) {
        if let Some(policy) = self.policies.first() {
            self.polling.observe(&policy.target, &report.next_poll);
        }
    }

    fn refresh_policies(&mut self) {
        let Some(target) = self.policies.first().map(|policy| policy.target.clone()) else {
            return;
        };
        let Ok(all) = self.store.policies() else {
            tracing::warn!(
                %target,
                "the policy journal could not be read this pass; continuing with the set already                  loaded"
            );
            return;
        };
        let refreshed: Vec<ScalePolicy> = all
            .into_iter()
            .filter(|policy| {
                policy.target == target
                    && (policy.may_start_runners() || policy.state() == PolicyState::Draining)
            })
            .collect();
        // An empty answer means the policy was removed. The loop keeps its last
        // known copy so that `local_active` still names something and any runner
        // still up is supervised; the target-set watch is what ends the loop.
        if !refreshed.is_empty() {
            self.policies = refreshed;
        }
    }

    fn local_active(&self) -> Option<u16> {
        let mut total = 0_u16;
        for policy in &self.policies {
            // A journal this pass could not read is `None`, not zero: reporting
            // an unreadable set as empty would let the upgrade stop a daemon
            // that is in the middle of somebody's job.
            let attempts = self.store.attempts_for_policy(policy.id).ok()?;
            total = total.saturating_add(active_count_for(policy.id, attempts.iter()));
        }
        Some(total)
    }
}

trait ContactRecorder: Send + Sync + 'static {
    fn record(&self) -> Result<(), CliError>;

    /// GitHub rejected the credential this pass. Best effort: the record only
    /// informs `service status` and the TUI, and losing it must not stop the
    /// daemon that would otherwise recover the moment a sign-in lands.
    fn rejected(&self) {}

    /// GitHub accepted the credential this pass.
    fn accepted(&self) {}
}

struct FileContactRecorder {
    paths: runner_manager_platform::paths::AppPaths,
    clock: Arc<dyn Clock>,
    /// Serialises the three records, and holds when the contact record was
    /// last written. Every target loop records a contact on every pass, and an
    /// idle loop passes every few seconds; the record answers "is the service
    /// reaching GitHub now", which a timestamp a few seconds old answers as
    /// well as a fresh one, so it is rewritten at most every
    /// [`CONTACT_RECORD_EVERY`] instead of several times a second. Stamped only
    /// after a write succeeds, so a failed write is retried on the next pass.
    write: Mutex<Option<std::time::Instant>>,
}

/// See [`FileContactRecorder::write`].
const CONTACT_RECORD_EVERY: Duration = Duration::from_secs(30);

impl ContactRecorder for FileContactRecorder {
    fn record(&self) -> Result<(), CliError> {
        let mut written_at = self.write.lock().map_err(|_| {
            CliError::new(
                Failure::LocalState,
                "cannot lock the last successful GitHub contact record",
            )
        })?;
        if written_at.is_some_and(|at| at.elapsed() < CONTACT_RECORD_EVERY) {
            return Ok(());
        }
        record_github_contact(&self.paths, self.clock.now()).map_err(|source| {
            CliError::new(
                Failure::LocalState,
                format!("cannot record the last successful GitHub contact: {source}"),
            )
        })?;
        *written_at = Some(std::time::Instant::now());
        Ok(())
    }

    fn rejected(&self) {
        let Ok(_write) = self.write.lock() else {
            return;
        };
        if let Err(error) = record_github_credential_rejected(&self.paths, self.clock.now()) {
            tracing::warn!(%error, "cannot record that GitHub rejected the credential");
        }
    }

    fn accepted(&self) {
        let Ok(_write) = self.write.lock() else {
            return;
        };
        if let Err(error) = clear_github_credential_rejection(&self.paths) {
            tracing::warn!(%error, "cannot clear the recorded credential rejection");
        }
    }
}

/// How long a drain may run before the daemon stops regardless.
///
/// # Why a drain needs a deadline at all
///
/// The graceful path below ends when the target reports no owned runners left.
/// That reading is an `Option`, and the `None` is not a zero: a target GitHub
/// could not be read contributes no allocation to the report at all
/// (`reconcile.rs`, the `PollOutcome::Failed` arm), so `active_owned` answers
/// `None` and the equality against `Some(0)` is false however long the drain
/// runs. Without a second exit, a daemon whose credential GitHub has rejected
/// drains **forever** — and since that is precisely the state a daemon sits in
/// after a token is revoked, the practical effect was a service that could not
/// be stopped or restarted at all, only killed. It was found that way: seven
/// minutes of `STOP_PENDING`, still polling, on the machine this was reported
/// from.
///
/// # Why stopping anyway is safe
///
/// Nothing is lost by stopping with runners still up. A runner this host owns
/// survives in the journal, and startup recovery is the thing that reconciles
/// it on the next run — adopting one still doing its job, concluding one that
/// is gone. That path is not a fallback added for this; it is how the daemon
/// already starts after any crash or reboot, which is the same situation.
///
/// Sixty seconds is chosen to be longer than a normal poll cycle, so an
/// ordinary drain still ends the graceful way, and short enough that the
/// service manager's own stop timeout does not expire first.
const DRAIN_DEADLINE: Duration = Duration::from_secs(60);

/// Why a target loop is draining, which is the whole of what separates the two
/// drains.
///
/// A **shutdown** is an operator or a machine waiting: it is bounded, and past
/// its deadline the daemon stops with runners still up, because startup
/// recovery will adopt them on the next run.
///
/// An **upgrade** has nobody waiting. Its whole purpose is to replace the
/// binary without interrupting work, so a deadline would defeat it — the one
/// thing it must not do is cut a job short to be timely. It waits on the local
/// journal instead of on GitHub precisely so that waiting forever is safe; see
/// [`TargetReconciler::local_active`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DrainKind {
    Shutdown,
    Upgrade,
}

async fn run_target_loop<T: TargetReconciler>(
    mut target: T,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
    mut upgrade: tokio::sync::watch::Receiver<bool>,
    contacts: Arc<dyn ContactRecorder>,
) -> Result<(), CliError> {
    let mut draining = shutdown_kind(&shutdown, &upgrade);
    let mut drain_deadline = match draining {
        Some(DrainKind::Shutdown) => Some(tokio::time::Instant::now() + DRAIN_DEADLINE),
        _ => None,
    };
    if draining.is_some() {
        target.begin_drain();
    }
    loop {
        if drain_deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline) {
            // Whatever is still running is startup recovery's to adopt.
            return Ok(());
        }
        // Before deciding anything, so a policy an operator changed a moment
        // ago governs this pass rather than the next one.
        target.refresh_policies();
        target.refresh_intervals();
        let report = target.reconcile().await;
        if draining.is_none() {
            target.observe(&report);
        }
        // `reached_github()` alone, and not `failure.is_none()`. Two changes in
        // one line, both deliberate.
        //
        // The conjunction of the two is a tautology: an unauthorized reading
        // *is* a failure, so a non-empty `unreadable` already implies
        // `failure.is_some()`. See `ReconcileReport::reached_github`, which
        // records how long that was believed otherwise.
        //
        // And `failure.is_none()` was the wrong question anyway. It asks
        // whether anything went wrong; what a *last successful contact* needs
        // to know is whether anything went right. A pass that read two targets
        // and lost a third did reach GitHub, and a pass that polled nothing at
        // all did not -- which is the case that really let a host report
        // `healthy` while doing nothing.
        if draining.is_none() && report.reached_github() {
            contacts.record()?;
        }
        // Published so that `service status` and the TUI can say what the log
        // alone used to: this service is signed out, even while an operator's
        // own commands authenticate fine with a copy it cannot see.
        if draining.is_none() {
            if matches!(report.failure, Some(RefreshState::Unauthorized)) {
                contacts.rejected();
            } else if report.reached_github() {
                contacts.accepted();
            }
        }
        match draining {
            Some(DrainKind::Shutdown) if target.active_owned(&report) == Some(0) => {
                return Ok(());
            }
            // GitHub is not consulted: an upgrade must complete even when the
            // credential is the reason the operator is upgrading.
            Some(DrainKind::Upgrade) if target.local_active() == Some(0) => {
                return Ok(());
            }
            _ => {}
        }
        // Capped by the deadline, so a long back-off delay cannot outlast it. A
        // drain that has run out of time sleeps zero and exits at the top.
        let delay = drain_deadline.map_or(report.next_poll.delay, |deadline| {
            report
                .next_poll
                .delay
                .min(deadline.saturating_duration_since(tokio::time::Instant::now()))
        });
        tokio::select! {
            () = tokio::time::sleep(delay) => {}
            changed = shutdown.changed(), if draining != Some(DrainKind::Shutdown) => {
                if changed.is_err() || *shutdown.borrow() {
                    // A shutdown arriving mid-upgrade supersedes it, deadline
                    // and all: the machine is going down either way, and the
                    // upgrade's patience is no longer anybody's benefit.
                    target.begin_drain();
                    draining = Some(DrainKind::Shutdown);
                    drain_deadline = Some(tokio::time::Instant::now() + DRAIN_DEADLINE);
                }
            }
            changed = upgrade.changed(), if draining.is_none() => {
                if changed.is_err() || *upgrade.borrow() {
                    target.begin_drain();
                    draining = Some(DrainKind::Upgrade);
                }
            }
        }
    }
}

/// The drain a loop is already in when it starts, if any. Shutdown wins.
fn shutdown_kind(
    shutdown: &tokio::sync::watch::Receiver<bool>,
    upgrade: &tokio::sync::watch::Receiver<bool>,
) -> Option<DrainKind> {
    if *shutdown.borrow() {
        Some(DrainKind::Shutdown)
    } else if *upgrade.borrow() {
        Some(DrainKind::Upgrade)
    } else {
        None
    }
}

fn active_owned(report: &ReconcileReport, policies: &[ScalePolicy]) -> Option<u16> {
    policies.iter().try_fold(0_u16, |total, policy| {
        report
            .allocations
            .iter()
            .find(|allocation| allocation.policy_id == policy.id)
            .map(|allocation| total.saturating_add(allocation.active_owned))
    })
}

fn begin_drain(policies: &mut [runner_manager_domain::policy::ScalePolicy]) {
    for policy in policies {
        if policy.can_request_disable() {
            let _ = policy.request_disable();
        }
    }
}

/// How long a refusal of the single-instance lock has to persist before it is
/// believed.
///
/// # A refusal that means nothing
///
/// `flock` belongs to the open file description, and `fork` copies every
/// descriptor the parent holds. `FD_CLOEXEC` closes the copy at `exec` — but
/// not before it, so between the two the child holds the parent's lock. Any
/// process this daemon spawns opens that window: `upgraded_version` runs the
/// candidate binary to read its version, and every runner is a child too.
///
/// A daemon starting inside somebody's fork window is therefore refused by a
/// lock nobody actually wants, and reports `another daemon already owns this
/// host` about a child that has already gone.
///
/// # Why waiting distinguishes the two
///
/// The window is bounded by one `exec`; a real second daemon holds the lock for
/// as long as it runs. So a refusal that survives a few retries is the real
/// thing and a refusal that does not was never a conflict. This does not make a
/// genuine conflict quieter -- it still fails, with the same message -- which
/// is the objection [`HostLock::acquire`]'s own documentation raises against
/// waiting on this lock.
///
/// Measured rather than guessed: with this at zero the daemon's own test for
/// the conflict path failed 33 times in 40 runs of the suite under Linux, and
/// never once with the suite serialised. See
/// `a_second_daemon_names_the_holder_and_uses_the_conflict_exit_class`.
const SINGLE_INSTANCE_SETTLE: Duration = Duration::from_millis(250);

fn acquire_instance(context: &Context) -> Result<HostLock, CliError> {
    HostLock::acquire(
        context.paths(),
        LockKind::SingleInstance,
        SINGLE_INSTANCE_SETTLE,
    )
    .map_err(|source| match source {
        held @ LockError::Held { .. } => CliError::with_remedy(
            Failure::Conflict,
            format!("another daemon already owns this host: {held}"),
            "runner-manager service status",
        ),
        other => CliError::new(
            Failure::LocalState,
            format!("cannot acquire the daemon's single-instance lock: {other}"),
        ),
    })
}

/// Start each daemon generation with nothing inherited about launches.
///
/// A blocked-launch record describes the refusals one daemon saw, so a new one
/// starts without it and counts afresh. And with the single-instance lock held,
/// no other launcher can be using this distribution's guest claim, so a claim
/// whose owner is gone -- a daemon killed mid-launch, a WSL restart -- is
/// reclaimed here rather than at the first launch, which would otherwise be a
/// minute or more away.
fn reset_launch_state(context: &Context) {
    use runner_manager_platform::launch_health::clear_launches_blocked;
    use runner_manager_platform::wsl::fence::reclaim_stale_guest_claim;

    if let Err(error) = clear_launches_blocked(context.paths()) {
        tracing::warn!(%error, "the previous daemon's blocked-launch record cannot be cleared");
    }
    let Ok(Some(config)) = GuestRecoveryConfig::read(context.paths()) else {
        return;
    };
    match reclaim_stale_guest_claim(&config.shared_root) {
        Ok(Some(reclaimed)) => tracing::warn!(
            shared_root = %config.shared_root.display(),
            %reclaimed,
            "reclaimed a WSL launch fence claim whose owner is gone"
        ),
        Ok(None) => {}
        Err(error) => tracing::warn!(
            shared_root = %config.shared_root.display(),
            %error,
            "the WSL launch fence cannot be inspected at daemon start"
        ),
    }
}

fn local_store_failure(source: runner_manager_domain::store::StoreError) -> CliError {
    CliError::new(
        Failure::LocalState,
        format!("cannot read the daemon's local database: {source}"),
    )
}

fn github_failure(source: GithubError) -> CliError {
    CliError::with_remedy(
        Failure::GithubUnavailable,
        source.to_string(),
        "runner-manager auth status",
    )
}

fn signal_failure(source: std::io::Error) -> CliError {
    CliError::new(
        Failure::UnsupportedHost,
        format!("cannot listen for the daemon shutdown signal: {source}"),
    )
}

#[cfg(unix)]
async fn shutdown_signal() -> std::io::Result<()> {
    let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
    tokio::select! {
        result = tokio::signal::ctrl_c() => result,
        _ = terminate.recv() => Ok(()),
    }
}

#[cfg(not(unix))]
async fn shutdown_signal() -> std::io::Result<()> {
    tokio::signal::ctrl_c().await
}

#[derive(Debug)]
struct GithubLifecycle {
    jit: Arc<RestJit>,
    inventory: Arc<RestInventory>,
    clock: Arc<dyn Clock>,
}

impl LifecycleGithub for GithubLifecycle {
    fn register<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        target: &'life1 ScaleTarget,
        request: &'life2 JitRunnerRequest,
        cancel: &'life3 CancelToken,
    ) -> Pin<
        Box<
            dyn Future<
                    Output = Result<
                        runner_manager_github::jit::JitRegistration,
                        runner_manager_agent::lifecycle::JitRequestFailure,
                    >,
                > + Send
                + 'async_trait,
        >,
    >
    where
        'life0: 'async_trait,
        'life1: 'async_trait,
        'life2: 'async_trait,
        'life3: 'async_trait,
        Self: 'async_trait,
    {
        Box::pin(async move {
            self.jit
                .generate_jit_config(target, request, cancel)
                .await
                .map_err(|error| runner_manager_agent::lifecycle::JitRequestFailure {
                    terminal: error.is_terminal(),
                    retry_after: error.rate_limited().map(|limit| limit.delay_from(self.clock.now())),
                    reason: if matches!(error, JitError::Forbidden { .. }) {
                        FailureReason::Other("GitHub refused JIT registration; check the App runner permission and runner-group access".into())
                    } else {
                        FailureReason::JitRequestFailed
                    },
                })
        })
    }

    fn observe<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        target: &'life1 ScaleTarget,
        attempt: AttemptId,
        cancel: &'life2 CancelToken,
    ) -> Pin<Box<dyn Future<Output = LifecycleGithubObservation> + Send + 'async_trait>>
    where
        'life0: 'async_trait,
        'life1: 'async_trait,
        'life2: 'async_trait,
        Self: 'async_trait,
    {
        Box::pin(async move {
            let name = format!("runner-manager-{attempt}");
            match self.inventory.list_runners(target, cancel).await {
                Ok(inventory) => inventory
                    .runners()
                    .iter()
                    .find(|runner| runner.name == name)
                    .map_or(LifecycleGithubObservation::not_registered(), |runner| {
                        LifecycleGithubObservation::registered(runner.id, runner.busy)
                    }),
                Err(_) => LifecycleGithubObservation::unreachable(),
            }
        })
    }

    fn deregister<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        target: &'life1 ScaleTarget,
        runner_id: u64,
        cancel: &'life2 CancelToken,
    ) -> Pin<Box<dyn Future<Output = bool> + Send + 'async_trait>>
    where
        'life0: 'async_trait,
        'life1: 'async_trait,
        'life2: 'async_trait,
        Self: 'async_trait,
    {
        Box::pin(async move {
            self.inventory
                .remove_runner(target, runner_id, cancel)
                .await
                .is_ok()
        })
    }
}

#[derive(Debug)]
struct GithubDirectory {
    client: Arc<AuthenticatedClient>,
    app: AppRegistration,
}

impl RepositoryDirectory for GithubDirectory {
    fn repositories<'life0, 'life1, 'async_trait>(
        &'life0 self,
        org: &'life1 Org,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<OwnerRepo>, InventoryError>> + Send + 'async_trait>>
    where
        'life0: 'async_trait,
        'life1: 'async_trait,
        Self: 'async_trait,
    {
        Box::pin(async move {
            let discovery = self
                .client
                .discover_installations(&self.app)
                .await
                .map_err(InventoryError::from)?;
            Ok(discovery
                .targets()
                .map(|targets| {
                    targets
                        .repositories()
                        .into_iter()
                        .filter(|repository| repository.owner().eq_ignore_ascii_case(org.as_str()))
                        .collect()
                })
                .unwrap_or_default())
        })
    }
}

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

    #[tokio::test]
    async fn diagnostics_maintenance_expires_history_without_waiting_for_a_job() {
        let root = tempfile::tempdir().unwrap();
        let summary = root.path().join(format!("{}.log", AttemptId::from_u128(1)));
        let file = std::fs::File::create(&summary).unwrap();
        file.set_modified(
            std::time::SystemTime::now()
                - runner_manager_platform::log_retention::ATTEMPT_LOG_MAX_AGE
                - Duration::from_secs(1),
        )
        .unwrap();
        drop(file);
        let service = root.path().join("runner-manager.service.log.2020-01-01");
        std::fs::write(&service, b"not an attempt summary").unwrap();

        let maintenance = tokio::spawn(maintain_diagnostics(root.path().to_path_buf()));
        let expired = tokio::time::timeout(Duration::from_secs(10), async {
            while summary.exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await;
        maintenance.abort();
        let _ = maintenance.await;
        expired.expect("the first maintenance pass runs at startup without any job");
        assert!(service.exists());
    }

    fn rooted(root: &std::path::Path) -> Context {
        let endpoints =
            runner_manager_github::Endpoints::for_test_server("http://127.0.0.1:9").unwrap();
        Context::rooted_against(root, endpoints).unwrap()
    }

    /// Host checks blocked behind a volume that never answers must not hold
    /// the daemon's loop, and leave the host unfit with what to do.
    #[tokio::test(start_paused = true)]
    async fn host_checks_that_never_finish_make_the_host_unfit() {
        let root = tempfile::tempdir().unwrap();
        let context = rooted(root.path());
        context.paths().create_all().unwrap();
        let stuck_root = std::path::PathBuf::from("/Volumes/NVME/runners");
        // Bounded from outside too, so a regression fails here rather than
        // hanging the run.
        let verdict = tokio::time::timeout(
            Duration::from_secs(600),
            bounded_preflight(
                std::future::pending(),
                Duration::from_secs(120),
                || vec![stuck_root.clone()],
                &context,
            ),
        )
        .await
        .expect("host checks that never finish are given up on")
        .expect("a verdict");
        assert_eq!(verdict.required, [BLOCKED_CHECKS_ID]);
        let record = runner_manager_platform::host_fitness::host_unfit(context.paths())
            .unwrap()
            .expect("recorded for status");
        assert!(
            record
                .describe()
                .contains("/Volumes/NVME/runners does not answer"),
            "{record:?}"
        );

        // A recheck while that evaluation still runs says the same.
        let again = bounded_preflight(
            async { Err(super::super::doctor::PREFLIGHT_STILL_RUNNING.to_owned()) },
            Duration::from_secs(120),
            || vec![stuck_root.clone()],
            &context,
        )
        .await
        .unwrap();
        assert_eq!(again.required, [BLOCKED_CHECKS_ID]);

        // Slow, with no directory blocked: nothing is held back.
        let slow = bounded_preflight(
            std::future::pending(),
            Duration::from_secs(120),
            Vec::new,
            &context,
        )
        .await;
        assert!(slow.is_err(), "{slow:?}");

        // A preflight that answers is passed through untouched.
        let answered = bounded_preflight(
            async { Ok(super::super::doctor::DaemonVerdict::default()) },
            Duration::from_secs(120),
            Vec::new,
            &context,
        )
        .await
        .unwrap();
        assert!(answered.required.is_empty());
    }

    /// The heartbeat beats while the generation runs and is gone once it
    /// returns.
    #[tokio::test]
    async fn the_heartbeat_beats_until_the_generation_ends() {
        use runner_manager_platform::daemon_heartbeat::{Liveness, liveness};
        let root = tempfile::tempdir().unwrap();
        let paths = runner_manager_platform::paths::AppPaths::rooted_at(root.path());
        paths.create_all().unwrap();
        let heartbeat = Heartbeat::start(paths.clone());
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(matches!(
            liveness(&paths, chrono::Utc::now()),
            Liveness::Beating { .. }
        ));
        drop(heartbeat);
        assert_eq!(liveness(&paths, chrono::Utc::now()), Liveness::Unknown);
    }
    use std::collections::VecDeque;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use runner_manager_agent::lifecycle::{
        JitRequestFailure, NoAttemptEvents, PruneAuthority, RuntimePackages,
    };
    use runner_manager_agent::package::RunnerVersion;
    use runner_manager_domain::attempt::RunnerAttempt;
    use runner_manager_domain::model::PolicyId;
    use runner_manager_domain::store::SqliteStore;
    use runner_manager_github::RenewalError;
    use runner_manager_github::jit::JitRegistration;
    use runner_manager_github::rest::RefreshState;
    use runner_manager_testkit::clock::FakeClock;
    use runner_manager_testkit::fixtures;

    #[test]
    fn scm_service_never_claims_authority_over_user_owned_wsl() {
        assert!(!wsl_recovery_is_available(true));
        assert_eq!(wsl_recovery_is_available(false), cfg!(windows));
    }

    #[derive(Debug)]
    struct CountingRenewal {
        calls: Arc<AtomicUsize>,
    }

    #[async_trait::async_trait]
    impl CredentialRenewal for CountingRenewal {
        async fn renew(
            &self,
            _refresh_token: &SecretString,
        ) -> Result<UserAccessToken, RenewalError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(UserAccessToken::from_stored(SecretString::from(
                serde_json::json!({
                    "access_token": "ghu_daemon_renewed_fixture",
                    "refresh_token": "ghr_daemon_renewed_fixture",
                    "access_expires_at": "2027-08-21T00:00:00Z",
                    "refresh_expires_at": "2028-02-21T00:00:00Z"
                })
                .to_string(),
            )))
        }
    }

    #[derive(Debug)]
    struct RecoveryGithub {
        expected: ScaleTarget,
    }

    impl LifecycleGithub for RecoveryGithub {
        fn register<'life0, 'life1, 'life2, 'life3, 'async_trait>(
            &'life0 self,
            _target: &'life1 ScaleTarget,
            _request: &'life2 JitRunnerRequest,
            _cancel: &'life3 CancelToken,
        ) -> Pin<
            Box<
                dyn Future<Output = Result<JitRegistration, JitRequestFailure>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            'life3: 'async_trait,
            Self: 'async_trait,
        {
            Box::pin(async { panic!("startup recovery must not register a runner") })
        }

        fn observe<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            target: &'life1 ScaleTarget,
            _attempt: AttemptId,
            _cancel: &'life2 CancelToken,
        ) -> Pin<Box<dyn Future<Output = LifecycleGithubObservation> + Send + 'async_trait>>
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: 'async_trait,
        {
            assert_eq!(
                target, &self.expected,
                "a target launcher observed another target's startup attempt"
            );
            Box::pin(std::future::ready(LifecycleGithubObservation::registered(
                73, false,
            )))
        }

        fn deregister<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            target: &'life1 ScaleTarget,
            _runner_id: u64,
            _cancel: &'life2 CancelToken,
        ) -> Pin<Box<dyn Future<Output = bool> + Send + 'async_trait>>
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: 'async_trait,
        {
            assert_eq!(
                target, &self.expected,
                "a target launcher deregistered another target's runner"
            );
            Box::pin(std::future::ready(true))
        }
    }

    #[derive(Debug)]
    struct UnusedPackages;

    impl RuntimePackages for UnusedPackages {
        fn materialize<'life0, 'life1, 'async_trait>(
            &'life0 self,
            _attempt: &'life1 RunnerAttempt,
        ) -> Pin<Box<dyn Future<Output = Result<RunnerVersion, FailureReason>> + Send + 'async_trait>>
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            Self: 'async_trait,
        {
            Box::pin(async { panic!("startup recovery must not materialize a package") })
        }

        fn release(&self, _attempt: AttemptId) -> Result<(), FailureReason> {
            Ok(())
        }

        fn prune_obsolete_guarded(
            &self,
            _authority: PruneAuthority<'_>,
            _current: &RunnerVersion,
            _attempts: &[RunnerAttempt],
        ) -> Result<(), FailureReason> {
            Ok(())
        }
    }

    #[derive(Debug)]
    struct FakeTarget {
        policy: ScalePolicy,
        reports: VecDeque<ReconcileReport>,
        calls: Arc<AtomicUsize>,
        active: u16,
        draining: bool,
        busy_was_terminated: Arc<AtomicUsize>,
        /// Models a target GitHub could not be read: `reconcile` contributes no
        /// allocation for it, so `active_owned` can only answer "unknown".
        unreadable: bool,
        /// How many times the loop asked for a fresh policy set.
        refreshes: Arc<AtomicUsize>,
    }

    impl FakeTarget {
        fn repeating(policy: ScalePolicy, report: ReconcileReport) -> (Self, Arc<AtomicUsize>) {
            let calls = Arc::new(AtomicUsize::new(0));
            (
                Self {
                    policy,
                    reports: VecDeque::from([report]),
                    calls: Arc::clone(&calls),
                    active: 0,
                    draining: false,
                    busy_was_terminated: Arc::new(AtomicUsize::new(0)),
                    unreadable: false,
                    refreshes: Arc::new(AtomicUsize::new(0)),
                },
                calls,
            )
        }

        fn busy_then_finished(policy: ScalePolicy) -> Self {
            Self {
                policy,
                reports: VecDeque::new(),
                calls: Arc::new(AtomicUsize::new(0)),
                active: 1,
                draining: false,
                busy_was_terminated: Arc::new(AtomicUsize::new(0)),
                unreadable: false,
                refreshes: Arc::new(AtomicUsize::new(0)),
            }
        }

        /// A target that never reports a runner count, because GitHub never
        /// answers for it.
        fn never_readable(policy: ScalePolicy) -> Self {
            Self {
                policy,
                reports: VecDeque::new(),
                calls: Arc::new(AtomicUsize::new(0)),
                active: 1,
                draining: false,
                busy_was_terminated: Arc::new(AtomicUsize::new(0)),
                unreadable: true,
                refreshes: Arc::new(AtomicUsize::new(0)),
            }
        }
    }

    impl TargetReconciler for FakeTarget {
        fn policies(&self) -> &[ScalePolicy] {
            std::slice::from_ref(&self.policy)
        }

        fn begin_drain(&mut self) {
            self.draining = true;
            begin_drain(std::slice::from_mut(&mut self.policy));
        }

        fn reconcile(&mut self) -> Pin<Box<dyn Future<Output = ReconcileReport> + Send + '_>> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            if self.draining && self.active > 0 {
                // The first drain pass supervises the busy child without any
                // termination action. The next pass observes its ordinary
                // completion and permits daemon exit.
                if self.calls.load(Ordering::SeqCst) >= 3 {
                    self.active = 0;
                }
            }
            let report = self.reports.front().cloned().unwrap_or_else(|| {
                let mut report = ReconcileReport::default();
                report.next_poll.delay = Duration::from_secs(1);
                report
            });
            Box::pin(std::future::ready(report))
        }

        fn active_owned(&self, _report: &ReconcileReport) -> Option<u16> {
            if self.unreadable {
                return None;
            }
            Some(self.active)
        }

        /// Deliberately answers even when `unreadable`: that is the whole point
        /// of the local count, and a fake that hid it could not show the
        /// difference the upgrade drain depends on.
        fn refresh_policies(&mut self) {
            self.refreshes.fetch_add(1, Ordering::SeqCst);
        }

        fn local_active(&self) -> Option<u16> {
            Some(self.active)
        }
    }

    /// An upgrade signal that never fires, for the tests that are about
    /// shutdown.
    fn never_upgraded() -> tokio::sync::watch::Receiver<bool> {
        let (sender, receiver) = tokio::sync::watch::channel(false);
        // Kept alive for the receiver's lifetime; a dropped sender would make
        // `changed()` resolve immediately and read as an upgrade.
        Box::leak(Box::new(sender));
        receiver
    }

    #[tokio::test(start_paused = true)]
    async fn an_upgrade_waits_for_a_running_job_however_long_it_takes() {
        // The promise this feature makes: a new binary never interrupts work.
        // `busy_then_finished` holds one runner for the first two passes.
        let target = FakeTarget::busy_then_finished(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
        );
        let terminations = Arc::clone(&target.busy_was_terminated);
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let (upgrade, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            upgrade.subscribe(),
            contacts as Arc<dyn ContactRecorder>,
        ));

        tokio::task::yield_now().await;
        upgrade.send(true).unwrap();

        // Far past the shutdown deadline, which must not apply here: an upgrade
        // that gave up after a minute would cut the job it exists to protect.
        for _ in 0..DRAIN_DEADLINE.as_secs() {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }

        // The fake releases its runner on the third pass, and only then does
        // the loop end.
        for _ in 0..5 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        tokio::time::timeout(Duration::from_secs(1), daemon)
            .await
            .expect("the upgrade drain ends once the job is done")
            .unwrap()
            .unwrap();
        assert_eq!(
            terminations.load(Ordering::SeqCst),
            0,
            "an upgrade must never terminate a running job"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn an_upgrade_completes_even_when_github_cannot_be_read() {
        // The reason the upgrade drain counts locally. `never_readable` answers
        // `None` from `active_owned` forever -- the revoked-credential state --
        // while its local count still falls to zero. Waiting on the former
        // would mean the machine that most needs a new binary never gets one.
        let mut target = FakeTarget::never_readable(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
        );
        target.active = 0;
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let (upgrade, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            upgrade.subscribe(),
            contacts as Arc<dyn ContactRecorder>,
        ));

        tokio::task::yield_now().await;
        upgrade.send(true).unwrap();
        for _ in 0..5 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        tokio::time::timeout(Duration::from_secs(1), daemon)
            .await
            .expect("an unreadable target must not block an upgrade")
            .unwrap()
            .unwrap();
    }

    /// A rewrite of the *same* version is not an upgrade: restarting for it
    /// would interrupt runners to change nothing.
    #[test]
    fn the_running_version_is_not_an_upgrade_of_itself() {
        let own = std::env::current_exe().expect("the test binary's own path");
        assert!(
            BinaryStamp::of(&own).is_some(),
            "a running binary must be stat-able"
        );
        assert!(
            BinaryStamp::of(std::path::Path::new("no-such-binary")).is_none(),
            "a missing file has no stamp, and is not mistaken for a new one"
        );
        assert!(
            upgraded_version(std::path::Path::new("no-such-binary")).is_none(),
            "a path that cannot be executed is never reported as an upgrade"
        );
        assert_eq!(
            changed_build_version(concat!(
                "runner-manager ",
                env!("RUNNER_MANAGER_BUILD_VERSION"),
                "\n"
            )),
            None,
            "a commit-qualified source build must recognize its own identity"
        );
        assert_eq!(
            changed_build_version("runner-manager 9.9.9\n").as_deref(),
            Some("9.9.9"),
            "a genuinely different build must still trigger handover"
        );
    }

    #[test]
    fn an_idle_host_uses_the_normal_upgrade_handover() {
        let temporary = tempfile::tempdir().unwrap();
        let context = Context::resolve(Some(temporary.path()), &mut Vec::new()).unwrap();
        let mut output = Vec::new();
        let error = stop_for_upgrade(&context, StartMode::Login, None, "9.9.9", &mut output)
            .expect_err("an upgrade exits for the service manager to restart it");
        assert_eq!(error.class(), Failure::UpgradePending);
        let output = String::from_utf8(output).unwrap();
        assert!(output.contains("9.9.9"), "{output}");
        assert!(output.contains("every runner finished"), "{output}");
    }

    #[tokio::test(start_paused = true)]
    async fn daemon_maintenance_renews_an_expired_pair_without_tui_traffic() {
        let calls = Arc::new(AtomicUsize::new(0));
        let renewal: Arc<dyn CredentialRenewal> = Arc::new(CountingRenewal {
            calls: Arc::clone(&calls),
        });
        let expired = UserAccessToken::from_stored(SecretString::from(
            serde_json::json!({
                "access_token": "ghu_daemon_expired_fixture",
                "refresh_token": "ghr_daemon_expired_fixture",
                "access_expires_at": "2026-08-20T23:59:59Z",
                "refresh_expires_at": "2027-02-21T00:00:00Z"
            })
            .to_string(),
        ));
        let client = Arc::new(
            AuthenticatedClient::new(
                runner_manager_github::Endpoints::production(),
                expired,
                Arc::new(FakeClock::default()),
            )
            .unwrap()
            .with_renewal(renewal),
        );

        let maintenance = tokio::spawn(maintain_credential(client));
        tokio::task::yield_now().await;
        assert_eq!(calls.load(Ordering::SeqCst), 1);

        tokio::time::advance(CREDENTIAL_MAINTENANCE_INTERVAL).await;
        tokio::task::yield_now().await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "the fresh pair must not be rotated again on every maintenance tick"
        );
        maintenance.abort();
    }

    #[derive(Default)]
    struct CountingContacts(AtomicUsize);

    impl ContactRecorder for CountingContacts {
        fn record(&self) -> Result<(), CliError> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }

    #[test]
    fn a_second_daemon_names_the_holder_and_uses_the_conflict_exit_class() {
        let temporary = tempfile::tempdir().unwrap();
        let context = Context::resolve(Some(temporary.path()), &mut Vec::new()).unwrap();
        let held = acquire_instance(&context).expect("first daemon acquires the lock");
        let error = acquire_instance(&context).expect_err("second daemon must be refused");
        assert_eq!(error.class(), Failure::Conflict);
        assert!(error.message().contains(&std::process::id().to_string()));
        drop(held);
        acquire_instance(&context).expect("dropping the daemon releases the lock");
    }

    /// The other half of [`SINGLE_INSTANCE_SETTLE`]: a refusal that does not
    /// persist was never a conflict.
    ///
    /// # What this stands for
    ///
    /// A process this daemon spawns, in the moment between `fork` and `exec`.
    /// The child holds a copy of every descriptor the parent had, `FD_CLOEXEC`
    /// closes it only at the `exec`, and `flock` belongs to the open file
    /// description — so for that instant the child holds this host's
    /// single-instance lock without wanting it or knowing about it.
    ///
    /// The test above is what caught it, from the other side: with no settle
    /// window it failed 33 times in 40 runs of this suite under Linux, always
    /// on its last line, and never once with the suite serialised. The holder
    /// in the refusal was a process id *higher* than the test's own — a child
    /// spawned by a neighbouring test, which by then had already exited.
    ///
    /// # Why this measures a floor and not a race
    ///
    /// Two earlier versions of this test staged a transient holder — a thread
    /// that took the lock and released it again — and asserted that
    /// `acquire_instance` rode it out. Both flaked: 2 in 100, then 4 in 150.
    /// Releasing needs the holder to be *scheduled*, and a thread competing
    /// with 155 others on eight cores is not scheduled on any deadline, so both
    /// versions were really asserting "the scheduler will get to it within
    /// 250ms". Asserting an upper bound on someone else's timing is exactly the
    /// mistake that produced the flake this whole change is about, written a
    /// second and third time while fixing it.
    ///
    /// So this asserts a *lower* bound instead, which load can only make more
    /// true: a refusal is not believed until it has been retried across the
    /// settle window. That is the property that makes a fork window survivable,
    /// and it is the property `try_acquire` did not have.
    #[test]
    fn a_refused_lock_is_retried_before_the_refusal_is_believed() {
        let temporary = tempfile::tempdir().unwrap();
        let context = Context::resolve(Some(temporary.path()), &mut Vec::new()).unwrap();
        let _held = acquire_instance(&context).expect("first daemon acquires the lock");

        let started = std::time::Instant::now();
        let error = acquire_instance(&context).expect_err("a lock still held is still a conflict");
        let waited = started.elapsed();

        assert_eq!(error.class(), Failure::Conflict);
        assert!(
            waited >= SINGLE_INSTANCE_SETTLE,
            "a refusal must be retried across the settle window before it is believed, or a \
             child's fork window reads as a second daemon. Gave up after {waited:?}, which is \
             less than {SINGLE_INSTANCE_SETTLE:?}"
        );
    }

    #[test]
    fn only_active_autoscale_policies_are_loaded_in_stable_target_order() {
        let active_z = fixtures::policy()
            .repository("zeta/repo")
            .autoscale("home", 1)
            .active()
            .build();
        let active_a = fixtures::policy()
            .repository("alpha/repo")
            .autoscale("home", 1)
            .active()
            .build();
        let pending = fixtures::policy()
            .repository("pending/repo")
            .autoscale("home", 1)
            .build();
        let monitor = fixtures::policy()
            .repository("monitor/repo")
            .monitor_only()
            .active()
            .build();
        let mut draining = fixtures::policy()
            .repository("draining/repo")
            .autoscale("home", 1)
            .active()
            .build();
        draining.request_disable().unwrap();
        let mut disabled = fixtures::policy()
            .repository("disabled/repo")
            .autoscale("home", 1)
            .active()
            .build();
        disabled.request_disable().unwrap();
        disabled.drain_completed(0).unwrap();

        let active_a_second = fixtures::policy()
            .repository("alpha/repo")
            .autoscale("home", 1)
            .active()
            .build();
        let selected = active_autoscale_targets(vec![
            active_z,
            pending,
            disabled,
            monitor,
            active_a,
            draining,
            active_a_second,
        ]);
        let targets: Vec<_> = selected
            .iter()
            .map(|policies| policies[0].target.to_string())
            .collect();
        assert_eq!(
            targets,
            ["alpha/repo", "draining/repo", "zeta/repo"],
            "a draining policy is supervised until its last runner ends"
        );
        assert_eq!(selected[0].len(), 2, "same-target policies share one loop");

        // Everything loaded either starts runners or is finishing the ones it
        // has. Nothing else is here: `pending` has not been armed, `monitor` is
        // monitor-only, and `disabled` has already drained to zero.
        assert!(selected.iter().flatten().all(|policy| {
            policy.may_start_runners() || policy.state() == PolicyState::Draining
        }));
        assert!(
            !targets
                .iter()
                .any(|t| t == "pending/repo" || t == "monitor/repo"),
            "{targets:?}"
        );
        assert!(!targets.iter().any(|t| t == "disabled/repo"), "{targets:?}");
    }

    #[test]
    fn named_native_and_isolated_profiles_share_one_target_loop() {
        let native = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .active()
            .build();
        let mut isolated = fixtures::named_policy("py-isolated", PolicyId::from_u128(2));
        isolated.activate().unwrap();
        let monitor = fixtures::policy()
            .id(PolicyId::from_u128(3))
            .monitor_only()
            .active()
            .build();
        let groups = active_autoscale_targets(vec![isolated, monitor, native]);
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].len(), 2);
        assert_eq!(groups[0][0].target, groups[0][1].target);
        assert_ne!(groups[0][0].profile_name(), groups[0][1].profile_name());
    }

    #[test]
    fn case_variant_targets_stay_in_one_loop_when_another_target_interleaves() {
        let upper = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .repository("A/repo")
            .active()
            .build();
        let middle = fixtures::policy()
            .id(PolicyId::from_u128(2))
            .repository("B/repo")
            .active()
            .build();
        let lower = fixtures::policy()
            .id(PolicyId::from_u128(3))
            .repository("a/repo")
            .active()
            .build();
        let groups = active_autoscale_targets(vec![upper, middle, lower]);
        assert_eq!(groups.len(), 2, "equal targets must share a daemon loop");
        assert_eq!(groups[0].len(), 2);
        assert_eq!(groups[0][0].target, groups[0][1].target);
        assert_eq!(groups[1].len(), 1);
    }

    /// A policy disabled while it still held a runner used to be lost for good.
    ///
    /// `draining` was filtered out of the daemon's targets, so nothing
    /// supervised the runner it was waiting on: it was never reaped, the active
    /// count never reached zero, the drain never completed — and `active` is not
    /// a legal transition from `draining`, so it could not be re-enabled either.
    /// The runner stayed online at GitHub with a live process and a runtime
    /// directory that nothing would ever clean up.
    #[test]
    fn a_draining_policy_is_still_supervised_or_its_last_runner_is_abandoned() {
        let mut draining = fixtures::policy()
            .repository("acme/repo")
            .autoscale("home", 1)
            .active()
            .build();
        draining.request_disable().expect("an active policy drains");
        assert_eq!(draining.state(), PolicyState::Draining);
        assert!(
            !draining.may_start_runners(),
            "the discriminator: it admits no new runners, which is why the old              filter dropped it"
        );

        let selected = active_autoscale_targets(vec![draining]);
        assert_eq!(
            selected.len(),
            1,
            "without this the drain can never finish and the policy is stuck forever"
        );
        assert_eq!(selected[0][0].state(), PolicyState::Draining);
    }

    #[tokio::test]
    async fn startup_recovery_keeps_each_targets_replacement_intent_in_its_launcher() {
        let temporary = tempfile::tempdir().unwrap();
        let store = Arc::new(SqliteStore::open_in_memory().unwrap());
        let policy_a = fixtures::policy()
            .id(PolicyId::from_u128(1))
            .repository("acme/alpha")
            .autoscale("home", 1)
            .active()
            .build();
        let policy_b = fixtures::policy()
            .id(PolicyId::from_u128(2))
            .repository("acme/beta")
            .autoscale("home", 1)
            .active()
            .build();
        store.insert_policy(&policy_a).unwrap();
        store.insert_policy(&policy_b).unwrap();

        let attempt_for = |id: u128, policy: &ScalePolicy, directory: &str| {
            let runtime = temporary.path().join(directory);
            std::fs::create_dir_all(&runtime).unwrap();
            fixtures::attempt()
                .id(AttemptId::from_u128(id))
                .policy_id(policy.id)
                .runtime_path(runtime.to_string_lossy())
                .build()
        };
        let attempt_a = attempt_for(11, &policy_a, "alpha");
        let attempt_b = attempt_for(22, &policy_b, "beta");
        store.record_attempt(&attempt_a).unwrap();
        store.record_attempt(&attempt_b).unwrap();

        let build_launcher = |policy: &ScalePolicy| {
            let scoped = Arc::new(TargetRecoveryStore::new(
                Arc::clone(&store) as Arc<dyn Store>,
                std::slice::from_ref(policy),
            ));
            let app_paths = runner_manager_platform::paths::AppPaths::rooted_at(temporary.path());
            let launcher = LifecycleLauncher::new(
                policy.to_persisted().host_id,
                app_paths,
                temporary.path().join("logs"),
                1,
                runner_manager_domain::attempt::RecoveryTimeouts::provisional(),
                RetryPolicy::bounded(1, Duration::from_millis(1), Duration::from_millis(1)),
                LifecyclePorts {
                    store: Arc::clone(&scoped) as Arc<dyn Store>,
                    github: Arc::new(RecoveryGithub {
                        expected: policy.target.clone(),
                    }),
                    packages: Arc::new(UnusedPackages),
                    processes: Arc::new(NativeProcesses::new()),
                    clock: Arc::new(FakeClock::default()),
                    demand: Arc::new(PersistentDemand),
                    delay: Arc::new(TokioRetryDelay),
                    events: Arc::new(NoAttemptEvents),
                    reconcile_events: Arc::new(runner_manager_agent::reconcile::EventLog::new()),
                },
            );
            (launcher, scoped)
        };
        let (launcher_a, scoped_a) = build_launcher(&policy_a);
        let (launcher_b, scoped_b) = build_launcher(&policy_b);

        let placed_a = launcher_a
            .recover_startup(std::slice::from_ref(&policy_a))
            .await
            .unwrap();
        scoped_a.finish_recovery();
        let placed_b = launcher_b
            .recover_startup(std::slice::from_ref(&policy_b))
            .await
            .unwrap();
        scoped_b.finish_recovery();
        assert_eq!(placed_a.len(), 1);
        assert_eq!(placed_a[0].policy, policy_a.id);
        assert_eq!(placed_a[0].previous_attempt, attempt_a.id);
        assert_eq!(placed_b.len(), 1);
        assert_eq!(placed_b[0].policy, policy_b.id);
        assert_eq!(placed_b[0].previous_attempt, attempt_b.id);

        let consumed_a = launcher_a.supervise(&policy_a).await.unwrap();
        let consumed_b = launcher_b.supervise(&policy_b).await.unwrap();
        assert_eq!(consumed_a, placed_a);
        assert_eq!(consumed_b, placed_b);
        assert!(launcher_a.supervise(&policy_a).await.unwrap().is_empty());
        assert!(launcher_b.supervise(&policy_b).await.unwrap().is_empty());
        assert_eq!(scoped_a.attempts().unwrap().len(), 2);
        assert_eq!(scoped_b.attempts().unwrap().len(), 2);
    }

    #[tokio::test(start_paused = true)]
    async fn an_offline_target_neither_suppresses_contacts_nor_backs_off_a_healthy_target() {
        let policy = |repository| {
            fixtures::policy()
                .repository(repository)
                .autoscale("home", 1)
                .active()
                .build()
        };
        let healthy_report = ReconcileReport {
            // What makes this target healthy rather than merely uncomplaining.
            // The default report reads nothing and fails on nothing, which is
            // what a pass with no policy to poll looks like -- and such a pass
            // reaches GitHub not at all, so it records no contact. Saying so
            // here is the fixture describing the thing it is named for.
            targets_read: 1,
            next_poll: runner_manager_agent::reconcile::NextPoll {
                delay: Duration::from_secs(1),
                ..Default::default()
            },
            ..Default::default()
        };
        let offline_report = ReconcileReport {
            failure: Some(RefreshState::Offline),
            next_poll: runner_manager_agent::reconcile::NextPoll {
                delay: Duration::from_secs(60),
                ..Default::default()
            },
            ..Default::default()
        };
        let (healthy, healthy_calls) =
            FakeTarget::repeating(policy("acme/healthy"), healthy_report);
        let (offline, offline_calls) =
            FakeTarget::repeating(policy("acme/offline"), offline_report);
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let healthy_loop = tokio::spawn(run_target_loop(
            healthy,
            stop.subscribe(),
            never_upgraded(),
            Arc::clone(&contacts) as Arc<dyn ContactRecorder>,
        ));
        let offline_loop = tokio::spawn(run_target_loop(
            offline,
            stop.subscribe(),
            never_upgraded(),
            Arc::clone(&contacts) as Arc<dyn ContactRecorder>,
        ));

        tokio::task::yield_now().await;
        for _ in 0..3 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        assert!(healthy_calls.load(Ordering::SeqCst) >= 3);
        assert_eq!(offline_calls.load(Ordering::SeqCst), 1);
        assert!(contacts.0.load(Ordering::SeqCst) >= 3);

        stop.send(true).unwrap();
        tokio::time::advance(Duration::from_secs(60)).await;
        healthy_loop.await.unwrap().unwrap();
        offline_loop.await.unwrap().unwrap();
    }

    #[derive(Default)]
    struct AuthContacts {
        rejected: AtomicUsize,
        accepted: AtomicUsize,
    }

    impl ContactRecorder for AuthContacts {
        fn record(&self) -> Result<(), CliError> {
            Ok(())
        }

        fn rejected(&self) {
            self.rejected.fetch_add(1, Ordering::SeqCst);
        }

        fn accepted(&self) {
            self.accepted.fetch_add(1, Ordering::SeqCst);
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_rejected_credential_is_published_and_an_accepted_one_clears_it() {
        // The log was the only place a signed-out service said so, and it
        // started no runner for nine hours while `service status` and the TUI
        // both reported a healthy host.
        let mut rejected = ReconcileReport {
            failure: Some(RefreshState::Unauthorized),
            ..Default::default()
        };
        rejected.next_poll.delay = Duration::from_secs(1);
        let mut accepted = ReconcileReport {
            targets_read: 1,
            ..Default::default()
        };
        accepted.next_poll.delay = Duration::from_secs(1);

        for (report, rejections, acceptances) in [(rejected, true, false), (accepted, false, true)]
        {
            let (target, _calls) = FakeTarget::repeating(
                fixtures::policy()
                    .repository("acme/repo")
                    .autoscale("home", 1)
                    .active()
                    .build(),
                report,
            );
            let contacts = Arc::new(AuthContacts::default());
            let (stop, _) = tokio::sync::watch::channel(false);
            let daemon = tokio::spawn(run_target_loop(
                target,
                stop.subscribe(),
                never_upgraded(),
                Arc::clone(&contacts) as Arc<dyn ContactRecorder>,
            ));
            for _ in 0..3 {
                tokio::time::advance(Duration::from_secs(1)).await;
                tokio::task::yield_now().await;
            }
            assert_eq!(contacts.rejected.load(Ordering::SeqCst) > 0, rejections);
            assert_eq!(contacts.accepted.load(Ordering::SeqCst) > 0, acceptances);
            stop.send(true).unwrap();
            tokio::time::advance(Duration::from_secs(60)).await;
            let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;
        }
    }

    #[tokio::test(start_paused = true)]
    async fn every_pass_re_reads_the_policy_before_deciding_anything() {
        // Without this, a policy the operator changed governs nothing until the
        // service is restarted -- and the CLI has already told them it worked.
        // Watched on a real host: a drained policy went on starting runners for
        // three more cycles.
        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(1);
        let (target, _calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        let refreshes = Arc::clone(&target.refreshes);
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            contacts as Arc<dyn ContactRecorder>,
        ));

        for _ in 0..4 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        let seen = refreshes.load(Ordering::SeqCst);
        assert!(
            seen >= 2,
            "the loop must ask for a fresh policy set on every pass, not once at startup: {seen}"
        );

        stop.send(true).unwrap();
        tokio::time::advance(Duration::from_secs(60)).await;
        let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;
    }

    #[tokio::test(start_paused = true)]
    async fn a_drain_whose_target_never_reports_a_count_still_ends() {
        // The state a revoked credential puts every target into: `reconcile`
        // contributes no allocation, so `active_owned` answers `None` and the
        // graceful exit's `== Some(0)` is false on every pass, forever. Before
        // the deadline this loop never returned, and the service could only be
        // killed.
        let target = FakeTarget::never_readable(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
        );
        let calls = Arc::clone(&target.calls);
        let terminations = Arc::clone(&target.busy_was_terminated);
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            contacts as Arc<dyn ContactRecorder>,
        ));

        tokio::task::yield_now().await;
        stop.send(true).unwrap();

        // Well inside the deadline the daemon is still draining, which is what
        // keeps this from passing on a loop that simply exits at once.
        for _ in 0..5 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        assert!(
            !daemon.is_finished(),
            "the drain gave up before its deadline"
        );

        // Past it, it stops anyway.
        for _ in 0..DRAIN_DEADLINE.as_secs() {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        tokio::time::timeout(Duration::from_secs(1), daemon)
            .await
            .expect("an unreadable target must not hold the daemon open forever")
            .unwrap()
            .unwrap();
        assert!(calls.load(Ordering::SeqCst) >= 2, "the drain never polled");
        assert_eq!(
            terminations.load(Ordering::SeqCst),
            0,
            "the deadline must not terminate a runner; startup recovery adopts it"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn shutdown_loop_supervises_a_busy_child_to_completion_without_terminating_it() {
        let target = FakeTarget::busy_then_finished(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
        );
        let calls = Arc::clone(&target.calls);
        let terminations = Arc::clone(&target.busy_was_terminated);
        let contacts = Arc::new(CountingContacts::default());
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            contacts as Arc<dyn ContactRecorder>,
        ));

        tokio::task::yield_now().await;
        stop.send(true).unwrap();
        for _ in 0..3 {
            tokio::time::advance(Duration::from_secs(1)).await;
            tokio::task::yield_now().await;
        }
        tokio::time::timeout(Duration::from_secs(1), daemon)
            .await
            .expect("the daemon exits after supervised completion")
            .unwrap()
            .unwrap();
        assert!(
            calls.load(Ordering::SeqCst) >= 3,
            "shutdown skipped supervision"
        );
        assert_eq!(
            terminations.load(Ordering::SeqCst),
            0,
            "busy child was terminated"
        );
    }

    // -----------------------------------------------------------------------
    // The upgrade handover
    // -----------------------------------------------------------------------

    use runner_manager_platform::secrets::{
        PlatformSecretStore, Removal, SecretScope, SecretStoreError,
    };

    fn handover_document() -> String {
        format!(
            r#"{{"access_token":"{}","refresh_token":"{}","access_expires_at":"2026-09-06T20:00:00Z","refresh_expires_at":"2027-03-05T12:00:00Z"}}"#,
            format_args!("{}{}", "ghu_", "handoverAccessCanary0000000000"),
            format_args!("{}{}", "ghr_", "handoverRefreshCanary000000000"),
        )
    }

    fn handover_store(root: &std::path::Path) -> PlatformSecretStore {
        PlatformSecretStore::rooted_at(SecretScope::User, root).expect("a rooted store resolves")
    }

    fn held(store: &dyn SecretStore) -> Option<String> {
        store
            .load()
            .expect("the store is readable")
            .map(|secret| secret.expose_secret().to_string())
    }

    #[test]
    fn the_new_binary_is_handed_exactly_the_stored_credential() {
        let root = tempfile::tempdir().unwrap();
        let store = handover_store(root.path());
        store
            .store(&SecretString::from(handover_document()))
            .unwrap();

        let mut handed = None;
        let outcome = hand_over(&store, |document| {
            handed = Some(document.expose_secret().to_string());
            Ok(())
        });

        assert_eq!(outcome, Handover::HandedOver);
        let handed: serde_json::Value =
            serde_json::from_str(&handed.expect("the new binary was run")).unwrap();
        let original: serde_json::Value = serde_json::from_str(&handover_document()).unwrap();
        assert_eq!(
            handed, original,
            "both halves of the pair reach the new binary"
        );
    }

    /// A bare token is what versions before the envelope stored, and `daemon
    /// adopt-credential` refuses anything that is not the envelope.
    #[test]
    fn a_bare_token_is_handed_over_as_the_envelope() {
        let root = tempfile::tempdir().unwrap();
        let store = handover_store(root.path());
        let bare = format!("{}{}", "ghu_", "handoverBareToken000000000000");
        store.store(&SecretString::from(bare.clone())).unwrap();

        let mut handed = None;
        hand_over(&store, |document| {
            handed = Some(document.expose_secret().to_string());
            Ok(())
        });

        let handed: serde_json::Value =
            serde_json::from_str(&handed.expect("the new binary was run")).unwrap();
        assert_eq!(handed["access_token"], serde_json::Value::String(bare));
    }

    #[test]
    fn nothing_is_run_when_nothing_is_stored() {
        let root = tempfile::tempdir().unwrap();
        let store = handover_store(root.path());
        let outcome = hand_over(&store, |_| panic!("there was nothing to hand over"));
        assert_eq!(outcome, Handover::NothingStored);
    }

    #[test]
    fn nothing_is_run_when_this_daemon_cannot_read_its_own_credential() {
        let outcome = hand_over(&RefusingStore, |_| panic!("there was nothing to hand over"));
        assert_eq!(outcome, Handover::Unreadable);
    }

    /// Replacing an item the new binary may not read starts by deleting it, so
    /// a write that fails after that must not leave the host with nothing.
    #[test]
    fn a_credential_the_failed_attempt_removed_is_put_back() {
        let root = tempfile::tempdir().unwrap();
        let store = handover_store(root.path());
        store
            .store(&SecretString::from(handover_document()))
            .unwrap();

        let outcome = hand_over(&store, |_| {
            store.delete().unwrap();
            Err("the keychain refused the write".to_string())
        });

        assert_eq!(outcome, Handover::Failed { restored: true });
        let restored: serde_json::Value =
            serde_json::from_str(&held(&store).expect("the credential is back")).unwrap();
        let original: serde_json::Value = serde_json::from_str(&handover_document()).unwrap();
        assert_eq!(restored, original);
    }

    /// An item that is still there is either the original or the new
    /// binary's, and neither is this daemon's to overwrite.
    #[test]
    fn a_credential_still_in_place_after_a_failure_is_left_alone() {
        let root = tempfile::tempdir().unwrap();
        let store = handover_store(root.path());
        store
            .store(&SecretString::from(handover_document()))
            .unwrap();
        let replacement = format!("{}{}", "ghu_", "handoverWrittenByTheNewBuild00");

        let outcome = hand_over(&store, |_| {
            store
                .store(&SecretString::from(replacement.clone()))
                .unwrap();
            Err("the new binary exited non-zero after writing".to_string())
        });

        assert_eq!(outcome, Handover::Failed { restored: false });
        assert_eq!(held(&store), Some(replacement));
    }

    /// A store whose every read is refused, as a keychain refuses a program it
    /// does not grant the item to.
    #[derive(Debug)]
    struct RefusingStore;

    impl RefusingStore {
        fn refusal() -> std::io::Error {
            std::io::Error::new(std::io::ErrorKind::PermissionDenied, "refused")
        }
    }

    impl SecretStore for RefusingStore {
        fn scope(&self) -> SecretScope {
            SecretScope::User
        }

        fn location(&self) -> String {
            "a store that refuses".to_string()
        }

        fn store(&self, _secret: &SecretString) -> Result<(), SecretStoreError> {
            panic!("the handover must not write a store it could not read")
        }

        fn ensure_writable(&self) -> Result<(), SecretStoreError> {
            Ok(())
        }

        fn load(&self) -> Result<Option<SecretString>, SecretStoreError> {
            Err(SecretStoreError::Load {
                scope: self.scope(),
                location: self.location(),
                source: Self::refusal(),
            })
        }

        fn delete(&self) -> Result<Removal, SecretStoreError> {
            panic!("the handover never deletes")
        }

        fn protection(
            &self,
        ) -> Result<runner_manager_platform::secrets::Protection, SecretStoreError> {
            panic!("not asked")
        }
    }
}