runner-manager 0.4.52

// 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, SharedPackages,
    TokioRetryDelay, TracingAttemptEvents,
};
use runner_manager_agent::package::{
    CachePorts, ExponentialBackoff, GatewayCatalog, HttpFetcher, PackageCache,
};
use runner_manager_agent::reconcile::{
    AllocationLock, DemandDirector, 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 _updates = super::auto_update::Worker::start(context.clone());
            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)
        .map(|path| super::update::restart_path(&path));

    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() {
        // Still linked, so the coordinator sees this host; it serves nothing,
        // so nothing is ever woken.
        let coordination = super::coordinator::coordination();
        let launches = Arc::new(super::launches::Launches::new(
            Arc::clone(&store) as Arc<dyn Store>,
            host.clone(),
            coordination.subscribe(),
        ));
        let _link = start_coordinator_link(
            context,
            &host,
            Arc::clone(&store) as Arc<dyn Store>,
            Arc::new(super::coordinator::WakeRoutes::default()),
            (
                Arc::new(AtomicBool::new(false)),
                Arc::new(AtomicBool::new(false)),
            ),
            (coordination, launches),
        );
        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_defaults(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,
    ));
    // The coordinator's wakes, one signal per target loop. Inert until this
    // host joins a coordinator; see `super::coordinator`.
    let routes = super::coordinator::WakeRoutes::new(Some(Arc::clone(&demand_gateway)));
    // What the link knows, for each loop's poll mode (standalone, coordinated,
    // fallback).
    let coordination = super::coordinator::coordination();
    // The coordinator's launches (stage 3): accepted by the link, started by
    // the target loops' passes for their pool profiles.
    let launches = Arc::new(super::launches::Launches::new(
        Arc::clone(&store) as Arc<dyn Store>,
        host.clone(),
        coordination.subscribe(),
    ));
    let routes = Arc::new(routes);
    // ------------------------------------------------------------------
    // ONE LOOP PER TARGET, ADDED AND REMOVED WHILE THE DAEMON RUNS.
    // ------------------------------------------------------------------
    // A repository added or removed changes the set of loops, not the rest
    // of the daemon: the coordinator link, its accepted launches and every
    // other target's loop keep running. See `follow_policy_set`.
    // The targets of this daemon share one package directory and one bound on
    // how many runners prepare at once.
    let shared_packages = SharedPackages::default();
    let preparation_turns = runner_manager_agent::lifecycle::preparation_turns();
    let build_target = |policies: Vec<ScalePolicy>| {
        let shared_packages = shared_packages.clone();
        let preparation_turns = Arc::clone(&preparation_turns);
        let target_paths = context.paths().clone();
        let client = Arc::clone(&client);
        let clock = Arc::clone(&clock);
        let host = host.clone();
        let store = Arc::clone(&store);
        let lifecycle_github = Arc::clone(&lifecycle_github);
        let dependency_caches = dependency_caches.clone();
        let events = Arc::clone(&events);
        let host_unfit = Arc::clone(&host_unfit);
        let shared_lock = Arc::clone(&shared_lock);
        let directory = Arc::clone(&directory);
        let demand_gateway = Arc::clone(&demand_gateway);
        let polling = Arc::clone(&polling);
        let launches = Arc::clone(&launches);
        let coordination = Arc::clone(&coordination);
        let routes = Arc::clone(&routes);
        async move {
            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(
                    &target_paths,
                    host.os,
                    host.architecture,
                    CachePorts {
                        catalog,
                        fetcher: Arc::new(HttpFetcher::default()),
                        backoff: Arc::new(ExponentialBackoff::default()),
                        clock: Arc::clone(&clock),
                    },
                )
                .with_last_check(shared_packages.last_check()),
            );
            let lifecycle_store = Arc::new(TargetRecoveryStore::new(
                Arc::clone(&store) as Arc<dyn Store>,
                &policies,
            ));
            let launcher = Arc::new(
                LifecycleLauncher::new(
                    host.id,
                    target_paths.clone(),
                    target_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, shared_packages)),
                        processes: Arc::new(
                            PlatformExecutionProvider::new(host.id)
                                .with_runner_env_file(runner_manager_platform::runner_env::path_in(
                                    target_paths.config_dir(),
                                ))
                                .with_dependency_caches(dependency_caches.clone())
                                .with_docker_cleanup(
                                    runner_manager_agent::docker_cleanup::DockerCleanup::new(
                                        host.id,
                                        target_paths.config_dir(),
                                        target_paths.state_dir(),
                                    ),
                                ),
                        ),
                        clock: Arc::clone(&clock),
                        demand: Arc::new(PersistentDemand),
                        delay: Arc::new(TokioRetryDelay),
                        events: Arc::new(TracingAttemptEvents),
                        reconcile_events: Arc::clone(&events),
                    },
                )
                .with_preparation_turns(preparation_turns),
            );
            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(),
            ));
            let signals = routes.add(package_target.clone());
            let ports = 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),
            };
            let director = Arc::clone(&launches) as Arc<dyn DemandDirector>;
            Ok(ManagedTarget {
                policies,
                store: Arc::clone(&store) as Arc<dyn Store>,
                launch_side: Some(
                    Reconciler::new(host.clone(), ports.clone())
                        .with_director(Arc::clone(&director)),
                ),
                reconciler: Reconciler::new(host.clone(), ports)
                    .with_director(director)
                    .with_launch_side(Arc::clone(&signals.launch)),
                cancel,
                polling: Arc::clone(&polling),
                wake: signals.wake,
                launch: signals.launch,
                launches,
                coordination: coordination.subscribe(),
                draining: false,
                pacing_floor: Duration::ZERO,
            })
        }
    };
    let mut managed_targets = Vec::with_capacity(targets.len());
    for policies in targets {
        managed_targets.push(build_target(policies).await?);
    }

    let draining = Arc::new(AtomicBool::new(false));
    let _link = start_coordinator_link(
        context,
        &host,
        Arc::clone(&store) as Arc<dyn Store>,
        Arc::clone(&routes),
        (Arc::clone(&host_unfit), Arc::clone(&draining)),
        (Arc::clone(&coordination), Arc::clone(&launches)),
    );

    let (shutdown, _) = 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),
    });
    // The loops running, by target: each with its own drain signal, which an
    // upgrade or a reload sends to all and a removed target to its own.
    let mut serving = Serving::default();
    for target in managed_targets {
        serving.spawn(&mut loops, target, &shutdown, &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
        }
        end = follow_policy_set(
            Arc::clone(&store) as Arc<dyn Store>,
            &build_target,
            &mut serving,
            &mut loops,
            (&shutdown, &contacts, &routes),
        ) => match end {
            // Nothing left to serve: the reload starts the idle daemon.
            None => {
                writeln!(
                    out,
                    "this host no longer serves any repository; finishing every running job before reloading"
                )
                .map_err(failed)?;
                restart_reason = Some("this host no longer serves any repository");
                None
            }
            ended => ended,
        },
        () = 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")
        }
    };
    // Told to the coordinator on the next heartbeat; the link stays up while
    // the loops drain.
    draining.store(true, Ordering::Relaxed);
    if upgraded_to.is_some() || restart_reason.is_some() {
        serving.drain_all();
        // 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)
}

/// Aborts the task it holds when dropped, so the coordinator link ends with
/// the generation that started it.
struct AbortOnDrop(tokio::task::JoinHandle<()>);

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Starts the coordinator link for this generation. It idles until the host
/// joins a coordinator, so it is started either way.
fn start_coordinator_link(
    context: &Context,
    host: &runner_manager_domain::model::Host,
    store: Arc<dyn Store>,
    routes: Arc<super::coordinator::WakeRoutes>,
    (unfit, draining): (Arc<AtomicBool>, Arc<AtomicBool>),
    (coordination, launches): (
        Arc<super::coordinator::Coordination>,
        Arc<super::launches::Launches>,
    ),
) -> Option<AbortOnDrop> {
    let key_store = match context.secret_store_for(
        host.service_start_mode,
        runner_manager_platform::secrets::SecretItem::CoordinatorKey,
    ) {
        Ok(store) => store,
        Err(error) => {
            tracing::warn!(%error, "no coordinator link: the secret store cannot be reached");
            return None;
        }
    };
    let runner_root = super::workspace::host_root(context.paths(), Some(host))
        .effective_text()
        .map_or_else(
            || context.paths().state_dir().to_path_buf(),
            std::path::PathBuf::from,
        );
    Some(AbortOnDrop(tokio::spawn(
        super::coordinator::maintain_link(super::coordinator::LinkPorts {
            paths: context.paths().clone(),
            key_store,
            store,
            routes,
            unfit,
            draining,
            host: host.clone(),
            runner_root,
            coordination,
            launches,
        }),
    )))
}

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.
pub(crate) 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.
pub(crate) 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(())
}

type LoopResult = Result<Result<(), CliError>, tokio::task::JoinError>;

/// The target loops running, by target.
#[derive(Default)]
struct Serving {
    loops: std::collections::BTreeMap<String, ServedTarget>,
    /// Loops of removed targets, draining their own runners away.
    retiring: std::collections::HashMap<tokio::task::Id, String>,
}

struct ServedTarget {
    target: ScaleTarget,
    drain: tokio::sync::watch::Sender<bool>,
    task: tokio::task::Id,
}

impl Serving {
    fn spawn(
        &mut self,
        loops: &mut tokio::task::JoinSet<Result<(), CliError>>,
        target: impl TargetReconciler,
        shutdown: &tokio::sync::watch::Sender<bool>,
        contacts: &Arc<dyn ContactRecorder>,
    ) {
        let Some(name) = target
            .policies()
            .first()
            .map(|policy| policy.target.clone())
        else {
            return;
        };
        let (drain, drained) = tokio::sync::watch::channel(false);
        let task = loops
            .spawn(run_target_loop(
                target,
                shutdown.subscribe(),
                drained,
                Arc::clone(contacts),
            ))
            .id();
        self.loops.insert(
            name.to_string(),
            ServedTarget {
                target: name,
                drain,
                task,
            },
        );
    }

    fn served(&self) -> BTreeSet<String> {
        self.loops.keys().cloned().collect()
    }

    /// An upgrade or a reload: every loop finishes its runners and stops.
    fn drain_all(&self) {
        for served in self.loops.values() {
            let _ = served.drain.send(true);
        }
    }

    /// `name` is no longer served: its loop drains away on its own.
    fn retire(&mut self, name: &str) -> Option<ScaleTarget> {
        let served = self.loops.remove(name)?;
        let _ = served.drain.send(true);
        self.retiring.insert(served.task, name.to_string());
        Some(served.target)
    }

    fn is_retiring(&self, name: &str) -> bool {
        self.retiring.values().any(|retiring| retiring == name)
    }
}

/// Follows the set of targets this daemon serves while it runs: a target
/// added gets a loop of its own (after its startup recovery), a target
/// removed has its loop drain its own runners and stop. Every other loop,
/// the coordinator link and its accepted launches are left alone; the link
/// describes the new set at its next heartbeat.
///
/// Returns when a loop that was not retired stopped (its result), or `None`
/// when nothing is left to serve, for the reload into the idle daemon.
async fn follow_policy_set<F, Fut, T>(
    store: Arc<dyn Store>,
    build_target: &F,
    serving: &mut Serving,
    loops: &mut tokio::task::JoinSet<Result<(), CliError>>,
    (shutdown, contacts, routes): (
        &tokio::sync::watch::Sender<bool>,
        &Arc<dyn ContactRecorder>,
        &super::coordinator::WakeRoutes,
    ),
) -> Option<LoopResult>
where
    F: Fn(Vec<ScalePolicy>) -> Fut,
    Fut: Future<Output = Result<T, CliError>>,
    T: TargetReconciler,
{
    loop {
        tokio::select! {
            groups = wait_for_policy_set_change(Arc::clone(&store), serving.served()) => {
                if groups.is_empty() {
                    return None;
                }
                let current: BTreeSet<String> =
                    groups.iter().map(|group| group[0].target.to_string()).collect();
                for name in serving.served().difference(&current) {
                    if let Some(target) = serving.retire(name) {
                        routes.remove(&target);
                        tracing::info!(%target, "a target is no longer served; its loop finishes its runners and stops");
                    }
                }
                for group in groups {
                    let target = group[0].target.clone();
                    let name = target.to_string();
                    // A removed target added back waits for its old loop.
                    if serving.loops.contains_key(&name) || serving.is_retiring(&name) {
                        continue;
                    }
                    match build_target(group).await {
                        Ok(managed) => {
                            serving.spawn(loops, managed, shutdown, contacts);
                            tracing::info!(%target, "a target was added; its loop runs");
                        }
                        Err(error) => {
                            tracing::warn!(%target, %error, "a target that was added could not start; retrying shortly");
                        }
                    }
                }
            }
            Some(joined) = loops.join_next_with_id(), if !loops.is_empty() => {
                let id = match &joined {
                    Ok((id, _)) => *id,
                    Err(error) => error.id(),
                };
                if let Some(target) = serving.retiring.remove(&id) {
                    if let Ok((_, Err(error))) = &joined {
                        tracing::warn!(%target, %error, "a removed target's loop stopped with an error");
                    }
                    continue;
                }
                return Some(joined.map(|(_, result)| result));
            }
        }
    }
}

/// 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 needs a loop built for it, and one *removed* leaves a loop with
/// nothing to serve: [`follow_policy_set`] does both, one target at a time.
/// Returns the new set, grouped by target as [`active_autoscale_targets`]
/// groups it.
async fn wait_for_policy_set_change(
    store: Arc<dyn Store>,
    initial: BTreeSet<String>,
) -> Vec<Vec<ScalePolicy>> {
    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 groups = active_autoscale_targets(policies);
        let current: BTreeSet<String> = groups
            .iter()
            .map(|group| group[0].target.to_string())
            .collect();
        if current != initial {
            return groups;
        }
    }
}

/// 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
}

/// The targets among `policies` that poll on their own schedule right now:
/// ones with a loop, and not `coordinated`.
fn polling_on_their_own(
    policies: &[ScalePolicy],
    view: &super::coordinator::LinkView,
    now: tokio::time::Instant,
) -> Vec<ScaleTarget> {
    policies
        .iter()
        .filter(|policy| policy.may_start_runners() || policy.state() == PolicyState::Draining)
        .filter(|policy| {
            view.mode(&policy.target, now) != super::coordinator::PollMode::Coordinated
        })
        .map(|policy| policy.target.clone())
        .collect()
}

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) {}

    /// The coordinator's "poll this target now" signal, when the daemon
    /// routes one to this loop. See [`super::coordinator`].
    fn wake(&self) -> Option<Arc<tokio::sync::Notify>> {
        None
    }

    /// A coordinator launch of this target is open with no runner yet.
    fn holds_unstarted_launches(&self) -> bool {
        false
    }

    /// The least time between this target's charged polls while it is not
    /// coordinated. See `super::host::pacing_floor`.
    fn pacing_floor(&self) -> Duration {
        Duration::ZERO
    }

    /// What serves this target's coordinator launches, beside its passes, until
    /// `stop`. See `Reconciler::serve_launches`. `None` where the daemon has no
    /// link to bring any, and on a second call.
    fn launch_driver(
        &mut self,
        _stop: tokio::sync::watch::Receiver<bool>,
    ) -> Option<Pin<Box<dyn Future<Output = ()> + Send + 'static>>> {
        None
    }

    /// The link's view, for this target's poll mode. `None` where no link
    /// is wired.
    fn coordination(&self) -> Option<tokio::sync::watch::Receiver<super::coordinator::LinkView>> {
        None
    }

    /// The interval a `coordinated` target is polled at.
    fn coordinated_interval(&self) -> Duration {
        runner_manager_domain::model::CoordinatedPollInterval::default().as_duration()
    }

    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,
    /// Over the same ports, for [`Reconciler::serve_launches`]: taken when the
    /// loop starts, and run beside `reconciler`'s passes.
    launch_side: Option<Reconciler>,
    cancel: CancelToken,
    store: Arc<dyn Store>,
    polling: Arc<super::polling::PollingMonitor>,
    wake: Arc<tokio::sync::Notify>,
    launch: Arc<tokio::sync::Notify>,
    launches: Arc<super::launches::Launches>,
    coordination: tokio::sync::watch::Receiver<super::coordinator::LinkView>,
    /// Draining: a policy read afresh is drained too, so a repository added
    /// back while this loop winds down is not served by it.
    draining: bool,
    /// From the policies last read: see `super::host::pacing_floor`.
    pacing_floor: Duration,
}

impl TargetReconciler for ManagedTarget {
    fn wake(&self) -> Option<Arc<tokio::sync::Notify>> {
        Some(Arc::clone(&self.wake))
    }

    fn pacing_floor(&self) -> Duration {
        self.pacing_floor
    }

    fn holds_unstarted_launches(&self) -> bool {
        self.launches.holds_unstarted(&self.policies)
    }

    fn launch_driver(
        &mut self,
        stop: tokio::sync::watch::Receiver<bool>,
    ) -> Option<Pin<Box<dyn Future<Output = ()> + Send + 'static>>> {
        let launch_side = self.launch_side.take()?;
        let signal = Arc::clone(&self.launch);
        let store = Arc::clone(&self.store);
        let target = self.policies.first()?.target.clone();
        Some(Box::pin(async move {
            // Read at each launch, so a profile an operator disabled a moment
            // ago starts nothing. One that cannot be read starts nothing now;
            // the launch stays open for the next pass.
            let policies = move || -> Vec<ScalePolicy> {
                store.policies().map_or_else(
                    |_| Vec::new(),
                    |all| {
                        all.into_iter()
                            .filter(|policy| policy.target == target)
                            .collect()
                    },
                )
            };
            launch_side.serve_launches(&policies, &signal, stop).await;
        }))
    }

    fn coordination(&self) -> Option<tokio::sync::watch::Receiver<super::coordinator::LinkView>> {
        Some(self.coordination.clone())
    }

    fn coordinated_interval(&self) -> Duration {
        self.polling.intervals().coordinated.as_duration()
    }

    fn policies(&self) -> &[ScalePolicy] {
        &self.policies
    }

    fn begin_drain(&mut self) {
        self.draining = true;
        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;
        };
        self.pacing_floor = {
            let view = self.coordination.borrow();
            super::host::pacing_floor(&polling_on_their_own(
                &all,
                &view,
                tokio::time::Instant::now(),
            ))
        };
        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;
            if self.draining {
                begin_drain(&mut self.policies);
            }
        }
    }

    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,
}

/// A target's passes, and beside them what serves its coordinator launches, so
/// a launch does not wait for a pass to finish. Launches stop being started when
/// the target drains, and the loop ends once those coming up are up.
async fn run_target_loop<T: TargetReconciler>(
    mut target: T,
    shutdown: tokio::sync::watch::Receiver<bool>,
    upgrade: tokio::sync::watch::Receiver<bool>,
    contacts: Arc<dyn ContactRecorder>,
) -> Result<(), CliError> {
    let (stop_launches, launches_stopped) = tokio::sync::watch::channel(false);
    let launch_driver = target.launch_driver(launches_stopped);
    let passes = run_target_passes(target, shutdown, upgrade, contacts, stop_launches);
    match launch_driver {
        Some(driver) => futures::future::join(passes, driver).await.0,
        None => passes.await,
    }
}

async fn run_target_passes<T: TargetReconciler>(
    mut target: T,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
    mut upgrade: tokio::sync::watch::Receiver<bool>,
    contacts: Arc<dyn ContactRecorder>,
    stop_launches: tokio::sync::watch::Sender<bool>,
) -> 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() {
        stop_launches.send_replace(true);
        target.begin_drain();
    }
    let wake = target.wake();
    let mut coordination = target.coordination();
    let target_name = target
        .policies()
        .first()
        .map(|policy| policy.target.clone());
    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(());
        }
        let polled_at = tokio::time::Instant::now();
        // A wake stored before this pass (one that arrived inside a back-off
        // or a drain) is answered by this very poll; left in place it would
        // pull the next poll forward for a job already seen.
        if let Some(wake) = &wake {
            let _ = futures::FutureExt::now_or_never(wake.notified());
        }
        // 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()))
        });
        let slept_at = tokio::time::Instant::now();
        let scheduled = slept_at + delay;
        // Targets polling on their own (not coordinated) may not fit the
        // budget together: charged polls stretch to what does.
        // Idle polls are answered 304, which is free, and a drain is bounded.
        let paced = if draining.is_some()
            || report.next_poll.pace == runner_manager_agent::reconcile::PollPace::Idle
        {
            scheduled
        } else {
            slept_at + delay.max(target.pacing_floor())
        };
        // Poll mode `coordinated`: the coordinator's sweep reads this target's
        // queue and wakes it, so its own poll stretches to the coordinated
        // interval. Never while draining, and never while this host runs a
        // runner for the target: a finished job frees a slot that queued work
        // the coordinator already announced may need now.
        let stretched = (coordination.is_some()
            && draining.is_none()
            && target.local_active() == Some(0)
            && !target.holds_unstarted_launches())
        .then(|| slept_at + delay.max(target.coordinated_interval()));
        // A coordinator's wake brings the next poll forward -- never while
        // draining, never while the schedule is a slowdown GitHub asked for
        // (rate limited, stretched, offline...), and never sooner than
        // `WAKE_MIN_GAP` after this poll began.
        let honour_wakes = draining.is_none() && !report.next_poll.pace.is_throttled();
        let wake = wake.as_deref().filter(|_| honour_wakes);
        let mut woken: Option<tokio::time::Instant> = None;
        loop {
            // Re-decided on every turn, so a link that drops or a coverage
            // that changes re-schedules a loop already asleep.
            let now = tokio::time::Instant::now();
            let (planned, fallback_at) = match (&mut coordination, &target_name) {
                (Some(view), Some(name)) => {
                    let view = view.borrow_and_update();
                    match stretched {
                        Some(stretched)
                            if view.mode(name, now)
                                == super::coordinator::PollMode::Coordinated =>
                        {
                            (stretched, view.fallback_at())
                        }
                        _ => (paced, None),
                    }
                }
                _ => (paced, None),
            };
            let deadline = woken.map_or(planned, |woken| woken.min(planned));
            tokio::select! {
                () = tokio::time::sleep_until(deadline) => break,
                () = async {
                    match wake {
                        Some(wake) => wake.notified().await,
                        None => std::future::pending().await,
                    }
                } => {
                    woken = Some(super::coordinator::woken_deadline(
                        deadline,
                        polled_at,
                        tokio::time::Instant::now(),
                    ));
                }
                changed = async {
                    match coordination.as_mut() {
                        Some(view) => view.changed().await,
                        None => std::future::pending().await,
                    }
                } => {
                    if changed.is_err() {
                        // The link is gone with its generation: poll on our own.
                        coordination = None;
                    }
                }
                () = async {
                    match fallback_at {
                        Some(at) => tokio::time::sleep_until(at).await,
                        None => std::future::pending().await,
                    }
                } => {}
                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.
                        stop_launches.send_replace(true);
                        target.begin_drain();
                        draining = Some(DrainKind::Shutdown);
                        drain_deadline = Some(tokio::time::Instant::now() + DRAIN_DEADLINE);
                    }
                    break;
                }
                changed = upgrade.changed(), if draining.is_none() => {
                    if changed.is_err() || *upgrade.borrow() {
                        stop_launches.send_replace(true);
                        target.begin_drain();
                        draining = Some(DrainKind::Upgrade);
                    }
                    break;
                }
            }
        }
    }
}

/// 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, 'life2, 'async_trait>(
            &'life0 self,
            _attempt: &'life1 RunnerAttempt,
            _timings: &'life2 mut runner_manager_agent::lifecycle::PackageTimings,
        ) -> Pin<Box<dyn Future<Output = Result<RunnerVersion, FailureReason>> + Send + 'async_trait>>
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: '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<'_>,
            _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>,
        /// The coordinator's wake signal, when a test routes one.
        wake: Option<Arc<tokio::sync::Notify>>,
        /// The link's view and the coordinated interval, when a test wires
        /// one.
        coordination: Option<tokio::sync::watch::Receiver<super::super::coordinator::LinkView>>,
        /// The link's launch signal, and how often accepted launches were
        /// started, when a test wires one.
        launch: Option<(Arc<tokio::sync::Notify>, Arc<AtomicUsize>)>,
        /// Each pass takes a permit first, when a test wires this.
        hold_pass: Option<Arc<tokio::sync::Semaphore>>,
        /// What the budget says about this target's charged polls.
        pacing_floor: Duration,
        /// A launch of this target stays open with no runner.
        unstarted: bool,
    }

    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)),
                    wake: None,
                    coordination: None,
                    launch: None,
                    hold_pass: None,
                    pacing_floor: Duration::ZERO,
                    unstarted: false,
                },
                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)),
                wake: None,
                coordination: None,
                launch: None,
                hold_pass: None,
                pacing_floor: Duration::ZERO,
                unstarted: false,
            }
        }

        /// 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)),
                wake: None,
                coordination: None,
                launch: None,
                hold_pass: None,
                pacing_floor: Duration::ZERO,
                unstarted: false,
            }
        }
    }

    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
            });
            let hold = self.hold_pass.clone();
            Box::pin(async move {
                if let Some(hold) = hold {
                    hold.acquire().await.expect("never closed").forget();
                }
                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)
        }

        fn wake(&self) -> Option<Arc<tokio::sync::Notify>> {
            self.wake.clone()
        }

        fn coordination(
            &self,
        ) -> Option<tokio::sync::watch::Receiver<super::super::coordinator::LinkView>> {
            self.coordination.clone()
        }

        fn pacing_floor(&self) -> Duration {
            self.pacing_floor
        }

        fn holds_unstarted_launches(&self) -> bool {
            self.unstarted
        }

        fn launch_driver(
            &mut self,
            mut stop: tokio::sync::watch::Receiver<bool>,
        ) -> Option<Pin<Box<dyn Future<Output = ()> + Send + 'static>>> {
            let (signal, started) = self.launch.take()?;
            Some(Box::pin(async move {
                loop {
                    tokio::select! {
                        () = signal.notified() => {
                            started.fetch_add(1, Ordering::SeqCst);
                        }
                        changed = stop.changed() => {
                            if changed.is_err() || *stop.borrow() {
                                return;
                            }
                        }
                    }
                }
            }))
        }
    }

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

    /// A target loop polling every 60 s, with a coordinator wake routed to
    /// it, and the counter of its polls.
    type WokenLoop = (
        tokio::task::JoinHandle<Result<(), CliError>>,
        Arc<AtomicUsize>,
        Arc<tokio::sync::Notify>,
        tokio::sync::watch::Sender<bool>,
    );

    fn woken_loop(pace: runner_manager_agent::reconcile::PollPace) -> WokenLoop {
        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        report.next_poll.pace = pace;
        let (mut target, calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        let wake = Arc::new(tokio::sync::Notify::new());
        target.wake = Some(Arc::clone(&wake));
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        (daemon, calls, wake, stop)
    }

    async fn after(secs: u64) {
        tokio::time::advance(Duration::from_secs(secs)).await;
        for _ in 0..8 {
            tokio::task::yield_now().await;
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_wake_polls_early_but_never_within_five_seconds_of_the_last_poll() {
        let (daemon, calls, wake, stop) =
            woken_loop(runner_manager_agent::reconcile::PollPace::Nominal);
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "the first pass");

        // Woken one second after the poll: held until five seconds.
        after(1).await;
        wake.notify_one();
        after(3).await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "too soon after the last poll"
        );
        after(1).await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "polled at five seconds, not sixty"
        );

        // A burst of wakes right after that poll is one more poll, five
        // seconds later.
        wake.notify_one();
        after(1).await;
        wake.notify_one();
        after(3).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        after(1).await;
        assert_eq!(calls.load(Ordering::SeqCst), 3);

        // With no wake, the schedule is the schedule.
        after(59).await;
        assert_eq!(calls.load(Ordering::SeqCst), 3);
        after(1).await;
        assert_eq!(calls.load(Ordering::SeqCst), 4);

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

    /// A target set over the allowance stretches its charged polls while it
    /// polls on its own; idle polls, answered 304 and free, keep their pace.
    #[tokio::test(start_paused = true)]
    async fn charged_polls_stretch_to_the_budget_and_idle_ones_do_not() {
        use runner_manager_agent::reconcile::PollPace;

        for (pace, delay, expected) in [(PollPace::Nominal, 60, 90), (PollPace::Idle, 10, 10)] {
            let mut report = ReconcileReport::default();
            report.next_poll.delay = Duration::from_secs(delay);
            report.next_poll.pace = pace;
            let (mut target, calls) = FakeTarget::repeating(
                fixtures::policy()
                    .repository("acme/repo")
                    .autoscale("home", 1)
                    .active()
                    .build(),
                report,
            );
            target.pacing_floor = Duration::from_secs(90);
            let (stop, _) = tokio::sync::watch::channel(false);
            let daemon = tokio::spawn(run_target_loop(
                target,
                stop.subscribe(),
                never_upgraded(),
                Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
            ));
            after(0).await;
            assert_eq!(calls.load(Ordering::SeqCst), 1, "the first pass");
            after(expected - 1).await;
            assert_eq!(calls.load(Ordering::SeqCst), 1, "{pace:?}: not yet");
            after(1).await;
            assert_eq!(calls.load(Ordering::SeqCst), 2, "{pace:?}: at {expected} s");
            stop.send(true).unwrap();
            after(120).await;
            let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;
        }
    }

    /// A coordinated target with nothing running sleeps its long interval,
    /// unless a launch of its own has no runner: its lease would run out first.
    /// A launch that does not start is retried by a pass within 30 seconds.
    #[tokio::test(start_paused = true)]
    async fn a_launch_with_no_runner_keeps_its_target_from_the_long_sleep() {
        use super::super::coordinator::LinkView;
        use runner_manager_domain::coordinator_protocol::{HelloAck, TargetRef, TargetScope};

        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        report.next_poll.pace = runner_manager_agent::reconcile::PollPace::Nominal;
        let (mut target, calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        target.active = 0;
        target.unstarted = true;
        let signal = Arc::new(tokio::sync::Notify::new());
        target.launch = Some((Arc::clone(&signal), Arc::new(AtomicUsize::new(0))));
        let (_link, view) = tokio::sync::watch::channel(LinkView {
            configured: true,
            connected: true,
            down_since: None,
            ack: Some(HelloAck {
                reconcile_interval_secs: Some(120),
                covered: vec![TargetRef {
                    scope: TargetScope::Repo,
                    name: "acme/repo".into(),
                }],
                any_pool: true,
            }),
            version: Some(4),
        });
        target.coordination = Some(view);
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "the first pass");
        // Accepted, and still not started: the 60 s schedule, not the 300 s one.
        signal.notify_one();
        after(59).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        after(1).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2, "at 60 s");
        stop.send(true).unwrap();
        after(120).await;
        let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;
    }

    /// The documented loss of coordination: a target that is not coordinated
    /// paces its charged polls to the budget; one coordinated waits for its
    /// own interval, and the floor counts only the targets polling on their
    /// own.
    #[tokio::test(start_paused = true)]
    async fn a_target_not_coordinated_is_paced_and_only_those_count_towards_the_floor() {
        use super::super::coordinator::LinkView;
        use runner_manager_domain::coordinator_protocol::{HelloAck, TargetRef, TargetScope};

        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        report.next_poll.pace = runner_manager_agent::reconcile::PollPace::Nominal;
        let policy = fixtures::policy()
            .repository("acme/repo")
            .autoscale("home", 1)
            .active()
            .build();
        let other = fixtures::policy()
            .id(runner_manager_domain::model::PolicyId::new_random())
            .repository("acme/other")
            .autoscale("home", 1)
            .active()
            .build();
        let (mut target, calls) = FakeTarget::repeating(policy.clone(), report);
        target.active = 0;
        target.pacing_floor = Duration::from_secs(90);
        // Joined, the link down well past the grace: fallback.
        let down = LinkView {
            configured: true,
            connected: false,
            down_since: tokio::time::Instant::now().checked_sub(Duration::from_secs(600)),
            ack: None,
            version: Some(4),
        };
        let (_link, view) = tokio::sync::watch::channel(down.clone());
        target.coordination = Some(view);
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "the first pass");
        after(89).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "not at 60 s");
        after(1).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2, "at the floor");
        stop.send(true).unwrap();
        after(120).await;
        let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;

        // The floor counts what polls on its own: both while the link is
        // down; only the uncovered one while it is up.
        let now = tokio::time::Instant::now();
        let policies = [policy.clone(), other.clone()];
        assert_eq!(polling_on_their_own(&policies, &down, now).len(), 2);
        let up = LinkView {
            configured: true,
            connected: true,
            down_since: None,
            ack: Some(HelloAck {
                reconcile_interval_secs: Some(120),
                covered: vec![TargetRef {
                    scope: TargetScope::Repo,
                    name: "acme/repo".into(),
                }],
                any_pool: true,
            }),
            version: Some(4),
        };
        assert_eq!(
            polling_on_their_own(&policies, &up, now),
            vec![other.target.clone()]
        );
    }

    /// Accepted launches start their runners at once, however recently the
    /// target was polled and however far its next poll is; a burst of them
    /// is served together, and none of it is a poll.
    #[tokio::test(start_paused = true)]
    async fn accepted_launches_start_at_once_together_and_poll_nothing() {
        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        let (mut target, calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        let signal = Arc::new(tokio::sync::Notify::new());
        let started = Arc::new(AtomicUsize::new(0));
        target.launch = Some((Arc::clone(&signal), Arc::clone(&started)));
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "the first pass");

        // Accepted one second after the poll: started then, not at five.
        after(1).await;
        signal.notify_one();
        after(0).await;
        assert_eq!(started.load(Ordering::SeqCst), 1);
        // Six accepted at once: the start they woke and at most one more
        // serve them all (each start serves every open launch).
        for _ in 0..6 {
            signal.notify_one();
        }
        after(0).await;
        assert_eq!(started.load(Ordering::SeqCst), 3);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "no poll was brought forward"
        );
        after(58).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        after(1).await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "the schedule is the schedule"
        );

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

    /// A launch does not wait for its target's pass: one running for as long as
    /// it likes, polling GitHub or bringing up a batch, does not hold it back.
    /// Draining stops launches, and the loop ends once they have stopped.
    #[tokio::test(start_paused = true)]
    async fn a_launch_is_served_while_a_pass_runs_and_a_drain_stops_serving() {
        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        let (mut target, calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        target.active = 0;
        let signal = Arc::new(tokio::sync::Notify::new());
        let started = Arc::new(AtomicUsize::new(0));
        target.launch = Some((Arc::clone(&signal), Arc::clone(&started)));
        let pass = Arc::new(tokio::sync::Semaphore::new(0));
        target.hold_pass = Some(Arc::clone(&pass));
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1, "in its first pass");

        signal.notify_one();
        after(600).await;
        assert_eq!(started.load(Ordering::SeqCst), 1, "served during it");
        assert_eq!(calls.load(Ordering::SeqCst), 1, "still in it");

        pass.add_permits(100);
        stop.send(true).unwrap();
        let ended = tokio::time::timeout(Duration::from_secs(60), daemon)
            .await
            .expect("the loop ends, the launch side with it")
            .expect("joined");
        assert!(ended.is_ok());
    }

    /// A repository added while the daemon runs gets a loop of its own; one
    /// removed has only its own loop drain and stop. The others keep
    /// running, and a removed target's loop stopping ends nothing.
    #[tokio::test(start_paused = true)]
    async fn the_set_of_target_loops_follows_the_policies_without_a_reload() {
        use runner_manager_domain::store::SqliteStore;

        let store = Arc::new(SqliteStore::open_in_memory().expect("store"));
        let host = fixtures::host().build();
        store.put_host(&host).expect("host");
        let policy = |repository: &str| {
            fixtures::policy()
                .id(runner_manager_domain::model::PolicyId::new_random())
                .repository(repository)
                .host(host.id)
                .autoscale("home", 1)
                .active()
                .build()
        };
        let (kept, removed, added) = (
            policy("acme/kept"),
            policy("acme/removed"),
            policy("acme/added"),
        );
        store.insert_policy(&kept).expect("policy");
        store.insert_policy(&removed).expect("policy");
        let built = Arc::new(Mutex::new(Vec::<String>::new()));
        let build = {
            let built = Arc::clone(&built);
            move |group: Vec<ScalePolicy>| {
                built.lock().unwrap().push(group[0].target.to_string());
                let mut report = ReconcileReport::default();
                report.next_poll.delay = Duration::from_millis(50);
                let (mut target, _) = FakeTarget::repeating(group[0].clone(), report);
                target.active = 0;
                std::future::ready(Ok::<_, CliError>(target))
            }
        };
        let (shutdown, _) = tokio::sync::watch::channel(false);
        let contacts: Arc<dyn ContactRecorder> = Arc::new(CountingContacts::default());
        let routes = super::super::coordinator::WakeRoutes::new(None);
        let mut loops = tokio::task::JoinSet::new();
        let mut serving = Serving::default();
        for policy in [&kept, &removed] {
            let target = build(vec![policy.clone()]).await.expect("built");
            routes.add(policy.target.clone());
            serving.spawn(&mut loops, target, &shutdown, &contacts);
        }
        built.lock().unwrap().clear();

        store
            .remove_policy(removed.id, removed.revision())
            .expect("removed");
        store.insert_policy(&added).expect("added");
        let followed = tokio::time::timeout(
            POLICY_SET_CHECK_INTERVAL * 3,
            follow_policy_set(
                Arc::clone(&store) as Arc<dyn Store>,
                &build,
                &mut serving,
                &mut loops,
                (&shutdown, &contacts, &routes),
            ),
        )
        .await;
        assert!(followed.is_err(), "nothing ended: the follower still runs");
        assert_eq!(
            *built.lock().unwrap(),
            vec![added.target.to_string()],
            "only the new target was built"
        );
        assert_eq!(
            serving.served(),
            BTreeSet::from([kept.target.to_string(), added.target.to_string()])
        );
        assert!(
            serving.retiring.is_empty(),
            "the removed loop drained and stopped"
        );
        assert_eq!(loops.len(), 2);
        let _ = shutdown.send(true);
    }

    /// standalone → coordinated → fallback → coordinated, and the interval
    /// each one polls at: 60 s on its own, 300 s while the coordinator's
    /// sweep covers the target, and back to 60 s at once when the link has
    /// been down for `FALLBACK_AFTER`.
    #[tokio::test(start_paused = true)]
    async fn a_target_polls_at_the_interval_its_mode_calls_for() {
        use super::super::coordinator::{LinkView, PollMode};
        use runner_manager_domain::coordinator_protocol::{HelloAck, TargetRef, TargetScope};

        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        report.next_poll.pace = runner_manager_agent::reconcile::PollPace::Nominal;
        let policy = fixtures::policy()
            .repository("acme/repo")
            .autoscale("home", 1)
            .active()
            .build();
        let name = policy.target.clone();
        let (mut target, calls) = FakeTarget::repeating(policy, report);
        let (link, view) = tokio::sync::watch::channel(LinkView::default());
        target.coordination = Some(view);
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        let polls = || calls.load(Ordering::SeqCst);
        let mode = |link: &tokio::sync::watch::Sender<LinkView>| {
            link.borrow().mode(&name, tokio::time::Instant::now())
        };

        // Standalone: every 60 s.
        after(0).await;
        assert_eq!(polls(), 1);
        assert_eq!(mode(&link), PollMode::Standalone);
        after(60).await;
        assert_eq!(polls(), 2);

        // Joined and covered: the next poll moves to 300 s after the last.
        let covered = HelloAck {
            reconcile_interval_secs: Some(120),
            covered: vec![TargetRef {
                scope: TargetScope::Repo,
                name: "acme/repo".into(),
            }],
            any_pool: true,
        };
        link.send_replace(LinkView {
            configured: true,
            connected: true,
            down_since: None,
            ack: Some(covered.clone()),
            version: None,
        });
        after(1).await;
        assert_eq!(mode(&link), PollMode::Coordinated);
        after(100).await;
        assert_eq!(polls(), 2, "no poll at 60 s while coordinated");
        after(199).await;
        assert_eq!(polls(), 3, "polled at 300 s");

        // The link drops: still coordinated for FALLBACK_AFTER, then
        // fallback, and the overdue 60 s poll happens at once.
        after(70).await;
        link.send_modify(|view| {
            view.connected = false;
            view.down_since = Some(tokio::time::Instant::now());
        });
        after(29).await;
        assert_eq!(polls(), 3, "a short outage changes nothing");
        assert_eq!(mode(&link), PollMode::Coordinated);
        after(1).await;
        assert_eq!(mode(&link), PollMode::Fallback);
        assert_eq!(polls(), 4, "back on its own at once");
        after(60).await;
        assert_eq!(polls(), 5, "every 60 s in fallback");

        // Reconnected: coordinated again.
        link.send_modify(|view| {
            view.connected = true;
            view.down_since = None;
        });
        after(60).await;
        assert_eq!(polls(), 5);
        after(240).await;
        assert_eq!(polls(), 6);

        // A coordinator that stops covering the target: fallback, so the
        // poll asleep for 300 s comes at 60 s.
        link.send_modify(|view| view.ack = Some(HelloAck::default()));
        after(1).await;
        assert_eq!(mode(&link), PollMode::Fallback);
        assert_eq!(polls(), 6);
        after(59).await;
        assert_eq!(polls(), 7, "60 s after the last poll, not 300");

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

    /// A covered target this host is running a runner for keeps the normal
    /// intervals: a finished job frees a slot queued work may need now.
    #[tokio::test(start_paused = true)]
    async fn a_coordinated_target_with_a_live_runner_keeps_its_normal_interval() {
        use super::super::coordinator::LinkView;
        use runner_manager_domain::coordinator_protocol::{HelloAck, TargetRef, TargetScope};

        let mut report = ReconcileReport::default();
        report.next_poll.delay = Duration::from_secs(60);
        let (mut target, calls) = FakeTarget::repeating(
            fixtures::policy()
                .repository("acme/repo")
                .autoscale("home", 1)
                .active()
                .build(),
            report,
        );
        target.active = 1;
        let (_link, view) = tokio::sync::watch::channel(LinkView {
            configured: true,
            connected: true,
            down_since: None,
            version: None,
            ack: Some(HelloAck {
                reconcile_interval_secs: Some(120),
                covered: vec![TargetRef {
                    scope: TargetScope::Repo,
                    name: "acme/repo".into(),
                }],
                any_pool: true,
            }),
        });
        target.coordination = Some(view);
        let (stop, _) = tokio::sync::watch::channel(false);
        let daemon = tokio::spawn(run_target_loop(
            target,
            stop.subscribe(),
            never_upgraded(),
            Arc::new(CountingContacts::default()) as Arc<dyn ContactRecorder>,
        ));
        after(0).await;
        after(60).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        stop.send(true).unwrap();
        after(60).await;
        let _ = tokio::time::timeout(Duration::from_secs(1), daemon).await;
    }

    #[tokio::test(start_paused = true)]
    async fn a_wake_never_shortens_a_slowdown_github_asked_for() {
        let (daemon, calls, wake, stop) =
            woken_loop(runner_manager_agent::reconcile::PollPace::RateLimited {
                kind: runner_manager_github::rest::RateLimitKind::Secondary,
            });
        after(0).await;
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        wake.notify_one();
        after(30).await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "a rate-limited target waits out its back-off, woken or not"
        );
        after(30).await;
        assert_eq!(calls.load(Ordering::SeqCst), 2);

        stop.send(true).unwrap();
        after(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")
        }
    }
}