cellos-fleet 0.6.0-pre

//! cellos-fleet — host-resident agent that polls a spec queue and dispatches
//! execution cells to cellos-supervisor.
//!
//! # Library seam (S24)
//!
//! This crate exposes a thin library target alongside its binary so that the
//! configuration surface ([`Config`]) and node-identity surface
//! ([`NodeIdentity`]) can be exercised by integration tests and reused by
//! future control-plane tooling without shelling out to the binary. The
//! binary (`src/main.rs`) is a thin wrapper: it parses `--version`, installs
//! the tracing subscriber, builds a [`Config`] via [`Config::from_env`], and
//! drives [`run`]. No runtime behaviour changed when the seam was carved —
//! every item below was lifted verbatim from the original `main.rs`.
//!
//! # Queue model
//!
//! The agent treats an S3 prefix as a simple work queue using key renaming as
//! the claim primitive:
//!
//! ```text
//! pending/<spec-id>.json   →  claimed/<spec-id>.json  →  supervisor runs
//!                                                      →  completed/<spec-id>.json  (exit 0)
//!                                                      →  failed/<spec-id>.json     (exit ≠ 0)
//! ```
//!
//! Claiming is performed by `aws s3 mv` (copy + delete), which is not atomic
//! but is safe enough for low-concurrency single-agent deployments. A future
//! version can replace this with a DynamoDB conditional write for true
//! atomic claim.
//!
//! # Environment variables
//!
//! | Variable | Required | Description |
//! |----------|----------|-------------|
//! | `CELLOS_FLEET_BUCKET` | yes | S3 bucket name |
//! | `CELLOS_FLEET_PREFIX` | no | Key prefix inside bucket (default: `fleet`) |
//! | `CELLOS_FLEET_QUEUE_NAME` | no | Optional queue lane under the prefix |
//! | `CELLOS_FLEET_POOL_ID` | no | Runner pool identifier; T11 placement gate. When set, the dispatcher skips specs whose `spec.placement.poolId` is set AND does not equal this value. Specs without a `poolId` constraint are accepted everywhere. |
//! | `CELLOS_FLEET_SUPERVISOR` | no | Path to cellos-supervisor binary (default: `cellos-supervisor`) |
//! | `CELLOS_FLEET_POLL_INTERVAL_MS` | no | Poll interval in milliseconds (default: `5000`) |
//! | `CELLOS_FLEET_HEARTBEAT_INTERVAL_MS` | no | Heartbeat interval in milliseconds (default: `30000`) |
//! | `CELLOS_FLEET_NODE_ID` | no | Unique node identifier (default: hostname) |
//!
//! The agent inherits AWS credentials from the environment (IAM role, env vars,
//! or instance metadata) — it does not manage its own identity.
//!
//! # Drain / graceful shutdown
//!
//! On SIGTERM the poll loop stops accepting new work. Any in-flight cell
//! finishes normally before the process exits. A clean drain log line is
//! emitted so operators can distinguish graceful shutdown from a crash.

use anyhow::{Context, Result};
use std::path::PathBuf;
use std::time::Duration;
use tokio::process::Command;
use tokio::time::interval;
use tracing::{error, info, warn};

#[cfg(unix)]
use tokio::signal::unix::{signal, SignalKind};

/// Runtime configuration resolved from environment variables at startup.
#[derive(Debug)]
pub struct Config {
    pub bucket: String,
    pub prefix: String,
    /// Optional queue name for placement-aware routing.
    ///
    /// When set, all S3 key paths gain a `{queue_name}/` sub-prefix so that
    /// multiple fleet agents on different nodes can each service a distinct
    /// named lane of work without stepping on each other.
    ///
    /// Empty string means "default queue" — uses the legacy flat layout and is
    /// backwards-compatible with existing deployments that predate placement routing.
    pub queue_name: String,
    /// T11 — runner pool identifier. When non-empty, the dispatcher skips
    /// specs whose `spec.placement.poolId` is set and does not match this
    /// value. Empty string means "no pool constraint" — every spec accepted.
    ///
    /// Sourced from `CELLOS_FLEET_POOL_ID`.
    pub pool_id: String,
    pub supervisor: PathBuf,
    pub poll_interval: Duration,
    pub heartbeat_interval: Duration,
    pub node_id: String,
}

impl Config {
    pub fn from_env() -> Result<Self> {
        let bucket =
            std::env::var("CELLOS_FLEET_BUCKET").context("CELLOS_FLEET_BUCKET is required")?;
        let prefix = std::env::var("CELLOS_FLEET_PREFIX").unwrap_or_else(|_| "fleet".to_string());
        let queue_name = std::env::var("CELLOS_FLEET_QUEUE_NAME")
            .map(|value| value.trim().to_string())
            .unwrap_or_default();
        let pool_id = std::env::var("CELLOS_FLEET_POOL_ID")
            .map(|value| value.trim().to_string())
            .unwrap_or_default();
        let supervisor = PathBuf::from(
            std::env::var("CELLOS_FLEET_SUPERVISOR")
                .unwrap_or_else(|_| "cellos-supervisor".to_string()),
        );
        let poll_ms: u64 = std::env::var("CELLOS_FLEET_POLL_INTERVAL_MS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(5000);
        let heartbeat_ms: u64 = std::env::var("CELLOS_FLEET_HEARTBEAT_INTERVAL_MS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(30_000);
        let node_id = std::env::var("CELLOS_FLEET_NODE_ID").unwrap_or_else(|_| {
            hostname::get()
                .ok()
                .and_then(|h| h.into_string().ok())
                .unwrap_or_else(|| "unknown-node".to_string())
        });
        Ok(Config {
            bucket,
            prefix,
            queue_name,
            pool_id,
            supervisor,
            poll_interval: Duration::from_millis(poll_ms),
            heartbeat_interval: Duration::from_millis(heartbeat_ms),
            node_id,
        })
    }

    pub fn queue_prefix(&self) -> String {
        let base_prefix = self.prefix.trim_end_matches('/');
        if self.queue_name.is_empty() {
            base_prefix.to_string()
        } else {
            format!("{}/{}/", base_prefix, self.queue_name)
                .trim_end_matches('/')
                .to_string()
        }
    }

    pub fn pending_prefix(&self) -> String {
        format!("{}/pending/", self.queue_prefix())
    }

    pub fn claimed_key(&self, spec_id: &str) -> String {
        format!("{}/claimed/{}.json", self.queue_prefix(), spec_id)
    }

    pub fn completed_key(&self, spec_id: &str) -> String {
        format!("{}/completed/{}.json", self.queue_prefix(), spec_id)
    }

    pub fn failed_key(&self, spec_id: &str) -> String {
        format!("{}/failed/{}.json", self.queue_prefix(), spec_id)
    }

    /// T11 — placement gate.
    ///
    /// Returns `true` when this runner should dispatch the spec.
    ///
    /// - If the runner has no `pool_id` configured (env var unset / empty),
    ///   the spec is always accepted (legacy behaviour).
    /// - If the spec carries no `spec.placement.poolId`, it is accepted
    ///   everywhere (a spec without a pool constraint is portable).
    /// - Otherwise the spec is accepted only when the two values match
    ///   exactly. Mismatches return `false` and the dispatcher logs a
    ///   `skipping spec <id>: placement.poolId=<X> != runner poolId=<Y>`
    ///   line at the call site.
    pub fn should_dispatch(&self, spec_pool_id: Option<&str>) -> bool {
        if self.pool_id.is_empty() {
            return true;
        }
        match spec_pool_id {
            None => true,
            Some(spec_pool) => spec_pool == self.pool_id,
        }
    }
}

/// Stable identity a fleet node presents to a control plane.
///
/// S24 carved this as a forward-looking seam: today the fleet agent inherits
/// AWS credentials from the ambient environment and identifies itself only by
/// `node_id`. As accredited-deployment work lands (signed heartbeats, attested
/// runners), the control plane needs a typed handle for *who* a node claims to
/// be and *how* that claim is backed.
///
/// The struct is intentionally minimal and additive:
///
/// - `node_id` mirrors [`Config::node_id`] — the operator-visible name a node
///   answers to (hostname by default).
/// - `public_key_ref` is an opaque reference to the node's signing key (e.g. a
///   KMS key ARN or a key fingerprint). It is a *reference*, never key
///   material — the fleet agent does not hold private keys.
/// - `attestation` is a placeholder for a future hardware/software attestation
///   document. It defaults to `None`; no attestation path is wired today, so
///   the type parameter is the unit type until a concrete attestation shape is
///   chosen in a later track.
///
/// No runtime path constructs or consumes this type yet — it is a library
/// surface only, so the binary's behaviour is unchanged.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NodeIdentity {
    /// Operator-visible node name (matches [`Config::node_id`]).
    pub node_id: String,
    /// Opaque reference to the node's signing key (ARN / fingerprint / URI).
    /// Never the key material itself.
    pub public_key_ref: String,
    /// Future attestation document. `None` until an attestation path is wired.
    pub attestation: Option<()>,
}

impl NodeIdentity {
    /// Construct a `NodeIdentity` with no attestation (the only shape wired
    /// today). `attestation` defaults to `None`.
    pub fn new(node_id: impl Into<String>, public_key_ref: impl Into<String>) -> Self {
        NodeIdentity {
            node_id: node_id.into(),
            public_key_ref: public_key_ref.into(),
            attestation: None,
        }
    }
}

// ── S25: signed node attestation ────────────────────────────────────────────
//
// A fleet node proves *who it claims to be* to a control plane by emitting a
// signed CloudEvent carrying its `nodeId` + `publicKeyRef`. Today the only
// custody mode is key-possession (the node holds an Ed25519 signing seed behind
// a `Signer`), so the attestation is stamped `identityKind: "key-possession"`
// and `attestation: null` — an *explicit* marker that this node is NOT
// hardware-attested. The event is signed through the same per-event signing
// seam (`sign_event_with` over a `SoftwareSigner`, S32/ADR-0031) the rest of
// the fleet uses, so it verifies offline against the org-root
// `TrustAnchorPublicKey` only — no network, no PKI walk.
//
// Doctrine: this is identity *assertion*, not hardware *attestation*. The
// `attestation: null` field is load-bearing — a future hardware-attested node
// would carry a non-null attestation document and a different `identityKind`,
// and a verifier can branch on the marker without guessing.

/// CloudEvent `type` carried by a fleet node-attestation event (S25).
pub const NODE_ATTESTATION_EVENT_TYPE: &str = "dev.cellos.events.fleet.v1.node-attestation";

/// Identity-kind marker stamped into a key-possession attestation (S25).
///
/// "key-possession" means the node proved control of its signing key by
/// signing the attestation — it is explicitly NOT a hardware attestation.
pub const IDENTITY_KIND_KEY_POSSESSION: &str = "key-possession";

/// Build a signed node-attestation event for `identity` (S25).
///
/// Constructs a [`cellos_core::CloudEventV1`] of type
/// [`NODE_ATTESTATION_EVENT_TYPE`] whose `data` payload carries:
///
/// - `nodeId` — the node's operator-visible name ([`NodeIdentity::node_id`]),
/// - `publicKeyRef` — the opaque signing-key reference
///   ([`NodeIdentity::public_key_ref`]), never key material,
/// - `identityKind` — always [`IDENTITY_KIND_KEY_POSSESSION`] here, and
/// - `attestation` — an explicit JSON `null`, marking the node as NOT
///   hardware-attested.
///
/// The event is then signed through the per-event signing seam
/// ([`cellos_core::sign_event_with`]) using the supplied [`Signer`] (a
/// [`cellos_core::SoftwareSigner`] in the default key-possession custody mode).
/// The returned envelope verifies offline against the org-root verifying key.
///
/// `event_id` / `time` are caller-supplied so the function stays pure and
/// deterministic (no clock / RNG dependence) — the binary stamps a ULID + RFC
/// 3339 timestamp at the call site.
///
/// # Errors
///
/// Returns an error if the canonical signing payload cannot be produced or the
/// signer rejects the signing operation.
pub fn build_node_attestation(
    identity: &NodeIdentity,
    signer: &dyn cellos_core::Signer,
    event_id: impl Into<String>,
    source: impl Into<String>,
    time: Option<String>,
) -> Result<cellos_core::SignedEventEnvelopeV1, cellos_core::CellosError> {
    let event = cellos_core::CloudEventV1 {
        specversion: "1.0".to_string(),
        id: event_id.into(),
        source: source.into(),
        ty: NODE_ATTESTATION_EVENT_TYPE.to_string(),
        datacontenttype: Some("application/json".to_string()),
        data: Some(serde_json::json!({
            "nodeId": identity.node_id,
            "publicKeyRef": identity.public_key_ref,
            "identityKind": IDENTITY_KIND_KEY_POSSESSION,
            // Explicit null: this node is NOT hardware-attested. A future
            // hardware-attested node carries a non-null attestation document.
            "attestation": serde_json::Value::Null,
        })),
        time,
        traceparent: None,
        cex: None,
    };
    cellos_core::sign_event_with(signer, &event)
}

/// Verify a signed node-attestation envelope offline (S25).
///
/// Verifies the envelope's signature against `verifying_keys` (the org-root
/// keyring) via [`cellos_core::verify_signed_event_envelope`] — Ed25519 only,
/// so an empty HMAC keyring is passed — then re-asserts the structural
/// invariants of a node attestation:
///
/// - the event `type` is [`NODE_ATTESTATION_EVENT_TYPE`],
/// - the `data` payload carries an `identityKind` of
///   [`IDENTITY_KIND_KEY_POSSESSION`], and
/// - the `attestation` field is explicitly JSON `null` (NOT hardware-attested).
///
/// On success returns the asserted [`NodeIdentity`] reconstructed from the
/// signed payload. Any signature, type, or structural failure returns `Err` —
/// the verifier is fail-closed and never returns a partially-trusted identity.
///
/// # Errors
///
/// Returns an error when the signature does not verify under `verifying_keys`,
/// the event type is wrong, the payload is missing/malformed, the identity kind
/// is not key-possession, or the attestation marker is not explicit `null`.
pub fn verify_node_attestation(
    envelope: &cellos_core::SignedEventEnvelopeV1,
    verifying_keys: &std::collections::HashMap<String, cellos_core::crypto::TrustAnchorPublicKey>,
) -> Result<NodeIdentity, cellos_core::CellosError> {
    use cellos_core::CellosError;

    // 1. Cryptographic verification against the org-root keyring (Ed25519 only).
    let empty_hmac: std::collections::HashMap<String, Vec<u8>> = std::collections::HashMap::new();
    let event = cellos_core::verify_signed_event_envelope(envelope, verifying_keys, &empty_hmac)?;

    // 2. Re-assert the event type (defense in depth: the signature covers it,
    //    but a verifier asking "is this an attestation?" must not assume).
    if event.ty != NODE_ATTESTATION_EVENT_TYPE {
        return Err(CellosError::InvalidSpec(format!(
            "node attestation: expected event type {NODE_ATTESTATION_EVENT_TYPE}, got {:?}",
            event.ty
        )));
    }

    // 3. Re-assert the structural invariants of the signed payload.
    let data = event.data.as_ref().ok_or_else(|| {
        CellosError::InvalidSpec("node attestation: event has no data payload".into())
    })?;

    let identity_kind = data.get("identityKind").and_then(|v| v.as_str());
    if identity_kind != Some(IDENTITY_KIND_KEY_POSSESSION) {
        return Err(CellosError::InvalidSpec(format!(
            "node attestation: expected identityKind {IDENTITY_KIND_KEY_POSSESSION:?}, got {:?}",
            identity_kind
        )));
    }

    // The attestation marker MUST be present AND explicitly null — a missing
    // key is ambiguous, a non-null value would be a hardware attestation this
    // key-possession verifier does not understand.
    match data.get("attestation") {
        Some(serde_json::Value::Null) => {}
        other => {
            return Err(CellosError::InvalidSpec(format!(
                "node attestation: attestation must be explicit null (NOT hardware-attested), got {:?}",
                other
            )));
        }
    }

    let node_id = data.get("nodeId").and_then(|v| v.as_str()).ok_or_else(|| {
        CellosError::InvalidSpec("node attestation: missing/invalid nodeId".into())
    })?;
    let public_key_ref = data
        .get("publicKeyRef")
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            CellosError::InvalidSpec("node attestation: missing/invalid publicKeyRef".into())
        })?;

    Ok(NodeIdentity::new(node_id, public_key_ref))
}

// ── S26: advisory pre-claim ceiling-epoch floor ─────────────────────────────
//
// Defense-in-depth ONLY. The authoritative admission-ceiling control lives in
// the control plane / supervisor; this is an advisory pre-claim floor a fleet
// node reads from `CELLOS_FLEET_MIN_CEILING_EPOCH` so a node that has *locally*
// observed a stale ceiling epoch can refuse to claim work before it even
// reaches the authoritative gate. It is NOT a substitute for that gate — a
// node with no floor configured proceeds exactly as before.

/// Environment variable carrying the advisory minimum ceiling-epoch floor (S26).
pub const MIN_CEILING_EPOCH_ENV: &str = "CELLOS_FLEET_MIN_CEILING_EPOCH";

/// Outcome of the advisory pre-claim ceiling-epoch floor check (S26).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CeilingEpochDecision {
    /// No floor configured, or `observed >= floor`: the node may proceed to
    /// claim. The authoritative control still runs downstream.
    Admit,
    /// `observed < floor`: the node refuses to claim (advisory, fail-safe).
    /// Carries the stable reason code `ceiling_epoch_stale`.
    Refuse { reason: &'static str },
}

impl CeilingEpochDecision {
    /// `true` when the decision admits the work to the claim path.
    pub fn admits(&self) -> bool {
        matches!(self, CeilingEpochDecision::Admit)
    }
}

/// Reason code emitted when the advisory floor refuses a claim (S26).
pub const CEILING_EPOCH_STALE: &str = "ceiling_epoch_stale";

/// Pure advisory decision: does `observed_epoch` clear the `floor`? (S26)
///
/// - `floor == None` → [`CeilingEpochDecision::Admit`] (no floor configured;
///   legacy behaviour — the authoritative control is the only gate).
/// - `observed_epoch >= floor` → [`CeilingEpochDecision::Admit`].
/// - `observed_epoch < floor` → [`CeilingEpochDecision::Refuse`] with reason
///   [`CEILING_EPOCH_STALE`].
///
/// This is defense-in-depth: it can only ever *refuse* work a node was about to
/// claim, never *authorize* anything the authoritative ceiling control would
/// deny.
pub fn ceiling_epoch_admits(observed_epoch: u64, floor: Option<u64>) -> CeilingEpochDecision {
    match floor {
        None => CeilingEpochDecision::Admit,
        Some(floor) if observed_epoch >= floor => CeilingEpochDecision::Admit,
        Some(_) => CeilingEpochDecision::Refuse {
            reason: CEILING_EPOCH_STALE,
        },
    }
}

/// Read the advisory ceiling-epoch floor from [`MIN_CEILING_EPOCH_ENV`] (S26).
///
/// Returns `Ok(None)` when the variable is unset (no floor — proceed), `Ok(Some)`
/// when it parses as a `u64`, and `Err` when it is set but unparsable (fail-safe
/// for a misconfigured floor: a garbage value must not silently disable the
/// floor).
pub fn read_min_ceiling_epoch() -> Result<Option<u64>> {
    match std::env::var(MIN_CEILING_EPOCH_ENV) {
        Err(std::env::VarError::NotPresent) => Ok(None),
        Err(e) => Err(anyhow::anyhow!(
            "{MIN_CEILING_EPOCH_ENV} is not valid unicode: {e}"
        )),
        Ok(raw) => {
            let parsed = raw.trim().parse::<u64>().with_context(|| {
                format!("{MIN_CEILING_EPOCH_ENV}={raw:?} is not a valid u64 epoch")
            })?;
            Ok(Some(parsed))
        }
    }
}

// ── S29: profile-gated fail-open / fail-closed admission ─────────────────────
//
// `CELLOS_FLEET_PROFILE=hardened` flips the fleet node's posture when a
// pre-flight precondition cannot be satisfied — it cannot sign its attestation,
// it observed a stale ceiling epoch, or its audit sink is unreachable. Under
// the hardened profile any such failure fails CLOSED: the node refuses to
// dispatch. Under the default (non-hardened) profile the same failures fail
// OPEN with a *named* advisory so the gap is visible in logs rather than
// silent.

/// Environment variable selecting the fleet admission profile (S29).
pub const FLEET_PROFILE_ENV: &str = "CELLOS_FLEET_PROFILE";

/// Admission profile governing fail-open vs fail-closed behaviour (S29).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FleetProfile {
    /// Default posture: preconditions are best-effort; a failure fails OPEN
    /// with a named advisory.
    Default,
    /// Hardened posture: a precondition failure fails CLOSED (refuse dispatch).
    Hardened,
}

impl FleetProfile {
    /// Resolve the profile from `raw` (the `CELLOS_FLEET_PROFILE` value).
    /// `"hardened"` (case-insensitive, trimmed) selects [`FleetProfile::Hardened`];
    /// anything else — including unset — is [`FleetProfile::Default`].
    pub fn from_raw(raw: Option<&str>) -> Self {
        match raw.map(|s| s.trim().to_ascii_lowercase()) {
            Some(s) if s == "hardened" => FleetProfile::Hardened,
            _ => FleetProfile::Default,
        }
    }

    /// Resolve the profile from [`FLEET_PROFILE_ENV`].
    pub fn from_env() -> Self {
        Self::from_raw(std::env::var(FLEET_PROFILE_ENV).ok().as_deref())
    }

    /// `true` for the hardened (fail-closed) posture.
    pub fn is_hardened(&self) -> bool {
        matches!(self, FleetProfile::Hardened)
    }
}

/// The pre-flight preconditions a fleet node checks before dispatching (S29).
///
/// Each field is `true` when the precondition is SATISFIED. The decision
/// function [`profile_admits`] folds these against the active [`FleetProfile`].
#[derive(Debug, Clone, Copy)]
pub struct PreflightChecks {
    /// The node can sign its attestation (holds a usable signer).
    pub can_sign_attestation: bool,
    /// The locally observed ceiling epoch is not stale (cleared the S26 floor).
    pub ceiling_epoch_fresh: bool,
    /// The audit sink is reachable (signed events can be durably recorded).
    pub audit_sink_reachable: bool,
}

impl PreflightChecks {
    /// All preconditions satisfied.
    pub fn all_ok() -> Self {
        PreflightChecks {
            can_sign_attestation: true,
            ceiling_epoch_fresh: true,
            audit_sink_reachable: true,
        }
    }

    /// The stable reason codes for every UNSATISFIED precondition, in a fixed
    /// order. Empty when all preconditions hold.
    pub fn failure_reasons(&self) -> Vec<&'static str> {
        let mut reasons = Vec::new();
        if !self.can_sign_attestation {
            reasons.push("cannot_sign_attestation");
        }
        if !self.ceiling_epoch_fresh {
            reasons.push(CEILING_EPOCH_STALE);
        }
        if !self.audit_sink_reachable {
            reasons.push("audit_sink_unreachable");
        }
        reasons
    }
}

/// Outcome of the profile-gated admission decision (S29).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProfileAdmission {
    /// Dispatch may proceed. `advisories` is empty when every precondition
    /// holds, or carries the named reason codes that failed OPEN under the
    /// default profile (visible-but-permitted gaps).
    Admit { advisories: Vec<&'static str> },
    /// Dispatch is refused (hardened profile, a precondition failed CLOSED).
    /// `reasons` names every unsatisfied precondition.
    Refuse { reasons: Vec<&'static str> },
}

impl ProfileAdmission {
    /// `true` when dispatch may proceed.
    pub fn admits(&self) -> bool {
        matches!(self, ProfileAdmission::Admit { .. })
    }
}

/// Fold pre-flight preconditions against the active profile (S29).
///
/// - All preconditions satisfied → [`ProfileAdmission::Admit`] with no
///   advisories, regardless of profile.
/// - A precondition failed under [`FleetProfile::Hardened`] →
///   [`ProfileAdmission::Refuse`] naming every unsatisfied precondition
///   (fail CLOSED).
/// - A precondition failed under [`FleetProfile::Default`] →
///   [`ProfileAdmission::Admit`] carrying the failures as *named advisories*
///   (fail OPEN, but visibly — the gap is logged, not swallowed).
pub fn profile_admits(profile: FleetProfile, checks: &PreflightChecks) -> ProfileAdmission {
    let failures = checks.failure_reasons();
    if failures.is_empty() {
        return ProfileAdmission::Admit {
            advisories: Vec::new(),
        };
    }
    if profile.is_hardened() {
        ProfileAdmission::Refuse { reasons: failures }
    } else {
        ProfileAdmission::Admit {
            advisories: failures,
        }
    }
}

/// T11 — minimal spec view used by the placement gate.
///
/// We deserialize only the fields the gate needs so that fleet remains
/// resilient to other spec evolutions: a `pool_id` mismatch should be
/// detectable even if some other unrelated field rejects strict parsing.
#[derive(Debug, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FleetSpecView {
    #[serde(default)]
    spec: FleetSpecBody,
}

#[derive(Debug, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FleetSpecBody {
    #[serde(default)]
    placement: Option<FleetPlacementView>,
}

#[derive(Debug, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct FleetPlacementView {
    #[serde(default)]
    pool_id: Option<String>,
}

/// Read `spec.placement.poolId` from a spec JSON file on disk. Returns
/// `Ok(None)` when the spec has no `placement` block or no `poolId` within
/// it. Returns `Err` if the file cannot be read or parsed.
pub fn read_spec_pool_id(spec_path: &std::path::Path) -> Result<Option<String>> {
    let bytes = std::fs::read(spec_path)
        .with_context(|| format!("read spec file {}", spec_path.display()))?;
    let view: FleetSpecView = serde_json::from_slice(&bytes)
        .with_context(|| format!("parse spec file {}", spec_path.display()))?;
    Ok(view.spec.placement.and_then(|p| p.pool_id))
}

/// List pending spec keys from the S3 queue prefix.
///
/// Uses `aws s3api list-objects-v2` so we don't depend on the S3 SDK crate.
/// Returns key names (not full URIs) for all `.json` objects under `pending/`.
pub async fn list_pending(cfg: &Config) -> Result<Vec<String>> {
    let output = Command::new("aws")
        .args([
            "s3api",
            "list-objects-v2",
            "--bucket",
            &cfg.bucket,
            "--prefix",
            &cfg.pending_prefix(),
            "--query",
            "Contents[?ends_with(Key, '.json')].Key",
            "--output",
            "json",
        ])
        .output()
        .await
        .context("aws s3api list-objects-v2 failed")?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        warn!("list_pending stderr: {stderr}");
        return Ok(vec![]);
    }

    let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
    if stdout == "null" || stdout.is_empty() {
        return Ok(vec![]);
    }

    let keys: Vec<String> =
        serde_json::from_str(&stdout).context("failed to parse list-objects-v2 output")?;
    Ok(keys)
}

/// Attempt to claim a pending spec key by moving it to the claimed prefix.
///
/// Returns `true` if the claim succeeded (key moved); `false` if the key
/// was already claimed by another agent (race condition, non-fatal).
pub async fn try_claim(cfg: &Config, pending_key: &str) -> Result<bool> {
    // Extract spec-id from the key: pending/<spec-id>.json → <spec-id>
    let spec_id = pending_key
        .trim_start_matches(&cfg.pending_prefix())
        .trim_end_matches(".json");

    let src = format!("s3://{}/{}", cfg.bucket, pending_key);
    let dst = format!("s3://{}/{}", cfg.bucket, cfg.claimed_key(spec_id));

    let status = Command::new("aws")
        .args(["s3", "mv", &src, &dst])
        .status()
        .await
        .context("aws s3 mv (claim) failed")?;

    Ok(status.success())
}

/// Download the claimed spec to a local temp file and return its path.
pub async fn download_spec(cfg: &Config, spec_id: &str) -> Result<tempfile::NamedTempFile> {
    let tmp = tempfile::Builder::new()
        .prefix("cellos-fleet-spec-")
        .suffix(".json")
        .tempfile()
        .context("failed to create temp file for spec")?;

    let s3_key = format!("s3://{}/{}", cfg.bucket, cfg.claimed_key(spec_id));

    let status = Command::new("aws")
        .args(["s3", "cp", &s3_key, tmp.path().to_str().unwrap()])
        .status()
        .await
        .context("aws s3 cp (download spec) failed")?;

    anyhow::ensure!(status.success(), "aws s3 cp exited non-zero");
    Ok(tmp)
}

/// T11 — peek a pending spec without claiming it, so we can run the
/// placement gate before stealing the work from another pool's runner.
pub async fn peek_pending_spec(cfg: &Config, pending_key: &str) -> Result<tempfile::NamedTempFile> {
    let tmp = tempfile::Builder::new()
        .prefix("cellos-fleet-peek-")
        .suffix(".json")
        .tempfile()
        .context("failed to create temp file for peek")?;

    let s3_key = format!("s3://{}/{}", cfg.bucket, pending_key);

    let status = Command::new("aws")
        .args(["s3", "cp", &s3_key, tmp.path().to_str().unwrap()])
        .status()
        .await
        .context("aws s3 cp (peek pending spec) failed")?;

    anyhow::ensure!(status.success(), "aws s3 cp (peek) exited non-zero");
    Ok(tmp)
}

/// Run cellos-supervisor with the downloaded spec file.
///
/// Returns the exit code (0 = success, non-zero = cell failed or supervisor error).
pub async fn run_cell(cfg: &Config, spec_path: &std::path::Path) -> Result<i32> {
    let status = Command::new(&cfg.supervisor)
        .arg(spec_path)
        .status()
        .await
        .context("cellos-supervisor failed to launch")?;

    Ok(status.code().unwrap_or(1))
}

/// Move the claimed spec to completed/ or failed/ based on exit code.
pub async fn finalize(cfg: &Config, spec_id: &str, exit_code: i32) -> Result<()> {
    let src = format!("s3://{}/{}", cfg.bucket, cfg.claimed_key(spec_id));
    let dst = if exit_code == 0 {
        format!("s3://{}/{}", cfg.bucket, cfg.completed_key(spec_id))
    } else {
        format!("s3://{}/{}", cfg.bucket, cfg.failed_key(spec_id))
    };

    Command::new("aws")
        .args(["s3", "mv", &src, &dst])
        .status()
        .await
        .context("aws s3 mv (finalize) failed")?;

    Ok(())
}

/// Process one spec: (T11 placement peek →) claim → download → run → finalize.
pub async fn process_spec(cfg: &Config, pending_key: &str) -> Result<()> {
    let spec_id = pending_key
        .trim_start_matches(&cfg.pending_prefix())
        .trim_end_matches(".json");

    // T11 — placement gate: peek the spec's `placement.poolId` BEFORE we
    // attempt to claim. This way another runner whose pool matches gets a
    // fair shot at the work; we never steal a spec just to immediately
    // bounce it back. Runners with no `pool_id` configured skip the peek
    // entirely (every spec is in scope) and fall through to the legacy
    // claim path.
    if !cfg.pool_id.is_empty() {
        let peek_tmp = peek_pending_spec(cfg, pending_key).await?;
        let spec_pool = read_spec_pool_id(peek_tmp.path()).unwrap_or_else(|e| {
            warn!(spec_id, error = %e, "failed to read placement.poolId — treating as no constraint");
            None
        });
        if !cfg.should_dispatch(spec_pool.as_deref()) {
            info!(
                node = %cfg.node_id,
                spec_id,
                "skipping spec {}: placement.poolId={} != runner poolId={}",
                spec_id,
                spec_pool.as_deref().unwrap_or("<none>"),
                cfg.pool_id,
            );
            return Ok(());
        }
    }

    info!(node = %cfg.node_id, spec_id, "claiming spec");

    if !try_claim(cfg, pending_key).await? {
        info!(spec_id, "spec already claimed by another node, skipping");
        return Ok(());
    }

    info!(node = %cfg.node_id, spec_id, "claimed — downloading");
    let tmp = download_spec(cfg, spec_id).await?;

    info!(node = %cfg.node_id, spec_id, path = %tmp.path().display(), "running cell");
    let exit_code = run_cell(cfg, tmp.path()).await?;

    info!(node = %cfg.node_id, spec_id, exit_code, "cell completed — finalizing");
    finalize(cfg, spec_id, exit_code).await?;

    if exit_code == 0 {
        info!(node = %cfg.node_id, spec_id, "spec completed successfully");
    } else {
        warn!(node = %cfg.node_id, spec_id, exit_code, "spec completed with failure");
    }

    Ok(())
}

#[cfg(unix)]
async fn wait_for_shutdown_signal() -> Result<()> {
    let mut sigterm =
        signal(SignalKind::terminate()).context("failed to install SIGTERM handler")?;
    sigterm.recv().await;
    Ok(())
}

#[cfg(not(unix))]
async fn wait_for_shutdown_signal() -> Result<()> {
    tokio::signal::ctrl_c()
        .await
        .context("failed to install Ctrl+C handler")?;
    Ok(())
}

/// Main poll loop with heartbeat emission and SIGTERM drain.
///
/// The loop runs three concurrent timers via `tokio::select!`:
/// - **poll tick**: list pending specs and dispatch cells
/// - **heartbeat tick**: emit a structured `fleet.v1.heartbeat` event so
///   a control plane (or log aggregator) can track node liveness
/// - **SIGTERM**: stop accepting new work; current in-flight cell finishes
///   before the loop exits
pub async fn run(cfg: Config) -> Result<()> {
    info!(
        node = %cfg.node_id,
        bucket = %cfg.bucket,
        prefix = %cfg.prefix,
        queue_name = %cfg.queue_name,
        pool_id = %cfg.pool_id,
        supervisor = %cfg.supervisor.display(),
        poll_interval_ms = cfg.poll_interval.as_millis(),
        heartbeat_interval_ms = cfg.heartbeat_interval.as_millis(),
        "cellos-fleet agent starting"
    );

    let mut poll_tick = interval(cfg.poll_interval);
    let mut heartbeat_tick = interval(cfg.heartbeat_interval);
    let shutdown = wait_for_shutdown_signal();
    tokio::pin!(shutdown);

    // Fire immediately on first tick rather than waiting a full interval.
    poll_tick.tick().await;
    heartbeat_tick.tick().await;

    loop {
        tokio::select! {
            // Poll tick: check for pending specs and dispatch.
            _ = poll_tick.tick() => {
                match list_pending(&cfg).await {
                    Err(e) => error!("list_pending error: {e:#}"),
                    Ok(keys) if keys.is_empty() => {}
                    Ok(keys) => {
                        for key in &keys {
                            if let Err(e) = process_spec(&cfg, key).await {
                                error!(key, "process_spec error: {e:#}");
                            }
                        }
                    }
                }
            }

            // Heartbeat tick: emit liveness event.
            _ = heartbeat_tick.tick() => {
                info!(
                    event_type = "dev.cellos.events.fleet.v1.heartbeat",
                    node = %cfg.node_id,
                    bucket = %cfg.bucket,
                    prefix = %cfg.prefix,
                    queue_name = %cfg.queue_name,
                    pool_id = %cfg.pool_id,
                    "heartbeat"
                );
            }

            // SIGTERM: drain — finish no new work, let in-flight cells complete.
            _ = &mut shutdown => {
                info!(
                    node = %cfg.node_id,
                    "SIGTERM received — draining (no new work accepted)"
                );
                break;
            }
        }
    }

    info!(node = %cfg.node_id, "cellos-fleet agent stopped (drain complete)");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{Config, NodeIdentity};
    use std::path::PathBuf;
    use std::time::Duration;

    fn config(prefix: &str, queue_name: &str) -> Config {
        config_with_pool(prefix, queue_name, "")
    }

    fn config_with_pool(prefix: &str, queue_name: &str, pool_id: &str) -> Config {
        Config {
            bucket: "bucket".into(),
            prefix: prefix.into(),
            queue_name: queue_name.into(),
            pool_id: pool_id.into(),
            supervisor: PathBuf::from("cellos-supervisor"),
            poll_interval: Duration::from_secs(5),
            heartbeat_interval: Duration::from_secs(30),
            node_id: "node-a".into(),
        }
    }

    #[test]
    fn uses_legacy_layout_when_queue_name_is_empty() {
        let cfg = config("fleet", "");

        assert_eq!(cfg.pending_prefix(), "fleet/pending/");
        assert_eq!(cfg.claimed_key("spec-1"), "fleet/claimed/spec-1.json");
        assert_eq!(cfg.completed_key("spec-1"), "fleet/completed/spec-1.json");
        assert_eq!(cfg.failed_key("spec-1"), "fleet/failed/spec-1.json");
    }

    #[test]
    fn uses_queue_qualified_layout_when_queue_name_is_set() {
        let cfg = config("fleet", "gpu-runners");

        assert_eq!(cfg.pending_prefix(), "fleet/gpu-runners/pending/");
        assert_eq!(
            cfg.claimed_key("spec-1"),
            "fleet/gpu-runners/claimed/spec-1.json"
        );
        assert_eq!(
            cfg.completed_key("spec-1"),
            "fleet/gpu-runners/completed/spec-1.json"
        );
        assert_eq!(
            cfg.failed_key("spec-1"),
            "fleet/gpu-runners/failed/spec-1.json"
        );
    }

    #[test]
    fn trims_trailing_slash_from_prefix() {
        let cfg = config("fleet/", "gpu-runners");

        assert_eq!(cfg.pending_prefix(), "fleet/gpu-runners/pending/");
        assert_eq!(
            cfg.claimed_key("spec-1"),
            "fleet/gpu-runners/claimed/spec-1.json"
        );
    }

    // T11-2 — placement gate matrix. The dispatcher accepts a spec only
    // when the runner's pool constraint either is empty (legacy) or
    // matches the spec's `placement.poolId` exactly. Specs without a
    // `poolId` constraint are accepted everywhere — they are portable.
    #[test]
    fn dispatch_matrix_for_pool_id_placement_gate() {
        // Legacy runner (no pool_id configured): accepts everything.
        let unbounded = config_with_pool("fleet", "", "");
        assert!(
            unbounded.should_dispatch(None),
            "no-pool runner must accept specs without a poolId constraint"
        );
        assert!(
            unbounded.should_dispatch(Some("runner-pool-amd64")),
            "no-pool runner must accept specs with any poolId constraint"
        );

        // Bound runner: skips mismatches, accepts matches AND
        // accepts portable specs (placement.poolId not set).
        let amd64 = config_with_pool("fleet", "", "runner-pool-amd64");
        assert!(
            amd64.should_dispatch(None),
            "pool-bound runner must accept specs with no poolId constraint"
        );
        assert!(
            amd64.should_dispatch(Some("runner-pool-amd64")),
            "pool-bound runner must accept matching poolId"
        );
        assert!(
            !amd64.should_dispatch(Some("runner-pool-arm64")),
            "pool-bound runner must skip mismatching poolId"
        );
    }

    #[test]
    fn read_spec_pool_id_parses_placement_and_handles_absence() {
        use std::io::Write;

        // Spec WITH placement.poolId — should be Some.
        let mut with_pool = tempfile::NamedTempFile::new().unwrap();
        write!(
            with_pool,
            r#"{{
                "apiVersion": "cellos.io/v1",
                "kind": "ExecutionCell",
                "spec": {{
                    "id": "test",
                    "placement": {{ "poolId": "runner-pool-amd64" }}
                }}
            }}"#
        )
        .unwrap();
        let pool = super::read_spec_pool_id(with_pool.path()).unwrap();
        assert_eq!(pool.as_deref(), Some("runner-pool-amd64"));

        // Spec WITHOUT placement — should be None (portable).
        let mut without = tempfile::NamedTempFile::new().unwrap();
        write!(
            without,
            r#"{{
                "apiVersion": "cellos.io/v1",
                "kind": "ExecutionCell",
                "spec": {{ "id": "test" }}
            }}"#
        )
        .unwrap();
        assert_eq!(super::read_spec_pool_id(without.path()).unwrap(), None);
    }

    // S24 — NodeIdentity is an additive library seam. These tests pin the
    // documented shape: `attestation` defaults to `None`, the convenience
    // constructor mirrors the field order, and the struct is comparable.
    #[test]
    fn node_identity_new_defaults_attestation_to_none() {
        let id = NodeIdentity::new("node-a", "arn:aws:kms:us-east-1:0:key/abc");
        assert_eq!(id.node_id, "node-a");
        assert_eq!(id.public_key_ref, "arn:aws:kms:us-east-1:0:key/abc");
        assert_eq!(id.attestation, None);
    }

    #[test]
    fn node_identity_is_constructible_with_explicit_fields() {
        let id = NodeIdentity {
            node_id: "node-b".into(),
            public_key_ref: "fingerprint:deadbeef".into(),
            attestation: None,
        };
        assert_eq!(id, NodeIdentity::new("node-b", "fingerprint:deadbeef"));
    }

    // ── S25: signed node attestation ───────────────────────────────────────
    use super::{
        build_node_attestation, ceiling_epoch_admits, profile_admits, verify_node_attestation,
        CeilingEpochDecision, FleetProfile, PreflightChecks, ProfileAdmission,
        IDENTITY_KIND_KEY_POSSESSION, NODE_ATTESTATION_EVENT_TYPE,
    };
    use cellos_core::crypto::{dalek::public_key_from_seed, TrustAnchorPublicKey};
    use cellos_core::SoftwareSigner;
    use std::collections::HashMap;

    /// Org-root verifying-key map for a deterministic test seed.
    fn org_root_keys(kid: &str, seed: [u8; 32]) -> HashMap<String, TrustAnchorPublicKey> {
        let mut keys = HashMap::new();
        keys.insert(
            kid.to_string(),
            TrustAnchorPublicKey::from_bytes_unchecked(public_key_from_seed(&seed).unwrap()),
        );
        keys
    }

    #[test]
    fn node_attestation_verifies_offline_against_org_root_only() {
        // S25: a key-possession attestation signed by the node verifies offline
        // against the org-root TrustAnchorPublicKey only — no network, no HMAC.
        let seed = [9u8; 32];
        let kid = "org-root-2026";
        let signer = SoftwareSigner::from_seed(kid, seed).expect("signer");
        let identity = NodeIdentity::new("node-alpha", "arn:aws:kms:us-east-1:0:key/abc");

        let envelope = build_node_attestation(
            &identity,
            &signer,
            "ev-attest-1",
            "/cellos-fleet/node-alpha",
            Some("2026-06-25T00:00:00Z".into()),
        )
        .expect("build attestation");

        // The event carries the type and the key-possession marker.
        assert_eq!(envelope.event.ty, NODE_ATTESTATION_EVENT_TYPE);
        let data = envelope.event.data.as_ref().unwrap();
        assert_eq!(
            data.get("identityKind").and_then(|v| v.as_str()),
            Some(IDENTITY_KIND_KEY_POSSESSION)
        );
        // attestation is explicit null — NOT hardware-attested.
        assert!(data.get("attestation").unwrap().is_null());

        let keys = org_root_keys(kid, seed);
        let asserted = verify_node_attestation(&envelope, &keys).expect("verify offline");
        assert_eq!(asserted, identity);
    }

    #[test]
    fn node_attestation_tamper_fails_closed() {
        // S25: tampering with the signed event invalidates the signature.
        let seed = [9u8; 32];
        let kid = "org-root-2026";
        let signer = SoftwareSigner::from_seed(kid, seed).expect("signer");
        let identity = NodeIdentity::new("node-alpha", "ref-1");

        let mut envelope =
            build_node_attestation(&identity, &signer, "ev-1", "/src", None).expect("build");
        // Tamper: swap the asserted nodeId in the signed payload.
        envelope.event.data = Some(serde_json::json!({
            "nodeId": "node-impostor",
            "publicKeyRef": "ref-1",
            "identityKind": IDENTITY_KIND_KEY_POSSESSION,
            "attestation": serde_json::Value::Null,
        }));

        let keys = org_root_keys(kid, seed);
        let err = verify_node_attestation(&envelope, &keys)
            .expect_err("tampered attestation must fail verify");
        assert!(
            format!("{err}").contains("ed25519 verify failed"),
            "expected signature failure, got: {err}"
        );
    }

    #[test]
    fn node_attestation_wrong_org_root_fails() {
        // S25: a verifier holding a different org-root key cannot verify.
        let signer = SoftwareSigner::from_seed("org-root-2026", [9u8; 32]).expect("signer");
        let identity = NodeIdentity::new("node-alpha", "ref-1");
        let envelope =
            build_node_attestation(&identity, &signer, "ev-1", "/src", None).expect("build");

        // Verifier knows the kid but with the WRONG public key.
        let wrong_keys = org_root_keys("org-root-2026", [1u8; 32]);
        let err = verify_node_attestation(&envelope, &wrong_keys)
            .expect_err("wrong org-root key must fail");
        assert!(format!("{err}").contains("ed25519 verify failed"));
    }

    // ── S26: advisory pre-claim ceiling-epoch floor ────────────────────────
    #[test]
    fn ceiling_epoch_floor_refuses_when_observed_below_floor() {
        // floor 5 + epoch 4 -> refuse (stale).
        let decision = ceiling_epoch_admits(4, Some(5));
        assert_eq!(
            decision,
            CeilingEpochDecision::Refuse {
                reason: super::CEILING_EPOCH_STALE
            }
        );
        assert!(!decision.admits());

        // 5 + 5 (at the floor) -> proceeds.
        assert!(ceiling_epoch_admits(5, Some(5)).admits());
        // 5 + 6 (above the floor) -> proceeds.
        assert!(ceiling_epoch_admits(6, Some(5)).admits());

        // No floor configured -> proceeds regardless of observed epoch.
        assert!(ceiling_epoch_admits(0, None).admits());
        assert!(ceiling_epoch_admits(u64::MAX, None).admits());
    }

    // ── S29: profile-gated fail-open / fail-closed admission ────────────────
    #[test]
    fn profile_from_raw_only_hardened_selects_hardened() {
        assert_eq!(
            FleetProfile::from_raw(Some("hardened")),
            FleetProfile::Hardened
        );
        assert_eq!(
            FleetProfile::from_raw(Some(" Hardened ")),
            FleetProfile::Hardened
        );
        assert_eq!(
            FleetProfile::from_raw(Some("HARDENED")),
            FleetProfile::Hardened
        );
        assert_eq!(
            FleetProfile::from_raw(Some("default")),
            FleetProfile::Default
        );
        assert_eq!(FleetProfile::from_raw(Some("")), FleetProfile::Default);
        assert_eq!(FleetProfile::from_raw(None), FleetProfile::Default);
    }

    #[test]
    fn profile_admits_all_ok_regardless_of_profile() {
        let ok = PreflightChecks::all_ok();
        assert_eq!(
            profile_admits(FleetProfile::Hardened, &ok),
            ProfileAdmission::Admit { advisories: vec![] }
        );
        assert_eq!(
            profile_admits(FleetProfile::Default, &ok),
            ProfileAdmission::Admit { advisories: vec![] }
        );
    }

    #[test]
    fn hardened_fails_closed_on_each_precondition() {
        // Cannot sign attestation -> refuse closed under hardened.
        let cannot_sign = PreflightChecks {
            can_sign_attestation: false,
            ..PreflightChecks::all_ok()
        };
        let d = profile_admits(FleetProfile::Hardened, &cannot_sign);
        assert!(!d.admits());
        assert_eq!(
            d,
            ProfileAdmission::Refuse {
                reasons: vec!["cannot_sign_attestation"]
            }
        );

        // Stale ceiling epoch -> refuse closed.
        let stale = PreflightChecks {
            ceiling_epoch_fresh: false,
            ..PreflightChecks::all_ok()
        };
        assert_eq!(
            profile_admits(FleetProfile::Hardened, &stale),
            ProfileAdmission::Refuse {
                reasons: vec![super::CEILING_EPOCH_STALE]
            }
        );

        // Audit sink unreachable -> refuse closed.
        let sink_down = PreflightChecks {
            audit_sink_reachable: false,
            ..PreflightChecks::all_ok()
        };
        assert_eq!(
            profile_admits(FleetProfile::Hardened, &sink_down),
            ProfileAdmission::Refuse {
                reasons: vec!["audit_sink_unreachable"]
            }
        );
    }

    #[test]
    fn default_profile_fails_open_with_named_advisory() {
        // Default profile: same failures admit, but carry named advisories.
        let cannot_sign = PreflightChecks {
            can_sign_attestation: false,
            ..PreflightChecks::all_ok()
        };
        let d = profile_admits(FleetProfile::Default, &cannot_sign);
        assert!(d.admits());
        assert_eq!(
            d,
            ProfileAdmission::Admit {
                advisories: vec!["cannot_sign_attestation"]
            }
        );

        // Multiple failures: all named, in fixed order.
        let multi = PreflightChecks {
            can_sign_attestation: false,
            ceiling_epoch_fresh: false,
            audit_sink_reachable: false,
        };
        assert_eq!(
            profile_admits(FleetProfile::Default, &multi),
            ProfileAdmission::Admit {
                advisories: vec![
                    "cannot_sign_attestation",
                    super::CEILING_EPOCH_STALE,
                    "audit_sink_unreachable"
                ]
            }
        );
    }
}