autumn-cli 0.9.0

//! `MediaMTX` host provisioning for autumn-media (issue #1974, Slice 7).
//!
//! [autumn-media](../../../../autumn-media-plugin/index.html) turns `MediaMTX` +
//! `FFmpeg` into the streaming substrate for an Autumn app: RTMP/WHIP ingest,
//! HLS/WebRTC/WHEP playback, recording, and a control API — served by a Go
//! binary that is a **host prerequisite, not a cargo dependency**. This module
//! provisions that binary on a bare host from `autumn deploy`, mirroring exactly
//! how [`super::proxy::KamalProxyController`] provisions kamal-proxy: a
//! non-Rust host daemon is a systemd unit written and enabled by ordered
//! [`DeployOp`]s over ssh. When nothing is at `binary_path`, `deploy up` copies
//! the binary out of a digest-pinned image before cutover
//! ([`MediaMtxController::binary_install_ops`]).
//!
//! ## What is real here
//!
//! - [`MediaMtxHostConfig`] deserializes the `[media.mediamtx]` deploy section
//!   (defaults everywhere, so a project that never runs the media controller is
//!   unaffected). The `MediaMTX` ports are real [`u16`] constants with defaults
//!   ([`MEDIAMTX_API_PORT`] etc.), NOT parsed out of URL strings.
//! - [`render_mediamtx_yml`] renders the ingest/playback/recording config
//!   (generalized from arroyo's `mediamtx.yml`), and [`render_mediamtx_unit`]
//!   renders the `/etc/systemd/system/<unit>.service` unit — both pure, so the
//!   exact rendered text is unit-testable.
//! - [`MediaMtxController::ensure_installed_ops`] emits the ordered
//!   [`DeployOp`]s (mkdir the config dir, write the yml, write the unit,
//!   `daemon-reload && enable`, restart only on a change) driven over the injectable
//!   [`DeployExecutor`], so the remote command
//!   sequence is assertable against a recording fake with no live host. It
//!   no-ops (empty op vector) when the section is not `enabled`.
//! - Six fail-closed doctor checks — [`mediamtx_ports_distinct`],
//!   [`mediamtx_ports_match_bases`], [`ffmpeg_preflight`],
//!   [`mediamtx_binary_preflight`], [`recordings_dir_writable`], and
//!   [`mediamtx_ports_available`] — most run over the same executor and all
//!   return the shared [`PreflightCheck`] type, so a media host can be graded the
//!   same way `autumn deploy check` grades the app host. The first two are **pure
//!   config** checks that run before any probe: [`mediamtx_ports_distinct`]
//!   (issue #2051, Finding P) rejects a config binding two same-protocol
//!   listeners to one port, so `systemctl restart` cannot fail after cutover on a
//!   config a fresh-host `ss` scan would pass; [`mediamtx_ports_match_bases`]
//!   (issue #1974) rejects a listener port the app's own base URL does not call,
//!   so a customized port cannot silently strand the app on an unbound origin.
//!   The binary preflight runs before cutover so a host with no usable
//!   `MediaMTX` binary, and no way to install one, fails fast instead of
//!   committing the app and then failing the post-cutover restart. An unverifiable result
//!   (transport error, unparseable output) reports a clear failure — never a
//!   silent pass.
//!
//! ## `FFmpeg` path verification (concrete literals only)
//!
//! `[media.ffmpeg] bin` may be a concrete path (the default `/usr/bin/ffmpeg` or
//! any explicit path), a `${VAR}` placeholder, or overridden at runtime by the
//! `AUTUMN_MEDIA__FFMPEG__BIN` environment variable — the plugin resolves the
//! latter two from the **service's own environment** before it ever shells out to
//! `FFmpeg`. The deploy side does not have and must not guess that service
//! environment: the generated service env-file (`build_env_file`)
//! propagates only `AUTUMN_ENV`, the signing secret, and the database URL — never
//! the operator's local shell. Resolving `[media.ffmpeg] bin` against the CLI's
//! local process env (as an earlier revision did) therefore probes the *wrong*
//! environment, so [`ffmpeg_preflight`] verifies **only a concrete literal** `bin`
//! — an environment-independent value it can probe correctly.
//!
//! **Deferred, not failed.** An env/interpolation-indirected path — an empty
//! value, or one carrying a `${...}` placeholder (including
//! `${AUTUMN_MEDIA__FFMPEG__BIN}`) — is **deferred to runtime** with a clear
//! non-passing, **non-blocking** outcome ("`FFmpeg` path uses env/interpolation
//! indirection resolved in the service environment; deferring verification to
//! runtime") rather than resolved against the CLI's local env and probed. It sets
//! [`PreflightCheck::deferred`], so [`PreflightCheck::blocking`] is `false` and
//! the deferred result is surfaced as a **warning that does not abort `deploy
//! up`** — the deployed service resolves and runs whatever *its* environment
//! names, so blocking the deploy on a path the deploy side cannot (and must not)
//! verify would be wrong, while reporting it as a plain pass would falsely claim
//! success. A **concrete literal** `FFmpeg` path that is missing / not executable
//! / not actually `FFmpeg` still fails closed (blocking) as before.
//!
//! ## `MediaMTX` = a SEPARATE host daemon
//!
//! Compose/Fly deployments can still run `MediaMTX` as a separate service/app
//! (arroyo does exactly this). This controller is for the `autumn deploy`
//! (bare-host) path: it stands `MediaMTX` up as a host systemd unit alongside the
//! reverse-proxied app. The two share nothing but the host.
//!
//! ## CSP requirement (the app owns its CSP profile)
//!
//! Apps embedding the players must allow the `MediaMTX` origins in their
//! `content_security_policy`: `connect-src`/`media-src` for the WebRTC (`:8889`),
//! HLS (`:8888`), and playback (`:9996`) origins, plus `frame-src` for WebRTC,
//! and the object-store origin in `media-src` for recorded playback. This module
//! documents the required origins ([`MediaMtxHostConfig::required_csp_origins`]);
//! the app owns the actual directive. See arroyo's `autumn.toml` /
//! `autumn-fly.toml` for the concrete `connect-src`/`media-src`/`frame-src`
//! strings (local origins collapse to `https://*.fly.dev` + the object store in
//! production).
//!
//! ## Where the checks run
//!
//! `deploy up` runs all six before cutover. When `[deploy]` is set,
//! `autumn doctor` runs the two config-only checks offline, and all six over
//! SSH with `--online` ([`collect_media_doctor_checks`]).
//!
//! ## Placement
//!
//! This controller lives in `autumn-cli`, not in the plugin, so `autumn-cli`
//! does not depend on the plugin's dependency tree.

use serde::Deserialize;

use super::PreflightCheck;
use super::exec::{DeployExecutor, DeployOp, FileContents, RemoteCommand};

// ── MediaMTX ports (real u16 constants, NOT parsed from URLs) ────────────────

/// `MediaMTX` RTMP ingest port. (`WHIP` ingest publishes on the WebRTC port.)
pub const MEDIAMTX_RTMP_PORT: u16 = 1935;
/// `MediaMTX` HLS playback port.
pub const MEDIAMTX_HLS_PORT: u16 = 8888;
/// `MediaMTX` WebRTC/WHEP playback port.
pub const MEDIAMTX_WEBRTC_PORT: u16 = 8889;
/// `MediaMTX` recording-playback port.
pub const MEDIAMTX_PLAYBACK_PORT: u16 = 9996;
/// `MediaMTX` control-API port.
pub const MEDIAMTX_API_PORT: u16 = 9997;
/// `MediaMTX` WebRTC local UDP port (ICE host candidates).
pub const MEDIAMTX_WEBRTC_LOCAL_UDP_PORT: u16 = 8189;

// ── LL-HLS window (copied verbatim from arroyo's mediamtx.yml) ───────────────
//
// The DVR-on-live window = `HLS_SEGMENT_COUNT * HLS_SEGMENT_DURATION` = 60 * 2s
// = 120s. Kept as constants (not config) so the low-latency segmenting contract
// matches the players' assumptions; changing them means updating the app's
// advertised DVR window to match.

/// LL-HLS variant selector.
const HLS_VARIANT: &str = "lowLatency";
/// LL-HLS partial-segment duration.
const HLS_PART_DURATION: &str = "200ms";
/// LL-HLS full-segment duration.
const HLS_SEGMENT_DURATION: &str = "2s";
/// LL-HLS retained-segment count (× segment duration = the DVR window).
const HLS_SEGMENT_COUNT: u32 = 60;

// ── Config-section defaults (free fns so serde + `Default` share one source) ──

fn default_recordings_dir() -> String {
    "/recordings".to_owned()
}

fn default_record_delete_after() -> String {
    "72h".to_owned()
}

fn default_config_path() -> String {
    "/etc/mediamtx/mediamtx.yml".to_owned()
}

fn default_binary_path() -> String {
    "/usr/local/bin/mediamtx".to_owned()
}

fn default_unit_name() -> String {
    "mediamtx".to_owned()
}

const fn default_api_port() -> u16 {
    MEDIAMTX_API_PORT
}

const fn default_rtmp_port() -> u16 {
    MEDIAMTX_RTMP_PORT
}

const fn default_hls_port() -> u16 {
    MEDIAMTX_HLS_PORT
}

const fn default_webrtc_port() -> u16 {
    MEDIAMTX_WEBRTC_PORT
}

const fn default_playback_port() -> u16 {
    MEDIAMTX_PLAYBACK_PORT
}

const fn default_webrtc_local_udp() -> u16 {
    MEDIAMTX_WEBRTC_LOCAL_UDP_PORT
}

// ── App-side base-URL defaults (MIRRORED from the plugin) ────────────────────
//
// The value an unset `*_base` resolves to, so `mediamtx_ports_match_bases` can
// grade it. Mirrored rather than imported, like `DEFAULT_FFMPEG_BIN`: this
// module does not depend on the plugin.
//
// KEEP IN SYNC with `MediaMtxConfig` in `autumn-media-plugin/src/config.rs`.
// `ports_match_bases_passes_for_default_config` pins them against the port
// constants above.

/// Plugin default for `[media.mediamtx] api_base`.
const DEFAULT_API_BASE: &str = "http://127.0.0.1:9997";
/// Plugin default for `[media.mediamtx] rtmp_base`.
const DEFAULT_RTMP_BASE: &str = "rtmp://127.0.0.1:1935/live";
/// Plugin default for `[media.mediamtx] hls_base`.
const DEFAULT_HLS_BASE: &str = "http://127.0.0.1:8888";
/// Plugin default for `[media.mediamtx] webrtc_base`.
const DEFAULT_WEBRTC_BASE: &str = "http://127.0.0.1:8889";
/// Plugin default for `[media.mediamtx] playback_base`.
const DEFAULT_PLAYBACK_BASE: &str = "http://127.0.0.1:9996";

/// Default `FFmpeg` binary path — the plugin's `[media.ffmpeg] bin` default.
/// Shared so the deploy-side `FFmpeg` preflight and the plugin agree on the
/// fallback path.
pub const DEFAULT_FFMPEG_BIN: &str = "/usr/bin/ffmpeg";

fn default_ffmpeg_bin() -> String {
    DEFAULT_FFMPEG_BIN.to_owned()
}

/// Host-provisioning settings for `MediaMTX`, read from the `[media.mediamtx]`
/// deploy section.
///
/// Every field has a serde default (via the struct-level `#[serde(default)]` +
/// the [`Default`] impl), so a bare `[media.mediamtx]` table — or none at all —
/// yields the localhost dev shape with the controller disabled. Deserialized
/// out of the merged `autumn.toml` value via [`media_host_config_from_value`]
/// (mirroring how the plugin reads `[media]`), so it never touches
/// `autumn-web`'s strict `AutumnConfig` schema — no `autumn-media-plugin`
/// dependency and no change to the app config type.
///
/// ## Deploy-side ports ↔ app runtime base URLs
///
/// The `*_port` fields here (`api_port` / `rtmp_port` / `hls_port` /
/// `webrtc_port` / `playback_port`) only **provision the daemon's listeners** in
/// the rendered `mediamtx.yml` (see [`render_mediamtx_yml`]). The *app's*
/// runtime `MediaMtxConfig` `*_base` URLs — which the deployed app/clients build
/// request URLs from — are a separate surface in the same `[media.mediamtx]`
/// table, so customizing a port without updating its base URL leaves the app
/// calling an origin the daemon no longer binds.
///
/// This module still does **not** depend on `autumn-media-plugin` (that coupling
/// was out of slice-7 scope, and the plugin owns its own config model). Instead
/// the `*_base` keys are deserialized here as opaque strings and read by exactly
/// one thing — [`mediamtx_ports_match_bases`], which fails the deploy closed when
/// the two disagree (issue #1974). Provisioning never reads them. The two
/// surfaces are default-aligned, so the common case needs no action.
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct MediaMtxHostConfig {
    /// When `false` (the default) the controller is a no-op — it emits no
    /// [`DeployOp`]s, so an app that never provisions `MediaMTX` is unaffected.
    pub enabled: bool,
    /// Control-API listen port.
    pub api_port: u16,
    /// RTMP ingest listen port. `WHIP` ingest is WebRTC and publishes on
    /// [`webrtc_port`](Self::webrtc_port), not here.
    pub rtmp_port: u16,
    /// HLS playback listen port.
    pub hls_port: u16,
    /// WebRTC/WHEP playback listen port.
    pub webrtc_port: u16,
    /// Recording-playback listen port.
    pub playback_port: u16,
    /// WebRTC local UDP port for ICE host candidates.
    pub webrtc_local_udp: u16,
    /// Root directory recordings are written under (`recordPath` prefix).
    pub recordings_dir: String,
    /// `MediaMTX` `recordDeleteAfter` retention window (e.g. `72h`).
    pub record_delete_after: String,
    /// Remote path the rendered `mediamtx.yml` is written to.
    pub config_path: String,
    /// Remote path of the `MediaMTX` binary.
    pub binary_path: String,
    /// When `true` (the default), `deploy up` installs the pinned `MediaMTX`
    /// ([`MEDIAMTX_KNOWN_GOOD_VERSION`]) at `binary_path` if nothing is there.
    /// It never replaces a file that is there. Set `false` to install it
    /// yourself.
    pub install_binary: bool,
    /// systemd unit name (without the `.service` suffix).
    pub unit_name: String,
    /// Extra hostnames/IPs announced as WebRTC ICE candidates
    /// (`webrtcAdditionalHosts`) — e.g. a public address in production.
    pub webrtc_additional_hosts: Vec<String>,

    // App-side base URLs, read only by `mediamtx_ports_match_bases`.
    // `None` = unset, which resolves to the mirrored `DEFAULT_*_BASE`.
    /// App-side `api_base` (control API), when set.
    pub api_base: Option<String>,
    /// App-side `rtmp_base` (RTMP ingest), when set.
    pub rtmp_base: Option<String>,
    /// App-side `hls_base` (HLS playback), when set.
    pub hls_base: Option<String>,
    /// App-side `hls_probe_base` (server-side HLS probe), when set.
    pub hls_probe_base: Option<String>,
    /// App-side `webrtc_base` (WebRTC/WHEP playback and WHIP ingest), when set.
    pub webrtc_base: Option<String>,
    /// App-side `playback_base` (recording playback), when set.
    pub playback_base: Option<String>,
    /// App-side `playback_probe_base` (server-side playback probe), when set.
    pub playback_probe_base: Option<String>,
}

impl Default for MediaMtxHostConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            api_port: default_api_port(),
            rtmp_port: default_rtmp_port(),
            hls_port: default_hls_port(),
            webrtc_port: default_webrtc_port(),
            playback_port: default_playback_port(),
            webrtc_local_udp: default_webrtc_local_udp(),
            recordings_dir: default_recordings_dir(),
            record_delete_after: default_record_delete_after(),
            config_path: default_config_path(),
            binary_path: default_binary_path(),
            install_binary: true,
            unit_name: default_unit_name(),
            webrtc_additional_hosts: Vec::new(),
            api_base: None,
            rtmp_base: None,
            hls_base: None,
            hls_probe_base: None,
            webrtc_base: None,
            playback_base: None,
            playback_probe_base: None,
        }
    }
}

/// Label of the op that installs the `MediaMTX` binary. It runs before
/// `migrate`, so it is in [`super::exec::PRE_MIGRATE_LABELS`].
pub const MEDIA_INSTALL_BINARY_LABEL: &str = "media-install-binary";

/// The `MediaMTX` release `deploy up` installs on a host that has none.
///
/// The `mediamtx.yml` renderer targets this release line (v1.19.x).
pub const MEDIAMTX_KNOWN_GOOD_VERSION: &str = "1.19.3";

/// OCI image-index digest of the `bluenviron/mediamtx` image at
/// [`MEDIAMTX_KNOWN_GOOD_VERSION`].
///
/// The install pulls `tag@digest`, never the bare tag. A Docker Hub tag can
/// move, and the binary runs as root. Change this only together with the
/// version, and read the new digest from the registry:
///
/// ```text
/// T=$(curl -sS "https://auth.docker.io/token?service=registry.docker.io&scope=repository:bluenviron/mediamtx:pull" | jq -r .token)
/// curl -sSI -H "Authorization: Bearer $T" \
///   -H "Accept: application/vnd.oci.image.index.v1+json" \
///   https://registry-1.docker.io/v2/bluenviron/mediamtx/manifests/<version> | grep -i docker-content-digest
/// ```
pub const MEDIAMTX_KNOWN_GOOD_DIGEST: &str =
    "sha256:7797ed3df88df21e8c04ecd0aff08ce49a5232d1db453e51f5480ef36bc80865";

/// The `[media]` table wrapper used to deserialize a full `autumn.toml` string
/// down to the `MediaMTX` host section, sidestepping `autumn-web`'s strict
/// `AutumnConfig` schema (mirrors the plugin's own `AutumnTomlRoot`).
#[derive(Debug, Default, Deserialize)]
struct MediaTomlRoot {
    #[serde(default)]
    media: MediaSection,
}

#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct MediaSection {
    mediamtx: MediaMtxHostConfig,
    ffmpeg: FfmpegSection,
}

/// The `[media.ffmpeg]` sub-table — only the `bin` path is relevant to the
/// deploy-side `FFmpeg` preflight.
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
struct FfmpegSection {
    bin: String,
}

impl Default for FfmpegSection {
    fn default() -> Self {
        Self {
            bin: default_ffmpeg_bin(),
        }
    }
}

impl MediaMtxHostConfig {
    /// The `MediaMTX` origins an embedding app must allow in its
    /// `connect-src`/`media-src` (and `frame-src` for WebRTC) CSP directives, in
    /// the local `http://127.0.0.1:<port>` shape.
    ///
    /// Returns the WebRTC (`webrtc_port`), HLS (`hls_port`), and playback
    /// (`playback_port`) origins — the three browser-facing `MediaMTX` ports. The
    /// control API (`api_port`) is server-side only (browsers never hit it) and
    /// so is deliberately excluded. In production these origins collapse to the
    /// public `MediaMTX` origin (e.g. `https://<app>-mediamtx.fly.dev`), and the
    /// object-store origin must additionally be allowed in `media-src` for
    /// recorded playback — see the module docs.
    #[must_use]
    pub fn required_csp_origins(&self) -> Vec<String> {
        vec![
            format!("http://127.0.0.1:{}", self.webrtc_port),
            format!("http://127.0.0.1:{}", self.hls_port),
            format!("http://127.0.0.1:{}", self.playback_port),
        ]
    }
}

/// Deserialize the `[media]` host config out of an **already-merged**
/// `autumn.toml` root [`toml::Value`] — base ← inline `[profile.<name>]` ←
/// `autumn-<profile>.toml` — returning both the [`MediaMtxHostConfig`] and the
/// `[media.ffmpeg] bin` path (issue #2051, Finding O).
///
/// The deploy-side loader resolves the media config under the **target deploy
/// profile** using the SAME base+profile TOML layering `AutumnConfig::load()`
/// applies to the app config, so a profiled deploy (`prod`/`staging`) sees the
/// profile's `[media.mediamtx]` / `[media.ffmpeg]` (from
/// `[profile.<name>.media.*]` or `autumn-<profile>.toml`) instead of the base
/// default. Rather than round-tripping the merged value back through a string
/// (which can panic on `toml::to_string` table-ordering), the caller merges raw
/// `toml::Value`s and deserializes the `[media]` subtree here.
///
/// Fail-closed: an *absent*
/// `[media]` subtree maps to the disabled default, but a subtree present with an
/// **ill-typed value in any contributing layer** is a hard [`Err`] (the merged
/// value carries the bad type), so the caller aborts `deploy plan`/`up` rather
/// than deploying a media-enabled app WITHOUT provisioning `MediaMTX`.
///
/// # Errors
///
/// Returns the [`toml::de::Error`] when the merged `[media.mediamtx]` /
/// `[media.ffmpeg]` subtree has a wrong-typed value.
pub fn media_host_config_from_value(
    root: toml::Value,
) -> Result<(MediaMtxHostConfig, String), toml::de::Error> {
    let parsed: MediaTomlRoot = root.try_into()?;
    Ok((parsed.media.mediamtx, parsed.media.ffmpeg.bin))
}

/// Render the `MediaMTX` `mediamtx.yml`, parameterized by [`MediaMtxHostConfig`].
///
/// Generalized from arroyo's broadcast-only `mediamtx.yml`: same LL-HLS window
/// (`hlsVariant: lowLatency`, `hlsPartDuration: 200ms`, `hlsSegmentDuration: 2s`,
/// `hlsSegmentCount: 60`), same fmp4 recording defaults under `recordings_dir`,
/// and the same WebRTC config — plus a **`~^room/.+$` path matcher** for
/// autumn-media's Rooms primitive (Slice 6), which arroyo lacks (it is
/// broadcast-only). Pure — exposed for unit assertions.
///
/// # `MediaMTX` schema currency (validated against v1.19.x / `bluenviron/mediamtx:1`)
///
/// `MediaMTX` rejects an **unknown** config field at startup (`json: unknown
/// field ...`), and this config is installed by a post-cutover `systemctl
/// restart` — so a single stale key would commit the app release then leave the
/// daemon dead and playback stranded. Every key rendered here is therefore
/// cross-checked against the current `MediaMTX` reference config
/// (<https://github.com/bluenviron/mediamtx/blob/main/mediamtx.yml> and
/// `internal/conf/conf.go`), the schema targeted by arroyo's pinned
/// `bluenviron/mediamtx:1` image. Notably the HLS CORS key is the plural array
/// `hlsAllowOrigins: ['*']` — the singular scalar `hlsAllowOrigin` is a
/// deprecated alias, so the current plural form is emitted (issue #2051,
/// Finding S). The globals (`api`/`apiAddress`, `playback`/`playbackAddress`,
/// `rtmp`/`rtmpAddress`, `hls`/`hlsAddress`/`hlsVariant`/`hlsPartDuration`/
/// `hlsSegmentCount`/`hlsSegmentDuration`, `webrtc`/`webrtcAddress`/
/// `webrtcAllowOrigins`/`webrtcLocalUDPAddress`/`webrtcAdditionalHosts`, the
/// `rtsp`/`srt`/`moq`/`metrics`/`pprof` disable toggles), the `pathDefaults`
/// recording keys (`record`/`recordPath`/`recordFormat`/`recordPartDuration`/
/// `recordSegmentDuration`/`recordDeleteAfter`), and the `paths` matchers all
/// match that reference in name and value shape. Future edits: re-validate any
/// added key against that reference before shipping it.
#[must_use]
pub fn render_mediamtx_yml(cfg: &MediaMtxHostConfig) -> String {
    // `webrtcAdditionalHosts` renders as an inline YAML list. Each host is
    // single-quoted so an IPv6 literal or a host with special chars stays a
    // single scalar; empty -> `[]`.
    let webrtc_hosts = if cfg.webrtc_additional_hosts.is_empty() {
        "[]".to_owned()
    } else {
        let joined = cfg
            .webrtc_additional_hosts
            .iter()
            .map(|h| yaml_single_quote(h))
            .collect::<Vec<_>>()
            .join(", ");
        format!("[{joined}]")
    };

    // The two config-derived string values in this template (the `recordPath`
    // prefix built from `recordings_dir`, and the `recordDeleteAfter` retention
    // window) are rendered as single-quoted YAML scalars (issue #2051, Finding T
    // sweep). A plain scalar carrying a leading/trailing space or a YAML
    // structural char (`:`/`#`/`{`/`[`/…) would otherwise break parsing — and a
    // `mediamtx.yml` that fails to parse strands the daemon on the deferred
    // post-cutover `systemctl restart`, the worst time. Quoting a path/duration
    // is loss-free for MediaMTX (both parse the same string either way).
    let record_path = yaml_single_quote(&format!(
        "{}/%path/%Y-%m-%d_%H-%M-%S-%f",
        cfg.recordings_dir
    ));
    let record_delete = yaml_single_quote(&cfg.record_delete_after);

    format!(
        "logLevel: info\n\
         \n\
         # Only the protocols autumn-media manages are enabled below (RTMP ingest,\n\
         # HLS, WebRTC/WHEP playback, the control API, and the recording-playback\n\
         # server). Every other protocol MediaMTX turns on by default is disabled\n\
         # explicitly, so the systemd unit only ever opens the ports this deploy\n\
         # plan declares — an occupied default RTSP (:8554), SRT (:8890), or MoQ\n\
         # (:8892) port cannot make `systemctl restart mediamtx` fail after\n\
         # preflight passed. (`rtsp: no` also disables RTSPS; `moq: no` closes the\n\
         # default-on MoQ HTTP/2+HTTP/3 listeners — autumn-media does not use MoQ;\n\
         # `metrics`/`pprof` default off but are pinned off for the same 'declared\n\
         # ports only' contract.)\n\
         rtsp: no\n\
         srt: no\n\
         moq: no\n\
         metrics: no\n\
         pprof: no\n\
         \n\
         # The control API is the server-side control plane: the co-located app\n\
         # reaches it at 127.0.0.1:{api} and it is intentionally excluded from the\n\
         # browser-facing CSP origins, so bind it to loopback only rather than\n\
         # every interface. The viewer/ingest listeners below (RTMP, HLS,\n\
         # WebRTC/WHEP, playback) stay bound on all interfaces so encoders and\n\
         # browsers can reach them.\n\
         api: true\n\
         apiAddress: 127.0.0.1:{api}\n\
         \n\
         playback: true\n\
         playbackAddress: :{playback}\n\
         \n\
         rtmp: true\n\
         rtmpAddress: :{rtmp}\n\
         \n\
         hls: true\n\
         hlsAddress: :{hls}\n\
         hlsAllowOrigins: ['*']\n\
         hlsVariant: {hls_variant}\n\
         hlsPartDuration: {hls_part}\n\
         # DVR-on-live window: retained window = hlsSegmentCount x hlsSegmentDuration\n\
         # = {seg_count} x {seg_dur}. Surfaced to the player so the UI never promises\n\
         # more rewind than MediaMTX actually keeps.\n\
         hlsSegmentCount: {seg_count}\n\
         hlsSegmentDuration: {seg_dur}\n\
         \n\
         webrtc: true\n\
         webrtcAddress: :{webrtc}\n\
         webrtcAllowOrigins: ['*']\n\
         webrtcLocalUDPAddress: :{webrtc_udp}\n\
         webrtcAdditionalHosts: {webrtc_hosts}\n\
         \n\
         pathDefaults:\n\
         \x20\x20source: publisher\n\
         \x20\x20record: true\n\
         \x20\x20recordPath: {rec_path}\n\
         \x20\x20recordFormat: fmp4\n\
         \x20\x20recordPartDuration: 1s\n\
         \x20\x20recordSegmentDuration: 5m\n\
         \x20\x20recordDeleteAfter: {rec_delete}\n\
         \n\
         paths:\n\
         # Broadcast ingest (single-publisher channels), as in arroyo.\n\
         \x20\x20~^live/.+$:\n\
         \x20\x20\x20\x20source: publisher\n\
         # autumn-media Rooms (Slice 6): room/{{namespace?}}/{{room_id}}/{{participant}}.\n\
         # arroyo is broadcast-only and has no room matcher; this is the addition.\n\
         \x20\x20~^room/.+$:\n\
         \x20\x20\x20\x20source: publisher\n",
        api = cfg.api_port,
        playback = cfg.playback_port,
        rtmp = cfg.rtmp_port,
        hls = cfg.hls_port,
        webrtc = cfg.webrtc_port,
        webrtc_udp = cfg.webrtc_local_udp,
        hls_variant = HLS_VARIANT,
        hls_part = HLS_PART_DURATION,
        seg_count = HLS_SEGMENT_COUNT,
        seg_dur = HLS_SEGMENT_DURATION,
        webrtc_hosts = webrtc_hosts,
        rec_path = record_path,
        rec_delete = record_delete,
    )
}

/// Render `value` as a single-quoted YAML scalar (the only escape inside a YAML
/// single-quoted scalar is a doubled `''` for a literal quote). Used for the
/// config-derived string values in [`render_mediamtx_yml`] (record path, retention
/// window, WebRTC additional hosts) so a value carrying whitespace or a YAML
/// structural char stays a single scalar and never breaks `mediamtx.yml` parsing.
#[must_use]
fn yaml_single_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "''"))
}

/// Render `value` as a systemd double-quoted `ExecStart` argument (issue #2051,
/// Finding T). systemd is **not** a shell: it splits `ExecStart` on whitespace but
/// honors its own C-style quoting, where a double-quoted token is one argument and
/// `\\` / `\"` are the escapes for a literal backslash / quote. So a
/// `binary_path`/`config_path` containing a space (e.g. `/opt/media mtx/mediamtx`)
/// must be wrapped in double quotes or the deferred post-cutover
/// `systemctl restart` parses the command as the truncated path `/opt/media`.
/// Backslash is escaped first, then the quote, so the result round-trips exactly.
#[must_use]
fn systemd_quote(value: &str) -> String {
    format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
}

/// Render the systemd unit that supervises `MediaMTX`, styled like
/// [`super::render_app_unit`] / [`super::proxy::KamalProxyController::render_proxy_unit`].
///
/// `ExecStart` is `"<binary_path>" "<config_path>"` — both paths are
/// [`systemd_quote`]d (issue #2051, Finding T) so a path containing whitespace
/// stays a single argument instead of splitting the command; `Restart=always` (a
/// media daemon that dies must come straight back so ingest/playback recover
/// without operator intervention). Pure — exposed for unit assertions.
#[must_use]
pub fn render_mediamtx_unit(cfg: &MediaMtxHostConfig) -> String {
    format!(
        "[Unit]\n\
         Description=MediaMTX (autumn-media host daemon)\n\
         After=network-online.target\n\
         Wants=network-online.target\n\
         \n\
         [Service]\n\
         Type=simple\n\
         ExecStart={bin} {config}\n\
         Restart=always\n\
         RestartSec=2\n\
         \n\
         [Install]\n\
         WantedBy=multi-user.target\n",
        bin = systemd_quote(&cfg.binary_path),
        config = systemd_quote(&cfg.config_path),
    )
}

/// The parent directory of a remote unix path — everything before the last `/`,
/// or `None` when the path has no parent segment (a bare filename, or a
/// root-level path whose parent `/` always exists).
///
/// Used to `mkdir -p` a config file's directory before `scp` writes into it:
/// [`super::exec::SshExecutor::upload`] shells out to `scp`, which does **not**
/// create missing parent directories, so the default
/// `config_path = /etc/mediamtx/mediamtx.yml` would otherwise fail at
/// `media-write-config` on the first deploy (the `/etc/mediamtx` dir does not
/// exist on a fresh host). Pure — unit-tested.
#[must_use]
fn parent_dir(path: &str) -> Option<&str> {
    match path.rsplit_once('/') {
        Some((parent, _)) if !parent.is_empty() => Some(parent),
        _ => None,
    }
}

/// Provisions `MediaMTX` as a host systemd unit, modeled on
/// [`super::proxy::KamalProxyController`].
///
/// The single [`Self::ensure_installed_ops`] method returns the ordered
/// [`DeployOp`]s (rather than driving the executor directly) so the media steps
/// slot into the same recorded op sequence as the rest of the deploy and are
/// assertable in-order against the recording fake.
#[derive(Debug, Clone)]
pub struct MediaMtxController {
    cfg: MediaMtxHostConfig,
}

impl MediaMtxController {
    /// Build a controller for the given host config.
    #[must_use]
    pub const fn new(cfg: MediaMtxHostConfig) -> Self {
        Self { cfg }
    }

    /// Remote path of the systemd unit (`/etc/systemd/system/<unit>.service`).
    #[must_use]
    pub fn unit_path(&self) -> String {
        format!("/etc/systemd/system/{}.service", self.cfg.unit_name)
    }

    /// Ops that install the pinned `MediaMTX` binary at `binary_path`.
    ///
    /// Empty when the section is not `enabled` or `install_binary` is `false`.
    /// Otherwise one `Run` op (`media-install-binary`). `deploy up` runs it
    /// before cutover, because it only adds a file at an empty path.
    ///
    /// The shell does these steps:
    ///
    /// 1. If an executable file is at `binary_path`, stop with success. Deploy
    ///    never replaces an operator's binary.
    /// 2. If any other file, directory or symlink is there, stop with an error.
    /// 3. Install `docker.io` with apt if `docker` is missing, and start it.
    ///    apt waits up to 5 minutes for its lock.
    /// 4. Remove any old staged path (a symlink is removed, not followed).
    ///    Copy `/mediamtx` out of `bluenviron/mediamtx:<version>@<digest>` to
    ///    the staged path next to `binary_path`, and remove the scratch container.
    /// 5. Make sure the staged file prints `v<version>` for `--version`. If not,
    ///    delete it and stop with an error.
    /// 6. Move the staged file into place.
    ///
    /// Each step that must succeed ends in `|| exit 1`. Cleanup steps end in
    /// `|| true`. `set -e` is not used: a shell ignores it inside a
    /// `( … ) || …` list.
    #[must_use]
    pub fn binary_install_ops(&self) -> Vec<DeployOp> {
        if !self.cfg.enabled || !self.cfg.install_binary {
            return Vec::new();
        }
        vec![DeployOp::Run(RemoteCommand::new(
            MEDIA_INSTALL_BINARY_LABEL,
            self.binary_install_shell(),
        ))]
    }

    fn binary_install_shell(&self) -> String {
        let quote = super::exec::shell_quote;
        let path = &self.cfg.binary_path;
        let bin = quote(path);
        let staged = quote(&format!("{path}.autumn-new"));
        let image = quote(&format!(
            "bluenviron/mediamtx:{MEDIAMTX_KNOWN_GOOD_VERSION}@{MEDIAMTX_KNOWN_GOOD_DIGEST}"
        ));
        let want = quote(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        let mkdir = parent_dir(path)
            .map(|parent| format!("mkdir -p {} || exit 1; ", quote(parent)))
            .unwrap_or_default();
        // Exit 2 = a refusal with its own message; the outer handler then
        // skips the apt / Docker Hub text.
        format!(
            "( if [ -f {bin} ] && [ -x {bin} ]; then exit 0; fi; \
             if [ -e {bin} ] || [ -L {bin} ]; then \
             echo 'refusing to replace '{bin}': it is not an executable file' >&2; exit 2; fi; \
             if ! command -v docker >/dev/null 2>&1; then \
             export DEBIAN_FRONTEND=noninteractive; \
             apt-get -o DPkg::Lock::Timeout=300 update -qq || exit 1; \
             apt-get -o DPkg::Lock::Timeout=300 install -y -qq --no-install-recommends docker.io \
             || exit 1; fi; \
             systemctl start docker >/dev/null 2>&1 || true; \
             {mkdir}\
             rm -rf {staged} || exit 1; \
             cid=$(docker create {image}) || exit 1; \
             rc=0; docker cp \"$cid:/mediamtx\" {staged} || rc=1; \
             docker rm -f \"$cid\" >/dev/null 2>&1 || true; \
             if [ \"$rc\" -ne 0 ]; then rm -rf {staged}; exit 1; fi; \
             chmod 755 {staged} || exit 1; \
             if [ \"$({staged} --version 2>/dev/null)\" != {want} ]; then \
             rm -f {staged}; \
             echo 'the copied MediaMTX is not '{want}' or does not run on this host' >&2; exit 2; fi; \
             mv -f {staged} {bin} || exit 1 \
             ) || {{ rc=$?; \
             if [ \"$rc\" -ne 2 ]; then \
             echo 'autumn deploy could not install MediaMTX {MEDIAMTX_KNOWN_GOOD_VERSION}. It needs an \
             apt-based (Debian/Ubuntu) host, outbound HTTPS to the distro mirrors and to Docker Hub, \
             and an SSH user that can install packages.' >&2; fi; \
             echo 'Install MediaMTX at '{bin}' yourself and set \
             `[media.mediamtx] install_binary = false` in autumn.toml.' >&2; \
             exit 1; }}"
        )
    }

    /// Ordered ops that render `mediamtx.yml` + the systemd unit to the host and
    /// make `MediaMTX` enabled, running, and reloaded — **restarting the daemon
    /// only when the rendered config/unit actually changed, or when the unit is
    /// not already running.**
    ///
    /// Returns an **empty vector when the section is not `enabled`** — the
    /// controller is a no-op for an app that does not provision `MediaMTX`. When
    /// enabled it emits, in order:
    /// 1. `Run` one op that `mkdir -p`s the config's parent dir and, mode
    ///    `0750`, `recordings_dir`. `scp` (the upload transport) does not create
    ///    parents, so the config dir must exist before op 2 writes the staged
    ///    config into it (omitted for a parent-less `config_path`), and
    ///    `recordings_dir` is the `recordPath` root the fail-closed preflight
    ///    expects provisioning to create. Duplicates collapse. The unit's parent
    ///    (`/etc/systemd/system`) is a standard existing dir and needs no
    ///    `mkdir`.
    /// 2. `WriteFile` the rendered `mediamtx.yml` to a **temp path**
    ///    (`<config_path>.autumn-new`, mode `0644`) — a sibling of the live file,
    ///    never over it, so the live config is untouched until op 4 proves a real
    ///    change.
    /// 3. `WriteFile` the rendered unit to a **temp path**
    ///    (`/etc/systemd/system/<unit>.service.autumn-new`, mode `0644`).
    /// 4. `Run` a single idempotent shell command that `cmp -s`-compares each
    ///    staged temp file to its live counterpart (a missing live file counts as
    ///    changed), copies over only the changed ones (tracking a `changed`
    ///    flag), removes the temp files, runs `systemctl daemon-reload` +
    ///    `systemctl enable <unit>.service` (both idempotent), and restarts the
    ///    unit **only if** `changed` is set **or** the unit is not currently
    ///    active (`systemctl is-active --quiet <unit>.service` fails).
    ///
    /// The load-bearing behaviour (issue #2051, Finding H): a pure app redeploy
    /// or a no-op media-config redeploy leaves the config and unit byte-identical
    /// and the unit already active, so **no restart runs** — active
    /// RTMP/WebRTC/HLS sessions are never killed before cutover. A genuine config
    /// or unit change (or a dead/never-started unit) still restarts, so the
    /// freshest config always takes effect.
    #[must_use]
    pub fn ensure_installed_ops(&self) -> Vec<DeployOp> {
        if !self.cfg.enabled {
            return Vec::new();
        }
        let unit = self.cfg.unit_name.clone();
        let config_tmp = format!("{}.autumn-new", self.cfg.config_path);
        let unit_path = self.unit_path();
        let unit_tmp = format!("{unit_path}.autumn-new");
        let mut ops = Vec::new();
        // Create the host dirs provisioning writes into, before anything writes
        // into them (the rustdoc above says why each is needed). `mkdir -p -m`
        // applies the mode only to a dir it creates, so an existing recordings
        // root keeps the mode the operator gave it.
        let config_parent = parent_dir(&self.cfg.config_path);
        // Skip a recordings root this deploy cannot resolve to one host
        // directory: `recordings_dir_writable` fails closed on an empty or
        // relative path and defers a `${...}` one, so creating anything for it
        // here would only make a wrongly-named dir.
        let recordings = Some(self.cfg.recordings_dir.as_str())
            .filter(|dir| unusable_recordings_dir(dir).is_none() && !dir.contains("${"));
        let mut mkdirs: Vec<String> = Vec::new();
        if let Some(parent) = config_parent.filter(|parent| Some(*parent) != recordings) {
            mkdirs.push(format!("mkdir -p {}", super::exec::shell_quote(parent)));
        }
        if let Some(dir) = recordings {
            mkdirs.push(format!(
                "mkdir -p -m 0750 {}",
                super::exec::shell_quote(dir)
            ));
        }
        if !mkdirs.is_empty() {
            ops.push(DeployOp::Run(RemoteCommand::new(
                "media-prepare-dirs",
                mkdirs.join("; "),
            )));
        }
        // Stage config + unit to TEMP paths (never directly over the live files),
        // so op 4 can diff them and restart only on a real change.
        ops.push(DeployOp::WriteFile {
            label: "media-write-config",
            contents: FileContents::Plain(render_mediamtx_yml(&self.cfg)),
            remote_path: config_tmp.clone(),
            mode: Some(0o644),
        });
        ops.push(DeployOp::WriteFile {
            label: "media-write-unit",
            contents: FileContents::Plain(render_mediamtx_unit(&self.cfg)),
            remote_path: unit_tmp.clone(),
            mode: Some(0o644),
        });

        let cfg_live = super::exec::shell_quote(&self.cfg.config_path);
        let cfg_new = super::exec::shell_quote(&config_tmp);
        let unit_live = super::exec::shell_quote(&unit_path);
        let unit_new = super::exec::shell_quote(&unit_tmp);
        // The systemd unit id passed to `systemctl` is a config-derived string
        // (`unit_name`), so shell-quote `<unit_name>.service` (issue #2051, Finding
        // T sweep) — an unquoted unit name containing whitespace/special chars
        // would otherwise split the `systemctl enable/is-active/restart` arguments.
        let unit_service = super::exec::shell_quote(&format!("{unit}.service"));
        // POSIX-sh, fail-fast (`set -e`): compare each staged file to its live
        // counterpart; a missing live file (`cmp` non-zero) or a real diff copies
        // it over and flags `changed`. daemon-reload + enable are idempotent. The
        // unit is restarted ONLY when the config/unit changed OR it is not
        // already active — an app-only redeploy over unchanged, already-running
        // media never interrupts live sessions.
        let install = format!(
            "set -e; \
             changed=0; \
             if ! cmp -s {cfg_new} {cfg_live}; then cp {cfg_new} {cfg_live}; chmod 0644 {cfg_live}; changed=1; fi; \
             if ! cmp -s {unit_new} {unit_live}; then cp {unit_new} {unit_live}; chmod 0644 {unit_live}; changed=1; fi; \
             rm -f {cfg_new} {unit_new}; \
             systemctl daemon-reload; \
             systemctl enable {unit_service}; \
             if [ \"$changed\" -ne 0 ] || ! systemctl is-active --quiet {unit_service}; then \
             systemctl restart {unit_service}; \
             fi",
        );
        ops.push(DeployOp::Run(RemoteCommand::new("media-install", install)));
        ops
    }
}

// ── Doctor checks (fail-closed, executor-driven) ─────────────────────────────
//
// Each check runs a bounded remote command over the injectable `DeployExecutor`
// and returns the shared `PreflightCheck` type. An unverifiable result (the
// command could not run, or its output does not prove the healthy case) reports
// a clear failure — never a silent pass — so a broken media host is caught
// before ingest/playback/encode silently fail at runtime.

/// The check name for the `FFmpeg` preflight.
pub const CHECK_FFMPEG_PREFLIGHT: &str = "media_ffmpeg_preflight";
/// The check name for the recordings-directory-writable probe.
pub const CHECK_RECORDINGS_DIR_WRITABLE: &str = "media_recordings_dir_writable";
/// The check name for the `MediaMTX` port-availability probe.
pub const CHECK_MEDIAMTX_PORTS_AVAILABLE: &str = "media_mediamtx_ports_available";
/// The check name for the `MediaMTX` binary-executable preflight.
pub const CHECK_MEDIAMTX_BINARY: &str = "media_mediamtx_binary";
/// The check name for the `MediaMTX` distinct-listener-port config preflight.
pub const CHECK_MEDIAMTX_PORTS_DISTINCT: &str = "media_mediamtx_ports_distinct";

/// Remediation hint shown when `FFmpeg` does not resolve.
const FFMPEG_HINT: &str =
    "Install FFmpeg on the host (apt package) and set `[media.ffmpeg] bin` to its path";
/// Remediation hint shown when the recordings dir is not writable.
const RECORDINGS_HINT: &str =
    "Create `[media.mediamtx] recordings_dir` on the host and make it writable by the media user";
/// Remediation hint shown when a `MediaMTX` port is occupied or unverifiable.
const PORTS_HINT: &str =
    "Free the conflicting MediaMTX ports on the host (or stop the service already bound to them)";
/// Remediation hint shown when two same-protocol `MediaMTX` listeners are
/// configured onto one port.
const PORTS_DISTINCT_HINT: &str = "Give each MediaMTX TCP listener (`rtmp_port`/`hls_port`/`webrtc_port`/`playback_port`/`api_port`) \
     a distinct port in `[media.mediamtx]`";
/// Remediation hint shown when the `MediaMTX` binary is missing / not executable.
const MEDIAMTX_BINARY_HINT: &str = "Set `[media.mediamtx] install_binary = true` to let `deploy up` install the pinned \
     MediaMTX, or put an executable MediaMTX at `binary_path` yourself";

/// Hint for a `binary_path` that holds something other than an executable.
const MEDIAMTX_BINARY_OCCUPIED_HINT: &str =
    "Remove the file at `binary_path`, or replace it with an executable MediaMTX";

/// Remediation hint shown when `[media.ffmpeg] bin` is env/interpolation
/// indirected and therefore deferred to the service's own runtime resolution.
const FFMPEG_UNRESOLVED_HINT: &str = "Set `[media.ffmpeg] bin` to a concrete path to verify it here, or ensure \
     AUTUMN_MEDIA__FFMPEG__BIN / the referenced ${VAR} is set in the SERVICE \
     environment (the deploy side defers env-indirected paths to runtime)";

/// Grade that `FFmpeg` resolves on the host and reports a version banner
/// (generalized from arroyo's boot preflight).
///
/// The deploy side verifies **only a concrete literal** `[media.ffmpeg] bin` (the
/// default `/usr/bin/ffmpeg` or any explicit path): that value is
/// environment-independent, so probing it here matches the binary the service
/// will run. Any **env/interpolation-indirected** path — an empty value, or one
/// carrying a `${...}` placeholder (including `${AUTUMN_MEDIA__FFMPEG__BIN}`) — is
/// resolved by the deployed service from ITS OWN environment, which the deploy
/// side neither has nor may guess (`build_env_file` does not propagate the
/// operator's local shell). Probing a value resolved against the CLI's local env
/// would be a false pass/fail against a binary the service will not use, so such a
/// path is **deferred to runtime**: a non-passing but **non-blocking** result
/// (`deferred: true`, so [`PreflightCheck::blocking`] is `false`) that surfaces as
/// a warning and does **not** abort `deploy up` — the deployed service resolves
/// `FFmpeg` from its own environment at runtime. Otherwise it runs `<bin>
/// -version` over the executor and requires the standard `ffmpeg version` banner
/// in stdout — a **concrete literal** binary that is missing, not executable, or
/// is not actually `FFmpeg` (no banner) is a clear, **blocking** failure.
/// Fail-closed: a transport error is a blocking failure, not a pass.
#[must_use]
pub fn ffmpeg_preflight(exec: &impl DeployExecutor, ffmpeg_bin: &str) -> PreflightCheck {
    // Env/interpolation indirection the deployed service resolves from its OWN
    // environment (empty value, or a `${...}` placeholder such as
    // `${AUTUMN_MEDIA__FFMPEG__BIN}`): defer rather than resolve/probe against the
    // CLI's local env, which the service does not share.
    if ffmpeg_bin.trim().is_empty() || ffmpeg_bin.contains("${") {
        return PreflightCheck {
            name: CHECK_FFMPEG_PREFLIGHT,
            scope: None,
            passed: false,
            deferred: true,
            detail: format!(
                "FFmpeg path uses env/interpolation indirection (`{ffmpeg_bin}`) resolved in \
                 the service environment; deferring verification to runtime"
            ),
            hint: Some(FFMPEG_UNRESOLVED_HINT),
        };
    }
    let cmd = RemoteCommand::new(
        "media-ffmpeg-preflight",
        format!("{} -version", super::exec::shell_quote(ffmpeg_bin)),
    );
    match exec.run(&cmd) {
        Ok(out) if out.stdout.contains("ffmpeg version") => PreflightCheck {
            name: CHECK_FFMPEG_PREFLIGHT,
            scope: None,
            passed: true,
            deferred: false,
            detail: format!("FFmpeg resolves at `{ffmpeg_bin}` and reports a version banner"),
            hint: None,
        },
        Ok(_) => PreflightCheck {
            name: CHECK_FFMPEG_PREFLIGHT,
            scope: None,
            passed: false,
            deferred: false,
            detail: format!(
                "`{ffmpeg_bin} -version` ran but did not report an `ffmpeg version` banner — \
                 the binary is not FFmpeg"
            ),
            hint: Some(FFMPEG_HINT),
        },
        Err(err) => PreflightCheck {
            name: CHECK_FFMPEG_PREFLIGHT,
            scope: None,
            passed: false,
            deferred: false,
            detail: format!("could not run `{ffmpeg_bin} -version` on the host: {err}"),
            hint: Some(FFMPEG_HINT),
        },
    }
}

/// Reject a `recordings_dir` that cannot name one directory on the host, before
/// any probe runs.
///
/// Three shapes are never usable and each used to fail somewhere later and less
/// clearly: an **empty** value (renders `recordPath: '/%path/…'`, i.e. the
/// filesystem root), a **relative** value (the probe, the `mkdir` and the daemon
/// each resolve it against a different working directory), and a value the
/// deploy side cannot resolve at all. Returns the reason, or `None` when the
/// path is a usable absolute one.
fn unusable_recordings_dir(dir: &str) -> Option<String> {
    if dir.trim().is_empty() {
        return Some("`[media.mediamtx] recordings_dir` is empty".to_owned());
    }
    if !dir.starts_with('/') {
        return Some(format!(
            "`[media.mediamtx] recordings_dir` must be an absolute path, got `{dir}`"
        ));
    }
    None
}

/// Grade that the `MediaMTX` recordings directory is usable on the host —
/// either it exists and is writable, or it is absent under a writable ancestor
/// so provisioning can create it.
///
/// An absent dir passes: `MediaMtxController::ensure_installed_ops` creates it.
/// The check therefore grades whether the path can hold recordings after
/// provisioning, not whether it exists now.
///
/// The probe always exits `0` and prints one marker, so a non-zero exit means
/// only "the command could not run". `present` and `creatable` pass;
/// `unwritable` and `blocked` name the two host failures apart; any other output
/// is unverifiable. Fail-closed: only the two pass markers pass.
///
/// A `${...}` placeholder is **deferred** to runtime, like an indirected
/// `[media.ffmpeg] bin`: the deploy side does not hold the service environment
/// and must not resolve it against its own. An empty or relative path fails
/// closed (see [`unusable_recordings_dir`]).
#[must_use]
pub fn recordings_dir_writable(
    exec: &impl DeployExecutor,
    cfg: &MediaMtxHostConfig,
) -> PreflightCheck {
    let dir = &cfg.recordings_dir;
    let check = |passed: bool, deferred: bool, detail: String| PreflightCheck {
        name: CHECK_RECORDINGS_DIR_WRITABLE,
        scope: None,
        passed,
        deferred,
        detail,
        hint: (!passed && !deferred).then_some(RECORDINGS_HINT),
    };
    // The indirection test comes first: an indirected path legitimately starts
    // with `$`, and only the service can say what it resolves to.
    if dir.contains("${") {
        return check(
            false,
            true,
            format!(
                "recordings directory `{dir}` uses env/interpolation indirection resolved in \
                 the service environment; deferring verification to runtime"
            ),
        );
    }
    if let Some(reason) = unusable_recordings_dir(dir) {
        return check(false, false, reason);
    }
    let quoted = super::exec::shell_quote(dir);
    // Walk to the nearest existing ancestor when the dir itself is absent.
    // The walk uses parameter expansion, not `dirname`: an absent `dirname`, or
    // a path `dirname` reads as an option, would otherwise leave `$d` empty and
    // spin forever. `[ ! -L ]` stops the walk at a dangling symlink or a symlink
    // loop — `[ -e ]` alone resolves it and would step past it as if absent, and
    // `mkdir -p` then fails after cutover.
    let cmd = RemoteCommand::new(
        "media-recordings-dir",
        format!(
            "d={quoted}; \
             if [ -d \"$d\" ]; then if [ -w \"$d\" ]; then printf present; \
             else printf unwritable; fi; \
             else while [ ! -e \"$d\" ] && [ ! -L \"$d\" ]; do \
             case \"$d\" in */*) d=${{d%/*}}; [ -n \"$d\" ] || d=/ ;; *) d=. ;; esac; done; \
             if [ -d \"$d\" ] && [ -w \"$d\" ]; then printf creatable; \
             else printf blocked; fi; fi"
        ),
    );
    match exec.run(&cmd) {
        Ok(output) => match output.stdout.trim() {
            "present" => check(
                true,
                false,
                format!("recordings directory `{dir}` exists and is writable"),
            ),
            "creatable" => check(
                true,
                false,
                format!(
                    "recordings directory `{dir}` does not exist yet and will be created by \
                     provisioning (its nearest existing parent is writable)"
                ),
            ),
            "unwritable" => check(
                false,
                false,
                format!("recordings directory `{dir}` exists but is not writable"),
            ),
            "blocked" => check(
                false,
                false,
                format!(
                    "recordings directory `{dir}` cannot be created: its nearest existing \
                     ancestor is not a writable directory"
                ),
            ),
            other => check(
                false,
                false,
                format!(
                    "recordings directory `{dir}` could not be verified (probe output: \
                     `{other}`)"
                ),
            ),
        },
        Err(err) => check(
            false,
            false,
            format!("recordings directory `{dir}` could not be probed: {err}"),
        ),
    }
}

/// The transport protocol of a listening socket (the `ss` Netid column). Only
/// `tcp` / `udp` are relevant to `MediaMTX`; any other Netid is ignored.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Protocol {
    Tcp,
    Udp,
}

/// The `MediaMTX` ports a bare host must have free before provisioning, **each
/// tagged with the protocol `MediaMTX` actually binds** (issue #2051, Finding I):
/// the five TCP listeners (RTMP, HLS, WebRTC/WHEP, playback, control API) plus the
/// WebRTC local **UDP** port (`webrtcLocalUDPAddress`, ICE host candidates). A
/// listener conflicts only when it matches BOTH the port AND the protocol — a
/// same-port/different-protocol listener (e.g. a TCP service on the UDP-only
/// `:8189`) is not a conflict.
#[must_use]
fn mediamtx_required_ports(cfg: &MediaMtxHostConfig) -> Vec<(u16, Protocol)> {
    vec![
        (cfg.api_port, Protocol::Tcp),
        (cfg.rtmp_port, Protocol::Tcp),
        (cfg.hls_port, Protocol::Tcp),
        (cfg.webrtc_port, Protocol::Tcp),
        (cfg.playback_port, Protocol::Tcp),
        (cfg.webrtc_local_udp, Protocol::Udp),
    ]
}

/// A listening socket parsed from `ss -H -tulnp`: the local port, its transport
/// protocol (the Netid column — issue #2051, Finding I), and the owning process
/// name (`None` when `ss` could not attribute the socket — e.g. the
/// `users:(("name",…))` column is absent because the caller lacked privilege).
#[derive(Debug, Clone, PartialEq, Eq)]
struct ListeningSocket {
    port: u16,
    protocol: Protocol,
    owner: Option<String>,
}

/// Parse the listening sockets out of `ss -H -tulnp` output.
///
/// Each row's 1st column (index 0) is the `Netid` (`tcp` / `udp`) and its 5th
/// column (index 4: `Netid State Recv-Q Send-Q Local Peer …`) is the
/// `Local Address:Port` — the port is the tail after the last `:` (covers IPv4
/// `0.0.0.0:8888`, IPv6 `[::]:8888`, and `*:8888`). The owning process is read
/// from the trailing `users:(("<name>",pid=…,fd=…))` column (added by `-p`), or
/// `None` when that column is absent. Rows whose Netid is not `tcp`/`udp`, or
/// whose port does not parse, are skipped. Pure — unit-tested against captured
/// `ss` output.
#[must_use]
fn parse_listening_sockets(ss_output: &str) -> Vec<ListeningSocket> {
    ss_output
        .lines()
        .filter_map(|line| {
            let mut cols = line.split_whitespace();
            let protocol = match cols.next()? {
                "tcp" => Protocol::Tcp,
                "udp" => Protocol::Udp,
                _ => return None,
            };
            // We already consumed index 0 (Netid); the Local column is index 4,
            // i.e. the 4th remaining column (skip State/Recv-Q/Send-Q).
            let local = cols.nth(3)?;
            let port = local.rsplit(':').next()?.parse::<u16>().ok()?;
            Some(ListeningSocket {
                port,
                protocol,
                owner: process_name_from_ss_line(line),
            })
        })
        .collect()
}

/// Extract the owning process name from an `ss -p` line's
/// `users:(("<name>",pid=…,fd=…))` column, if present. Returns the first process
/// name (a socket shared by several PIDs of the same binary lists them all, but
/// the first name is the owning binary). `None` when the column is absent.
#[must_use]
fn process_name_from_ss_line(line: &str) -> Option<String> {
    // `…users:(("mediamtx",pid=10,fd=7))` → the text after the opening `"`.
    let after_quote = line.split_once("users:((\"")?.1;
    let name = after_quote.split_once('"')?.0;
    (!name.is_empty()).then(|| name.to_owned())
}

/// The managed `MediaMTX` process name, derived from the configured binary path's
/// basename (`/usr/local/bin/mediamtx` → `mediamtx`). This is the `comm` `ss`
/// reports for our own daemon; the kernel truncates `comm` to 15 chars, which the
/// short `mediamtx` name never hits.
#[must_use]
fn managed_process_name(cfg: &MediaMtxHostConfig) -> &str {
    cfg.binary_path
        .rsplit('/')
        .next()
        .filter(|s| !s.is_empty())
        .unwrap_or(cfg.binary_path.as_str())
}

/// Whether our configured `MediaMTX` systemd unit is currently active, via
/// `systemctl is-active --quiet <unit>.service` (exit `0` ⇒ active). Any failure
/// (inactive, unknown unit, or transport error) is read as **not active** — the
/// fail-closed direction for [`mediamtx_ports_available`]'s Finding-J gate. Run
/// at most once per check and the result reused.
#[must_use]
fn unit_is_active(exec: &impl DeployExecutor, unit_name: &str) -> bool {
    // `unit_name` is a config-derived string, so shell-quote `<unit>.service`
    // (issue #2051, Finding T sweep) rather than interpolating it bare.
    let unit_service = super::exec::shell_quote(&format!("{unit_name}.service"));
    let cmd = RemoteCommand::new(
        "media-unit-active",
        format!("systemctl is-active --quiet {unit_service}"),
    );
    exec.run(&cmd).is_ok()
}

/// Grade that the `MediaMTX` ports are free on the host (no *foreign* service is
/// already bound to them).
///
/// **Protocol-aware and unit-verified** (issue #2051, Findings I + J). It runs a
/// single port scan — `ss -H -tulnp` (the `-p` adds the owning-process column,
/// the leading Netid gives the protocol) over the executor — and classifies each
/// configured target `(port, protocol)`:
///
/// - No **same-protocol** listener on the port → free → pass. A same-port but
///   *different-protocol* listener is NOT a conflict: the WebRTC local port
///   `:8189` is UDP-only, so a TCP service on `8189` (or a UDP service on a TCP
///   `MediaMTX` port) never cross-conflicts (Finding I).
/// - Every same-protocol listener owned by *our own* managed `MediaMTX` (process
///   basename of [`MediaMtxHostConfig::binary_path`]) **and** our configured unit
///   reports active (`systemctl is-active --quiet <unit>.service`) → a same-port
///   redeploy of our own daemon, not a conflict → pass (informational note).
/// - A same-protocol listener owned by a `MediaMTX` process while our configured
///   unit is **not** active (manual run, a differently-named service, or a
///   crashed unit) → conflict → fail-closed (Finding J): `enable --now` would
///   start a *second* instance over ports the old process still owns.
/// - A same-protocol listener owned by any *other* process, **or** whose owner
///   cannot be attributed (empty process column — e.g. insufficient privilege) →
///   conflict → fail-closed. `autumn deploy` runs as root managing systemd, so
///   `-p` attribution is normally available; an unattributable listener on a
///   target port is deliberately treated as a conflict rather than assumed benign.
///
/// The unit `is-active` query runs at most **once** (only when a MediaMTX-owned
/// port is actually found) and the result is reused across the held ports.
///
/// Fail-closed: if `ss` cannot run, or its output does not parse into any
/// recognizable listener, the check fails with a "could not verify" message
/// rather than passing blind.
#[must_use]
pub fn mediamtx_ports_available(
    exec: &impl DeployExecutor,
    cfg: &MediaMtxHostConfig,
) -> PreflightCheck {
    // `-p` attaches the owning-process column so a target port held by our own
    // managed MediaMTX (a same-port redeploy) is distinguished from a foreign
    // conflict; the Netid column gives the protocol. Deploy runs as root, so `-p`
    // attribution is available.
    let cmd = RemoteCommand::new("media-ports", "ss -H -tulnp".to_owned());
    let output = match exec.run(&cmd) {
        Ok(out) => out.stdout,
        Err(err) => {
            return PreflightCheck {
                name: CHECK_MEDIAMTX_PORTS_AVAILABLE,
                scope: None,
                passed: false,
                deferred: false,
                detail: format!("could not verify MediaMTX port availability (`ss` failed): {err}"),
                hint: Some(PORTS_HINT),
            };
        }
    };

    let sockets = parse_listening_sockets(&output);
    // Fail-closed: a non-empty `ss` run that yields zero parseable sockets means we
    // could not actually read the host's listener state — treat it as
    // unverifiable rather than a pass.
    if sockets.is_empty() && !output.trim().is_empty() {
        return PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_AVAILABLE,
            scope: None,
            passed: false,
            deferred: false,
            detail: "could not verify MediaMTX port availability: `ss` output did not parse into \
                     any listening ports"
                .to_owned(),
            hint: Some(PORTS_HINT),
        };
    }

    let ours = managed_process_name(cfg);
    // Ports held by a foreign / unattributable same-protocol listener.
    let mut foreign: Vec<u16> = Vec::new();
    // Ports held only by our own MediaMTX process (pending the Finding-J
    // is-active gate below).
    let mut ours_ports: Vec<u16> = Vec::new();
    for (port, protocol) in mediamtx_required_ports(cfg) {
        // A listener conflicts only when it matches BOTH the port AND the
        // protocol (Finding I) — a same-port/different-protocol listener is not
        // a conflict.
        let mut matching = sockets
            .iter()
            .filter(|s| s.port == port && s.protocol == protocol)
            .peekable();
        if matching.peek().is_none() {
            continue; // no same-protocol listener → free
        }
        // A listener owned by our own mediamtx is (provisionally) ours; a foreign
        // process OR an unattributable listener (owner `None`) is a conflict.
        if matching.all(|s| s.owner.as_deref() == Some(ours)) {
            ours_ports.push(port);
        } else {
            foreign.push(port);
        }
    }

    // Finding J: a MediaMTX-owned port counts as "ours" only when our configured
    // unit is actually active. If the ports are held by a `mediamtx` process that
    // is NOT our managed unit (manual run / differently-named service / crashed),
    // treat them as conflicts (fail-closed) — `enable --now` must not start a
    // second instance over them. The is-active query runs at most once.
    let mut held_by_us: Vec<u16> = Vec::new();
    let mut unmanaged: Vec<u16> = Vec::new();
    if !ours_ports.is_empty() {
        if unit_is_active(exec, &cfg.unit_name) {
            held_by_us = ours_ports;
        } else {
            unmanaged = ours_ports;
        }
    }

    let mut conflict_msgs: Vec<String> = Vec::new();
    foreign.sort_unstable();
    foreign.dedup();
    if !foreign.is_empty() {
        conflict_msgs.push(format!(
            "MediaMTX port(s) already in use by another service on the host: {}",
            join_ports(&foreign)
        ));
    }
    unmanaged.sort_unstable();
    unmanaged.dedup();
    if !unmanaged.is_empty() {
        conflict_msgs.push(format!(
            "MediaMTX port(s) {} are held by a `{ours}` process, but the configured \
             `{}` unit is not active — an unmanaged/manual MediaMTX (or a crashed unit) still \
             owns them, so starting our unit would launch a second instance over them",
            join_ports(&unmanaged),
            cfg.unit_name,
        ));
    }
    if !conflict_msgs.is_empty() {
        return PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_AVAILABLE,
            scope: None,
            passed: false,
            deferred: false,
            detail: conflict_msgs.join("; "),
            hint: Some(PORTS_HINT),
        };
    }

    let detail = if held_by_us.is_empty() {
        "all MediaMTX ports are free on the host".to_owned()
    } else {
        format!(
            "MediaMTX ports are available; port(s) {} are already held by our own managed \
             MediaMTX (same-port redeploy)",
            join_ports(&held_by_us),
        )
    };
    PreflightCheck {
        name: CHECK_MEDIAMTX_PORTS_AVAILABLE,
        scope: None,
        passed: true,
        deferred: false,
        detail,
        hint: None,
    }
}

/// Join a sorted list of ports as a `"1935, 8888"` string for messages.
#[must_use]
fn join_ports(ports: &[u16]) -> String {
    ports
        .iter()
        .map(u16::to_string)
        .collect::<Vec<_>>()
        .join(", ")
}

/// Grade that the `MediaMTX` binary is on the host, or that `deploy up` can
/// install it.
///
/// **Why this must run in the pre-cutover preflight:** `provision_media_host`
/// (which writes the unit and runs `systemctl … restart`) runs after the app
/// cutover. A host with no usable binary must fail here, before the app goes
/// live without its media daemon.
///
/// A read-only probe reports one of three states for `binary_path`:
///
/// - `present`: an executable file. Pass.
/// - `absent`: nothing there. Pass when `install_binary` is on, because
///   [`MediaMtxController::binary_install_ops`] installs it before cutover.
///   Fail when it is off.
/// - `occupied`: some other file, a directory or a dangling symlink. Fail.
///   Deploy never replaces it.
///
/// Fail-closed: a transport error or any other output is a failure.
#[must_use]
pub fn mediamtx_binary_preflight(
    exec: &impl DeployExecutor,
    cfg: &MediaMtxHostConfig,
) -> PreflightCheck {
    let bin = &cfg.binary_path;
    // systemd needs an absolute `ExecStart`, and the install needs a real
    // directory. Fail before the probe, so a bad path never reaches cutover.
    if !bin.starts_with('/') {
        return PreflightCheck {
            name: CHECK_MEDIAMTX_BINARY,
            scope: None,
            passed: false,
            deferred: false,
            detail: format!(
                "`[media.mediamtx] binary_path` {bin:?} must be an absolute path — systemd \
                 cannot start a relative `ExecStart`"
            ),
            hint: Some("Set `[media.mediamtx] binary_path` to an absolute path"),
        };
    }
    let quoted = super::exec::shell_quote(bin);
    let cmd = RemoteCommand::new(
        "media-mediamtx-binary",
        format!(
            "if [ -f {quoted} ] && [ -x {quoted} ]; then echo present; \
             elif [ -e {quoted} ] || [ -L {quoted} ]; then echo occupied; \
             else echo absent; fi"
        ),
    );
    let fail = |detail: String, hint: &'static str| PreflightCheck {
        name: CHECK_MEDIAMTX_BINARY,
        scope: None,
        passed: false,
        deferred: false,
        detail,
        hint: Some(hint),
    };
    let pass = |detail: String| PreflightCheck {
        name: CHECK_MEDIAMTX_BINARY,
        scope: None,
        passed: true,
        deferred: false,
        detail,
        hint: None,
    };
    let output = match exec.run(&cmd) {
        Ok(output) => output,
        Err(err) => {
            return fail(
                format!("could not check the MediaMTX binary `{bin}`: {err}"),
                MEDIAMTX_BINARY_HINT,
            );
        }
    };
    match output.stdout.trim() {
        "present" => pass(format!("MediaMTX binary `{bin}` exists and is executable")),
        "absent" if cfg.install_binary => pass(format!(
            "MediaMTX binary `{bin}` is absent; `deploy up` installs MediaMTX \
             {MEDIAMTX_KNOWN_GOOD_VERSION} there before cutover"
        )),
        "absent" => fail(
            format!(
                "MediaMTX binary `{bin}` is absent and `install_binary = false` — MediaMTX \
                 cannot start, so the post-cutover `systemctl restart` would fail"
            ),
            MEDIAMTX_BINARY_HINT,
        ),
        "occupied" => fail(
            format!(
                "`{bin}` exists but is not an executable file — deploy does not replace it, \
                 and the post-cutover `systemctl restart` would fail"
            ),
            MEDIAMTX_BINARY_OCCUPIED_HINT,
        ),
        other => fail(
            format!(
                "could not check the MediaMTX binary `{bin}`: unexpected probe output {other:?}"
            ),
            MEDIAMTX_BINARY_HINT,
        ),
    }
}

/// Grade that the configured `MediaMTX` **TCP listener ports are all distinct**
/// (issue #2051, Finding P) — a **pure config** check, no executor, no host I/O.
///
/// `MediaMTX` runs a separate server per protocol, each binding its own TCP
/// socket: RTMP (`rtmp_port`), HLS (`hls_port`), WebRTC/WHEP (`webrtc_port`),
/// recording-playback (`playback_port`), and the control API (`api_port`). If two
/// of those are configured to the same value, the rendered `mediamtx.yml` asks two
/// servers to bind one port. [`mediamtx_ports_available`]'s `ss` scan can still
/// pass on a fresh host (nothing is listening yet), so the conflict would only
/// surface as a failed **deferred** `systemctl restart` AFTER app cutover — the
/// worst time. This check catches it up front and fails closed.
///
/// The WebRTC local **UDP** port (`webrtc_local_udp`) is a *separate protocol
/// namespace*, so it may legitimately equal a TCP port and is deliberately not
/// compared against the TCP set (mirroring [`mediamtx_required_ports`]'s
/// protocol-aware conflict rule). Fail-closed message names the duplicated value
/// and every field bound to it.
#[must_use]
pub fn mediamtx_ports_distinct(cfg: &MediaMtxHostConfig) -> PreflightCheck {
    // The five TCP listeners MediaMTX binds — each its own server, so all must be
    // distinct. `webrtc_local_udp` is UDP and intentionally excluded.
    let tcp: [(u16, &str); 5] = [
        (cfg.rtmp_port, "rtmp_port"),
        (cfg.hls_port, "hls_port"),
        (cfg.webrtc_port, "webrtc_port"),
        (cfg.playback_port, "playback_port"),
        (cfg.api_port, "api_port"),
    ];
    // Group fields by port (BTreeMap → deterministic ascending-port ordering) and
    // report every port claimed by more than one field.
    let mut by_port: std::collections::BTreeMap<u16, Vec<&str>> = std::collections::BTreeMap::new();
    for (port, field) in tcp {
        by_port.entry(port).or_default().push(field);
    }
    let conflicts: Vec<String> = by_port
        .iter()
        .filter(|(_, fields)| fields.len() > 1)
        .map(|(port, fields)| format!("TCP port {port} is bound by {}", fields.join(", ")))
        .collect();

    if conflicts.is_empty() {
        PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_DISTINCT,
            scope: None,
            passed: true,
            deferred: false,
            detail: "all MediaMTX TCP listener ports are distinct".to_owned(),
            hint: None,
        }
    } else {
        PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_DISTINCT,
            scope: None,
            passed: false,
            deferred: false,
            detail: format!(
                "MediaMTX listener ports must be unique per protocol, but {} — two servers \
                 cannot bind the same TCP port, so `systemctl restart mediamtx` would fail after \
                 cutover",
                conflicts.join("; ")
            ),
            hint: Some(PORTS_DISTINCT_HINT),
        }
    }
}

/// The check name for the listener-port ↔ app-base-URL consistency preflight.
pub const CHECK_MEDIAMTX_PORTS_MATCH_BASES: &str = "media_mediamtx_ports_match_bases";

/// Remediation hint shown when a listener port and the app's base URL disagree.
const PORTS_MATCH_BASES_HINT: &str = "Update the app-side `[media.mediamtx] *_base` URL to the port the daemon is provisioned on \
     (or drop the explicit port from the base URL if a reverse proxy fronts it)";

/// What a base URL's authority says about its port.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BasePort {
    /// The URL names no port — an implicit `80`/`443`, which the deploy side
    /// cannot compare against a listener port.
    Implicit,
    /// The URL names this port.
    Explicit(u16),
    /// The URL names something in the port position that is not a port. Never a
    /// valid shape, so it is reported rather than skipped.
    Malformed,
}

/// Read the port an app-side base URL addresses.
///
/// Strips the scheme, any `user:pw@` userinfo, and the path/query/fragment, then
/// reads a `:port` off the authority (`[::1]:8888` included).
fn explicit_port(url: &str) -> BasePort {
    let after_scheme = url.split_once("://").map_or(url, |(_, rest)| rest);
    let Some(authority) = after_scheme.split(['/', '?', '#']).next() else {
        return BasePort::Implicit;
    };
    let host_port = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
    // An IPv6 literal's own colons are inside the brackets; only a `:` after the
    // closing `]` introduces a port.
    let port = match host_port.rsplit_once(']') {
        Some((_, rest)) => match rest.strip_prefix(':') {
            Some(port) => port,
            None if rest.is_empty() => return BasePort::Implicit,
            None => return BasePort::Malformed,
        },
        None => match host_port.split_once(':') {
            Some((_, port)) => port,
            None => return BasePort::Implicit,
        },
    };
    port.parse().map_or(BasePort::Malformed, BasePort::Explicit)
}

/// Whether a base URL addresses the `MediaMTX` listener **directly**, i.e. its
/// host is loopback.
///
/// A base on any other host is fronted by a reverse proxy or CDN, whose public
/// port has no reason to equal the daemon's listener port
/// (`https://media.example.com:8443` → `hls_port = 8888` is a correct
/// production shape). Only a loopback base is comparable, so only a loopback
/// base can produce a mismatch.
fn addresses_listener_directly(url: &str) -> bool {
    let after_scheme = url.split_once("://").map_or(url, |(_, rest)| rest);
    let authority = after_scheme
        .split(['/', '?', '#'])
        .next()
        .unwrap_or_default();
    let host_port = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
    let host = match host_port.rsplit_once(']') {
        Some((host, _)) => host.trim_start_matches('['),
        None => host_port.split(':').next().unwrap_or_default(),
    };
    matches!(
        host.trim().to_ascii_lowercase().as_str(),
        "127.0.0.1" | "localhost" | "::1"
    )
}

/// Grade that every provisioned `MediaMTX` listener port agrees with the app's
/// base URL for that listener.
///
/// A **pure config** check (no host I/O), like [`mediamtx_ports_distinct`]. The
/// `*_port` fields provision the daemon's listeners; the app and its clients
/// call those listeners on the `[media.mediamtx] *_base` URLs. Both live in one
/// TOML table, and nothing else keeps them in step: customizing a port without
/// updating the base points the app at an origin the daemon no longer binds, and
/// the deploy succeeds while every request fails at runtime.
///
/// Comparison rules:
/// - A base is **compared** only when it is set in TOML, addresses loopback, and
///   names a port. A mismatch there is a contradiction between two values the
///   operator wrote, so it **blocks**.
/// - A base on any other host is fronted by a proxy or CDN
///   ([`addresses_listener_directly`]) and is **skipped** — its public port has
///   no reason to equal the listener port.
/// - A base with no port (implicit `80`/`443`) is likewise skipped.
/// - A base naming a **malformed** port is reported: it is never a valid shape.
/// - An **unset** base is **deferred**, not compared, when its plugin default
///   disagrees with the provisioned port. The app may take that base from
///   `AUTUMN_MEDIA__MEDIAMTX__*_BASE` in the service environment, which the
///   deploy side does not hold — so this is a loud non-blocking warning, like an
///   indirected [`ffmpeg_preflight`], never a blocked deploy.
/// - A `${...}` placeholder is deferred for the same reason.
/// - A real mismatch **outranks** any deferral.
/// - The two probe bases are graded only when explicitly set (unset, they fall
///   back to their public base, which is already graded).
#[must_use]
pub fn mediamtx_ports_match_bases(cfg: &MediaMtxHostConfig) -> PreflightCheck {
    let mut mismatches: Vec<String> = Vec::new();
    let mut deferred: Vec<String> = Vec::new();
    for (port, port_field, base_field, base, set_in_toml) in port_base_pairs(cfg) {
        let Some(base) = base else { continue };
        if base.contains("${") {
            deferred.push(format!("`{base_field}` uses env/interpolation indirection"));
            continue;
        }
        let base_port = match explicit_port(base) {
            BasePort::Explicit(port) => port,
            BasePort::Implicit => continue,
            BasePort::Malformed => {
                mismatches.push(format!("`{base_field}` names no valid port (`{base}`)"));
                continue;
            }
        };
        if base_port == port {
            continue;
        }
        if !set_in_toml {
            // The default disagrees with the provisioned port, but the app may
            // be told otherwise by its own environment. Warn, never block.
            deferred.push(format!(
                "`{port_field}` provisions port {port} while the default `{base_field}` \
                 (`{base}`) calls port {base_port}; set `{base_field}` here, or in the \
                 service environment"
            ));
            continue;
        }
        if !addresses_listener_directly(base) {
            // A proxied base's public port is its own business.
            continue;
        }
        mismatches.push(format!(
            "`{port_field}` provisions port {port} but `{base_field}` calls port \
             {base_port} (`{base}`)"
        ));
    }

    if !mismatches.is_empty() {
        return PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_MATCH_BASES,
            scope: None,
            passed: false,
            deferred: false,
            detail: format!(
                "MediaMTX listener ports and the app's base URLs disagree, so the deployed app \
                 would call an origin the daemon does not bind: {}",
                mismatches.join("; ")
            ),
            hint: Some(PORTS_MATCH_BASES_HINT),
        };
    }
    if !deferred.is_empty() {
        return PreflightCheck {
            name: CHECK_MEDIAMTX_PORTS_MATCH_BASES,
            scope: None,
            passed: false,
            deferred: true,
            detail: format!(
                "every comparable MediaMTX base URL matches its listener port, but some are \
                 resolved in the service environment, so verification is deferred to runtime: {}",
                deferred.join("; ")
            ),
            hint: None,
        };
    }
    PreflightCheck {
        name: CHECK_MEDIAMTX_PORTS_MATCH_BASES,
        scope: None,
        passed: true,
        deferred: false,
        detail: "every MediaMTX base URL matches its provisioned listener port".to_owned(),
        hint: None,
    }
}

/// The rows [`mediamtx_ports_match_bases`] grades:
/// `(listener port, port field, base field, effective base URL, set in TOML)`.
///
/// An unset public base resolves to the mirrored plugin default, flagged
/// `false` so the caller can defer rather than block on it. An unset **probe**
/// base is `None` and skipped: it falls back to its public base, which the row
/// above it already grades.
fn port_base_pairs(
    cfg: &MediaMtxHostConfig,
) -> [(u16, &'static str, &'static str, Option<&str>, bool); 7] {
    /// Resolve one public base to `(effective value, was set in TOML)`.
    const fn public<'a>(configured: Option<&'a str>, default: &'a str) -> (Option<&'a str>, bool) {
        match configured {
            Some(base) => (Some(base), true),
            None => (Some(default), false),
        }
    }
    let (api, api_set) = public(cfg.api_base.as_deref(), DEFAULT_API_BASE);
    let (rtmp, rtmp_set) = public(cfg.rtmp_base.as_deref(), DEFAULT_RTMP_BASE);
    let (hls, hls_set) = public(cfg.hls_base.as_deref(), DEFAULT_HLS_BASE);
    let (webrtc, webrtc_set) = public(cfg.webrtc_base.as_deref(), DEFAULT_WEBRTC_BASE);
    let (playback, playback_set) = public(cfg.playback_base.as_deref(), DEFAULT_PLAYBACK_BASE);
    [
        (cfg.api_port, "api_port", "api_base", api, api_set),
        (cfg.rtmp_port, "rtmp_port", "rtmp_base", rtmp, rtmp_set),
        (cfg.hls_port, "hls_port", "hls_base", hls, hls_set),
        (
            cfg.webrtc_port,
            "webrtc_port",
            "webrtc_base",
            webrtc,
            webrtc_set,
        ),
        (
            cfg.playback_port,
            "playback_port",
            "playback_base",
            playback,
            playback_set,
        ),
        (
            cfg.hls_port,
            "hls_port",
            "hls_probe_base",
            cfg.hls_probe_base.as_deref(),
            true,
        ),
        (
            cfg.playback_port,
            "playback_port",
            "playback_probe_base",
            cfg.playback_probe_base.as_deref(),
            true,
        ),
    ]
}

/// Collect the six `MediaMTX` host doctor checks against a live executor.
///
/// Runs the two pure **config** checks first (no host I/O, so a duplicated-port
/// or port/base-URL-inconsistent config aborts before any probe), then the
/// `FFmpeg` preflight
/// (against `ffmpeg_bin`), the `MediaMTX` binary-executable preflight, the
/// recordings-dir probe, and the port-availability probe — returning the shared
/// [`PreflightCheck`] results so a caller that already holds a [`DeployExecutor`]
/// can grade a media host the same way `autumn deploy check` grades the app host.
/// `ffmpeg_bin` is the resolved `[media.ffmpeg] bin` (the plugin's default is
/// `/usr/bin/ffmpeg`).
#[must_use]
pub fn collect_media_doctor_checks(
    exec: &impl DeployExecutor,
    cfg: &MediaMtxHostConfig,
    ffmpeg_bin: &str,
) -> Vec<PreflightCheck> {
    vec![
        mediamtx_ports_distinct(cfg),
        mediamtx_ports_match_bases(cfg),
        ffmpeg_preflight(exec, ffmpeg_bin),
        mediamtx_binary_preflight(exec, cfg),
        recordings_dir_writable(exec, cfg),
        mediamtx_ports_available(exec, cfg),
    ]
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::cell::RefCell;
    use std::path::Path;

    use super::super::exec::{CommandOutput, DeployExecError};

    /// A recording fake executor mirroring `exec.rs`'s test fake (which is
    /// private to that module): records `run` calls in order and returns
    /// scripted stdout / scripted failures per command label, so the controller
    /// ops and the doctor checks are assertable with no live host.
    #[derive(Default)]
    struct RecordingExecutor {
        run_labels: RefCell<Vec<&'static str>>,
        /// The rendered shell of each `run`, so a probe's own command text is
        /// assertable (not just that it was invoked).
        run_shells: RefCell<Vec<String>>,
        fail_labels: Vec<&'static str>,
        stdout_by_label: Vec<(&'static str, String)>,
    }

    impl RecordingExecutor {
        fn new() -> Self {
            Self::default()
        }

        fn failing_on(label: &'static str) -> Self {
            Self {
                fail_labels: vec![label],
                ..Self::default()
            }
        }

        fn with_stdout(mut self, label: &'static str, stdout: impl Into<String>) -> Self {
            self.stdout_by_label.push((label, stdout.into()));
            self
        }

        /// Chainable variant of [`Self::failing_on`]: mark an additional command
        /// label as failing (so a check can succeed on one command and fail on
        /// another — e.g. `ss` succeeds but `systemctl is-active` reports the unit
        /// inactive).
        fn failing(mut self, label: &'static str) -> Self {
            self.fail_labels.push(label);
            self
        }

        fn labels(&self) -> Vec<&'static str> {
            self.run_labels.borrow().clone()
        }

        /// The rendered shell of the one command run with `label`.
        fn shell_for(&self, label: &str) -> String {
            let index = self
                .run_labels
                .borrow()
                .iter()
                .position(|recorded| *recorded == label)
                .unwrap_or_else(|| panic!("no command ran with label `{label}`"));
            self.run_shells.borrow()[index].clone()
        }
    }

    impl DeployExecutor for RecordingExecutor {
        fn run(&self, cmd: &RemoteCommand) -> Result<CommandOutput, DeployExecError> {
            self.run_labels.borrow_mut().push(cmd.label);
            self.run_shells.borrow_mut().push(cmd.shell.clone());
            if self.fail_labels.contains(&cmd.label) {
                return Err(DeployExecError::CommandFailed {
                    label: cmd.label,
                    message: "scripted failure".to_owned(),
                });
            }
            let stdout = self
                .stdout_by_label
                .iter()
                .find(|(l, _)| *l == cmd.label)
                .map(|(_, out)| out.clone())
                .unwrap_or_default();
            Ok(CommandOutput {
                stdout,
                stderr: String::new(),
            })
        }

        fn upload(
            &self,
            _local: &Path,
            _remote_path: &str,
            _mode: Option<u32>,
        ) -> Result<(), DeployExecError> {
            Ok(())
        }
    }

    // ── Config defaults / deserialization ────────────────────────────────────

    #[test]
    fn config_defaults_are_the_localhost_disabled_shape() {
        let cfg = MediaMtxHostConfig::default();
        assert!(!cfg.enabled, "controller is disabled by default");
        assert_eq!(cfg.api_port, MEDIAMTX_API_PORT);
        assert_eq!(cfg.rtmp_port, MEDIAMTX_RTMP_PORT);
        assert_eq!(cfg.hls_port, MEDIAMTX_HLS_PORT);
        assert_eq!(cfg.webrtc_port, MEDIAMTX_WEBRTC_PORT);
        assert_eq!(cfg.playback_port, MEDIAMTX_PLAYBACK_PORT);
        assert_eq!(cfg.webrtc_local_udp, MEDIAMTX_WEBRTC_LOCAL_UDP_PORT);
        assert_eq!(cfg.recordings_dir, "/recordings");
        assert_eq!(cfg.record_delete_after, "72h");
        assert_eq!(cfg.config_path, "/etc/mediamtx/mediamtx.yml");
        assert_eq!(cfg.binary_path, "/usr/local/bin/mediamtx");
        assert_eq!(cfg.unit_name, "mediamtx");
        assert!(cfg.webrtc_additional_hosts.is_empty());
    }

    /// Parse a raw `autumn.toml` string and deserialize its `[media]` subtree via
    /// the production [`media_host_config_from_value`] path — the deploy loader
    /// merges raw `toml::Value`s and calls that fn, so tests exercise the same
    /// deserializer over a single (unmerged) document.
    fn host_cfg_from_toml_str(s: &str) -> Result<(MediaMtxHostConfig, String), toml::de::Error> {
        media_host_config_from_value(toml::from_str(s)?)
    }

    #[test]
    fn empty_toml_yields_disabled_defaults() {
        let (cfg, bin) = host_cfg_from_toml_str("").expect("empty toml parses");
        assert!(!cfg.enabled);
        assert_eq!(cfg.api_port, MEDIAMTX_API_PORT);
        assert_eq!(bin, DEFAULT_FFMPEG_BIN);
    }

    #[test]
    fn toml_reads_the_media_mediamtx_section() {
        let toml = "\
[media.mediamtx]
enabled = true
recordings_dir = \"/srv/recordings\"
record_delete_after = \"48h\"
webrtc_additional_hosts = [\"stream.example.com\", \"203.0.113.7\"]
unit_name = \"mediamtx-prod\"
";
        let (cfg, _) = host_cfg_from_toml_str(toml).expect("parses");
        assert!(cfg.enabled);
        assert_eq!(cfg.recordings_dir, "/srv/recordings");
        assert_eq!(cfg.record_delete_after, "48h");
        assert_eq!(
            cfg.webrtc_additional_hosts,
            vec!["stream.example.com".to_owned(), "203.0.113.7".to_owned()]
        );
        assert_eq!(cfg.unit_name, "mediamtx-prod");
        // Unspecified fields keep their defaults.
        assert_eq!(cfg.hls_port, MEDIAMTX_HLS_PORT);
    }

    #[test]
    fn toml_without_media_table_is_default() {
        let (cfg, _) = host_cfg_from_toml_str("[deploy]\nhost = \"h\"\n").expect("parses");
        assert!(!cfg.enabled);
    }

    #[test]
    fn ffmpeg_bin_defaults_and_reads_from_media_ffmpeg() {
        assert_eq!(
            host_cfg_from_toml_str("").expect("empty parses").1,
            DEFAULT_FFMPEG_BIN
        );
        let (_, bin) =
            host_cfg_from_toml_str("[media.ffmpeg]\nbin = \"/opt/ffmpeg\"\n").expect("parses");
        assert_eq!(bin, "/opt/ffmpeg");
    }

    #[test]
    fn media_config_fails_closed_on_malformed_mediamtx_section() {
        // Finding C: a PRESENT `[media.mediamtx]` with an ill-typed value is a hard
        // error, so the caller must abort rather than silently disable provisioning.
        // `enabled` as a string, not a bool:
        assert!(host_cfg_from_toml_str("[media.mediamtx]\nenabled = \"yes\"\n").is_err());
        // A port field as a non-number:
        assert!(host_cfg_from_toml_str("[media.mediamtx]\napi_port = \"nope\"\n").is_err());
    }

    #[test]
    fn media_config_fails_closed_on_malformed_ffmpeg_section() {
        // Finding C: a malformed `[media.ffmpeg]` is also a hard error — the deploy
        // deserializer reads the whole `[media]` subtree, so a broken ffmpeg table
        // cannot be tricked into the disabled default.
        let toml = "[media.ffmpeg]\nbin = 123\n";
        assert!(host_cfg_from_toml_str(toml).is_err());
    }

    // ── mediamtx.yml rendering ───────────────────────────────────────────────

    #[test]
    fn rendered_yml_contains_every_port() {
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        // The control API binds loopback-only (Finding M) — server-side control
        // plane for the co-located app, excluded from browser-facing CSP origins.
        assert!(
            yml.contains("apiAddress: 127.0.0.1:9997"),
            "api port: {yml}"
        );
        assert!(yml.contains("playbackAddress: :9996"), "playback: {yml}");
        assert!(yml.contains("rtmpAddress: :1935"), "rtmp: {yml}");
        assert!(yml.contains("hlsAddress: :8888"), "hls: {yml}");
        assert!(yml.contains("webrtcAddress: :8889"), "webrtc: {yml}");
        assert!(yml.contains("webrtcLocalUDPAddress: :8189"), "udp: {yml}");
    }

    #[test]
    fn rendered_yml_binds_api_to_loopback_but_leaves_public_listeners_open() {
        // Finding M: the control API is the server-side control plane (the
        // co-located app reaches it at 127.0.0.1) and is excluded from the
        // browser-facing CSP origins, so it must NOT listen on every interface.
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        assert!(
            yml.contains("apiAddress: 127.0.0.1:9997"),
            "control API must bind loopback only: {yml}"
        );
        assert!(
            !yml.contains("apiAddress: :9997"),
            "control API must NOT bind all interfaces: {yml}"
        );
        // The viewer/ingest-facing listeners MUST stay bound on all interfaces so
        // encoders and browsers can reach them — none is loopback-bound.
        for public in [
            "rtmpAddress: :1935",
            "hlsAddress: :8888",
            "webrtcAddress: :8889",
            "playbackAddress: :9996",
        ] {
            assert!(yml.contains(public), "public listener {public}: {yml}");
        }
        assert!(
            !yml.contains("rtmpAddress: 127.0.0.1")
                && !yml.contains("hlsAddress: 127.0.0.1")
                && !yml.contains("webrtcAddress: 127.0.0.1")
                && !yml.contains("playbackAddress: 127.0.0.1"),
            "no public listener may be loopback-bound: {yml}"
        );
    }

    #[test]
    fn rendered_yml_has_ll_hls_recording_and_webrtc_config() {
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        // LL-HLS window copied from arroyo.
        assert!(yml.contains("hlsVariant: lowLatency"));
        assert!(yml.contains("hlsPartDuration: 200ms"));
        assert!(yml.contains("hlsSegmentDuration: 2s"));
        assert!(yml.contains("hlsSegmentCount: 60"));
        // Recording config under the configured recordings dir.
        assert!(yml.contains("record: true"));
        // Config-derived path/duration values are single-quoted YAML scalars
        // (Finding T sweep) so a value with whitespace/special chars can't break
        // parsing.
        assert!(yml.contains("recordPath: '/recordings/%path/%Y-%m-%d_%H-%M-%S-%f'"));
        assert!(yml.contains("recordFormat: fmp4"));
        assert!(yml.contains("recordDeleteAfter: '72h'"));
        // WebRTC config.
        assert!(yml.contains("webrtc: true"));
        assert!(yml.contains("webrtcAdditionalHosts: []"));
    }

    #[test]
    fn rendered_yml_uses_current_mediamtx_cors_schema() {
        // Finding S (issue #2051): MediaMTX rejects unknown config fields at
        // startup, and this config is installed by a post-cutover `systemctl
        // restart`. The HLS CORS key is the plural array `hlsAllowOrigins` in the
        // current schema (v1.19.x / `bluenviron/mediamtx:1`); the singular scalar
        // `hlsAllowOrigin` is a deprecated alias, so the template must emit the
        // current plural form, matching the already-plural `webrtcAllowOrigins`.
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        assert!(
            yml.contains("hlsAllowOrigins: ['*']"),
            "HLS CORS must use the current plural array key: {yml}"
        );
        assert!(
            !yml.contains("hlsAllowOrigin:"),
            "the deprecated singular hlsAllowOrigin key must not be rendered: {yml}"
        );
        assert!(
            yml.contains("webrtcAllowOrigins: ['*']"),
            "WebRTC CORS must use the plural array key: {yml}"
        );
    }

    #[test]
    fn rendered_yml_disables_unmanaged_protocols_but_keeps_managed_ones() {
        // Finding E/Q: MediaMTX fills omitted protocol flags from its own defaults, so
        // RTSP (:8554), SRT (:8890), and MoQ (:8892) would otherwise listen even though
        // the deploy plan never declares/checks those ports — an occupied default port
        // could then fail `systemctl restart mediamtx` after preflight passed. The
        // template pins every unmanaged default-on protocol OFF so the unit only opens
        // the declared ports.
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        assert!(yml.contains("rtsp: no"), "RTSP must be disabled: {yml}");
        assert!(yml.contains("srt: no"), "SRT must be disabled: {yml}");
        assert!(yml.contains("moq: no"), "MoQ must be disabled: {yml}");
        assert!(yml.contains("metrics: no"), "metrics must be off: {yml}");
        assert!(yml.contains("pprof: no"), "pprof must be off: {yml}");
        // The managed protocols stay enabled (RTMP/HLS/WebRTC/API/playback).
        assert!(yml.contains("rtmp: true"), "rtmp managed: {yml}");
        assert!(yml.contains("hls: true"), "hls managed: {yml}");
        assert!(yml.contains("webrtc: true"), "webrtc managed: {yml}");
        assert!(yml.contains("api: true"), "api managed: {yml}");
        assert!(yml.contains("playback: true"), "playback managed: {yml}");
    }

    #[test]
    fn rendered_yml_has_both_live_and_room_path_matchers() {
        let yml = render_mediamtx_yml(&MediaMtxHostConfig::default());
        assert!(yml.contains("~^live/.+$:"), "broadcast matcher: {yml}");
        // The autumn-media addition arroyo lacks.
        assert!(yml.contains("~^room/.+$:"), "room matcher: {yml}");
    }

    #[test]
    fn rendered_yml_honors_custom_ports_dirs_and_hosts() {
        let cfg = MediaMtxHostConfig {
            api_port: 19997,
            hls_port: 18888,
            recordings_dir: "/data/rec".to_owned(),
            record_delete_after: "24h".to_owned(),
            webrtc_additional_hosts: vec!["a.example.com".to_owned(), "b.example.com".to_owned()],
            ..MediaMtxHostConfig::default()
        };
        let yml = render_mediamtx_yml(&cfg);
        assert!(yml.contains("apiAddress: 127.0.0.1:19997"));
        assert!(yml.contains("hlsAddress: :18888"));
        assert!(yml.contains("recordPath: '/data/rec/%path/%Y-%m-%d_%H-%M-%S-%f'"));
        assert!(yml.contains("recordDeleteAfter: '24h'"));
        assert!(yml.contains("webrtcAdditionalHosts: ['a.example.com', 'b.example.com']"));
    }

    // ── systemd unit rendering ───────────────────────────────────────────────

    #[test]
    fn rendered_unit_uses_binary_and_config_paths_and_restarts_always() {
        let unit = render_mediamtx_unit(&MediaMtxHostConfig::default());
        // Finding T: the binary/config paths are systemd double-quoted so a path
        // with whitespace stays one argument.
        assert!(
            unit.contains("ExecStart=\"/usr/local/bin/mediamtx\" \"/etc/mediamtx/mediamtx.yml\"\n"),
            "ExecStart: {unit}"
        );
        assert!(unit.contains("Restart=always\n"), "restart: {unit}");
        assert!(unit.contains("After=network-online.target\n"));
        assert!(unit.contains("WantedBy=multi-user.target\n"));
    }

    #[test]
    fn rendered_unit_quotes_whitespace_paths_in_execstart() {
        // Finding T (issue #2051): a `binary_path`/`config_path` containing
        // whitespace must be systemd double-quoted in `ExecStart`, or systemd
        // parses `ExecStart=/opt/media mtx/mediamtx ...` as the truncated command
        // `/opt/media` and the deferred post-cutover `systemctl restart` fails.
        let cfg = MediaMtxHostConfig {
            binary_path: "/opt/media mtx/mediamtx".to_owned(),
            config_path: "/etc/media mtx/mediamtx.yml".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let unit = render_mediamtx_unit(&cfg);
        // Each path is wrapped in systemd double quotes → one argument each.
        assert!(
            unit.contains(
                "ExecStart=\"/opt/media mtx/mediamtx\" \"/etc/media mtx/mediamtx.yml\"\n"
            ),
            "ExecStart must double-quote whitespace paths: {unit}"
        );
        // And NOT the bare, space-splitting form.
        assert!(
            !unit.contains("ExecStart=/opt/media mtx/mediamtx"),
            "ExecStart must not interpolate the path unquoted: {unit}"
        );
    }

    #[test]
    fn systemd_quote_escapes_backslash_and_double_quote() {
        // systemd C-style quoting: `\\` and `\"` are the escapes inside a
        // double-quoted token, escaped in that order so the value round-trips.
        assert_eq!(systemd_quote("/usr/bin/mediamtx"), "\"/usr/bin/mediamtx\"");
        assert_eq!(systemd_quote("/opt/a b/mtx"), "\"/opt/a b/mtx\"");
        assert_eq!(systemd_quote(r#"/a"b"#), "\"/a\\\"b\"");
        assert_eq!(systemd_quote(r"/a\b"), "\"/a\\\\b\"");
    }

    #[test]
    fn rendered_yml_quotes_whitespace_and_special_config_values() {
        // Finding T sweep: config-derived path/duration values are single-quoted
        // YAML scalars, so a recordings dir with a space or a YAML structural char
        // stays a single scalar instead of breaking `mediamtx.yml` parsing.
        let cfg = MediaMtxHostConfig {
            recordings_dir: "/mnt/rec ords".to_owned(),
            record_delete_after: "72h".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let yml = render_mediamtx_yml(&cfg);
        assert!(
            yml.contains("recordPath: '/mnt/rec ords/%path/%Y-%m-%d_%H-%M-%S-%f'"),
            "recordPath must be single-quoted: {yml}"
        );
        // A `#` in the recordings dir would otherwise start a YAML comment.
        let hashy = MediaMtxHostConfig {
            recordings_dir: "/mnt/rec#1".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let yml = render_mediamtx_yml(&hashy);
        assert!(
            yml.contains("recordPath: '/mnt/rec#1/%path/%Y-%m-%d_%H-%M-%S-%f'"),
            "a `#` path must be quoted, not truncated by a YAML comment: {yml}"
        );
    }

    #[test]
    fn install_op_quotes_whitespace_unit_name() {
        // Finding T sweep: the config-derived unit name is shell-quoted in every
        // systemctl invocation, so a unit name with whitespace does not split the
        // arguments.
        let cfg = MediaMtxHostConfig {
            enabled: true,
            unit_name: "media mtx".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();
        let DeployOp::Run(cmd) = &ops[3] else {
            panic!("op 3 must be the install Run op");
        };
        assert!(
            cmd.shell.contains("systemctl enable 'media mtx.service'"),
            "unit name must be shell-quoted: {}",
            cmd.shell
        );
        assert!(
            cmd.shell
                .contains("systemctl is-active --quiet 'media mtx.service'"),
            "{}",
            cmd.shell
        );
        assert!(
            cmd.shell.contains("systemctl restart 'media mtx.service'"),
            "{}",
            cmd.shell
        );
    }

    // ── Controller op sequence ───────────────────────────────────────────────

    #[test]
    fn controller_noops_when_disabled() {
        let controller = MediaMtxController::new(MediaMtxHostConfig::default());
        assert!(
            controller.ensure_installed_ops().is_empty(),
            "a disabled controller emits no ops"
        );
    }

    #[test]
    fn controller_emits_write_write_install_when_enabled() {
        let cfg = MediaMtxHostConfig {
            enabled: true,
            ..MediaMtxHostConfig::default()
        };
        let controller = MediaMtxController::new(cfg);
        let ops = controller.ensure_installed_ops();
        assert_eq!(ops.len(), 4);

        // op 0: mkdir -p the config's parent dir BEFORE scp writes into it.
        let DeployOp::Run(mkdir) = &ops[0] else {
            panic!("op 0 must be the mkdir Run op, got {:?}", ops[0]);
        };
        assert_eq!(mkdir.label, "media-prepare-dirs");
        // Both the config parent AND the recordings root, in one op. `-m 0750`
        // keeps a freshly-created recordings root off world-readable.
        assert_eq!(
            mkdir.shell,
            "mkdir -p '/etc/mediamtx'; mkdir -p -m 0750 '/recordings'"
        );

        // Finding H: op 1 writes mediamtx.yml to a TEMP path (never over the live
        // file), so op 3 can diff it and restart only on a real change.
        match &ops[1] {
            DeployOp::WriteFile {
                label,
                contents,
                remote_path,
                mode,
            } => {
                assert_eq!(*label, "media-write-config");
                assert_eq!(remote_path, "/etc/mediamtx/mediamtx.yml.autumn-new");
                assert_eq!(*mode, Some(0o644));
                let FileContents::Plain(yml) = contents else {
                    panic!("config must be plain text");
                };
                assert!(yml.contains("~^room/.+$:"));
            }
            other => panic!("op 1 must write the config, got {other:?}"),
        }

        // op 2: write the unit to a TEMP path next to the live unit (0644).
        match &ops[2] {
            DeployOp::WriteFile {
                label,
                contents,
                remote_path,
                mode,
            } => {
                assert_eq!(*label, "media-write-unit");
                assert_eq!(
                    remote_path,
                    "/etc/systemd/system/mediamtx.service.autumn-new"
                );
                assert_eq!(*mode, Some(0o644));
                let FileContents::Plain(unit) = contents else {
                    panic!("unit must be plain text");
                };
                assert!(unit.contains("ExecStart=\"/usr/local/bin/mediamtx\""));
            }
            other => panic!("op 2 must write the unit, got {other:?}"),
        }

        // op 3 (the conditional-restart install) is asserted in
        // `controller_install_op_is_conditional_restart` below.
        assert!(matches!(&ops[3], DeployOp::Run(c) if c.label == "media-install"));
    }

    #[test]
    fn controller_install_op_is_conditional_restart() {
        // Finding H: op 3 must `cmp` the staged temp files against the live ones,
        // copy only the changed, and restart the unit ONLY on a real change or an
        // inactive unit — never unconditionally (which would kill live sessions on
        // a pure app redeploy).
        let cfg = MediaMtxHostConfig {
            enabled: true,
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();
        let DeployOp::Run(cmd) = &ops[3] else {
            panic!("op 3 must be the install Run op");
        };
        assert_eq!(cmd.label, "media-install");
        let shell = &cmd.shell;
        // Diffs the staged temp files against the live counterparts.
        assert!(
            shell.contains(
                "cmp -s '/etc/mediamtx/mediamtx.yml.autumn-new' '/etc/mediamtx/mediamtx.yml'"
            ),
            "must cmp the staged config against the live config: {shell}"
        );
        assert!(
            shell.contains(
                "cmp -s '/etc/systemd/system/mediamtx.service.autumn-new' \
                 '/etc/systemd/system/mediamtx.service'"
            ),
            "must cmp the staged unit against the live unit: {shell}"
        );
        // Copies over only the changed files and tracks a `changed` flag.
        assert!(
            shell.contains("changed=1"),
            "must track a changed flag: {shell}"
        );
        assert!(
            shell.contains(
                "cp '/etc/mediamtx/mediamtx.yml.autumn-new' '/etc/mediamtx/mediamtx.yml'"
            ),
            "must copy the staged config into place on change: {shell}"
        );
        // Cleans up the temp files.
        assert!(
            shell.contains(
                "rm -f '/etc/mediamtx/mediamtx.yml.autumn-new' \
                 '/etc/systemd/system/mediamtx.service.autumn-new'"
            ),
            "must clean up the staged temp files: {shell}"
        );
        // Idempotent daemon-reload + enable (NOT `enable --now` — start is handled
        // by the conditional restart).
        assert!(shell.contains("systemctl daemon-reload"), "{shell}");
        // The unit id is shell-quoted (`'mediamtx.service'`, Finding T sweep) in
        // every systemctl invocation.
        assert!(
            shell.contains("systemctl enable 'mediamtx.service'"),
            "{shell}"
        );
        assert!(
            !shell.contains("enable --now"),
            "must not `enable --now` (that would start unconditionally): {shell}"
        );
        // Restart gated on change-OR-inactive, never unconditional.
        assert!(
            shell.contains(
                "if [ \"$changed\" -ne 0 ] || ! systemctl is-active --quiet 'mediamtx.service'; then"
            ),
            "restart must be gated on change or an inactive unit: {shell}"
        );
        assert!(
            shell.contains("systemctl restart 'mediamtx.service'"),
            "{shell}"
        );
    }

    #[test]
    fn controller_honors_custom_unit_name_in_paths_and_command() {
        let cfg = MediaMtxHostConfig {
            enabled: true,
            unit_name: "mediamtx-prod".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let controller = MediaMtxController::new(cfg);
        assert_eq!(
            controller.unit_path(),
            "/etc/systemd/system/mediamtx-prod.service"
        );
        let ops = controller.ensure_installed_ops();
        let DeployOp::Run(cmd) = &ops[3] else {
            panic!("op 3 must be the install Run op");
        };
        assert!(
            cmd.shell
                .contains("systemctl enable 'mediamtx-prod.service'")
        );
        assert!(
            cmd.shell
                .contains("is-active --quiet 'mediamtx-prod.service'")
        );
        assert!(cmd.shell.contains("restart 'mediamtx-prod.service'"));
        // The staged temp unit tracks the custom unit name too.
        assert!(
            cmd.shell
                .contains("/etc/systemd/system/mediamtx-prod.service.autumn-new")
        );
    }

    #[test]
    fn controller_mkdirs_config_parent_before_writing_config() {
        // Finding B: `scp` never creates parent dirs, so a `mkdir -p` of the
        // config's parent must precede the `media-write-config` upload.
        let cfg = MediaMtxHostConfig {
            enabled: true,
            config_path: "/opt/media/etc/mediamtx.yml".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();

        let mkdir_idx = ops
            .iter()
            .position(|op| op.label() == "media-prepare-dirs")
            .expect("a media-prepare-dirs op must be emitted");
        let write_idx = ops
            .iter()
            .position(|op| op.label() == "media-write-config")
            .expect("a media-write-config op must be emitted");
        assert!(
            mkdir_idx < write_idx,
            "mkdir must precede the config write (mkdir at {mkdir_idx}, write at {write_idx})"
        );

        let DeployOp::Run(mkdir) = &ops[mkdir_idx] else {
            panic!("media-prepare-dirs must be a Run op");
        };
        // The parent is derived from `config_path`, not hardcoded (shell-quoted).
        assert_eq!(
            mkdir.shell,
            "mkdir -p '/opt/media/etc'; mkdir -p -m 0750 '/recordings'"
        );
    }

    #[test]
    fn controller_creates_the_recordings_dir_it_preflights() {
        // The recordings root is fail-closed preflighted, so provisioning must
        // create it — otherwise a fresh host can never satisfy its own check.
        let cfg = MediaMtxHostConfig {
            enabled: true,
            recordings_dir: "/srv/media recordings".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();

        let mkdir_idx = ops
            .iter()
            .position(|op| op.label() == "media-prepare-dirs")
            .expect("a media-prepare-dirs op must be emitted");
        let DeployOp::Run(mkdir) = &ops[mkdir_idx] else {
            panic!("media-prepare-dirs must be a Run op");
        };
        // Shell-quoted, so a path with a space stays one argument.
        assert_eq!(
            mkdir.shell,
            "mkdir -p '/etc/mediamtx'; mkdir -p -m 0750 '/srv/media recordings'"
        );
        let write_idx = ops
            .iter()
            .position(|op| op.label() == "media-write-config")
            .expect("a media-write-config op must be emitted");
        assert!(mkdir_idx < write_idx);
    }

    #[test]
    fn controller_mkdirs_the_recordings_dir_even_with_a_parentless_config_path() {
        // A parent-less `config_path` skips the config mkdir; the recordings
        // root must still be created.
        let cfg = MediaMtxHostConfig {
            enabled: true,
            config_path: "mediamtx.yml".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();
        let DeployOp::Run(mkdir) = &ops[0] else {
            panic!("op 0 must be the mkdir Run op, got {:?}", ops[0]);
        };
        assert_eq!(mkdir.shell, "mkdir -p -m 0750 '/recordings'");
    }

    #[test]
    fn controller_emits_one_mkdir_when_recordings_dir_is_the_config_parent() {
        // Deduplicated: the same dir is never listed twice.
        let cfg = MediaMtxHostConfig {
            enabled: true,
            config_path: "/recordings/mediamtx.yml".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let ops = MediaMtxController::new(cfg).ensure_installed_ops();
        let DeployOp::Run(mkdir) = &ops[0] else {
            panic!("op 0 must be the mkdir Run op, got {:?}", ops[0]);
        };
        // One mkdir, and it is the mode-bearing one.
        assert_eq!(mkdir.shell, "mkdir -p -m 0750 '/recordings'");
    }

    #[test]
    fn parent_dir_extracts_parent_or_none() {
        assert_eq!(
            parent_dir("/etc/mediamtx/mediamtx.yml"),
            Some("/etc/mediamtx")
        );
        assert_eq!(parent_dir("/opt/a/b/c.yml"), Some("/opt/a/b"));
        // Root-level and bare-filename paths have no parent to create.
        assert_eq!(parent_dir("/mediamtx.yml"), None);
        assert_eq!(parent_dir("mediamtx.yml"), None);
    }

    // ── Doctor: ffmpeg preflight ─────────────────────────────────────────────

    #[test]
    fn ffmpeg_preflight_passes_on_version_banner() {
        let exec = RecordingExecutor::new().with_stdout(
            "media-ffmpeg-preflight",
            "ffmpeg version 6.1.1 Copyright (c)",
        );
        let check = ffmpeg_preflight(&exec, "/usr/bin/ffmpeg");
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(check.name, CHECK_FFMPEG_PREFLIGHT);
        assert_eq!(exec.labels(), vec!["media-ffmpeg-preflight"]);
    }

    #[test]
    fn ffmpeg_preflight_fails_without_banner() {
        // A binary that runs but is not FFmpeg (no banner) fails — fail-closed.
        let exec =
            RecordingExecutor::new().with_stdout("media-ffmpeg-preflight", "not-ffmpeg output");
        let check = ffmpeg_preflight(&exec, "/usr/bin/ffmpeg");
        assert!(!check.passed);
        // A concrete literal that is not FFmpeg is a BLOCKING failure (not
        // deferred) — it must still abort the deploy (Finding U, #2051).
        assert!(!check.deferred);
        assert!(check.blocking());
        assert!(check.hint.is_some());
    }

    #[test]
    fn ffmpeg_preflight_fails_when_binary_missing() {
        // A non-zero exit (missing / not executable) surfaces as a transport
        // error — fail-closed, not a pass.
        let exec = RecordingExecutor::failing_on("media-ffmpeg-preflight");
        let check = ffmpeg_preflight(&exec, "/usr/bin/ffmpeg");
        assert!(!check.passed);
        // A missing concrete literal is a BLOCKING failure, not deferred.
        assert!(!check.deferred);
        assert!(check.blocking());
        assert!(check.detail.contains("could not run"));
    }

    // ── Finding L: verify only a concrete literal; defer env-indirected paths ──
    //
    // The deploy side must NOT resolve `[media.ffmpeg] bin` against the operator's
    // LOCAL shell — the generated service env-file does not propagate it, so the
    // service would resolve a different path. Only an environment-independent
    // concrete literal is probed; anything env/interpolation-indirected (a
    // `${...}` placeholder — including `${AUTUMN_MEDIA__FFMPEG__BIN}` — or an empty
    // value) is deferred to runtime with NO probe, regardless of any local env.

    #[test]
    fn ffmpeg_preflight_defers_env_indirected_paths_without_probing() {
        // A whole-string `${VAR}`, an embedded placeholder, a value that relies on
        // the AUTUMN_MEDIA__FFMPEG__BIN override, and an empty value all DEFER — a
        // clear non-passing outcome, and crucially NO probe runs.
        for configured in [
            "${FF_BIN_UNSET_XYZ}",
            "/opt/${VER}/ffmpeg",
            // A value that relies on the AUTUMN_MEDIA__FFMPEG__BIN override — the
            // service resolves it from ITS own env; the deploy side must defer.
            "${AUTUMN_MEDIA__FFMPEG__BIN}",
            "",
        ] {
            let exec = RecordingExecutor::new();
            let check = ffmpeg_preflight(&exec, configured);
            assert!(!check.passed, "must be non-passing for `{configured}`");
            // Deferred, not failed: the service resolves FFmpeg from its own
            // runtime env, so this must NOT abort the deploy (Finding U, #2051).
            assert!(check.deferred, "must be deferred for `{configured}`");
            assert!(
                !check.blocking(),
                "deferred check must be non-blocking for `{configured}`"
            );
            assert!(
                check.detail.contains("indirection") && check.detail.contains("deferring"),
                "detail must name the deferral for `{configured}`: {}",
                check.detail
            );
            assert!(check.hint.is_some());
            // Fail-closed-honest: never probed a locally-resolved / placeholder path.
            assert!(
                exec.labels().is_empty(),
                "no probe must run for `{configured}`, ran {:?}",
                exec.labels()
            );
        }
    }

    #[test]
    fn ffmpeg_preflight_probes_a_concrete_literal_path() {
        // A concrete (non-placeholder) literal path IS probed verbatim — the
        // healthy, environment-independent path.
        let exec = RecordingExecutor::new()
            .with_stdout("media-ffmpeg-preflight", "ffmpeg version 6.1.1 Copyright");
        let check = ffmpeg_preflight(&exec, "/usr/bin/ffmpeg");
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(exec.labels(), vec!["media-ffmpeg-preflight"]);
    }

    // ── Doctor: MediaMTX binary preflight ────────────────────────────────────
    //
    // Provisioning (`systemctl … restart`) runs after the app cutover, so the
    // binary must be there, or be installable, before cutover. The probe is
    // read-only and reports one of three states.

    fn binary_probe(stdout: &str) -> RecordingExecutor {
        RecordingExecutor::new().with_stdout("media-mediamtx-binary", stdout)
    }

    #[test]
    fn mediamtx_binary_preflight_passes_when_present() {
        let exec = binary_probe("present\n");
        let check = mediamtx_binary_preflight(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(check.name, CHECK_MEDIAMTX_BINARY);
        assert_eq!(exec.labels(), vec!["media-mediamtx-binary"]);
        assert!(check.hint.is_none());
    }

    #[test]
    fn mediamtx_binary_preflight_passes_when_absent_and_deploy_installs_it() {
        let check =
            mediamtx_binary_preflight(&binary_probe("absent\n"), &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert!(!check.deferred);
        assert!(
            check.detail.contains(MEDIAMTX_KNOWN_GOOD_VERSION) && check.detail.contains("install"),
            "detail must say deploy installs the pinned release: {}",
            check.detail
        );
    }

    #[test]
    fn mediamtx_binary_preflight_fails_when_absent_and_install_is_off() {
        let cfg = MediaMtxHostConfig {
            install_binary: false,
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_binary_preflight(&binary_probe("absent\n"), &cfg);
        assert!(!check.passed);
        assert!(check.blocking());
        assert!(check.detail.contains(&cfg.binary_path), "{}", check.detail);
        assert!(
            check.detail.contains("systemctl restart"),
            "detail must state the post-cutover restart would fail: {}",
            check.detail
        );
        assert!(
            check
                .hint
                .is_some_and(|hint| hint.contains("install_binary"))
        );
    }

    #[test]
    fn mediamtx_binary_preflight_fails_when_another_file_is_there() {
        // A non-executable file, a directory or a dangling symlink: deploy
        // never replaces it, so it cannot install either.
        let check =
            mediamtx_binary_preflight(&binary_probe("occupied\n"), &MediaMtxHostConfig::default());
        assert!(!check.passed);
        assert!(check.blocking());
        assert!(
            check.detail.contains("not an executable file"),
            "{}",
            check.detail
        );
    }

    #[test]
    fn mediamtx_binary_preflight_fails_on_a_relative_or_empty_path() {
        // systemd needs an absolute `ExecStart`, so these fail before cutover
        // and the host is not probed.
        for path in ["", "mediamtx", "bin/mediamtx"] {
            let exec = binary_probe("absent\n");
            let cfg = MediaMtxHostConfig {
                binary_path: path.to_owned(),
                ..MediaMtxHostConfig::default()
            };
            let check = mediamtx_binary_preflight(&exec, &cfg);
            assert!(check.blocking(), "{path:?}: {}", check.detail);
            assert!(check.detail.contains("absolute"), "{}", check.detail);
            assert!(exec.labels().is_empty(), "{path:?} must not be probed");
        }
    }

    #[test]
    fn mediamtx_binary_preflight_hint_for_an_occupied_path_says_remove() {
        let check =
            mediamtx_binary_preflight(&binary_probe("occupied\n"), &MediaMtxHostConfig::default());
        assert!(
            check.hint.is_some_and(|hint| hint.contains("Remove")),
            "{:?}",
            check.hint
        );
    }

    #[test]
    fn mediamtx_binary_preflight_fails_closed_when_it_cannot_tell() {
        for exec in [
            RecordingExecutor::failing_on("media-mediamtx-binary"),
            binary_probe("something else\n"),
            binary_probe(""),
        ] {
            let check = mediamtx_binary_preflight(&exec, &MediaMtxHostConfig::default());
            assert!(!check.passed, "detail: {}", check.detail);
            assert!(check.blocking());
        }
    }

    #[test]
    fn mediamtx_binary_preflight_probe_is_read_only_and_quotes_the_path() {
        let exec = binary_probe("present\n");
        let cfg = MediaMtxHostConfig {
            binary_path: "/opt/media mtx/mediamtx".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let _ = mediamtx_binary_preflight(&exec, &cfg);
        let cmd = exec.shell_for("media-mediamtx-binary");
        assert!(
            cmd.contains("'/opt/media mtx/mediamtx'"),
            "path must be quoted: {cmd}"
        );
        for mutating in ["mkdir", "mv ", "docker", "chmod", "rm "] {
            assert!(
                !cmd.contains(mutating),
                "probe must not run `{mutating}`: {cmd}"
            );
        }
    }

    // ── MediaMTX binary install (pinned image) ───────────────────────────────

    #[test]
    fn install_binary_defaults_on_and_parses_off() {
        assert!(MediaMtxHostConfig::default().install_binary);
        let root: toml::Value =
            toml::from_str("[media.mediamtx]\nenabled = true\ninstall_binary = false\n").unwrap();
        let (cfg, _) = media_host_config_from_value(root).unwrap();
        assert!(!cfg.install_binary);
    }

    #[test]
    fn binary_install_ops_are_empty_when_disabled_or_opted_out() {
        let disabled = MediaMtxController::new(MediaMtxHostConfig::default());
        assert!(disabled.binary_install_ops().is_empty());
        let opted_out = MediaMtxController::new(MediaMtxHostConfig {
            enabled: true,
            install_binary: false,
            ..MediaMtxHostConfig::default()
        });
        assert!(opted_out.binary_install_ops().is_empty());
    }

    /// The install shell for `cfg` (enabled, install on).
    fn install_shell_for(binary_path: &str) -> String {
        let ops = MediaMtxController::new(MediaMtxHostConfig {
            enabled: true,
            binary_path: binary_path.to_owned(),
            ..MediaMtxHostConfig::default()
        })
        .binary_install_ops();
        let [DeployOp::Run(cmd)] = ops.as_slice() else {
            panic!("expected one Run op, got {ops:?}");
        };
        assert_eq!(cmd.label, "media-install-binary");
        cmd.shell.clone()
    }

    #[test]
    fn binary_install_op_pulls_the_pinned_digest() {
        let shell = install_shell_for("/opt/media mtx/mediamtx");
        let image = format!(
            "bluenviron/mediamtx:{MEDIAMTX_KNOWN_GOOD_VERSION}@{MEDIAMTX_KNOWN_GOOD_DIGEST}"
        );
        assert!(shell.contains(&image), "must pull tag@digest: {shell}");
        assert!(shell.contains("\"$cid:/mediamtx\""), "{shell}");
        assert!(
            shell.contains("'/opt/media mtx/mediamtx'"),
            "path must be quoted: {shell}"
        );
        assert!(
            shell.contains("--version"),
            "must check the version: {shell}"
        );
    }

    #[test]
    fn pinned_digest_is_a_sha256() {
        let hex = MEDIAMTX_KNOWN_GOOD_DIGEST
            .strip_prefix("sha256:")
            .expect("sha256: prefix");
        assert_eq!(hex.len(), 64);
        assert!(
            hex.bytes()
                .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
        );
    }

    /// Runs the real install shell under `sh` with a fake `docker` and
    /// `systemctl` on `PATH`.
    #[cfg(unix)]
    struct InstallRun {
        dir: tempfile::TempDir,
        bin: std::path::PathBuf,
    }

    #[cfg(unix)]
    impl InstallRun {
        fn new() -> Self {
            use std::os::unix::fs::PermissionsExt as _;
            let dir = tempfile::TempDir::new().expect("tempdir");
            let fake = dir.path().join("fakebin");
            std::fs::create_dir(&fake).unwrap();
            // `docker cp <cid>:/mediamtx <dest>` writes a script that prints
            // `$FAKE_VERSION` for `--version`.
            let docker = "#!/bin/sh\n\
                echo \"$*\" >> \"$DOCKER_LOG\"\n\
                case \"$1\" in\n\
                create) echo fakecid ;;\n\
                cp) printf '#!/bin/sh\\necho %s\\n' \"$FAKE_VERSION\" > \"$3\" ;;\n\
                esac\n";
            for (name, body) in [("docker", docker), ("systemctl", "#!/bin/sh\nexit 0\n")] {
                let path = fake.join(name);
                std::fs::write(&path, body).unwrap();
                std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
            }
            let bin = dir.path().join("usr local/bin/mediamtx");
            Self { dir, bin }
        }

        fn run(&self, fake_version: &str) -> std::process::Output {
            let path = format!(
                "{}:{}",
                self.dir.path().join("fakebin").display(),
                std::env::var("PATH").unwrap_or_default()
            );
            std::process::Command::new("sh")
                .arg("-c")
                .arg(install_shell_for(self.bin.to_str().unwrap()))
                .env("PATH", path)
                .env("DOCKER_LOG", self.dir.path().join("docker.log"))
                .env("FAKE_VERSION", fake_version)
                .output()
                .expect("run sh")
        }

        fn docker_log(&self) -> String {
            std::fs::read_to_string(self.dir.path().join("docker.log")).unwrap_or_default()
        }

        fn staged(&self) -> std::path::PathBuf {
            let mut staged = self.bin.clone().into_os_string();
            staged.push(".autumn-new");
            staged.into()
        }
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_installs_the_pinned_binary_when_absent() {
        let run = InstallRun::new();
        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(
            out.status.success(),
            "stderr: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        let version = std::process::Command::new(&run.bin)
            .arg("--version")
            .output()
            .unwrap();
        assert_eq!(
            String::from_utf8_lossy(&version.stdout).trim(),
            format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}")
        );
        assert!(run.docker_log().contains(&format!(
            "create bluenviron/mediamtx:{MEDIAMTX_KNOWN_GOOD_VERSION}@{MEDIAMTX_KNOWN_GOOD_DIGEST}"
        )));
        assert!(
            run.docker_log().contains("rm -f fakecid"),
            "scratch container removed"
        );
        assert!(!run.staged().exists(), "no staged file left");
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_keeps_an_executable_that_is_already_there() {
        use std::os::unix::fs::PermissionsExt as _;
        let run = InstallRun::new();
        std::fs::create_dir_all(run.bin.parent().unwrap()).unwrap();
        std::fs::write(&run.bin, "#!/bin/sh\necho operator-build\n").unwrap();
        std::fs::set_permissions(&run.bin, std::fs::Permissions::from_mode(0o755)).unwrap();

        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(
            out.status.success(),
            "stderr: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        assert_eq!(
            std::fs::read_to_string(&run.bin).unwrap(),
            "#!/bin/sh\necho operator-build\n"
        );
        assert!(run.docker_log().is_empty(), "docker must not run");
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_refuses_to_replace_any_other_file() {
        // A plain file, then a dangling symlink.
        let run = InstallRun::new();
        std::fs::create_dir_all(run.bin.parent().unwrap()).unwrap();
        std::fs::write(&run.bin, "not a binary").unwrap();
        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(!out.status.success());
        assert_eq!(std::fs::read_to_string(&run.bin).unwrap(), "not a binary");

        std::fs::remove_file(&run.bin).unwrap();
        std::os::unix::fs::symlink(run.dir.path().join("gone"), &run.bin).unwrap();
        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(!out.status.success());
        assert!(
            std::fs::symlink_metadata(&run.bin)
                .unwrap()
                .file_type()
                .is_symlink()
        );
        assert!(run.docker_log().is_empty(), "docker must not run");
    }

    #[test]
    fn install_shell_waits_for_the_apt_lock() {
        let shell = install_shell_for("/usr/local/bin/mediamtx");
        assert_eq!(
            shell.matches("-o DPkg::Lock::Timeout=300").count(),
            2,
            "{shell}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_removes_a_leftover_staged_path_without_following_it() {
        let run = InstallRun::new();
        std::fs::create_dir_all(run.bin.parent().unwrap()).unwrap();
        // A symlink at the staged path must not redirect the copy.
        let victim = run.dir.path().join("victim");
        std::fs::write(&victim, "keep me").unwrap();
        std::os::unix::fs::symlink(&victim, run.staged()).unwrap();
        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(
            out.status.success(),
            "stderr: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        assert_eq!(std::fs::read_to_string(&victim).unwrap(), "keep me");
        assert!(
            !std::fs::symlink_metadata(&run.bin)
                .unwrap()
                .file_type()
                .is_symlink()
        );

        // A leftover directory must not block the next install.
        let run = InstallRun::new();
        std::fs::create_dir_all(run.staged()).unwrap();
        let out = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        assert!(
            out.status.success(),
            "stderr: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        assert!(!run.staged().exists());
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_names_the_real_cause_of_a_refusal() {
        let run = InstallRun::new();
        std::fs::create_dir_all(run.bin.parent().unwrap()).unwrap();
        std::fs::write(&run.bin, "not a binary").unwrap();
        let refused = run.run(&format!("v{MEDIAMTX_KNOWN_GOOD_VERSION}"));
        let stderr = String::from_utf8_lossy(&refused.stderr);
        assert!(stderr.contains("refusing to replace"), "{stderr}");
        assert!(
            !stderr.contains("apt-based"),
            "no apt hint for a refusal: {stderr}"
        );

        let wrong = InstallRun::new().run("v1.18.0");
        let stderr = String::from_utf8_lossy(&wrong.stderr);
        assert!(stderr.contains("v1.19.3"), "{stderr}");
        assert!(
            !stderr.contains("apt-based"),
            "no apt hint for a version mismatch: {stderr}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn install_shell_installs_nothing_when_the_version_is_wrong() {
        let run = InstallRun::new();
        let out = run.run("v1.18.0");
        assert!(!out.status.success());
        assert!(!run.bin.exists(), "a wrong build must not be installed");
        assert!(!run.staged().exists(), "no staged file left");
    }

    // ── Doctor: recordings dir writable ──────────────────────────────────────

    #[test]
    fn recordings_dir_passes_when_present_and_writable() {
        let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", "present\n");
        let check = recordings_dir_writable(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(check.name, CHECK_RECORDINGS_DIR_WRITABLE);
        assert!(check.detail.contains("exists and is writable"));
        assert_eq!(exec.labels(), vec!["media-recordings-dir"]);
    }

    #[test]
    fn recordings_dir_passes_when_absent_but_creatable() {
        // A fresh host has no recordings dir yet; provisioning creates it, so an
        // absent dir under a writable ancestor is a pass, not a blocked deploy.
        let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", "creatable\n");
        let check = recordings_dir_writable(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert!(
            check.detail.contains("will be created"),
            "detail must say the dir is created by provisioning: {}",
            check.detail
        );
    }

    #[test]
    fn recordings_dir_fails_when_not_writable() {
        // `test -w` exits non-zero → transport error → fail-closed.
        let exec = RecordingExecutor::failing_on("media-recordings-dir");
        let check = recordings_dir_writable(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed);
        assert!(check.detail.contains("/recordings"));
        assert!(check.hint.is_some());
    }

    #[test]
    fn recordings_dir_fails_when_output_proves_nothing() {
        // Unverifiable output is never a silent pass (module discipline).
        for stdout in ["", "yes", "presentish"] {
            let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", stdout);
            let check = recordings_dir_writable(&exec, &MediaMtxHostConfig::default());
            assert!(!check.passed, "stdout {stdout:?} must not pass");
            assert!(check.hint.is_some());
        }
    }

    #[test]
    fn recordings_dir_reports_the_two_host_failures_apart() {
        for (marker, needle) in [
            ("unwritable", "exists but is not writable"),
            ("blocked", "nearest existing ancestor"),
        ] {
            let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", marker);
            let check = recordings_dir_writable(&exec, &MediaMtxHostConfig::default());
            assert!(!check.passed, "{marker} must not pass");
            assert!(!check.deferred);
            assert!(check.detail.contains(needle), "detail: {}", check.detail);
        }
    }

    #[test]
    fn recordings_dir_probe_is_quoted_and_walks_without_dirname() {
        // The probe's own command text is the load-bearing part of this check,
        // so assert it rather than only that the command ran.
        let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", "present");
        let cfg = MediaMtxHostConfig {
            recordings_dir: "/srv/media recordings".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let check = recordings_dir_writable(&exec, &cfg);
        assert!(check.passed, "detail: {}", check.detail);

        let shell = exec.shell_for("media-recordings-dir");
        // A path with a space stays one argument.
        assert!(
            shell.contains("d='/srv/media recordings';"),
            "path must be shell-quoted: {shell}"
        );
        // The ancestor walk uses parameter expansion: an absent `dirname`, or a
        // path it reads as an option, would otherwise leave `$d` empty and spin.
        assert!(
            !shell.contains("dirname"),
            "walk must not shell out: {shell}"
        );
        // `[ -e ]` resolves symlinks, so `[ ! -L ]` is what stops the walk at a
        // dangling link instead of stepping past it as if absent.
        assert!(
            shell.contains(r#"[ ! -L "$d" ]"#),
            "walk must stop at a symlink: {shell}"
        );
        // The probe reports by marker and always exits 0, so a non-zero exit
        // means only that the command could not run.
        for marker in [
            "printf present",
            "printf unwritable",
            "printf creatable",
            "printf blocked",
        ] {
            assert!(shell.contains(marker), "missing `{marker}`: {shell}");
        }
    }

    #[test]
    fn recordings_dir_fails_closed_on_a_path_that_names_no_host_directory() {
        // Empty renders `recordPath: '/%path/…'` (the filesystem root); relative
        // resolves to a different directory in the probe, the mkdir and the
        // daemon. Neither may reach the host.
        for (dir, needle) in [("", "is empty"), ("recordings", "must be an absolute path")] {
            let exec = RecordingExecutor::new().with_stdout("media-recordings-dir", "creatable");
            let cfg = MediaMtxHostConfig {
                recordings_dir: dir.to_owned(),
                ..MediaMtxHostConfig::default()
            };
            let check = recordings_dir_writable(&exec, &cfg);
            assert!(!check.passed, "`{dir}` must fail closed");
            assert!(!check.deferred, "`{dir}` is a config error, not a deferral");
            assert!(check.detail.contains(needle), "detail: {}", check.detail);
            // Rejected before any host command runs.
            assert!(exec.labels().is_empty(), "no probe may run for `{dir}`");
        }
    }

    #[test]
    fn recordings_dir_defers_an_env_indirected_path() {
        // Same rule as the FFmpeg preflight: the service resolves this from its
        // own environment, which the deploy side must not guess.
        let exec = RecordingExecutor::new();
        let cfg = MediaMtxHostConfig {
            recordings_dir: "${MEDIA_REC_ROOT}/mediamtx".to_owned(),
            ..MediaMtxHostConfig::default()
        };
        let check = recordings_dir_writable(&exec, &cfg);
        assert!(!check.passed);
        assert!(check.deferred);
        assert!(!check.blocking(), "a deferred path must not abort a deploy");
        assert!(exec.labels().is_empty(), "no probe may run");
    }

    // ── Doctor: mediamtx ports available ─────────────────────────────────────

    #[test]
    fn parse_listening_sockets_extracts_port_protocol_and_owner() {
        // Finding I: the Netid (tcp/udp) column is parsed onto each socket.
        let ss = "\
tcp   LISTEN 0 128  0.0.0.0:8888  0.0.0.0:*  users:((\"mediamtx\",pid=10,fd=5))
tcp   LISTEN 0 128  [::]:22       [::]:*     users:((\"sshd\",pid=5,fd=3))
udp   UNCONN 0 0    *:8189        *:*
sctp  LISTEN 0 128  0.0.0.0:9999  0.0.0.0:*
";
        let sockets = parse_listening_sockets(ss);
        assert!(sockets.iter().any(|s| s.port == 8888
            && s.protocol == Protocol::Tcp
            && s.owner.as_deref() == Some("mediamtx")));
        assert!(sockets.iter().any(|s| s.port == 22
            && s.protocol == Protocol::Tcp
            && s.owner.as_deref() == Some("sshd")));
        // The UDP row has no `users:` column → unattributable owner, protocol UDP.
        assert!(
            sockets
                .iter()
                .any(|s| s.port == 8189 && s.protocol == Protocol::Udp && s.owner.is_none())
        );
        // A non-tcp/udp Netid (e.g. `sctp`) is skipped entirely.
        assert_eq!(sockets.len(), 3);
    }

    #[test]
    fn ports_available_passes_when_ports_free() {
        // ss shows only sshd — none of MediaMTX's ports are taken.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(check.name, CHECK_MEDIAMTX_PORTS_AVAILABLE);
        assert!(check.detail.contains("free"), "detail: {}", check.detail);
    }

    #[test]
    fn ports_available_fails_when_foreign_process_holds_port() {
        // Finding D (b): a foreign process (nginx) is bound to the HLS port (8888).
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n\
             tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:* users:((\"nginx\",pid=99,fd=6))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8888"), "detail: {}", check.detail);
        assert!(check.detail.contains("another service"));
        assert!(check.hint.is_some());
    }

    #[test]
    fn ports_available_passes_when_our_mediamtx_owns_all_ports() {
        // Finding D (a): every required port is held by OUR managed mediamtx (a
        // same-port redeploy) — not a conflict, so the check passes and never
        // aborts the 2nd+ `deploy up`.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:9997 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=3))\n\
             tcp LISTEN 0 128 0.0.0.0:1935 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=4))\n\
             tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=5))\n\
             tcp LISTEN 0 128 0.0.0.0:8889 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=6))\n\
             tcp LISTEN 0 128 0.0.0.0:9996 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=7))\n\
             udp UNCONN 0 0 0.0.0.0:8189 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=8))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert!(
            check.detail.contains("our own managed"),
            "detail: {}",
            check.detail
        );
        // Finding J: because MediaMTX-owned ports were found, the is-active gate
        // runs exactly once after the `ss` scan (the fake reports the unit active
        // by default), so the same-port redeploy passes.
        assert_eq!(exec.labels(), vec!["media-ports", "media-unit-active"]);
    }

    #[test]
    fn ports_available_fails_when_changed_port_held_by_foreign_service() {
        // Finding D core: a redeploy that MOVED the HLS port to 8080, where a
        // foreign service is already bound. The old managed unit only held the old
        // ports, so a process-aware scan of the NEW target port must still fail —
        // exactly the case the removed is-active skip missed.
        let cfg = MediaMtxHostConfig {
            hls_port: 8080,
            ..MediaMtxHostConfig::default()
        };
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            // Our mediamtx still holds its OTHER (unchanged) ports, but the
            // newly-configured HLS port 8080 is held by a foreign nginx.
            "tcp LISTEN 0 128 0.0.0.0:9997 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=3))\n\
             tcp LISTEN 0 128 0.0.0.0:1935 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=4))\n\
             tcp LISTEN 0 128 0.0.0.0:8080 0.0.0.0:* users:((\"nginx\",pid=99,fd=6))\n",
        );
        let check = mediamtx_ports_available(&exec, &cfg);
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8080"), "detail: {}", check.detail);
    }

    #[test]
    fn ports_available_ignores_same_port_different_protocol_listener() {
        // Finding I: the WebRTC local port 8189 is UDP-only. A TCP service on 8189
        // is a DIFFERENT protocol, so it must NOT be flagged as a conflict.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n\
             tcp LISTEN 0 128 0.0.0.0:8189 0.0.0.0:* users:((\"someapp\",pid=77,fd=9))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("free"), "detail: {}", check.detail);

        // The mirror case: a UDP service on a TCP MediaMTX port (8888 is TCP) is
        // also a different protocol and not a conflict.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "udp UNCONN 0 0 0.0.0.0:8888 0.0.0.0:* users:((\"someapp\",pid=88,fd=4))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
    }

    #[test]
    fn ports_available_fails_on_same_protocol_foreign_listener() {
        // Finding I: a TCP listener on the TCP HLS port (8888), owned by a foreign
        // process, IS a same-protocol conflict.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:* users:((\"nginx\",pid=99,fd=6))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8888"), "detail: {}", check.detail);
        assert!(check.detail.contains("another service"));
    }

    #[test]
    fn ports_available_fails_on_udp_foreign_listener_on_udp_target() {
        // Finding I: a UDP foreign listener on the UDP-target 8189 IS a conflict.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "udp UNCONN 0 0 0.0.0.0:8189 0.0.0.0:* users:((\"coturn\",pid=42,fd=8))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8189"), "detail: {}", check.detail);
    }

    #[test]
    fn ports_available_fails_closed_when_mediamtx_owns_ports_but_unit_inactive() {
        // Finding J: the ports are held by a `mediamtx` process, but OUR configured
        // unit is NOT active (manual run / differently-named service / crashed).
        // Starting our unit would launch a second instance over them → fail-closed.
        let exec = RecordingExecutor::new()
            .with_stdout(
                "media-ports",
                "tcp LISTEN 0 128 0.0.0.0:9997 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=3))\n\
                 tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:* users:((\"mediamtx\",pid=10,fd=5))\n",
            )
            // `systemctl is-active --quiet` exits non-zero → transport error →
            // read as NOT active.
            .failing("media-unit-active");
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8888"), "detail: {}", check.detail);
        assert!(
            check.detail.contains("not active"),
            "detail: {}",
            check.detail
        );
        // The is-active gate ran exactly once, after the single ss scan.
        assert_eq!(exec.labels(), vec!["media-ports", "media-unit-active"]);
    }

    #[test]
    fn ports_available_does_not_query_is_active_when_no_mediamtx_port_held() {
        // Finding J: the is-active gate runs at most once, and only when a
        // MediaMTX-owned port is actually found. A foreign-only conflict (or a
        // free host) never queries it.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:* users:((\"nginx\",pid=99,fd=6))\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed);
        assert_eq!(exec.labels(), vec!["media-ports"]);
    }

    #[test]
    fn ports_available_fails_closed_on_unattributable_listener() {
        // Finding D (e): a listener on a required port whose owner cannot be
        // attributed (no `users:` column — e.g. insufficient privilege) is treated
        // as a conflict, not assumed benign. Deploy runs as root so this is rare,
        // but fail-closed is the safe default.
        let exec = RecordingExecutor::new().with_stdout(
            "media-ports",
            "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n\
             tcp LISTEN 0 128 0.0.0.0:8888 0.0.0.0:*\n",
        );
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed, "detail: {}", check.detail);
        assert!(check.detail.contains("8888"), "detail: {}", check.detail);
    }

    #[test]
    fn ports_available_fails_closed_when_ss_errors() {
        let exec = RecordingExecutor::failing_on("media-ports");
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed);
        assert!(check.detail.contains("could not verify"));
    }

    #[test]
    fn ports_available_fails_closed_on_unparseable_output() {
        // A non-empty ss body that yields zero ports is unverifiable, not a pass.
        let exec =
            RecordingExecutor::new().with_stdout("media-ports", "garbage that has no ports\n");
        let check = mediamtx_ports_available(&exec, &MediaMtxHostConfig::default());
        assert!(!check.passed);
        assert!(check.detail.contains("could not verify"));
    }

    #[test]
    fn collect_runs_all_six_checks_in_order() {
        let exec = RecordingExecutor::new()
            .with_stdout("media-ffmpeg-preflight", "ffmpeg version 6.1")
            .with_stdout("media-mediamtx-binary", "present")
            .with_stdout("media-recordings-dir", "present")
            .with_stdout(
                "media-ports",
                "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n",
            );
        let checks =
            collect_media_doctor_checks(&exec, &MediaMtxHostConfig::default(), "/usr/bin/ffmpeg");
        assert_eq!(checks.len(), 6);
        assert!(
            checks.iter().all(|c| c.passed),
            "all should pass: {:?}",
            checks.iter().filter(|c| !c.passed).collect::<Vec<_>>()
        );
        // The pure ports-distinct config check (Finding P) leads and runs NO
        // command, so the recorded command order is unchanged from the four
        // executor-driven checks.
        assert_eq!(checks[0].name, CHECK_MEDIAMTX_PORTS_DISTINCT);
        assert_eq!(
            exec.labels(),
            vec![
                "media-ffmpeg-preflight",
                // The MediaMTX binary preflight (Finding N) runs before cutover,
                // symmetric with the FFmpeg preflight, so a host missing the
                // deferred-provisioned binary fails fast.
                "media-mediamtx-binary",
                "media-recordings-dir",
                // The ports check is a single process-aware `ss -tulnp` scan; the
                // old `systemctl is-active` gate was removed (Finding D).
                "media-ports"
            ]
        );
    }

    #[test]
    fn collect_with_env_indirected_ffmpeg_has_no_blocking_failures() {
        // Finding U (#2051): an env/interpolation-indirected `[media.ffmpeg] bin`
        // is DEFERRED to the service runtime — non-passing but non-blocking — so
        // the aggregate has zero blocking checks and `deploy up` proceeds (the
        // deploy gate counts `blocking()`, not `!passed`). A concrete missing
        // FFmpeg would instead block.
        let exec = RecordingExecutor::new()
            .with_stdout("media-mediamtx-binary", "present")
            .with_stdout("media-recordings-dir", "present")
            .with_stdout(
                "media-ports",
                "tcp LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=5,fd=3))\n",
            );
        let checks = collect_media_doctor_checks(
            &exec,
            &MediaMtxHostConfig::default(),
            "${AUTUMN_MEDIA__FFMPEG__BIN}",
        );
        // The ffmpeg check is present, deferred, and non-passing …
        let ffmpeg = checks
            .iter()
            .find(|c| c.name == CHECK_FFMPEG_PREFLIGHT)
            .expect("ffmpeg check present");
        assert!(ffmpeg.deferred && !ffmpeg.passed);
        // … and no probe ran for it (no ffmpeg-preflight label recorded).
        assert!(!exec.labels().contains(&"media-ffmpeg-preflight"));
        // The deploy gate keys on `blocking()`: the deferred ffmpeg does NOT
        // count, so the aggregate is non-blocking and the deploy is not aborted.
        assert_eq!(
            checks.iter().filter(|c| c.blocking()).count(),
            0,
            "deferred ffmpeg must not block the deploy"
        );
    }

    // ── Ports ↔ app base URLs (#1974 slice-7 deferral 6) ─────────────────────

    #[test]
    fn ports_match_bases_passes_for_default_config() {
        // The two surfaces ship default-aligned, so the common case is silent.
        let check = mediamtx_ports_match_bases(&MediaMtxHostConfig::default());
        assert!(check.passed, "detail: {}", check.detail);
        assert_eq!(check.name, CHECK_MEDIAMTX_PORTS_MATCH_BASES);
        assert!(!check.deferred);
    }

    #[test]
    fn ports_match_bases_fails_when_a_custom_port_leaves_the_base_behind() {
        // Two values the operator wrote, contradicting each other: the daemon
        // binds 1936 and the app calls 1935 on the same host. This is the silent
        // runtime break the check exists to catch before cutover.
        let cfg = MediaMtxHostConfig {
            rtmp_port: 1936,
            rtmp_base: Some("rtmp://127.0.0.1:1935/live".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.blocking(), "a real mismatch must block the deploy");
        assert!(check.detail.contains("rtmp_port"), "{}", check.detail);
        assert!(check.detail.contains("rtmp_base"), "{}", check.detail);
        assert!(check.detail.contains("1936"), "{}", check.detail);
        assert!(check.detail.contains("1935"), "{}", check.detail);
        assert!(check.hint.is_some());
    }

    #[test]
    fn ports_match_bases_passes_when_the_base_is_updated_to_match() {
        let cfg = MediaMtxHostConfig {
            rtmp_port: 1936,
            webrtc_port: 8899,
            rtmp_base: Some("rtmp://media.example.com:1936/live".to_owned()),
            webrtc_base: Some("https://media.example.com:8899".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(check.passed, "detail: {}", check.detail);
    }

    #[test]
    fn ports_match_bases_ignores_a_base_with_no_explicit_port() {
        // A reverse-proxied / CDN-fronted base (implicit 80/443) is a legitimate
        // shape the deploy side cannot compare against a listener port.
        let cfg = MediaMtxHostConfig {
            hls_port: 9999,
            hls_base: Some("https://media.example.com/hls".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(check.passed, "detail: {}", check.detail);
    }

    #[test]
    fn ports_match_bases_defers_an_env_indirected_base() {
        // Same rule as the FFmpeg preflight: a value the service resolves from
        // its own environment is deferred (non-passing, NON-blocking), never
        // guessed against the operator's shell.
        let cfg = MediaMtxHostConfig {
            hls_port: 9999,
            hls_base: Some("${MEDIA_HLS_BASE}".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.deferred);
        assert!(!check.blocking(), "a deferred base must not abort a deploy");
        assert!(check.detail.contains("hls_base"), "{}", check.detail);
    }

    #[test]
    fn ports_match_bases_reports_a_mismatch_even_when_another_base_is_indirected() {
        // A real mismatch outranks a deferral: the deploy still fails closed.
        let cfg = MediaMtxHostConfig {
            rtmp_port: 1936,
            rtmp_base: Some("rtmp://127.0.0.1:1935/live".to_owned()),
            hls_base: Some("${MEDIA_HLS_BASE}".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(!check.deferred);
        assert!(check.blocking());
        assert!(check.detail.contains("rtmp_base"), "{}", check.detail);
    }

    #[test]
    fn ports_match_bases_checks_the_probe_bases_too() {
        // The server-side probe bases hit the same listeners, so they are graded
        // when set (and skipped when they fall back to their public base).
        let cfg = MediaMtxHostConfig {
            hls_port: 8888,
            hls_probe_base: Some("http://127.0.0.1:8890".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.detail.contains("hls_probe_base"), "{}", check.detail);
    }

    #[test]
    fn ports_match_bases_reports_every_mismatched_pair() {
        let cfg = MediaMtxHostConfig {
            api_port: 1,
            playback_port: 2,
            api_base: Some("http://127.0.0.1:9997".to_owned()),
            playback_base: Some("http://127.0.0.1:9996".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.detail.contains("api_base"), "{}", check.detail);
        assert!(check.detail.contains("playback_base"), "{}", check.detail);
    }

    #[test]
    fn explicit_port_separates_a_named_port_from_a_portless_and_a_malformed_one() {
        use BasePort::{Explicit, Implicit, Malformed};
        assert_eq!(explicit_port("http://127.0.0.1:9997"), Explicit(9997));
        assert_eq!(explicit_port("rtmp://host:1935/live"), Explicit(1935));
        assert_eq!(explicit_port("https://user:pw@host:8443/x"), Explicit(8443));
        assert_eq!(explicit_port("http://[::1]:8888/hls"), Explicit(8888));
        // No port — proxied, or a bare authority. Userinfo colons are not ports.
        assert_eq!(explicit_port("https://media.example.com/hls"), Implicit);
        assert_eq!(explicit_port("http://[::1]/hls"), Implicit);
        assert_eq!(explicit_port("host-only"), Implicit);
        assert_eq!(explicit_port("https://user:pw@host/x"), Implicit);
        assert_eq!(explicit_port(""), Implicit);
        // Something in the port position that is not a port. Never a valid
        // reverse-proxy shape, so it must not be silently skipped.
        assert_eq!(explicit_port("http://host:notaport/x"), Malformed);
        assert_eq!(explicit_port("http://host:99999/x"), Malformed);
        assert_eq!(explicit_port("http://host: 8888/x"), Malformed);
        assert_eq!(explicit_port("http://[::1]junk/x"), Malformed);
    }

    #[test]
    fn addresses_listener_directly_recognizes_only_loopback() {
        for base in [
            "http://127.0.0.1:8888",
            "http://localhost:8888/hls",
            "http://[::1]:8888",
            "rtmp://LOCALHOST:1935/live",
            "http://user:pw@127.0.0.1:8888",
        ] {
            assert!(addresses_listener_directly(base), "{base} is loopback");
        }
        for base in [
            "https://media.example.com:8443/hls",
            "https://cdn.example.net",
            "http://10.0.0.4:8888",
            "http://[2001:db8::1]:8888",
        ] {
            assert!(!addresses_listener_directly(base), "{base} is fronted");
        }
    }

    #[test]
    fn ports_match_bases_skips_a_proxied_base_on_an_explicit_port() {
        // A TLS proxy in front of the daemon publishes its own port. That is the
        // documented production shape, not a misconfiguration, so it must not
        // block a deploy.
        let cfg = MediaMtxHostConfig {
            hls_port: 8888,
            hls_base: Some("https://media.example.com:8443/hls".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(check.passed, "detail: {}", check.detail);
    }

    #[test]
    fn ports_match_bases_defers_rather_than_blocks_when_the_base_is_unset() {
        // The app can be told its base by `AUTUMN_MEDIA__MEDIAMTX__*_BASE`, which
        // the deploy side does not hold — so an unset base warns loudly and lets
        // the deploy through, rather than blocking a correct configuration.
        let cfg = MediaMtxHostConfig {
            rtmp_port: 1936,
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.deferred);
        assert!(!check.blocking(), "an unset base must not abort a deploy");
        assert!(check.detail.contains("rtmp_base"), "{}", check.detail);
        assert!(check.detail.contains("1936"), "{}", check.detail);
        assert!(check.detail.contains("1935"), "{}", check.detail);
    }

    #[test]
    fn ports_match_bases_reports_a_base_whose_port_is_malformed() {
        let cfg = MediaMtxHostConfig {
            hls_base: Some("http://127.0.0.1:88888".to_owned()),
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_match_bases(&cfg);
        assert!(!check.passed);
        assert!(check.blocking());
        assert!(check.detail.contains("no valid port"), "{}", check.detail);
    }

    // ── Distinct listener ports (Finding P) ──────────────────────────────────

    #[test]
    fn ports_distinct_passes_for_default_config() {
        let check = mediamtx_ports_distinct(&MediaMtxHostConfig::default());
        assert!(
            check.passed,
            "default ports are all distinct: {}",
            check.detail
        );
        assert_eq!(check.name, CHECK_MEDIAMTX_PORTS_DISTINCT);
    }

    #[test]
    fn ports_distinct_fails_when_two_tcp_listeners_share_a_port() {
        // hls_port == playback_port: two MediaMTX servers asked to bind one TCP
        // port. `ss` passes on a fresh host, but the deferred restart fails.
        let cfg = MediaMtxHostConfig {
            hls_port: 8888,
            playback_port: 8888,
            ..MediaMtxHostConfig::default()
        };
        let check = mediamtx_ports_distinct(&cfg);
        assert!(!check.passed);
        // Names the duplicated value and BOTH conflicting fields.
        assert!(check.detail.contains("8888"), "detail: {}", check.detail);
        assert!(
            check.detail.contains("hls_port"),
            "detail: {}",
            check.detail
        );
        assert!(
            check.detail.contains("playback_port"),
            "detail: {}",
            check.detail
        );
        assert!(check.hint.is_some());
    }

    #[test]
    fn ports_distinct_allows_tcp_port_equal_to_udp_port() {
        // The WebRTC local UDP port (8189) is a separate protocol namespace, so a
        // TCP listener numerically equal to it is NOT a conflict.
        let cfg = MediaMtxHostConfig {
            rtmp_port: MEDIAMTX_WEBRTC_LOCAL_UDP_PORT,
            ..MediaMtxHostConfig::default()
        };
        assert_eq!(cfg.rtmp_port, cfg.webrtc_local_udp);
        let check = mediamtx_ports_distinct(&cfg);
        assert!(
            check.passed,
            "a TCP port equal to the UDP port is not a same-protocol conflict: {}",
            check.detail
        );
    }

    // ── media_host_config_from_value (Finding O merge target) ────────────────

    #[test]
    fn media_host_config_from_value_reads_merged_media_subtree() {
        let root: toml::Value = toml::from_str(
            "[media.mediamtx]\nenabled = true\napi_port = 19997\n[media.ffmpeg]\nbin = \"/opt/ff\"\n",
        )
        .expect("parses");
        let (cfg, bin) = media_host_config_from_value(root).expect("deserializes");
        assert!(cfg.enabled);
        assert_eq!(cfg.api_port, 19997);
        assert_eq!(bin, "/opt/ff");
    }

    #[test]
    fn media_host_config_from_value_absent_media_is_disabled_default() {
        let root: toml::Value = toml::from_str("[deploy]\nhost = \"h\"\n").expect("parses");
        let (cfg, bin) = media_host_config_from_value(root).expect("deserializes");
        assert!(!cfg.enabled);
        assert_eq!(bin, DEFAULT_FFMPEG_BIN);
    }

    #[test]
    fn media_host_config_from_value_fails_closed_on_ill_typed_value() {
        let root: toml::Value =
            toml::from_str("[media.mediamtx]\nenabled = \"yes\"\n").expect("parses");
        assert!(media_host_config_from_value(root).is_err());
    }

    // ── CSP origins ──────────────────────────────────────────────────────────

    #[test]
    fn required_csp_origins_lists_browser_facing_ports() {
        let origins = MediaMtxHostConfig::default().required_csp_origins();
        assert!(
            origins.contains(&"http://127.0.0.1:8889".to_owned()),
            "webrtc"
        );
        assert!(origins.contains(&"http://127.0.0.1:8888".to_owned()), "hls");
        assert!(
            origins.contains(&"http://127.0.0.1:9996".to_owned()),
            "playback"
        );
        // The control API (9997) is server-side only and must NOT be advertised.
        assert!(!origins.iter().any(|o| o.contains("9997")));
    }
}