Skip to main content

runner_manager_agent/
lifecycle.rs

1// owner: e3-jit-lifecycle-recovery
2
3//! One ephemeral runner, from an allocation decision to a scrubbed runtime.
4//!
5//! The ordering in this module is intentional.  An attempt is written before
6//! the package or GitHub is touched, the JIT value exists only in a restrictive
7//! handoff, and a registration-timeout termination is journalled before the
8//! process is signalled.  Recovery uses the same code as ordinary supervision;
9//! startup merely supplies the first observation.
10
11#[cfg(test)]
12use std::collections::VecDeque;
13use std::collections::{BTreeMap, BTreeSet};
14use std::ffi::OsStr;
15use std::fmt;
16use std::fs;
17use std::io::Write;
18use std::num::NonZeroU16;
19use std::path::{Path, PathBuf};
20use std::sync::{Arc, Mutex};
21use std::time::Duration;
22
23use async_trait::async_trait;
24use runner_manager_domain::attempt::{
25    AttemptOutcome, AttemptState, FailureReason, GithubRunnerObservation, RecoveryDecision,
26    RecoveryObservation, RecoveryTimeouts, RunnerAttempt, authorize, recovery_decision,
27};
28use runner_manager_domain::model::{AttemptId, Clock, HostId, PolicyId, ScaleTarget};
29use runner_manager_domain::path::LocalAbsolutePath;
30use runner_manager_domain::policy::ScalePolicy;
31use runner_manager_domain::store::{Store, StoreError};
32use runner_manager_domain::workspace::{AttemptWorkspace, WorkspacePolicy};
33use runner_manager_github::jit::{
34    DEFAULT_WORK_FOLDER, EncodedJitConfig, JitError, JitGateway, JitRegistration, JitRunnerRequest,
35};
36use runner_manager_github::rest::{CancelToken, InventoryGateway};
37use runner_manager_platform::process::{
38    Adoption, ChildProcess, ProcessIdentity, RestrictiveHandoff, SpawnSpec, Termination,
39};
40use runner_manager_platform::runner_root::{
41    self, RootOwner, RootPreflight, RunnerRootError, default_runner_root,
42};
43use secrecy::SecretString;
44
45use crate::package::{PackageCache, PackageError, RunnerVersion};
46use crate::reconcile::{
47    AllocationGuard, EventSink, LaunchFailure, LaunchRequest, LifecycleEvent, OutcomeKind,
48    ReplacementIntent, RunnerLauncher,
49};
50
51const IDENTITY_FILE: &str = ".runner-process.json";
52const FALLBACK_IDENTITY_FILE: &str = ".runner-process.recovery.json";
53const UNRESOLVED_PROCESS_FILE: &str = ".runner-process.unresolved";
54const RUNNER_ID_FILE: &str = ".github-runner-id";
55const TERMINATE_INTENT_FILE: &str = ".terminate-registration-timeout";
56const MAX_POST_SPAWN_STOP_ATTEMPTS: usize = 3;
57
58/// Slot-root names a cleaned persistent attempt must not have left behind.
59///
60/// This is not the rule — the rule is that *nothing* but a real `_work`
61/// survives, and [`verify_slot_scrubbed`] enforces that by counting. This list
62/// is the second, independent question asked of the same directory: each name
63/// is stat-ed directly, so a scrub that skipped one is caught even if the
64/// enumeration that was supposed to find it under-reported. Every entry is one
65/// of the things `04-security-recovery.md` requires to be proven absent before a
66/// slot is released; the encoded JIT handoff is the one exception, matched by
67/// its published prefix in [`verify_slot_scrubbed`] because the rest of its name
68/// is a UUID. Being compile-time constants, these are also the only entry names
69/// a refusal message is allowed to print.
70const SENSITIVE_SLOT_ENTRIES: &[&str] = &[
71    // Runner binaries and the launchers beside them.
72    "bin",
73    "externals",
74    "run.sh",
75    "run.cmd",
76    "config.sh",
77    "config.cmd",
78    // The registration identity GitHub's runner writes for itself, and the
79    // per-run environment it reads back.
80    ".runner",
81    ".credentials",
82    ".credentials_rsaparams",
83    ".env",
84    ".path",
85    "_diag",
86    // This agent's own process-identity and lifecycle sidecars.
87    IDENTITY_FILE,
88    FALLBACK_IDENTITY_FILE,
89    UNRESOLVED_PROCESS_FILE,
90    RUNNER_ID_FILE,
91    TERMINATE_INTENT_FILE,
92];
93#[cfg(test)]
94const TEST_LISTENER_READY: &str = ".test-listener-ready";
95
96/// GitHub Runner v2.336.0 accepts JIT configuration for `run` through its
97/// secret `ACTIONS_RUNNER_INPUT_JITCONFIG` input. The platform spawn boundary
98/// supplies that input from the restrictive handoff; the listener command line
99/// must contain only the supported `run` command.
100fn runner_listener_spec(program: PathBuf, runtime: &Path) -> SpawnSpec {
101    let tmp = runtime.join("tmp");
102    let _ = std::fs::create_dir_all(&tmp);
103    SpawnSpec::new(program)
104        .arg("run")
105        .working_dir(runtime)
106        .env("TMPDIR", &tmp)
107        .env("TEMP", &tmp)
108        .env("TMP", &tmp)
109}
110
111/// Retry bounds for failures that can resolve without operator action.
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub struct RetryPolicy {
114    pub max_attempts: u32,
115    pub initial: Duration,
116    pub maximum: Duration,
117}
118
119impl RetryPolicy {
120    #[must_use]
121    pub const fn bounded(max_attempts: u32, initial: Duration, maximum: Duration) -> Self {
122        Self {
123            max_attempts,
124            initial,
125            maximum,
126        }
127    }
128
129    fn delay(self, failure_index: u32) -> Duration {
130        let shift = failure_index.saturating_sub(1).min(31);
131        self.initial
132            .saturating_mul(1_u32 << shift)
133            .min(self.maximum)
134    }
135}
136
137/// Non-secret lifecycle evidence.  Payloads are identifiers and closed enums;
138/// neither the encoded configuration nor child output can enter this type.
139#[derive(Debug, Clone, PartialEq, Eq)]
140pub enum AttemptEvent {
141    State {
142        attempt: AttemptId,
143        state: AttemptState,
144    },
145    Retry {
146        attempt: AttemptId,
147        operation: &'static str,
148        delay: Duration,
149    },
150    Adopted {
151        attempt: AttemptId,
152    },
153    RemoteIdentityRecovered {
154        attempt: AttemptId,
155        runner_id: u64,
156    },
157    TerminateIntent {
158        attempt: AttemptId,
159    },
160    Terminated {
161        attempt: AttemptId,
162    },
163    /// The attempt's GitHub registration was removed by this agent. Carries the
164    /// runner id because that is the identifier an operator sees in the
165    /// target's runner settings, and the attempt id is not shown there.
166    Deregistered {
167        attempt: AttemptId,
168        runner_id: u64,
169    },
170    Concluded {
171        attempt: AttemptId,
172        outcome: OutcomeKind,
173    },
174    Cleaned {
175        attempt: AttemptId,
176        outcome: OutcomeKind,
177    },
178}
179
180pub trait AttemptEventSink: fmt::Debug + Send + Sync {
181    fn emit(&self, event: AttemptEvent);
182}
183
184#[derive(Debug, Default)]
185pub struct AttemptEventLog(Mutex<Vec<AttemptEvent>>);
186
187impl AttemptEventLog {
188    #[must_use]
189    pub fn events(&self) -> Vec<AttemptEvent> {
190        self.0
191            .lock()
192            .map(|events| events.clone())
193            .unwrap_or_default()
194    }
195}
196
197impl AttemptEventSink for AttemptEventLog {
198    fn emit(&self, event: AttemptEvent) {
199        if let Ok(mut events) = self.0.lock() {
200            events.push(event);
201        }
202    }
203}
204
205#[derive(Debug, Clone, Copy, Default)]
206pub struct NoAttemptEvents;
207
208impl AttemptEventSink for NoAttemptEvents {
209    fn emit(&self, _event: AttemptEvent) {}
210}
211
212/// Whether the demand that justified a retry still exists.
213#[async_trait]
214pub trait DemandPersistence: fmt::Debug + Send + Sync {
215    async fn persists(&self, policy: PolicyId) -> bool;
216}
217
218#[derive(Debug, Clone, Copy, Default)]
219pub struct PersistentDemand;
220
221#[async_trait]
222impl DemandPersistence for PersistentDemand {
223    async fn persists(&self, _policy: PolicyId) -> bool {
224        true
225    }
226}
227
228#[async_trait]
229pub trait RetryDelay: fmt::Debug + Send + Sync {
230    async fn wait(&self, duration: Duration);
231}
232
233#[derive(Debug, Clone, Copy, Default)]
234pub struct TokioRetryDelay;
235
236#[async_trait]
237impl RetryDelay for TokioRetryDelay {
238    async fn wait(&self, duration: Duration) {
239        tokio::time::sleep(duration).await;
240    }
241}
242
243#[derive(Debug, Clone, PartialEq, Eq)]
244pub struct JitRequestFailure {
245    pub terminal: bool,
246    pub reason: FailureReason,
247    pub retry_after: Option<Duration>,
248}
249
250/// GitHub's authoritative runner state plus the identity returned by inventory.
251/// The id is carried independently of the local sidecar so recovery can close
252/// the crash boundary immediately after a successful remote registration.
253#[derive(Debug, Clone, Copy, PartialEq, Eq)]
254pub struct LifecycleGithubObservation {
255    pub status: GithubRunnerObservation,
256    pub runner_id: Option<u64>,
257}
258
259impl LifecycleGithubObservation {
260    #[must_use]
261    pub const fn unreachable() -> Self {
262        Self {
263            status: GithubRunnerObservation::Unreachable,
264            runner_id: None,
265        }
266    }
267
268    #[must_use]
269    pub const fn not_registered() -> Self {
270        Self {
271            status: GithubRunnerObservation::NotRegistered,
272            runner_id: None,
273        }
274    }
275
276    #[must_use]
277    pub const fn registered(runner_id: u64, busy: bool) -> Self {
278        Self {
279            status: GithubRunnerObservation::Registered { busy },
280            runner_id: Some(runner_id),
281        }
282    }
283}
284
285/// The two GitHub views the lifecycle needs, combined so one fake can drive
286/// registration and authoritative runner telemetry.
287#[async_trait]
288pub trait LifecycleGithub: fmt::Debug + Send + Sync {
289    async fn register(
290        &self,
291        target: &ScaleTarget,
292        request: &JitRunnerRequest,
293        cancel: &CancelToken,
294    ) -> Result<JitRegistration, JitRequestFailure>;
295
296    async fn observe(
297        &self,
298        target: &ScaleTarget,
299        attempt: AttemptId,
300        cancel: &CancelToken,
301    ) -> LifecycleGithubObservation;
302
303    /// Remove one runner registration this agent created.
304    ///
305    /// Answers whether the registration is gone, and is deliberately not
306    /// fallible in the `Result` sense: no caller may abandon a conclusion
307    /// because GitHub was unreachable. See
308    /// [`LifecycleLauncher::deregister_runner`].
309    async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool;
310}
311
312#[async_trait]
313impl<T> LifecycleGithub for T
314where
315    T: JitGateway + InventoryGateway + fmt::Debug + Send + Sync,
316{
317    async fn register(
318        &self,
319        target: &ScaleTarget,
320        request: &JitRunnerRequest,
321        cancel: &CancelToken,
322    ) -> Result<JitRegistration, JitRequestFailure> {
323        self.generate_jit_config(target, request, cancel)
324            .await
325            .map_err(|error| {
326                let reason = if matches!(&error, JitError::Forbidden { .. }) {
327                    FailureReason::Other(
328                        "GitHub refused JIT registration with 403; check the App's runner permission and runner-group access"
329                            .into(),
330                    )
331                } else {
332                    FailureReason::JitRequestFailed
333                };
334                JitRequestFailure {
335                    terminal: error.is_terminal(),
336                    reason,
337                    retry_after: error
338                        .rate_limited()
339                        .map(|limit| limit.delay_from(self.now())),
340                }
341            })
342    }
343
344    async fn observe(
345        &self,
346        target: &ScaleTarget,
347        attempt: AttemptId,
348        cancel: &CancelToken,
349    ) -> LifecycleGithubObservation {
350        let expected_name = runner_name(attempt);
351        match self.list_runners(target, cancel).await {
352            Ok(inventory) => inventory
353                .runners()
354                .iter()
355                .find(|runner| runner.name == expected_name)
356                .map_or(LifecycleGithubObservation::not_registered(), |runner| {
357                    LifecycleGithubObservation::registered(runner.id, runner.busy)
358                }),
359            Err(_) => LifecycleGithubObservation::unreachable(),
360        }
361    }
362
363    async fn deregister(&self, target: &ScaleTarget, runner_id: u64, cancel: &CancelToken) -> bool {
364        self.remove_runner(target, runner_id, cancel).await.is_ok()
365    }
366}
367
368/// Package/cache operations used by one attempt.
369#[async_trait]
370pub trait RuntimePackages: fmt::Debug + Send + Sync {
371    async fn materialize(&self, attempt: &RunnerAttempt) -> Result<RunnerVersion, FailureReason>;
372    fn release(&self, attempt: AttemptId) -> Result<(), FailureReason>;
373    fn prune_obsolete_guarded(
374        &self,
375        authority: PruneAuthority<'_>,
376        current: &RunnerVersion,
377        attempts: &[RunnerAttempt],
378    ) -> Result<(), FailureReason>;
379}
380
381/// Unforgeable evidence that pruning was reached through e1's launch request.
382/// The type is public only because it appears in the public adapter trait; its
383/// private field and constructor prevent callers from substituting a guard
384/// acquired from an unrelated lock.
385pub struct PruneAuthority<'a> {
386    _guard: &'a AllocationGuard,
387}
388
389impl fmt::Debug for PruneAuthority<'_> {
390    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
391        f.write_str("PruneAuthority")
392    }
393}
394
395impl<'a> PruneAuthority<'a> {
396    fn from_launch_request(guard: &'a AllocationGuard) -> Self {
397        Self { _guard: guard }
398    }
399}
400
401/// Production package adapter.  It takes an e2 lease before returning, so a
402/// cache entry can never look unused while its runtime is starting.
403#[derive(Debug)]
404pub struct CachedRuntimePackages {
405    cache: Arc<PackageCache>,
406}
407
408impl CachedRuntimePackages {
409    #[must_use]
410    pub fn new(cache: Arc<PackageCache>) -> Self {
411        Self { cache }
412    }
413}
414
415#[async_trait]
416impl RuntimePackages for CachedRuntimePackages {
417    async fn materialize(&self, attempt: &RunnerAttempt) -> Result<RunnerVersion, FailureReason> {
418        let installed = self
419            .cache
420            .ensure_installed()
421            .await
422            .map_err(package_failure)?;
423        copy_package_tree(installed.root(), attempt.runtime_path())
424            .map_err(|_| FailureReason::ProcessStartFailed)?;
425        if let Err(error) = self.cache.lease(attempt, installed.version()) {
426            // Undoing the copy must not undo the *job* workspace: a persistent
427            // slot's `_work` is retained across attempts, and this rollback
428            // runs before the attempt that would have owned it ever started.
429            let _ = remove_materialized_package(attempt);
430            return Err(package_failure(error));
431        }
432        Ok(installed.version().clone())
433    }
434
435    fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
436        self.cache.release(attempt).map_err(package_failure)
437    }
438
439    fn prune_obsolete_guarded(
440        &self,
441        _authority: PruneAuthority<'_>,
442        current: &RunnerVersion,
443        attempts: &[RunnerAttempt],
444    ) -> Result<(), FailureReason> {
445        for installed in self.cache.installed().map_err(package_failure)? {
446            if installed.version() != current {
447                match self.cache.prune(installed.version(), attempts) {
448                    Ok(()) | Err(PackageError::VersionInUse { .. }) => {}
449                    Err(error) => return Err(package_failure(error)),
450                }
451            }
452        }
453        Ok(())
454    }
455}
456
457fn package_failure(error: PackageError) -> FailureReason {
458    error.failure_reason().unwrap_or(FailureReason::Other(
459        "runner package cache operation failed".into(),
460    ))
461}
462
463fn package_failure_is_terminal(reason: &FailureReason) -> bool {
464    matches!(
465        reason,
466        FailureReason::RunnerPackageUnverified | FailureReason::RunnerVersionRejected
467    )
468}
469
470/// How many hex characters of the attempt id name its workspace.
471///
472/// # Why this is not the whole identifier, and why the policy is not in the path
473///
474/// Windows refuses a path over `MAX_PATH`, and the runner writes deep inside
475/// this directory: `_work/<repo>/<repo>/.git/objects/pack/pack-<40 hex>.keep`
476/// is 100 characters on its own before the repository is named twice. The
477/// layout used to add two full identifiers -- the policy's and the attempt's,
478/// 74 characters between them -- and that was enough to put a real checkout
479/// over the line. Measured, not guessed: this repository's own CI failed here
480/// three times in a row at 264 characters against a limit of 260, with
481/// `fatal: cannot write keep file ...: Filename too long`. A repository whose
482/// name is ten characters longer would have missed by fourteen.
483///
484/// The policy identifier is simply redundant -- an attempt identifier is
485/// unique on its own, and nothing reads the directory tree to find a policy's
486/// attempts, because [`crate::lifecycle::LifecycleLauncher`] asks the journal.
487/// Twelve hex characters of the attempt is 48 bits, which for the handful of
488/// directories one host holds at once is not a collision anybody will see, and
489/// the journal keeps the full identifier either way.
490///
491/// Together that is 61 characters returned to the repository name.
492const WORKSPACE_NAME_LEN: usize = 12;
493
494/// The directory name for one attempt's workspace.
495fn workspace_name(id: AttemptId) -> String {
496    let full = id.to_string();
497    full.chars()
498        .filter(|c| *c != '-')
499        .take(WORKSPACE_NAME_LEN)
500        .collect()
501}
502
503/// Where one attempt's files go, and which cleanup algorithm they are owed.
504///
505/// The pair travels together because journalling them apart is exactly the bug
506/// `AttemptWorkspace` exists to prevent: a `runtime_path` under a persistent
507/// root recorded as ephemeral would be removed whole, taking the retained job
508/// workspace with it.
509#[derive(Debug, Clone)]
510struct Placement {
511    runtime: PathBuf,
512    workspace: AttemptWorkspace,
513}
514
515/// A runner-root refusal, rendered for the operator who has to fix it.
516///
517/// `RunnerRootError`'s `Display` already names the path, the relation and the
518/// remediation command, and none of its variants can carry a credential — they
519/// are paths, and `03-migration-rollout.md` requires the remediation command to
520/// reach the operator verbatim.
521fn root_failure(error: RunnerRootError) -> LifecycleError {
522    // Logged here because here is the last place this is known at all. A launch
523    // refused by the runner root fails *before* `record_allocation`, so no
524    // attempt row is ever written: `b2` has nothing to carry the failure on and
525    // `g2` has nothing to show it from. The lifecycle event keeps only
526    // `reason=other`, by `failure_reason_kind`'s rule that no free text may
527    // reach an event -- which left the whole refusal reading
528    // `runner_start_failed reason=other`, once per poll, naming nothing.
529    //
530    // What travels is the *kind* and not the sentence, and that is forced rather
531    // than chosen: `crate::logging` redacts every field it does not allow-list
532    // and then scrubs anything path-shaped out of the ones it does, so a
533    // rendered `RunnerRootError` -- which is mostly paths -- reaches the log as
534    // `[redacted]`. `error_kind` is allow-listed and `RunnerRootError::kind` is
535    // a closed vocabulary that survives the scrub, so this names which of a
536    // dozen causes the operator has. The paths and the remediation reach them
537    // through the command line, which is not redacted.
538    tracing::warn!(
539        error_kind = error.kind(),
540        "the runner root refused this launch, so no attempt was created; the host will \
541         retry every poll until the cause is resolved. Re-running `host set-runtime-root` \
542         with the same path re-runs this check and prints the directory and the \
543         remediation in full"
544    );
545    LifecycleError::Failed(FailureReason::Other(error.to_string()))
546}
547
548/// The lowest positive slot inside `ceiling` that no uncleaned attempt holds.
549///
550/// `leases` is the journal's answer to "which slots are leased"
551/// (`Store::slot_leases_for_policy`), which deliberately includes a terminal
552/// attempt whose cleanup has not finished: that attempt still owns its
553/// directory, so its slot is not free even though it no longer counts against
554/// host capacity. `None` means the ceiling is reached, which is a refusal and
555/// not a reason to allocate `s(ceiling + 1)`.
556fn lowest_free_slot(leases: &[RunnerAttempt], ceiling: NonZeroU16) -> Option<NonZeroU16> {
557    let held: BTreeSet<u16> = leases
558        .iter()
559        .filter_map(|attempt| attempt.workspace().slot_number())
560        .collect();
561    (1..=ceiling.get())
562        .find(|slot| !held.contains(slot))
563        .and_then(NonZeroU16::new)
564}
565
566/// Create `<root>/sN`, or prove that what is already there is a real directory.
567///
568/// A symlink, junction or reparse point standing where the slot should be is
569/// refused rather than followed: it is the one thing that could put an attempt's
570/// files outside the root the operator configured, and
571/// `04-security-recovery.md` requires that case to fail closed rather than to
572/// be repaired here.
573fn create_or_validate_slot(slot: &Path) -> Result<(), LifecycleError> {
574    match fs::symlink_metadata(slot) {
575        // [`is_link_like`] and not `is_symlink`, so that this is the same
576        // question cleanup asks in [`slot_is_present`]: a reparse tag the
577        // standard library has no name for is refused here rather than
578        // allocated into and then quarantined forever by a cleanup that will
579        // not scrub it.
580        Ok(metadata) if is_link_like(&metadata) => Err(slot_refusal(
581            slot,
582            "is a symbolic link, junction or other reparse point, which could place runner \
583             files outside the configured root",
584        )),
585        Ok(metadata) if !metadata.is_dir() => Err(slot_refusal(slot, "is not a directory")),
586        Ok(_) => Ok(()),
587        Err(error) if error.kind() == std::io::ErrorKind::NotFound => fs::create_dir(slot)
588            .map_err(|source| slot_refusal(slot, format!("could not be created: {source}"))),
589        Err(source) => Err(slot_refusal(
590            slot,
591            format!("could not be inspected: {source}"),
592        )),
593    }
594}
595
596/// Accept a slot for reuse only when it is empty or holds one real `_work`.
597///
598/// `02-target-architecture.md`: "Before materialization, a reusable slot must
599/// contain only a valid real `_work` directory or be empty." Everything else —
600/// a leftover `bin/`, a link-shaped `_work`, a stray file — is refused here
601/// rather than cleaned, because deciding whether those bytes are safe is
602/// cleanup's and recovery's job (`c3`), and quietly reusing them would hand one
603/// repository's retained state to the next attempt without anybody choosing to.
604///
605/// The inspection is one level deep and uses `symlink_metadata`, so nothing is
606/// followed while it is being judged.
607fn accept_reusable_slot(slot: &Path) -> Result<(), LifecycleError> {
608    let unreadable =
609        |source: std::io::Error| slot_refusal(slot, format!("could not be read: {source}"));
610    let entries = fs::read_dir(slot).map_err(unreadable)?;
611    let mut refused: Vec<String> = Vec::new();
612    for entry in entries {
613        let entry = entry.map_err(unreadable)?;
614        let name = entry.file_name();
615        let metadata = fs::symlink_metadata(entry.path()).map_err(|source| {
616            slot_refusal(
617                slot,
618                format!("entry {name:?} could not be inspected: {source}"),
619            )
620        })?;
621        // The same predicate cleanup retains by, so a `_work` this accepts is
622        // one [`scrub_slot_entries`] will keep rather than refuse: a link, a
623        // junction or any other reparse point is not a job workspace to either
624        // of them.
625        if is_retainable_work_folder(&name, &metadata) {
626            continue;
627        }
628        refused.push(name.to_string_lossy().into_owned());
629    }
630    if refused.is_empty() {
631        return Ok(());
632    }
633    refused.sort();
634    Err(slot_refusal(
635        slot,
636        format!(
637            "holds {} that this attempt may not reuse: [{}]. A reusable slot is empty or holds \
638             one real `{DEFAULT_WORK_FOLDER}` directory and nothing else; remove or move the \
639             entries listed, or let cleanup and recovery resolve them",
640            if refused.len() == 1 {
641                "an entry"
642            } else {
643                "entries"
644            },
645            refused.join(", ")
646        ),
647    ))
648}
649
650fn slot_refusal(slot: &Path, detail: impl fmt::Display) -> LifecycleError {
651    LifecycleError::Failed(FailureReason::Other(format!(
652        "the persistent slot {} {detail}",
653        slot.display()
654    )))
655}
656
657/// Whether a directory entry names the retained job workspace.
658///
659/// The comparison folds case on Windows because the filesystem does: there
660/// `_Work` and `_work` are one directory, so a case-sensitive test would let
661/// [`scrub_slot_entries`] delete the very directory it exists to keep, let
662/// [`accept_reusable_slot`] refuse a slot that holds nothing but a valid job
663/// workspace, and let a package's top-level `_Work` merge itself into the
664/// previous attempt's `_work`. Elsewhere the two names really are two
665/// directories and only the exact one is the job workspace.
666fn is_work_folder(name: &OsStr) -> bool {
667    if cfg!(windows) {
668        name.eq_ignore_ascii_case(DEFAULT_WORK_FOLDER)
669    } else {
670        name == OsStr::new(DEFAULT_WORK_FOLDER)
671    }
672}
673
674/// Whether the operating system would follow this entry somewhere else.
675///
676/// `FileType::is_symlink` is the whole answer on Unix. On Windows it is not:
677/// the standard library reports only the symlink and mount-point reparse tags,
678/// and the substitution this has to refuse is *any* reparse point standing
679/// where a real directory should be. So the attribute bit is the test there,
680/// and a tag the standard library has no name for fails closed with the two it
681/// does.
682fn is_link_like(metadata: &fs::Metadata) -> bool {
683    if metadata.file_type().is_symlink() {
684        return true;
685    }
686    #[cfg(windows)]
687    {
688        use std::os::windows::fs::MetadataExt;
689
690        const FILE_ATTRIBUTE_REPARSE_POINT: u32 = 0x0000_0400;
691        metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0
692    }
693    #[cfg(not(windows))]
694    false
695}
696
697/// Whether an entry is the retained job workspace and is safe to retain.
698///
699/// The two halves are one question. A `_work` that is a file, a symlink, a
700/// junction or any other reparse point is not a job workspace, and it is also
701/// the exact substitution a hostile workflow makes to send cleanup somewhere
702/// else (`04-security-recovery.md`, "A workflow replaces `_work` with a
703/// junction or symlink to escape cleanup").
704fn is_retainable_work_folder(name: &OsStr, metadata: &fs::Metadata) -> bool {
705    is_work_folder(name) && metadata.is_dir() && !is_link_like(metadata)
706}
707
708/// Undo one package materialization, dispatching on the journalled workspace.
709///
710/// The ephemeral half is what this always did: the directory is the attempt's
711/// alone, so it goes whole. The persistent half removes the copy and nothing
712/// else, because the slot's `_work` predates this attempt and outlives it
713/// (`02-target-architecture.md`, "Persistent repository").
714fn remove_materialized_package(attempt: &RunnerAttempt) -> std::io::Result<()> {
715    match attempt.workspace() {
716        AttemptWorkspace::Ephemeral => remove_runtime_tree(attempt.runtime_path()),
717        AttemptWorkspace::PersistentSlot { .. } => scrub_slot_entries(attempt.runtime_path())
718            .map_err(|quarantine| std::io::Error::other(quarantine.to_string())),
719    }
720}
721
722/// Remove a disposable runner tree without opening workflow-created special files.
723///
724/// On Unix, `remove_dir_all` 1.0 can open a FIFO while walking a directory and
725/// wait forever for its peer. The .NET runner routinely leaves diagnostic FIFOs
726/// in its private `tmp`, so use the standard library's fd-relative Unix remover,
727/// which unlinks non-directories without opening them. Keep the external remover
728/// on Windows for its existing read-only and junction handling.
729fn remove_runtime_tree(path: &Path) -> std::io::Result<()> {
730    #[cfg(windows)]
731    {
732        remove_dir_all::remove_dir_all(path)
733    }
734    #[cfg(not(windows))]
735    {
736        fs::remove_dir_all(path)
737    }
738}
739
740// ---------------------------------------------------------------------------
741// Persistent cleanup (`04-security-recovery.md`, "Safe path handling")
742// ---------------------------------------------------------------------------
743
744/// Why a persistent slot could not be proven safe to scrub.
745///
746/// A closed set rather than a formatted string, for two reasons that point the
747/// same way. [`LifecycleEvent::AttemptCleanFailed`] takes a `&'static str` for
748/// exactly the reason [`crate::reconcile::failure_reason_kind`] documents —
749/// free text is the one shape that can carry a credential past a field
750/// allow-list. And the entries under a slot root are *workflow-controlled*: a
751/// job that writes a file named after a secret would publish it through any
752/// message that echoed a directory listing, which is why nothing here ever
753/// renders an entry name that did not come from this module's own constants.
754#[derive(Debug, Clone, Copy, PartialEq, Eq)]
755enum SlotRefusal {
756    /// The journalled runtime path is not `<root>/sN` for the journalled slot.
757    NotTheJournalledSlot,
758    /// A policy that still exists names a different root than the journal does.
759    PolicyRootDisagrees,
760    /// The slot is not strictly inside its root once components resolve.
761    Containment,
762    /// The slot itself is a file, a link, or could not be inspected.
763    SlotNotADirectory,
764    /// The slot's direct entries could not be listed.
765    Enumeration,
766    /// `_work` is a file, a symlink, a junction or another reparse point.
767    WorkNotADirectory,
768    /// An entry that had to go could not be removed.
769    Deletion,
770    /// Something other than the job workspace survived removal.
771    Residue,
772}
773
774impl SlotRefusal {
775    /// The event field: a fixed vocabulary, never operator or workflow text.
776    const fn class(self) -> &'static str {
777        match self {
778            Self::NotTheJournalledSlot => "slot_path_is_not_the_journalled_slot",
779            Self::PolicyRootDisagrees => "slot_root_disagrees_with_policy",
780            Self::Containment => "slot_escapes_its_root",
781            Self::SlotNotADirectory => "slot_is_not_a_directory",
782            Self::Enumeration => "slot_could_not_be_enumerated",
783            Self::WorkNotADirectory => "retained_work_is_not_a_directory",
784            Self::Deletion => "slot_entry_could_not_be_removed",
785            Self::Residue => "slot_still_holds_runner_state",
786        }
787    }
788
789    /// What the operator has to do, in one sentence and with no path in it.
790    const fn remediation(self) -> &'static str {
791        match self {
792            Self::NotTheJournalledSlot | Self::PolicyRootDisagrees | Self::Containment => {
793                "the attempt keeps its slot lease and nothing was removed; correct the \
794                 repository's persistent workspace path, or remove the slot directory by hand \
795                 once you have confirmed what is in it"
796            }
797            Self::SlotNotADirectory | Self::WorkNotADirectory => {
798                "the attempt keeps its slot lease and nothing was removed; a job replaced the \
799                 slot or its `_work` with a link, so inspect it before deleting anything and \
800                 treat the retained workspace as untrusted"
801            }
802            Self::Enumeration | Self::Deletion | Self::Residue => {
803                "the attempt keeps its slot lease and will be cleaned again on the next pass; \
804                 release whatever is holding the files open, or remove the slot's contents by \
805                 hand leaving only `_work`"
806            }
807        }
808    }
809}
810
811/// A refusal that leaves one persistent slot quarantined.
812///
813/// `detail` is redacted by construction: it may hold paths this product
814/// configured, `std::io::ErrorKind` values, counts, and names drawn from
815/// [`SENSITIVE_SLOT_ENTRIES`] — and nothing that came out of a directory
816/// listing.
817#[derive(Debug, Clone, PartialEq, Eq)]
818struct SlotQuarantine {
819    refusal: SlotRefusal,
820    detail: String,
821}
822
823impl SlotQuarantine {
824    fn new(refusal: SlotRefusal, detail: impl Into<String>) -> Self {
825        Self {
826            refusal,
827            detail: detail.into(),
828        }
829    }
830}
831
832impl fmt::Display for SlotQuarantine {
833    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
834        write!(f, "{}; {}", self.detail, self.refusal.remediation())
835    }
836}
837
838/// Steps 1 to 3: the journalled root, the journalled slot, and containment.
839///
840/// Everything comes from the two immutable allocation facts — the exact runtime
841/// path and the slot number. Not from the filesystem: scanning a root to decide
842/// which directories are "mine" is what invariant 6 forbids, and it is also
843/// impossible for an attempt whose policy has since been deleted. `configured`
844/// is therefore a *cross-check* and not a source. A policy that survives has to
845/// agree; a policy that does not survive removes a check rather than the
846/// ability to clean the directory the journal already names.
847///
848/// The name test comes before the parent is used, so `<root>/s1` for a journal
849/// row that says `s2` — and anything at all that is not one `sN` component —
850/// is refused before a root is derived from it.
851fn verify_journalled_slot(
852    runtime: &Path,
853    slot: NonZeroU16,
854    configured: Option<&LocalAbsolutePath>,
855) -> Result<(), SlotQuarantine> {
856    let mislaid = || {
857        SlotQuarantine::new(
858            SlotRefusal::NotTheJournalledSlot,
859            format!(
860                "the journalled runtime {} is not the slot s{slot} this attempt was allocated as",
861                runtime.display()
862            ),
863        )
864    };
865    let local = |path: &Path| {
866        path.to_str()
867            .and_then(|raw| LocalAbsolutePath::new(raw).ok())
868            .ok_or_else(mislaid)
869    };
870
871    let runtime_path = local(runtime)?;
872    let root = local(runtime.parent().ok_or_else(mislaid)?)?;
873    // Containment's lexical half, by construction: `derive_child` accepts one
874    // component, so the equality below can only hold when the journalled path
875    // really is this root's `sN` and nothing else.
876    // The directory name comes from the domain that allocation named it with,
877    // never from a second `s{n}` spelled out here: a convention with two
878    // spellings would let cleanup refuse every slot the allocator created.
879    let name = AttemptWorkspace::persistent_slot(slot)
880        .slot_directory_name()
881        .expect("a persistent workspace names its slot directory");
882    let derived = runner_root::derive_child(&root, &name).map_err(|_| mislaid())?;
883    if derived != runtime_path {
884        return Err(mislaid());
885    }
886    if let Some(configured) = configured
887        && configured != &root
888    {
889        return Err(SlotQuarantine::new(
890            SlotRefusal::PolicyRootDisagrees,
891            format!(
892                "the journalled slot {} is not under the repository's configured persistent root \
893                 {}",
894                runtime.display(),
895                configured.as_str()
896            ),
897        ));
898    }
899    // And containment's canonical half, which is what a junction planted inside
900    // the root between allocation and cleanup has to get past.
901    runner_root::verify_containment(&root, &derived).map_err(|source| {
902        SlotQuarantine::new(
903            SlotRefusal::Containment,
904            format!("the journalled slot is not inside the root it was allocated from: {source}"),
905        )
906    })
907}
908
909/// Whether the slot is there to be scrubbed at all, before its entries are.
910///
911/// `Ok(false)` — the directory is gone — is not a refusal. There is nothing to
912/// remove and nothing to prove absent, which is the same tolerance the
913/// disposable arm has always had for a runtime that vanished under it.
914///
915/// A slot that is a file, a link, a junction or any other reparse point *is* a
916/// refusal, and it is checked here rather than inside the enumeration so that
917/// the reason an operator reads names the shape rather than reporting that a
918/// directory could not be listed.
919fn slot_is_present(slot: &Path) -> Result<bool, SlotQuarantine> {
920    match fs::symlink_metadata(slot) {
921        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
922        Err(source) => Err(SlotQuarantine::new(
923            SlotRefusal::SlotNotADirectory,
924            format!(
925                "the slot {} could not be inspected: {:?}",
926                slot.display(),
927                source.kind()
928            ),
929        )),
930        Ok(metadata) if !metadata.is_dir() || is_link_like(&metadata) => Err(SlotQuarantine::new(
931            SlotRefusal::SlotNotADirectory,
932            format!(
933                "the slot {} is a link or a file rather than a real directory",
934                slot.display()
935            ),
936        )),
937        Ok(_) => Ok(true),
938    }
939}
940
941/// Steps 4 to 6: list the slot's direct entries, keep one real `_work`, remove
942/// every other one.
943///
944/// Every call is a literal filesystem API against a path built from the slot
945/// root and one entry name — no glob, no shell string, no repository-controlled
946/// fragment. Nothing is followed: `symlink_metadata` says what each entry *is*,
947/// and a link-shaped entry is unlinked rather than descended into, so a junction
948/// planted where `bin` used to be cannot take the deletion outside the slot.
949///
950/// A `_work` that is not a real directory is the one entry this refuses to act
951/// on at all. Removing it could destroy an operator's data if the link were
952/// theirs; keeping it would hand the next attempt a workspace pointing
953/// anywhere. `04-security-recovery.md` requires that case to quarantine the
954/// slot, so the whole scrub stops there with nothing removed after it.
955fn scrub_slot_entries(slot: &Path) -> Result<(), SlotQuarantine> {
956    let unreadable = |source: std::io::Error| {
957        SlotQuarantine::new(
958            SlotRefusal::Enumeration,
959            format!(
960                "the entries of {} could not be listed: {:?}",
961                slot.display(),
962                source.kind()
963            ),
964        )
965    };
966    for entry in fs::read_dir(slot).map_err(unreadable)? {
967        let name = entry.map_err(unreadable)?.file_name();
968        let path = slot.join(&name);
969        // Nothing is assumed from an entry that vanished: `verify_slot_scrubbed`
970        // asks the filesystem again afterwards and refuses if it is still there.
971        let Some(metadata) = listed_entry_metadata(&path).map_err(unreadable)? else {
972            continue;
973        };
974        if is_work_folder(&name) {
975            if is_retainable_work_folder(&name, &metadata) {
976                continue;
977            }
978            return Err(SlotQuarantine::new(
979                SlotRefusal::WorkNotADirectory,
980                format!(
981                    "the retained `{DEFAULT_WORK_FOLDER}` in {} is a link or a file rather than a \
982                     real directory",
983                    slot.display()
984                ),
985            ));
986        }
987        remove_slot_entry(&path, &metadata).map_err(|source| {
988            SlotQuarantine::new(
989                SlotRefusal::Deletion,
990                format!(
991                    "an entry of {} could not be removed: {:?}",
992                    slot.display(),
993                    source.kind()
994                ),
995            )
996        })?;
997    }
998    Ok(())
999}
1000
1001/// What a listed entry *is*, or `None` when it is no longer there.
1002///
1003/// An entry named by a listing and gone by the time it is stat-ed is absent,
1004/// which is a fact both passes over a slot want rather than an enumeration that
1005/// failed. Nothing is followed: `symlink_metadata` reports a link as a link.
1006fn listed_entry_metadata(path: &Path) -> std::io::Result<Option<fs::Metadata>> {
1007    match fs::symlink_metadata(path) {
1008        Ok(metadata) => Ok(Some(metadata)),
1009        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
1010        Err(source) => Err(source),
1011    }
1012}
1013
1014/// Remove one slot entry without following it.
1015///
1016/// An entry that is already gone is removed: the post-condition this serves is
1017/// "not there", and racing with whatever removed it first is not a refusal.
1018fn remove_slot_entry(path: &Path, metadata: &fs::Metadata) -> std::io::Result<()> {
1019    let removed = if is_link_like(metadata) {
1020        // A file symlink unlinks with `remove_file`; a directory symlink or a
1021        // Windows junction needs `remove_dir`. Neither follows the link.
1022        fs::remove_file(path).or_else(|_| fs::remove_dir(path))
1023    } else if metadata.is_dir() {
1024        remove_runtime_tree(path)
1025    } else {
1026        fs::remove_file(path)
1027    };
1028    match removed {
1029        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
1030        other => other,
1031    }
1032}
1033
1034/// Step 7: prove that nothing but the retained job workspace is left.
1035///
1036/// Two passes, deliberately independent, because the interesting failure is an
1037/// enumeration that under-reports. The first asks the directory what remains
1038/// and counts everything that is not a real `_work`. The second ignores the
1039/// listing entirely and stats the names an attempt is known to write — the
1040/// runner's binaries, the registration identity it stores beside them, the
1041/// process-identity sidecars and this agent's lifecycle marks — so a scrub that
1042/// silently skipped one is caught by a question that never consulted the
1043/// listing that skipped it.
1044///
1045/// Only the second pass names anything, and the encoded JIT handoff is reported
1046/// by its published prefix rather than by the UUID that follows it. A slot root
1047/// is workflow-writable, so a job that named a file after a secret would publish
1048/// it through any message that echoed the listing; the first pass therefore
1049/// reports a count.
1050fn verify_slot_scrubbed(slot: &Path) -> Result<(), SlotQuarantine> {
1051    let unreadable = |source: std::io::Error| {
1052        SlotQuarantine::new(
1053            SlotRefusal::Enumeration,
1054            format!(
1055                "the entries of {} could not be listed to verify the scrub: {:?}",
1056                slot.display(),
1057                source.kind()
1058            ),
1059        )
1060    };
1061    let mut residue = 0_usize;
1062    let mut named: Vec<String> = Vec::new();
1063    for entry in fs::read_dir(slot).map_err(unreadable)? {
1064        let name = entry.map_err(unreadable)?.file_name();
1065        // Absent is what this pass is here to establish. The named half below
1066        // stats every sensitive entry again, so an entry that only *looks*
1067        // absent to this listing is still caught.
1068        let Some(metadata) = listed_entry_metadata(&slot.join(&name)).map_err(unreadable)? else {
1069            continue;
1070        };
1071        if is_retainable_work_folder(&name, &metadata) {
1072            continue;
1073        }
1074        residue = residue.saturating_add(1);
1075        if name
1076            .to_string_lossy()
1077            .starts_with(RestrictiveHandoff::NAME_PREFIX)
1078        {
1079            named.push("an encoded JIT handoff".to_owned());
1080        }
1081    }
1082    named.extend(
1083        SENSITIVE_SLOT_ENTRIES
1084            .iter()
1085            .filter(|entry| fs::symlink_metadata(slot.join(entry)).is_ok())
1086            .map(|entry| format!("`{entry}`")),
1087    );
1088    if residue == 0 && named.is_empty() {
1089        return Ok(());
1090    }
1091    named.sort_unstable();
1092    named.dedup();
1093    Err(SlotQuarantine::new(
1094        SlotRefusal::Residue,
1095        residue_detail(slot, residue, &named),
1096    ))
1097}
1098
1099/// Word a [`SlotRefusal::Residue`] refusal from the two facts that produced it.
1100///
1101/// Separate from [`verify_slot_scrubbed`] because the disagreement it has to
1102/// report — a listing that counted nothing and a filesystem that answered
1103/// otherwise — is a race no test can stage, and a message that contradicts
1104/// itself is exactly what an operator reads at three in the morning.
1105///
1106/// `named` is the sanitized half: entries this crate published the names of.
1107/// A name a workflow chose is only ever counted, never echoed.
1108fn residue_detail(slot: &Path, residue: usize, named: &[String]) -> String {
1109    if residue == 0 {
1110        // The whole reason the second pass ignores the listing: the listing
1111        // reported a clean slot and the filesystem disagrees. Saying "0
1112        // entries survived" here would report the under-count as the fact.
1113        format!(
1114            "the listing of {} reported nothing but `{DEFAULT_WORK_FOLDER}`, yet {} survived \
1115                 cleanup",
1116            slot.display(),
1117            named.join(", ")
1118        )
1119    } else {
1120        format!(
1121            "{residue} entr{} other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}{}",
1122            if residue == 1 { "y" } else { "ies" },
1123            slot.display(),
1124            if named.is_empty() {
1125                String::new()
1126            } else {
1127                format!(", including {}", named.join(", "))
1128            }
1129        )
1130    }
1131}
1132
1133fn replacement_operation(outcome: &AttemptOutcome) -> Option<&'static str> {
1134    match outcome {
1135        AttemptOutcome::Failed {
1136            reason: FailureReason::JitExpired,
1137        } => Some("jit_expired_replacement"),
1138        AttemptOutcome::Failed {
1139            reason: FailureReason::ProcessExitedUnexpectedly,
1140        } => Some("exit_before_acceptance_replacement"),
1141        _ => None,
1142    }
1143}
1144
1145/// Lay the verified runner package out *around* whatever the slot retains.
1146///
1147/// `02-target-architecture.md`: "The verified runner package is copied into the
1148/// slot for the attempt", beside a `_work` that survives every attempt. Two
1149/// properties make that safe, and both are structural rather than documented:
1150///
1151/// * the walk is of the **source** tree, so a retained `_work` in the
1152///   destination is never opened, never descended into, and cannot be followed
1153///   wherever it might point;
1154/// * a top-level source entry named `_work` is refused rather than copied, so a
1155///   package that ever grew one could not merge itself into, or replace, the
1156///   job workspace of the attempt before it. The refusal is top-level only,
1157///   because the retained directory is a direct child of the slot; a `_work`
1158///   nested inside the package's own tree is an ordinary name.
1159fn copy_package_tree(source: &Path, destination: &Path) -> std::io::Result<()> {
1160    if source.join(DEFAULT_WORK_FOLDER).exists() {
1161        return Err(std::io::Error::new(
1162            std::io::ErrorKind::InvalidData,
1163            "a cached runner package contains a _work folder, which means it was used \
1164             to run a job before it was archived; the cache must only contain clean \
1165             extracts to prevent data leakage",
1166        ));
1167    }
1168
1169    #[cfg(unix)]
1170    {
1171        let status = std::process::Command::new("cp")
1172            .arg("-a")
1173            .arg(format!("{}/.", source.display()))
1174            .arg(destination)
1175            .status()?;
1176        if status.success() {
1177            Ok(())
1178        } else {
1179            Err(std::io::Error::other("cp failed"))
1180        }
1181    }
1182    #[cfg(not(unix))]
1183    copy_package_entries(source, destination, true)
1184}
1185
1186#[cfg(not(unix))]
1187fn copy_package_entries(source: &Path, destination: &Path, top_level: bool) -> std::io::Result<()> {
1188    fs::create_dir_all(destination)?;
1189    for entry in fs::read_dir(source)? {
1190        let entry = entry?;
1191        if top_level && is_work_folder(&entry.file_name()) {
1192            return Err(std::io::Error::new(
1193                std::io::ErrorKind::InvalidData,
1194                format!(
1195                    "the runner package holds a top-level `{DEFAULT_WORK_FOLDER}`; copying \
1196                     it would overwrite the job workspace a persistent slot retains"
1197                ),
1198            ));
1199        }
1200        let target = destination.join(entry.file_name());
1201        if entry.file_type()?.is_dir() {
1202            copy_package_entries(&entry.path(), &target, false)?;
1203        } else {
1204            fs::copy(entry.path(), target)?;
1205        }
1206    }
1207    Ok(())
1208}
1209
1210/// Process operations are attempt-addressed so a recovered process and a child
1211/// started in this invocation are supervised through one port.
1212#[derive(Debug, Clone, PartialEq, Eq)]
1213pub struct ProcessStartFailure {
1214    pub reason: FailureReason,
1215    /// False once a child existed: the one-shot JIT value may have been
1216    /// consumed, so retrying it could start a duplicate.
1217    pub retryable: bool,
1218    /// Set only when cleanup could not prove the spawned process dead.  The
1219    /// caller must journal `starting` and retain capacity/supervision.
1220    pub live_pid: Option<u32>,
1221}
1222
1223impl ProcessStartFailure {
1224    fn before_spawn(reason: FailureReason) -> Self {
1225        Self {
1226            reason,
1227            retryable: true,
1228            live_pid: None,
1229        }
1230    }
1231
1232    fn after_spawn_stopped() -> Self {
1233        Self {
1234            reason: FailureReason::ProcessStartFailed,
1235            retryable: false,
1236            live_pid: None,
1237        }
1238    }
1239
1240    fn after_spawn_live(pid: u32) -> Self {
1241        Self::after_spawn_live_with_reason(pid, FailureReason::ProcessStartFailed)
1242    }
1243
1244    fn after_spawn_live_with_reason(pid: u32, reason: FailureReason) -> Self {
1245        Self {
1246            reason,
1247            retryable: false,
1248            live_pid: Some(pid),
1249        }
1250    }
1251}
1252
1253pub trait ProcessSupervisor: fmt::Debug + Send + Sync {
1254    fn spawn(
1255        &self,
1256        attempt: &RunnerAttempt,
1257        config: &EncodedJitConfig,
1258    ) -> Result<u32, ProcessStartFailure>;
1259    fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason>;
1260    /// Durable identity observed for a process that spawned before the
1261    /// `starting` journal write survived.
1262    fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason>;
1263    /// True only for a child this invocation owned and reaped with a successful
1264    /// exit status.  A recovered process that is merely gone answers false.
1265    fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool;
1266    fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
1267    fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool;
1268    fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason>;
1269}
1270
1271/// Native process supervision.  The start token is stored beside the runtime;
1272/// recovery never trusts a recycled PID merely because SQLite contains it.
1273#[derive(Debug, Default)]
1274pub struct NativeProcesses {
1275    children: Mutex<BTreeMap<AttemptId, ChildProcess>>,
1276    successful_exits: Mutex<BTreeMap<AttemptId, bool>>,
1277    #[cfg(test)]
1278    post_spawn_faults: Mutex<VecDeque<PostSpawnBoundary>>,
1279    #[cfg(test)]
1280    post_spawn_reaps: std::sync::atomic::AtomicUsize,
1281    #[cfg(test)]
1282    post_spawn_stop_failures: std::sync::atomic::AtomicUsize,
1283    #[cfg(test)]
1284    use_long_lived_test_listener: std::sync::atomic::AtomicBool,
1285}
1286
1287#[cfg(test)]
1288#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1289enum PostSpawnBoundary {
1290    HandoffDelete,
1291    IdentitySerialize,
1292    IdentityWrite,
1293    ChildMapInsert,
1294}
1295
1296impl NativeProcesses {
1297    #[must_use]
1298    pub fn new() -> Self {
1299        Self::default()
1300    }
1301
1302    #[cfg(test)]
1303    fn fail_post_spawn_at(&self, boundary: PostSpawnBoundary) {
1304        self.post_spawn_faults.lock().unwrap().push_back(boundary);
1305    }
1306
1307    #[cfg(test)]
1308    fn faults_at(&self, boundary: PostSpawnBoundary) -> bool {
1309        let mut faults = self.post_spawn_faults.lock().unwrap();
1310        if faults.front() == Some(&boundary) {
1311            faults.pop_front();
1312            true
1313        } else {
1314            false
1315        }
1316    }
1317
1318    #[cfg(test)]
1319    fn fail_post_spawn_stops(&self, count: usize) {
1320        self.post_spawn_stop_failures
1321            .fetch_add(count, std::sync::atomic::Ordering::SeqCst);
1322    }
1323
1324    #[cfg(test)]
1325    fn fail_next_post_spawn_stop(&self) {
1326        self.fail_post_spawn_stops(1);
1327    }
1328
1329    #[cfg(test)]
1330    fn use_long_lived_test_listener(&self) {
1331        self.use_long_lived_test_listener
1332            .store(true, std::sync::atomic::Ordering::SeqCst);
1333    }
1334
1335    fn stop_spawned_child(&self, child: &mut ChildProcess) -> Result<(), FailureReason> {
1336        #[cfg(test)]
1337        if self
1338            .post_spawn_stop_failures
1339            .fetch_update(
1340                std::sync::atomic::Ordering::SeqCst,
1341                std::sync::atomic::Ordering::SeqCst,
1342                |left| if left > 0 { Some(left - 1) } else { None },
1343            )
1344            .is_ok()
1345        {
1346            return Err(FailureReason::Other("injected runner stop failure".into()));
1347        }
1348        child
1349            .stop(Duration::from_secs(1))
1350            .map(|_| ())
1351            .map_err(|_| FailureReason::Other("spawned runner process could not be stopped".into()))
1352    }
1353
1354    fn abort_spawned_child(
1355        &self,
1356        mut child: ChildProcess,
1357        attempt: &RunnerAttempt,
1358        remove_identity: bool,
1359    ) -> ProcessStartFailure {
1360        let mut reaped = self.stop_spawned_child(&mut child).is_ok();
1361        if reaped {
1362            if remove_identity {
1363                Self::remove_identity_files(attempt);
1364            }
1365        } else {
1366            // A failed stop is not a failed attempt yet.  Persist enough truth
1367            // for crash recovery, and retain the owned child when possible.
1368            let identity_durable =
1369                serde_json::to_vec(child.identity())
1370                    .ok()
1371                    .is_some_and(|identity| {
1372                        self.persist_identity(attempt, &identity).is_ok()
1373                            || self.persist_fallback_identity(attempt, &identity).is_ok()
1374                    });
1375            if !identity_durable {
1376                // Returning a live PID as though recovery were complete would
1377                // make the next boot trust a recyclable PID. Reaping is bounded;
1378                // if it cannot finish, the durable `starting` journal entry is
1379                // deliberately unresolved on restart and blocks new launches.
1380                for _ in 1..MAX_POST_SPAWN_STOP_ATTEMPTS {
1381                    if self.stop_spawned_child(&mut child).is_ok() {
1382                        reaped = true;
1383                        break;
1384                    }
1385                }
1386                if !reaped {
1387                    // The attempt journal will durably record `starting` and
1388                    // its PID. Recovery treats a missing full identity as
1389                    // unresolved and starts nothing, so bounded stop failure
1390                    // cannot turn into either a hang or a duplicate runner.
1391                    let pid = child.pid();
1392                    let marker = write_durable_file(
1393                        &Self::unresolved_process_path(attempt),
1394                        pid.to_string().as_bytes(),
1395                    );
1396                    self.children
1397                        .lock()
1398                        .unwrap_or_else(std::sync::PoisonError::into_inner)
1399                        .insert(attempt.id, child);
1400                    let reason = if marker.is_ok() {
1401                        FailureReason::Other(
1402                            "spawn cleanup exhausted its bounded stop attempts; the live process remains under durable unresolved supervision"
1403                                .into(),
1404                        )
1405                    } else {
1406                        FailureReason::Other(
1407                            "spawn cleanup exhausted its bounded stop attempts and the unresolved-process marker could not be journalled"
1408                                .into(),
1409                        )
1410                    };
1411                    return ProcessStartFailure::after_spawn_live_with_reason(pid, reason);
1412                }
1413            } else {
1414                let pid = child.pid();
1415                self.children
1416                    .lock()
1417                    .unwrap_or_else(std::sync::PoisonError::into_inner)
1418                    .insert(attempt.id, child);
1419                return ProcessStartFailure::after_spawn_live(pid);
1420            }
1421        }
1422        #[cfg(test)]
1423        if reaped {
1424            self.post_spawn_reaps
1425                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1426        }
1427        #[cfg(not(test))]
1428        let _ = reaped;
1429        ProcessStartFailure::after_spawn_stopped()
1430    }
1431
1432    fn identity_path(attempt: &RunnerAttempt) -> PathBuf {
1433        attempt.runtime_path().join(IDENTITY_FILE)
1434    }
1435
1436    fn fallback_identity_path(attempt: &RunnerAttempt) -> PathBuf {
1437        attempt.runtime_path().join(FALLBACK_IDENTITY_FILE)
1438    }
1439
1440    fn unresolved_process_path(attempt: &RunnerAttempt) -> PathBuf {
1441        attempt.runtime_path().join(UNRESOLVED_PROCESS_FILE)
1442    }
1443
1444    fn remove_identity_files(attempt: &RunnerAttempt) {
1445        let _ = fs::remove_file(Self::identity_path(attempt));
1446        let _ = fs::remove_file(Self::fallback_identity_path(attempt));
1447        let _ = fs::remove_file(Self::unresolved_process_path(attempt));
1448    }
1449
1450    fn persist_identity(&self, attempt: &RunnerAttempt, bytes: &[u8]) -> std::io::Result<()> {
1451        self.persist_identity_at(&Self::identity_path(attempt), bytes)
1452    }
1453
1454    fn persist_fallback_identity(
1455        &self,
1456        attempt: &RunnerAttempt,
1457        bytes: &[u8],
1458    ) -> std::io::Result<()> {
1459        self.persist_identity_at(&Self::fallback_identity_path(attempt), bytes)
1460    }
1461
1462    fn persist_identity_at(&self, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
1463        #[cfg(test)]
1464        if self.faults_at(PostSpawnBoundary::IdentityWrite) {
1465            return Err(std::io::Error::other("injected identity write failure"));
1466        }
1467        write_durable_file(path, bytes)
1468    }
1469
1470    fn intent_path(attempt: &RunnerAttempt) -> PathBuf {
1471        attempt.runtime_path().join(TERMINATE_INTENT_FILE)
1472    }
1473
1474    fn read_identity(attempt: &RunnerAttempt) -> Result<Option<ProcessIdentity>, FailureReason> {
1475        match Self::read_identity_at(&Self::identity_path(attempt))? {
1476            Some(identity) => Ok(Some(identity)),
1477            None => Self::read_identity_at(&Self::fallback_identity_path(attempt)),
1478        }
1479    }
1480
1481    fn read_identity_at(path: &Path) -> Result<Option<ProcessIdentity>, FailureReason> {
1482        match fs::read(path) {
1483            Ok(bytes) => serde_json::from_slice(&bytes)
1484                .map(Some)
1485                .map_err(|_| FailureReason::Other("process identity journal is unreadable".into())),
1486            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
1487            Err(_) => Err(FailureReason::Other(
1488                "process identity journal could not be read".into(),
1489            )),
1490        }
1491    }
1492}
1493
1494impl ProcessSupervisor for NativeProcesses {
1495    fn spawn(
1496        &self,
1497        attempt: &RunnerAttempt,
1498        config: &EncodedJitConfig,
1499    ) -> Result<u32, ProcessStartFailure> {
1500        let handoff = RestrictiveHandoff::create(
1501            attempt.runtime_path(),
1502            SecretString::from(config.expose().to_owned()),
1503        )
1504        .map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
1505        #[cfg(windows)]
1506        let program = attempt
1507            .runtime_path()
1508            .join("bin")
1509            .join("Runner.Listener.exe");
1510        #[cfg(not(windows))]
1511        let program = attempt.runtime_path().join("bin").join("Runner.Listener");
1512        // Checked after the handoff exists on purpose: the error path below is
1513        // a real post-handoff launch failure, and unwinding must delete it.
1514        if !program.is_file() {
1515            return Err(ProcessStartFailure::before_spawn(
1516                FailureReason::ProcessStartFailed,
1517            ));
1518        }
1519        #[cfg(test)]
1520        let spec = if self
1521            .use_long_lived_test_listener
1522            .load(std::sync::atomic::Ordering::SeqCst)
1523        {
1524            SpawnSpec::new(program)
1525                .args([
1526                    "--ignored",
1527                    "--exact",
1528                    "lifecycle::tests::long_lived_native_listener_helper",
1529                    "--nocapture",
1530                ])
1531                .env(
1532                    "RUNNER_MANAGER_TEST_LISTENER_READY",
1533                    attempt.runtime_path().join(TEST_LISTENER_READY),
1534                )
1535                .working_dir(attempt.runtime_path())
1536        } else {
1537            runner_listener_spec(program, attempt.runtime_path())
1538        };
1539        #[cfg(not(test))]
1540        let spec = runner_listener_spec(program, attempt.runtime_path());
1541        let child = spec
1542            .spawn_runner_with_handoff(&handoff)
1543            .map_err(|_| ProcessStartFailure::before_spawn(FailureReason::ProcessStartFailed))?;
1544        // The payload is gone before any state saying "starting" is persisted.
1545        #[cfg(test)]
1546        if self.faults_at(PostSpawnBoundary::HandoffDelete) {
1547            drop(handoff);
1548            return Err(self.abort_spawned_child(child, attempt, false));
1549        }
1550        if handoff.delete().is_err() {
1551            return Err(self.abort_spawned_child(child, attempt, false));
1552        }
1553        #[cfg(test)]
1554        if self.faults_at(PostSpawnBoundary::IdentitySerialize) {
1555            return Err(self.abort_spawned_child(child, attempt, false));
1556        }
1557        let identity = match serde_json::to_vec(child.identity()) {
1558            Ok(identity) => identity,
1559            Err(_) => {
1560                return Err(self.abort_spawned_child(child, attempt, false));
1561            }
1562        };
1563        if self.persist_identity(attempt, &identity).is_err() {
1564            return Err(self.abort_spawned_child(child, attempt, true));
1565        }
1566        let pid = child.pid();
1567        #[cfg(test)]
1568        if self.faults_at(PostSpawnBoundary::ChildMapInsert) {
1569            return Err(self.abort_spawned_child(child, attempt, true));
1570        }
1571        let mut children = self
1572            .children
1573            .lock()
1574            .unwrap_or_else(std::sync::PoisonError::into_inner);
1575        children.insert(attempt.id, child);
1576        Ok(pid)
1577    }
1578
1579    fn is_alive(&self, attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
1580        let mut children = self
1581            .children
1582            .lock()
1583            .unwrap_or_else(std::sync::PoisonError::into_inner);
1584        if let Some(child) = children.get_mut(&attempt.id) {
1585            return match child
1586                .try_exit_status()
1587                .map_err(|_| FailureReason::Other("runner process could not be observed".into()))?
1588            {
1589                None => Ok(true),
1590                Some(status) => {
1591                    if let Ok(mut exits) = self.successful_exits.lock() {
1592                        exits.insert(attempt.id, status.success());
1593                    }
1594                    Ok(false)
1595                }
1596            };
1597        }
1598        let Some(identity) = Self::read_identity(attempt)? else {
1599            if attempt.process_id().is_some() || Self::unresolved_process_path(attempt).is_file() {
1600                return Err(FailureReason::Other(
1601                    "runner process identity is missing; refusing recovery until the process is resolved"
1602                        .into(),
1603                ));
1604            }
1605            return Ok(false);
1606        };
1607        match identity.recheck() {
1608            Ok(Adoption::Live) => Ok(true),
1609            Ok(Adoption::Gone | Adoption::PidRecycled { .. }) => Ok(false),
1610            Err(_) => Ok(false),
1611        }
1612    }
1613
1614    fn recovered_pid(&self, attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
1615        Ok(Self::read_identity(attempt)?.map(|identity| identity.pid()))
1616    }
1617
1618    fn completed_successfully(&self, attempt: &RunnerAttempt) -> bool {
1619        self.successful_exits
1620            .lock()
1621            .ok()
1622            .and_then(|exits| exits.get(&attempt.id).copied())
1623            .unwrap_or(false)
1624    }
1625
1626    fn record_terminate_intent(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
1627        let path = Self::intent_path(attempt);
1628        write_durable_file(&path, b"registration-timeout\n")
1629            .map_err(|_| FailureReason::Other("terminate intent could not be journalled".into()))
1630    }
1631
1632    fn has_terminate_intent(&self, attempt: &RunnerAttempt) -> bool {
1633        Self::intent_path(attempt).is_file()
1634    }
1635
1636    fn terminate(&self, attempt: &RunnerAttempt) -> Result<(), FailureReason> {
1637        let mut children = self
1638            .children
1639            .lock()
1640            .unwrap_or_else(std::sync::PoisonError::into_inner);
1641        if let Some(child) = children.get_mut(&attempt.id) {
1642            child
1643                .stop(Duration::from_secs(10))
1644                .map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?;
1645            return Ok(());
1646        }
1647        let Some(identity) = Self::read_identity(attempt)? else {
1648            return Ok(());
1649        };
1650        match identity
1651            .terminate(Duration::from_secs(10))
1652            .map_err(|_| FailureReason::Other("runner process could not be stopped".into()))?
1653        {
1654            Termination::Terminated | Termination::AlreadyGone => Ok(()),
1655            Termination::RefusedPidRecycled { .. } => Err(FailureReason::Other(
1656                "runner PID was recycled; refusing to signal it".into(),
1657            )),
1658        }
1659    }
1660}
1661
1662pub struct LifecyclePorts {
1663    pub store: Arc<dyn Store>,
1664    pub github: Arc<dyn LifecycleGithub>,
1665    pub packages: Arc<dyn RuntimePackages>,
1666    pub processes: Arc<dyn ProcessSupervisor>,
1667    pub clock: Arc<dyn Clock>,
1668    pub demand: Arc<dyn DemandPersistence>,
1669    pub delay: Arc<dyn RetryDelay>,
1670    pub events: Arc<dyn AttemptEventSink>,
1671    pub reconcile_events: Arc<dyn EventSink>,
1672}
1673
1674impl fmt::Debug for LifecyclePorts {
1675    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1676        f.debug_struct("LifecyclePorts")
1677            .field("store", &self.store)
1678            .field("github", &self.github)
1679            .field("packages", &self.packages)
1680            .field("processes", &self.processes)
1681            .finish_non_exhaustive()
1682    }
1683}
1684
1685#[derive(Debug, thiserror::Error)]
1686pub enum LifecycleError {
1687    #[error("attempt journal operation failed")]
1688    Journal,
1689    #[error("attempt {0} is not in the journal")]
1690    Missing(AttemptId),
1691    #[error("attempt lifecycle transition was refused")]
1692    Transition,
1693    #[error("startup recovery has not completed")]
1694    RecoveryIncomplete,
1695    /// A persistent slot could not be proven safe to scrub, so the attempt keeps
1696    /// its uncleaned state and, with it, its slot lease.
1697    ///
1698    /// Separate from [`Self::Failed`] because the two demand opposite handling
1699    /// from the same call sites. A failure aborts the pass; a quarantine must
1700    /// not, or one stuck slot would stop the host launching anything at all —
1701    /// which is precisely what `04-security-recovery.md` rules out when it says
1702    /// such an attempt "does not count as active host capacity" and "recovery
1703    /// retries the same cleanup".
1704    #[error("the persistent slot was not cleaned: {detail}")]
1705    SlotQuarantined {
1706        /// The closed-vocabulary event field; never operator or workflow text.
1707        class: &'static str,
1708        detail: String,
1709    },
1710    #[error("runner lifecycle failed: {0}")]
1711    Failed(FailureReason),
1712}
1713
1714impl LifecycleError {
1715    fn reason(&self) -> FailureReason {
1716        match self {
1717            Self::Failed(reason) => reason.clone(),
1718            Self::RecoveryIncomplete => FailureReason::Other("startup recovery incomplete".into()),
1719            Self::Journal => FailureReason::Other("attempt journal operation failed".into()),
1720            Self::Missing(_) => FailureReason::Other("attempt disappeared from the journal".into()),
1721            Self::Transition => FailureReason::Other("attempt transition was refused".into()),
1722            // Rendered through `Display` rather than a second copy of the same
1723            // sentence, so the two cannot drift apart.
1724            Self::SlotQuarantined { .. } => FailureReason::Other(self.to_string()),
1725        }
1726    }
1727}
1728
1729/// Production implementation of e1's launcher port.
1730#[derive(Debug)]
1731pub struct LifecycleLauncher {
1732    host_id: HostId,
1733    app_paths: runner_manager_platform::paths::AppPaths,
1734    diagnostics_root: PathBuf,
1735    runner_group_id: u64,
1736    timeouts: RecoveryTimeouts,
1737    retry: RetryPolicy,
1738    cancel: CancelToken,
1739    ports: LifecyclePorts,
1740    recovery_complete: Mutex<bool>,
1741    versions: Mutex<BTreeMap<AttemptId, RunnerVersion>>,
1742    pending_replacements: Mutex<BTreeMap<AttemptId, ReplacementIntent>>,
1743}
1744
1745#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1746enum ReconcileProgress {
1747    Reconciled,
1748    Deferred,
1749    Replacement {
1750        attempt: AttemptId,
1751        operation: &'static str,
1752    },
1753}
1754
1755impl LifecycleLauncher {
1756    #[must_use]
1757    pub fn new(
1758        host_id: HostId,
1759        app_paths: runner_manager_platform::paths::AppPaths,
1760        diagnostics_root: impl Into<PathBuf>,
1761        runner_group_id: u64,
1762        timeouts: RecoveryTimeouts,
1763        retry: RetryPolicy,
1764        ports: LifecyclePorts,
1765    ) -> Self {
1766        Self {
1767            host_id,
1768            app_paths,
1769            diagnostics_root: diagnostics_root.into(),
1770            runner_group_id,
1771            timeouts,
1772            retry,
1773            cancel: CancelToken::new(),
1774            ports,
1775            recovery_complete: Mutex::new(false),
1776            versions: Mutex::new(BTreeMap::new()),
1777            pending_replacements: Mutex::new(BTreeMap::new()),
1778        }
1779    }
1780
1781    /// Reconcile the entire journal before allowing a launch.  Unknown-policy
1782    /// attempts are left untouched rather than being acted on without an
1783    /// ownership proof.
1784    pub async fn recover_startup(
1785        &self,
1786        policies: &[ScalePolicy],
1787    ) -> Result<Vec<ReplacementIntent>, LifecycleError> {
1788        let by_id: BTreeMap<_, _> = policies.iter().map(|policy| (policy.id, policy)).collect();
1789        let attempts = self
1790            .ports
1791            .store
1792            .attempts()
1793            .map_err(|_| LifecycleError::Journal)?;
1794        let mut unresolved = false;
1795        for attempt in attempts {
1796            let Some(policy) = by_id.get(&attempt.policy_id) else {
1797                if !attempt.is_terminal() && attempt.state() != AttemptState::Cleaned {
1798                    unresolved = true;
1799                }
1800                continue;
1801            };
1802            authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
1803            match self.reconcile_one(policy, attempt).await? {
1804                ReconcileProgress::Deferred => unresolved = true,
1805                ReconcileProgress::Replacement { attempt, operation } => {
1806                    self.pending_replacements
1807                        .lock()
1808                        .map_err(|_| LifecycleError::Journal)?
1809                        .insert(
1810                            attempt,
1811                            ReplacementIntent {
1812                                policy: policy.id,
1813                                previous_attempt: attempt,
1814                                operation,
1815                            },
1816                        );
1817                }
1818                ReconcileProgress::Reconciled => {}
1819            }
1820        }
1821        if unresolved {
1822            return Err(LifecycleError::RecoveryIncomplete);
1823        }
1824        *self
1825            .recovery_complete
1826            .lock()
1827            .map_err(|_| LifecycleError::Journal)? = true;
1828        Ok(self
1829            .pending_replacements
1830            .lock()
1831            .map_err(|_| LifecycleError::Journal)?
1832            .values()
1833            .copied()
1834            .collect())
1835    }
1836
1837    /// Supervise all attempts of one policy during an ordinary poll.
1838    pub async fn supervise(
1839        &self,
1840        policy: &ScalePolicy,
1841    ) -> Result<Vec<ReplacementIntent>, LifecycleError> {
1842        let mut replacements = Vec::new();
1843        self.pending_replacements
1844            .lock()
1845            .map_err(|_| LifecycleError::Journal)?
1846            .retain(|_, intent| {
1847                if intent.policy == policy.id {
1848                    replacements.push(*intent);
1849                    false
1850                } else {
1851                    true
1852                }
1853            });
1854        let attempts = self
1855            .ports
1856            .store
1857            .attempts_for_policy(policy.id)
1858            .map_err(|_| LifecycleError::Journal)?;
1859        for attempt in attempts {
1860            authorize(self.host_id, policy, &attempt).map_err(|_| LifecycleError::Journal)?;
1861            if let ReconcileProgress::Replacement { attempt, operation } =
1862                self.reconcile_one(policy, attempt).await?
1863            {
1864                replacements.push(ReplacementIntent {
1865                    policy: policy.id,
1866                    previous_attempt: attempt,
1867                    operation,
1868                });
1869            }
1870        }
1871        Ok(replacements)
1872    }
1873
1874    async fn reconcile_one(
1875        &self,
1876        policy: &ScalePolicy,
1877        mut attempt: RunnerAttempt,
1878    ) -> Result<ReconcileProgress, LifecycleError> {
1879        if attempt.state() == AttemptState::Cleaned {
1880            return Ok(ReconcileProgress::Reconciled);
1881        }
1882        if attempt.is_terminal() {
1883            self.clean_or_quarantine(&mut attempt)?;
1884            return Ok(ReconcileProgress::Reconciled);
1885        }
1886        let process_alive = self
1887            .ports
1888            .processes
1889            .is_alive(&attempt)
1890            .map_err(LifecycleError::Failed)?;
1891        let github = self
1892            .ports
1893            .github
1894            .observe(&policy.target, attempt.id, &self.cancel)
1895            .await;
1896
1897        // If the agent died after GitHub accepted the registration but before
1898        // the non-secret runner-id sidecar landed, inventory closes the gap.
1899        // Persist it before making any state decision so a second crash moves
1900        // the boundary forward rather than repeating it.
1901        if let Some(runner_id) = github.runner_id
1902            && read_runner_id(attempt.runtime_path()).is_none()
1903        {
1904            write_runner_id(attempt.runtime_path(), runner_id)?;
1905            self.ports
1906                .events
1907                .emit(AttemptEvent::RemoteIdentityRecovered {
1908                    attempt: attempt.id,
1909                    runner_id,
1910                });
1911        }
1912
1913        // The child identity is synced before `spawn` returns.  If the agent
1914        // crashed before the following `starting` journal write, recover that
1915        // exact PID and take the legal `jit_received -> starting` edge before
1916        // applying GitHub's authoritative idle/busy observation below.
1917        if attempt.state() == AttemptState::JitReceived
1918            && process_alive
1919            && let Some(pid) = self
1920                .ports
1921                .processes
1922                .recovered_pid(&attempt)
1923                .map_err(LifecycleError::Failed)?
1924        {
1925            attempt
1926                .started(pid, self.ports.clock.now())
1927                .map_err(|_| LifecycleError::Transition)?;
1928            self.record(&attempt)?;
1929        }
1930
1931        // A one-shot child owned by this invocation exited successfully after
1932        // GitHub had already reported it busy, and its ephemeral registration
1933        // is now gone.  This concludes the *runner attempt*, never the workflow
1934        // outcome; GitHub remains authoritative for that outcome.
1935        if attempt.state() == AttemptState::Busy
1936            && !process_alive
1937            && github.status == GithubRunnerObservation::NotRegistered
1938            && self.ports.processes.completed_successfully(&attempt)
1939        {
1940            self.conclude(&mut attempt, AttemptOutcome::CompletedJob)?;
1941            self.clean_or_quarantine(&mut attempt)?;
1942            return Ok(ReconcileProgress::Reconciled);
1943        }
1944
1945        // This durable mark is more authoritative than a later observation
1946        // which cannot distinguish an agent kill from a crash.
1947        if self.ports.processes.has_terminate_intent(&attempt) && !process_alive {
1948            self.deregister_runner(policy, &attempt).await;
1949            self.conclude(
1950                &mut attempt,
1951                AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout),
1952            )?;
1953            self.clean_or_quarantine(&mut attempt)?;
1954            return Ok(ReconcileProgress::Replacement {
1955                attempt: attempt.id,
1956                operation: "registration_timeout_replacement",
1957            });
1958        }
1959
1960        // A remote registration with no surviving process has lost its
1961        // one-shot JIT secret.  Walking it to `starting` would require inventing
1962        // a PID; retrying the same registration would require inventing the
1963        // secret.  Record the configuration as expired and let the bounded
1964        // replacement path request a fresh one only if demand remains.
1965        if matches!(
1966            attempt.state(),
1967            AttemptState::Allocated | AttemptState::JitReceived
1968        ) && !process_alive
1969            && matches!(github.status, GithubRunnerObservation::Registered { .. })
1970        {
1971            if attempt.state() == AttemptState::Allocated {
1972                attempt
1973                    .jit_received(self.ports.clock.now())
1974                    .map_err(|_| LifecycleError::Transition)?;
1975                self.record(&attempt)?;
1976            }
1977            self.deregister_runner(policy, &attempt).await;
1978            self.conclude(
1979                &mut attempt,
1980                AttemptOutcome::failed(FailureReason::JitExpired),
1981            )?;
1982            self.clean_or_quarantine(&mut attempt)?;
1983            return Ok(ReconcileProgress::Replacement {
1984                attempt: attempt.id,
1985                operation: "jit_expired_replacement",
1986            });
1987        }
1988
1989        match recovery_decision(
1990            &attempt,
1991            RecoveryObservation {
1992                process_alive,
1993                github: github.status,
1994            },
1995            self.timeouts,
1996            self.ports.clock.as_ref(),
1997        ) {
1998            RecoveryDecision::Nothing | RecoveryDecision::Wait => Ok(ReconcileProgress::Reconciled),
1999            RecoveryDecision::Defer => Ok(ReconcileProgress::Deferred),
2000            RecoveryDecision::Adopt => {
2001                self.ports.events.emit(AttemptEvent::Adopted {
2002                    attempt: attempt.id,
2003                });
2004                Ok(ReconcileProgress::Reconciled)
2005            }
2006            RecoveryDecision::Clean => {
2007                self.clean_or_quarantine(&mut attempt)?;
2008                Ok(ReconcileProgress::Reconciled)
2009            }
2010            RecoveryDecision::Observe(state) => {
2011                let runner_id = attempt
2012                    .github_runner_id()
2013                    .or(github.runner_id)
2014                    .or_else(|| read_runner_id(attempt.runtime_path()))
2015                    .ok_or(LifecycleError::Transition)?;
2016                match state {
2017                    AttemptState::JitReceived => attempt
2018                        .jit_received(self.ports.clock.now())
2019                        .map_err(|_| LifecycleError::Transition)?,
2020                    AttemptState::Starting => {
2021                        let pid = attempt.process_id().ok_or(LifecycleError::Transition)?;
2022                        attempt
2023                            .started(pid, self.ports.clock.now())
2024                            .map_err(|_| LifecycleError::Transition)?;
2025                    }
2026                    AttemptState::Idle => attempt
2027                        .registered_idle(runner_id, self.ports.clock.now())
2028                        .map_err(|_| LifecycleError::Transition)?,
2029                    AttemptState::Busy => attempt
2030                        .assigned_job(runner_id, self.ports.clock.now())
2031                        .map_err(|_| LifecycleError::Transition)?,
2032                    _ => return Err(LifecycleError::Transition),
2033                }
2034                self.record(&attempt)?;
2035                Ok(ReconcileProgress::Reconciled)
2036            }
2037            RecoveryDecision::Conclude(outcome) => {
2038                let replacement = replacement_operation(&outcome);
2039                // Only when GitHub still holds one. Every other conclusion here
2040                // was reached *because* the observation was `NotRegistered`, and
2041                // spending a DELETE to be told so again would put a request per
2042                // concluded attempt on a budget `rest.rs` prices to the request.
2043                if matches!(github.status, GithubRunnerObservation::Registered { .. }) {
2044                    self.deregister_runner(policy, &attempt).await;
2045                }
2046                self.conclude(&mut attempt, outcome)?;
2047                self.clean_or_quarantine(&mut attempt)?;
2048                Ok(
2049                    replacement.map_or(ReconcileProgress::Reconciled, |operation| {
2050                        ReconcileProgress::Replacement {
2051                            attempt: attempt.id,
2052                            operation,
2053                        }
2054                    }),
2055                )
2056            }
2057            RecoveryDecision::Terminate(payload) => {
2058                // The mark is synced first, and what proves the process died is
2059                // the `is_alive` re-read below -- not the outcome recorded after
2060                // it. Which outcome that is depends on why the termination was
2061                // ordered, and only the payload knows: a `starting` runner that
2062                // never registered is a failure this agent then stopped, while
2063                // an `idle` one past its timeout is flow 2.7's surplus exit and
2064                // no failure at all. Hardcoding the first reason here labelled
2065                // the second as a registration timeout and asked the allocator
2066                // for a replacement to boot.
2067                let idle_exit = payload.is_idle_exit();
2068                self.ports
2069                    .processes
2070                    .record_terminate_intent(&attempt)
2071                    .map_err(LifecycleError::Failed)?;
2072                self.ports.events.emit(AttemptEvent::TerminateIntent {
2073                    attempt: attempt.id,
2074                });
2075                self.ports
2076                    .processes
2077                    .terminate(&attempt)
2078                    .map_err(LifecycleError::Failed)?;
2079                if self
2080                    .ports
2081                    .processes
2082                    .is_alive(&attempt)
2083                    .map_err(LifecycleError::Failed)?
2084                {
2085                    return Ok(ReconcileProgress::Deferred);
2086                }
2087                self.ports.events.emit(AttemptEvent::Terminated {
2088                    attempt: attempt.id,
2089                });
2090                // The registration-timeout path keeps deriving its own reason
2091                // rather than applying the payload: on the pass that reads the
2092                // journalled mark back the process is dead, and
2093                // `TerminatedAfterRegistrationTimeout` is the reason that stays
2094                // true of a dead process. See `RecoveryDecision::Terminate`.
2095                let outcome = if idle_exit {
2096                    AttemptOutcome::ExitedIdleWithoutWork
2097                } else {
2098                    AttemptOutcome::failed(FailureReason::TerminatedAfterRegistrationTimeout)
2099                };
2100                self.deregister_runner(policy, &attempt).await;
2101                self.conclude(&mut attempt, outcome)?;
2102                self.clean_or_quarantine(&mut attempt)?;
2103                // A surplus runner is not replaced. It was stopped precisely
2104                // because the work it was started for went elsewhere; asking the
2105                // allocator for another one rebuilds it every idle timeout.
2106                if idle_exit {
2107                    Ok(ReconcileProgress::Reconciled)
2108                } else {
2109                    Ok(ReconcileProgress::Replacement {
2110                        attempt: attempt.id,
2111                        operation: "registration_timeout_replacement",
2112                    })
2113                }
2114            }
2115        }
2116    }
2117
2118    fn record(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2119        self.ports
2120            .store
2121            .record_attempt(attempt)
2122            .map_err(|_| LifecycleError::Journal)?;
2123        self.ports.events.emit(AttemptEvent::State {
2124            attempt: attempt.id,
2125            state: attempt.state(),
2126        });
2127        Ok(())
2128    }
2129
2130    /// Remove the GitHub registration an attempt is about to leave behind.
2131    ///
2132    /// # Why this is not fallible, and does not block the conclusion
2133    ///
2134    /// GitHub retires an ephemeral runner itself once that runner *completes a
2135    /// job*, and for the ordinary path that is the whole story. The paths that
2136    /// reach here are the ones where it does not: a runner stopped before it
2137    /// was ever assigned work, a registration whose process died still holding
2138    /// it, a JIT configuration that expired. Nothing else deletes those, and
2139    /// before this existed nothing did — they accumulated in the target's
2140    /// runner settings, one row per attempt, for the life of the repository.
2141    ///
2142    /// It returns `()` rather than a `Result` because the alternative is worse
2143    /// in both directions. The attempt is over: its process is gone and its
2144    /// slot has to come back, so a failed delete may not abort the conclusion
2145    /// or the host leaks capacity every time GitHub is unreachable. And a
2146    /// registration that outlives this call is not lost — it is exactly the
2147    /// `Registered` + dead-process observation that
2148    /// [`AttemptOutcome::Orphaned`] already names, which a later pass can still
2149    /// see. So a failure is logged and stepped over, deliberately.
2150    async fn deregister_runner(&self, policy: &ScalePolicy, attempt: &RunnerAttempt) {
2151        let Some(runner_id) = attempt
2152            .github_runner_id()
2153            .or_else(|| read_runner_id(attempt.runtime_path()))
2154        else {
2155            return;
2156        };
2157        if self
2158            .ports
2159            .github
2160            .deregister(&policy.target, runner_id, &self.cancel)
2161            .await
2162        {
2163            self.ports.events.emit(AttemptEvent::Deregistered {
2164                attempt: attempt.id,
2165                runner_id,
2166            });
2167        } else {
2168            tracing::warn!(
2169                attempt = %attempt.id,
2170                runner_id,
2171                "the runner registration could not be removed from GitHub; it will show in the \
2172                 target's runner settings until GitHub retires it or a later pass removes it"
2173            );
2174        }
2175    }
2176
2177    fn conclude(
2178        &self,
2179        attempt: &mut RunnerAttempt,
2180        outcome: AttemptOutcome,
2181    ) -> Result<(), LifecycleError> {
2182        attempt
2183            .conclude(outcome.clone(), self.ports.clock.now())
2184            .map_err(|_| LifecycleError::Transition)?;
2185        self.record(attempt)?;
2186        self.ports.events.emit(AttemptEvent::Concluded {
2187            attempt: attempt.id,
2188            outcome: OutcomeKind::of(&outcome),
2189        });
2190        Ok(())
2191    }
2192
2193    /// Clean a concluded attempt, tolerating a quarantined persistent slot.
2194    ///
2195    /// The quarantine is reported and stepped over rather than raised, because
2196    /// raising it aborts the whole pass: on the startup path that leaves
2197    /// `recovery_complete` false and stops the host launching anything, which is
2198    /// the opposite of `04-security-recovery.md`'s "it does not count as active
2199    /// host capacity" and "recovery retries the same cleanup". The attempt keeps
2200    /// its state, so it keeps its slot lease and its directory, and the next
2201    /// pass — [`crate::reconcile::Reconciler`]'s terminal sweep on every poll,
2202    /// or the next startup — attempts exactly the same cleanup again.
2203    ///
2204    /// Only a *quarantine* is tolerated. A journal failure or a package lease
2205    /// that cannot be released still propagates: those are not one slot's
2206    /// problem.
2207    fn clean_or_quarantine(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
2208        match self.clean_attempt(attempt) {
2209            Err(LifecycleError::SlotQuarantined { class, .. }) => {
2210                self.ports
2211                    .reconcile_events
2212                    .emit(LifecycleEvent::AttemptCleanFailed {
2213                        policy: attempt.policy_id,
2214                        attempt: attempt.id,
2215                        reason: class,
2216                    });
2217                Ok(())
2218            }
2219            other => other,
2220        }
2221    }
2222
2223    fn clean_attempt(&self, attempt: &mut RunnerAttempt) -> Result<(), LifecycleError> {
2224        let outcome = attempt
2225            .outcome()
2226            .cloned()
2227            .ok_or(LifecycleError::Transition)?;
2228        self.preserve_diagnostics(attempt, &outcome)?;
2229        self.scrub_workspace(attempt)?;
2230        self.ports
2231            .packages
2232            .release(attempt.id)
2233            .map_err(LifecycleError::Failed)?;
2234        attempt
2235            .clean(self.ports.clock.now())
2236            .map_err(|_| LifecycleError::Transition)?;
2237        self.record(attempt)?;
2238        let kind = OutcomeKind::of(&outcome);
2239        self.ports.events.emit(AttemptEvent::Cleaned {
2240            attempt: attempt.id,
2241            outcome: kind,
2242        });
2243        self.ports
2244            .reconcile_events
2245            .emit(LifecycleEvent::AttemptCleaned {
2246                policy: attempt.policy_id,
2247                attempt: attempt.id,
2248                outcome: kind,
2249            });
2250        Ok(())
2251    }
2252
2253    /// Undo an attempt's placement by the algorithm its journalled workspace
2254    /// kind makes legal (`02-target-architecture.md`, "Cleanup and recovery").
2255    ///
2256    /// The dispatch is on the *journal*, never on what the directory looks like
2257    /// now. A slot whose `_work` was replaced by a junction is still scrubbed as
2258    /// a slot rather than removed whole, and a disposable directory that happens
2259    /// to contain a `_work` still goes whole rather than being spared: the
2260    /// workspace kind is immutable precisely so that the shape of a directory a
2261    /// workflow can write to cannot choose the algorithm applied to it.
2262    fn scrub_workspace(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2263        #[cfg(test)]
2264        {
2265            // The two contamination mutants `f1` drives the security gates with.
2266            // They sit in front of the dispatch rather than inside one arm so
2267            // that a skipped cleanup is equally observable in both modes.
2268            if matches!(
2269                std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref(),
2270                Ok("skip_workspace_cleanup" | "reuse_job_workspace")
2271            ) {
2272                return Ok(());
2273            }
2274        }
2275        match attempt.workspace() {
2276            AttemptWorkspace::Ephemeral => match remove_runtime_tree(attempt.runtime_path()) {
2277                Ok(()) => Ok(()),
2278                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
2279                Err(_) => Err(LifecycleError::Failed(FailureReason::Other(
2280                    "attempt workspace could not be removed".into(),
2281                ))),
2282            },
2283            AttemptWorkspace::PersistentSlot { slot } => self.scrub_persistent_slot(attempt, slot),
2284        }
2285    }
2286
2287    /// Retain exactly `_work` and prove everything else is gone.
2288    ///
2289    /// The seven ordered checks of `04-security-recovery.md`, "Safe path
2290    /// handling": the journalled root and slot, a surviving policy's agreement,
2291    /// lexical and canonical containment, a literal enumeration, the one-entry
2292    /// allowlist, and a verification pass before the caller releases the package
2293    /// lease and marks the attempt cleaned. Any of them may refuse, and a
2294    /// refusal removes nothing after it.
2295    ///
2296    /// A slot directory that is already gone is not a refusal. There is nothing
2297    /// to scrub and nothing to prove absent, which is the same tolerance the
2298    /// disposable arm has always had for a runtime that vanished under it.
2299    fn scrub_persistent_slot(
2300        &self,
2301        attempt: &RunnerAttempt,
2302        slot: NonZeroU16,
2303    ) -> Result<(), LifecycleError> {
2304        // The policy is read from the journal, never inferred from the
2305        // directory tree, and its absence is legal: `Reconciler` cleans every
2306        // concluded attempt whether or not its policy still exists, and this
2307        // one's root is already journalled on the attempt itself.
2308        let configured = self
2309            .ports
2310            .store
2311            .policy(attempt.policy_id)
2312            .map_err(|_| LifecycleError::Journal)?
2313            .and_then(|policy| match policy.workspace_policy() {
2314                WorkspacePolicy::Persistent { root } => Some(root.clone()),
2315                WorkspacePolicy::Ephemeral => None,
2316            });
2317        let runtime = attempt.runtime_path();
2318        self.quarantine_on_refusal(
2319            attempt,
2320            verify_journalled_slot(runtime, slot, configured.as_ref())
2321                .and_then(|()| slot_is_present(runtime))
2322                .and_then(|present| {
2323                    if present {
2324                        scrub_slot_entries(runtime).and_then(|()| verify_slot_scrubbed(runtime))
2325                    } else {
2326                        Ok(())
2327                    }
2328                }),
2329        )
2330    }
2331
2332    /// Turn a slot refusal into the error that keeps the attempt uncleaned.
2333    ///
2334    /// The warning is emitted here, at the one place a quarantine is minted, so
2335    /// that both routes out of cleanup carry it: the reconciler's terminal sweep,
2336    /// which receives the error through the launcher port, and
2337    /// [`Self::clean_or_quarantine`], which swallows it to keep the pass alive.
2338    /// Everything logged is either a path this product configured or a constant
2339    /// from this module — see [`SlotQuarantine`] for why that matters.
2340    fn quarantine_on_refusal(
2341        &self,
2342        attempt: &RunnerAttempt,
2343        outcome: Result<(), SlotQuarantine>,
2344    ) -> Result<(), LifecycleError> {
2345        let Err(quarantine) = outcome else {
2346            return Ok(());
2347        };
2348        let detail = quarantine.to_string();
2349        tracing::warn!(
2350            attempt = %attempt.id,
2351            policy = %attempt.policy_id,
2352            slot = attempt.workspace().slot_number(),
2353            refusal = quarantine.refusal.class(),
2354            "{detail}"
2355        );
2356        Err(LifecycleError::SlotQuarantined {
2357            class: quarantine.refusal.class(),
2358            detail,
2359        })
2360    }
2361
2362    fn preserve_diagnostics(
2363        &self,
2364        attempt: &RunnerAttempt,
2365        outcome: &AttemptOutcome,
2366    ) -> Result<(), LifecycleError> {
2367        fs::create_dir_all(&self.diagnostics_root).map_err(|_| {
2368            LifecycleError::Failed(FailureReason::Other(
2369                "diagnostics directory could not be created".into(),
2370            ))
2371        })?;
2372        // Intentionally constructed from typed local facts, not runner output.
2373        // Raw child output can contain workflow secrets and is never copied.
2374        let diagnostic = format!(
2375            "attempt_id={}\npolicy_id={}\noutcome={}\n",
2376            attempt.id,
2377            attempt.policy_id,
2378            OutcomeKind::of(outcome).as_str()
2379        );
2380        fs::write(
2381            self.diagnostics_root.join(format!("{}.log", attempt.id)),
2382            diagnostic,
2383        )
2384        .map_err(|_| {
2385            LifecycleError::Failed(FailureReason::Other(
2386                "redacted diagnostics could not be preserved".into(),
2387            ))
2388        })
2389    }
2390
2391    async fn materialize_with_retry(
2392        &self,
2393        policy: &ScalePolicy,
2394        attempt: &RunnerAttempt,
2395    ) -> Result<RunnerVersion, FailureReason> {
2396        let mut issued = 0_u32;
2397        loop {
2398            issued = issued.saturating_add(1);
2399            match self.ports.packages.materialize(attempt).await {
2400                Ok(version) => return Ok(version),
2401                Err(reason)
2402                    if package_failure_is_terminal(&reason)
2403                        || issued >= self.retry.max_attempts.max(1) =>
2404                {
2405                    return Err(reason);
2406                }
2407                Err(reason) => {
2408                    if !self.ports.demand.persists(policy.id).await {
2409                        return Err(reason);
2410                    }
2411                    let delay = self.retry.delay(issued);
2412                    self.ports.events.emit(AttemptEvent::Retry {
2413                        attempt: attempt.id,
2414                        operation: "package_materialization",
2415                        delay,
2416                    });
2417                    self.ports.delay.wait(delay).await;
2418                    if !self.ports.demand.persists(policy.id).await {
2419                        return Err(reason);
2420                    }
2421                }
2422            }
2423        }
2424    }
2425
2426    async fn register_with_retry(
2427        &self,
2428        policy: &ScalePolicy,
2429        attempt: AttemptId,
2430        request: &JitRunnerRequest,
2431    ) -> Result<JitRegistration, LifecycleError> {
2432        let mut issued = 0_u32;
2433        loop {
2434            issued = issued.saturating_add(1);
2435            match self
2436                .ports
2437                .github
2438                .register(&policy.target, request, &self.cancel)
2439                .await
2440            {
2441                Ok(registration) => return Ok(registration),
2442                Err(error) if error.terminal => {
2443                    return Err(LifecycleError::Failed(error.reason));
2444                }
2445                Err(error) => {
2446                    if issued >= self.retry.max_attempts.max(1)
2447                        || !self.ports.demand.persists(policy.id).await
2448                    {
2449                        return Err(LifecycleError::Failed(error.reason));
2450                    }
2451                    let delay = error
2452                        .retry_after
2453                        .unwrap_or_else(|| self.retry.delay(issued));
2454                    self.ports.events.emit(AttemptEvent::Retry {
2455                        attempt,
2456                        operation: "jit_request",
2457                        delay,
2458                    });
2459                    self.ports.delay.wait(delay).await;
2460                    if !self.ports.demand.persists(policy.id).await {
2461                        return Err(LifecycleError::Failed(error.reason));
2462                    }
2463                }
2464            }
2465        }
2466    }
2467
2468    /// Where one attempt's files go, decided while the host allocation lock is
2469    /// held and before anything external happens.
2470    ///
2471    /// The branch is on the *repository's configured* workspace policy, so an
2472    /// organization policy and an ephemeral repository never reach slot
2473    /// selection at all: a persistent policy is unrepresentable for an
2474    /// organization target (D7, refused by `WorkspacePolicy::permitted_for` in
2475    /// both the constructor and the loader), and an ephemeral repository takes
2476    /// the disposable arm that existed before slots did.
2477    fn allocate_workspace(
2478        &self,
2479        policy: &ScalePolicy,
2480        id: AttemptId,
2481    ) -> Result<Placement, LifecycleError> {
2482        let placement = match policy.workspace_policy() {
2483            // Precedence (`02-target-architecture.md`): the repository's
2484            // persistent root is selected *before* the host root, which is why
2485            // this arm is first and why it never makes resolving the host
2486            // default a precondition of its own success.
2487            WorkspacePolicy::Persistent { root } => self.allocate_persistent_slot(policy, root),
2488            WorkspacePolicy::Ephemeral => self.allocate_disposable(policy, id),
2489        };
2490        // Here rather than inside the two arms, so that every path a root can
2491        // accept clears the record and none can be forgotten.
2492        if placement.is_ok() {
2493            self.root_accepted(policy.id);
2494        }
2495        placement
2496    }
2497
2498    /// `Host.runner_root_override`, read from the journal.
2499    ///
2500    /// Separated from [`Self::effective_host_root`] so that the two failures it
2501    /// folds together stay apart: an unreadable or missing host row is a journal
2502    /// problem and is always fatal, while an unresolvable *platform default* is
2503    /// only fatal to a placement that actually needs the host root.
2504    fn configured_host_root(&self) -> Result<Option<LocalAbsolutePath>, LifecycleError> {
2505        let host = self
2506            .ports
2507            .store
2508            .host(self.host_id)
2509            .map_err(|_| LifecycleError::Journal)?
2510            .ok_or_else(|| LifecycleError::Failed(FailureReason::Other("host not found".into())))?;
2511        Ok(host.runner_root_override.clone())
2512    }
2513
2514    /// `Host.runner_root_override`, or the platform default standing in for it.
2515    ///
2516    /// Takes the policy because an unresolvable default is a refusal like any
2517    /// other, and the record it leaves is that policy's.
2518    fn effective_host_root(
2519        &self,
2520        policy: &ScalePolicy,
2521    ) -> Result<LocalAbsolutePath, LifecycleError> {
2522        match self.configured_host_root()? {
2523            Some(configured) => Ok(configured),
2524            None => default_runner_root(&self.app_paths).map_err(|error| {
2525                // No root resolved, so there is no path to name but the one the
2526                // platform would have produced.
2527                self.root_refused(policy.id, "the platform default runner root", error)
2528            }),
2529        }
2530    }
2531
2532    /// Turns a runner-root refusal into a failure, and leaves the sentence
2533    /// somewhere an operator can read it.
2534    ///
2535    /// Per policy, because the policies on a host do not share a fate: one with
2536    /// its own persistent root places runners while another on a withheld
2537    /// volume places none, and a host-wide record would have the first clear
2538    /// the second's on the same pass.
2539    ///
2540    /// The recording is best-effort and its failure is deliberately swallowed:
2541    /// this runs on the path that is already failing, and a host that cannot
2542    /// write a diagnostic file must still report the refusal it came to report.
2543    /// `service status` says so itself when the file is unreadable.
2544    fn root_refused(&self, policy: PolicyId, root: &str, error: RunnerRootError) -> LifecycleError {
2545        let _ = runner_manager_platform::service::record_runner_root_refusal(
2546            &self.app_paths,
2547            &policy.to_string(),
2548            self.ports.clock.now(),
2549            error.kind(),
2550            root,
2551            &error.to_string(),
2552        );
2553        root_failure(error)
2554    }
2555
2556    /// Clears that policy's record, because its root accepted a placement.
2557    ///
2558    /// Called on every successful placement rather than only after a failure:
2559    /// the daemon that recovers is often not the process that failed -- a
2560    /// self-update restarts it -- so "clear it if we wrote it" would leave a
2561    /// stale note on `service status` for as long as the host ran.
2562    fn root_accepted(&self, policy: PolicyId) {
2563        let _ = runner_manager_platform::service::clear_runner_root_refusal(
2564            &self.app_paths,
2565            &policy.to_string(),
2566        );
2567    }
2568
2569    /// D3's disposable placement: a unique child of the effective host root,
2570    /// removed whole on cleanup. `c1`'s behaviour, moved behind the branch.
2571    fn allocate_disposable(
2572        &self,
2573        policy: &ScalePolicy,
2574        id: AttemptId,
2575    ) -> Result<Placement, LifecycleError> {
2576        let effective_root = self.effective_host_root(policy)?;
2577        RootPreflight::new(&self.app_paths)
2578            .check(&RootOwner::Host, &effective_root)
2579            .map_err(|error| self.root_refused(policy.id, effective_root.as_str(), error))?;
2580        let runtime = effective_root.as_path().join({
2581            #[cfg(test)]
2582            {
2583                if std::env::var("RUNNER_MANAGER_TEST_MUTANT").as_deref()
2584                    == Ok("reuse_job_workspace")
2585                {
2586                    "mutant-shared-workspace".to_owned()
2587                } else {
2588                    workspace_name(id)
2589                }
2590            }
2591            #[cfg(not(test))]
2592            {
2593                workspace_name(id)
2594            }
2595        });
2596        fs::create_dir_all(&runtime)
2597            .map_err(|_| LifecycleError::Failed(FailureReason::ProcessStartFailed))?;
2598        Ok(Placement {
2599            runtime,
2600            workspace: AttemptWorkspace::Ephemeral,
2601        })
2602    }
2603
2604    /// D4/D5's persistent placement: the lowest free `sN` under the repository's
2605    /// configured root.
2606    ///
2607    /// Steps 1 to 6 of `02-target-architecture.md`, "Slot allocation", in order;
2608    /// step 7 is the journal write [`Self::record_allocation`] owns. All of them
2609    /// run under the host allocation lock, because [`Self::launch_attempt`] is
2610    /// reachable only through a `LaunchRequest` and that carries the guard.
2611    ///
2612    /// **The filesystem is never consulted to decide which slots are taken**
2613    /// (invariant 6). The leases come from the journal; the directory is
2614    /// inspected only to decide whether *this* slot is safe to reuse.
2615    fn allocate_persistent_slot(
2616        &self,
2617        policy: &ScalePolicy,
2618        root: &LocalAbsolutePath,
2619    ) -> Result<Placement, LifecycleError> {
2620        // 1-4. The lowest positive slot no uncleaned attempt holds, refused
2621        // above the policy ceiling.
2622        let ceiling = policy.max_capacity().ok_or_else(|| {
2623            LifecycleError::Failed(FailureReason::Other(
2624                "a persistent workspace needs the policy's max_capacity to bound its slots"
2625                    .to_string(),
2626            ))
2627        })?;
2628        let leases = self
2629            .ports
2630            .store
2631            .slot_leases_for_policy(policy.id)
2632            .map_err(|_| LifecycleError::Journal)?;
2633        let slot = lowest_free_slot(&leases, ceiling).ok_or_else(|| {
2634            LifecycleError::Failed(FailureReason::Other(format!(
2635                "every persistent slot s1 to s{ceiling} for {} is leased by an attempt that has \
2636                 not been cleaned, so no slot is free; raise the repository's max capacity, or \
2637                 finish cleaning a concluded attempt",
2638                policy.target
2639            )))
2640        })?;
2641        let workspace = AttemptWorkspace::persistent_slot(slot);
2642        let name = workspace
2643            .slot_directory_name()
2644            .expect("a persistent allocation names its slot directory");
2645
2646        // The operational preflight, for the reasons the host root gets one: a
2647        // root that is remote, unwritable, or overlapping application data has
2648        // to fail before a directory is created rather than after. The host
2649        // root is registered only as something *not* to overlap; a host default
2650        // that cannot be resolved is a host-root problem and does not block a
2651        // repository that configured a root of its own.
2652        //
2653        // Only *that* failure is tolerated. An unreadable host row is a journal
2654        // failure and propagates, because silently continuing would drop the
2655        // overlap check entirely and accept a repository root that sits inside
2656        // the host root — the pair `RootPreflight` exists to refuse.
2657        let host_root = self
2658            .configured_host_root()?
2659            .or_else(|| default_runner_root(&self.app_paths).ok());
2660        let mut preflight = RootPreflight::new(&self.app_paths);
2661        if let Some(host_root) = host_root {
2662            preflight = preflight.against(RootOwner::Host, host_root);
2663        }
2664        let checked = preflight
2665            .check(&RootOwner::Repository(policy.target.to_string()), root)
2666            .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2667        if let Some(leaf) = checked.leaf_to_create() {
2668            fs::create_dir(leaf).map_err(|source| {
2669                LifecycleError::Failed(FailureReason::Other(format!(
2670                    "the persistent workspace root {} could not be created: {source}",
2671                    leaf.display()
2672                )))
2673            })?;
2674        }
2675
2676        // 5-6. `<root>/sN`, contained lexically by construction, then created or
2677        // validated, then contained canonically now that it resolves, and only
2678        // then accepted for reuse.
2679        let slot_path = runner_root::derive_child(root, &name)
2680            .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2681        create_or_validate_slot(slot_path.as_path())?;
2682        runner_root::verify_containment(root, &slot_path)
2683            .map_err(|error| self.root_refused(policy.id, root.as_str(), error))?;
2684        accept_reusable_slot(slot_path.as_path())?;
2685        Ok(Placement {
2686            runtime: slot_path.as_path().to_path_buf(),
2687            workspace,
2688        })
2689    }
2690
2691    /// The first journal write of an attempt, where a duplicate slot lease is
2692    /// still possible and has to be reported as itself.
2693    ///
2694    /// [`Self::record`] flattens every store failure into
2695    /// [`LifecycleError::Journal`], which is right for a state transition and
2696    /// wrong here: the partial unique index
2697    /// `one_uncleaned_persistent_attempt_per_slot` is the final race fence
2698    /// (`04-security-recovery.md`, "two attempts use one slot concurrently"),
2699    /// and an operator who reaches it needs to read that rather than "attempt
2700    /// journal operation failed". Nothing was written, so the caller returns
2701    /// without concluding an attempt that is not in the journal.
2702    fn record_allocation(&self, attempt: &RunnerAttempt) -> Result<(), LifecycleError> {
2703        match self.ports.store.record_attempt(attempt) {
2704            Ok(()) => {
2705                self.ports.events.emit(AttemptEvent::State {
2706                    attempt: attempt.id,
2707                    state: attempt.state(),
2708                });
2709                Ok(())
2710            }
2711            Err(error @ StoreError::SlotAlreadyLeased { .. }) => Err(LifecycleError::Failed(
2712                FailureReason::Other(error.to_string()),
2713            )),
2714            Err(_) => Err(LifecycleError::Journal),
2715        }
2716    }
2717
2718    async fn launch_attempt(
2719        &self,
2720        policy: &ScalePolicy,
2721        allocation_guard: &AllocationGuard,
2722    ) -> Result<RunnerAttempt, LifecycleError> {
2723        if !*self
2724            .recovery_complete
2725            .lock()
2726            .map_err(|_| LifecycleError::Journal)?
2727        {
2728            return Err(LifecycleError::RecoveryIncomplete);
2729        }
2730        let labels = policy
2731            .routing_labels()
2732            .ok_or(LifecycleError::Failed(FailureReason::JitRequestFailed))?;
2733        let id = AttemptId::new_random();
2734        let placement = self.allocate_workspace(policy, id)?;
2735        let mut attempt = RunnerAttempt::allocate_in(
2736            id,
2737            policy.id,
2738            placement.runtime,
2739            placement.workspace,
2740            self.ports.clock.now(),
2741        );
2742        // This is deliberately the first effect after directory allocation, and
2743        // for a persistent attempt it is also what makes the slot lease durable
2744        // before any package or GitHub effect
2745        // (`02-target-architecture.md`, "Slot allocation", step 7).
2746        self.record_allocation(&attempt)?;
2747
2748        let version = match self.materialize_with_retry(policy, &attempt).await {
2749            Ok(version) => version,
2750            Err(reason) => return self.fail_launch(&mut attempt, reason),
2751        };
2752        self.prune_under_allocation_lock(allocation_guard, &version)?;
2753        self.versions
2754            .lock()
2755            .map_err(|_| LifecycleError::Journal)?
2756            .insert(id, version);
2757
2758        let jit_request =
2759            JitRunnerRequest::for_policy(runner_name(id), self.runner_group_id, labels);
2760        let registration = match self.register_with_retry(policy, id, &jit_request).await {
2761            Ok(registration) => registration,
2762            Err(error) => return self.fail_launch(&mut attempt, error.reason()),
2763        };
2764        let runner_id = registration.runner().id;
2765        write_runner_id(attempt.runtime_path(), runner_id)?;
2766        attempt
2767            .jit_received(self.ports.clock.now())
2768            .map_err(|_| LifecycleError::Transition)?;
2769        self.record(&attempt)?;
2770        let config = registration.into_config();
2771        let mut issued = 0_u32;
2772        let pid = loop {
2773            issued = issued.saturating_add(1);
2774            match self.ports.processes.spawn(&attempt, &config) {
2775                Ok(pid) => break pid,
2776                Err(error) => {
2777                    if let Some(pid) = error.live_pid {
2778                        attempt
2779                            .started(pid, self.ports.clock.now())
2780                            .map_err(|_| LifecycleError::Transition)?;
2781                        self.record(&attempt)?;
2782                        return Err(LifecycleError::Failed(error.reason));
2783                    }
2784                    if !error.retryable
2785                        || issued >= self.retry.max_attempts.max(1)
2786                        || !self.ports.demand.persists(policy.id).await
2787                    {
2788                        return self.fail_launch(&mut attempt, error.reason);
2789                    }
2790                    let delay = self.retry.delay(issued);
2791                    self.ports.events.emit(AttemptEvent::Retry {
2792                        attempt: attempt.id,
2793                        operation: "process_start",
2794                        delay,
2795                    });
2796                    self.ports.delay.wait(delay).await;
2797                    if !self.ports.demand.persists(policy.id).await {
2798                        return self.fail_launch(&mut attempt, error.reason);
2799                    }
2800                }
2801            }
2802        };
2803        attempt
2804            .started(pid, self.ports.clock.now())
2805            .map_err(|_| LifecycleError::Transition)?;
2806        self.record(&attempt)?;
2807        Ok(attempt)
2808    }
2809
2810    fn fail_launch<T>(
2811        &self,
2812        attempt: &mut RunnerAttempt,
2813        reason: FailureReason,
2814    ) -> Result<T, LifecycleError> {
2815        self.conclude(attempt, AttemptOutcome::failed(reason.clone()))?;
2816        Err(LifecycleError::Failed(reason))
2817    }
2818
2819    /// e2's prune guard is invoked only with e1's allocation guard borrowed.
2820    /// The otherwise-unused argument is a compile-time witness of the ordering.
2821    fn prune_under_allocation_lock(
2822        &self,
2823        guard: &AllocationGuard,
2824        version: &RunnerVersion,
2825    ) -> Result<(), LifecycleError> {
2826        let attempts = self
2827            .ports
2828            .store
2829            .attempts()
2830            .map_err(|_| LifecycleError::Journal)?;
2831        self.ports
2832            .packages
2833            .prune_obsolete_guarded(
2834                PruneAuthority::from_launch_request(guard),
2835                version,
2836                &attempts,
2837            )
2838            .map_err(LifecycleError::Failed)
2839    }
2840}
2841
2842#[async_trait]
2843impl RunnerLauncher for LifecycleLauncher {
2844    async fn supervise(
2845        &self,
2846        policy: &ScalePolicy,
2847    ) -> Result<Vec<ReplacementIntent>, LaunchFailure> {
2848        LifecycleLauncher::supervise(self, policy)
2849            .await
2850            .map_err(|error| LaunchFailure::new(error.reason()))
2851    }
2852
2853    async fn attempts(&self) -> Result<Vec<RunnerAttempt>, LaunchFailure> {
2854        self.ports.store.attempts().map_err(|_| {
2855            LaunchFailure::new(FailureReason::Other(
2856                "attempt journal could not be read".into(),
2857            ))
2858        })
2859    }
2860
2861    async fn launch(&self, request: LaunchRequest<'_>) -> Result<RunnerAttempt, LaunchFailure> {
2862        self.launch_attempt(request.policy, request.allocation_guard)
2863            .await
2864            .map_err(|error| LaunchFailure::new(error.reason()))
2865    }
2866
2867    async fn clean(&self, id: AttemptId) -> Result<(), LaunchFailure> {
2868        let mut attempt = self
2869            .ports
2870            .store
2871            .attempt(id)
2872            .map_err(|_| {
2873                LaunchFailure::new(FailureReason::Other(
2874                    "attempt journal could not be read".into(),
2875                ))
2876            })?
2877            .ok_or_else(|| {
2878                LaunchFailure::new(FailureReason::Other(
2879                    "attempt disappeared from the journal".into(),
2880                ))
2881            })?;
2882        self.clean_attempt(&mut attempt)
2883            .map_err(|error| LaunchFailure::new(error.reason()))
2884    }
2885}
2886
2887fn runner_name(attempt: AttemptId) -> String {
2888    format!("runner-manager-{attempt}")
2889}
2890
2891fn read_runner_id(runtime: &Path) -> Option<u64> {
2892    fs::read_to_string(runtime.join(RUNNER_ID_FILE))
2893        .ok()?
2894        .trim()
2895        .parse()
2896        .ok()
2897}
2898
2899fn write_runner_id(runtime: &Path, runner_id: u64) -> Result<(), LifecycleError> {
2900    let target = runtime.join(RUNNER_ID_FILE);
2901    if let Some(existing) = read_runner_id(runtime) {
2902        return (existing == runner_id)
2903            .then_some(())
2904            .ok_or(LifecycleError::Journal);
2905    }
2906    let temporary = runtime.join(format!("{RUNNER_ID_FILE}.{}.tmp", uuid::Uuid::new_v4()));
2907    write_durable_file(&temporary, runner_id.to_string().as_bytes())
2908        .map_err(|_| LifecycleError::Journal)?;
2909    match fs::rename(&temporary, &target) {
2910        Ok(()) => sync_directory(runtime).map_err(|_| LifecycleError::Journal),
2911        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
2912            let _ = fs::remove_file(&temporary);
2913            (read_runner_id(runtime) == Some(runner_id))
2914                .then_some(())
2915                .ok_or(LifecycleError::Journal)
2916        }
2917        Err(_) => {
2918            let _ = fs::remove_file(&temporary);
2919            Err(LifecycleError::Journal)
2920        }
2921    }
2922}
2923
2924fn write_durable_file(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
2925    let mut file = fs::OpenOptions::new()
2926        .create(true)
2927        .truncate(true)
2928        .write(true)
2929        .open(path)?;
2930    file.write_all(bytes)?;
2931    file.sync_all()?;
2932    let parent = path.parent().ok_or_else(|| {
2933        std::io::Error::new(
2934            std::io::ErrorKind::InvalidInput,
2935            "file has no parent directory",
2936        )
2937    })?;
2938    sync_directory(parent)
2939}
2940
2941#[cfg(unix)]
2942fn sync_directory(path: &Path) -> std::io::Result<()> {
2943    fs::File::open(path)?.sync_all()
2944}
2945
2946#[cfg(windows)]
2947fn sync_directory(path: &Path) -> std::io::Result<()> {
2948    use std::os::windows::fs::OpenOptionsExt;
2949
2950    const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
2951    const FILE_SHARE_ALL: u32 = 0x0000_0007;
2952    const GENERIC_WRITE: u32 = 0x4000_0000;
2953    fs::OpenOptions::new()
2954        .access_mode(GENERIC_WRITE)
2955        .share_mode(FILE_SHARE_ALL)
2956        .custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
2957        .open(path)?
2958        .sync_all()
2959}
2960
2961#[cfg(test)]
2962mod tests {
2963    use super::*;
2964    use std::collections::BTreeSet;
2965    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2966
2967    use crate::reconcile::{AllocationLock, InProcessAllocationLock};
2968    use runner_manager_domain::model::{Elapsed, TargetScope};
2969    use runner_manager_domain::store::SqliteStore;
2970    use runner_manager_github::jit::JitRunner;
2971    use runner_manager_testkit::clock::FakeClock;
2972    use runner_manager_testkit::fixtures;
2973
2974    /// One event's fields, in the order they were recorded.
2975    type CapturedFields = Vec<(String, String)>;
2976
2977    /// Keeps the `(name, value)` fields of every `tracing` event emitted while
2978    /// it is installed.
2979    ///
2980    /// The names are kept and not just the rendered line, so a test can hold the
2981    /// event to `crate::logging`'s two rules -- the field allow-list and the
2982    /// value scrub -- instead of asserting that some string was passed to a
2983    /// macro.
2984    #[derive(Clone, Default)]
2985    struct CapturedEvents(std::sync::Arc<std::sync::Mutex<Vec<CapturedFields>>>);
2986
2987    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedEvents {
2988        fn on_event(
2989            &self,
2990            event: &tracing::Event<'_>,
2991            _context: tracing_subscriber::layer::Context<'_, S>,
2992        ) {
2993            struct Collect(Vec<(String, String)>);
2994            impl tracing::field::Visit for Collect {
2995                fn record_debug(
2996                    &mut self,
2997                    field: &tracing::field::Field,
2998                    value: &dyn std::fmt::Debug,
2999                ) {
3000                    // `{value:?}` on a `&str` field would keep the quotes, and
3001                    // the redaction rules are about the value, not its literal.
3002                    self.0.push((
3003                        field.name().to_owned(),
3004                        format!("{value:?}").trim_matches('"').to_owned(),
3005                    ));
3006                }
3007            }
3008            let mut collected = Collect(Vec::new());
3009            event.record(&mut collected);
3010            self.0
3011                .lock()
3012                .expect("the capture mutex is not poisoned")
3013                .push(collected.0);
3014        }
3015    }
3016
3017    /// The regression behind a three-hour outage that showed nothing an operator
3018    /// could act on.
3019    ///
3020    /// A root the daemon cannot use refuses the launch *before*
3021    /// `record_allocation`, so no attempt row is ever written: `b2` has nothing
3022    /// to carry the failure on and `g2` has nothing to show it from. The
3023    /// lifecycle event keeps only the variant -- `failure_reason_kind` allows no
3024    /// free text on an event -- so the whole failure read
3025    /// `runner_start_failed reason=other`, once per poll, for hours.
3026    ///
3027    /// The assertion is deliberately made against the **production redaction
3028    /// rules** and not merely against what was emitted. An earlier attempt at
3029    /// this fix logged the rendered error on a `detail` field and passed a test
3030    /// exactly like this one, while shipping `detail="[redacted]"`: the field
3031    /// was not allow-listed, and had it been, the value is mostly paths and
3032    /// would have been scrubbed to `[path]`. A test that does not ask
3033    /// `crate::logging` what survives is a test that proves nothing.
3034    #[test]
3035    fn a_launch_the_runner_root_refused_names_the_cause_in_the_log_that_ships() {
3036        use runner_manager_platform::logging;
3037        use tracing_subscriber::layer::SubscriberExt as _;
3038
3039        let captured = CapturedEvents::default();
3040        let error = RunnerRootError::DeniedByPrivacyPolicy {
3041            requested: PathBuf::from("/Volumes/NVME/runners"),
3042            refused: PathBuf::from("/Volumes/NVME"),
3043            remediation: RootOwner::Host.remediation(),
3044        };
3045        let kind = error.kind();
3046
3047        let failure = tracing::subscriber::with_default(
3048            tracing_subscriber::registry().with(captured.clone()),
3049            || root_failure(error),
3050        );
3051
3052        // The reason still carries the whole sentence. Nothing on *this* path
3053        // renders it -- there is no attempt row, and the event carries only the
3054        // variant -- so this is a guard against a refactor that drops the detail
3055        // before some future surface can show it, and not a claim that one does.
3056        assert!(
3057            matches!(
3058                &failure,
3059                LifecycleError::Failed(FailureReason::Other(detail))
3060                    if detail.contains("/Volumes/NVME/runners")
3061                        && detail.contains("Full Disk Access")
3062            ),
3063            "the reason must still carry the detail: {failure:?}"
3064        );
3065
3066        let events = captured
3067            .0
3068            .lock()
3069            .expect("the capture mutex is not poisoned")
3070            .clone();
3071        let event = events
3072            .iter()
3073            .find(|fields| fields.iter().any(|(_, value)| value.contains(kind)))
3074            .unwrap_or_else(|| panic!("the refusal did not name its cause: {events:?}"));
3075
3076        // Held to the rules the real sink applies, by name and by value. An
3077        // earlier fix put the rendered error on an unlisted `detail` field and
3078        // shipped `[redacted]`; asserting only that something was emitted would
3079        // have passed then too.
3080        for (name, value) in event {
3081            assert!(
3082                logging::is_field_allowed(name),
3083                "`{name}` is not allow-listed, so it ships as `{}`: {event:?}",
3084                logging::REDACTION
3085            );
3086            assert_eq!(
3087                &logging::redact(value),
3088                value,
3089                "`{name}` does not survive value-shape scrubbing: {event:?}"
3090            );
3091        }
3092    }
3093
3094    /// A slot number, for the tests that name one.
3095    fn nz(slot: u16) -> NonZeroU16 {
3096        NonZeroU16::new(slot).expect("a positive slot")
3097    }
3098
3099    const JIT: &str = "eyJzZWNyZXQiOiJnaHBfRE9fTk9UX0xFQUsifQ==";
3100
3101    #[derive(Debug, Default)]
3102    struct FakeGithubLifecycle {
3103        registration_failures: Mutex<VecDeque<bool>>,
3104        observations: Mutex<VecDeque<LifecycleGithubObservation>>,
3105        registrations: AtomicUsize,
3106        remaining_runners: AtomicUsize,
3107        /// Every runner id `deregister` was asked to remove, in order. A count
3108        /// would not do: the assertions worth making are that the *right*
3109        /// registration was deleted and that it was deleted once.
3110        deregistrations: Mutex<Vec<u64>>,
3111        /// Set to make `deregister` answer `false`, standing for a GitHub that
3112        /// could not be reached at the moment the attempt concluded.
3113        deregistration_fails: AtomicBool,
3114        /// The journal to read *during* a registration, for the ordering
3115        /// assertion `02-target-architecture.md` makes: the slot lease is
3116        /// written "before package or GitHub effects". Reading it afterwards
3117        /// would pass even if the write happened second.
3118        journal: Mutex<Option<Arc<SqliteStore>>>,
3119        /// One entry per registration, in order.
3120        registration_facts: Mutex<Vec<RegistrationFact>>,
3121    }
3122
3123    /// What one JIT registration saw of the world at the moment it was issued.
3124    ///
3125    /// `runner_name` is what ties the other two fields to *one* attempt: with
3126    /// two allocators racing, "some slot was journalled" is a much weaker claim
3127    /// than "the slot this very request belongs to was journalled", and only the
3128    /// name distinguishes them.
3129    #[derive(Debug, Clone)]
3130    struct RegistrationFact {
3131        /// The persistent slots the journal already held.
3132        leased_slots: Vec<u16>,
3133        /// The `work_folder` the request carried.
3134        work_folder: String,
3135        /// The runner name the request carried, i.e. [`runner_name`] of the
3136        /// registering attempt.
3137        runner_name: String,
3138    }
3139
3140    impl FakeGithubLifecycle {
3141        fn fail(mut self, terminal: bool) -> Self {
3142            self.registration_failures
3143                .get_mut()
3144                .expect("unpoisoned")
3145                .push_back(terminal);
3146            self
3147        }
3148
3149        fn watch_journal(&self, store: Arc<SqliteStore>) {
3150            *self.journal.lock().unwrap() = Some(store);
3151        }
3152
3153        fn registration_facts(&self) -> Vec<RegistrationFact> {
3154            self.registration_facts.lock().unwrap().clone()
3155        }
3156
3157        fn observe(&self, observation: GithubRunnerObservation) {
3158            let observation = match observation {
3159                GithubRunnerObservation::Unreachable => LifecycleGithubObservation::unreachable(),
3160                GithubRunnerObservation::NotRegistered => {
3161                    LifecycleGithubObservation::not_registered()
3162                }
3163                GithubRunnerObservation::Registered { busy } => {
3164                    LifecycleGithubObservation::registered(73, busy)
3165                }
3166            };
3167            self.observations.lock().unwrap().push_back(observation);
3168        }
3169    }
3170
3171    #[async_trait]
3172    impl LifecycleGithub for FakeGithubLifecycle {
3173        async fn register(
3174            &self,
3175            _target: &ScaleTarget,
3176            request: &JitRunnerRequest,
3177            _cancel: &CancelToken,
3178        ) -> Result<JitRegistration, JitRequestFailure> {
3179            self.registrations.fetch_add(1, Ordering::SeqCst);
3180            if let Some(store) = self.journal.lock().unwrap().as_ref() {
3181                let slots = store
3182                    .attempts()
3183                    .expect("the journal is readable")
3184                    .iter()
3185                    .filter_map(|attempt| attempt.workspace().slot_number())
3186                    .collect();
3187                self.registration_facts
3188                    .lock()
3189                    .unwrap()
3190                    .push(RegistrationFact {
3191                        leased_slots: slots,
3192                        work_folder: request.work_folder().to_string(),
3193                        runner_name: request.name().to_string(),
3194                    });
3195            }
3196            if let Some(terminal) = self.registration_failures.lock().unwrap().pop_front() {
3197                return Err(JitRequestFailure {
3198                    terminal,
3199                    reason: if terminal {
3200                        FailureReason::Other("GitHub refused JIT registration with 403".into())
3201                    } else {
3202                        FailureReason::JitRequestFailed
3203                    },
3204                    retry_after: None,
3205                });
3206            }
3207            self.remaining_runners.store(1, Ordering::SeqCst);
3208            Ok(JitRegistration::new(
3209                EncodedJitConfig::new(JIT),
3210                JitRunner {
3211                    id: 73,
3212                    name: request.name().to_string(),
3213                    os: "windows".into(),
3214                    status: "offline".into(),
3215                    busy: false,
3216                    runner_group_id: Some(1),
3217                    labels: request.labels().to_vec(),
3218                },
3219            ))
3220        }
3221
3222        async fn observe(
3223            &self,
3224            _target: &ScaleTarget,
3225            _attempt: AttemptId,
3226            _cancel: &CancelToken,
3227        ) -> LifecycleGithubObservation {
3228            let observation = self
3229                .observations
3230                .lock()
3231                .unwrap()
3232                .pop_front()
3233                .unwrap_or(LifecycleGithubObservation::not_registered());
3234            if observation.status == GithubRunnerObservation::NotRegistered {
3235                self.remaining_runners.store(0, Ordering::SeqCst);
3236            }
3237            observation
3238        }
3239
3240        async fn deregister(
3241            &self,
3242            _target: &ScaleTarget,
3243            runner_id: u64,
3244            _cancel: &CancelToken,
3245        ) -> bool {
3246            self.deregistrations.lock().unwrap().push(runner_id);
3247            if self.deregistration_fails.load(Ordering::SeqCst) {
3248                return false;
3249            }
3250            self.remaining_runners.store(0, Ordering::SeqCst);
3251            true
3252        }
3253    }
3254
3255    #[derive(Debug)]
3256    struct FakePackages {
3257        version: RunnerVersion,
3258        leases: Mutex<BTreeSet<AttemptId>>,
3259        materializations: AtomicUsize,
3260        materialization_failures: AtomicUsize,
3261        releases: AtomicUsize,
3262        prunes: AtomicUsize,
3263        prune_currents: Mutex<Vec<RunnerVersion>>,
3264    }
3265
3266    impl Default for FakePackages {
3267        fn default() -> Self {
3268            Self {
3269                version: RunnerVersion::parse("2.330.0").unwrap(),
3270                leases: Mutex::new(BTreeSet::new()),
3271                materializations: AtomicUsize::new(0),
3272                materialization_failures: AtomicUsize::new(0),
3273                releases: AtomicUsize::new(0),
3274                prunes: AtomicUsize::new(0),
3275                prune_currents: Mutex::new(Vec::new()),
3276            }
3277        }
3278    }
3279
3280    impl FakePackages {
3281        fn fail_materializations(&self, count: usize) {
3282            self.materialization_failures.store(count, Ordering::SeqCst);
3283        }
3284    }
3285
3286    #[async_trait]
3287    impl RuntimePackages for FakePackages {
3288        async fn materialize(
3289            &self,
3290            attempt: &RunnerAttempt,
3291        ) -> Result<RunnerVersion, FailureReason> {
3292            self.materializations.fetch_add(1, Ordering::SeqCst);
3293            if self
3294                .materialization_failures
3295                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
3296                    if left > 0 { Some(left - 1) } else { None }
3297                })
3298                .is_ok()
3299            {
3300                return Err(FailureReason::Other(
3301                    "runner package materialization failed transiently".into(),
3302                ));
3303            }
3304            fs::create_dir_all(attempt.runtime_path()).unwrap();
3305            fs::write(attempt.runtime_path().join("runner-package"), b"verified").unwrap();
3306            self.leases.lock().unwrap().insert(attempt.id);
3307            Ok(self.version.clone())
3308        }
3309
3310        fn release(&self, attempt: AttemptId) -> Result<(), FailureReason> {
3311            self.leases.lock().unwrap().remove(&attempt);
3312            self.releases.fetch_add(1, Ordering::SeqCst);
3313            Ok(())
3314        }
3315
3316        fn prune_obsolete_guarded(
3317            &self,
3318            _authority: PruneAuthority<'_>,
3319            current: &RunnerVersion,
3320            _attempts: &[RunnerAttempt],
3321        ) -> Result<(), FailureReason> {
3322            self.prunes.fetch_add(1, Ordering::SeqCst);
3323            self.prune_currents.lock().unwrap().push(current.clone());
3324            Ok(())
3325        }
3326    }
3327
3328    #[derive(Debug, Default)]
3329    struct FakeProcesses {
3330        alive: AtomicBool,
3331        completed_successfully: AtomicBool,
3332        spawns: AtomicUsize,
3333        spawn_failures: AtomicUsize,
3334        live_spawn_failure: AtomicBool,
3335        terminations: AtomicUsize,
3336        intent: AtomicBool,
3337        intent_failure: AtomicBool,
3338        actions: Mutex<Vec<&'static str>>,
3339        saw_secret: AtomicBool,
3340    }
3341
3342    impl FakeProcesses {
3343        fn fail_spawns(&self, count: usize) {
3344            self.spawn_failures.store(count, Ordering::SeqCst);
3345        }
3346
3347        fn fail_spawn_with_live_child(&self) {
3348            self.live_spawn_failure.store(true, Ordering::SeqCst);
3349        }
3350
3351        fn set_alive(&self, alive: bool) {
3352            self.alive.store(alive, Ordering::SeqCst);
3353        }
3354
3355        fn finish_successfully(&self) {
3356            self.completed_successfully.store(true, Ordering::SeqCst);
3357            self.alive.store(false, Ordering::SeqCst);
3358        }
3359
3360        fn fail_intent(&self) {
3361            self.intent_failure.store(true, Ordering::SeqCst);
3362        }
3363    }
3364
3365    impl ProcessSupervisor for FakeProcesses {
3366        fn spawn(
3367            &self,
3368            attempt: &RunnerAttempt,
3369            config: &EncodedJitConfig,
3370        ) -> Result<u32, ProcessStartFailure> {
3371            self.spawns.fetch_add(1, Ordering::SeqCst);
3372            // Model the production handoff on both paths: the sensitive file is
3373            // scoped to this call and absent when it returns.
3374            let handoff = RestrictiveHandoff::create(
3375                attempt.runtime_path(),
3376                SecretString::from(config.expose().to_owned()),
3377            )
3378            .unwrap();
3379            self.saw_secret
3380                .store(config.expose() == JIT, Ordering::SeqCst);
3381            let handoff_path = handoff.path().to_path_buf();
3382            let failing = self
3383                .spawn_failures
3384                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
3385                    if left > 0 { Some(left - 1) } else { None }
3386                })
3387                .is_ok();
3388            drop(handoff);
3389            assert!(!handoff_path.exists(), "handoff must be absent on return");
3390            if self.live_spawn_failure.swap(false, Ordering::SeqCst) {
3391                self.alive.store(true, Ordering::SeqCst);
3392                return Err(ProcessStartFailure::after_spawn_live(4242));
3393            }
3394            if failing {
3395                return Err(ProcessStartFailure::before_spawn(
3396                    FailureReason::ProcessStartFailed,
3397                ));
3398            }
3399            self.alive.store(true, Ordering::SeqCst);
3400            Ok(4242)
3401        }
3402
3403        fn is_alive(&self, _attempt: &RunnerAttempt) -> Result<bool, FailureReason> {
3404            self.actions.lock().unwrap().push("observe_process");
3405            Ok(self.alive.load(Ordering::SeqCst))
3406        }
3407
3408        fn recovered_pid(&self, _attempt: &RunnerAttempt) -> Result<Option<u32>, FailureReason> {
3409            Ok(self.alive.load(Ordering::SeqCst).then_some(4242))
3410        }
3411
3412        fn completed_successfully(&self, _attempt: &RunnerAttempt) -> bool {
3413            self.completed_successfully.load(Ordering::SeqCst)
3414        }
3415
3416        fn record_terminate_intent(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
3417            self.actions.lock().unwrap().push("terminate_intent");
3418            if self.intent_failure.load(Ordering::SeqCst) {
3419                return Err(FailureReason::Other(
3420                    "terminate intent directory sync failed".into(),
3421                ));
3422            }
3423            self.intent.store(true, Ordering::SeqCst);
3424            Ok(())
3425        }
3426
3427        fn has_terminate_intent(&self, _attempt: &RunnerAttempt) -> bool {
3428            self.intent.load(Ordering::SeqCst)
3429        }
3430
3431        fn terminate(&self, _attempt: &RunnerAttempt) -> Result<(), FailureReason> {
3432            assert!(
3433                self.intent.load(Ordering::SeqCst),
3434                "the durable intent must exist before signalling"
3435            );
3436            self.actions.lock().unwrap().push("terminate");
3437            self.terminations.fetch_add(1, Ordering::SeqCst);
3438            self.alive.store(false, Ordering::SeqCst);
3439            Ok(())
3440        }
3441    }
3442
3443    #[derive(Debug, Default)]
3444    struct FakeDemand {
3445        answers: Mutex<VecDeque<bool>>,
3446    }
3447
3448    impl FakeDemand {
3449        fn answering(answers: impl IntoIterator<Item = bool>) -> Self {
3450            Self {
3451                answers: Mutex::new(answers.into_iter().collect()),
3452            }
3453        }
3454    }
3455
3456    #[async_trait]
3457    impl DemandPersistence for FakeDemand {
3458        async fn persists(&self, _policy: PolicyId) -> bool {
3459            self.answers.lock().unwrap().pop_front().unwrap_or(true)
3460        }
3461    }
3462
3463    #[derive(Debug, Default)]
3464    struct FakeDelay(Mutex<Vec<Duration>>);
3465
3466    #[async_trait]
3467    impl RetryDelay for FakeDelay {
3468        async fn wait(&self, duration: Duration) {
3469            self.0.lock().unwrap().push(duration);
3470        }
3471    }
3472
3473    struct Harness {
3474        _root: tempfile::TempDir,
3475        app_paths: runner_manager_platform::paths::AppPaths,
3476        launcher: LifecycleLauncher,
3477        demand: Arc<dyn DemandPersistence>,
3478        store: Arc<SqliteStore>,
3479        github: Arc<FakeGithubLifecycle>,
3480        packages: Arc<FakePackages>,
3481        processes: Arc<FakeProcesses>,
3482        clock: Arc<FakeClock>,
3483        events: Arc<AttemptEventLog>,
3484        reconcile_events: Arc<crate::reconcile::EventLog>,
3485        delay: Arc<FakeDelay>,
3486        host: runner_manager_domain::model::Host,
3487        policy: ScalePolicy,
3488        allocation_lock: InProcessAllocationLock,
3489        /// The repository persistent root, once one is configured.
3490        workspace_root: Option<LocalAbsolutePath>,
3491    }
3492
3493    impl Harness {
3494        fn new(github: FakeGithubLifecycle, demand: Arc<dyn DemandPersistence>) -> Self {
3495            let root = tempfile::tempdir().unwrap();
3496            let paths = runner_manager_platform::paths::AppPaths::rooted_at(root.path());
3497            paths.create_all().unwrap();
3498            let policy = fixtures::policy()
3499                .repository("octo/repo")
3500                .autoscale("home", 2)
3501                .active()
3502                .build();
3503            // A unit harness must never inspect or write the workstation's
3504            // production runner root. On Windows that is C:\rman, which is
3505            // intentionally inaccessible to an ordinary account after a boot
3506            // service is installed. Using it made the same suite pass on a
3507            // clean CI image and fail on a correctly secured developer host.
3508            let host_root = root.path().join("host-root");
3509            fs::create_dir_all(&host_root).unwrap();
3510            let mut host = fixtures::host().build();
3511            host.runner_root_override = Some(
3512                LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
3513                    .expect("a local absolute host root"),
3514            );
3515            let store = Arc::new(SqliteStore::open_in_memory().unwrap());
3516            store.put_host(&host).unwrap();
3517            let github = Arc::new(github);
3518            let packages = Arc::new(FakePackages::default());
3519            let processes = Arc::new(FakeProcesses::default());
3520            let clock = Arc::new(FakeClock::default());
3521            let events = Arc::new(AttemptEventLog::default());
3522            let reconcile_events = Arc::new(crate::reconcile::EventLog::new());
3523            let delay = Arc::new(FakeDelay::default());
3524            let ports = LifecyclePorts {
3525                store: Arc::clone(&store) as Arc<dyn Store>,
3526                github: Arc::clone(&github) as Arc<dyn LifecycleGithub>,
3527                packages: Arc::clone(&packages) as Arc<dyn RuntimePackages>,
3528                processes: Arc::clone(&processes) as Arc<dyn ProcessSupervisor>,
3529                clock: Arc::clone(&clock) as Arc<dyn Clock>,
3530                demand: Arc::clone(&demand),
3531                delay: Arc::clone(&delay) as Arc<dyn RetryDelay>,
3532                events: Arc::clone(&events) as Arc<dyn AttemptEventSink>,
3533                reconcile_events: Arc::clone(&reconcile_events) as Arc<dyn EventSink>,
3534            };
3535            let launcher = Self::launcher_over(policy.host_id, &paths, ports);
3536            Self {
3537                _root: root,
3538                app_paths: paths,
3539                launcher,
3540                demand,
3541                store,
3542                github,
3543                packages,
3544                processes,
3545                clock,
3546                events,
3547                reconcile_events,
3548                delay,
3549                host,
3550                policy,
3551                allocation_lock: InProcessAllocationLock::new(),
3552                workspace_root: None,
3553            }
3554        }
3555
3556        /// Put disposable attempts under a host root of this harness's own.
3557        ///
3558        /// Without it the launcher resolves the *platform* default, which on
3559        /// Windows is `%SystemDrive%\rman` — a real directory on the machine
3560        /// running the suite. Every test added by `c2` places its files inside
3561        /// its own temporary directory instead.
3562        fn with_host_runner_root(mut self) -> Self {
3563            let host_root = self.host_root();
3564            fs::create_dir_all(&host_root).unwrap();
3565            self.host.runner_root_override = Some(
3566                LocalAbsolutePath::new(host_root.to_str().expect("a UTF-8 temporary path"))
3567                    .expect("a local absolute host root"),
3568            );
3569            self.store.put_host(&self.host).unwrap();
3570            self
3571        }
3572
3573        /// Opt this harness's repository into a persistent workspace (D4).
3574        fn with_persistent_workspace(mut self, capacity: u16) -> Self {
3575            self = self.with_host_runner_root();
3576            let root = self._root.path().join("persist");
3577            let root = LocalAbsolutePath::new(root.to_str().expect("a UTF-8 temporary path"))
3578                .expect("a local absolute workspace root");
3579            self.policy = fixtures::policy()
3580                .repository("octo/repo")
3581                .autoscale("home", capacity)
3582                .active()
3583                .build();
3584            self.policy
3585                .set_workspace_policy(
3586                    WorkspacePolicy::persistent(root.clone(), TargetScope::Repository)
3587                        .expect("a repository may be persistent"),
3588                )
3589                .expect("a repository may be persistent");
3590            self.workspace_root = Some(root);
3591            // The journal's copy, so that cleanup's cross-check against a
3592            // surviving policy is exercised rather than skipped.
3593            self.store.insert_policy(&self.policy).unwrap();
3594            self
3595        }
3596
3597        fn workspace_root(&self) -> &LocalAbsolutePath {
3598            self.workspace_root
3599                .as_ref()
3600                .expect("this harness configured a persistent workspace")
3601        }
3602
3603        fn slot_path(&self, slot: u16) -> PathBuf {
3604            self.workspace_root().as_path().join(format!("s{slot}"))
3605        }
3606
3607        fn host_root(&self) -> PathBuf {
3608            self._root.path().join("host-root")
3609        }
3610
3611        fn attempt(&self, id: AttemptId) -> RunnerAttempt {
3612            self.store
3613                .attempt(id)
3614                .unwrap()
3615                .expect("the attempt is journalled")
3616        }
3617
3618        /// Conclude an attempt and run the real cleanup over it.
3619        fn conclude(&self, id: AttemptId) -> RunnerAttempt {
3620            let mut attempt = self.attempt(id);
3621            attempt
3622                .conclude(
3623                    AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
3624                    self.clock.now(),
3625                )
3626                .unwrap();
3627            self.store.record_attempt(&attempt).unwrap();
3628            attempt
3629        }
3630
3631        async fn cleanup_retaining_work(&self, id: AttemptId) {
3632            self.conclude(id);
3633            self.launcher
3634                .clean(id)
3635                .await
3636                .expect("the slot is scrubbed and the lease released");
3637        }
3638
3639        /// The launcher configuration every launcher in this harness shares, so
3640        /// that the one a restart mints cannot drift from the original.
3641        fn launcher_over(
3642            host: HostId,
3643            paths: &runner_manager_platform::paths::AppPaths,
3644            ports: LifecyclePorts,
3645        ) -> LifecycleLauncher {
3646            LifecycleLauncher::new(
3647                host,
3648                paths.clone(),
3649                paths.logs_dir(),
3650                1,
3651                RecoveryTimeouts::new(
3652                    Elapsed::seconds(10),
3653                    Elapsed::seconds(10),
3654                    Elapsed::seconds(10),
3655                ),
3656                RetryPolicy::bounded(3, Duration::from_millis(10), Duration::from_millis(25)),
3657                ports,
3658            )
3659        }
3660
3661        /// The same journal, the same directories, a launcher that remembers
3662        /// nothing — which is what a daemon restart is.
3663        fn restart(&self) -> LifecycleLauncher {
3664            Self::launcher_over(
3665                self.policy.host_id,
3666                &self.app_paths,
3667                LifecyclePorts {
3668                    store: Arc::clone(&self.store) as Arc<dyn Store>,
3669                    github: Arc::clone(&self.github) as Arc<dyn LifecycleGithub>,
3670                    packages: Arc::clone(&self.packages) as Arc<dyn RuntimePackages>,
3671                    processes: Arc::clone(&self.processes) as Arc<dyn ProcessSupervisor>,
3672                    clock: Arc::clone(&self.clock) as Arc<dyn Clock>,
3673                    demand: Arc::clone(&self.demand),
3674                    delay: Arc::clone(&self.delay) as Arc<dyn RetryDelay>,
3675                    events: Arc::clone(&self.events) as Arc<dyn AttemptEventSink>,
3676                    reconcile_events: Arc::clone(&self.reconcile_events) as Arc<dyn EventSink>,
3677                },
3678            )
3679        }
3680
3681        async fn ready(&self) {
3682            self.launcher
3683                .recover_startup(std::slice::from_ref(&self.policy))
3684                .await
3685                .unwrap();
3686        }
3687
3688        async fn launch(&self) -> RunnerAttempt {
3689            self.launch_result().await.unwrap()
3690        }
3691
3692        async fn launch_result(&self) -> Result<RunnerAttempt, LaunchFailure> {
3693            let guard = self.allocation_lock.acquire().await.unwrap();
3694            self.launcher
3695                .launch(LaunchRequest {
3696                    host: &self.host,
3697                    policy: &self.policy,
3698                    allocation_guard: &guard,
3699                })
3700                .await
3701        }
3702
3703        fn only_attempt(&self) -> RunnerAttempt {
3704            self.store.attempts().unwrap().into_iter().next().unwrap()
3705        }
3706    }
3707
3708    /// The wiring the three-hour outage needed and did not have.
3709    ///
3710    /// A root that refuses a launch does so before `record_allocation`, so no
3711    /// attempt row carries it, and the daemon's log scrubs the paths out of the
3712    /// sentence. The record this asserts is the only surface left, and
3713    /// `service status` reads it -- so if this wiring is ever dropped, the
3714    /// failure goes back to reading `runner_start_failed reason=other` once per
3715    /// poll and nothing else.
3716    #[tokio::test]
3717    async fn a_root_that_refuses_a_launch_is_recorded_and_cleared_when_one_succeeds() {
3718        use runner_manager_platform::service::{clear_runner_root_refusal, runner_root_refusals};
3719
3720        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
3721            .with_host_runner_root();
3722        harness.ready().await;
3723
3724        // A root under two directories that do not exist: refused by the
3725        // preflight, for a reason that has nothing to do with this platform, so
3726        // the assertion holds on all three.
3727        let unusable = harness
3728            ._root
3729            .path()
3730            .join("absent")
3731            .join("deeper")
3732            .join("runners");
3733        let mut host = harness.host.clone();
3734        host.runner_root_override = Some(
3735            LocalAbsolutePath::new(unusable.to_str().expect("a UTF-8 temporary path"))
3736                .expect("a local absolute host root"),
3737        );
3738        harness.store.put_host(&host).unwrap();
3739
3740        let failure = harness
3741            .launch_result()
3742            .await
3743            .expect_err("a root whose parents are missing cannot hold a runner");
3744        assert!(
3745            matches!(failure.reason, FailureReason::Other(_)),
3746            "{failure:?}"
3747        );
3748
3749        let refusals = runner_root_refusals(&harness.app_paths).expect("readable");
3750        let refusal = refusals
3751            .first()
3752            .expect("the refusal reached the one surface that can hold it");
3753        assert_eq!(refusal.policy, harness.policy.id.to_string());
3754        assert_eq!(refusal.kind, "missing_parents");
3755        assert!(
3756            refusal.root.contains("runners") && refusal.detail.contains("runners"),
3757            "the directory must be named in full: {refusal:?}"
3758        );
3759
3760        // And a root that works clears it, so a host that has been fixed stops
3761        // reporting a fault it no longer has.
3762        harness.store.put_host(&harness.host).unwrap();
3763        harness.launch().await;
3764        assert!(
3765            runner_root_refusals(&harness.app_paths)
3766                .expect("readable")
3767                .is_empty(),
3768            "a successful placement clears that policy's record"
3769        );
3770
3771        clear_runner_root_refusal(&harness.app_paths, &harness.policy.id.to_string())
3772            .expect("cleanup");
3773    }
3774
3775    #[tokio::test]
3776    async fn a_job_walks_every_state_and_cleans_every_artifact() {
3777        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3778        harness.ready().await;
3779        let started = harness.launch().await;
3780        assert_eq!(started.state(), AttemptState::Starting);
3781        assert_eq!(read_runner_id(started.runtime_path()), Some(73));
3782
3783        harness
3784            .github
3785            .observe(GithubRunnerObservation::Registered { busy: false });
3786        harness.launcher.supervise(&harness.policy).await.unwrap();
3787        assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
3788
3789        harness
3790            .github
3791            .observe(GithubRunnerObservation::Registered { busy: true });
3792        harness.launcher.supervise(&harness.policy).await.unwrap();
3793        assert_eq!(harness.only_attempt().state(), AttemptState::Busy);
3794
3795        harness.processes.finish_successfully();
3796        harness
3797            .github
3798            .observe(GithubRunnerObservation::NotRegistered);
3799        harness.launcher.supervise(&harness.policy).await.unwrap();
3800        let cleaned = harness.only_attempt();
3801        assert_eq!(cleaned.state(), AttemptState::Cleaned);
3802        assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
3803        assert!(!started.runtime_path().exists());
3804        assert_eq!(harness.packages.releases.load(Ordering::SeqCst), 1);
3805        assert_eq!(harness.github.remaining_runners.load(Ordering::SeqCst), 0);
3806
3807        let states: Vec<_> = harness
3808            .events
3809            .events()
3810            .into_iter()
3811            .filter_map(|event| match event {
3812                AttemptEvent::State { state, .. } => Some(state),
3813                _ => None,
3814            })
3815            .collect();
3816        assert_eq!(
3817            states,
3818            vec![
3819                AttemptState::Allocated,
3820                AttemptState::JitReceived,
3821                AttemptState::Starting,
3822                AttemptState::Idle,
3823                AttemptState::Busy,
3824                AttemptState::Finished,
3825                AttemptState::Cleaned,
3826            ]
3827        );
3828    }
3829
3830    #[tokio::test]
3831    async fn an_idle_exit_is_not_a_failure_in_the_journal_or_events() {
3832        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3833        harness.ready().await;
3834        let started = harness.launch().await;
3835        harness
3836            .github
3837            .observe(GithubRunnerObservation::Registered { busy: false });
3838        harness.launcher.supervise(&harness.policy).await.unwrap();
3839        harness.clock.advance_secs(11);
3840        harness.processes.set_alive(false);
3841        harness
3842            .github
3843            .observe(GithubRunnerObservation::NotRegistered);
3844        harness.launcher.supervise(&harness.policy).await.unwrap();
3845
3846        let cleaned = harness.only_attempt();
3847        assert!(cleaned.outcome().unwrap().is_idle_exit());
3848        assert!(!cleaned.outcome().unwrap().is_failure());
3849        assert!(!started.runtime_path().exists());
3850        assert!(
3851            harness
3852                .reconcile_events
3853                .events()
3854                .iter()
3855                .any(|event| matches!(
3856                    event,
3857                    LifecycleEvent::AttemptCleaned {
3858                        outcome: OutcomeKind::IdleExit,
3859                        ..
3860                    }
3861                ))
3862        );
3863        assert!(!harness.events.events().iter().any(|event| matches!(
3864            event,
3865            AttemptEvent::Concluded {
3866                outcome: OutcomeKind::Failed,
3867                ..
3868            }
3869        )));
3870    }
3871
3872    #[tokio::test]
3873    async fn handoff_is_absent_after_success_and_every_failed_spawn_retry() {
3874        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3875        harness.processes.fail_spawns(2);
3876        harness.ready().await;
3877        let attempt = harness.launch().await;
3878        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 3);
3879        assert!(harness.processes.saw_secret.load(Ordering::SeqCst));
3880        let names: Vec<_> = fs::read_dir(attempt.runtime_path())
3881            .unwrap()
3882            .map(|entry| entry.unwrap().file_name())
3883            .collect();
3884        assert!(
3885            names.iter().all(|name| {
3886                !name
3887                    .to_string_lossy()
3888                    .starts_with(RestrictiveHandoff::NAME_PREFIX)
3889            }),
3890            "JIT artifact survived: {names:?}"
3891        );
3892        assert_eq!(
3893            *harness.delay.0.lock().unwrap(),
3894            vec![Duration::from_millis(10), Duration::from_millis(20)]
3895        );
3896    }
3897
3898    #[tokio::test]
3899    async fn jit_retry_stops_with_demand_and_a_terminal_403_never_retries() {
3900        let gone = Harness::new(
3901            FakeGithubLifecycle::default().fail(false),
3902            Arc::new(FakeDemand::answering([false])),
3903        );
3904        gone.ready().await;
3905        assert!(gone.launch_result().await.is_err());
3906        assert_eq!(gone.github.registrations.load(Ordering::SeqCst), 1);
3907        assert!(gone.delay.0.lock().unwrap().is_empty());
3908
3909        let forbidden = Harness::new(
3910            FakeGithubLifecycle::default().fail(true),
3911            Arc::new(PersistentDemand),
3912        );
3913        forbidden.ready().await;
3914        assert!(forbidden.launch_result().await.is_err());
3915        assert_eq!(forbidden.github.registrations.load(Ordering::SeqCst), 1);
3916        assert!(forbidden.delay.0.lock().unwrap().is_empty());
3917        assert!(matches!(
3918            forbidden.only_attempt().outcome(),
3919            Some(AttemptOutcome::Failed {
3920                reason: FailureReason::Other(action)
3921            }) if action.contains("403")
3922        ));
3923
3924        let transient = Harness::new(
3925            FakeGithubLifecycle::default().fail(false).fail(false),
3926            Arc::new(PersistentDemand),
3927        );
3928        transient.ready().await;
3929        transient.launch().await;
3930        assert_eq!(transient.github.registrations.load(Ordering::SeqCst), 3);
3931        assert_eq!(
3932            *transient.delay.0.lock().unwrap(),
3933            vec![Duration::from_millis(10), Duration::from_millis(20)]
3934        );
3935    }
3936
3937    /// The layout has to leave room for what the runner writes underneath it.
3938    ///
3939    /// Windows refuses a path over `MAX_PATH`, and this product's own CI hit
3940    /// that: 264 characters against a limit of 260, failing three checkout
3941    /// retries with `Filename too long`. The two identifiers in the old layout
3942    /// cost 74 characters between them for no benefit -- an attempt id is
3943    /// unique on its own.
3944    #[test]
3945    fn a_workspace_leaves_room_for_the_deepest_path_a_checkout_writes() {
3946        const MAX_PATH: usize = 260;
3947        // The real root on the machine this was found on.
3948        let root = r"C:\Users\IvanD\AppData\Local\IvanMurzak\runner-manager\data\runtime";
3949        // What `actions/checkout` writes at its deepest: the work directory,
3950        // the repository named twice, and a pack keep-file with a 40-character
3951        // object name.
3952        let repo = "GitHub-Runner-Scaler-UI";
3953        let deepest = format!(
3954            r"_work\{repo}\{repo}\.git\objects\pack\pack-{}.keep",
3955            "0".repeat(40)
3956        );
3957
3958        let name = workspace_name(AttemptId::new_random());
3959        assert_eq!(name.len(), WORKSPACE_NAME_LEN, "{name}");
3960        assert!(
3961            name.chars().all(|c| c.is_ascii_hexdigit()),
3962            "a directory name must not carry the identifier's dashes: {name}"
3963        );
3964
3965        let full = format!(r"{root}\{name}\{deepest}");
3966        assert!(
3967            full.len() < MAX_PATH,
3968            "the deepest path a checkout writes must fit: {} characters, limit {MAX_PATH}",
3969            full.len()
3970        );
3971
3972        // The discriminator: the layout this replaced does not fit, so a test
3973        // that passed for both would be proving nothing.
3974        let old = format!(
3975            r"{root}\{}\{}\{deepest}",
3976            PolicyId::new_random(),
3977            AttemptId::new_random()
3978        );
3979        assert!(
3980            old.len() > MAX_PATH,
3981            "the old layout is supposed to be the thing that did not fit: {} characters",
3982            old.len()
3983        );
3984    }
3985
3986    #[tokio::test]
3987    async fn two_attempts_never_share_a_workspace_even_after_failure() {
3988        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
3989        harness.ready().await;
3990        let first = harness.launch().await;
3991        fs::write(first.runtime_path().join("hostile-leftover"), b"first job").unwrap();
3992        harness
3993            .github
3994            .observe(GithubRunnerObservation::Registered { busy: false });
3995        harness.launcher.supervise(&harness.policy).await.unwrap();
3996        harness.clock.advance_secs(11);
3997        harness.processes.set_alive(false);
3998        harness
3999            .github
4000            .observe(GithubRunnerObservation::NotRegistered);
4001        harness.launcher.supervise(&harness.policy).await.unwrap();
4002        assert!(!first.runtime_path().exists());
4003
4004        let second = harness.launch().await;
4005        assert_ne!(first.runtime_path(), second.runtime_path());
4006        assert!(!second.runtime_path().join("hostile-leftover").exists());
4007
4008        fs::write(
4009            second.runtime_path().join("hostile-on-failure"),
4010            b"second job",
4011        )
4012        .unwrap();
4013        harness.processes.set_alive(false);
4014        harness
4015            .github
4016            .observe(GithubRunnerObservation::NotRegistered);
4017        harness.launcher.supervise(&harness.policy).await.unwrap();
4018        assert!(
4019            !second.runtime_path().exists(),
4020            "failed workspace was retained"
4021        );
4022    }
4023
4024    #[tokio::test]
4025    async fn a_runner_that_never_gets_a_job_is_stopped_deregistered_and_not_replaced() {
4026        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4027        harness.ready().await;
4028        let attempt = harness.launch().await;
4029
4030        // Registered and waiting, which is where it stays: the fake keeps
4031        // answering the same observation, exactly as GitHub does for a runner
4032        // nobody assigns work to.
4033        harness
4034            .github
4035            .observe(GithubRunnerObservation::Registered { busy: false });
4036        harness.launcher.supervise(&harness.policy).await.unwrap();
4037        assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
4038
4039        // One second inside the ten-second idle timeout nothing happens, which
4040        // is what keeps this from being a test that would pass on any clock.
4041        harness.clock.advance_secs(9);
4042        harness
4043            .github
4044            .observe(GithubRunnerObservation::Registered { busy: false });
4045        let none_yet = harness.launcher.supervise(&harness.policy).await.unwrap();
4046        assert_eq!(harness.only_attempt().state(), AttemptState::Idle);
4047        assert!(none_yet.is_empty());
4048        assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
4049
4050        // Past it, the agent ends the runner itself.
4051        harness.clock.advance_secs(1);
4052        harness
4053            .github
4054            .observe(GithubRunnerObservation::Registered { busy: false });
4055        let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4056
4057        let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
4058        assert_eq!(
4059            concluded.outcome(),
4060            Some(&AttemptOutcome::ExitedIdleWithoutWork),
4061            "a surplus runner did not fail; recording one as a failure sends an operator \
4062             hunting a fault that does not exist"
4063        );
4064        assert_eq!(concluded.state(), AttemptState::Cleaned);
4065        assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
4066        assert!(!attempt.runtime_path().exists());
4067
4068        // The registration goes with it. Without this the runner stays listed
4069        // in the target's runner settings after the process it named is gone.
4070        assert_eq!(
4071            *harness.github.deregistrations.lock().unwrap(),
4072            vec![73],
4073            "the attempt's own runner id, deleted exactly once"
4074        );
4075
4076        // And nothing is started in its place: the work it was launched for went
4077        // elsewhere, so a replacement would rebuild it every idle timeout.
4078        assert!(
4079            replacements.is_empty(),
4080            "a surplus exit must not request a replacement"
4081        );
4082    }
4083
4084    #[tokio::test]
4085    async fn a_registration_github_will_not_delete_still_concludes_the_attempt() {
4086        // The delete is best-effort by construction: the process is gone and the
4087        // slot has to come back. Holding the conclusion until GitHub cooperates
4088        // would leak a capacity slot on every unreachable moment.
4089        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4090        harness.ready().await;
4091        let attempt = harness.launch().await;
4092        harness
4093            .github
4094            .observe(GithubRunnerObservation::Registered { busy: false });
4095        harness.launcher.supervise(&harness.policy).await.unwrap();
4096
4097        harness
4098            .github
4099            .deregistration_fails
4100            .store(true, Ordering::SeqCst);
4101        harness.clock.advance_secs(11);
4102        harness
4103            .github
4104            .observe(GithubRunnerObservation::Registered { busy: false });
4105        harness.launcher.supervise(&harness.policy).await.unwrap();
4106
4107        assert_eq!(
4108            *harness.github.deregistrations.lock().unwrap(),
4109            vec![73],
4110            "the delete was attempted"
4111        );
4112        let concluded = harness.store.attempt(attempt.id).unwrap().unwrap();
4113        assert_eq!(
4114            concluded.outcome(),
4115            Some(&AttemptOutcome::ExitedIdleWithoutWork),
4116            "the attempt concluded anyway"
4117        );
4118        assert_eq!(concluded.state(), AttemptState::Cleaned);
4119        assert!(!attempt.runtime_path().exists());
4120    }
4121
4122    #[tokio::test]
4123    async fn exit_before_acceptance_returns_replacement_intent_without_launching() {
4124        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4125        harness.ready().await;
4126        let first = harness.launch().await;
4127        harness.processes.set_alive(false);
4128        harness
4129            .github
4130            .observe(GithubRunnerObservation::NotRegistered);
4131        let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4132        let failed = harness.store.attempt(first.id).unwrap().unwrap();
4133        assert!(matches!(
4134            failed.outcome(),
4135            Some(AttemptOutcome::Failed {
4136                reason: FailureReason::ProcessExitedUnexpectedly
4137            })
4138        ));
4139        assert!(!first.runtime_path().exists());
4140
4141        assert_eq!(
4142            replacements,
4143            vec![ReplacementIntent {
4144                policy: harness.policy.id,
4145                previous_attempt: first.id,
4146                operation: "exit_before_acceptance_replacement",
4147            }]
4148        );
4149        assert_eq!(harness.store.attempts().unwrap().len(), 1);
4150        assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
4151        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4152        assert!(harness.delay.0.lock().unwrap().is_empty());
4153    }
4154
4155    #[tokio::test]
4156    async fn expired_jit_is_removed_and_does_not_reregister_after_demand_disappears() {
4157        let harness = Harness::new(
4158            FakeGithubLifecycle::default(),
4159            Arc::new(FakeDemand::answering([false])),
4160        );
4161        let id = AttemptId::new_random();
4162        let runtime = harness
4163            .launcher
4164            .app_paths
4165            .runtime_dir()
4166            .join(harness.policy.id.to_string())
4167            .join(id.to_string());
4168        fs::create_dir_all(&runtime).unwrap();
4169        let mut attempt =
4170            RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4171        attempt.jit_received(harness.clock.now()).unwrap();
4172        harness.store.record_attempt(&attempt).unwrap();
4173        harness.clock.advance_secs(11);
4174        let replacements = harness
4175            .launcher
4176            .recover_startup(std::slice::from_ref(&harness.policy))
4177            .await
4178            .unwrap();
4179        assert_eq!(
4180            replacements,
4181            vec![ReplacementIntent {
4182                policy: harness.policy.id,
4183                previous_attempt: id,
4184                operation: "jit_expired_replacement",
4185            }]
4186        );
4187
4188        let cleaned = harness.store.attempt(id).unwrap().unwrap();
4189        assert_eq!(cleaned.state(), AttemptState::Cleaned);
4190        assert!(matches!(
4191            cleaned.outcome(),
4192            Some(AttemptOutcome::Failed {
4193                reason: FailureReason::JitExpired
4194            })
4195        ));
4196        assert!(!runtime.exists());
4197        assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
4198        assert!(harness.delay.0.lock().unwrap().is_empty());
4199    }
4200
4201    #[tokio::test]
4202    async fn expired_jit_returns_intent_but_never_launches_inside_lifecycle() {
4203        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4204        let id = AttemptId::new_random();
4205        let runtime = harness
4206            .launcher
4207            .app_paths
4208            .runtime_dir()
4209            .join("expired-with-demand");
4210        fs::create_dir_all(&runtime).unwrap();
4211        let mut attempt =
4212            RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4213        attempt.jit_received(harness.clock.now()).unwrap();
4214        harness.store.record_attempt(&attempt).unwrap();
4215        harness.clock.advance_secs(11);
4216        let replacements = harness
4217            .launcher
4218            .recover_startup(std::slice::from_ref(&harness.policy))
4219            .await
4220            .unwrap();
4221
4222        let attempts = harness.store.attempts().unwrap();
4223        assert_eq!(attempts.len(), 1);
4224        assert_eq!(
4225            attempts
4226                .iter()
4227                .find(|attempt| attempt.id == id)
4228                .unwrap()
4229                .state(),
4230            AttemptState::Cleaned
4231        );
4232        assert_eq!(
4233            replacements,
4234            vec![ReplacementIntent {
4235                policy: harness.policy.id,
4236                previous_attempt: id,
4237                operation: "jit_expired_replacement",
4238            }]
4239        );
4240        assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 0);
4241        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4242        assert!(harness.delay.0.lock().unwrap().is_empty());
4243    }
4244
4245    #[tokio::test]
4246    async fn package_materialization_retries_are_bounded_and_demand_adjacent() {
4247        let persistent = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4248        persistent.packages.fail_materializations(2);
4249        persistent.ready().await;
4250        persistent.launch().await;
4251        assert_eq!(
4252            persistent.packages.materializations.load(Ordering::SeqCst),
4253            3
4254        );
4255        assert_eq!(
4256            *persistent.delay.0.lock().unwrap(),
4257            vec![Duration::from_millis(10), Duration::from_millis(20)]
4258        );
4259
4260        let gone_before_wait = Harness::new(
4261            FakeGithubLifecycle::default(),
4262            Arc::new(FakeDemand::answering([false])),
4263        );
4264        gone_before_wait.packages.fail_materializations(3);
4265        gone_before_wait.ready().await;
4266        assert!(gone_before_wait.launch_result().await.is_err());
4267        assert_eq!(
4268            gone_before_wait
4269                .packages
4270                .materializations
4271                .load(Ordering::SeqCst),
4272            1
4273        );
4274        assert!(gone_before_wait.delay.0.lock().unwrap().is_empty());
4275
4276        let gone_during_wait = Harness::new(
4277            FakeGithubLifecycle::default(),
4278            Arc::new(FakeDemand::answering([true, false])),
4279        );
4280        gone_during_wait.packages.fail_materializations(3);
4281        gone_during_wait.ready().await;
4282        assert!(gone_during_wait.launch_result().await.is_err());
4283        assert_eq!(
4284            gone_during_wait
4285                .packages
4286                .materializations
4287                .load(Ordering::SeqCst),
4288            1
4289        );
4290        assert_eq!(
4291            *gone_during_wait.delay.0.lock().unwrap(),
4292            vec![Duration::from_millis(10)]
4293        );
4294    }
4295
4296    #[tokio::test]
4297    async fn replacement_is_intent_only_and_never_launches_inside_lifecycle() {
4298        let harness = Harness::new(
4299            FakeGithubLifecycle::default(),
4300            Arc::new(FakeDemand::answering([true, false])),
4301        );
4302        harness.ready().await;
4303        let first = harness.launch().await;
4304        harness.processes.set_alive(false);
4305        harness
4306            .github
4307            .observe(GithubRunnerObservation::NotRegistered);
4308        let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4309
4310        assert_eq!(harness.store.attempts().unwrap().len(), 1);
4311        assert_eq!(harness.github.registrations.load(Ordering::SeqCst), 1);
4312        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4313        assert!(harness.delay.0.lock().unwrap().is_empty());
4314        assert_eq!(
4315            replacements,
4316            vec![ReplacementIntent {
4317                policy: harness.policy.id,
4318                previous_attempt: first.id,
4319                operation: "exit_before_acceptance_replacement",
4320            }]
4321        );
4322        assert_eq!(
4323            harness.store.attempt(first.id).unwrap().unwrap().state(),
4324            AttemptState::Cleaned
4325        );
4326    }
4327
4328    #[tokio::test]
4329    async fn startup_adopts_a_live_process_and_refuses_launch_before_recovery() {
4330        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4331        let before = harness.launch_result().await;
4332        assert!(before.is_err());
4333        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4334
4335        let id = AttemptId::new_random();
4336        let runtime = harness.launcher.app_paths.runtime_dir().join("adopt");
4337        fs::create_dir_all(&runtime).unwrap();
4338        let mut attempt =
4339            RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4340        attempt.jit_received(harness.clock.now()).unwrap();
4341        attempt.started(4242, harness.clock.now()).unwrap();
4342        harness.store.record_attempt(&attempt).unwrap();
4343        harness.processes.set_alive(true);
4344        harness
4345            .github
4346            .observe(GithubRunnerObservation::NotRegistered);
4347        let replacements = harness
4348            .launcher
4349            .recover_startup(std::slice::from_ref(&harness.policy))
4350            .await
4351            .unwrap();
4352        assert!(replacements.is_empty());
4353        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 0);
4354        assert!(
4355            harness
4356                .events
4357                .events()
4358                .contains(&AttemptEvent::Adopted { attempt: id })
4359        );
4360    }
4361
4362    #[tokio::test]
4363    async fn spawn_before_starting_crash_recovers_pid_then_completes_and_cleans() {
4364        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4365        let id = AttemptId::new_random();
4366        let runtime = harness
4367            .launcher
4368            .app_paths
4369            .runtime_dir()
4370            .join("spawn-before-starting");
4371        fs::create_dir_all(&runtime).unwrap();
4372        let mut attempt =
4373            RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4374        attempt.jit_received(harness.clock.now()).unwrap();
4375        harness.store.record_attempt(&attempt).unwrap();
4376        harness.processes.set_alive(true);
4377        harness
4378            .github
4379            .observe(GithubRunnerObservation::Registered { busy: true });
4380
4381        let replacements = harness
4382            .launcher
4383            .recover_startup(std::slice::from_ref(&harness.policy))
4384            .await
4385            .unwrap();
4386        assert!(replacements.is_empty());
4387        let recovered = harness.store.attempt(id).unwrap().unwrap();
4388        assert_eq!(recovered.state(), AttemptState::Busy);
4389        assert_eq!(recovered.process_id(), Some(4242));
4390        assert_eq!(recovered.github_runner_id(), Some(73));
4391        let events = harness.events.events();
4392        let starting = events
4393            .iter()
4394            .position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Starting } if *attempt == id))
4395            .unwrap();
4396        let busy = events
4397            .iter()
4398            .position(|event| matches!(event, AttemptEvent::State { attempt, state: AttemptState::Busy } if *attempt == id))
4399            .unwrap();
4400        assert!(starting < busy, "recovery skipped a legal edge: {events:?}");
4401
4402        harness.processes.finish_successfully();
4403        harness
4404            .github
4405            .observe(GithubRunnerObservation::NotRegistered);
4406        assert!(
4407            harness
4408                .launcher
4409                .supervise(&harness.policy)
4410                .await
4411                .unwrap()
4412                .is_empty()
4413        );
4414        let cleaned = harness.store.attempt(id).unwrap().unwrap();
4415        assert_eq!(cleaned.state(), AttemptState::Cleaned);
4416        assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::CompletedJob));
4417        assert!(!runtime.exists());
4418    }
4419
4420    #[tokio::test]
4421    async fn failed_post_spawn_stop_keeps_capacity_until_supervision_proves_death() {
4422        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4423        harness.processes.fail_spawn_with_live_child();
4424        harness.ready().await;
4425        assert!(harness.launch_result().await.is_err());
4426
4427        let attempt = harness.only_attempt();
4428        assert_eq!(attempt.state(), AttemptState::Starting);
4429        assert_eq!(attempt.process_id(), Some(4242));
4430        assert!(attempt.outcome().is_none());
4431        assert!(attempt.state().counts_against_capacity());
4432        assert_eq!(harness.processes.spawns.load(Ordering::SeqCst), 1);
4433        assert!(harness.delay.0.lock().unwrap().is_empty());
4434
4435        harness.processes.set_alive(false);
4436        harness
4437            .github
4438            .observe(GithubRunnerObservation::NotRegistered);
4439        let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4440        assert_eq!(replacements.len(), 1);
4441        assert_eq!(
4442            harness.store.attempt(attempt.id).unwrap().unwrap().state(),
4443            AttemptState::Cleaned
4444        );
4445    }
4446
4447    #[tokio::test]
4448    async fn remote_runner_identity_closes_both_sides_of_the_registration_crash_boundary() {
4449        for sidecar_already_present in [false, true] {
4450            let harness = Harness::new(
4451                FakeGithubLifecycle::default(),
4452                Arc::new(FakeDemand::answering([false])),
4453            );
4454            let id = AttemptId::new_random();
4455            let runtime =
4456                harness
4457                    .launcher
4458                    .app_paths
4459                    .runtime_dir()
4460                    .join(if sidecar_already_present {
4461                        "after-id-sidecar"
4462                    } else {
4463                        "before-id-sidecar"
4464                    });
4465            fs::create_dir_all(&runtime).unwrap();
4466            let mut attempt =
4467                RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4468            if sidecar_already_present {
4469                write_runner_id(&runtime, 73).unwrap();
4470                attempt.jit_received(harness.clock.now()).unwrap();
4471            }
4472            harness.store.record_attempt(&attempt).unwrap();
4473            harness.processes.set_alive(true);
4474            harness
4475                .github
4476                .observe(GithubRunnerObservation::Registered { busy: false });
4477            harness
4478                .launcher
4479                .recover_startup(std::slice::from_ref(&harness.policy))
4480                .await
4481                .unwrap();
4482
4483            assert_eq!(read_runner_id(&runtime), Some(73));
4484            assert!(
4485                harness
4486                    .store
4487                    .attempt(id)
4488                    .unwrap()
4489                    .unwrap()
4490                    .outcome()
4491                    .is_none()
4492            );
4493            let events = harness.events.events();
4494            let recovered = events.iter().position(|event| {
4495                matches!(
4496                    event,
4497                    AttemptEvent::RemoteIdentityRecovered {
4498                        attempt,
4499                        runner_id: 73
4500                    } if *attempt == id
4501                )
4502            });
4503            assert_eq!(recovered.is_some(), !sidecar_already_present);
4504            if let Some(recovered) = recovered {
4505                let adopted = events
4506                    .iter()
4507                    .position(|event| matches!(event, AttemptEvent::Adopted { attempt } if *attempt == id))
4508                    .unwrap();
4509                assert!(
4510                    recovered < adopted,
4511                    "identity was not durable before adoption: {events:?}"
4512                );
4513            }
4514            assert!(runtime.exists());
4515        }
4516    }
4517
4518    #[tokio::test]
4519    async fn recovery_stays_closed_for_unknown_policy_and_unreachable_attempts() {
4520        let unknown = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4521        let unknown_attempt = RunnerAttempt::allocate(
4522            AttemptId::new_random(),
4523            PolicyId::from_u128(0xfeed),
4524            unknown
4525                .launcher
4526                .app_paths
4527                .runtime_dir()
4528                .join("unknown-policy"),
4529            unknown.clock.now(),
4530        );
4531        unknown.store.record_attempt(&unknown_attempt).unwrap();
4532        let expired_id = AttemptId::new_random();
4533        let expired_runtime = unknown
4534            .launcher
4535            .app_paths
4536            .runtime_dir()
4537            .join("expired-beside-unknown");
4538        fs::create_dir_all(&expired_runtime).unwrap();
4539        let mut expired = RunnerAttempt::allocate(
4540            expired_id,
4541            unknown.policy.id,
4542            expired_runtime,
4543            unknown.clock.now(),
4544        );
4545        expired.jit_received(unknown.clock.now()).unwrap();
4546        unknown.store.record_attempt(&expired).unwrap();
4547        unknown.clock.advance_secs(11);
4548        assert!(matches!(
4549            unknown
4550                .launcher
4551                .recover_startup(std::slice::from_ref(&unknown.policy))
4552                .await,
4553            Err(LifecycleError::RecoveryIncomplete)
4554        ));
4555        assert!(unknown.launch_result().await.is_err());
4556        assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
4557        let recovered_policy = fixtures::policy()
4558            .id(PolicyId::from_u128(0xfeed))
4559            .repository("octo/repo")
4560            .autoscale("home", 2)
4561            .active()
4562            .build();
4563        let pending = unknown
4564            .launcher
4565            .recover_startup(&[unknown.policy.clone(), recovered_policy])
4566            .await
4567            .unwrap();
4568        assert_eq!(
4569            pending,
4570            vec![ReplacementIntent {
4571                policy: unknown.policy.id,
4572                previous_attempt: expired_id,
4573                operation: "jit_expired_replacement",
4574            }]
4575        );
4576        assert_eq!(unknown.processes.spawns.load(Ordering::SeqCst), 0);
4577
4578        let unreachable = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4579        let id = AttemptId::new_random();
4580        let runtime = unreachable
4581            .launcher
4582            .app_paths
4583            .runtime_dir()
4584            .join("unreachable");
4585        fs::create_dir_all(&runtime).unwrap();
4586        unreachable
4587            .store
4588            .record_attempt(&RunnerAttempt::allocate(
4589                id,
4590                unreachable.policy.id,
4591                runtime,
4592                unreachable.clock.now(),
4593            ))
4594            .unwrap();
4595        unreachable
4596            .github
4597            .observe(GithubRunnerObservation::Unreachable);
4598        assert!(matches!(
4599            unreachable
4600                .launcher
4601                .recover_startup(std::slice::from_ref(&unreachable.policy))
4602                .await,
4603            Err(LifecycleError::RecoveryIncomplete)
4604        ));
4605        assert!(unreachable.launch_result().await.is_err());
4606        assert_eq!(unreachable.processes.spawns.load(Ordering::SeqCst), 0);
4607    }
4608
4609    #[tokio::test]
4610    async fn a_dead_busy_process_unknown_to_github_is_orphaned_and_cleaned() {
4611        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4612        let id = AttemptId::new_random();
4613        let runtime = harness.launcher.app_paths.runtime_dir().join("orphan");
4614        fs::create_dir_all(&runtime).unwrap();
4615        let mut attempt =
4616            RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4617        attempt.jit_received(harness.clock.now()).unwrap();
4618        attempt.started(4242, harness.clock.now()).unwrap();
4619        attempt.assigned_job(73, harness.clock.now()).unwrap();
4620        harness.store.record_attempt(&attempt).unwrap();
4621        harness.processes.set_alive(false);
4622        harness
4623            .github
4624            .observe(GithubRunnerObservation::NotRegistered);
4625        harness
4626            .launcher
4627            .recover_startup(std::slice::from_ref(&harness.policy))
4628            .await
4629            .unwrap();
4630        let cleaned = harness.store.attempt(id).unwrap().unwrap();
4631        assert_eq!(cleaned.state(), AttemptState::Cleaned);
4632        assert_eq!(cleaned.outcome(), Some(&AttemptOutcome::Orphaned));
4633        assert!(!runtime.exists());
4634    }
4635
4636    #[tokio::test]
4637    async fn registration_timeout_journals_intent_stops_then_concludes_with_dead_reason() {
4638        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4639        harness.ready().await;
4640        let id = AttemptId::new_random();
4641        let runtime = harness.launcher.app_paths.runtime_dir().join("timeout");
4642        fs::create_dir_all(&runtime).unwrap();
4643        let mut attempt =
4644            RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4645        attempt.jit_received(harness.clock.now()).unwrap();
4646        attempt.started(4242, harness.clock.now()).unwrap();
4647        harness.store.record_attempt(&attempt).unwrap();
4648        harness.clock.advance_secs(11);
4649        harness.processes.set_alive(true);
4650        harness
4651            .github
4652            .observe(GithubRunnerObservation::NotRegistered);
4653        let replacements = harness.launcher.supervise(&harness.policy).await.unwrap();
4654        assert_eq!(
4655            replacements,
4656            vec![ReplacementIntent {
4657                policy: harness.policy.id,
4658                previous_attempt: id,
4659                operation: "registration_timeout_replacement",
4660            }]
4661        );
4662
4663        assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 1);
4664        assert!(!harness.processes.alive.load(Ordering::SeqCst));
4665        let actions = harness.processes.actions.lock().unwrap().clone();
4666        let intent = actions
4667            .iter()
4668            .position(|action| *action == "terminate_intent")
4669            .unwrap();
4670        let signal = actions
4671            .iter()
4672            .position(|action| *action == "terminate")
4673            .unwrap();
4674        assert!(
4675            intent < signal,
4676            "intent was not durable before signal: {actions:?}"
4677        );
4678
4679        let cleaned = harness.store.attempt(id).unwrap().unwrap();
4680        assert!(matches!(
4681            cleaned.outcome(),
4682            Some(AttemptOutcome::Failed {
4683                reason: FailureReason::TerminatedAfterRegistrationTimeout
4684            })
4685        ));
4686        let events = harness.events.events();
4687        let intent = events
4688            .iter()
4689            .position(|event| matches!(event, AttemptEvent::TerminateIntent { .. }))
4690            .unwrap();
4691        let stopped = events
4692            .iter()
4693            .position(|event| matches!(event, AttemptEvent::Terminated { .. }))
4694            .unwrap();
4695        let concluded = events
4696            .iter()
4697            .position(|event| matches!(event, AttemptEvent::Concluded { .. }))
4698            .unwrap();
4699        assert!(intent < stopped && stopped < concluded, "{events:?}");
4700    }
4701
4702    #[tokio::test]
4703    async fn timeout_crash_recovery_returns_the_same_replacement_intent() {
4704        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4705        let id = AttemptId::new_random();
4706        let runtime = harness
4707            .launcher
4708            .app_paths
4709            .runtime_dir()
4710            .join("timeout-after-crash");
4711        fs::create_dir_all(&runtime).unwrap();
4712        let mut attempt =
4713            RunnerAttempt::allocate(id, harness.policy.id, runtime, harness.clock.now());
4714        attempt.jit_received(harness.clock.now()).unwrap();
4715        attempt.started(4242, harness.clock.now()).unwrap();
4716        harness.store.record_attempt(&attempt).unwrap();
4717        harness.processes.intent.store(true, Ordering::SeqCst);
4718        harness.processes.set_alive(false);
4719        harness
4720            .github
4721            .observe(GithubRunnerObservation::NotRegistered);
4722
4723        let replacements = harness
4724            .launcher
4725            .recover_startup(std::slice::from_ref(&harness.policy))
4726            .await
4727            .unwrap();
4728        assert_eq!(
4729            replacements,
4730            vec![ReplacementIntent {
4731                policy: harness.policy.id,
4732                previous_attempt: id,
4733                operation: "registration_timeout_replacement",
4734            }]
4735        );
4736        let consumed = RunnerLauncher::supervise(&harness.launcher, &harness.policy)
4737            .await
4738            .unwrap();
4739        assert_eq!(consumed, replacements);
4740        assert!(
4741            RunnerLauncher::supervise(&harness.launcher, &harness.policy)
4742                .await
4743                .unwrap()
4744                .is_empty(),
4745            "startup replacement evidence must be consumed exactly once by e1"
4746        );
4747        assert!(matches!(
4748            harness.store.attempt(id).unwrap().unwrap().outcome(),
4749            Some(AttemptOutcome::Failed {
4750                reason: FailureReason::TerminatedAfterRegistrationTimeout
4751            })
4752        ));
4753    }
4754
4755    #[tokio::test]
4756    async fn terminate_intent_sync_failure_prevents_signal_and_conclusion() {
4757        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4758        let id = AttemptId::new_random();
4759        let runtime = harness
4760            .launcher
4761            .app_paths
4762            .runtime_dir()
4763            .join("timeout-sync-failure");
4764        fs::create_dir_all(&runtime).unwrap();
4765        let mut attempt =
4766            RunnerAttempt::allocate(id, harness.policy.id, &runtime, harness.clock.now());
4767        attempt.jit_received(harness.clock.now()).unwrap();
4768        attempt.started(4242, harness.clock.now()).unwrap();
4769        harness.store.record_attempt(&attempt).unwrap();
4770        harness.clock.advance_secs(11);
4771        harness.processes.set_alive(true);
4772        harness.processes.fail_intent();
4773        harness
4774            .github
4775            .observe(GithubRunnerObservation::NotRegistered);
4776
4777        assert!(
4778            harness
4779                .launcher
4780                .recover_startup(std::slice::from_ref(&harness.policy))
4781                .await
4782                .is_err()
4783        );
4784        assert_eq!(harness.processes.terminations.load(Ordering::SeqCst), 0);
4785        assert!(harness.processes.alive.load(Ordering::SeqCst));
4786        assert_eq!(
4787            harness.store.attempt(id).unwrap().unwrap().state(),
4788            AttemptState::Starting
4789        );
4790        assert!(!harness.events.events().iter().any(|event| matches!(
4791            event,
4792            AttemptEvent::Terminated { attempt } | AttemptEvent::Concluded { attempt, .. }
4793                if *attempt == id
4794        )));
4795    }
4796
4797    #[tokio::test]
4798    async fn diagnostics_survive_cleanup_without_the_jit_or_a_token() {
4799        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
4800        harness.ready().await;
4801        let attempt = harness.launch().await;
4802        harness
4803            .github
4804            .observe(GithubRunnerObservation::Registered { busy: false });
4805        harness.launcher.supervise(&harness.policy).await.unwrap();
4806        harness.clock.advance_secs(11);
4807        harness.processes.set_alive(false);
4808        harness
4809            .github
4810            .observe(GithubRunnerObservation::NotRegistered);
4811        harness.launcher.supervise(&harness.policy).await.unwrap();
4812        let diagnostic = fs::read_to_string(
4813            harness
4814                .launcher
4815                .diagnostics_root
4816                .join(format!("{}.log", attempt.id)),
4817        )
4818        .unwrap();
4819        assert!(diagnostic.contains("exited_idle_without_work"));
4820        assert!(!diagnostic.contains(JIT));
4821        assert!(!diagnostic.contains("ghp_"));
4822        assert!(!attempt.runtime_path().exists());
4823    }
4824
4825    #[test]
4826    fn native_process_listing_never_contains_jit_and_handoffs_never_survive() {
4827        let root = tempfile::tempdir().unwrap();
4828        let policy = fixtures::policy()
4829            .repository("octo/repo")
4830            .autoscale("home", 1)
4831            .active()
4832            .build();
4833        let runtime = root.path().join("successful");
4834        fs::create_dir_all(&runtime).unwrap();
4835        let processes = NativeProcesses::new();
4836        let config = EncodedJitConfig::new(JIT);
4837        let handoff =
4838            RestrictiveHandoff::create(&runtime, SecretString::from(config.expose().to_owned()))
4839                .unwrap();
4840        let mut child = native_inspection_spec()
4841            .spawn_runner_with_handoff(&handoff)
4842            .expect("native child starts");
4843        let pid = child.pid();
4844        handoff.delete().unwrap();
4845        let command_line = native_command_line(pid);
4846        assert!(
4847            !command_line.contains(JIT),
4848            "the encoded JIT configuration appeared in the native process listing"
4849        );
4850        assert_no_jit_file(&runtime);
4851        child
4852            .stop(Duration::from_secs(1))
4853            .expect("native child stops");
4854
4855        let failed_runtime = root.path().join("failed");
4856        fs::create_dir_all(&failed_runtime).unwrap();
4857        let failed = RunnerAttempt::allocate(
4858            AttemptId::new_random(),
4859            policy.id,
4860            &failed_runtime,
4861            FakeClock::default().now(),
4862        );
4863        assert!(
4864            processes
4865                .spawn(&failed, &EncodedJitConfig::new(JIT))
4866                .is_err(),
4867            "a runtime with no runner executable must fail"
4868        );
4869        assert_no_jit_file(&failed_runtime);
4870        processes
4871            .record_terminate_intent(&failed)
4872            .expect("the intent file and its directory entry are durably synced");
4873        assert_eq!(
4874            fs::read(NativeProcesses::intent_path(&failed)).unwrap(),
4875            b"registration-timeout\n"
4876        );
4877    }
4878
4879    #[test]
4880    fn post_spawn_boundaries_are_bounded_durable_and_never_retry_jit() {
4881        let root = tempfile::tempdir().unwrap();
4882        let policy = fixtures::policy()
4883            .repository("octo/repo")
4884            .autoscale("home", 1)
4885            .active()
4886            .build();
4887        let processes = NativeProcesses::new();
4888        processes.use_long_lived_test_listener();
4889        for (index, boundary) in [
4890            PostSpawnBoundary::HandoffDelete,
4891            PostSpawnBoundary::IdentitySerialize,
4892            PostSpawnBoundary::IdentityWrite,
4893            PostSpawnBoundary::ChildMapInsert,
4894        ]
4895        .into_iter()
4896        .enumerate()
4897        {
4898            let runtime = root.path().join(format!("post-spawn-{index}"));
4899            let bin = runtime.join("bin");
4900            fs::create_dir_all(&bin).unwrap();
4901            #[cfg(windows)]
4902            let listener = bin.join("Runner.Listener.exe");
4903            #[cfg(not(windows))]
4904            let listener = bin.join("Runner.Listener");
4905            fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4906            let attempt = RunnerAttempt::allocate(
4907                AttemptId::new_random(),
4908                policy.id,
4909                &runtime,
4910                FakeClock::default().now(),
4911            );
4912            processes.fail_post_spawn_at(boundary);
4913            let failure = processes
4914                .spawn(&attempt, &EncodedJitConfig::new(JIT))
4915                .expect_err("fault must cross the post-spawn cleanup path");
4916            assert!(!failure.retryable, "{boundary:?} allowed duplicate retry");
4917            assert!(
4918                !processes.is_alive(&attempt).unwrap(),
4919                "{boundary:?} left a child"
4920            );
4921            assert!(!NativeProcesses::identity_path(&attempt).exists());
4922            assert_no_jit_file(&runtime);
4923        }
4924        assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
4925
4926        let runtime = root.path().join("identity-and-stop-fail");
4927        let bin = runtime.join("bin");
4928        fs::create_dir_all(&bin).unwrap();
4929        #[cfg(windows)]
4930        let listener = bin.join("Runner.Listener.exe");
4931        #[cfg(not(windows))]
4932        let listener = bin.join("Runner.Listener");
4933        fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4934        let attempt = RunnerAttempt::allocate(
4935            AttemptId::new_random(),
4936            policy.id,
4937            &runtime,
4938            FakeClock::default().now(),
4939        );
4940        // The first fault rejects the normal identity write; the second rejects
4941        // its retry after the first stop fails. The fallback sidecar must make
4942        // the live-child result durable without an unbounded reap loop.
4943        processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4944        processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4945        processes.fail_next_post_spawn_stop();
4946        let failure = processes
4947            .spawn(&attempt, &EncodedJitConfig::new(JIT))
4948            .expect_err("the identity boundary must fail closed");
4949        assert!(failure.live_pid.is_some());
4950        assert_long_lived_listener_ready(&processes, &attempt);
4951        assert!(processes.is_alive(&attempt).unwrap());
4952        assert!(!NativeProcesses::identity_path(&attempt).exists());
4953        assert!(NativeProcesses::fallback_identity_path(&attempt).is_file());
4954        assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
4955        processes.terminate(&attempt).unwrap();
4956
4957        let runtime = root.path().join("persistent-stop-and-identity-failures");
4958        let bin = runtime.join("bin");
4959        fs::create_dir_all(&bin).unwrap();
4960        #[cfg(windows)]
4961        let listener = bin.join("Runner.Listener.exe");
4962        #[cfg(not(windows))]
4963        let listener = bin.join("Runner.Listener");
4964        fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
4965        let mut unresolved = RunnerAttempt::allocate(
4966            AttemptId::new_random(),
4967            policy.id,
4968            &runtime,
4969            FakeClock::default().now(),
4970        );
4971        for _ in 0..3 {
4972            processes.fail_post_spawn_at(PostSpawnBoundary::IdentityWrite);
4973        }
4974        processes.fail_post_spawn_stops(MAX_POST_SPAWN_STOP_ATTEMPTS);
4975        let failure = processes
4976            .spawn(&unresolved, &EncodedJitConfig::new(JIT))
4977            .expect_err("bounded cleanup must return even when every stop errors");
4978        let pid = failure
4979            .live_pid
4980            .expect("the owned child remains supervised in this invocation");
4981        assert!(matches!(failure.reason, FailureReason::Other(_)));
4982        assert_long_lived_listener_ready(&processes, &unresolved);
4983        unresolved.jit_received(FakeClock::default().now()).unwrap();
4984        unresolved.started(pid, FakeClock::default().now()).unwrap();
4985        let journal = SqliteStore::open_in_memory().unwrap();
4986        journal.record_attempt(&unresolved).unwrap();
4987        let recovered = journal.attempt(unresolved.id).unwrap().unwrap();
4988        assert_eq!(recovered.process_id(), Some(pid));
4989        assert_eq!(recovered.state(), AttemptState::Starting);
4990        assert!(processes.is_alive(&unresolved).unwrap());
4991        assert!(!NativeProcesses::identity_path(&unresolved).exists());
4992        assert!(!NativeProcesses::fallback_identity_path(&unresolved).exists());
4993        assert_eq!(
4994            fs::read_to_string(NativeProcesses::unresolved_process_path(&unresolved)).unwrap(),
4995            pid.to_string(),
4996            "bounded cleanup must leave durable unresolved-process evidence before returning"
4997        );
4998        assert!(
4999            NativeProcesses::new().is_alive(&recovered).is_err(),
5000            "restart must fail closed on the durable starting/PID journal rather than trust a bare PID"
5001        );
5002        processes.terminate(&unresolved).unwrap();
5003
5004        let runtime = root.path().join("post-spawn-stop-failed");
5005        let bin = runtime.join("bin");
5006        fs::create_dir_all(&bin).unwrap();
5007        #[cfg(windows)]
5008        let listener = bin.join("Runner.Listener.exe");
5009        #[cfg(not(windows))]
5010        let listener = bin.join("Runner.Listener");
5011        fs::copy(std::env::current_exe().unwrap(), &listener).unwrap();
5012        let attempt = RunnerAttempt::allocate(
5013            AttemptId::new_random(),
5014            policy.id,
5015            &runtime,
5016            FakeClock::default().now(),
5017        );
5018        processes.fail_post_spawn_at(PostSpawnBoundary::ChildMapInsert);
5019        processes.fail_next_post_spawn_stop();
5020        let failure = processes
5021            .spawn(&attempt, &EncodedJitConfig::new(JIT))
5022            .expect_err("the injected stop failure must preserve supervision");
5023        let live_pid = failure
5024            .live_pid
5025            .expect("live PID is returned to the journal");
5026        assert!(!failure.retryable);
5027        assert_long_lived_listener_ready(&processes, &attempt);
5028        assert!(NativeProcesses::identity_path(&attempt).is_file());
5029        assert_eq!(
5030            NativeProcesses::read_identity(&attempt)
5031                .unwrap()
5032                .unwrap()
5033                .pid(),
5034            live_pid
5035        );
5036        assert_eq!(processes.post_spawn_reaps.load(Ordering::SeqCst), 4);
5037        processes.terminate(&attempt).unwrap();
5038    }
5039
5040    #[test]
5041    #[ignore = "spawned only as the platform-stable native listener fixture"]
5042    fn long_lived_native_listener_helper() {
5043        let ready = std::env::var_os("RUNNER_MANAGER_TEST_LISTENER_READY")
5044            .map(PathBuf::from)
5045            .expect("the parent supplies the readiness path");
5046        fs::write(ready, b"ready\n").expect("the listener publishes readiness");
5047        std::thread::sleep(Duration::from_secs(30));
5048    }
5049
5050    fn assert_long_lived_listener_ready(processes: &NativeProcesses, attempt: &RunnerAttempt) {
5051        let ready = attempt.runtime_path().join(TEST_LISTENER_READY);
5052        let deadline = std::time::Instant::now() + Duration::from_secs(5);
5053        loop {
5054            if ready.is_file() {
5055                assert_eq!(fs::read(&ready).unwrap(), b"ready\n");
5056                return;
5057            }
5058            assert!(
5059                processes.is_alive(attempt).unwrap(),
5060                "the native listener exited before publishing readiness"
5061            );
5062            assert!(
5063                std::time::Instant::now() < deadline,
5064                "the native listener stayed alive but never published readiness"
5065            );
5066            std::thread::sleep(Duration::from_millis(10));
5067        }
5068    }
5069
5070    #[tokio::test]
5071    async fn every_production_launch_prunes_under_the_same_allocation_guard() {
5072        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
5073        harness.ready().await;
5074        assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 0);
5075        harness.launch().await;
5076        assert_eq!(harness.packages.prunes.load(Ordering::SeqCst), 1);
5077        assert_eq!(
5078            *harness.packages.prune_currents.lock().unwrap(),
5079            vec![harness.packages.version.clone()],
5080            "the leased current version is an exclusion, never the prune target"
5081        );
5082    }
5083
5084    fn assert_no_jit_file(runtime: &Path) {
5085        for entry in fs::read_dir(runtime).unwrap() {
5086            let path = entry.unwrap().path();
5087            if path.is_file() {
5088                let bytes = fs::read(&path).unwrap();
5089                assert!(
5090                    !bytes
5091                        .windows(JIT.len())
5092                        .any(|window| window == JIT.as_bytes()),
5093                    "a JIT payload survived in a runtime file"
5094                );
5095            }
5096        }
5097    }
5098
5099    #[test]
5100    fn production_listener_command_uses_the_supported_jit_contract() {
5101        let runtime = Path::new("runtime");
5102        let spec = runner_listener_spec(PathBuf::from("Runner.Listener"), runtime);
5103        let arguments: Vec<_> = spec
5104            .arguments()
5105            .iter()
5106            .map(|argument| argument.to_string_lossy().into_owned())
5107            .collect();
5108
5109        assert_eq!(arguments, ["run"]);
5110        assert!(
5111            !arguments
5112                .iter()
5113                .any(|argument| argument == "--jit-config-file"),
5114            "the obsolete file option would be rejected by Runner.Listener 2.336.0"
5115        );
5116    }
5117
5118    #[cfg(windows)]
5119    fn native_inspection_spec() -> SpawnSpec {
5120        SpawnSpec::new("powershell.exe").args([
5121            "-NoProfile",
5122            "-NonInteractive",
5123            "-Command",
5124            "Start-Sleep -Seconds 30",
5125        ])
5126    }
5127
5128    #[cfg(unix)]
5129    fn native_inspection_spec() -> SpawnSpec {
5130        SpawnSpec::new("/bin/sh").args(["-c", "sleep 30"])
5131    }
5132
5133    #[cfg(windows)]
5134    fn native_command_line(pid: u32) -> String {
5135        let output = std::process::Command::new("powershell.exe")
5136            .args([
5137                "-NoProfile",
5138                "-NonInteractive",
5139                "-Command",
5140                &format!("(Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}').CommandLine"),
5141            ])
5142            .output()
5143            .expect("PowerShell can inspect the native child");
5144        assert!(output.status.success(), "native process inspection failed");
5145        String::from_utf8(output.stdout).expect("Windows command lines are Unicode")
5146    }
5147
5148    #[cfg(target_os = "linux")]
5149    fn native_command_line(pid: u32) -> String {
5150        fs::read(format!("/proc/{pid}/cmdline"))
5151            .map(|bytes| String::from_utf8_lossy(&bytes).replace('\0', " "))
5152            .expect("/proc exposes the native child command line")
5153    }
5154
5155    #[cfg(target_os = "macos")]
5156    fn native_command_line(pid: u32) -> String {
5157        let output = std::process::Command::new("ps")
5158            .args(["-o", "command=", "-p", &pid.to_string()])
5159            .output()
5160            .expect("ps can inspect the native child");
5161        assert!(output.status.success(), "native process inspection failed");
5162        String::from_utf8(output.stdout).expect("the command line is UTF-8")
5163    }
5164
5165    // -- c2: persistent slot allocation -------------------------------------
5166
5167    #[tokio::test]
5168    async fn a_persistent_repository_leases_s1_and_journals_it_before_any_github_effect() {
5169        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5170            .with_persistent_workspace(2);
5171        harness.github.watch_journal(Arc::clone(&harness.store));
5172        harness.ready().await;
5173
5174        let attempt = harness.launch().await;
5175
5176        assert_eq!(
5177            attempt.workspace(),
5178            AttemptWorkspace::persistent_slot(nz(1)),
5179            "the lowest free slot is leased"
5180        );
5181        assert_eq!(attempt.runtime_path(), harness.slot_path(1));
5182        assert!(attempt.holds_slot_lease());
5183        // The exact runtime path is journalled, not re-derived later.
5184        assert_eq!(
5185            harness.attempt(attempt.id).runtime_path(),
5186            harness.slot_path(1)
5187        );
5188
5189        // Step 7 of "Slot allocation": the lease exists before GitHub is asked
5190        // for anything, and the runner's work folder stays the relative `_work`
5191        // the slot root is laid out around.
5192        let facts = harness.github.registration_facts();
5193        assert_eq!(facts.len(), 1);
5194        assert_eq!(
5195            facts[0].leased_slots,
5196            vec![1],
5197            "the lease was journalled first"
5198        );
5199        assert_eq!(facts[0].work_folder, DEFAULT_WORK_FOLDER);
5200    }
5201
5202    #[tokio::test]
5203    async fn a_terminal_but_uncleaned_attempt_keeps_its_slot_without_holding_capacity() {
5204        let harness = Harness::new(
5205            FakeGithubLifecycle::default().fail(true),
5206            Arc::new(PersistentDemand),
5207        )
5208        .with_persistent_workspace(2);
5209        harness.ready().await;
5210
5211        // A terminal JIT refusal concludes the attempt without cleaning it.
5212        harness.launch_result().await.unwrap_err();
5213        let first = harness.store.attempts().unwrap().remove(0);
5214        assert_eq!(first.state(), AttemptState::Failed);
5215        assert!(
5216            !first.state().counts_against_capacity(),
5217            "a concluded attempt is invisible to host capacity"
5218        );
5219        assert!(
5220            first.holds_slot_lease(),
5221            "and still owns its directory, so its slot is not free"
5222        );
5223
5224        let second = harness.launch().await;
5225        assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5226        assert_eq!(second.runtime_path(), harness.slot_path(2));
5227        assert_eq!(
5228            harness
5229                .store
5230                .slot_leases_for_policy(harness.policy.id)
5231                .unwrap()
5232                .len(),
5233            2
5234        );
5235    }
5236
5237    #[tokio::test]
5238    async fn two_sequential_allocations_at_capacity_one_reuse_s1_and_its_retained_work() {
5239        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5240            .with_persistent_workspace(1);
5241        harness.ready().await;
5242
5243        let first = harness.launch().await;
5244        assert_eq!(first.runtime_path(), harness.slot_path(1));
5245
5246        // What a job leaves behind, at the path the runner writes it to.
5247        let checkout = harness.slot_path(1).join(DEFAULT_WORK_FOLDER).join("repo");
5248        fs::create_dir_all(&checkout).unwrap();
5249        fs::write(checkout.join("checkout.txt"), b"from the first job").unwrap();
5250
5251        harness.cleanup_retaining_work(first.id).await;
5252
5253        let second = harness.launch().await;
5254        assert_ne!(second.id, first.id);
5255        assert_eq!(
5256            second.workspace(),
5257            AttemptWorkspace::persistent_slot(nz(1)),
5258            "a released slot is leased again rather than skipped"
5259        );
5260        assert_eq!(
5261            second.runtime_path(),
5262            first.runtime_path(),
5263            "the same slot is the same exact path"
5264        );
5265        assert_eq!(
5266            fs::read_to_string(checkout.join("checkout.txt")).unwrap(),
5267            "from the first job",
5268            "the retained job workspace survived the second allocation"
5269        );
5270        // The attempt's own runner material was recreated for this attempt.
5271        assert!(harness.slot_path(1).join("runner-package").exists());
5272    }
5273
5274    #[tokio::test]
5275    async fn lowering_capacity_leaves_higher_slots_alone_and_raising_it_permits_them_again() {
5276        let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5277            .with_persistent_workspace(2);
5278        harness.ready().await;
5279
5280        let first = harness.launch().await;
5281        let second = harness.launch().await;
5282        assert_eq!(second.runtime_path(), harness.slot_path(2));
5283        let kept = harness
5284            .slot_path(2)
5285            .join(DEFAULT_WORK_FOLDER)
5286            .join("kept.txt");
5287        fs::create_dir_all(kept.parent().unwrap()).unwrap();
5288        fs::write(&kept, b"s2 was here").unwrap();
5289        harness.cleanup_retaining_work(second.id).await;
5290
5291        // The operator lowers the ceiling while s1 is still leased.
5292        harness.policy.set_max_capacity(nz(1)).unwrap();
5293        let refusal = harness.launch_result().await.unwrap_err().to_string();
5294        assert!(
5295            refusal.contains("s1 to s1"),
5296            "the refusal names the ceiling it reached: {refusal}"
5297        );
5298        assert!(
5299            harness.slot_path(2).exists() && kept.exists(),
5300            "lowering capacity deletes nothing; the higher slot is merely unusable"
5301        );
5302
5303        // Raising it again makes the free higher slot available.
5304        harness.policy.set_max_capacity(nz(2)).unwrap();
5305        let third = harness.launch().await;
5306        assert_eq!(third.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5307        assert_eq!(third.runtime_path(), harness.slot_path(2));
5308        assert_eq!(fs::read_to_string(&kept).unwrap(), "s2 was here");
5309        assert!(first.holds_slot_lease(), "s1 was never disturbed");
5310    }
5311
5312    #[tokio::test]
5313    async fn organization_and_ephemeral_policies_never_enter_slot_allocation() {
5314        for policy in [
5315            fixtures::policy()
5316                .organization("octo")
5317                .autoscale("home", 2)
5318                .active()
5319                .build(),
5320            fixtures::policy()
5321                .repository("octo/repo")
5322                .autoscale("home", 2)
5323                .active()
5324                .build(),
5325        ] {
5326            let mut harness =
5327                Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5328                    .with_host_runner_root();
5329            assert_eq!(policy.workspace_policy(), &WorkspacePolicy::Ephemeral);
5330            harness.policy = policy;
5331            harness.ready().await;
5332
5333            let attempt = harness.launch().await;
5334            assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
5335            assert_eq!(attempt.workspace().slot_number(), None);
5336            assert!(!attempt.holds_slot_lease());
5337            assert_eq!(
5338                attempt.runtime_path().parent().unwrap(),
5339                harness.host_root(),
5340                "a disposable attempt is a child of the host root, never of a slot"
5341            );
5342            assert!(
5343                harness
5344                    .store
5345                    .slot_leases_for_policy(harness.policy.id)
5346                    .unwrap()
5347                    .is_empty()
5348            );
5349        }
5350    }
5351
5352    #[tokio::test]
5353    async fn two_concurrent_allocations_never_share_a_slot() {
5354        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5355            .with_persistent_workspace(2);
5356        harness.github.watch_journal(Arc::clone(&harness.store));
5357        harness.ready().await;
5358
5359        // Both allocators race for the same host allocation lock, which is what
5360        // orders slot *selection*; each one then journals its lease before it
5361        // asks GitHub for anything.
5362        let (first, second) = tokio::join!(harness.launch_result(), harness.launch_result());
5363        let first = first.unwrap();
5364        let second = second.unwrap();
5365
5366        let slots: BTreeSet<u16> = [&first, &second]
5367            .iter()
5368            .map(|attempt| {
5369                attempt
5370                    .workspace()
5371                    .slot_number()
5372                    .expect("a persistent attempt leases a slot")
5373            })
5374            .collect();
5375        assert_eq!(slots, BTreeSet::from([1, 2]), "one slot each, never shared");
5376        assert_ne!(first.runtime_path(), second.runtime_path());
5377        assert_eq!(
5378            harness
5379                .store
5380                .slot_leases_for_policy(harness.policy.id)
5381                .unwrap()
5382                .len(),
5383            2
5384        );
5385
5386        // Every registration saw *its own* lease already in the journal. Reading
5387        // the whole journal and asking only that it be non-empty would pass on
5388        // the other allocator's lease, which is precisely the ordering bug this
5389        // test exists to exclude.
5390        let facts = harness.github.registration_facts();
5391        assert_eq!(facts.len(), 2);
5392        for fact in facts {
5393            let attempt = [&first, &second]
5394                .into_iter()
5395                .find(|attempt| runner_name(attempt.id) == fact.runner_name)
5396                .expect("every registration belongs to one of the two attempts");
5397            let slot = attempt
5398                .workspace()
5399                .slot_number()
5400                .expect("a persistent attempt leases a slot");
5401            assert!(
5402                fact.leased_slots.contains(&slot),
5403                "a JIT request never precedes its own lease: s{slot} not in {:?}",
5404                fact.leased_slots
5405            );
5406            assert_eq!(fact.work_folder, DEFAULT_WORK_FOLDER);
5407        }
5408    }
5409
5410    #[tokio::test]
5411    async fn the_database_is_the_final_fence_against_two_attempts_in_one_slot() {
5412        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5413            .with_persistent_workspace(2);
5414        harness.ready().await;
5415        let first = harness.launch().await;
5416
5417        // What a second allocator that lost the race would write: the lock
5418        // orders selection, and this index is what catches a writer the lock
5419        // could not see.
5420        let clash = RunnerAttempt::allocate_in(
5421            AttemptId::new_random(),
5422            harness.policy.id,
5423            first.runtime_path(),
5424            AttemptWorkspace::persistent_slot(nz(1)),
5425            harness.clock.now(),
5426        );
5427        assert!(matches!(
5428            harness.store.record_attempt(&clash).unwrap_err(),
5429            StoreError::SlotAlreadyLeased { slot: 1, .. }
5430        ));
5431
5432        // And the launcher reports it as itself rather than as a generic
5433        // journal failure, so the operator reads what actually happened.
5434        let error = harness.launcher.record_allocation(&clash).unwrap_err();
5435        let rendered = error.to_string();
5436        assert!(rendered.contains("slot s1"), "{rendered}");
5437        assert!(rendered.contains("nothing was written"), "{rendered}");
5438        assert_eq!(
5439            harness.store.attempts().unwrap().len(),
5440            1,
5441            "the losing allocator journalled nothing"
5442        );
5443    }
5444
5445    #[test]
5446    fn slot_selection_fills_the_lowest_gap_and_stops_at_the_ceiling() {
5447        let leased = |slots: &[u16]| -> Vec<RunnerAttempt> {
5448            slots
5449                .iter()
5450                .map(|slot| {
5451                    RunnerAttempt::allocate_in(
5452                        AttemptId::new_random(),
5453                        fixtures::POLICY_ID,
5454                        format!("/srv/rman/acme/s{slot}"),
5455                        AttemptWorkspace::persistent_slot(nz(*slot)),
5456                        fixtures::created_at(),
5457                    )
5458                })
5459                .collect()
5460        };
5461
5462        assert_eq!(lowest_free_slot(&[], nz(1)), Some(nz(1)));
5463        assert_eq!(lowest_free_slot(&leased(&[1]), nz(4)), Some(nz(2)));
5464        // The gap a released middle slot leaves is filled before the tail.
5465        assert_eq!(lowest_free_slot(&leased(&[1, 3]), nz(4)), Some(nz(2)));
5466        // The ceiling is a refusal, never a reason to allocate past it.
5467        assert_eq!(lowest_free_slot(&leased(&[1]), nz(1)), None);
5468        assert_eq!(lowest_free_slot(&leased(&[1, 2]), nz(2)), None);
5469        // An ephemeral attempt holds no slot and cannot block one.
5470        let ephemeral = vec![RunnerAttempt::allocate(
5471            AttemptId::new_random(),
5472            fixtures::POLICY_ID,
5473            "/srv/rman/host/abc",
5474            fixtures::created_at(),
5475        )];
5476        assert_eq!(lowest_free_slot(&ephemeral, nz(1)), Some(nz(1)));
5477    }
5478
5479    #[test]
5480    fn a_slot_is_reusable_only_when_it_is_empty_or_holds_one_real_work_directory() {
5481        let root = tempfile::tempdir().unwrap();
5482        let slot = root.path().join("s1");
5483        fs::create_dir(&slot).unwrap();
5484        accept_reusable_slot(&slot).expect("an empty slot is reusable");
5485
5486        fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
5487        accept_reusable_slot(&slot).expect("a retained job workspace is reusable");
5488
5489        // Runner material a previous attempt left behind is refused rather than
5490        // reused or removed: deciding those bytes are safe is cleanup's job.
5491        fs::create_dir(slot.join("bin")).unwrap();
5492        fs::write(slot.join(".github-runner-id"), b"73").unwrap();
5493        let refusal = accept_reusable_slot(&slot).unwrap_err().to_string();
5494        assert!(refusal.contains("bin"), "{refusal}");
5495        assert!(refusal.contains(".github-runner-id"), "{refusal}");
5496
5497        // A `_work` that is not a real directory is not a job workspace.
5498        let file_work = root.path().join("s2");
5499        fs::create_dir(&file_work).unwrap();
5500        fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
5501        assert!(accept_reusable_slot(&file_work).is_err());
5502    }
5503
5504    #[cfg(unix)]
5505    #[test]
5506    fn a_link_shaped_work_directory_is_refused_rather_than_followed() {
5507        // Windows needs a privilege to create either kind of link, so the
5508        // link-shaped cases are asserted here; the rule itself is
5509        // platform-independent because it is `symlink_metadata`'s answer.
5510        let root = tempfile::tempdir().unwrap();
5511        let elsewhere = root.path().join("elsewhere");
5512        fs::create_dir(&elsewhere).unwrap();
5513
5514        let slot = root.path().join("s1");
5515        fs::create_dir(&slot).unwrap();
5516        std::os::unix::fs::symlink(&elsewhere, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
5517        assert!(accept_reusable_slot(&slot).is_err());
5518
5519        let linked_slot = root.path().join("s2");
5520        std::os::unix::fs::symlink(&elsewhere, &linked_slot).unwrap();
5521        assert!(create_or_validate_slot(&linked_slot).is_err());
5522    }
5523
5524    #[test]
5525    fn a_slot_standing_where_a_file_is_refuses_rather_than_replacing_it() {
5526        let root = tempfile::tempdir().unwrap();
5527        let occupied = root.path().join("s1");
5528        fs::write(&occupied, b"an operator's file").unwrap();
5529        let refusal = create_or_validate_slot(&occupied).unwrap_err().to_string();
5530        assert!(refusal.contains("is not a directory"), "{refusal}");
5531        assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
5532
5533        let fresh = root.path().join("s2");
5534        create_or_validate_slot(&fresh).expect("a missing slot is created");
5535        assert!(fresh.is_dir());
5536        create_or_validate_slot(&fresh).expect("an existing directory is accepted");
5537    }
5538
5539    #[test]
5540    fn the_retained_work_directory_is_matched_the_way_the_filesystem_matches_it() {
5541        assert!(is_work_folder(OsStr::new(DEFAULT_WORK_FOLDER)));
5542        assert!(!is_work_folder(OsStr::new("_work2")));
5543        // A Windows filesystem is case-insensitive, so `_Work` *is* the retained
5544        // job workspace there and must never be removed as a leftover; on a
5545        // case-sensitive filesystem it is a different directory entirely.
5546        assert_eq!(is_work_folder(OsStr::new("_Work")), cfg!(windows));
5547    }
5548
5549    #[test]
5550    fn package_materialization_never_overwrites_or_follows_a_retained_work_directory() {
5551        let root = tempfile::tempdir().unwrap();
5552        let package = root.path().join("package");
5553        fs::create_dir_all(package.join("bin")).unwrap();
5554        fs::write(package.join("bin").join("Runner.Listener"), b"binary").unwrap();
5555        // A nested `_work` inside the package's own tree is an ordinary name.
5556        fs::create_dir_all(package.join("externals").join(DEFAULT_WORK_FOLDER)).unwrap();
5557
5558        let slot = root.path().join("s1");
5559        let retained = slot.join(DEFAULT_WORK_FOLDER).join("repo");
5560        fs::create_dir_all(&retained).unwrap();
5561        fs::write(retained.join("checkout.txt"), b"from the first job").unwrap();
5562
5563        copy_package_tree(&package, &slot).expect("the package lays out around `_work`");
5564        assert!(slot.join("bin").join("Runner.Listener").exists());
5565        assert!(
5566            slot.join("externals").join(DEFAULT_WORK_FOLDER).is_dir(),
5567            "the guard is top-level only"
5568        );
5569        assert_eq!(
5570            fs::read_to_string(retained.join("checkout.txt")).unwrap(),
5571            "from the first job"
5572        );
5573
5574        // A package that ever grew a top-level `_work` is refused, not merged.
5575        fs::create_dir(package.join(DEFAULT_WORK_FOLDER)).unwrap();
5576        let error = copy_package_tree(&package, &slot).unwrap_err();
5577        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
5578        assert_eq!(
5579            fs::read_to_string(retained.join("checkout.txt")).unwrap(),
5580            "from the first job"
5581        );
5582    }
5583
5584    #[test]
5585    fn rolling_back_a_materialization_keeps_a_slot_but_removes_a_disposable_directory() {
5586        let root = tempfile::tempdir().unwrap();
5587
5588        let slot = root.path().join("s1");
5589        let retained = slot.join(DEFAULT_WORK_FOLDER);
5590        fs::create_dir_all(retained.join("repo")).unwrap();
5591        fs::write(retained.join("repo").join("checkout.txt"), b"kept").unwrap();
5592        fs::create_dir_all(slot.join("bin")).unwrap();
5593        fs::write(slot.join(".github-runner-id"), b"73").unwrap();
5594        let persistent = RunnerAttempt::allocate_in(
5595            AttemptId::new_random(),
5596            fixtures::POLICY_ID,
5597            &slot,
5598            AttemptWorkspace::persistent_slot(nz(1)),
5599            fixtures::created_at(),
5600        );
5601
5602        remove_materialized_package(&persistent).unwrap();
5603        assert!(slot.is_dir(), "the slot itself is not removed");
5604        assert!(!slot.join("bin").exists());
5605        assert!(!slot.join(".github-runner-id").exists());
5606        assert_eq!(
5607            fs::read_to_string(retained.join("repo").join("checkout.txt")).unwrap(),
5608            "kept"
5609        );
5610
5611        let disposable_path = root.path().join("abcdef012345");
5612        fs::create_dir_all(disposable_path.join(DEFAULT_WORK_FOLDER)).unwrap();
5613        let disposable = RunnerAttempt::allocate(
5614            AttemptId::new_random(),
5615            fixtures::POLICY_ID,
5616            &disposable_path,
5617            fixtures::created_at(),
5618        );
5619        remove_materialized_package(&disposable).unwrap();
5620        assert!(
5621            !disposable_path.exists(),
5622            "a disposable directory is still removed whole"
5623        );
5624    }
5625
5626    // -- c3: persistent cleanup and recovery --------------------------------
5627
5628    /// The runner state one attempt leaves at a slot root, as a real attempt
5629    /// leaves it: binaries, registration identity, a JIT handoff that outlived
5630    /// its process, and this agent's own lifecycle sidecars.
5631    ///
5632    /// Driven by [`SENSITIVE_SLOT_ENTRIES`] rather than by a second copy of it,
5633    /// so a name added to the thing cleanup must prove absent is a name every
5634    /// test here starts leaving behind.
5635    fn litter_the_slot(slot: &Path) {
5636        for directory in ["bin", "externals", "_diag"] {
5637            fs::create_dir_all(slot.join(directory)).unwrap();
5638        }
5639        fs::write(slot.join("bin").join("Runner.Listener"), b"binary").unwrap();
5640        for file in SENSITIVE_SLOT_ENTRIES
5641            .iter()
5642            .filter(|entry| !slot.join(entry).is_dir())
5643        {
5644            fs::write(slot.join(file), b"runner state").unwrap();
5645        }
5646        // A handoff whose owning process died before `Drop` could delete it.
5647        fs::write(
5648            slot.join(format!(
5649                "{}0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0.tmp",
5650                RestrictiveHandoff::NAME_PREFIX
5651            )),
5652            JIT.as_bytes(),
5653        )
5654        .unwrap();
5655    }
5656
5657    /// A marker under `_work` of the kind a job leaves for the next one.
5658    fn retain_under_work(slot: &Path) -> PathBuf {
5659        let checkout = slot.join(DEFAULT_WORK_FOLDER).join("repo").join("target");
5660        fs::create_dir_all(&checkout).unwrap();
5661        let marker = checkout.join("build-output.bin");
5662        fs::write(&marker, RETAINED).unwrap();
5663        marker
5664    }
5665
5666    /// What a job leaves under `_work` for the next job to reuse.
5667    const RETAINED: &str = "a Git-ignored build output the next job reuses";
5668
5669    /// Every direct entry of a directory, sorted, as plain strings.
5670    fn entries_of(directory: &Path) -> Vec<String> {
5671        let mut names: Vec<String> = fs::read_dir(directory)
5672            .unwrap()
5673            .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
5674            .collect();
5675        names.sort();
5676        names
5677    }
5678
5679    /// The one entry a cleaned slot is allowed to hold.
5680    fn only_the_job_workspace() -> Vec<String> {
5681        vec![DEFAULT_WORK_FOLDER.to_owned()]
5682    }
5683
5684    #[cfg(unix)]
5685    #[test]
5686    fn disposable_tree_removal_does_not_open_a_dotnet_diagnostic_fifo() {
5687        use std::sync::mpsc;
5688
5689        let temporary = tempfile::tempdir().unwrap();
5690        let tree = temporary.path().join("attempt");
5691        let diagnostic = tree.join("tmp/clr-debug-pipe-runner-in");
5692        fs::create_dir_all(diagnostic.parent().unwrap()).unwrap();
5693        assert!(
5694            std::process::Command::new("mkfifo")
5695                .arg(&diagnostic)
5696                .status()
5697                .unwrap()
5698                .success()
5699        );
5700
5701        let (finished, result) = mpsc::channel();
5702        std::thread::spawn(move || {
5703            let removed = remove_runtime_tree(&tree);
5704            let _ = finished.send(removed);
5705        });
5706
5707        result
5708            .recv_timeout(Duration::from_secs(2))
5709            .expect("runtime deletion must not wait for a FIFO peer")
5710            .unwrap();
5711        assert!(!diagnostic.exists());
5712    }
5713
5714    /// One slot entry that refuses to be removed, and the undo that lets the
5715    /// temporary directory be torn down afterwards.
5716    ///
5717    /// The two operating systems refuse for different reasons and there is no
5718    /// portable third. Windows will not open a file for deletion while a handle
5719    /// with share mode zero is held on it; Unix will not unlink from a directory
5720    /// the caller cannot write. Both are states a real machine reaches -- a
5721    /// scanner holding a file open, a job that left a directory read-only -- so
5722    /// the injection is a filesystem fact rather than a seam cut into the
5723    /// product for a test to pull.
5724    ///
5725    /// The Unix half is a permission, and permissions do not apply to `root`.
5726    /// [`Self::inject`] proves the block on a throwaway directory before
5727    /// claiming it, so a suite running as `root` says it could not inject rather
5728    /// than asserting nothing and passing.
5729    struct BlockedDeletion {
5730        directory: PathBuf,
5731        #[cfg(windows)]
5732        _handle: fs::File,
5733    }
5734
5735    impl BlockedDeletion {
5736        const HELD: &'static str = "held-open";
5737
5738        /// Fill `directory` with a file that cannot be removed, or answer `None`
5739        /// when this account cannot be stopped from removing anything.
5740        fn inject(directory: &Path) -> Option<Self> {
5741            #[cfg(unix)]
5742            if !Self::refusal_is_possible() {
5743                return None;
5744            }
5745            fs::create_dir_all(directory).unwrap();
5746            fs::write(
5747                directory.join(Self::HELD),
5748                b"a file the scrub cannot remove",
5749            )
5750            .unwrap();
5751            #[cfg(windows)]
5752            let handle = {
5753                use std::os::windows::fs::OpenOptionsExt;
5754
5755                fs::OpenOptions::new()
5756                    .read(true)
5757                    .share_mode(0)
5758                    .open(directory.join(Self::HELD))
5759                    .expect("the blocking handle opens")
5760            };
5761            #[cfg(unix)]
5762            Self::set_mode(directory, 0o555);
5763            Some(Self {
5764                directory: directory.to_path_buf(),
5765                #[cfg(windows)]
5766                _handle: handle,
5767            })
5768        }
5769
5770        fn release(self) {
5771            drop(self);
5772        }
5773
5774        #[cfg(unix)]
5775        fn refusal_is_possible() -> bool {
5776            let probe = tempfile::tempdir().unwrap();
5777            let directory = probe.path().join("probe");
5778            fs::create_dir(&directory).unwrap();
5779            fs::write(directory.join("file"), b"probe").unwrap();
5780            Self::set_mode(&directory, 0o555);
5781            let refused = fs::remove_dir_all(&directory).is_err();
5782            Self::set_mode(&directory, 0o755);
5783            refused
5784        }
5785
5786        #[cfg(unix)]
5787        fn set_mode(directory: &Path, mode: u32) {
5788            use std::os::unix::fs::PermissionsExt;
5789
5790            let mut permissions = fs::metadata(directory).unwrap().permissions();
5791            permissions.set_mode(mode);
5792            fs::set_permissions(directory, permissions).unwrap();
5793        }
5794    }
5795
5796    impl Drop for BlockedDeletion {
5797        fn drop(&mut self) {
5798            #[cfg(unix)]
5799            Self::set_mode(&self.directory, 0o755);
5800            #[cfg(not(unix))]
5801            let _ = &self.directory;
5802        }
5803    }
5804
5805    #[tokio::test]
5806    async fn two_sequential_jobs_keep_the_checkout_and_start_without_the_earlier_runner_state() {
5807        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5808            .with_persistent_workspace(1);
5809        harness.ready().await;
5810
5811        let first = harness.launch().await;
5812        let slot = harness.slot_path(1);
5813        assert_eq!(first.runtime_path(), slot);
5814        assert_eq!(
5815            read_runner_id(&slot),
5816            Some(73),
5817            "the attempt registered, so its identity is on disk"
5818        );
5819        let marker = retain_under_work(&slot);
5820        litter_the_slot(&slot);
5821
5822        harness.cleanup_retaining_work(first.id).await;
5823
5824        // The allowlist is exactly one entry, so this assertion is the security
5825        // property in full: what is retained, and that nothing else is.
5826        assert_eq!(entries_of(&slot), only_the_job_workspace());
5827        assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
5828        assert_eq!(
5829            read_runner_id(&slot),
5830            None,
5831            "the first attempt's registration identity is gone before the second starts"
5832        );
5833        assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
5834        assert!(!harness.attempt(first.id).holds_slot_lease());
5835
5836        let second = harness.launch().await;
5837        assert_ne!(second.id, first.id);
5838        assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(1)));
5839        assert_eq!(
5840            second.runtime_path(),
5841            slot,
5842            "the same slot, so the same retained `_work`"
5843        );
5844        assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
5845    }
5846
5847    #[tokio::test]
5848    async fn cleaning_a_persistent_slot_needs_no_policy_and_scans_no_directory_for_ownership() {
5849        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5850            .with_persistent_workspace(1);
5851        harness.ready().await;
5852        let attempt = harness.launch().await;
5853        let slot = harness.slot_path(1);
5854        let marker = retain_under_work(&slot);
5855        litter_the_slot(&slot);
5856        harness.conclude(attempt.id);
5857
5858        // A repository removed from the product between the attempt concluding
5859        // and the sweep reaching it. The journalled runtime path and slot are
5860        // the only facts left, and `04-security-recovery.md` requires them to be
5861        // enough: the alternative is scanning a root to work out which
5862        // directories were ours, which invariant 6 forbids.
5863        harness
5864            .store
5865            .remove_policy(harness.policy.id, harness.policy.revision())
5866            .unwrap();
5867        assert!(harness.store.policy(harness.policy.id).unwrap().is_none());
5868
5869        harness
5870            .launcher
5871            .clean(attempt.id)
5872            .await
5873            .expect("journal facts alone are enough to clean the slot");
5874
5875        assert_eq!(entries_of(&slot), only_the_job_workspace());
5876        assert!(marker.exists());
5877        assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
5878    }
5879
5880    #[tokio::test]
5881    async fn an_injected_partial_deletion_quarantines_the_slot_across_a_restart() {
5882        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
5883            .with_persistent_workspace(2);
5884        harness.ready().await;
5885        let first = harness.launch().await;
5886        let slot = harness.slot_path(1);
5887        let marker = retain_under_work(&slot);
5888        litter_the_slot(&slot);
5889        harness.conclude(first.id);
5890
5891        let Some(block) = BlockedDeletion::inject(&slot.join("bin")) else {
5892            eprintln!(
5893                "skipped: this account cannot be refused a deletion, so no partial deletion can \
5894                 be injected"
5895            );
5896            return;
5897        };
5898
5899        let refusal = harness
5900            .launcher
5901            .clean(first.id)
5902            .await
5903            .expect_err("a deletion that failed may not report a cleaned slot");
5904        let rendered = refusal.reason.to_string();
5905        assert!(rendered.contains("could not be removed"), "{rendered}");
5906
5907        let held = harness.attempt(first.id);
5908        assert_eq!(held.state(), AttemptState::Failed, "still not cleaned");
5909        assert!(held.holds_slot_lease(), "so the slot is still leased");
5910        assert!(
5911            !held.state().counts_against_capacity(),
5912            "and a concluded attempt still costs the host no capacity"
5913        );
5914
5915        // The same journal and the same directories, under a launcher that
5916        // remembers nothing. Recovery must complete: a host that can launch
5917        // nothing at all because one slot is stuck is not what "does not count
5918        // as active host capacity" means.
5919        let restarted = harness.restart();
5920        restarted
5921            .recover_startup(std::slice::from_ref(&harness.policy))
5922            .await
5923            .expect("one quarantined slot does not stop the host recovering");
5924        assert_eq!(
5925            harness.attempt(first.id).state(),
5926            AttemptState::Failed,
5927            "the quarantine survived the restart"
5928        );
5929        assert!(
5930            harness
5931                .reconcile_events
5932                .events()
5933                .iter()
5934                .any(|event| matches!(
5935                    event,
5936                    LifecycleEvent::AttemptCleanFailed {
5937                        reason: "slot_entry_could_not_be_removed",
5938                        ..
5939                    }
5940                )),
5941            "the refusal is reported rather than retried in silence"
5942        );
5943
5944        // Capacity two, slot one quarantined: the next attempt goes to s2 and
5945        // never to the slot still holding runner state.
5946        let guard = harness.allocation_lock.acquire().await.unwrap();
5947        let second = restarted
5948            .launch(LaunchRequest {
5949                host: &harness.host,
5950                policy: &harness.policy,
5951                allocation_guard: &guard,
5952            })
5953            .await
5954            .expect("the host can still launch");
5955        assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
5956        drop(guard);
5957
5958        // And the same cleanup succeeds once the obstruction is gone, which is
5959        // what "retry through normal recovery" has to mean.
5960        block.release();
5961        restarted
5962            .clean(first.id)
5963            .await
5964            .expect("the retried cleanup completes");
5965        assert_eq!(entries_of(&slot), only_the_job_workspace());
5966        assert!(marker.exists());
5967        assert_eq!(harness.attempt(first.id).state(), AttemptState::Cleaned);
5968    }
5969
5970    /// The disposable half of the same injection, which had no test of its own.
5971    ///
5972    /// `scrub_workspace`'s ephemeral arm turns a failed `remove_dir_all` into
5973    /// `"attempt workspace could not be removed"`, and until now that branch was
5974    /// only reachable in theory: every injected-deletion test drove a persistent
5975    /// slot. The property is the same one and matters for the same reason --
5976    /// `04-security-recovery.md`'s "Cleanup partly fails and the slot is reused
5977    /// anyway" -- but the disposable guarantee is stronger, so a cleanup that
5978    /// reported success over a directory it had not removed would be the
5979    /// contamination gate failing silently rather than a slot being held.
5980    #[tokio::test]
5981    async fn an_injected_deletion_failure_leaves_a_disposable_attempt_uncleaned() {
5982        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand));
5983        harness.ready().await;
5984        let attempt = harness.launch().await;
5985        let runtime = attempt.runtime_path().to_path_buf();
5986        assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
5987        harness.conclude(attempt.id);
5988
5989        let Some(block) = BlockedDeletion::inject(&runtime.join("held-open-subdirectory")) else {
5990            eprintln!(
5991                "skipped: this account cannot be refused a deletion, so no partial deletion can be injected"
5992            );
5993            return;
5994        };
5995
5996        let refusal = harness
5997            .launcher
5998            .clean(attempt.id)
5999            .await
6000            .expect_err("a deletion that failed may not report a removed workspace");
6001        let rendered = refusal.reason.to_string();
6002        assert!(
6003            rendered.contains("could not be removed"),
6004            "the refusal names what happened: {rendered}"
6005        );
6006        assert_ne!(
6007            harness.attempt(attempt.id).state(),
6008            AttemptState::Cleaned,
6009            "an attempt whose directory is still on disk is not cleaned"
6010        );
6011        assert!(
6012            runtime.is_dir(),
6013            "the directory the removal could not finish is still there, which is the fact the journal must keep agreeing with"
6014        );
6015
6016        // And the ordinary retry -- the reconciler's next terminal sweep --
6017        // finishes it once the obstruction is gone.
6018        block.release();
6019        harness
6020            .launcher
6021            .clean(attempt.id)
6022            .await
6023            .expect("the retried cleanup completes");
6024        assert!(!runtime.exists(), "the whole attempt directory goes");
6025        assert_eq!(harness.attempt(attempt.id).state(), AttemptState::Cleaned);
6026    }
6027
6028    #[tokio::test]
6029    async fn changing_a_repository_back_to_ephemeral_leaves_every_old_slot_untouched() {
6030        let mut harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6031            .with_persistent_workspace(1);
6032        harness.ready().await;
6033        let first = harness.launch().await;
6034        let slot = harness.slot_path(1);
6035        let marker = retain_under_work(&slot);
6036        harness.cleanup_retaining_work(first.id).await;
6037
6038        // Every attempt for this policy is cleaned, so the mutation is allowed
6039        // (`04-security-recovery.md`, "Recovery rules"). What it must not do is
6040        // move or delete anything the operator still owns.
6041        harness
6042            .policy
6043            .set_workspace_policy(WorkspacePolicy::Ephemeral)
6044            .unwrap();
6045
6046        let second = harness.launch().await;
6047        assert_eq!(second.workspace(), AttemptWorkspace::Ephemeral);
6048        assert_eq!(
6049            second.runtime_path().parent().unwrap(),
6050            harness.host_root(),
6051            "a disposable attempt is a child of the host root"
6052        );
6053        assert!(slot.is_dir(), "the old slot is left where it stands");
6054        assert_eq!(fs::read_to_string(&marker).unwrap(), RETAINED);
6055
6056        // And cleaning the disposable attempt removes its own directory whole
6057        // without reaching the retained slot beside it.
6058        harness.conclude(second.id);
6059        harness.launcher.clean(second.id).await.unwrap();
6060        assert!(!second.runtime_path().exists());
6061        assert!(marker.exists());
6062    }
6063
6064    #[tokio::test]
6065    async fn a_persistent_slot_is_scrubbed_only_after_the_process_is_signalled_and_gone() {
6066        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6067            .with_persistent_workspace(1);
6068        harness.ready().await;
6069        let slot = harness.slot_path(1);
6070        fs::create_dir_all(&slot).unwrap();
6071        let marker = retain_under_work(&slot);
6072        litter_the_slot(&slot);
6073
6074        let id = AttemptId::new_random();
6075        let mut attempt = RunnerAttempt::allocate_in(
6076            id,
6077            harness.policy.id,
6078            &slot,
6079            AttemptWorkspace::persistent_slot(nz(1)),
6080            harness.clock.now(),
6081        );
6082        attempt.jit_received(harness.clock.now()).unwrap();
6083        attempt.started(4242, harness.clock.now()).unwrap();
6084        harness.store.record_attempt(&attempt).unwrap();
6085        harness.clock.advance_secs(11);
6086        harness.processes.set_alive(true);
6087        harness
6088            .github
6089            .observe(GithubRunnerObservation::NotRegistered);
6090
6091        harness.launcher.supervise(&harness.policy).await.unwrap();
6092
6093        // The identity and termination ordering `e3` established is unchanged by
6094        // the workspace kind: the intent is durable before the signal, and the
6095        // slot is scrubbed only once the process is gone.
6096        let actions = harness.processes.actions.lock().unwrap().clone();
6097        let intent = actions
6098            .iter()
6099            .position(|action| *action == "terminate_intent")
6100            .unwrap();
6101        let signal = actions
6102            .iter()
6103            .position(|action| *action == "terminate")
6104            .unwrap();
6105        assert!(intent < signal, "{actions:?}");
6106        assert!(!harness.processes.alive.load(Ordering::SeqCst));
6107
6108        let cleaned = harness.attempt(id);
6109        assert_eq!(cleaned.state(), AttemptState::Cleaned);
6110        assert!(matches!(
6111            cleaned.outcome(),
6112            Some(AttemptOutcome::Failed {
6113                reason: FailureReason::TerminatedAfterRegistrationTimeout
6114            })
6115        ));
6116        assert_eq!(entries_of(&slot), only_the_job_workspace());
6117        assert!(marker.exists());
6118    }
6119
6120    #[test]
6121    fn a_scrub_retains_one_real_work_directory_and_removes_every_other_entry() {
6122        let root = tempfile::tempdir().unwrap();
6123        let slot = root.path().join("s1");
6124        fs::create_dir(&slot).unwrap();
6125        let marker = retain_under_work(&slot);
6126        litter_the_slot(&slot);
6127        fs::write(slot.join("runner-package"), b"verified").unwrap();
6128
6129        scrub_slot_entries(&slot).expect("a slot of ordinary runner state scrubs");
6130        verify_slot_scrubbed(&slot).expect("and proves it afterwards");
6131
6132        assert_eq!(entries_of(&slot), only_the_job_workspace());
6133        assert!(marker.exists());
6134    }
6135
6136    #[test]
6137    fn a_residue_refusal_never_reports_the_under_count_as_the_fact() {
6138        let slot = Path::new("/runners/s1");
6139
6140        // The ordinary case: the listing counted, so the count is the fact and
6141        // the published names qualify it.
6142        let counted = residue_detail(slot, 2, &["`bin`".to_owned()]);
6143        assert!(counted.contains("2 entries other than"), "{counted}");
6144        assert!(counted.contains("including `bin`"), "{counted}");
6145        assert_eq!(
6146            residue_detail(slot, 1, &[]),
6147            format!(
6148                "1 entry other than `{DEFAULT_WORK_FOLDER}` survived cleanup of {}",
6149                slot.display()
6150            )
6151        );
6152
6153        // The race the second pass exists for: the listing saw nothing and the
6154        // filesystem answered otherwise. Saying "0 entries survived" here would
6155        // state the under-count as the fact and contradict the rest of the
6156        // sentence.
6157        let raced = residue_detail(slot, 0, &["`.credentials`".to_owned()]);
6158        assert!(!raced.contains('0'), "{raced}");
6159        assert!(raced.contains("reported nothing but"), "{raced}");
6160        assert!(raced.contains("`.credentials` survived cleanup"), "{raced}");
6161    }
6162
6163    #[test]
6164    fn verification_asks_the_filesystem_rather_than_the_listing_that_missed_an_entry() {
6165        let root = tempfile::tempdir().unwrap();
6166        let slot = root.path().join("s1");
6167        fs::create_dir(&slot).unwrap();
6168        fs::create_dir(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6169        verify_slot_scrubbed(&slot).expect("only `_work` is a clean slot");
6170
6171        // Runner binaries, registration identity, process identity and this
6172        // agent's lifecycle marks, one at a time, so a scrub that skipped
6173        // exactly one is still caught.
6174        for survivor in ["bin", ".credentials", IDENTITY_FILE, RUNNER_ID_FILE] {
6175            fs::write(slot.join(survivor), b"left behind").unwrap();
6176            let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6177            assert_eq!(quarantine.refusal, SlotRefusal::Residue);
6178            assert!(
6179                quarantine.detail.contains(&format!("`{survivor}`")),
6180                "{quarantine}"
6181            );
6182            fs::remove_file(slot.join(survivor)).unwrap();
6183        }
6184
6185        // A handoff is named by its published prefix, never by the UUID that
6186        // follows it, and never by the payload it holds.
6187        let handoff = slot.join(format!("{}whatever.tmp", RestrictiveHandoff::NAME_PREFIX));
6188        fs::write(&handoff, JIT.as_bytes()).unwrap();
6189        let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6190        assert!(
6191            quarantine.detail.contains("an encoded JIT handoff"),
6192            "{quarantine}"
6193        );
6194        assert!(!quarantine.detail.contains(JIT), "{quarantine}");
6195        fs::remove_file(&handoff).unwrap();
6196
6197        // A name a workflow chose is counted and never echoed: a slot root is
6198        // writable by the job, so a file named after a secret would be published
6199        // by any message that repeated the listing.
6200        fs::write(slot.join("ghp_DO_NOT_LEAK"), b"named by the job").unwrap();
6201        let quarantine = verify_slot_scrubbed(&slot).unwrap_err();
6202        assert!(
6203            quarantine.detail.contains("1 entry other than"),
6204            "{quarantine}"
6205        );
6206        assert!(
6207            !quarantine.detail.contains("ghp_DO_NOT_LEAK"),
6208            "{quarantine}"
6209        );
6210    }
6211
6212    #[test]
6213    fn a_slot_is_derived_from_the_journal_and_refused_when_it_disagrees() {
6214        let root = tempfile::tempdir().unwrap();
6215        let configured =
6216            LocalAbsolutePath::new(root.path().to_str().unwrap()).expect("a local absolute root");
6217        let slot = configured.as_path().join("s1");
6218        fs::create_dir(&slot).unwrap();
6219
6220        verify_journalled_slot(&slot, nz(1), Some(&configured))
6221            .expect("the journalled slot agrees");
6222        verify_journalled_slot(&slot, nz(1), None)
6223            .expect("and a policy that is gone removes a check, not the ability to clean");
6224
6225        // The journalled slot number is what names the directory. `s1` recorded
6226        // as slot two is corrupt state, not a slot to clean.
6227        assert_eq!(
6228            verify_journalled_slot(&slot, nz(2), None)
6229                .unwrap_err()
6230                .refusal,
6231            SlotRefusal::NotTheJournalledSlot
6232        );
6233        for stray in ["s1/nested", "not-a-slot", "s01"] {
6234            let path = configured.as_path().join(stray);
6235            assert_eq!(
6236                verify_journalled_slot(&path, nz(1), None)
6237                    .unwrap_err()
6238                    .refusal,
6239                SlotRefusal::NotTheJournalledSlot,
6240                "{}",
6241                path.display()
6242            );
6243        }
6244
6245        // A surviving policy that names a different root does not get to have
6246        // its disagreement resolved by deleting something.
6247        let elsewhere = tempfile::tempdir().unwrap();
6248        let other =
6249            LocalAbsolutePath::new(elsewhere.path().to_str().unwrap()).expect("a second root");
6250        assert_eq!(
6251            verify_journalled_slot(&slot, nz(1), Some(&other))
6252                .unwrap_err()
6253                .refusal,
6254            SlotRefusal::PolicyRootDisagrees
6255        );
6256    }
6257
6258    #[cfg(unix)]
6259    #[test]
6260    fn a_substituted_work_directory_quarantines_the_slot_and_deletes_nothing_outside_it() {
6261        // Windows needs a privilege to create a junction or a symlink, so the
6262        // substitution is made here; the rule is platform-independent because it
6263        // is `symlink_metadata`'s answer plus the reparse attribute.
6264        let root = tempfile::tempdir().unwrap();
6265        let outside = root.path().join("operator-data");
6266        fs::create_dir(&outside).unwrap();
6267        let sentinel = outside.join("do-not-delete.txt");
6268        fs::write(
6269            &sentinel,
6270            b"an operator's data, outside every approved root",
6271        )
6272        .unwrap();
6273
6274        let slot = root.path().join("s1");
6275        fs::create_dir(&slot).unwrap();
6276        fs::create_dir(slot.join("bin")).unwrap();
6277        std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6278
6279        let quarantine = scrub_slot_entries(&slot).unwrap_err();
6280        assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
6281        assert!(
6282            sentinel.exists(),
6283            "the deletion followed the link out of the slot"
6284        );
6285        assert!(outside.is_dir());
6286        assert!(
6287            slot.join(DEFAULT_WORK_FOLDER).symlink_metadata().is_ok(),
6288            "the substituted link is left for the operator, never unlinked as if it were ours"
6289        );
6290
6291        // A `_work` that is a plain file is the same refusal for the same
6292        // reason: it is not a job workspace, and this is not the code that
6293        // decides what to do about it.
6294        let file_work = root.path().join("s2");
6295        fs::create_dir(&file_work).unwrap();
6296        fs::write(file_work.join(DEFAULT_WORK_FOLDER), b"not a directory").unwrap();
6297        assert_eq!(
6298            scrub_slot_entries(&file_work).unwrap_err().refusal,
6299            SlotRefusal::WorkNotADirectory
6300        );
6301    }
6302
6303    #[cfg(unix)]
6304    #[test]
6305    fn a_slot_replaced_by_a_link_out_of_its_root_is_refused_before_anything_is_read() {
6306        let root = tempfile::tempdir().unwrap();
6307        let outside = root.path().join("operator-data");
6308        fs::create_dir(&outside).unwrap();
6309        let sentinel = outside.join("do-not-delete.txt");
6310        fs::write(
6311            &sentinel,
6312            b"an operator's data, outside every approved root",
6313        )
6314        .unwrap();
6315
6316        // The lexical half of containment passes -- the name is right and the
6317        // parent is right -- and canonical resolution is what catches it.
6318        let inside = root.path().join("inside");
6319        fs::create_dir(&inside).unwrap();
6320        let slot = inside.join("s1");
6321        std::os::unix::fs::symlink(&outside, &slot).unwrap();
6322
6323        assert_eq!(
6324            verify_journalled_slot(&slot, nz(1), None)
6325                .unwrap_err()
6326                .refusal,
6327            SlotRefusal::Containment
6328        );
6329        assert!(sentinel.exists());
6330        assert!(
6331            slot.symlink_metadata().is_ok(),
6332            "the link is left for the operator rather than removed as if it were ours"
6333        );
6334    }
6335
6336    /// The Windows half of the case above.
6337    ///
6338    /// Containment was proven only on Unix, and a symbolic link is the wrong
6339    /// instrument to prove it with on Windows: creating one needs a privilege an
6340    /// ordinary workflow does not have, so it is not the substitution an
6341    /// attacker would reach for. A **junction** needs none, which makes it both
6342    /// the realistic attack and the one this repository must refuse -- and
6343    /// `04-security-recovery.md` names it in the same breath as the symlink for
6344    /// exactly that reason.
6345    #[cfg(windows)]
6346    #[test]
6347    fn a_slot_root_replaced_by_a_junction_is_refused_before_anything_is_read() {
6348        let root = tempfile::tempdir().unwrap();
6349        let outside = root.path().join("operator-data");
6350        fs::create_dir(&outside).unwrap();
6351        let sentinel = outside.join("do-not-delete.txt");
6352        fs::write(
6353            &sentinel,
6354            b"an operator's data, outside every approved root",
6355        )
6356        .unwrap();
6357
6358        // The lexical half of containment passes -- `s1` under the root the
6359        // journal names -- and canonical resolution is what catches it.
6360        let inside = root.path().join("inside");
6361        fs::create_dir(&inside).unwrap();
6362        let slot = inside.join("s1");
6363        let Some(()) = plant_junction(&slot, &outside) else {
6364            eprintln!("skipped: this machine would not create a directory junction");
6365            return;
6366        };
6367
6368        assert_eq!(
6369            verify_journalled_slot(&slot, nz(1), None)
6370                .unwrap_err()
6371                .refusal,
6372            SlotRefusal::Containment
6373        );
6374        assert!(
6375            sentinel.exists(),
6376            "the refusal resolved the junction and reached the operator's data"
6377        );
6378        assert!(
6379            slot.symlink_metadata().is_ok(),
6380            "the junction is left for the operator rather than removed as if it were ours"
6381        );
6382    }
6383
6384    /// Plant a directory junction at `link` pointing at `target`.
6385    ///
6386    /// A junction is the Windows substitution this has to refuse, and unlike a
6387    /// symbolic link it needs no privilege — which is exactly why it is the one
6388    /// an unprivileged workflow would reach for. `mklink` is a `cmd` builtin, so
6389    /// there is no binary to find and nothing to install; `None` means this
6390    /// machine would not make one and the caller says so rather than asserting
6391    /// nothing.
6392    #[cfg(windows)]
6393    fn plant_junction(link: &Path, target: &Path) -> Option<()> {
6394        let made = std::process::Command::new("cmd")
6395            .arg("/C")
6396            .arg("mklink")
6397            .arg("/J")
6398            .arg(link)
6399            .arg(target)
6400            .output()
6401            .ok()?;
6402        (made.status.success() && link.symlink_metadata().is_ok()).then_some(())
6403    }
6404
6405    #[cfg(windows)]
6406    #[test]
6407    fn a_work_directory_replaced_by_a_junction_fails_closed_and_deletes_nothing_beyond_it() {
6408        let root = tempfile::tempdir().unwrap();
6409        let outside = root.path().join("operator-data");
6410        fs::create_dir(&outside).unwrap();
6411        let sentinel = outside.join("do-not-delete.txt");
6412        fs::write(
6413            &sentinel,
6414            b"an operator's data, outside every approved root",
6415        )
6416        .unwrap();
6417
6418        let slot = root.path().join("s1");
6419        fs::create_dir(&slot).unwrap();
6420        fs::create_dir(slot.join("bin")).unwrap();
6421        let Some(()) = plant_junction(&slot.join(DEFAULT_WORK_FOLDER), &outside) else {
6422            eprintln!("skipped: this machine would not create a directory junction");
6423            return;
6424        };
6425
6426        // The reparse point is what `is_link_like` answers on, so a junction is
6427        // refused for the same reason a symbolic link is and neither is
6428        // descended into.
6429        let work = fs::symlink_metadata(slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6430        assert!(is_link_like(&work), "a junction is a reparse point");
6431        let quarantine = scrub_slot_entries(&slot).unwrap_err();
6432        assert_eq!(quarantine.refusal, SlotRefusal::WorkNotADirectory);
6433        assert!(
6434            sentinel.exists(),
6435            "the deletion followed the junction out of the slot"
6436        );
6437        assert!(outside.is_dir());
6438
6439        // A junction standing where an ordinary entry was is unlinked rather
6440        // than followed, so the removal still cannot reach through it.
6441        let elsewhere = root.path().join("s2");
6442        fs::create_dir(&elsewhere).unwrap();
6443        fs::create_dir(elsewhere.join(DEFAULT_WORK_FOLDER)).unwrap();
6444        if plant_junction(&elsewhere.join("externals"), &outside).is_some() {
6445            scrub_slot_entries(&elsewhere).expect("an ordinary entry is removed, junction or not");
6446            verify_slot_scrubbed(&elsewhere).expect("and the slot verifies");
6447            assert!(sentinel.exists(), "the junction was followed, not unlinked");
6448            assert_eq!(entries_of(&elsewhere), only_the_job_workspace());
6449        }
6450    }
6451
6452    #[cfg(unix)]
6453    #[tokio::test]
6454    async fn a_substituted_work_directory_leaves_the_attempt_uncleaned_and_still_leased() {
6455        let harness = Harness::new(FakeGithubLifecycle::default(), Arc::new(PersistentDemand))
6456            .with_persistent_workspace(2);
6457        harness.ready().await;
6458        let first = harness.launch().await;
6459        let slot = harness.slot_path(1);
6460        harness.conclude(first.id);
6461
6462        let outside = harness._root.path().join("operator-data");
6463        fs::create_dir_all(&outside).unwrap();
6464        let sentinel = outside.join("do-not-delete.txt");
6465        fs::write(&sentinel, b"outside every approved root").unwrap();
6466        std::os::unix::fs::symlink(&outside, slot.join(DEFAULT_WORK_FOLDER)).unwrap();
6467
6468        harness
6469            .launcher
6470            .clean(first.id)
6471            .await
6472            .expect_err("a slot whose `_work` was substituted is quarantined");
6473        assert!(sentinel.exists());
6474
6475        let held = harness.attempt(first.id);
6476        assert_eq!(held.state(), AttemptState::Failed);
6477        assert!(held.holds_slot_lease());
6478
6479        // The quarantined slot is not silently chosen again.
6480        let second = harness.launch().await;
6481        assert_eq!(second.workspace(), AttemptWorkspace::persistent_slot(nz(2)));
6482    }
6483
6484    #[test]
6485    fn a_slot_that_is_a_file_is_refused_and_a_slot_that_is_gone_is_not() {
6486        let root = tempfile::tempdir().unwrap();
6487        let occupied = root.path().join("s1");
6488        fs::write(&occupied, b"an operator's file").unwrap();
6489        // The journal check passes -- it is the right name under the right root
6490        // -- and the shape check is what refuses.
6491        verify_journalled_slot(&occupied, nz(1), None).expect("the path is the journalled slot");
6492        assert_eq!(
6493            slot_is_present(&occupied).unwrap_err().refusal,
6494            SlotRefusal::SlotNotADirectory
6495        );
6496        assert_eq!(fs::read_to_string(&occupied).unwrap(), "an operator's file");
6497
6498        // A directory that is simply not there leaves nothing to remove and
6499        // nothing to prove absent, so it is not a refusal.
6500        assert!(!slot_is_present(&root.path().join("s2")).unwrap());
6501        let present = root.path().join("s3");
6502        fs::create_dir(&present).unwrap();
6503        assert!(slot_is_present(&present).unwrap());
6504    }
6505
6506    #[test]
6507    fn cleanup_dispatches_on_the_journalled_kind_and_not_on_what_the_directory_holds() {
6508        let root = tempfile::tempdir().unwrap();
6509
6510        // A disposable directory that happens to contain a `_work` still goes
6511        // whole: the workspace kind is immutable so that the shape of a
6512        // directory a workflow can write to cannot choose its own algorithm.
6513        let disposable = root.path().join("abcdef012345");
6514        fs::create_dir_all(disposable.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
6515        let ephemeral = RunnerAttempt::allocate(
6516            AttemptId::new_random(),
6517            fixtures::POLICY_ID,
6518            &disposable,
6519            fixtures::created_at(),
6520        );
6521        remove_materialized_package(&ephemeral).unwrap();
6522        assert!(!disposable.exists());
6523
6524        // And a slot keeps its `_work` with the same contents beneath it.
6525        let slot = root.path().join("s1");
6526        fs::create_dir_all(slot.join(DEFAULT_WORK_FOLDER).join("repo")).unwrap();
6527        fs::create_dir_all(slot.join("bin")).unwrap();
6528        let persistent = RunnerAttempt::allocate_in(
6529            AttemptId::new_random(),
6530            fixtures::POLICY_ID,
6531            &slot,
6532            AttemptWorkspace::persistent_slot(nz(1)),
6533            fixtures::created_at(),
6534        );
6535        remove_materialized_package(&persistent).unwrap();
6536        assert_eq!(entries_of(&slot), only_the_job_workspace());
6537        assert!(slot.join(DEFAULT_WORK_FOLDER).join("repo").is_dir());
6538    }
6539
6540    #[test]
6541    fn every_slot_refusal_names_a_distinct_event_class_and_keeps_the_lease() {
6542        let refusals = [
6543            SlotRefusal::NotTheJournalledSlot,
6544            SlotRefusal::PolicyRootDisagrees,
6545            SlotRefusal::Containment,
6546            SlotRefusal::SlotNotADirectory,
6547            SlotRefusal::Enumeration,
6548            SlotRefusal::WorkNotADirectory,
6549            SlotRefusal::Deletion,
6550            SlotRefusal::Residue,
6551        ];
6552        let classes: BTreeSet<&str> = refusals.iter().map(|refusal| refusal.class()).collect();
6553        assert_eq!(
6554            classes.len(),
6555            refusals.len(),
6556            "an event class shared by two refusals tells an operator less than it appears to"
6557        );
6558        for refusal in refusals {
6559            // The event field is a closed vocabulary, so it has to look like
6560            // one: `d1`'s sink allows the name verbatim.
6561            assert!(
6562                refusal
6563                    .class()
6564                    .chars()
6565                    .all(|c| c.is_ascii_lowercase() || c == '_'),
6566                "{}",
6567                refusal.class()
6568            );
6569            assert!(
6570                refusal.remediation().contains("slot lease"),
6571                "every refusal has to say the lease is still held: {}",
6572                refusal.class()
6573            );
6574        }
6575    }
6576
6577    #[test]
6578    fn copy_package_tree_copies_files_and_preserves_paths_with_spaces() {
6579        let root = tempfile::tempdir().unwrap();
6580        let source = root.path().join("source with spaces");
6581        let dest = root.path().join("dest with spaces");
6582
6583        fs::create_dir_all(&source).unwrap();
6584        fs::write(source.join("file1.txt"), b"hello").unwrap();
6585
6586        let nested = source.join("nested dir");
6587        fs::create_dir_all(&nested).unwrap();
6588        fs::write(nested.join("file2.txt"), b"world").unwrap();
6589
6590        // This is not a top-level `_work`, so it should be allowed
6591        let nested_work = nested.join(DEFAULT_WORK_FOLDER);
6592        fs::create_dir_all(&nested_work).unwrap();
6593        fs::write(nested_work.join("allowed.txt"), b"allowed").unwrap();
6594
6595        copy_package_tree(&source, &dest).unwrap();
6596
6597        assert_eq!(fs::read_to_string(dest.join("file1.txt")).unwrap(), "hello");
6598        assert_eq!(
6599            fs::read_to_string(dest.join("nested dir").join("file2.txt")).unwrap(),
6600            "world"
6601        );
6602        assert_eq!(
6603            fs::read_to_string(
6604                dest.join("nested dir")
6605                    .join(DEFAULT_WORK_FOLDER)
6606                    .join("allowed.txt")
6607            )
6608            .unwrap(),
6609            "allowed"
6610        );
6611    }
6612
6613    #[test]
6614    fn copy_package_tree_refuses_top_level_work_folder() {
6615        let root = tempfile::tempdir().unwrap();
6616        let source = root.path().join("source");
6617        let dest = root.path().join("dest");
6618
6619        fs::create_dir_all(&source).unwrap();
6620        fs::write(source.join("file1.txt"), b"hello").unwrap();
6621
6622        // Top-level `_work` should be refused
6623        let top_work = source.join(DEFAULT_WORK_FOLDER);
6624        fs::create_dir_all(&top_work).unwrap();
6625
6626        let err = copy_package_tree(&source, &dest).unwrap_err();
6627        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
6628    }
6629}