runner_manager_domain/attempt.rs
1// owner: b1-domain-core
2
3//! One runner attempt: its lifecycle, its outcome, who may act on it, and what
4//! to do about it after a restart.
5//!
6//! Two things here are easy to get subtly wrong, so they are stated before the
7//! code:
8//!
9//! **The idle exit is a normal outcome, not a failure.** Because there is no
10//! `AcquireJobs` equivalent on the REST path, nothing reserves a queued job for
11//! this host. A runner may start, find that another host took the work, and exit
12//! on its idle timeout (`03-control-flows.md`, flow 2.7). That attempt is
13//! terminal and is cleaned like any other, but [`AttemptOutcome`] records *which*
14//! terminal thing happened, because `g2` must render it distinctly from a
15//! failure and can only do so if the domain wrote the distinction down. Showing a
16//! normal surplus exit as an error sends an operator hunting a fault that does
17//! not exist.
18//!
19//! **A `busy` attempt is never cleaned to free capacity.** Scale-down reclaims a
20//! slot when an attempt reaches a terminal state and at no other time
21//! (`04-subsystem-contracts.md`: "`busy` cannot transition to cleanup due to a
22//! scale-down request"). [`RunnerAttempt::clean`] refuses it by name rather than
23//! by a generic transition error, so the refusal is legible in a log.
24
25use std::fmt;
26use std::path::{Path, PathBuf};
27
28use serde::{Deserialize, Serialize};
29
30use crate::execution::{AttemptExecution, ExecutionError};
31use crate::model::{AttemptId, Clock, Elapsed, HostId, PolicyId, Timestamp};
32use crate::policy::ScalePolicy;
33use crate::workspace::{AttemptWorkspace, WorkspaceError, WorkspaceKind};
34
35// ---------------------------------------------------------------------------
36// Errors
37// ---------------------------------------------------------------------------
38
39#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
40pub enum AttemptError {
41 #[error(transparent)]
42 Execution(#[from] ExecutionError),
43 #[error("{to} is not a legal transition from {from}")]
44 IllegalTransition {
45 from: AttemptState,
46 to: AttemptState,
47 },
48
49 /// A stored workspace kind and slot that cannot describe one allocation, so
50 /// the legal cleanup algorithm for the journalled path is undecidable.
51 #[error(transparent)]
52 Workspace(#[from] WorkspaceError),
53
54 #[error(
55 "a busy attempt must not be cleaned; capacity is reclaimed only when an \
56 attempt reaches a terminal state, never by stopping a runner that is \
57 executing a job"
58 )]
59 BusyCannotBeCleaned,
60
61 #[error("the outcome {outcome:?} cannot be reached from {from}")]
62 OutcomeUnreachable {
63 from: AttemptState,
64 outcome: AttemptOutcome,
65 },
66
67 #[error("{state} is a terminal state and requires an outcome, but none was recorded")]
68 TerminalWithoutOutcome { state: AttemptState },
69
70 #[error("{state} is not terminal, so it must not carry the outcome {outcome:?}")]
71 NonTerminalWithOutcome {
72 state: AttemptState,
73 outcome: AttemptOutcome,
74 },
75
76 #[error("the recorded outcome {outcome:?} does not belong to the recorded state {state}")]
77 OutcomeStateMismatch {
78 state: AttemptState,
79 outcome: AttemptOutcome,
80 },
81
82 #[error("{state} is a terminal state and requires a terminal_at, but none was recorded")]
83 TerminalWithoutTimestamp { state: AttemptState },
84
85 #[error("{state} is not terminal, so it must not carry a terminal_at")]
86 NonTerminalWithTimestamp { state: AttemptState },
87
88 #[error(
89 "{state} carries {field} at {found}, which is before its created_at of \
90 {created_at}; an attempt cannot have changed state or concluded before \
91 it was allocated"
92 )]
93 TimestampsOutOfOrder {
94 state: AttemptState,
95 /// Which of the two orderable timestamps is out of order.
96 field: &'static str,
97 created_at: Timestamp,
98 found: Timestamp,
99 },
100}
101
102/// Ownership rule 2: "A host agent may act only on attempts persisted under its
103/// `host_id`."
104#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
105pub enum OwnershipError {
106 #[error(
107 "attempt {attempt} belongs to policy {attempt_policy}, but was checked \
108 against policy {policy}"
109 )]
110 PolicyMismatch {
111 attempt: AttemptId,
112 attempt_policy: PolicyId,
113 policy: PolicyId,
114 },
115
116 #[error(
117 "policy {policy} belongs to host {owner}; this agent runs on host \
118 {agent} and must not act on its attempts"
119 )]
120 ForeignHost {
121 policy: PolicyId,
122 owner: HostId,
123 agent: HostId,
124 },
125}
126
127// ---------------------------------------------------------------------------
128// AttemptState
129// ---------------------------------------------------------------------------
130
131/// The runner-attempt lifecycle, exactly as `04-subsystem-contracts.md` draws
132/// it after its 2026-08-21 amendment:
133///
134/// ```text
135/// allocated -> jit_received -> starting -> idle | busy
136/// idle -> busy
137/// allocated | jit_received | starting -> failed | orphaned
138/// idle | busy -> finished | failed | orphaned
139/// finished | failed | orphaned -> cleaned
140/// ```
141///
142/// `idle` means the runner process is registered and awaiting its single job
143/// assignment. It is short-lived and is **not** an idle persistent runner — this
144/// product has none, by D7.
145///
146/// **Every transition outside that diagram is rejected**, which `b1`'s
147/// Definition of Done requires. `cleaned` is absorbing, which is correct and
148/// intended.
149///
150/// **The last two edge sets were added by amendment, and the reasons matter to
151/// anyone reading this state machine.** `b1` first implemented the original
152/// diagram faithfully and surfaced its gaps as an explicit
153/// `NoLegalTransition` decision rather than inventing edges; the amendment
154/// closed them at the design level, and that decision value is gone with them:
155///
156/// * **`idle -> busy`.** `e3`'s Scope step 4 moves an attempt through
157/// `jit_received`, `starting`, `idle`, `busy` *in sequence*, and the
158/// definition of `idle` above describes a state that by construction precedes
159/// a job. Without this edge a runner observed idle and then assigned a job had
160/// nowhere legal to go.
161/// * **Terminal edges out of the three pre-registration states.** An attempt
162/// counts against host capacity for exactly as long as it is non-terminal
163/// ([`Self::counts_against_capacity`]), so an attempt that could not reach a
164/// terminal state **held a host capacity slot permanently**: two failed JIT
165/// requests on a `host_capacity: 2` host wedged that host into starting zero
166/// runners, with no error state and no cleanup path. `orphaned` is included
167/// for the restart case, where a pre-registration attempt is found after the
168/// agent restarts.
169///
170/// Those edges are also what make seven of the nine [`FailureReason`] variants
171/// reachable at all — `JitRequestFailed`, `JitExpired`,
172/// `RunnerPackageUnverified`, `RunnerVersionRejected`, `ProcessStartFailed`,
173/// `RegistrationTimedOut` and `TerminatedAfterRegistrationTimeout` each occur at
174/// a pre-registration state. `03-control-flows.md` flow 2 names the first five
175/// as conditions the agent must record; the last two are named by no document
176/// and are this crate's own, added because [`recovery_decision`] needs to say
177/// "alive, past its deadline, still unregistered" without calling it a crash,
178/// and because `e3` then needs to say what became of that runner without
179/// calling a process it stopped itself either a crash or still running.
180#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
181#[serde(rename_all = "snake_case")]
182pub enum AttemptState {
183 Allocated,
184 Preparing,
185 Prepared,
186 JitReceived,
187 Starting,
188 Idle,
189 Busy,
190 Finished,
191 Failed,
192 Orphaned,
193 Destroying,
194 CleanupDeferred,
195 Cleaned,
196}
197
198impl AttemptState {
199 pub const ALL: [AttemptState; 13] = [
200 AttemptState::Allocated,
201 AttemptState::Preparing,
202 AttemptState::Prepared,
203 AttemptState::JitReceived,
204 AttemptState::Starting,
205 AttemptState::Idle,
206 AttemptState::Busy,
207 AttemptState::Finished,
208 AttemptState::Failed,
209 AttemptState::Orphaned,
210 AttemptState::Destroying,
211 AttemptState::CleanupDeferred,
212 AttemptState::Cleaned,
213 ];
214
215 /// The complete legal transition list. A self-transition is not in it.
216 pub const LEGAL: &'static [(AttemptState, AttemptState)] = &[
217 // `allocated -> jit_received -> starting -> idle | busy`.
218 (AttemptState::Allocated, AttemptState::JitReceived),
219 (AttemptState::Allocated, AttemptState::Preparing),
220 (AttemptState::Preparing, AttemptState::Prepared),
221 (AttemptState::Prepared, AttemptState::JitReceived),
222 (AttemptState::JitReceived, AttemptState::Starting),
223 (AttemptState::Starting, AttemptState::Idle),
224 (AttemptState::Starting, AttemptState::Busy),
225 // `idle -> busy`.
226 (AttemptState::Idle, AttemptState::Busy),
227 // `busy -> idle`: GitHub showed the runner assigned, and the runner
228 // never started the job. See [`RunnerAttempt::assignment_not_taken`].
229 (AttemptState::Busy, AttemptState::Idle),
230 // `allocated | jit_received | starting -> failed | orphaned`.
231 (AttemptState::Allocated, AttemptState::Failed),
232 (AttemptState::Allocated, AttemptState::Orphaned),
233 (AttemptState::Preparing, AttemptState::Failed),
234 (AttemptState::Preparing, AttemptState::Orphaned),
235 (AttemptState::Prepared, AttemptState::Failed),
236 (AttemptState::Prepared, AttemptState::Orphaned),
237 (AttemptState::JitReceived, AttemptState::Failed),
238 (AttemptState::JitReceived, AttemptState::Orphaned),
239 (AttemptState::Starting, AttemptState::Failed),
240 (AttemptState::Starting, AttemptState::Orphaned),
241 // `idle | busy -> finished | failed | orphaned`.
242 (AttemptState::Idle, AttemptState::Finished),
243 (AttemptState::Idle, AttemptState::Failed),
244 (AttemptState::Idle, AttemptState::Orphaned),
245 (AttemptState::Busy, AttemptState::Finished),
246 (AttemptState::Busy, AttemptState::Failed),
247 (AttemptState::Busy, AttemptState::Orphaned),
248 // `finished | failed | orphaned -> cleaned`.
249 (AttemptState::Finished, AttemptState::Cleaned),
250 (AttemptState::Failed, AttemptState::Cleaned),
251 (AttemptState::Orphaned, AttemptState::Cleaned),
252 (AttemptState::Finished, AttemptState::Destroying),
253 (AttemptState::Failed, AttemptState::Destroying),
254 (AttemptState::Orphaned, AttemptState::Destroying),
255 (AttemptState::Destroying, AttemptState::CleanupDeferred),
256 (AttemptState::CleanupDeferred, AttemptState::Destroying),
257 (AttemptState::Destroying, AttemptState::Cleaned),
258 ];
259
260 /// The five states an attempt can still be concluded from: every
261 /// non-terminal state. After the amendment this is exactly
262 /// "not [`Self::is_terminal`]", which is what closed the permanently-held
263 /// capacity slot — but it is written out rather than derived, because the
264 /// two are only equal while the diagram gives every live state a terminal
265 /// edge, and that equality is a property worth failing loudly on.
266 pub const CONCLUDABLE_FROM: &'static [AttemptState] = &[
267 AttemptState::Allocated,
268 AttemptState::Preparing,
269 AttemptState::Prepared,
270 AttemptState::JitReceived,
271 AttemptState::Starting,
272 AttemptState::Idle,
273 AttemptState::Busy,
274 ];
275
276 #[must_use]
277 pub fn can_transition_to(self, next: AttemptState) -> bool {
278 Self::LEGAL.contains(&(self, next))
279 }
280
281 /// Terminal states, at which capacity is reclaimed.
282 ///
283 /// `e1`: "capacity is reclaimed only when an attempt reaches a terminal
284 /// state." `cleaned` is included because it follows a terminal state.
285 #[must_use]
286 pub const fn is_terminal(self) -> bool {
287 matches!(
288 self,
289 AttemptState::Finished
290 | AttemptState::Failed
291 | AttemptState::Orphaned
292 | AttemptState::Destroying
293 | AttemptState::CleanupDeferred
294 | AttemptState::Cleaned
295 )
296 }
297
298 /// Whether this attempt still occupies one of the host's capacity slots.
299 ///
300 /// This is the term the reconciliation formula subtracts, and getting it
301 /// wrong is silent: counting a terminal attempt starves the host, and failing
302 /// to count a `starting` one starts a second runner for a job already being
303 /// served.
304 #[must_use]
305 pub const fn counts_against_capacity(self) -> bool {
306 !self.is_terminal()
307 }
308
309 /// The three terminal states that require an outcome before `cleaned`.
310 #[must_use]
311 pub const fn is_concluded(self) -> bool {
312 matches!(
313 self,
314 AttemptState::Finished | AttemptState::Failed | AttemptState::Orphaned
315 )
316 }
317}
318
319impl fmt::Display for AttemptState {
320 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
321 f.write_str(match self {
322 AttemptState::Allocated => "allocated",
323 AttemptState::Preparing => "preparing",
324 AttemptState::Prepared => "prepared",
325 AttemptState::JitReceived => "jit_received",
326 AttemptState::Starting => "starting",
327 AttemptState::Idle => "idle",
328 AttemptState::Busy => "busy",
329 AttemptState::Finished => "finished",
330 AttemptState::Failed => "failed",
331 AttemptState::Orphaned => "orphaned",
332 AttemptState::Destroying => "destroying",
333 AttemptState::CleanupDeferred => "cleanup_deferred",
334 AttemptState::Cleaned => "cleaned",
335 })
336 }
337}
338
339// ---------------------------------------------------------------------------
340// Outcome
341// ---------------------------------------------------------------------------
342
343/// Why an attempt failed.
344///
345/// `Other` exists so `e3` is not blocked by a reason this task did not
346/// anticipate; `crates/domain/src/attempt.rs` belongs to `b1` and `e3` cannot
347/// extend this enum itself.
348///
349/// **Nothing that reaches this type may contain a credential.** It is written to
350/// the journal by `b2` and rendered by `g2`, and `07-security.md`'s log scan runs
351/// over both. A JIT blob, an `Authorization` header, or a token in an `Other`
352/// string would defeat that gate from inside the domain, where the redacting log
353/// sink (`d1`) never gets a chance to see it.
354#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
355#[serde(rename_all = "snake_case")]
356pub enum FailureReason {
357 /// `generate-jitconfig` did not return a configuration.
358 JitRequestFailed,
359 /// The configuration was never claimed and expired (flow 4.4).
360 JitExpired,
361 /// The runner package's published checksum was absent or did not match
362 /// (`05-infrastructure.md`: the agent fails closed).
363 RunnerPackageUnverified,
364 /// GitHub rejects runners more than 30 days behind the latest release
365 /// (`01-current-architecture.md`, edge case 7). Terminal and
366 /// operator-actionable, never retried.
367 RunnerVersionRejected,
368 /// The child process could not be spawned.
369 ProcessStartFailed,
370 /// The child process exited before it could do its one job.
371 ///
372 /// **Only for a process that is actually gone.** A runner still running past
373 /// its startup deadline is [`Self::RegistrationTimedOut`], not this: `g2`
374 /// renders these strings to an operator, and telling one that a process
375 /// "exited unexpectedly" while it is visible in Task Manager spends the
376 /// credibility of every other message this product prints.
377 ProcessExitedUnexpectedly,
378 /// The runner process is up but never registered with GitHub inside its
379 /// startup window.
380 ///
381 /// Split from [`Self::ProcessExitedUnexpectedly`] because it is accurate and
382 /// because it points an operator somewhere else entirely. A process that
383 /// exited is a crash to investigate — logs, exit code, a corrupt runner
384 /// package. A process that is alive and unregistered has almost always
385 /// failed to *reach* GitHub: a proxy, a firewall, a DNS answer, an expired
386 /// or wrong-scoped configuration. Those are configuration and networking
387 /// fixes, and an operator sent to the wrong one of the two loses the time
388 /// this distinction exists to save.
389 RegistrationTimedOut,
390 /// The agent stopped a runner process that had not registered inside its
391 /// startup window.
392 ///
393 /// **The dead-process counterpart of [`Self::RegistrationTimedOut`], and it
394 /// exists because that one cannot be reused here.** By the time this reason
395 /// is recorded the process is gone — the agent signalled it — so rendering
396 /// "the runner process is running but did not register" would tell an
397 /// operator a process is up that they can see is not. That is the same false
398 /// liveness claim [`Self::ProcessExitedUnexpectedly`]'s documentation says
399 /// spends the credibility of every other message this product prints, and
400 /// `tests::no_decision_calls_a_dead_process_live` is what holds the line.
401 ///
402 /// **It points where [`Self::RegistrationTimedOut`] points, not where
403 /// [`Self::ProcessExitedUnexpectedly`] points.** The runner never reached
404 /// GitHub — a proxy, a firewall, a DNS answer, an expired or wrong-scoped
405 /// configuration — and the exit is the agent's own doing rather than
406 /// evidence of a crash. An operator sent to logs and exit codes for this is
407 /// investigating the wrong machine.
408 ///
409 /// **Who records it.** `e3`, and only `e3`. [`recovery_decision`] cannot
410 /// derive it: a process this agent killed and a process that crashed on its
411 /// own present the *same* [`RecoveryObservation`], so the distinguishing
412 /// fact has to be journalled as terminate-intent before the signal is sent
413 /// and read back afterwards. See [`RecoveryDecision::Terminate`] for the
414 /// window that obligation closes.
415 TerminatedAfterRegistrationTimeout,
416 /// A closed, credential-free isolation provider failure. Its category is
417 /// preserved in the attempt journal and operator activity rather than
418 /// flattened into free-form subprocess or adapter text.
419 IsolationProvider(IsolationProviderFailure),
420 /// An ephemeral attempt's workspace could not be removed yet, typically
421 /// because a late child still holds a file open or a directory refuses to
422 /// be listed; the reconciler retries it with backoff.
423 ///
424 /// **A cleanup report, never an attempt outcome.** It names why the
425 /// reconciler's cleanup of an already concluded attempt failed, so the
426 /// event and the log say *which* failure it was instead of `other`. No code
427 /// path concludes an attempt with it, which also keeps it out of the
428 /// journal that an older binary might read.
429 WorkspaceCleanupDeferred,
430 /// A required host capability is missing (`runner-manager host doctor`),
431 /// so the daemon refused to start a runner at all.
432 ///
433 /// **A launch refusal, never an attempt outcome**, for the reason
434 /// [`Self::WorkspaceCleanupDeferred`] gives: no attempt exists yet when
435 /// the daemon refuses, no runner is registered with GitHub, and the event
436 /// and the log name the refusal instead of `other`.
437 HostUnfit,
438 /// Anything else. Must carry no credential.
439 Other(String),
440}
441
442/// Operator-actionable isolation failures. No variant carries provider output,
443/// image text, JIT content, or another string that could contain a credential.
444#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
445#[serde(rename_all = "snake_case")]
446pub enum IsolationProviderFailure {
447 Unsupported,
448 NotInstalled,
449 PermissionDenied,
450 ImageUnavailableOrIncompatible,
451 DiskQuotaUnavailable,
452 Degraded,
453 RuntimeOperationFailed,
454 OwnershipMismatch,
455 UnsafeRuntimePath,
456 JitHandoffRejected,
457}
458
459impl fmt::Display for IsolationProviderFailure {
460 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
461 f.write_str(match self {
462 Self::Unsupported => "isolation provider unsupported on this host",
463 Self::NotInstalled => "isolation provider not installed",
464 Self::PermissionDenied => "isolation provider permission denied",
465 Self::ImageUnavailableOrIncompatible => "isolated image unavailable or incompatible",
466 Self::DiskQuotaUnavailable => "rootless OCI writable-layer disk quota unavailable",
467 Self::Degraded => "isolation provider degraded",
468 Self::RuntimeOperationFailed => "rootless OCI runtime operation failed",
469 Self::OwnershipMismatch => "OCI environment ownership mismatch",
470 Self::UnsafeRuntimePath => "OCI runtime must be on the Linux filesystem",
471 Self::JitHandoffRejected => "OCI JIT handoff rejected before start",
472 })
473 }
474}
475
476impl FailureReason {
477 /// One value of every variant, the counterpart of [`AttemptState::ALL`].
478 ///
479 /// `Other`'s detail is empty because what a caller enumerates is the
480 /// *variant*; no consumer should read the string out of this constant.
481 ///
482 /// **This list is hand-written, and what keeps it honest is not its own
483 /// length.** A length written as `9` next to nine elements asserts
484 /// nothing — that was the defect in the assertion this constant replaced.
485 /// What catches a new variant is the exhaustive, wildcard-free `match` in
486 /// `tests::earliest_state_producing`, which stops the test target compiling
487 /// the moment one is added and so puts the author in front of this list.
488 ///
489 /// **The residual gap, measured rather than assumed.** An author who adds a
490 /// variant, writes its `Display` arm and its `earliest_state_producing` arm,
491 /// and then adds it to *neither* this constant nor the test's `cases` table,
492 /// gets a green suite with the variant untested. Adding it to exactly one of
493 /// the two fails the length check; adding it to neither does not.
494 ///
495 /// Re-measured when `TerminatedAfterRegistrationTimeout` was added, because
496 /// a gap described once and never re-run is a gap nobody knows still exists.
497 /// With the variant declared and both match arms written but neither list
498 /// touched, `cargo test -p runner-manager-domain` was green — its lib target
499 /// reported `128 passed; 0 failed` — with the ninth variant unreachable and
500 /// untested. Adding it to this constant alone then failed the length check
501 /// with `left: 8 / right: 9`.
502 ///
503 /// **And the compiler never points at this constant.** A `const` array is
504 /// unaffected by a new variant, so nothing here errors. What stops the
505 /// author is `Display::fmt`'s match (`E0004`) and then
506 /// `tests::earliest_state_producing`'s, and neither of those mentions this
507 /// list — which is why a note pointing back here sits at each of those two
508 /// match sites, where the author is actually standing.
509 ///
510 /// **This is closable in stable Rust, and is hand-written anyway.** A local
511 /// `macro_rules!` that declares the enum and emits `ALL` from the same
512 /// variant list needs no dependency and no unstable feature
513 /// (`std::mem::variant_count` is unstable, but a declarative macro is not
514 /// the same thing). It is not used here because every variant of this enum
515 /// carries several paragraphs of its own documentation explaining what an
516 /// operator should do about it, and variants declared inside a macro
517 /// invocation are markedly worse to read and to `rustdoc`. That is a
518 /// legibility trade, deliberately taken — not an impossibility. If the
519 /// documentation ever thins out, the macro is the better answer.
520 pub const ALL: [FailureReason; 12] = [
521 FailureReason::JitRequestFailed,
522 FailureReason::JitExpired,
523 FailureReason::RunnerPackageUnverified,
524 FailureReason::RunnerVersionRejected,
525 FailureReason::ProcessStartFailed,
526 FailureReason::ProcessExitedUnexpectedly,
527 FailureReason::RegistrationTimedOut,
528 FailureReason::TerminatedAfterRegistrationTimeout,
529 FailureReason::IsolationProvider(IsolationProviderFailure::RuntimeOperationFailed),
530 FailureReason::WorkspaceCleanupDeferred,
531 FailureReason::HostUnfit,
532 FailureReason::Other(String::new()),
533 ];
534}
535
536impl fmt::Display for FailureReason {
537 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
538 // Adding a variant? This `E0004` is where the compiler stops you, and it
539 // is the *only* place it does for `FailureReason::ALL`, which is a const
540 // array and errors nowhere. Add the variant to `ALL` and to the `cases`
541 // table in `tests::all_failure_reasons_are_reachable_from_the_state_that
542 // _produces_them` as well, or it ships untested and unreachable.
543 match self {
544 FailureReason::JitRequestFailed => f.write_str("the JIT configuration request failed"),
545 FailureReason::JitExpired => f.write_str("the JIT configuration expired unclaimed"),
546 FailureReason::RunnerPackageUnverified => {
547 f.write_str("the runner package could not be verified")
548 }
549 FailureReason::RunnerVersionRejected => {
550 f.write_str("GitHub rejected the runner version")
551 }
552 FailureReason::ProcessStartFailed => f.write_str("the runner process failed to start"),
553 FailureReason::ProcessExitedUnexpectedly => {
554 f.write_str("the runner process exited unexpectedly")
555 }
556 FailureReason::RegistrationTimedOut => f.write_str(
557 "the runner process is running but did not register with GitHub \
558 before its startup deadline",
559 ),
560 // Past tense, and no claim that anything is still running: by the
561 // time this is recorded the agent has signalled the process and the
562 // operator can see it is gone. `tests::a_terminated_runner_is_never
563 // _described_as_running` pins that.
564 FailureReason::TerminatedAfterRegistrationTimeout => f.write_str(
565 "the agent stopped the runner process after it failed to \
566 register with GitHub before its startup deadline",
567 ),
568 FailureReason::IsolationProvider(category) => category.fmt(f),
569 FailureReason::WorkspaceCleanupDeferred => {
570 f.write_str("the attempt workspace could not be removed yet; cleanup will retry")
571 }
572 FailureReason::HostUnfit => f.write_str(
573 "a required host check fails, so no runner is started; run \
574 `runner-manager host doctor`",
575 ),
576 FailureReason::Other(detail) => write!(f, "{detail}"),
577 }
578 }
579}
580
581/// What terminally happened to an attempt.
582///
583/// The state and the outcome are not two independent fields that a caller has to
584/// keep consistent: [`AttemptOutcome::terminal_state`] derives the state from the
585/// outcome, and [`RunnerAttempt::conclude`] is the only way to set either. A
586/// `failed` attempt whose outcome says it ran a job is therefore not a bug this
587/// code can have.
588#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
589#[serde(tag = "outcome", rename_all = "snake_case")]
590pub enum AttemptOutcome {
591 /// The runner accepted its one job and the job ended. This says the *runner*
592 /// finished, never that the workflow succeeded — GitHub remains the source of
593 /// truth for workflow outcome (`03-control-flows.md`, flow 2, Failure).
594 CompletedJob,
595 /// The surplus case. The runner registered, no job arrived, and it exited on
596 /// its idle timeout (flow 2.7). Normal, bounded, and **not** a failure.
597 ExitedIdleWithoutWork,
598 /// The attempt failed.
599 Failed { reason: FailureReason },
600 /// Supervision was lost: the process is gone and the attempt could not be
601 /// reconciled with GitHub (`e3`, restart recovery).
602 Orphaned,
603}
604
605impl AttemptOutcome {
606 #[must_use]
607 pub fn failed(reason: FailureReason) -> Self {
608 Self::Failed { reason }
609 }
610
611 /// The one terminal state this outcome corresponds to.
612 #[must_use]
613 pub const fn terminal_state(&self) -> AttemptState {
614 match self {
615 AttemptOutcome::CompletedJob | AttemptOutcome::ExitedIdleWithoutWork => {
616 AttemptState::Finished
617 }
618 AttemptOutcome::Failed { .. } => AttemptState::Failed,
619 AttemptOutcome::Orphaned => AttemptState::Orphaned,
620 }
621 }
622
623 /// True for the outcomes an operator should be told to look at.
624 ///
625 /// `g2` renders this differently from [`Self::is_idle_exit`], and `e1`'s
626 /// Definition of Done requires that a surplus attempt "is not reported as a
627 /// failure".
628 #[must_use]
629 pub const fn is_failure(&self) -> bool {
630 matches!(
631 self,
632 AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned
633 )
634 }
635
636 /// True only for the surplus case.
637 #[must_use]
638 pub const fn is_idle_exit(&self) -> bool {
639 matches!(self, AttemptOutcome::ExitedIdleWithoutWork)
640 }
641
642 /// True only when this runner actually took a job.
643 #[must_use]
644 pub const fn ran_a_job(&self) -> bool {
645 matches!(self, AttemptOutcome::CompletedJob)
646 }
647
648 /// The states an attempt must be in for this outcome to be reachable.
649 ///
650 /// Two of the four outcomes are narrower than the diagram's terminal edges,
651 /// and deliberately so: `finished` is reachable from both `idle` and `busy`,
652 /// but only one of them can have produced each of the two outcomes that lead
653 /// there. Failure and orphaning are as wide as the diagram — every live
654 /// state has a `-> failed | orphaned` edge since the amendment, and
655 /// narrowing them per [`FailureReason`] here would invent product rules no
656 /// document states.
657 const fn required_from(&self) -> &'static [AttemptState] {
658 match self {
659 // Only a runner that was assigned a job can have run one.
660 AttemptOutcome::CompletedJob => &[AttemptState::Busy],
661 // Only a runner that was registered and waiting can have exited idle.
662 AttemptOutcome::ExitedIdleWithoutWork => &[AttemptState::Idle],
663 AttemptOutcome::Failed { .. } | AttemptOutcome::Orphaned => {
664 AttemptState::CONCLUDABLE_FROM
665 }
666 }
667 }
668}
669
670impl fmt::Display for AttemptOutcome {
671 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
672 match self {
673 AttemptOutcome::CompletedJob => f.write_str("ran a job"),
674 AttemptOutcome::ExitedIdleWithoutWork => f.write_str("exited idle without work"),
675 AttemptOutcome::Failed { reason } => write!(f, "failed: {reason}"),
676 AttemptOutcome::Orphaned => f.write_str("orphaned"),
677 }
678 }
679}
680
681// ---------------------------------------------------------------------------
682// RunnerAttempt
683// ---------------------------------------------------------------------------
684
685/// One ephemeral runner, from directory allocation to cleanup.
686///
687/// The fields `04-subsystem-contracts.md` names are all here. Two are not in that
688/// list and are added deliberately:
689///
690/// * `outcome`, because `b1`'s Scope requires the attempt to "carry an outcome
691/// distinguishing 'ran a job' from 'exited idle without work'".
692/// * `last_state_change_at`, because recovery decisions measure elapsed time in
693/// the *current* state — an idle timeout runs from entering `idle`, not from
694/// `created_at` — and `b1`'s Definition of Done requires those decisions to be
695/// testable against a fake clock.
696///
697/// `b2` persists both.
698#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
699pub struct RunnerAttempt {
700 pub id: AttemptId,
701 pub policy_id: PolicyId,
702 github_runner_id: Option<u64>,
703 state: AttemptState,
704 outcome: Option<AttemptOutcome>,
705 process_id: Option<u32>,
706 execution: AttemptExecution,
707 runtime_path: PathBuf,
708 /// Which cleanup algorithm this attempt's directory is entitled to, and the
709 /// slot it leases if any.
710 ///
711 /// Immutable after allocation, by `02-target-architecture.md`: "The
712 /// workspace kind and slot number tell recovery which cleanup algorithm is
713 /// legal. Neither may change after allocation." Private with no setter is
714 /// how that is enforced — a mutable kind would let a running attempt convert
715 /// a disposable directory into a retained one, which is exactly the
716 /// two-job contamination path `04-security-recovery.md` measures.
717 workspace: AttemptWorkspace,
718 pub created_at: Timestamp,
719 terminal_at: Option<Timestamp>,
720 last_state_change_at: Timestamp,
721}
722
723/// Every stored column of one attempt, named rather than positional.
724///
725/// **Why this is a struct.** [`RunnerAttempt::from_persisted`] took ten
726/// positional arguments, and two of them — `created_at` and
727/// `last_state_change_at` — are both `Timestamp`. Transposing them type-checked,
728/// compiled, and silently reverted `last_state_change_at` to `created_at`, which
729/// is precisely the bug that field exists to prevent: every recovery timeout
730/// would then have measured from allocation rather than from the current state,
731/// so a long-running `busy` attempt would be read as a stuck one. `terminal_at`
732/// is a third `Option<Timestamp>` in the same list.
733///
734/// `b2` maps database columns onto this type. With a struct that mapping is
735/// checked by name at compile time; positionally it was checked by nothing.
736///
737/// **That guarantee covers the Rust side of the mapping and no more.** It is
738/// the *field* names that the compiler checks, not the column names they are
739/// read from: `PersistedAttempt { created_at: row.get("last_state_change_at")?,
740/// … }` compiles exactly as happily as the correct version, and reintroduces the
741/// very transposition described above. `b2` still owes a test that loads a row
742/// whose columns hold distinguishable values and asserts each landed in the
743/// field of the same name; this type does not supply one.
744///
745/// Construct it with a struct literal so every field is written down at the call
746/// site — that is the whole point, and a builder or a `Default` would give the
747/// omission back.
748#[derive(Debug, Clone, PartialEq, Eq)]
749pub struct PersistedAttempt {
750 pub id: AttemptId,
751 pub policy_id: PolicyId,
752 pub github_runner_id: Option<u64>,
753 pub state: AttemptState,
754 pub outcome: Option<AttemptOutcome>,
755 pub process_id: Option<u32>,
756 pub execution: AttemptExecution,
757 pub runtime_path: PathBuf,
758 /// `ephemeral` or `persistent`, stored beside the slot below.
759 pub workspace_kind: WorkspaceKind,
760 /// The leased slot: `Some` exactly when `workspace_kind` is `persistent`,
761 /// and positive. It is a raw `u16` here rather than a `NonZeroU16` so that
762 /// the `0` a hand-edited row can hold is refused by
763 /// [`AttemptWorkspace::from_persisted`] instead of being unrepresentable at
764 /// the column boundary and panicking somewhere else.
765 pub workspace_slot: Option<u16>,
766 /// When the runtime directory was allocated. Never moves.
767 pub created_at: Timestamp,
768 /// Set when, and only when, the attempt concluded.
769 pub terminal_at: Option<Timestamp>,
770 /// When the attempt entered its **current** state. Recovery timeouts run
771 /// from here, not from `created_at`.
772 pub last_state_change_at: Timestamp,
773}
774
775impl RunnerAttempt {
776 /// Every stored column of this attempt, for `b2` to write back.
777 ///
778 /// The exact inverse of [`Self::from_persisted`], so a round trip through
779 /// the journal is expressible without reaching for a field accessor per
780 /// column and without this type exposing its private fields for writing.
781 #[must_use]
782 pub fn to_persisted(&self) -> PersistedAttempt {
783 PersistedAttempt {
784 id: self.id,
785 policy_id: self.policy_id,
786 github_runner_id: self.github_runner_id,
787 state: self.state,
788 outcome: self.outcome.clone(),
789 process_id: self.process_id,
790 execution: self.execution.clone(),
791 runtime_path: self.runtime_path.clone(),
792 workspace_kind: self.workspace.kind(),
793 workspace_slot: self.workspace.slot_number(),
794 created_at: self.created_at,
795 terminal_at: self.terminal_at,
796 last_state_change_at: self.last_state_change_at,
797 }
798 }
799
800 /// The first step of `e3`'s per-attempt flow: a runtime directory is
801 /// allocated and journalled **before** anything remote happens, so a crash
802 /// leaves a recoverable trace rather than an invisible one.
803 /// D3 keeps this the disposable path: an attempt allocated through it is
804 /// [`AttemptWorkspace::Ephemeral`], so every existing caller and test goes on
805 /// producing the behaviour it produced before persistent slots existed.
806 /// [`Self::allocate_in`] is the one that leases a slot.
807 #[must_use]
808 pub fn allocate(
809 id: AttemptId,
810 policy_id: PolicyId,
811 runtime_path: impl Into<PathBuf>,
812 now: Timestamp,
813 ) -> Self {
814 Self::allocate_in(
815 id,
816 policy_id,
817 runtime_path,
818 AttemptWorkspace::Ephemeral,
819 now,
820 )
821 }
822
823 /// The same first step, recording which workspace the directory came from.
824 ///
825 /// `c2` calls this with [`AttemptWorkspace::PersistentSlot`] while holding
826 /// the host allocation lock, so the slot lease is journalled "before package
827 /// or GitHub effects" and a crash between the two leaves a recoverable trace
828 /// rather than an orphaned slot.
829 #[must_use]
830 pub fn allocate_in(
831 id: AttemptId,
832 policy_id: PolicyId,
833 runtime_path: impl Into<PathBuf>,
834 workspace: AttemptWorkspace,
835 now: Timestamp,
836 ) -> Self {
837 Self {
838 id,
839 policy_id,
840 github_runner_id: None,
841 state: AttemptState::Allocated,
842 outcome: None,
843 process_id: None,
844 execution: AttemptExecution::Native { process_id: None },
845 runtime_path: runtime_path.into(),
846 workspace,
847 created_at: now,
848 terminal_at: None,
849 last_state_change_at: now,
850 }
851 }
852
853 /// Rebuild a journalled attempt.
854 ///
855 /// # Errors
856 /// Any state/outcome/timestamp combination that this crate's own transitions
857 /// cannot produce, so a hand-edited journal cannot inject a `failed` attempt
858 /// that claims to have run a job, or a `finished` one that never reached a
859 /// terminal state.
860 pub fn from_persisted(fields: PersistedAttempt) -> Result<Self, AttemptError> {
861 let PersistedAttempt {
862 id,
863 policy_id,
864 github_runner_id,
865 state,
866 outcome,
867 process_id,
868 execution,
869 runtime_path,
870 workspace_kind,
871 workspace_slot,
872 created_at,
873 terminal_at,
874 last_state_change_at,
875 } = fields;
876
877 // The workspace pair is checked first because it decides which cleanup
878 // algorithm recovery is allowed to run on `runtime_path`. A row that
879 // claims `persistent` with no slot, or `ephemeral` with one, names a
880 // directory whose safe cleanup is undecidable, and
881 // `04-security-recovery.md` requires that to fail closed rather than to
882 // fall back to the destructive branch.
883 let workspace = AttemptWorkspace::from_persisted(workspace_kind, workspace_slot)?;
884 execution.validate()?;
885 match &execution {
886 AttemptExecution::Native {
887 process_id: identity_pid,
888 } if *identity_pid == process_id => {}
889 AttemptExecution::Isolated { .. } if process_id.is_none() => {}
890 _ => return Err(ExecutionError::AttemptIdentityMismatch.into()),
891 }
892
893 match (&outcome, state.is_terminal()) {
894 (None, true) => return Err(AttemptError::TerminalWithoutOutcome { state }),
895 (Some(outcome), false) => {
896 return Err(AttemptError::NonTerminalWithOutcome {
897 state,
898 outcome: outcome.clone(),
899 });
900 }
901 (Some(outcome), true) => {
902 let expected = outcome.terminal_state();
903 let allowed_isolated_cleanup = !execution.is_native()
904 && matches!(
905 state,
906 AttemptState::Destroying | AttemptState::CleanupDeferred
907 );
908 if ![expected, AttemptState::Cleaned].contains(&state) && !allowed_isolated_cleanup
909 {
910 return Err(AttemptError::OutcomeStateMismatch {
911 state,
912 outcome: outcome.clone(),
913 });
914 }
915 }
916 (None, false) => {}
917 }
918
919 // `terminal_at` is validated on exactly the same footing as `outcome`,
920 // and for the same stated reason. `conclude` is the only writer of both
921 // and sets them together, so `state.is_terminal()` and
922 // `terminal_at.is_some()` are equivalent in anything this crate
923 // produced; a row where they disagree was edited by hand. Without this,
924 // a `finished` attempt with no `terminal_at` loaded cleanly and every
925 // consumer of `terminal_at()` -- retention, reporting, `g2`'s ordering
926 // -- silently saw an attempt that had never concluded.
927 match (terminal_at, state.is_terminal()) {
928 (None, true) => return Err(AttemptError::TerminalWithoutTimestamp { state }),
929 (Some(_), false) => return Err(AttemptError::NonTerminalWithTimestamp { state }),
930 _ => {}
931 }
932
933 if matches!(
934 state,
935 AttemptState::Preparing
936 | AttemptState::Prepared
937 | AttemptState::Destroying
938 | AttemptState::CleanupDeferred
939 ) && execution.is_native()
940 {
941 return Err(ExecutionError::AttemptIdentityMismatch.into());
942 }
943 if matches!(
944 state,
945 AttemptState::Prepared
946 | AttemptState::JitReceived
947 | AttemptState::Starting
948 | AttemptState::Idle
949 | AttemptState::Busy
950 ) && matches!(
951 &execution,
952 AttemptExecution::Isolated {
953 environment_id: None,
954 ..
955 }
956 ) {
957 return Err(ExecutionError::AttemptIdentityMismatch.into());
958 }
959
960 // Presence was checked above; *ordering* is checked here, and it is a
961 // separate hazard. `created_at` never moves and every other timestamp is
962 // written by a transition that happens after it, so a row where either
963 // precedes it is one this crate cannot have produced. Without this, a
964 // `finished` row with `created_at: ts(100), terminal_at: Some(ts(0))`
965 // loaded cleanly -- an attempt that concluded a hundred seconds before
966 // it was created -- and every duration computed from the pair came out
967 // negative or wrapped. The hand-edited-journal threat model that
968 // motivates the presence gate covers this equally.
969 if last_state_change_at < created_at {
970 return Err(AttemptError::TimestampsOutOfOrder {
971 state,
972 field: "last_state_change_at",
973 created_at,
974 found: last_state_change_at,
975 });
976 }
977 if let Some(terminal_at) = terminal_at
978 && terminal_at < created_at
979 {
980 return Err(AttemptError::TimestampsOutOfOrder {
981 state,
982 field: "terminal_at",
983 created_at,
984 found: terminal_at,
985 });
986 }
987
988 Ok(Self {
989 id,
990 policy_id,
991 github_runner_id,
992 state,
993 outcome,
994 process_id,
995 execution,
996 runtime_path,
997 workspace,
998 created_at,
999 terminal_at,
1000 last_state_change_at,
1001 })
1002 }
1003
1004 #[must_use]
1005 pub const fn state(&self) -> AttemptState {
1006 self.state
1007 }
1008
1009 #[must_use]
1010 pub const fn outcome(&self) -> Option<&AttemptOutcome> {
1011 self.outcome.as_ref()
1012 }
1013
1014 #[must_use]
1015 pub const fn github_runner_id(&self) -> Option<u64> {
1016 self.github_runner_id
1017 }
1018
1019 #[must_use]
1020 pub const fn process_id(&self) -> Option<u32> {
1021 self.process_id
1022 }
1023
1024 #[must_use]
1025 pub const fn execution(&self) -> &AttemptExecution {
1026 &self.execution
1027 }
1028
1029 /// Choose the provider intent before the first journal write.
1030 pub fn allocate_execution(&mut self, execution: AttemptExecution) -> Result<(), AttemptError> {
1031 if self.state != AttemptState::Allocated || self.process_id.is_some() {
1032 return Err(ExecutionError::AttemptIdentityMismatch.into());
1033 }
1034 execution.validate()?;
1035 if matches!(
1036 execution,
1037 AttemptExecution::Native {
1038 process_id: Some(_)
1039 }
1040 ) {
1041 return Err(ExecutionError::AttemptIdentityMismatch.into());
1042 }
1043 self.execution = execution;
1044 Ok(())
1045 }
1046
1047 /// Fill the provider identity through a durable journal write before an
1048 /// isolated runner receives JIT. A second or changed identity is refused.
1049 pub fn prepared_environment(&mut self, environment_id: String) -> Result<(), AttemptError> {
1050 if !matches!(
1051 self.state,
1052 AttemptState::Allocated
1053 | AttemptState::Preparing
1054 | AttemptState::Failed
1055 | AttemptState::Orphaned
1056 | AttemptState::Destroying
1057 | AttemptState::CleanupDeferred
1058 ) {
1059 return Err(ExecutionError::AttemptIdentityMismatch.into());
1060 }
1061 let AttemptExecution::Isolated {
1062 environment_id: stored,
1063 ..
1064 } = &mut self.execution
1065 else {
1066 return Err(ExecutionError::AttemptIdentityMismatch.into());
1067 };
1068 if stored.is_some() {
1069 return Err(ExecutionError::AttemptIdentityMismatch.into());
1070 }
1071 *stored = Some(environment_id);
1072 if let Err(error) = self.execution.validate() {
1073 if let AttemptExecution::Isolated {
1074 environment_id: stored,
1075 ..
1076 } = &mut self.execution
1077 {
1078 *stored = None;
1079 }
1080 return Err(error.into());
1081 }
1082 Ok(())
1083 }
1084
1085 /// Persist intent before a provider performs an external prepare effect.
1086 pub fn begin_prepare(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1087 if self.execution.is_native() {
1088 return Err(ExecutionError::AttemptIdentityMismatch.into());
1089 }
1090 self.move_to(AttemptState::Preparing, now)
1091 }
1092
1093 /// Persist the returned provider identity before requesting JIT.
1094 pub fn mark_prepared(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1095 if !matches!(
1096 self.execution,
1097 AttemptExecution::Isolated {
1098 environment_id: Some(_),
1099 ..
1100 }
1101 ) {
1102 return Err(ExecutionError::AttemptIdentityMismatch.into());
1103 }
1104 self.move_to(AttemptState::Prepared, now)
1105 }
1106
1107 #[must_use]
1108 pub fn runtime_path(&self) -> &Path {
1109 &self.runtime_path
1110 }
1111
1112 /// The immutable allocation fact: disposable, or the persistent slot leased.
1113 ///
1114 /// There is no setter. Cleanup and recovery dispatch on this
1115 /// (`02-target-architecture.md`, "Cleanup and recovery"), so a value that
1116 /// could be changed after allocation would let the algorithm chosen for a
1117 /// directory disagree with the one it was created under.
1118 #[must_use]
1119 pub const fn workspace(&self) -> AttemptWorkspace {
1120 self.workspace
1121 }
1122
1123 /// Whether this attempt holds a persistent slot lease.
1124 ///
1125 /// Every uncleaned persistent attempt is a lease, including a terminal one
1126 /// whose cleanup failed — which is why this asks about the workspace and not
1127 /// about the state.
1128 #[must_use]
1129 pub const fn holds_slot_lease(&self) -> bool {
1130 self.workspace.is_persistent() && !matches!(self.state, AttemptState::Cleaned)
1131 }
1132
1133 #[must_use]
1134 pub const fn terminal_at(&self) -> Option<Timestamp> {
1135 self.terminal_at
1136 }
1137
1138 /// When the attempt entered its current state. Recovery timeouts run from
1139 /// here.
1140 #[must_use]
1141 pub const fn last_state_change_at(&self) -> Timestamp {
1142 self.last_state_change_at
1143 }
1144
1145 #[must_use]
1146 pub const fn is_terminal(&self) -> bool {
1147 self.state.is_terminal()
1148 }
1149
1150 /// Whether this attempt still holds one of the host's capacity slots.
1151 /// Isolated environments retain the slot through terminal runner states
1152 /// until provider absence has been proved and cleanup is journalled.
1153 #[must_use]
1154 pub const fn counts_against_capacity(&self) -> bool {
1155 self.state.counts_against_capacity()
1156 || (!self.execution.is_native() && !matches!(self.state, AttemptState::Cleaned))
1157 }
1158
1159 fn move_to(&mut self, next: AttemptState, now: Timestamp) -> Result<(), AttemptError> {
1160 if !self.state.can_transition_to(next) {
1161 return Err(AttemptError::IllegalTransition {
1162 from: self.state,
1163 to: next,
1164 });
1165 }
1166 self.state = next;
1167 self.last_state_change_at = now;
1168 Ok(())
1169 }
1170
1171 /// `allocated -> jit_received`.
1172 ///
1173 /// # Errors
1174 /// [`AttemptError::IllegalTransition`] from any other state.
1175 pub fn jit_received(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1176 self.move_to(AttemptState::JitReceived, now)
1177 }
1178
1179 /// `jit_received -> starting`, recording the child process identity.
1180 ///
1181 /// # Errors
1182 /// [`AttemptError::IllegalTransition`] from any other state.
1183 pub fn started(&mut self, process_id: u32, now: Timestamp) -> Result<(), AttemptError> {
1184 if !self.execution.is_native() {
1185 return Err(ExecutionError::AttemptIdentityMismatch.into());
1186 }
1187 self.move_to(AttemptState::Starting, now)?;
1188 self.process_id = Some(process_id);
1189 self.execution = AttemptExecution::Native {
1190 process_id: Some(process_id),
1191 };
1192 Ok(())
1193 }
1194
1195 /// An isolated environment has no host PID; its immutable identity is already journalled.
1196 pub fn started_isolated(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1197 if !matches!(
1198 self.execution,
1199 AttemptExecution::Isolated {
1200 environment_id: Some(_),
1201 ..
1202 }
1203 ) {
1204 return Err(ExecutionError::AttemptIdentityMismatch.into());
1205 }
1206 self.move_to(AttemptState::Starting, now)
1207 }
1208
1209 /// `starting -> idle`: the runner registered and is awaiting its one
1210 /// assignment.
1211 ///
1212 /// # Errors
1213 /// [`AttemptError::IllegalTransition`] from any other state.
1214 pub fn registered_idle(
1215 &mut self,
1216 github_runner_id: u64,
1217 now: Timestamp,
1218 ) -> Result<(), AttemptError> {
1219 self.move_to(AttemptState::Idle, now)?;
1220 self.github_runner_id = Some(github_runner_id);
1221 Ok(())
1222 }
1223
1224 /// `starting | idle -> busy`: the runner was assigned its one job.
1225 ///
1226 /// Both sources are real. A runner may be observed taking a job directly out
1227 /// of `starting`, or it may be seen `idle` first and pick the job up on a
1228 /// later pass — `e3`'s Scope step 4 walks the second sequence explicitly.
1229 ///
1230 /// # Errors
1231 /// [`AttemptError::IllegalTransition`] from any other state.
1232 pub fn assigned_job(
1233 &mut self,
1234 github_runner_id: u64,
1235 now: Timestamp,
1236 ) -> Result<(), AttemptError> {
1237 self.move_to(AttemptState::Busy, now)?;
1238 self.github_runner_id = Some(github_runner_id);
1239 Ok(())
1240 }
1241
1242 /// `busy -> idle`: GitHub showed this runner assigned, but the runner never
1243 /// started the job.
1244 ///
1245 /// GitHub's inventory marks a runner busy when it hands it a job, before the
1246 /// runner has acquired it. An assignment the runner does not take up is
1247 /// re-queued to another runner, and this one goes back to waiting (or, as a
1248 /// one-shot JIT runner, exits). Without this edge the journal went on saying
1249 /// `busy`, so the attempt was later recorded `completed_job` for a job it
1250 /// never ran, and a live runner in that position was never held to the idle
1251 /// timeout. The caller takes the edge only with local proof that no job ran:
1252 /// the runner's own diagnostics show its listener but no job worker.
1253 ///
1254 /// # Errors
1255 /// [`AttemptError::IllegalTransition`] from any state but `busy`.
1256 pub fn assignment_not_taken(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1257 if self.state != AttemptState::Busy {
1258 return Err(AttemptError::IllegalTransition {
1259 from: self.state,
1260 to: AttemptState::Idle,
1261 });
1262 }
1263 self.move_to(AttemptState::Idle, now)
1264 }
1265
1266 /// Record the terminal outcome, moving to the state it implies.
1267 ///
1268 /// # Errors
1269 /// [`AttemptError::OutcomeUnreachable`] when the outcome does not belong to
1270 /// the current state — `CompletedJob` from anything but `busy`, or
1271 /// `ExitedIdleWithoutWork` from anything but `idle` — and
1272 /// [`AttemptError::IllegalTransition`] otherwise.
1273 pub fn conclude(
1274 &mut self,
1275 outcome: AttemptOutcome,
1276 now: Timestamp,
1277 ) -> Result<(), AttemptError> {
1278 if !outcome.required_from().contains(&self.state) {
1279 return Err(AttemptError::OutcomeUnreachable {
1280 from: self.state,
1281 outcome,
1282 });
1283 }
1284 self.move_to(outcome.terminal_state(), now)?;
1285 self.terminal_at = Some(now);
1286 self.outcome = Some(outcome);
1287 Ok(())
1288 }
1289
1290 /// `finished | failed | orphaned -> cleaned`.
1291 ///
1292 /// # Errors
1293 /// [`AttemptError::BusyCannotBeCleaned`] for a `busy` attempt — the
1294 /// scale-down case `04-subsystem-contracts.md` forbids — and
1295 /// [`AttemptError::IllegalTransition`] for any other non-terminal state or
1296 /// for an already-cleaned attempt.
1297 pub fn clean(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1298 if self.state == AttemptState::Busy {
1299 return Err(AttemptError::BusyCannotBeCleaned);
1300 }
1301 if !self.execution.is_native() {
1302 return Err(ExecutionError::IsolatedCleanupUnproven.into());
1303 }
1304 if !self.state.is_concluded() {
1305 return Err(AttemptError::IllegalTransition {
1306 from: self.state,
1307 to: AttemptState::Cleaned,
1308 });
1309 }
1310 self.move_to(AttemptState::Cleaned, now)
1311 }
1312
1313 /// Durable cleanup intent, recorded before stop or destroy is attempted.
1314 pub fn begin_destroy(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1315 if self.execution.is_native() || self.outcome.is_none() {
1316 return Err(ExecutionError::AttemptIdentityMismatch.into());
1317 }
1318 self.move_to(AttemptState::Destroying, now)
1319 }
1320
1321 /// The provider still owns resources; preserve the capacity lease for a retry.
1322 pub fn defer_cleanup(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1323 if self.execution.is_native() {
1324 return Err(ExecutionError::AttemptIdentityMismatch.into());
1325 }
1326 self.move_to(AttemptState::CleanupDeferred, now)
1327 }
1328
1329 /// Only the lifecycle may call this after an independent missing observation.
1330 pub fn clean_isolated(&mut self, now: Timestamp) -> Result<(), AttemptError> {
1331 if self.execution.is_native() {
1332 return Err(ExecutionError::AttemptIdentityMismatch.into());
1333 }
1334 self.move_to(AttemptState::Cleaned, now)
1335 }
1336}
1337
1338// ---------------------------------------------------------------------------
1339// Ownership
1340// ---------------------------------------------------------------------------
1341
1342/// Ownership rules 1 and 2 (`04-subsystem-contracts.md`).
1343///
1344/// An attempt records its `policy_id`, and the policy records its `host_id`, so
1345/// authorising an attempt is a two-link check and both links matter. `e3`'s
1346/// restart recovery runs this before it touches a process it found on the
1347/// machine: adopting, terminating, or cleaning another host's attempt is the
1348/// failure this rule exists to prevent.
1349///
1350/// # Errors
1351/// [`OwnershipError::PolicyMismatch`] when the attempt does not belong to the
1352/// policy, and [`OwnershipError::ForeignHost`] when the policy does not belong to
1353/// this agent's host.
1354pub fn authorize(
1355 agent_host: HostId,
1356 policy: &ScalePolicy,
1357 attempt: &RunnerAttempt,
1358) -> Result<(), OwnershipError> {
1359 if attempt.policy_id != policy.id {
1360 return Err(OwnershipError::PolicyMismatch {
1361 attempt: attempt.id,
1362 attempt_policy: attempt.policy_id,
1363 policy: policy.id,
1364 });
1365 }
1366 if !policy.is_owned_by(agent_host) {
1367 return Err(OwnershipError::ForeignHost {
1368 policy: policy.id,
1369 owner: policy.host_id,
1370 agent: agent_host,
1371 });
1372 }
1373 Ok(())
1374}
1375
1376// ---------------------------------------------------------------------------
1377// Recovery
1378// ---------------------------------------------------------------------------
1379
1380/// How long an attempt may sit in each pre-terminal state before recovery treats
1381/// it as stuck.
1382///
1383/// **No document in this taskflow states these durations.** `03-control-flows.md`
1384/// flow 2.7 says a surplus runner "exits on its idle timeout" without giving one,
1385/// and flow 4.4 says an expired JIT configuration is discarded without saying
1386/// when it expires. [`RecoveryTimeouts::provisional`] therefore returns values
1387/// chosen here, named so that a caller cannot mistake them for a product
1388/// decision, and there is deliberately no `Default` impl — `e1` and `e3` should
1389/// have to write the numbers down.
1390#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1391pub struct RecoveryTimeouts {
1392 /// How long `allocated` or `jit_received` may last before the JIT
1393 /// configuration is assumed lost.
1394 pub jit_handoff: Elapsed,
1395 /// How long `starting` may last before the runner is assumed not to be
1396 /// coming up.
1397 pub startup: Elapsed,
1398 /// How long `idle` may last, in both directions it is read.
1399 ///
1400 /// For an attempt whose process is already gone it separates the two
1401 /// readings of that exit: past this, the surplus case; before it, a crash.
1402 ///
1403 /// For one whose process is still alive and still registered it is a
1404 /// deadline rather than a reading — the point at which the agent stops the
1405 /// runner itself. Nothing else does: `Runner.Listener run` long-polls for
1406 /// an assignment indefinitely, so this value, and only this value, bounds
1407 /// how long a runner that never gets a job holds its capacity slot and its
1408 /// entry in the target's runner settings.
1409 pub idle: Elapsed,
1410}
1411
1412impl RecoveryTimeouts {
1413 #[must_use]
1414 pub const fn new(jit_handoff: Elapsed, startup: Elapsed, idle: Elapsed) -> Self {
1415 Self {
1416 jit_handoff,
1417 startup,
1418 idle,
1419 }
1420 }
1421
1422 /// Placeholder values, with one exception — see the type documentation.
1423 ///
1424 /// `idle` is no longer a placeholder. It stopped being one when it became
1425 /// the only thing that bounds a live runner: five minutes is long enough
1426 /// that a runner GitHub is about to assign is not stopped out from under
1427 /// the assignment, and short enough that a machine does not carry an
1428 /// unusable slot, and a repository an unusable runner row, for hours. The
1429 /// other two still only separate readings of an event that already
1430 /// happened, and nothing here has had to decide what they should be.
1431 #[must_use]
1432 pub fn provisional() -> Self {
1433 Self {
1434 jit_handoff: Elapsed::seconds(120),
1435 startup: Elapsed::seconds(300),
1436 idle: Elapsed::seconds(300),
1437 }
1438 }
1439}
1440
1441/// What GitHub says about this attempt's runner.
1442///
1443/// Precedence rule 3: "GitHub runner status is authoritative for remote job
1444/// status; local process state is authoritative only for a child process owned by
1445/// this agent." Both halves are inputs here, and neither is allowed to stand in
1446/// for the other.
1447#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1448pub enum GithubRunnerObservation {
1449 /// GitHub could not be reached this cycle. Flow 3.3: start nothing, retain
1450 /// what is running, back off. It is emphatically **not** the same as
1451 /// `NotRegistered`, and conflating the two would delete live runners during
1452 /// an outage.
1453 Unreachable,
1454 /// GitHub knows no runner for this attempt.
1455 NotRegistered,
1456 /// GitHub knows the runner.
1457 Registered { busy: bool },
1458}
1459
1460/// One attempt's observed reality at recovery time.
1461#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1462pub struct RecoveryObservation {
1463 /// Whether the child process this agent recorded is still alive. `d1` supplies
1464 /// a process identity that survives a reboot, because a bare PID is reused.
1465 pub process_alive: bool,
1466 pub github: GithubRunnerObservation,
1467}
1468
1469/// What to do about one attempt found in the journal at startup.
1470#[derive(Debug, Clone, PartialEq, Eq)]
1471pub enum RecoveryDecision {
1472 /// Already cleaned; there is nothing left to do.
1473 Nothing,
1474 /// Terminal but not yet cleaned: remove the runtime directory and mark it
1475 /// `cleaned`.
1476 Clean,
1477 /// The process is alive and this attempt is still ours. `e3`: "an attempt
1478 /// whose process still runs is adopted, not duplicated."
1479 Adopt,
1480 /// Nothing is decidable yet; look again next cycle.
1481 Wait,
1482 /// GitHub is unreachable. Decide nothing destructive during an outage
1483 /// (flow 3.3).
1484 Defer,
1485 /// Move the attempt to this state to match what was observed.
1486 ///
1487 /// **Where the observed state is live, the caller must adopt supervision of
1488 /// the process independently of this decision.** `RecoveryDecision` has no
1489 /// way to express "adopt *and* observe" — the two are separate variants —
1490 /// so `Starting` with `process_alive` and `Registered { busy: false }`
1491 /// returns `Observe(Idle)` and never [`Self::Adopt`], because GitHub is
1492 /// authoritative for the remote status while the local process is
1493 /// authoritative for supervision, and both facts are true at once.
1494 /// `Idle` with `process_alive` and `Registered { busy: true }` returns
1495 /// `Observe(Busy)` for exactly the same reason; it did not always, and the
1496 /// arm's own comment records what that cost.
1497 ///
1498 /// **An `Observe` decision must be applied *and persisted*, or it repeats
1499 /// forever.** The decision is a pure function of the journalled state and
1500 /// the observation, so a caller that moves the in-memory attempt without
1501 /// writing the new state back will read the same stale state on the next
1502 /// pass and be handed the same decision, indefinitely.
1503 Observe(AttemptState),
1504 /// Conclude the attempt with this outcome.
1505 ///
1506 /// **Terminal, and therefore a capacity release.** An attempt is only
1507 /// concluded where the thing it was supervising is already gone; see
1508 /// [`Self::Terminate`] for the case where it is not.
1509 Conclude(AttemptOutcome),
1510 /// The process is still alive but the attempt cannot go on: **stop the
1511 /// process, and only once it is gone record this outcome.**
1512 ///
1513 /// **Two producers, and the payload is what separates them.** `starting`
1514 /// past its registration timeout carries
1515 /// [`FailureReason::RegistrationTimedOut`]; `idle` past its idle timeout,
1516 /// with GitHub still reporting the runner registered and unassigned,
1517 /// carries [`AttemptOutcome::ExitedIdleWithoutWork`] — flow 2.7's surplus
1518 /// runner, which is a normal outcome and not a failure at all. The two
1519 /// share this variant because they need the identical *sequence* — signal,
1520 /// confirm the process is gone, then record — and differ only in what is
1521 /// recorded at the end. A caller that hardcodes either reason will
1522 /// mislabel the other, so the payload is the caller's instruction, not
1523 /// decoration.
1524 ///
1525 /// **Why this is not a [`Self::Conclude`].** `Conclude` moves the attempt to
1526 /// a terminal state, and a terminal attempt no longer
1527 /// [counts against capacity](AttemptState::counts_against_capacity) — so
1528 /// concluding one whose process is still running hands the host back a slot
1529 /// it is still using. The agent then starts a replacement runner beside a
1530 /// live, unregistered one that may yet register and take a job, for an
1531 /// attempt the journal already calls `failed`. There is no
1532 /// `RecoveryDecision` that would have expressed the fix: [`Self::Adopt`]
1533 /// means take over supervision and [`Self::Clean`] means delete a runtime
1534 /// directory, and neither stops anything.
1535 ///
1536 /// **What happens to the capacity slot.** Nothing, until the process is
1537 /// actually gone. The attempt stays in its current, non-terminal state and
1538 /// keeps holding its slot for as long as the runner it started is running,
1539 /// which is the honest answer — the resources are genuinely occupied. The
1540 /// slot comes back at the moment the caller applies the payload through
1541 /// [`RunnerAttempt::conclude`], which it does only after the process has
1542 /// exited.
1543 ///
1544 /// **It is safe to re-derive in one direction, and owes a debt in the
1545 /// other.** The decision is a pure function of the journalled state and the
1546 /// observation, so an agent that dies *before* terminating sees a live
1547 /// process and a larger `elapsed` on the next pass and is handed this same
1548 /// decision again. That half costs nothing and needs nothing.
1549 ///
1550 /// **The other half is a known defect, and it is not harmless.** An agent
1551 /// that terminates the process and then dies *before* writing the outcome
1552 /// observes `process_alive: false` next time and reaches the ordinary
1553 /// [`Self::Conclude`] arm, which records
1554 /// [`FailureReason::ProcessExitedUnexpectedly`]. The process did not exit
1555 /// unexpectedly; this agent killed it. That is precisely the diagnosis
1556 /// [`FailureReason::RegistrationTimedOut`] was split out to prevent — the
1557 /// two reasons send an operator to different places, and this window sends
1558 /// them to a crash investigation (logs, exit code, a corrupt package) for
1559 /// what was a registration failure.
1560 ///
1561 /// **It cannot be fixed here, and the alternative was weighed rather than
1562 /// waved off.** This function's inputs are the journalled state and a
1563 /// [`RecoveryObservation`], and neither carries the fact that separates the
1564 /// two cases: a process this agent killed and a process that crashed on its
1565 /// own present the *same observation*. Using `RegistrationTimedOut` in the
1566 /// dead-process arm once `elapsed >= startup` would close this window at the
1567 /// price of a wider one — every genuine early crash first observed after a
1568 /// restart longer than the startup window would then be reported as a runner
1569 /// that "is running but did not register", to an operator who can see that
1570 /// it is not running. That trades a rare wrong reason for a common false
1571 /// claim about liveness, in the one direction
1572 /// [`FailureReason::ProcessExitedUnexpectedly`]'s own documentation says
1573 /// spends the credibility of every other message this product prints.
1574 /// `tests::no_decision_calls_a_dead_process_live` is what stops that trade
1575 /// being made by accident later.
1576 ///
1577 /// **So the obligation is `e3`'s, and it is a persistence one.** The
1578 /// distinguishing fact exists only at the moment terminate-intent is formed,
1579 /// and the only way to carry it across a crash is to write it down. `e3`
1580 /// journals the intent *before* it signals the process, and on a later pass
1581 /// concludes an attempt it finds so marked with
1582 /// [`FailureReason::TerminatedAfterRegistrationTimeout`] rather than with
1583 /// whatever this function derived from an observation that could not know.
1584 /// Until `e3` does that, the window stands.
1585 ///
1586 /// **Why that closure needs a reason of its own, and not
1587 /// `RegistrationTimedOut`.** On the pass where `e3` reads the mark back, the
1588 /// process is dead — `e3` killed it, which is the whole reason the mark is
1589 /// there. `RegistrationTimedOut` renders as "the runner process is running
1590 /// but did not register", so concluding with it would print exactly the
1591 /// false liveness claim the paragraph above rejects option A for: the same
1592 /// sentence, about a process an operator can see is gone, moved one pass
1593 /// later. Rewording that string instead is not open either —
1594 /// `tests::the_two_starting_failures_read_differently_to_an_operator` pins
1595 /// it on "running", which is correct for the live case it names.
1596 ///
1597 /// [`FailureReason::TerminatedAfterRegistrationTimeout`] is that reason.
1598 /// It is true of a dead process, it says who stopped it and why, and it
1599 /// sends an operator to the networking and configuration fix rather than to
1600 /// a crash investigation — which is the whole distinction
1601 /// [`FailureReason::RegistrationTimedOut`] was split out to draw. Nothing in
1602 /// this function derives it, and nothing should: it is a claim about an
1603 /// action this agent took, not about anything a [`RecoveryObservation`]
1604 /// reports. `tests::no_decision_calls_a_dead_process_live` covers it
1605 /// alongside the other two liveness-claiming reasons, so the day something
1606 /// here does start deriving it — which it legitimately might, once the mark
1607 /// is journalled where this function can read it — it may only do so beside
1608 /// a process the observation says is gone.
1609 ///
1610 /// **The one thing this does not bound** is a process that refuses to die.
1611 /// The slot is held until it does. That is a worse outcome than concluding
1612 /// early only if the runner was never going to register, and a better one in
1613 /// every case where it was — and unlike the early conclusion it cannot
1614 /// oversubscribe the host.
1615 Terminate(AttemptOutcome),
1616}
1617
1618/// Decide what to do about one journalled attempt after a restart, or on any
1619/// reconciliation pass.
1620///
1621/// Time enters only through `clock`, and only as `now - attempt.
1622/// last_state_change_at()`, so the whole function is exercised by advancing a
1623/// [`crate::model::Clock`] the test controls.
1624#[must_use]
1625pub fn recovery_decision(
1626 attempt: &RunnerAttempt,
1627 observation: RecoveryObservation,
1628 timeouts: RecoveryTimeouts,
1629 clock: &dyn Clock,
1630) -> RecoveryDecision {
1631 use AttemptState as S;
1632 use GithubRunnerObservation as G;
1633
1634 let state = attempt.state();
1635
1636 if state == S::Cleaned {
1637 return RecoveryDecision::Nothing;
1638 }
1639 if state.is_terminal() {
1640 return RecoveryDecision::Clean;
1641 }
1642 if observation.github == G::Unreachable {
1643 return RecoveryDecision::Defer;
1644 }
1645
1646 let elapsed = clock.now() - attempt.last_state_change_at();
1647
1648 match state {
1649 S::Allocated | S::JitReceived => {
1650 if observation.process_alive {
1651 // Precedence rule 3's other half: local process state is
1652 // authoritative for a child process this agent owns. `e3`: an
1653 // attempt whose process still runs is adopted, not duplicated.
1654 RecoveryDecision::Adopt
1655 } else {
1656 match observation.github {
1657 // The crash window `e3` exists for. A runner cannot register
1658 // without having received its JIT configuration and started,
1659 // so a registration is proof the attempt got further than
1660 // the journal records: the `starting` write was lost, not
1661 // the attempt. This branch used to consult only
1662 // `process_alive` and the clock, which abandoned a live
1663 // runner as `failed` and left its registration at GitHub
1664 // unreconciled and unremoved.
1665 //
1666 // Each state takes the one *forward* edge out of itself, so
1667 // recovery walks the diagram rather than jumping across it:
1668 // `allocated -> jit_received` (the registration proves the
1669 // JIT configuration arrived) and `jit_received -> starting`.
1670 // `busy` is deliberately not consulted here -- neither state
1671 // has an edge to it, and the next pass, from `starting`, is
1672 // where that distinction becomes legal and is drawn.
1673 //
1674 // This is a genuine recovery path and not a workaround for a
1675 // missing edge: GitHub really is reporting a live runner,
1676 // and each step really did happen. It survived the diagram
1677 // amendment unchanged.
1678 G::Registered { .. } => RecoveryDecision::Observe(if state == S::Allocated {
1679 S::JitReceived
1680 } else {
1681 S::Starting
1682 }),
1683 // Past the handoff deadline with nothing at GitHub and no
1684 // process: the attempt died before registering, and since
1685 // the amendment it can say so directly. It previously had to
1686 // report `NoLegalTransition` and hold its host capacity slot
1687 // for ever.
1688 //
1689 // The two states get different reasons because they know
1690 // different things. At `allocated` no configuration was ever
1691 // recorded as arriving, so the request itself did not
1692 // complete; at `jit_received` one arrived and was never
1693 // claimed, which is flow 4.4's expiry.
1694 G::NotRegistered => {
1695 if elapsed >= timeouts.jit_handoff {
1696 RecoveryDecision::Conclude(AttemptOutcome::failed(
1697 if state == S::Allocated {
1698 FailureReason::JitRequestFailed
1699 } else {
1700 FailureReason::JitExpired
1701 },
1702 ))
1703 } else {
1704 RecoveryDecision::Wait
1705 }
1706 }
1707 G::Unreachable => unreachable!("handled above"),
1708 }
1709 }
1710 }
1711
1712 S::Starting => match observation.github {
1713 // GitHub is authoritative for remote job status, and both of these
1714 // are legal edges out of `starting`.
1715 G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
1716 G::Registered { busy: false } => RecoveryDecision::Observe(S::Idle),
1717 // Three cases, not two, and the third is the one that matters.
1718 //
1719 // A live process inside its startup window is adopted, the same as
1720 // in every other pre-terminal state: `e3` must take over supervision
1721 // rather than start a second runner for the same work.
1722 //
1723 // A process that is *gone* is flow 2's "runner exit before job
1724 // acceptance". Since the amendment that is recordable, so the
1725 // attempt concludes and gives its slot back.
1726 //
1727 // A process that is alive and past its deadline is neither, and
1728 // collapsing it into the second was a real defect: `Conclude` makes
1729 // the attempt terminal, a terminal attempt stops counting against
1730 // capacity, and the host therefore got its slot back while the
1731 // runner was still running -- free to register late and take a job
1732 // for an attempt the journal already called `failed`, beside the
1733 // replacement the freed slot let the agent start. It also read
1734 // `ProcessExitedUnexpectedly` to an operator looking at the process
1735 // in a task manager. `Terminate` says the true thing and keeps the
1736 // slot until the process is actually gone.
1737 //
1738 // The dead-process arm below carries a residual this arm cannot
1739 // close: an agent that terminated the process and died before
1740 // writing the outcome lands there and records
1741 // `ProcessExitedUnexpectedly` for a process it killed itself. The
1742 // fact that would separate the two is not in this function's
1743 // inputs -- a killed process and a crashed one are the same
1744 // observation -- so the fix is `e3` journalling terminate-intent
1745 // before signalling and concluding the marked attempt with
1746 // `FailureReason::TerminatedAfterRegistrationTimeout`, not a
1747 // rearrangement here. That reason exists precisely because the
1748 // process is gone by then, so the closure cannot reuse
1749 // `RegistrationTimedOut` without printing the same false liveness
1750 // claim one pass later. See `RecoveryDecision::Terminate`, which
1751 // names the trade and the owner, and
1752 // `tests::no_decision_calls_a_dead_process_live`, which reds if
1753 // somebody closes it here instead.
1754 G::NotRegistered => {
1755 if !observation.process_alive {
1756 RecoveryDecision::Conclude(AttemptOutcome::failed(
1757 FailureReason::ProcessExitedUnexpectedly,
1758 ))
1759 } else if elapsed < timeouts.startup {
1760 RecoveryDecision::Adopt
1761 } else {
1762 RecoveryDecision::Terminate(AttemptOutcome::failed(
1763 FailureReason::RegistrationTimedOut,
1764 ))
1765 }
1766 }
1767 G::Unreachable => unreachable!("handled above"),
1768 },
1769
1770 // GitHub is consulted **first**, exactly as at `starting` above, and the
1771 // symmetry is load-bearing rather than tidy. This arm used to
1772 // short-circuit on `process_alive` before looking at GitHub, so the same
1773 // conflict -- a live process that GitHub reports as `busy` -- resolved
1774 // one way here and the opposite way one state earlier. What that cost
1775 // was not an inconsistency but a wrong outcome: `Adopt` left the journal
1776 // saying `idle`, so `last_state_change_at` went on pointing at the idle
1777 // entry, and a runner that later crashed *during its job* had its idle
1778 // timeout elapse and was concluded `ExitedIdleWithoutWork` -- the benign
1779 // surplus exit, recorded for a mid-job crash, inverting the one
1780 // distinction this module exists to keep. Nothing downstream could
1781 // catch it either: `required_from` for that outcome is `&[Idle]`, and
1782 // the attempt really was `idle`.
1783 //
1784 // `Observe(Busy)` is returned whether or not the process is alive: the
1785 // caller adopts supervision independently of this decision, which is
1786 // what [`RecoveryDecision::Observe`] already instructs and what
1787 // `starting` has always relied on.
1788 S::Idle => match observation.github {
1789 // GitHub says this runner took a job, and GitHub is authoritative
1790 // for remote job status (precedence rule 3). `idle -> busy` was
1791 // added by the amendment *for this observation*; before this change
1792 // the only arm that could reach it was the one where the process is
1793 // already dead.
1794 G::Registered { busy: true } => RecoveryDecision::Observe(S::Busy),
1795 // GitHub agrees with the journal, so there is no state to move to.
1796 // A dead process means supervision is lost while the remote
1797 // registration outlived it and needs removing.
1798 //
1799 // A *live* one is adopted only while it is still inside the idle
1800 // timeout. Past it, this is flow 2.7's surplus runner and the agent
1801 // has to end it, because nothing else will: the runner is spawned as
1802 // a bare `Runner.Listener run`, and that process has no idle timeout
1803 // of its own -- it long-polls for an assignment until something
1804 // stops it. This arm used to answer `Adopt` for every elapsed time,
1805 // which is why it never was stopped. A registered, unassigned runner
1806 // then holds its capacity slot and its entry in the target's runner
1807 // settings for as long as the host stays up; one observed in the
1808 // field sat here for 27 hours across two restarts of nothing.
1809 //
1810 // `Terminate`, not `Conclude`: the process is alive, so the slot may
1811 // not be returned until it is gone. The caller signals it, re-reads
1812 // liveness, and only then applies the payload -- the same sequence
1813 // `starting` above relies on, and the reason that decision carries
1814 // its outcome rather than the caller deriving one.
1815 G::Registered { busy: false } => {
1816 if !observation.process_alive {
1817 RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
1818 } else if elapsed >= timeouts.idle {
1819 RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork)
1820 } else {
1821 RecoveryDecision::Adopt
1822 }
1823 }
1824 // Nothing at GitHub. A live process is still ours to supervise.
1825 // Otherwise the surplus case and the crash case, separated by the
1826 // clock: a runner that sat out its whole idle timeout and then
1827 // exited with no registration left behind did what flow 2.7
1828 // describes. One that vanished early did not.
1829 G::NotRegistered => {
1830 if observation.process_alive {
1831 RecoveryDecision::Adopt
1832 } else if elapsed >= timeouts.idle {
1833 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork)
1834 } else {
1835 RecoveryDecision::Conclude(AttemptOutcome::failed(
1836 FailureReason::ProcessExitedUnexpectedly,
1837 ))
1838 }
1839 }
1840 G::Unreachable => unreachable!("handled above"),
1841 },
1842
1843 S::Busy => {
1844 if observation.process_alive {
1845 RecoveryDecision::Adopt
1846 } else {
1847 // `e3`: "an attempt whose process is gone and whose runner is
1848 // unknown to GitHub is `orphaned` and cleaned". The agent never
1849 // reports a job as complete, so a lost supervision is recorded as
1850 // exactly that and not guessed into a success.
1851 RecoveryDecision::Conclude(AttemptOutcome::Orphaned)
1852 }
1853 }
1854
1855 S::Preparing | S::Prepared => RecoveryDecision::Wait,
1856 S::Finished | S::Failed | S::Orphaned | S::Destroying | S::CleanupDeferred | S::Cleaned => {
1857 unreachable!("terminal states are handled above")
1858 }
1859 }
1860}
1861
1862/// How many of these attempts still occupy a host capacity slot.
1863///
1864/// Saturating rather than wrapping: a host cannot hold more than `u16::MAX`
1865/// attempts, and if some caller ever produced that many, reporting the ceiling is
1866/// safe where wrapping to zero would let the allocator start `u16::MAX` more.
1867#[must_use]
1868pub fn active_count<'a>(attempts: impl IntoIterator<Item = &'a RunnerAttempt>) -> u16 {
1869 attempts
1870 .into_iter()
1871 .filter(|a| a.counts_against_capacity())
1872 .fold(0u16, |acc, _| acc.saturating_add(1))
1873}
1874
1875/// How many of these attempts still occupy a slot **and belong to one policy**.
1876///
1877/// The per-policy counterpart of [`active_count`], and the term
1878/// [`crate::capacity::HostAllocator::allocate`] subtracts. The two are not
1879/// interchangeable: [`active_count`] is the host-wide total that bounds D9's
1880/// ceiling, this one is the per-policy figure that bounds D7's, and substituting
1881/// either for the other is silent — the host-wide count in a per-policy slot
1882/// starves every policy but the first, and a zero in place of this one starts a
1883/// duplicate runner on every poll.
1884#[must_use]
1885pub fn active_count_for<'a>(
1886 policy_id: PolicyId,
1887 attempts: impl IntoIterator<Item = &'a RunnerAttempt>,
1888) -> u16 {
1889 attempts
1890 .into_iter()
1891 .filter(|a| a.policy_id == policy_id && a.counts_against_capacity())
1892 .fold(0u16, |acc, _| acc.saturating_add(1))
1893}
1894
1895#[cfg(test)]
1896mod tests {
1897 use super::*;
1898 use crate::model::{Arch, CachePolicy, HostLabel, Os, ScaleTarget};
1899 use crate::policy::{PolicyMode, RoutingLabels};
1900 use std::num::NonZeroU16;
1901
1902 #[derive(Debug)]
1903 struct StubClock(std::sync::Mutex<Timestamp>);
1904
1905 impl StubClock {
1906 fn at(secs: i64) -> Self {
1907 Self(std::sync::Mutex::new(ts(secs)))
1908 }
1909 fn set(&self, secs: i64) {
1910 *self.0.lock().unwrap() = ts(secs);
1911 }
1912 }
1913
1914 impl Clock for StubClock {
1915 fn now(&self) -> Timestamp {
1916 *self.0.lock().unwrap()
1917 }
1918 }
1919
1920 fn ts(secs: i64) -> Timestamp {
1921 chrono::DateTime::from_timestamp(secs, 0).expect("valid timestamp")
1922 }
1923
1924 fn attempt_in(state: AttemptState, entered_at: i64) -> RunnerAttempt {
1925 let mut attempt = RunnerAttempt::allocate(
1926 AttemptId::from_u128(1),
1927 PolicyId::from_u128(1),
1928 "runtime/p/a",
1929 ts(0),
1930 );
1931 // Set the state directly: only possible from inside the module, which is
1932 // exactly why the state-machine tests live here.
1933 attempt.state = state;
1934 attempt.last_state_change_at = ts(entered_at);
1935 attempt
1936 }
1937
1938 /// The same, under a nominated policy, for the per-policy count.
1939 fn attempt_for(policy_id: PolicyId, state: AttemptState, entered_at: i64) -> RunnerAttempt {
1940 let mut attempt = attempt_in(state, entered_at);
1941 attempt.policy_id = policy_id;
1942 attempt
1943 }
1944
1945 fn a_policy(host: HostId, policy_id: PolicyId) -> ScalePolicy {
1946 ScalePolicy::new(
1947 policy_id,
1948 ScaleTarget::repository("o/r").unwrap(),
1949 1,
1950 host,
1951 PolicyMode::autoscale(
1952 RoutingLabels::derive(&HostLabel::new("home").unwrap(), Os::Windows, Arch::X64),
1953 0,
1954 NonZeroU16::new(1).unwrap(),
1955 )
1956 .unwrap(),
1957 CachePolicy::default(),
1958 )
1959 }
1960
1961 // =======================================================================
1962 // The state machine, both directions
1963 // =======================================================================
1964
1965 /// The diagram from `04-subsystem-contracts.md`, transcribed by hand.
1966 ///
1967 /// **Deliberately a second copy of [`AttemptState::LEGAL`]**, for the reason
1968 /// given on `policy::tests::diagram_edges`: a test whose expectation is read
1969 /// out of the constant under test asserts only that the constant equals
1970 /// itself, and would accept any edge someone added to it.
1971 ///
1972 /// ```text
1973 /// allocated -> jit_received -> starting -> idle | busy
1974 /// idle -> busy
1975 /// busy -> idle
1976 /// allocated | jit_received | starting -> failed | orphaned
1977 /// idle | busy -> finished | failed | orphaned
1978 /// finished | failed | orphaned -> cleaned
1979 /// ```
1980 fn diagram_edges() -> Vec<(AttemptState, AttemptState)> {
1981 use AttemptState::*;
1982 // Line 1.
1983 let mut edges = vec![
1984 (Allocated, JitReceived),
1985 (Allocated, Preparing),
1986 (Preparing, Prepared),
1987 (Prepared, JitReceived),
1988 (JitReceived, Starting),
1989 (Starting, Idle),
1990 (Starting, Busy),
1991 ];
1992 // Line 2, added by the 2026-08-21 amendment.
1993 edges.push((Idle, Busy));
1994 // Added when an assignment GitHub reported was shown not to have been
1995 // taken up: the runner's diagnostics hold no job worker.
1996 edges.push((Busy, Idle));
1997 // Line 3, added by the same amendment.
1998 for from in [Allocated, Preparing, Prepared, JitReceived, Starting] {
1999 for to in [Failed, Orphaned] {
2000 edges.push((from, to));
2001 }
2002 }
2003 // Line 4.
2004 for from in [Idle, Busy] {
2005 for to in [Finished, Failed, Orphaned] {
2006 edges.push((from, to));
2007 }
2008 }
2009 // Line 5.
2010 for from in [Finished, Failed, Orphaned] {
2011 edges.push((from, Cleaned));
2012 edges.push((from, Destroying));
2013 }
2014 edges.extend([
2015 (Destroying, CleanupDeferred),
2016 (CleanupDeferred, Destroying),
2017 (Destroying, Cleaned),
2018 ]);
2019 edges
2020 }
2021
2022 #[test]
2023 fn every_attempt_state_transition_is_legal_exactly_where_the_diagram_says() {
2024 let expected = diagram_edges();
2025 assert_eq!(
2026 expected.len(),
2027 34,
2028 "the transcription itself changed; check it against the diagram"
2029 );
2030
2031 let mut legal_seen = 0usize;
2032 let mut illegal_seen = 0usize;
2033
2034 for from in AttemptState::ALL {
2035 for to in AttemptState::ALL {
2036 let expected_legal = expected.contains(&(from, to));
2037 let mut attempt = attempt_in(from, 0);
2038 let result = attempt.move_to(to, ts(10));
2039
2040 if expected_legal {
2041 legal_seen += 1;
2042 assert!(
2043 result.is_ok(),
2044 "{from} -> {to} is in the diagram and must be accepted"
2045 );
2046 assert_eq!(attempt.state(), to);
2047 assert_eq!(attempt.last_state_change_at(), ts(10));
2048 } else {
2049 illegal_seen += 1;
2050 assert!(
2051 matches!(result, Err(AttemptError::IllegalTransition { .. })),
2052 "{from} -> {to} is not in the diagram and must be rejected"
2053 );
2054 assert_eq!(
2055 attempt.state(),
2056 from,
2057 "a refused transition changes nothing"
2058 );
2059 assert_eq!(attempt.last_state_change_at(), ts(0));
2060 }
2061 }
2062 }
2063
2064 assert_eq!(legal_seen, 34);
2065 assert_eq!(illegal_seen, 13 * 13 - 34);
2066
2067 // And the published constant matches the transcription.
2068 let mut published = AttemptState::LEGAL.to_vec();
2069 let mut transcribed = expected;
2070 published.sort_unstable();
2071 transcribed.sort_unstable();
2072 assert_eq!(published, transcribed);
2073 }
2074
2075 #[test]
2076 fn every_live_state_has_a_terminal_edge_so_no_attempt_can_hold_a_slot_forever() {
2077 // The operational half of the amendment, asserted as the property it is
2078 // rather than as a list of edges. An attempt occupies a host capacity
2079 // slot for exactly as long as it is non-terminal, so a live state with
2080 // no terminal edge is a state an attempt can be stranded in -- which is
2081 // how two failed JIT requests wedged a `host_capacity: 2` host into
2082 // starting zero runners, with no error and no cleanup path.
2083 for state in AttemptState::ALL {
2084 if state.is_terminal() {
2085 continue;
2086 }
2087 // Read through `active_count`, which is the function the
2088 // reconciliation formula actually calls, rather than through
2089 // `state.counts_against_capacity()`. The latter is defined as
2090 // `!self.is_terminal()`, so asserting it one line under
2091 // `if state.is_terminal() { continue; }` is `assert!(true)` and
2092 // catches nothing; this asserts the same property against a second
2093 // reader that can regress on its own -- an inverted or mistyped
2094 // filter predicate in `active_count` fails here and nowhere else in
2095 // this test.
2096 assert_eq!(
2097 active_count([&attempt_in(state, 0)]),
2098 1,
2099 "{state} is non-terminal, so the reconciliation formula must \
2100 still be subtracting it"
2101 );
2102 assert!(
2103 state.can_transition_to(AttemptState::Failed)
2104 || state.can_transition_to(AttemptState::Finished),
2105 "{state} has no way to conclude, so an attempt in it holds a host \
2106 capacity slot permanently"
2107 );
2108 assert!(
2109 state.can_transition_to(AttemptState::Orphaned),
2110 "{state} has no orphan edge, so an attempt found in it after a \
2111 restart cannot be recorded"
2112 );
2113 assert!(
2114 AttemptState::CONCLUDABLE_FROM.contains(&state),
2115 "{state} is live but missing from CONCLUDABLE_FROM"
2116 );
2117 }
2118 assert_eq!(
2119 AttemptState::CONCLUDABLE_FROM.len(),
2120 AttemptState::ALL
2121 .iter()
2122 .filter(|s| !s.is_terminal())
2123 .count(),
2124 "CONCLUDABLE_FROM must list every non-terminal state and nothing else"
2125 );
2126 }
2127
2128 #[test]
2129 fn an_attempt_state_cannot_transition_to_itself() {
2130 for state in AttemptState::ALL {
2131 assert!(
2132 !state.can_transition_to(state),
2133 "{state} -> {state} is not an edge in the diagram"
2134 );
2135 }
2136 }
2137
2138 #[test]
2139 fn cleaned_is_absorbing() {
2140 for to in AttemptState::ALL {
2141 assert!(
2142 !AttemptState::Cleaned.can_transition_to(to),
2143 "cleaned -> {to} must not exist; a cleaned runtime directory is gone"
2144 );
2145 }
2146 }
2147
2148 #[test]
2149 fn the_documented_happy_path_walks_allocated_to_cleaned() {
2150 // `e3`: "A full attempt runs allocated -> jit_received -> starting ->
2151 // busy -> finished -> cleaned".
2152 let mut attempt = RunnerAttempt::allocate(
2153 AttemptId::from_u128(1),
2154 PolicyId::from_u128(1),
2155 "runtime/p/a",
2156 ts(0),
2157 );
2158 assert_eq!(attempt.state(), AttemptState::Allocated);
2159 assert!(attempt.counts_against_capacity());
2160
2161 attempt.jit_received(ts(1)).unwrap();
2162 attempt.started(4242, ts(2)).unwrap();
2163 assert_eq!(attempt.process_id(), Some(4242));
2164
2165 attempt.assigned_job(73, ts(3)).unwrap();
2166 assert_eq!(attempt.state(), AttemptState::Busy);
2167 assert_eq!(attempt.github_runner_id(), Some(73));
2168
2169 attempt
2170 .conclude(AttemptOutcome::CompletedJob, ts(9))
2171 .unwrap();
2172 assert_eq!(attempt.state(), AttemptState::Finished);
2173 assert_eq!(attempt.terminal_at(), Some(ts(9)));
2174 assert!(!attempt.counts_against_capacity());
2175
2176 attempt.clean(ts(10)).unwrap();
2177 assert_eq!(attempt.state(), AttemptState::Cleaned);
2178 }
2179
2180 // =======================================================================
2181 // Outcome: the surplus attempt
2182 // =======================================================================
2183
2184 #[test]
2185 fn an_attempt_that_exits_idle_without_work_is_terminal_and_not_a_failure() {
2186 // `b1`: "An attempt that exits idle without work reaches a terminal state
2187 // carrying an outcome distinguishable from a failure."
2188 let mut surplus = RunnerAttempt::allocate(
2189 AttemptId::from_u128(1),
2190 PolicyId::from_u128(1),
2191 "runtime/p/a",
2192 ts(0),
2193 );
2194 surplus.jit_received(ts(1)).unwrap();
2195 surplus.started(1, ts(2)).unwrap();
2196 surplus.registered_idle(73, ts(3)).unwrap();
2197 surplus
2198 .conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(300))
2199 .unwrap();
2200
2201 assert!(surplus.is_terminal());
2202 assert_eq!(surplus.state(), AttemptState::Finished);
2203
2204 let outcome = surplus.outcome().expect("a terminal attempt carries one");
2205 assert!(outcome.is_idle_exit());
2206 assert!(
2207 !outcome.is_failure(),
2208 "the surplus runner is an accepted, bounded cost of having no job \
2209 reservation -- presenting it as a fault sends an operator hunting \
2210 something that did not happen"
2211 );
2212 assert!(!outcome.ran_a_job());
2213
2214 // And it is cleaned like any other terminal attempt.
2215 surplus.clean(ts(301)).unwrap();
2216 assert_eq!(surplus.state(), AttemptState::Cleaned);
2217 }
2218
2219 #[test]
2220 fn an_idle_exit_is_distinguishable_from_a_failure_and_from_a_completed_job() {
2221 // The three outcomes `g2` must render apart.
2222 let idle = AttemptOutcome::ExitedIdleWithoutWork;
2223 let failed = AttemptOutcome::failed(FailureReason::ProcessStartFailed);
2224 let done = AttemptOutcome::CompletedJob;
2225
2226 assert_ne!(idle, failed);
2227 assert_ne!(idle, done);
2228 assert_ne!(failed, done);
2229
2230 assert_eq!(idle.terminal_state(), AttemptState::Finished);
2231 assert_eq!(done.terminal_state(), AttemptState::Finished);
2232 assert_eq!(failed.terminal_state(), AttemptState::Failed);
2233 assert_eq!(
2234 AttemptOutcome::Orphaned.terminal_state(),
2235 AttemptState::Orphaned
2236 );
2237
2238 // `finished` alone does not say which happened; the outcome does. This is
2239 // the assertion that would fail if someone dropped the outcome field and
2240 // let `g2` infer from the state.
2241 assert_eq!(idle.terminal_state(), done.terminal_state());
2242 assert_ne!(idle, done);
2243
2244 assert!(!idle.is_failure());
2245 assert!(failed.is_failure());
2246 assert!(AttemptOutcome::Orphaned.is_failure());
2247 }
2248
2249 #[test]
2250 fn an_outcome_cannot_be_recorded_from_a_state_that_could_not_produce_it() {
2251 // A runner that never got a job cannot have run one.
2252 let mut idle = attempt_in(AttemptState::Idle, 0);
2253 assert!(matches!(
2254 idle.conclude(AttemptOutcome::CompletedJob, ts(1)),
2255 Err(AttemptError::OutcomeUnreachable {
2256 from: AttemptState::Idle,
2257 ..
2258 })
2259 ));
2260 assert_eq!(idle.state(), AttemptState::Idle);
2261 assert!(idle.outcome().is_none());
2262
2263 // And a runner that was executing a job did not exit idle without work.
2264 let mut busy = attempt_in(AttemptState::Busy, 0);
2265 assert!(matches!(
2266 busy.conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(1)),
2267 Err(AttemptError::OutcomeUnreachable {
2268 from: AttemptState::Busy,
2269 ..
2270 })
2271 ));
2272
2273 // Failure and orphaning are reachable from every live state, including
2274 // the three pre-registration ones the amendment gave terminal edges.
2275 for state in AttemptState::CONCLUDABLE_FROM {
2276 attempt_in(*state, 0)
2277 .conclude(AttemptOutcome::failed(FailureReason::JitExpired), ts(1))
2278 .unwrap_or_else(|e| panic!("{state} must be able to fail: {e}"));
2279 attempt_in(*state, 0)
2280 .conclude(AttemptOutcome::Orphaned, ts(1))
2281 .unwrap_or_else(|e| panic!("{state} must be able to orphan: {e}"));
2282 }
2283
2284 // And from a terminal state nothing can be concluded at all -- a second
2285 // conclusion would overwrite the first.
2286 for state in AttemptState::ALL.iter().filter(|s| s.is_terminal()) {
2287 assert!(
2288 attempt_in(*state, 0)
2289 .conclude(AttemptOutcome::Orphaned, ts(1))
2290 .is_err(),
2291 "{state} has already concluded"
2292 );
2293 }
2294 }
2295
2296 /// The earliest state at which the agent learns each fact, from flow 2's own
2297 /// step order.
2298 ///
2299 /// **This match is exhaustive and carries no wildcard, and that is the whole
2300 /// mechanism.** The assertion it replaced read
2301 /// `assert_eq!(cases.len(), 7, "FailureReason has seven variants; ...")`,
2302 /// and `cases.len()` on a `[_; 7]` is the compile-time constant `7`: the
2303 /// assertion was `7 == 7` and could not fail. Measured on the code as it
2304 /// stood, adding an eighth variant produced exactly one error — `E0004` from
2305 /// `Display`'s match — and once that arm was written the suite was green
2306 /// with the new variant unreachable and untested. Here, a tenth variant
2307 /// stops this file compiling until somebody says which state produces it,
2308 /// and `all_failure_reasons_are_reachable_from_the_state_that_produces_them`
2309 /// then proves the answer.
2310 ///
2311 /// Re-measured on the ninth. Declaring
2312 /// `TerminatedAfterRegistrationTimeout` and changing nothing else produced
2313 /// exactly one error, `E0004` at `Display`'s match; writing that arm
2314 /// produced exactly one more, `E0004` here; writing this one left
2315 /// `cargo check --all-targets --workspace` clean. Two stops, in that order,
2316 /// and no third — which is what the note above each of them promises.
2317 fn earliest_state_producing(reason: &FailureReason) -> AttemptState {
2318 // The second place a new variant stops the compiler, and the last one.
2319 // `FailureReason::ALL` is a const array, so it errors nowhere at all:
2320 // add the variant to `ALL` and to the `cases` table below too, or the
2321 // suite goes green with it untested.
2322 match reason {
2323 // Step 5: the package is verified before the JIT request is made.
2324 FailureReason::RunnerPackageUnverified => AttemptState::Allocated,
2325 // Step 5: `generate-jitconfig` did not return a configuration.
2326 FailureReason::JitRequestFailed => AttemptState::Allocated,
2327 // Flow 4.4: a configuration arrived and was never claimed.
2328 FailureReason::JitExpired => AttemptState::JitReceived,
2329 // Step 6: the child process could not be spawned.
2330 FailureReason::ProcessStartFailed => AttemptState::JitReceived,
2331 // Edge case 7: GitHub refuses the registration on version grounds.
2332 FailureReason::RunnerVersionRejected => AttemptState::Starting,
2333 // Flow 2's "runner exit before job acceptance".
2334 FailureReason::ProcessExitedUnexpectedly => AttemptState::Starting,
2335 // The live-but-unregistered process past its startup deadline; see
2336 // the `S::Starting` / `G::NotRegistered` arm of `recovery_decision`.
2337 FailureReason::RegistrationTimedOut => AttemptState::Starting,
2338 // The same attempt one step later: still `starting`, because the
2339 // agent acts on the `Terminate` payload without moving the state
2340 // first, and the process it signalled is the one that never
2341 // registered.
2342 FailureReason::TerminatedAfterRegistrationTimeout => AttemptState::Starting,
2343 FailureReason::IsolationProvider(_) => AttemptState::Allocated,
2344 // `WorkspaceCleanupDeferred` is never recorded as an outcome (see
2345 // the variant), so the only claim made for it is the one `Other`
2346 // makes: the type would accept it.
2347 FailureReason::WorkspaceCleanupDeferred
2348 | FailureReason::HostUnfit
2349 | FailureReason::Other(_) => AttemptState::Busy,
2350 }
2351 }
2352
2353 #[test]
2354 fn all_failure_reasons_are_reachable_from_the_state_that_produces_them() {
2355 // `03-control-flows.md` flow 2 names most of these by name as conditions
2356 // the agent must record. Before the diagram amendment five were
2357 // unreachable: each occurs at a pre-registration state, and `allocated`,
2358 // `jit_received` and `starting` had no terminal edge, so `conclude`
2359 // answered `OutcomeUnreachable` from every state that could actually
2360 // have produced them. A `FailureReason` variant that nothing can reach
2361 // is a variant `g2` renders a match arm for and no test can cover.
2362 //
2363 // The table is written out rather than derived so each pairing carries
2364 // its reason; `earliest_state_producing` above is what makes a new
2365 // variant a compile error, and the two are cross-checked below.
2366 let cases: [(FailureReason, AttemptState); 12] = [
2367 // Step 5: the package is verified before the JIT request is made.
2368 (
2369 FailureReason::RunnerPackageUnverified,
2370 AttemptState::Allocated,
2371 ),
2372 // Step 5: `generate-jitconfig` did not return a configuration.
2373 (FailureReason::JitRequestFailed, AttemptState::Allocated),
2374 // Flow 4.4: a configuration arrived and was never claimed.
2375 (FailureReason::JitExpired, AttemptState::JitReceived),
2376 // Step 6: the child process could not be spawned.
2377 (FailureReason::ProcessStartFailed, AttemptState::JitReceived),
2378 // Edge case 7: GitHub refuses the registration on version grounds.
2379 (FailureReason::RunnerVersionRejected, AttemptState::Starting),
2380 // Flow 2's "runner exit before job acceptance".
2381 (
2382 FailureReason::ProcessExitedUnexpectedly,
2383 AttemptState::Starting,
2384 ),
2385 // A live runner that never reached GitHub inside its startup window.
2386 (FailureReason::RegistrationTimedOut, AttemptState::Starting),
2387 // The same runner after `e3` acted on the `Terminate` payload: the
2388 // attempt never left `starting`, so this is where it concludes from.
2389 (
2390 FailureReason::TerminatedAfterRegistrationTimeout,
2391 AttemptState::Starting,
2392 ),
2393 (
2394 FailureReason::IsolationProvider(IsolationProviderFailure::NotInstalled),
2395 AttemptState::Allocated,
2396 ),
2397 (FailureReason::WorkspaceCleanupDeferred, AttemptState::Busy),
2398 (FailureReason::HostUnfit, AttemptState::Busy),
2399 (
2400 FailureReason::Other("a reason b1 did not anticipate".into()),
2401 AttemptState::Busy,
2402 ),
2403 ];
2404
2405 // Every variant is covered exactly once. Three separate things have to
2406 // hold, and none of them is a length compared against itself:
2407 //
2408 // * `FailureReason::ALL` and this table are the same size, so a variant
2409 // added to one and not the other is caught;
2410 // * every variant in `ALL` appears here -- by discriminant, because
2411 // `Other`'s payload differs between the two lists;
2412 // * each pairing agrees with `earliest_state_producing`, whose
2413 // wildcard-free match is what stops this file compiling when a
2414 // variant is added at all.
2415 assert_eq!(
2416 cases.len(),
2417 FailureReason::ALL.len(),
2418 "every FailureReason variant needs a state it is reachable from"
2419 );
2420 for listed in FailureReason::ALL {
2421 assert!(
2422 cases.iter().any(|(reason, _)| {
2423 std::mem::discriminant(reason) == std::mem::discriminant(&listed)
2424 }),
2425 "{listed:?} is in FailureReason::ALL but has no case here"
2426 );
2427 }
2428 for (reason, from) in &cases {
2429 assert_eq!(
2430 earliest_state_producing(reason),
2431 *from,
2432 "{reason:?}: the table and the exhaustive match disagree"
2433 );
2434 }
2435
2436 for (reason, from) in cases {
2437 let mut attempt = attempt_in(from, 0);
2438 attempt
2439 .conclude(AttemptOutcome::failed(reason.clone()), ts(5))
2440 .unwrap_or_else(|e| panic!("{reason:?} must be recordable from {from}, got {e}"));
2441
2442 assert_eq!(attempt.state(), AttemptState::Failed);
2443 assert_eq!(attempt.terminal_at(), Some(ts(5)));
2444 assert_eq!(
2445 attempt.outcome(),
2446 Some(&AttemptOutcome::failed(reason.clone()))
2447 );
2448 assert!(
2449 !attempt.counts_against_capacity(),
2450 "{reason:?} from {from} must give the host capacity slot back; \
2451 that is what the amendment was for"
2452 );
2453
2454 // And it survives the persistence gate, so the row `b2` writes for
2455 // it can be read back.
2456 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2457 .unwrap_or_else(|e| panic!("{reason:?} from {from} must reload: {e}"));
2458 assert_eq!(restored, attempt);
2459 }
2460 }
2461
2462 #[test]
2463 fn isolation_failure_categories_round_trip_without_free_text() {
2464 let categories = [
2465 IsolationProviderFailure::NotInstalled,
2466 IsolationProviderFailure::PermissionDenied,
2467 IsolationProviderFailure::ImageUnavailableOrIncompatible,
2468 IsolationProviderFailure::DiskQuotaUnavailable,
2469 IsolationProviderFailure::RuntimeOperationFailed,
2470 ];
2471 let mut rendered = std::collections::BTreeSet::new();
2472 for category in categories {
2473 let reason = FailureReason::IsolationProvider(category);
2474 let journal = serde_json::to_string(&reason).unwrap();
2475 assert!(!journal.contains("other"));
2476 assert!(!journal.contains("ghp_secret"));
2477 assert_eq!(
2478 serde_json::from_str::<FailureReason>(&journal).unwrap(),
2479 reason
2480 );
2481 rendered.insert(reason.to_string());
2482 }
2483 assert_eq!(rendered.len(), categories.len());
2484 }
2485
2486 // =======================================================================
2487 // Busy protection
2488 // =======================================================================
2489
2490 #[test]
2491 fn a_busy_attempt_cannot_be_cleaned() {
2492 // `04-subsystem-contracts.md`: "`busy` cannot transition to cleanup due
2493 // to a scale-down request."
2494 let mut busy = attempt_in(AttemptState::Busy, 0);
2495 let err = busy.clean(ts(1)).unwrap_err();
2496 assert_eq!(
2497 err,
2498 AttemptError::BusyCannotBeCleaned,
2499 "the refusal must be named, not a generic transition error, so that a \
2500 scale-down that tried it is legible in a log"
2501 );
2502 assert_eq!(busy.state(), AttemptState::Busy);
2503 assert!(busy.counts_against_capacity());
2504 }
2505
2506 #[test]
2507 fn only_a_terminal_attempt_can_be_cleaned() {
2508 for state in AttemptState::ALL {
2509 let mut attempt = attempt_in(state, 0);
2510 let result = attempt.clean(ts(1));
2511 if state.is_concluded() {
2512 assert!(result.is_ok(), "{state} is terminal and must be cleanable");
2513 assert_eq!(attempt.state(), AttemptState::Cleaned);
2514 } else {
2515 assert!(
2516 result.is_err(),
2517 "{state} is not terminal and must not be cleanable"
2518 );
2519 assert_eq!(attempt.state(), state);
2520 }
2521 }
2522 }
2523
2524 #[test]
2525 fn capacity_is_reclaimed_exactly_at_the_terminal_states() {
2526 for state in AttemptState::ALL {
2527 assert_eq!(
2528 attempt_in(state, 0).counts_against_capacity(),
2529 !state.is_terminal(),
2530 "{state}"
2531 );
2532 }
2533
2534 let attempts = vec![
2535 attempt_in(AttemptState::Allocated, 0),
2536 attempt_in(AttemptState::Starting, 0),
2537 attempt_in(AttemptState::Busy, 0),
2538 attempt_in(AttemptState::Finished, 0),
2539 attempt_in(AttemptState::Cleaned, 0),
2540 ];
2541 assert_eq!(active_count(&attempts), 3);
2542 }
2543
2544 #[test]
2545 fn the_per_policy_active_count_is_not_the_host_wide_one() {
2546 // Substituting either for the other is silent, so the difference is
2547 // pinned rather than left to the reader of two similar names.
2548 let mine = PolicyId::from_u128(1);
2549 let theirs = PolicyId::from_u128(2);
2550
2551 let mut attempts = vec![
2552 attempt_for(mine, AttemptState::Starting, 0),
2553 attempt_for(mine, AttemptState::Busy, 0),
2554 attempt_for(theirs, AttemptState::Busy, 0),
2555 attempt_for(theirs, AttemptState::Idle, 0),
2556 attempt_for(theirs, AttemptState::Allocated, 0),
2557 ];
2558
2559 assert_eq!(active_count(&attempts), 5, "the host-wide total, for D9");
2560 assert_eq!(active_count_for(mine, &attempts), 2, "this policy, for D7");
2561 assert_eq!(active_count_for(theirs, &attempts), 3);
2562 assert_eq!(
2563 active_count_for(PolicyId::from_u128(3), &attempts),
2564 0,
2565 "a policy with nothing in flight"
2566 );
2567
2568 // Terminal attempts drop out of the per-policy count on the same rule as
2569 // the host-wide one.
2570 attempts.push(attempt_for(mine, AttemptState::Finished, 0));
2571 attempts.push(attempt_for(mine, AttemptState::Cleaned, 0));
2572 assert_eq!(active_count_for(mine, &attempts), 2);
2573 }
2574
2575 // =======================================================================
2576 // Workspace allocation (D3, D5, D6)
2577 // =======================================================================
2578
2579 fn slot(n: u16) -> AttemptWorkspace {
2580 AttemptWorkspace::persistent_slot(std::num::NonZeroU16::new(n).expect("a positive slot"))
2581 }
2582
2583 /// The unremarkable outcome for a terminal state, so a fixture does not have
2584 /// to restate the state/outcome pairing the loader enforces.
2585 fn terminal_outcome(state: AttemptState) -> AttemptOutcome {
2586 match state {
2587 AttemptState::Failed => {
2588 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly)
2589 }
2590 AttemptState::Orphaned => AttemptOutcome::Orphaned,
2591 _ => AttemptOutcome::CompletedJob,
2592 }
2593 }
2594
2595 #[test]
2596 fn the_ordinary_constructor_still_allocates_a_disposable_workspace() {
2597 // D3: disposable mode remains the default, so every existing caller of
2598 // `allocate` keeps the cleanup behaviour it had.
2599 let attempt = RunnerAttempt::allocate(
2600 AttemptId::from_u128(1),
2601 PolicyId::from_u128(1),
2602 "runtime/p/a",
2603 ts(0),
2604 );
2605 assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
2606 assert_eq!(attempt.workspace().slot(), None);
2607 assert!(!attempt.holds_slot_lease());
2608 assert_eq!(
2609 attempt.to_persisted().workspace_kind,
2610 WorkspaceKind::Ephemeral
2611 );
2612 assert_eq!(attempt.to_persisted().workspace_slot, None);
2613 }
2614
2615 #[test]
2616 fn a_persistent_attempt_journals_the_slot_it_leased() {
2617 let attempt = RunnerAttempt::allocate_in(
2618 AttemptId::from_u128(1),
2619 PolicyId::from_u128(1),
2620 "/srv/rman/acme/s2",
2621 slot(2),
2622 ts(0),
2623 );
2624 assert_eq!(attempt.workspace(), slot(2));
2625 assert_eq!(attempt.workspace().slot_number(), Some(2));
2626 assert_eq!(
2627 attempt.workspace().slot_directory_name().as_deref(),
2628 Some("s2")
2629 );
2630 assert_eq!(attempt.runtime_path(), Path::new("/srv/rman/acme/s2"));
2631 assert!(attempt.holds_slot_lease());
2632 }
2633
2634 #[test]
2635 fn the_workspace_kind_and_slot_do_not_change_after_allocation() {
2636 // `02-target-architecture.md`: "Neither may change after allocation."
2637 // There is no setter, so the property is proved by driving the whole
2638 // lifecycle and reading the value back at every step.
2639 let mut attempt = RunnerAttempt::allocate_in(
2640 AttemptId::from_u128(1),
2641 PolicyId::from_u128(1),
2642 "/srv/rman/acme/s1",
2643 slot(1),
2644 ts(0),
2645 );
2646 attempt
2647 .jit_received(ts(1))
2648 .expect("allocated -> jit_received");
2649 assert_eq!(attempt.workspace(), slot(1));
2650 attempt
2651 .started(4242, ts(2))
2652 .expect("jit_received -> starting");
2653 assert_eq!(attempt.workspace(), slot(1));
2654 attempt
2655 .registered_idle(73, ts(3))
2656 .expect("starting -> idle");
2657 assert_eq!(attempt.workspace(), slot(1));
2658 attempt.assigned_job(73, ts(4)).expect("idle -> busy");
2659 assert_eq!(attempt.workspace(), slot(1));
2660 attempt
2661 .conclude(AttemptOutcome::CompletedJob, ts(5))
2662 .expect("busy -> finished");
2663 assert_eq!(attempt.workspace(), slot(1));
2664 assert!(
2665 attempt.holds_slot_lease(),
2666 "a terminal attempt still holds its slot until cleanup succeeds"
2667 );
2668
2669 attempt.clean(ts(6)).expect("finished -> cleaned");
2670 assert_eq!(attempt.workspace(), slot(1));
2671 assert!(
2672 !attempt.holds_slot_lease(),
2673 "a cleaned attempt releases the slot for reuse"
2674 );
2675 }
2676
2677 #[test]
2678 fn an_uncleaned_terminal_attempt_keeps_its_lease() {
2679 // `04-security-recovery.md`: "Attempt remains not-cleaned and continues
2680 // to hold the slot through the unique lease index".
2681 for state in [
2682 AttemptState::Finished,
2683 AttemptState::Failed,
2684 AttemptState::Orphaned,
2685 ] {
2686 let mut fields = row(state, Some(terminal_outcome(state)));
2687 fields.workspace_kind = WorkspaceKind::Persistent;
2688 fields.workspace_slot = Some(3);
2689 let attempt = RunnerAttempt::from_persisted(fields).expect("a row the domain accepts");
2690 assert!(attempt.holds_slot_lease(), "state {state}");
2691 }
2692 }
2693
2694 #[test]
2695 fn a_workspace_allocation_round_trips_through_the_journal() {
2696 for workspace in [AttemptWorkspace::Ephemeral, slot(1), slot(u16::MAX)] {
2697 let attempt = RunnerAttempt::allocate_in(
2698 AttemptId::from_u128(1),
2699 PolicyId::from_u128(1),
2700 "runtime/p/a",
2701 workspace,
2702 ts(0),
2703 );
2704 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2705 .expect("a row this crate wrote must load");
2706 assert_eq!(restored, attempt);
2707 assert_eq!(restored.workspace(), workspace);
2708 }
2709 }
2710
2711 #[test]
2712 fn a_journal_row_whose_workspace_columns_disagree_is_rejected() {
2713 // The kind decides which cleanup algorithm is legal, so an undecidable
2714 // pair must fail closed rather than fall back to the destructive branch.
2715 let mut persistent_without_slot = row(AttemptState::Allocated, None);
2716 persistent_without_slot.workspace_kind = WorkspaceKind::Persistent;
2717 assert_eq!(
2718 RunnerAttempt::from_persisted(persistent_without_slot),
2719 Err(AttemptError::Workspace(
2720 WorkspaceError::PersistentWithoutSlot
2721 ))
2722 );
2723
2724 let mut zero_slot = row(AttemptState::Allocated, None);
2725 zero_slot.workspace_kind = WorkspaceKind::Persistent;
2726 zero_slot.workspace_slot = Some(0);
2727 assert_eq!(
2728 RunnerAttempt::from_persisted(zero_slot),
2729 Err(AttemptError::Workspace(WorkspaceError::SlotNotPositive))
2730 );
2731
2732 let mut ephemeral_with_slot = row(AttemptState::Allocated, None);
2733 ephemeral_with_slot.workspace_slot = Some(1);
2734 assert_eq!(
2735 RunnerAttempt::from_persisted(ephemeral_with_slot),
2736 Err(AttemptError::Workspace(WorkspaceError::EphemeralWithSlot {
2737 slot: 1
2738 }))
2739 );
2740 }
2741
2742 #[test]
2743 fn an_attempt_serialises_its_workspace_without_credentials() {
2744 let attempt = RunnerAttempt::allocate_in(
2745 AttemptId::from_u128(1),
2746 PolicyId::from_u128(1),
2747 "/srv/rman/acme/s2",
2748 slot(2),
2749 ts(0),
2750 );
2751 let encoded = serde_json::to_string(&attempt).expect("serialisable");
2752 let decoded: RunnerAttempt = serde_json::from_str(&encoded).expect("deserialisable");
2753 assert_eq!(decoded, attempt);
2754 for needle in ["token", "secret", "jitconfig", "password"] {
2755 assert!(
2756 !encoded.to_ascii_lowercase().contains(needle),
2757 "an attempt leaked {needle:?}: {encoded}"
2758 );
2759 }
2760 }
2761
2762 // =======================================================================
2763 // Persistence gate
2764 // =======================================================================
2765
2766 /// A journal row the domain accepts, for a test to spoil one field of.
2767 fn row(state: AttemptState, outcome: Option<AttemptOutcome>) -> PersistedAttempt {
2768 PersistedAttempt {
2769 id: AttemptId::from_u128(1),
2770 policy_id: PolicyId::from_u128(1),
2771 github_runner_id: Some(73),
2772 state,
2773 outcome,
2774 process_id: Some(9),
2775 execution: AttemptExecution::Native {
2776 process_id: Some(9),
2777 },
2778 runtime_path: "runtime/p/a".into(),
2779 workspace_kind: WorkspaceKind::Ephemeral,
2780 workspace_slot: None,
2781 created_at: ts(0),
2782 terminal_at: state.is_terminal().then(|| ts(9)),
2783 last_state_change_at: ts(9),
2784 }
2785 }
2786
2787 #[test]
2788 fn a_hand_edited_journal_row_with_an_impossible_outcome_is_rejected() {
2789 assert!(
2790 RunnerAttempt::from_persisted(row(
2791 AttemptState::Finished,
2792 Some(AttemptOutcome::ExitedIdleWithoutWork)
2793 ))
2794 .is_ok()
2795 );
2796
2797 // Terminal with no outcome.
2798 assert!(matches!(
2799 RunnerAttempt::from_persisted(row(AttemptState::Failed, None)),
2800 Err(AttemptError::TerminalWithoutOutcome { .. })
2801 ));
2802
2803 // Non-terminal carrying one.
2804 assert!(matches!(
2805 RunnerAttempt::from_persisted(row(
2806 AttemptState::Busy,
2807 Some(AttemptOutcome::CompletedJob)
2808 )),
2809 Err(AttemptError::NonTerminalWithOutcome { .. })
2810 ));
2811
2812 // A `failed` row that claims to have run a job.
2813 assert!(matches!(
2814 RunnerAttempt::from_persisted(row(
2815 AttemptState::Failed,
2816 Some(AttemptOutcome::CompletedJob)
2817 )),
2818 Err(AttemptError::OutcomeStateMismatch { .. })
2819 ));
2820
2821 // `cleaned` keeps whichever outcome preceded it.
2822 assert!(
2823 RunnerAttempt::from_persisted(row(
2824 AttemptState::Cleaned,
2825 Some(AttemptOutcome::Orphaned)
2826 ))
2827 .is_ok()
2828 );
2829 }
2830
2831 #[test]
2832 fn a_hand_edited_journal_row_with_an_impossible_terminal_at_is_rejected() {
2833 // `terminal_at` is on the same footing as `outcome`: `conclude` is the
2834 // only writer of either and sets them together, so a row where
2835 // `state.is_terminal()` and `terminal_at.is_some()` disagree is one this
2836 // crate cannot have produced. Before this gate existed, a `finished` row
2837 // with `terminal_at: None` loaded cleanly and every reader of
2838 // `terminal_at()` saw an attempt that had never concluded.
2839 let mut terminal_without = row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
2840 terminal_without.terminal_at = None;
2841 assert!(
2842 matches!(
2843 RunnerAttempt::from_persisted(terminal_without),
2844 Err(AttemptError::TerminalWithoutTimestamp {
2845 state: AttemptState::Finished
2846 })
2847 ),
2848 "a terminal row with no terminal_at must be refused, exactly as a \
2849 terminal row with no outcome is"
2850 );
2851
2852 // The other direction: a live attempt that claims to have concluded.
2853 let mut live_with = row(AttemptState::Busy, None);
2854 live_with.terminal_at = Some(ts(9));
2855 assert!(matches!(
2856 RunnerAttempt::from_persisted(live_with),
2857 Err(AttemptError::NonTerminalWithTimestamp {
2858 state: AttemptState::Busy
2859 })
2860 ));
2861
2862 // `cleaned` follows a concluded state, so it keeps that state's
2863 // timestamp and is not a special case.
2864 let mut cleaned = row(AttemptState::Cleaned, Some(AttemptOutcome::Orphaned));
2865 cleaned.terminal_at = Some(ts(4));
2866 assert!(RunnerAttempt::from_persisted(cleaned).is_ok());
2867 }
2868
2869 #[test]
2870 fn a_hand_edited_journal_row_whose_timestamps_run_backwards_is_rejected() {
2871 // Presence was already gated; ordering was not, so a row saying an
2872 // attempt concluded a hundred seconds before it was created loaded
2873 // cleanly and every duration derived from the pair came out negative.
2874 // The same hand-edited-journal threat model that motivates the presence
2875 // gate covers this, and `b2` is the task that will meet it.
2876 let mut concluded_before_created =
2877 row(AttemptState::Finished, Some(AttemptOutcome::CompletedJob));
2878 concluded_before_created.created_at = ts(100);
2879 concluded_before_created.last_state_change_at = ts(100);
2880 concluded_before_created.terminal_at = Some(ts(0));
2881 assert_eq!(
2882 RunnerAttempt::from_persisted(concluded_before_created),
2883 Err(AttemptError::TimestampsOutOfOrder {
2884 state: AttemptState::Finished,
2885 field: "terminal_at",
2886 created_at: ts(100),
2887 found: ts(0),
2888 })
2889 );
2890
2891 // The same for the timestamp every recovery timeout is measured from. A
2892 // `last_state_change_at` before `created_at` makes `now - it` larger
2893 // than the attempt's whole life, so every timeout reads as expired.
2894 let mut changed_before_created = row(AttemptState::Busy, None);
2895 changed_before_created.created_at = ts(100);
2896 changed_before_created.last_state_change_at = ts(0);
2897 assert_eq!(
2898 RunnerAttempt::from_persisted(changed_before_created),
2899 Err(AttemptError::TimestampsOutOfOrder {
2900 state: AttemptState::Busy,
2901 field: "last_state_change_at",
2902 created_at: ts(100),
2903 found: ts(0),
2904 })
2905 );
2906
2907 // Equal is not out of order: `allocate` writes the same instant to both,
2908 // so the very first row of every attempt has `created_at ==
2909 // last_state_change_at`, and an attempt concluded in the same second it
2910 // was allocated has all three equal.
2911 let mut same_instant = row(AttemptState::Failed, Some(AttemptOutcome::Orphaned));
2912 same_instant.outcome = Some(AttemptOutcome::failed(FailureReason::JitRequestFailed));
2913 same_instant.created_at = ts(7);
2914 same_instant.last_state_change_at = ts(7);
2915 same_instant.terminal_at = Some(ts(7));
2916 assert!(RunnerAttempt::from_persisted(same_instant).is_ok());
2917
2918 // And a well-ordered row is untouched by the new arm.
2919 assert!(
2920 RunnerAttempt::from_persisted(row(
2921 AttemptState::Finished,
2922 Some(AttemptOutcome::CompletedJob)
2923 ))
2924 .is_ok()
2925 );
2926 }
2927
2928 #[test]
2929 fn an_attempt_round_trips_through_its_persisted_form() {
2930 let mut attempt = RunnerAttempt::allocate(
2931 AttemptId::from_u128(3),
2932 PolicyId::from_u128(4),
2933 "runtime/p/a",
2934 ts(0),
2935 );
2936 attempt.jit_received(ts(1)).unwrap();
2937 attempt.started(7, ts(2)).unwrap();
2938 attempt.registered_idle(7, ts(3)).unwrap();
2939 attempt
2940 .conclude(AttemptOutcome::ExitedIdleWithoutWork, ts(4))
2941 .unwrap();
2942
2943 let restored = RunnerAttempt::from_persisted(attempt.to_persisted())
2944 .expect("a row this crate produced must load");
2945 assert_eq!(restored, attempt);
2946 assert_eq!(restored.terminal_at(), Some(ts(4)));
2947 assert_ne!(
2948 restored.last_state_change_at(),
2949 restored.created_at,
2950 "the two timestamps must not collapse onto each other; that \
2951 transposition is what PersistedAttempt exists to prevent"
2952 );
2953 }
2954
2955 #[test]
2956 fn an_attempt_round_trips_through_serde() {
2957 let mut attempt = RunnerAttempt::allocate(
2958 AttemptId::from_u128(3),
2959 PolicyId::from_u128(4),
2960 "runtime/p/a",
2961 ts(0),
2962 );
2963 attempt.jit_received(ts(1)).unwrap();
2964 attempt.started(7, ts(2)).unwrap();
2965 attempt.registered_idle(73, ts(3)).unwrap();
2966 attempt
2967 .conclude(
2968 AttemptOutcome::failed(FailureReason::Other("no detail".into())),
2969 ts(4),
2970 )
2971 .unwrap();
2972
2973 let json = serde_json::to_string(&attempt).unwrap();
2974 let back: RunnerAttempt = serde_json::from_str(&json).unwrap();
2975 assert_eq!(attempt, back);
2976 }
2977
2978 // =======================================================================
2979 // Ownership
2980 // =======================================================================
2981
2982 #[test]
2983 fn an_attempt_belonging_to_another_host_is_rejected() {
2984 // Ownership rule 2: "A host agent may act only on attempts persisted
2985 // under its `host_id`."
2986 let mine = HostId::from_u128(7);
2987 let theirs = HostId::from_u128(8);
2988 let policy_id = PolicyId::from_u128(11);
2989
2990 let their_policy = a_policy(theirs, policy_id);
2991 let attempt =
2992 RunnerAttempt::allocate(AttemptId::from_u128(1), policy_id, "runtime/p/a", ts(0));
2993
2994 let err = authorize(mine, &their_policy, &attempt).unwrap_err();
2995 assert!(
2996 matches!(
2997 err,
2998 OwnershipError::ForeignHost {
2999 owner,
3000 agent,
3001 ..
3002 } if owner == theirs && agent == mine
3003 ),
3004 "got {err:?}"
3005 );
3006
3007 // The same attempt under our own policy is fine.
3008 let my_policy = a_policy(mine, policy_id);
3009 assert!(authorize(mine, &my_policy, &attempt).is_ok());
3010 }
3011
3012 #[test]
3013 fn an_attempt_checked_against_the_wrong_policy_is_rejected() {
3014 let host = HostId::from_u128(7);
3015 let policy = a_policy(host, PolicyId::from_u128(11));
3016 let attempt = RunnerAttempt::allocate(
3017 AttemptId::from_u128(1),
3018 PolicyId::from_u128(12),
3019 "runtime/p/a",
3020 ts(0),
3021 );
3022 assert!(matches!(
3023 authorize(host, &policy, &attempt),
3024 Err(OwnershipError::PolicyMismatch { .. })
3025 ));
3026 }
3027
3028 // =======================================================================
3029 // Recovery, against a controlled clock
3030 // =======================================================================
3031
3032 #[test]
3033 fn recovery_never_reads_the_system_clock() {
3034 // The whole decision surface moves when the fake clock moves, and by
3035 // nothing else. If any branch called `Utc::now()` the two decisions below
3036 // would be identical.
3037 let timeouts = RecoveryTimeouts::new(
3038 Elapsed::seconds(60),
3039 Elapsed::seconds(120),
3040 Elapsed::seconds(300),
3041 );
3042 let attempt = attempt_in(AttemptState::Idle, 1_000);
3043 let observation = RecoveryObservation {
3044 process_alive: false,
3045 github: GithubRunnerObservation::NotRegistered,
3046 };
3047
3048 let clock = StubClock::at(1_000 + 299);
3049 assert_eq!(
3050 recovery_decision(&attempt, observation, timeouts, &clock),
3051 RecoveryDecision::Conclude(AttemptOutcome::failed(
3052 FailureReason::ProcessExitedUnexpectedly
3053 )),
3054 "one second before its idle timeout, a vanished runner crashed"
3055 );
3056
3057 clock.set(1_000 + 300);
3058 assert_eq!(
3059 recovery_decision(&attempt, observation, timeouts, &clock),
3060 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
3061 "at its idle timeout, the same runner is the surplus case from flow 2.7"
3062 );
3063 }
3064
3065 #[test]
3066 fn a_live_process_is_adopted_rather_than_duplicated() {
3067 // `e3`: "an attempt whose process still runs is adopted, not duplicated."
3068 let timeouts = RecoveryTimeouts::provisional();
3069 let clock = StubClock::at(1_000_000);
3070 for state in [
3071 AttemptState::Allocated,
3072 AttemptState::JitReceived,
3073 AttemptState::Starting,
3074 AttemptState::Idle,
3075 AttemptState::Busy,
3076 ] {
3077 // `entered_at` is the clock's own instant, so every state is inside
3078 // its window and the only thing being asserted is the live-process
3079 // rule.
3080 let attempt = attempt_in(state, 1_000_000);
3081 assert_eq!(
3082 recovery_decision(
3083 &attempt,
3084 RecoveryObservation {
3085 process_alive: true,
3086 github: GithubRunnerObservation::NotRegistered,
3087 },
3088 timeouts,
3089 &clock,
3090 ),
3091 RecoveryDecision::Adopt,
3092 "{state} with a live process"
3093 );
3094 }
3095 }
3096
3097 #[test]
3098 fn a_dead_busy_attempt_is_orphaned_rather_than_guessed_into_a_success() {
3099 // The agent never reports a job as complete; GitHub remains the source of
3100 // truth for workflow outcome (flow 2, Failure).
3101 let clock = StubClock::at(1_000_000);
3102 for github in [
3103 GithubRunnerObservation::NotRegistered,
3104 GithubRunnerObservation::Registered { busy: true },
3105 GithubRunnerObservation::Registered { busy: false },
3106 ] {
3107 assert_eq!(
3108 recovery_decision(
3109 &attempt_in(AttemptState::Busy, 0),
3110 RecoveryObservation {
3111 process_alive: false,
3112 github,
3113 },
3114 RecoveryTimeouts::provisional(),
3115 &clock,
3116 ),
3117 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3118 "{github:?}"
3119 );
3120 }
3121 }
3122
3123 #[test]
3124 fn an_unreachable_github_defers_every_decision() {
3125 // Flow 3.3: while offline, start nothing and retain what is running. An
3126 // agent that treated "unreachable" as "not registered" would conclude and
3127 // clean every live attempt during a network outage.
3128 let clock = StubClock::at(1_000_000);
3129 for state in [
3130 AttemptState::Allocated,
3131 AttemptState::JitReceived,
3132 AttemptState::Starting,
3133 AttemptState::Idle,
3134 AttemptState::Busy,
3135 ] {
3136 assert_eq!(
3137 recovery_decision(
3138 &attempt_in(state, 0),
3139 RecoveryObservation {
3140 process_alive: false,
3141 github: GithubRunnerObservation::Unreachable,
3142 },
3143 RecoveryTimeouts::provisional(),
3144 &clock,
3145 ),
3146 RecoveryDecision::Defer,
3147 "{state} while GitHub is unreachable"
3148 );
3149 }
3150 }
3151
3152 #[test]
3153 fn a_starting_attempt_follows_what_github_reports() {
3154 // Precedence rule 3: GitHub runner status is authoritative for remote job
3155 // status. Both edges are in the diagram.
3156 let clock = StubClock::at(1_000);
3157 let attempt = attempt_in(AttemptState::Starting, 0);
3158
3159 assert_eq!(
3160 recovery_decision(
3161 &attempt,
3162 RecoveryObservation {
3163 process_alive: true,
3164 github: GithubRunnerObservation::Registered { busy: true },
3165 },
3166 RecoveryTimeouts::provisional(),
3167 &clock,
3168 ),
3169 RecoveryDecision::Observe(AttemptState::Busy)
3170 );
3171 assert_eq!(
3172 recovery_decision(
3173 &attempt,
3174 RecoveryObservation {
3175 process_alive: true,
3176 github: GithubRunnerObservation::Registered { busy: false },
3177 },
3178 RecoveryTimeouts::provisional(),
3179 &clock,
3180 ),
3181 RecoveryDecision::Observe(AttemptState::Idle)
3182 );
3183 }
3184
3185 #[test]
3186 fn a_pre_registration_attempt_believes_github_over_its_own_stale_journal() {
3187 // The crash window `e3` exists for: the process started and registered
3188 // at GitHub, but the write recording it was lost. GitHub is telling the
3189 // agent the runner is alive.
3190 //
3191 // This branch used to consult only `process_alive` and the clock, so the
3192 // observation below produced
3193 // `NoLegalTransition { from: JitReceived, wanted: Failed }` -- the
3194 // attempt was abandoned as failed while its registration stayed at
3195 // GitHub, never reconciled and never removed. Nothing in the suite
3196 // noticed, because nothing asked.
3197 let timeouts = RecoveryTimeouts::new(
3198 Elapsed::seconds(60),
3199 Elapsed::seconds(120),
3200 Elapsed::seconds(300),
3201 );
3202 // Well past the JIT handoff deadline, so the old code's timeout arm is
3203 // the one being displaced.
3204 let clock = StubClock::at(10_000);
3205
3206 for busy in [true, false] {
3207 assert_eq!(
3208 recovery_decision(
3209 &attempt_in(AttemptState::JitReceived, 0),
3210 RecoveryObservation {
3211 process_alive: false,
3212 github: GithubRunnerObservation::Registered { busy },
3213 },
3214 timeouts,
3215 &clock,
3216 ),
3217 RecoveryDecision::Observe(AttemptState::Starting),
3218 "jit_received + Registered{{busy:{busy}}}: the runner cannot have \
3219 registered without starting, and `jit_received -> starting` is \
3220 an edge the diagram already has"
3221 );
3222
3223 // `allocated` has no `-> starting` edge, so it takes the one legal
3224 // step it does have. The registration proves the JIT configuration
3225 // arrived, which is exactly what that edge records; the next pass
3226 // continues from `jit_received`.
3227 assert_eq!(
3228 recovery_decision(
3229 &attempt_in(AttemptState::Allocated, 0),
3230 RecoveryObservation {
3231 process_alive: false,
3232 github: GithubRunnerObservation::Registered { busy },
3233 },
3234 timeouts,
3235 &clock,
3236 ),
3237 RecoveryDecision::Observe(AttemptState::JitReceived),
3238 "allocated + Registered{{busy:{busy}}}"
3239 );
3240 }
3241
3242 // Unchanged where GitHub knows nothing: the timeout still decides, and a
3243 // live process is still adopted rather than duplicated. What changed at
3244 // the amendment is only what the timeout produces -- a conclusion
3245 // instead of a report that none was expressible.
3246 assert_eq!(
3247 recovery_decision(
3248 &attempt_in(AttemptState::JitReceived, 0),
3249 RecoveryObservation {
3250 process_alive: false,
3251 github: GithubRunnerObservation::NotRegistered,
3252 },
3253 timeouts,
3254 &clock,
3255 ),
3256 RecoveryDecision::Conclude(AttemptOutcome::failed(FailureReason::JitExpired))
3257 );
3258 assert_eq!(
3259 recovery_decision(
3260 &attempt_in(AttemptState::JitReceived, 0),
3261 RecoveryObservation {
3262 process_alive: true,
3263 github: GithubRunnerObservation::Registered { busy: false },
3264 },
3265 timeouts,
3266 &clock,
3267 ),
3268 RecoveryDecision::Adopt,
3269 "precedence rule 3's other half: local process state is \
3270 authoritative for a child process this agent owns"
3271 );
3272 }
3273
3274 #[test]
3275 fn a_terminal_attempt_is_cleaned_and_a_cleaned_one_is_left_alone() {
3276 let clock = StubClock::at(1_000_000);
3277 let observation = RecoveryObservation {
3278 process_alive: false,
3279 github: GithubRunnerObservation::NotRegistered,
3280 };
3281 for state in [
3282 AttemptState::Finished,
3283 AttemptState::Failed,
3284 AttemptState::Orphaned,
3285 ] {
3286 assert_eq!(
3287 recovery_decision(
3288 &attempt_in(state, 0),
3289 observation,
3290 RecoveryTimeouts::provisional(),
3291 &clock
3292 ),
3293 RecoveryDecision::Clean,
3294 "{state}"
3295 );
3296 }
3297 assert_eq!(
3298 recovery_decision(
3299 &attempt_in(AttemptState::Cleaned, 0),
3300 observation,
3301 RecoveryTimeouts::provisional(),
3302 &clock
3303 ),
3304 RecoveryDecision::Nothing
3305 );
3306 }
3307
3308 #[test]
3309 fn a_pre_registration_attempt_waits_until_its_deadline_then_concludes() {
3310 // This test used to document a gap in the state diagram: `allocated`,
3311 // `jit_received` and `starting` had no edge to `failed` or `orphaned`,
3312 // so a dead attempt that never registered could not be concluded at all
3313 // and the decision reported the missing edge instead of inventing one.
3314 // The 2026-08-21 amendment added those edges, and the whole point of
3315 // adding them was that this attempt now gives its capacity slot back.
3316 let timeouts = RecoveryTimeouts::new(
3317 Elapsed::seconds(60),
3318 Elapsed::seconds(120),
3319 Elapsed::seconds(300),
3320 );
3321 let clock = StubClock::at(1_059);
3322 let observation = RecoveryObservation {
3323 process_alive: false,
3324 github: GithubRunnerObservation::NotRegistered,
3325 };
3326
3327 // The two pre-JIT states conclude with different reasons, because they
3328 // know different things: at `allocated` no configuration was ever
3329 // recorded as arriving, at `jit_received` one arrived and was never
3330 // claimed.
3331 for (state, reason) in [
3332 (AttemptState::Allocated, FailureReason::JitRequestFailed),
3333 (AttemptState::JitReceived, FailureReason::JitExpired),
3334 ] {
3335 let attempt = attempt_in(state, 1_000);
3336 assert_eq!(
3337 recovery_decision(&attempt, observation, timeouts, &clock),
3338 RecoveryDecision::Wait,
3339 "{state} inside its handoff window"
3340 );
3341 clock.set(1_060);
3342 assert_eq!(
3343 recovery_decision(&attempt, observation, timeouts, &clock),
3344 RecoveryDecision::Conclude(AttemptOutcome::failed(reason)),
3345 "{state} past its handoff window concludes rather than stranding \
3346 the attempt"
3347 );
3348 clock.set(1_059);
3349 }
3350
3351 // `starting` past its own, longer deadline: flow 2's "runner exit before
3352 // job acceptance".
3353 clock.set(1_121);
3354 assert_eq!(
3355 recovery_decision(
3356 &attempt_in(AttemptState::Starting, 1_000),
3357 observation,
3358 timeouts,
3359 &clock
3360 ),
3361 RecoveryDecision::Conclude(AttemptOutcome::failed(
3362 FailureReason::ProcessExitedUnexpectedly
3363 ))
3364 );
3365
3366 // Every one of those decisions must be applicable to the attempt it was
3367 // made about; a decision the state machine then refuses would strand the
3368 // attempt just as the missing edges did.
3369 for (state, decision) in [
3370 (
3371 AttemptState::Allocated,
3372 AttemptOutcome::failed(FailureReason::JitRequestFailed),
3373 ),
3374 (
3375 AttemptState::JitReceived,
3376 AttemptOutcome::failed(FailureReason::JitExpired),
3377 ),
3378 (
3379 AttemptState::Starting,
3380 AttemptOutcome::failed(FailureReason::ProcessExitedUnexpectedly),
3381 ),
3382 ] {
3383 let mut attempt = attempt_in(state, 1_000);
3384 attempt
3385 .conclude(decision, ts(1_200))
3386 .unwrap_or_else(|e| panic!("recovery decided {state} concludes, but: {e}"));
3387 assert!(!attempt.counts_against_capacity());
3388 }
3389 }
3390
3391 #[test]
3392 fn an_idle_runner_that_github_reports_as_busy_is_recorded_as_busy() {
3393 // The `idle -> busy` half of the amendment, at the point it bites in
3394 // recovery. GitHub is authoritative for remote job status (precedence
3395 // rule 3), and before the amendment this observation was not
3396 // representable at all: the decision reported that no legal transition
3397 // existed and the attempt stayed `idle` for ever while a job ran on it.
3398 let timeouts = RecoveryTimeouts::provisional();
3399 let clock = StubClock::at(1_121);
3400 assert_eq!(
3401 recovery_decision(
3402 &attempt_in(AttemptState::Idle, 1_000),
3403 RecoveryObservation {
3404 process_alive: false,
3405 github: GithubRunnerObservation::Registered { busy: true },
3406 },
3407 timeouts,
3408 &clock,
3409 ),
3410 RecoveryDecision::Observe(AttemptState::Busy)
3411 );
3412
3413 // And the decision is applicable: `idle -> busy` is an edge, so the
3414 // caller can actually carry it out.
3415 let mut attempt = attempt_in(AttemptState::Idle, 1_000);
3416 attempt.assigned_job(73, ts(1_200)).unwrap();
3417 assert_eq!(attempt.state(), AttemptState::Busy);
3418 assert_eq!(attempt.github_runner_id(), Some(73));
3419
3420 // And a live process does not change the answer. GitHub is authoritative
3421 // for remote job status; the caller adopts supervision independently of
3422 // the decision, which is what `Observe`'s own documentation says and
3423 // what `starting` has always relied on.
3424 assert_eq!(
3425 recovery_decision(
3426 &attempt_in(AttemptState::Idle, 1_000),
3427 RecoveryObservation {
3428 process_alive: true,
3429 github: GithubRunnerObservation::Registered { busy: true },
3430 },
3431 timeouts,
3432 &clock,
3433 ),
3434 RecoveryDecision::Observe(AttemptState::Busy),
3435 "a live process must not hide a job GitHub is reporting"
3436 );
3437 }
3438
3439 #[test]
3440 fn a_restart_during_a_job_is_recorded_as_busy_and_not_left_reading_idle() {
3441 // A1. The same conflict -- a live process that GitHub reports as `busy`
3442 // -- must resolve the same way at `idle` as at `starting`, and this test
3443 // is written as the damage rather than as the symmetry, because the
3444 // symmetry is not what it costs to get wrong.
3445 //
3446 // Before the fix, `idle` short-circuited on `process_alive` and answered
3447 // `Adopt`. `Adopt` writes nothing, so the journal stayed `idle` and
3448 // `last_state_change_at` stayed pointed at the idle entry. Ten minutes
3449 // later the runner crashed mid-job, GitHub reaped its registration, and
3450 // the idle timeout had long since elapsed -- so the crash was concluded
3451 // `ExitedIdleWithoutWork`: the benign surplus exit, which `g2` renders
3452 // as a normal end and never alarms an operator about. Nothing
3453 // downstream could refuse it, because `required_from` for that outcome
3454 // is `&[Idle]` and the attempt genuinely was `idle`.
3455 let timeouts = RecoveryTimeouts::new(
3456 Elapsed::seconds(60),
3457 Elapsed::seconds(120),
3458 Elapsed::seconds(300),
3459 );
3460 let clock = StubClock::at(1_000);
3461
3462 // The two arms answer alike, which is the property. `starting` was
3463 // already right; `idle` was not.
3464 let mid_job = RecoveryObservation {
3465 process_alive: true,
3466 github: GithubRunnerObservation::Registered { busy: true },
3467 };
3468 assert_eq!(
3469 recovery_decision(
3470 &attempt_in(AttemptState::Idle, 1_000),
3471 mid_job,
3472 timeouts,
3473 &clock
3474 ),
3475 recovery_decision(
3476 &attempt_in(AttemptState::Starting, 1_000),
3477 mid_job,
3478 timeouts,
3479 &clock
3480 ),
3481 "the same observation must not resolve one way at idle and the \
3482 opposite way one state earlier"
3483 );
3484 assert_eq!(
3485 recovery_decision(
3486 &attempt_in(AttemptState::Idle, 1_000),
3487 mid_job,
3488 timeouts,
3489 &clock
3490 ),
3491 RecoveryDecision::Observe(AttemptState::Busy)
3492 );
3493
3494 // And now the consequence, walked end to end. The caller applies the
3495 // decision -- which `Observe` requires, on pain of repeating for ever --
3496 // and the attempt is `busy` with `last_state_change_at` moved to the
3497 // moment the job was observed.
3498 let mut attempt = attempt_in(AttemptState::Idle, 1_000);
3499 attempt.assigned_job(73, ts(1_000)).expect("idle -> busy");
3500 assert_eq!(attempt.last_state_change_at(), ts(1_000));
3501
3502 // Ten minutes on, the process is gone and GitHub has reaped the runner.
3503 // Well past the idle timeout, and irrelevant: a crash from `busy` is
3504 // `Orphaned`, and `ExitedIdleWithoutWork` is not reachable from `busy`
3505 // at all.
3506 clock.set(1_600);
3507 let crashed = RecoveryObservation {
3508 process_alive: false,
3509 github: GithubRunnerObservation::NotRegistered,
3510 };
3511 let decision = recovery_decision(&attempt, crashed, timeouts, &clock);
3512 assert_eq!(
3513 decision,
3514 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3515 "a crash during a job is a lost supervision, never the surplus exit"
3516 );
3517 let RecoveryDecision::Conclude(outcome) = decision else {
3518 unreachable!("asserted just above")
3519 };
3520 assert!(
3521 outcome.is_failure() && !outcome.is_idle_exit(),
3522 "`g2` reads these two flags to decide whether to alarm an operator, \
3523 and a mid-job crash must alarm one"
3524 );
3525
3526 // The old behaviour, spelled out so the regression is unmistakable: had
3527 // the attempt been left at `idle`, this is what the same crash would
3528 // have produced.
3529 assert_eq!(
3530 recovery_decision(
3531 &attempt_in(AttemptState::Idle, 1_000),
3532 crashed,
3533 timeouts,
3534 &clock
3535 ),
3536 RecoveryDecision::Conclude(AttemptOutcome::ExitedIdleWithoutWork),
3537 "which is exactly why the journal must not be left saying `idle`"
3538 );
3539 }
3540
3541 #[test]
3542 fn a_registered_runner_that_never_gets_a_job_is_stopped_at_its_idle_timeout() {
3543 // The surplus case of flow 2.7, in the shape it actually occurs in:
3544 // GitHub still lists the runner, it is not busy, and the process is
3545 // very much alive -- `Runner.Listener run` long-polls for an assignment
3546 // and has no idle timeout of its own, so it never leaves on its own.
3547 // This arm answered `Adopt` at every elapsed time, which is why a runner
3548 // observed in the field held its slot and its row in the target's runner
3549 // settings for 27 hours.
3550 let timeouts = RecoveryTimeouts::new(
3551 Elapsed::seconds(60),
3552 Elapsed::seconds(120),
3553 Elapsed::seconds(300),
3554 );
3555 let clock = StubClock::at(1_000);
3556 let attempt = attempt_in(AttemptState::Idle, 1_000);
3557 let idle_registered = |alive| RecoveryObservation {
3558 process_alive: alive,
3559 github: GithubRunnerObservation::Registered { busy: false },
3560 };
3561
3562 // Inside the window the runner is still plausibly about to be assigned,
3563 // so it is adopted and nothing is disturbed.
3564 clock.set(1_299);
3565 assert_eq!(
3566 recovery_decision(&attempt, idle_registered(true), timeouts, &clock),
3567 RecoveryDecision::Adopt,
3568 "a runner one second inside its idle timeout may still be given a job"
3569 );
3570
3571 // At the deadline it is surplus, and the agent has to end it.
3572 clock.set(1_300);
3573 let decision = recovery_decision(&attempt, idle_registered(true), timeouts, &clock);
3574 assert_eq!(
3575 decision,
3576 RecoveryDecision::Terminate(AttemptOutcome::ExitedIdleWithoutWork),
3577 "past the idle timeout a registered, unassigned runner is flow 2.7's surplus case"
3578 );
3579
3580 // `Terminate`, not `Conclude`, and the difference is the capacity slot:
3581 // the process is alive, so the slot may not come back until it is gone.
3582 let mut attempt = attempt;
3583 assert!(attempt.counts_against_capacity());
3584 let RecoveryDecision::Terminate(outcome) = decision else {
3585 unreachable!("asserted above")
3586 };
3587 assert!(
3588 outcome.is_idle_exit(),
3589 "the surplus exit is a normal outcome, and `g2` renders it as one -- an operator \
3590 sent to hunt a fault here would find nothing"
3591 );
3592 attempt
3593 .conclude(outcome, ts(1_310))
3594 .expect("the payload must be applicable to the attempt it was made about");
3595 assert!(!attempt.counts_against_capacity());
3596
3597 // A dead process is the pre-existing reading and is left alone: the
3598 // registration outlived supervision, which is what `Orphaned` names.
3599 clock.set(1_400);
3600 assert_eq!(
3601 recovery_decision(
3602 &attempt_in(AttemptState::Idle, 1_000),
3603 idle_registered(false),
3604 timeouts,
3605 &clock
3606 ),
3607 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3608 "the idle deadline governs a live runner; a dead one is orphaned however long it sat"
3609 );
3610
3611 // And a runner GitHub reports busy is never stopped, at any elapsed
3612 // time. This is the assertion that stops the deadline above from ever
3613 // being applied to a runner in the middle of a job.
3614 for offset in [0, 299, 300, 100_000] {
3615 clock.set(1_000 + offset);
3616 assert_eq!(
3617 recovery_decision(
3618 &attempt_in(AttemptState::Idle, 1_000),
3619 RecoveryObservation {
3620 process_alive: true,
3621 github: GithubRunnerObservation::Registered { busy: true },
3622 },
3623 timeouts,
3624 &clock
3625 ),
3626 RecoveryDecision::Observe(AttemptState::Busy),
3627 "at +{offset}s a runner that took a job was stopped by an idle deadline"
3628 );
3629 }
3630 }
3631
3632 #[test]
3633 fn a_live_unregistered_runner_past_its_deadline_is_stopped_before_its_slot_returns() {
3634 // A2. Three cases at `starting` with nothing at GitHub, and the middle
3635 // one used to be folded into the last.
3636 let timeouts = RecoveryTimeouts::new(
3637 Elapsed::seconds(60),
3638 Elapsed::seconds(120),
3639 Elapsed::seconds(300),
3640 );
3641 let clock = StubClock::at(1_000);
3642 let attempt = attempt_in(AttemptState::Starting, 1_000);
3643 let unregistered = |alive| RecoveryObservation {
3644 process_alive: alive,
3645 github: GithubRunnerObservation::NotRegistered,
3646 };
3647
3648 // Inside the window, alive: adopted.
3649 clock.set(1_119);
3650 assert_eq!(
3651 recovery_decision(&attempt, unregistered(true), timeouts, &clock),
3652 RecoveryDecision::Adopt
3653 );
3654
3655 // Gone: an accurate `ProcessExitedUnexpectedly`, and terminal, because
3656 // there is nothing left running to hold the slot.
3657 assert_eq!(
3658 recovery_decision(&attempt, unregistered(false), timeouts, &clock),
3659 RecoveryDecision::Conclude(AttemptOutcome::failed(
3660 FailureReason::ProcessExitedUnexpectedly
3661 )),
3662 "the one case that really is flow 2's runner exit"
3663 );
3664
3665 // Alive and past the deadline: neither of the above.
3666 clock.set(1_120);
3667 let decision = recovery_decision(&attempt, unregistered(true), timeouts, &clock);
3668 assert_eq!(
3669 decision,
3670 RecoveryDecision::Terminate(AttemptOutcome::failed(
3671 FailureReason::RegistrationTimedOut
3672 ))
3673 );
3674 assert_ne!(
3675 decision,
3676 RecoveryDecision::Conclude(AttemptOutcome::failed(
3677 FailureReason::ProcessExitedUnexpectedly
3678 )),
3679 "a process visible in a task manager did not exit unexpectedly, and \
3680 an operator told that it did stops believing the next message too"
3681 );
3682
3683 // The substance: the slot is not returned while the process runs. The
3684 // decision itself moves nothing, so the attempt is still `starting` and
3685 // still counted -- which is what stops the agent starting a replacement
3686 // beside a runner that could still register and take a job.
3687 let mut attempt = attempt;
3688 assert!(attempt.counts_against_capacity());
3689 assert_eq!(active_count([&attempt]), 1);
3690
3691 // The slot comes back at the moment the caller applies the payload,
3692 // which it does only after the process is gone.
3693 let RecoveryDecision::Terminate(outcome) = decision else {
3694 unreachable!("asserted above")
3695 };
3696 attempt
3697 .conclude(outcome, ts(1_130))
3698 .expect("the payload must be applicable to the attempt it was made about");
3699 assert_eq!(attempt.state(), AttemptState::Failed);
3700 assert!(!attempt.counts_against_capacity());
3701 assert_eq!(active_count([&attempt]), 0);
3702
3703 // Half the re-derivation is free: an agent that died between deciding
3704 // and terminating still sees a live process, and is handed the same
3705 // answer next pass.
3706 let pending = attempt_in(AttemptState::Starting, 1_000);
3707 assert_eq!(
3708 recovery_decision(&pending, unregistered(true), timeouts, &clock),
3709 RecoveryDecision::Terminate(AttemptOutcome::failed(
3710 FailureReason::RegistrationTimedOut
3711 ))
3712 );
3713
3714 // The other half is not, and this is the assertion that says so rather
3715 // than blessing it. An agent that terminated the process and died
3716 // before writing the outcome is, on the next pass, indistinguishable
3717 // from one whose runner crashed on its own: both present exactly the
3718 // same `RecoveryObservation`. The decision below is therefore the same
3719 // for both, and for the terminate-then-crash case the reason it carries
3720 // is wrong -- nothing exited unexpectedly, this agent killed it.
3721 let crashed_on_its_own = unregistered(false);
3722 let killed_by_us_then_lost = unregistered(false);
3723 assert_eq!(
3724 crashed_on_its_own, killed_by_us_then_lost,
3725 "this equality *is* the defect: the fact that separates the two \
3726 cases is not an input to `recovery_decision`, which is why it \
3727 cannot be repaired inside it"
3728 );
3729 assert_eq!(
3730 recovery_decision(&pending, killed_by_us_then_lost, timeouts, &clock),
3731 RecoveryDecision::Conclude(AttemptOutcome::failed(
3732 FailureReason::ProcessExitedUnexpectedly
3733 )),
3734 "pinned as the current behaviour of a known-wrong window, not as a \
3735 correct answer: `e3` closes it by journalling terminate-intent \
3736 before signalling and concluding the marked attempt with \
3737 `TerminatedAfterRegistrationTimeout`, which is true of the dead \
3738 process this pass is looking at. `RecoveryDecision::Terminate` \
3739 names the trade and the owner"
3740 );
3741 }
3742
3743 #[test]
3744 fn no_decision_calls_a_dead_process_live() {
3745 // `RegistrationTimedOut` renders as "the runner process is running but
3746 // did not register", `ProcessExitedUnexpectedly` is documented "only for
3747 // a process that is actually gone", and
3748 // `TerminatedAfterRegistrationTimeout` says the agent stopped the
3749 // process. All three are claims about liveness, and
3750 // `RecoveryObservation::process_alive` is the only source of truth for
3751 // it, so each reason may only ever appear beside the observation that
3752 // supports it.
3753 //
3754 // This is also the guard on the `Terminate` window's residual. The
3755 // rejected fix for that window -- reach for `RegistrationTimedOut` in
3756 // the dead-process arm at `starting` once `elapsed >= startup` --
3757 // compiles, and reds three tests. Measured by applying it:
3758 //
3759 // * here, at `starting at +120s with NotRegistered and a process the
3760 // observation says is gone was reported as still running`;
3761 // * `a_live_unregistered_runner_past_its_deadline_is_stopped_before_
3762 // its_slot_returns`, at the assertion that pins the known-wrong
3763 // window as current behaviour;
3764 // * `a_pre_registration_attempt_waits_until_its_deadline_then_
3765 // concludes`, with `left: Conclude(Failed { reason:
3766 // RegistrationTimedOut })` against `right: Conclude(Failed { reason:
3767 // ProcessExitedUnexpectedly })`.
3768 //
3769 // The third is the strongest of the three and is worth naming ahead of
3770 // this one. It is not a guard written to catch this mistake: it is a
3771 // real scenario already in the suite -- flow 2's "runner exit before job
3772 // acceptance", process dead, elapsed 121s against a 120s startup window
3773 // -- which the rejected fix relabels `RegistrationTimedOut`. So the
3774 // objection is not only that a synthetic sweep dislikes the change; an
3775 // ordinary early crash, observed after a restart longer than the startup
3776 // window, gets reported as a runner that is still running. This test
3777 // remains the argument's *general* form, because it holds the claim at
3778 // every state, every observation and both sides of every deadline rather
3779 // than at one scenario.
3780 let timeouts = RecoveryTimeouts::new(
3781 Elapsed::seconds(60),
3782 Elapsed::seconds(120),
3783 Elapsed::seconds(300),
3784 );
3785 let entered_at = 1_000i64;
3786 // Both sides of each deadline, the deadlines themselves, and one
3787 // instant far past all three.
3788 let offsets = [0, 59, 60, 61, 119, 120, 121, 299, 300, 301, 100_000];
3789 let observations = [
3790 GithubRunnerObservation::NotRegistered,
3791 GithubRunnerObservation::Registered { busy: false },
3792 GithubRunnerObservation::Registered { busy: true },
3793 GithubRunnerObservation::Unreachable,
3794 ];
3795
3796 // Set by the third arm below, and asserted `false` after the sweep. It
3797 // is what keeps that arm from being a vacuous assertion nobody notices
3798 // has stopped meaning anything: today the arm is unreached, and this
3799 // says so out loud rather than leaving it to be assumed.
3800 let mut agent_termination_derived_here = false;
3801
3802 for state in AttemptState::ALL {
3803 for github in observations {
3804 for process_alive in [true, false] {
3805 for offset in offsets {
3806 let clock = StubClock::at(entered_at + offset);
3807 let observation = RecoveryObservation {
3808 process_alive,
3809 github,
3810 };
3811 let decision = recovery_decision(
3812 &attempt_in(state, entered_at),
3813 observation,
3814 timeouts,
3815 &clock,
3816 );
3817 let reason = match &decision {
3818 RecoveryDecision::Conclude(AttemptOutcome::Failed { reason })
3819 | RecoveryDecision::Terminate(AttemptOutcome::Failed { reason }) => {
3820 Some(reason)
3821 }
3822 _ => None,
3823 };
3824 match reason {
3825 Some(FailureReason::RegistrationTimedOut) => assert!(
3826 process_alive,
3827 "{state} at +{offset}s with {github:?} and a process the \
3828 observation says is gone was reported as still running"
3829 ),
3830 Some(FailureReason::ProcessExitedUnexpectedly) => assert!(
3831 !process_alive,
3832 "{state} at +{offset}s with {github:?} reported an \
3833 unexpected exit for a process that is alive"
3834 ),
3835 // Nothing here derives this one today -- it is
3836 // `e3`'s, recorded after reading its own journalled
3837 // terminate-intent back, which is why the flag above
3838 // records that the arm was reached at all. It is not
3839 // idle cover: once the mark is journalled somewhere
3840 // `recovery_decision` can read it, deriving the
3841 // reason here becomes legitimate, and this is what
3842 // stops that derivation landing beside a process the
3843 // observation reports as still running.
3844 Some(FailureReason::TerminatedAfterRegistrationTimeout) => {
3845 agent_termination_derived_here = true;
3846 assert!(
3847 !process_alive,
3848 "{state} at +{offset}s with {github:?} said the agent had \
3849 stopped a process the observation reports as alive"
3850 );
3851 }
3852 _ => {}
3853 }
3854 }
3855 }
3856 }
3857 }
3858
3859 // The division of labour, asserted rather than described. The reason
3860 // that names an agent-initiated stop is not derivable from an
3861 // observation -- a process this agent killed and one that crashed on its
3862 // own look identical to `recovery_decision` -- so it must not appear
3863 // anywhere in the sweep above. If it ever does, the fact that separates
3864 // the two cases has become an input, and `RecoveryDecision::Terminate`'s
3865 // deferred obligation is either discharged or wrong; either way somebody
3866 // should be reading that doc rather than passing this test by accident.
3867 assert!(
3868 !agent_termination_derived_here,
3869 "`recovery_decision` derived `TerminatedAfterRegistrationTimeout`, \
3870 which is `e3`'s to record from journalled terminate-intent and not \
3871 this function's to infer from an observation that cannot know"
3872 );
3873 }
3874
3875 #[test]
3876 fn a_terminated_runner_is_never_described_as_running() {
3877 // The `Display` half of the same argument. `e3` concludes the window in
3878 // `RecoveryDecision::Terminate`'s doc with this reason, on a pass where
3879 // the process is already dead, so its rendering may not claim otherwise.
3880 // Reusing `RegistrationTimedOut` there -- the closure this file used to
3881 // promise -- would have printed "the runner process is running but did
3882 // not register" about a process the operator can see is gone: the same
3883 // false liveness claim option A is rejected for, one pass later.
3884 let terminated = FailureReason::TerminatedAfterRegistrationTimeout.to_string();
3885 assert!(
3886 !terminated.contains("running"),
3887 "this reason is only ever recorded about a process the agent has \
3888 already stopped: {terminated}"
3889 );
3890 assert!(
3891 !terminated.contains("exited unexpectedly"),
3892 "and nothing exited unexpectedly either -- the agent stopped it on \
3893 purpose: {terminated}"
3894 );
3895 assert!(
3896 terminated.contains("stopped") && terminated.contains("register"),
3897 "it has to say both what happened to the process and why, or it \
3898 sends an operator to a crash investigation: {terminated}"
3899 );
3900
3901 // Distinct from both neighbours, in the variant and in the string. A
3902 // reason that renders identically to another is a reason `g2` cannot
3903 // use to send an operator anywhere different.
3904 assert_ne!(
3905 terminated,
3906 FailureReason::RegistrationTimedOut.to_string(),
3907 "the live and the stopped case must read differently"
3908 );
3909 assert_ne!(
3910 terminated,
3911 FailureReason::ProcessExitedUnexpectedly.to_string()
3912 );
3913 assert!(
3914 !matches!(
3915 FailureReason::TerminatedAfterRegistrationTimeout,
3916 FailureReason::Other(_)
3917 ),
3918 "a known, named condition must not travel through the escape hatch"
3919 );
3920 }
3921
3922 #[test]
3923 fn the_two_starting_failures_read_differently_to_an_operator() {
3924 // `g2` renders `FailureReason` directly, and these two send an operator
3925 // in different directions: one to a crash investigation, the other to a
3926 // networking or configuration fix. Reaching for `Other` here instead of
3927 // a named variant would have put a free-form string on the same path.
3928 let exited = FailureReason::ProcessExitedUnexpectedly.to_string();
3929 let timed_out = FailureReason::RegistrationTimedOut.to_string();
3930 assert_ne!(exited, timed_out);
3931 assert!(
3932 timed_out.contains("running") && timed_out.contains("register"),
3933 "the message must say the process is up and unregistered: {timed_out}"
3934 );
3935 assert!(
3936 !timed_out.contains("exited"),
3937 "the whole point is that nothing exited: {timed_out}"
3938 );
3939 assert!(
3940 !matches!(FailureReason::RegistrationTimedOut, FailureReason::Other(_)),
3941 "a known, named condition must not travel through the escape hatch"
3942 );
3943 }
3944
3945 #[test]
3946 fn a_dead_idle_runner_still_registered_at_github_is_orphaned() {
3947 let clock = StubClock::at(1_000_000);
3948 assert_eq!(
3949 recovery_decision(
3950 &attempt_in(AttemptState::Idle, 0),
3951 RecoveryObservation {
3952 process_alive: false,
3953 github: GithubRunnerObservation::Registered { busy: false },
3954 },
3955 RecoveryTimeouts::provisional(),
3956 &clock,
3957 ),
3958 RecoveryDecision::Conclude(AttemptOutcome::Orphaned),
3959 "the remote registration outlived our process and needs removing"
3960 );
3961 }
3962}