subc_control/

lib.rs

//! Client-facing subc channel-0 control wire shapes.
//!
//! This crate is the client ↔ subc control-plane boundary. It depends only on
//! [`subc-protocol`] for shared primitives such as `RouteTarget` and
//! `BindIdentity`; clients can use it without depending on the
//! daemon implementation.

#![forbid(unsafe_code)]

use std::{collections::BTreeMap, path::PathBuf};

use serde::{
    de::{Error as _, MapAccess, SeqAccess, Visitor},
    ser::SerializeMap,
    Deserialize, Deserializer, Serialize, Serializer,
};
use subc_protocol::{
    manifest::{
        CapabilityDeclarations, EventDeclaration, ManifestProvenance, ProviderRole,
        SelfSignalDeclaration,
    },
    scope::ScopeSelector,
    session::HealthStatus,
    BindIdentity, RouteTarget,
};

pub use subc_protocol::RouteCloseReason;

mod activity;
pub use activity::*;

macro_rules! open_string_enum {
    (
        $(#[$meta:meta])*
        $name:ident {
            $( $(#[$variant_meta:meta])* $variant:ident => $wire_name:literal ),+ $(,)?
        }
    ) => {
        $(#[$meta])*
        #[derive(Debug, Clone, PartialEq, Eq)]
        pub enum $name {
            $( $(#[$variant_meta])* $variant, )+
            Unknown(String),
        }

        impl $name {
            fn wire_name(&self) -> &str {
                match self {
                    $( Self::$variant => $wire_name, )+
                    Self::Unknown(value) => value,
                }
            }
        }

        impl Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
            where
                S: serde::Serializer,
            {
                serializer.serialize_str(self.wire_name())
            }
        }

        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: serde::Deserializer<'de>,
            {
                let value = String::deserialize(deserializer)?;
                Ok(match value.as_str() {
                    $( $wire_name => Self::$variant, )+
                    _ => Self::Unknown(value),
                })
            }
        }
    };
}

/// Daemon-spawned consumer identity presented on route.open.
#[derive(Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct ConsumerIdentity {
    pub module_id: String,
    pub launch_nonce: String,
}

// Hand-written so the launch nonce is never printed. The nonce is the credential
// that attributes a connection to a supervised module, and a derived Debug would
// write it into any log line or panic message that formats this value. Same
// reasoning as ConnectionInfo's Debug in subc-transport.
impl std::fmt::Debug for ConsumerIdentity {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConsumerIdentity")
            .field("module_id", &self.module_id)
            .field(
                "launch_nonce",
                &format_args!("<{} bytes redacted>", self.launch_nonce.len()),
            )
            .finish()
    }
}

/// Reserved dotted operation prefixes for the v0.4 control vocabulary.
///
/// `scheduler.` and `watch.` were reserved here from v0.4 until 2026-08-10 and
/// were removed deliberately rather than left as placeholders: neither was ever
/// implemented, and both capabilities are now owned elsewhere by ruling --
/// scheduled tasks belong to the session runtime (prefrontal) because the
/// daemon is state-free routing, and external-event watching belongs to the
/// connectors module (plexus). A reserved name for something that will never be
/// built here reads as a roadmap commitment to anyone surveying the protocol,
/// and it recruited exactly that misunderstanding from an outside contributor.
pub mod ops {
    pub const SERVER: &str = "server.";
    pub const CATALOG: &str = "catalog.";
    pub const ROUTE: &str = "route.";
    pub const SUPERVISOR: &str = "supervisor.";
    pub const CONFIG: &str = "config.";

    pub const SERVER_DESCRIBE: &str = "server.describe";
    pub const CATALOG_LIST: &str = "catalog.list";
    pub const ROUTE_OPEN: &str = "route.open";
    pub const ROUTE_POLL: &str = "route.poll";
    pub const ROUTE_CLOSING: &str = "route.closing";
    pub const ROUTE_CLOSED: &str = "route.closed";
    pub const SUPERVISOR_LIST: &str = "supervisor.list";
    pub const SUPERVISOR_RESTART: &str = "supervisor.restart";
    pub const SUPERVISOR_SWAP: &str = "supervisor.swap";
    pub const SUPERVISOR_RELOAD: &str = "supervisor.reload";
    pub const SUPERVISOR_RESCAN: &str = "supervisor.rescan";
    pub const SUPERVISOR_RELEASE_RESERVED: &str = "supervisor.release_reserved";
    pub const SUPERVISOR_SET_ENABLED: &str = "supervisor.set_enabled";
    pub const SUPERVISOR_HEALTH_PROBE: &str = "supervisor.health_probe";
    pub const SUPERVISOR_HEALTH: &str = "supervisor.health";
    pub const SUPERVISOR_STDERR_TAIL: &str = "supervisor.stderr_tail";
    pub const SUPERVISOR_TERMINALS: &str = "supervisor.terminals";
    pub const SUPERVISOR_ROUTES: &str = "supervisor.routes";
    pub const SUPERVISOR_PROVENANCE: &str = "supervisor.provenance";
    pub const SUPERVISOR_SPAWN_SNAPSHOT: &str = "supervisor.spawn_snapshot";
    pub const SUPERVISOR_SPAWN_SUBSCRIBE: &str = "supervisor.spawn_subscribe";
    pub const ACTIVITY_RECORD: &str = "activity.record";
    pub const ACTIVITY_QUERY: &str = "activity.query";
    pub const ACTIVITY_SUBSCRIBE: &str = "activity.subscribe";
}

/// Client-originated channel-0 control RPC body.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "op")]
// RouteOpen carries the complete route metadata, while several control operations
// are markers; retain the direct public wire shape instead of boxing its fields.
#[allow(clippy::large_enum_variant)]
pub enum ClientControlRequest {
    #[serde(rename = "server.describe")]
    ServerDescribe {},
    #[serde(rename = "catalog.list")]
    CatalogList {
        /// Absent lists every registered module; present narrows to one. A
        /// narrowed list for an unregistered id is an empty list rather than an
        /// error, so absent and unregistered are distinguishable only by which
        /// question you asked.
        #[serde(default)]
        module_id: Option<String>,
    },
    #[serde(rename = "route.open")]
    RouteOpen {
        target: RouteTarget,
        identity: BindIdentity,
        /// The consumer's claim to a supervised launch, which the daemon verifies
        /// against its live spawn nonces before stamping a principal.
        ///
        /// Absent is a legitimate shape, not an omission: a direct key-holder has
        /// no launch nonce to present, and the daemon stamps `Direct`. So absence
        /// means NO CLAIM WAS MADE, never that a claim was refused — a refused
        /// claim is an error frame and the route never opens. A provider deciding
        /// what to trust reads the stamped principal on the bind, not this.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        consumer_identity: Option<ConsumerIdentity>,
        /// Consumer-declared reverse-request capabilities for the route. This is
        /// an unverified declaration, not a privilege grant; if a consumer
        /// over-declares, providers may still send reverse requests that later
        /// time out or deny. Providers must treat an absent field as no
        /// reverse-request capability. The vocabulary is open strings; known MCP
        /// method-family values today are "elicitation", "sampling", and
        /// "roots".
        #[serde(default, skip_serializing_if = "Option::is_none")]
        consumer_capabilities: Option<Vec<String>>,
        /// The versions of provider roles this consumer speaks on the route,
        /// role name to version (`{"tool-provider": "v1"}`). Like
        /// `consumer_capabilities` it is an unverified declaration that grants
        /// nothing: the daemon checks its shape
        /// (`subc_protocol::session::validate_role_versions`), refuses a
        /// malformed one as `invalid_request` naming `role_versions`, and
        /// copies it unchanged onto the module's bind. An empty map means the
        /// same as an absent one. Send it only to a daemon advertising
        /// `route-role-versions/v1`: an older daemon drops unknown fields, so
        /// the provider would never see it.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        role_versions: Option<BTreeMap<String, String>>,
        /// Opaque admission facts supplied by the configured carrier module.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        admission_facts: Option<serde_json::Value>,
        /// The scope to open the route under. The daemon admits the open only
        /// when the opener (its attested principal, never anything in this
        /// body) is the scope's owner or a listed carrier, and stamps the
        /// scope on the module's bind. Send it only to a daemon advertising
        /// `scopes/v1`: an older daemon drops unknown fields and would open an
        /// unscoped route.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        scope: Option<ScopeSelector>,
    },
    #[serde(rename = "route.poll")]
    RoutePoll {
        route_channel: u16,
        route_epoch: u32,
        kind: PollKind,
    },
    #[serde(rename = "supervisor.list")]
    SupervisorList {},
    /// Read the live supervised processes and the event cursor atomically.
    #[serde(rename = "supervisor.spawn_snapshot")]
    SupervisorSpawnSnapshot {},
    /// Replay spawn events after `since`, then remain open for live events.
    ///
    /// The cursor is one value copied from a snapshot or event. It includes the
    /// daemon incarnation so a restarted daemon rejects an earlier instance's
    /// sequence instead of treating it as a position in the current stream.
    ///
    /// Refusals and terminal errors, as `Error` frames on the request's corr:
    /// - `spawn_cursor_incarnation_mismatch` (detail `current_daemon_incarnation`):
    ///   `since` names another daemon incarnation.
    /// - `spawn_cursor_too_old` (detail `oldest_retained_cursor`): `since`
    ///   predates the retained event ring.
    /// - `spawn_subscriber_lagged` (detail `first_undelivered_cursor`): the open
    ///   stream fell too far behind and the daemon dropped it. It arrives after
    ///   every event that was already queued and ends the stream; resubscribe
    ///   with `since` set to the last cursor received.
    #[serde(rename = "supervisor.spawn_subscribe")]
    SupervisorSpawnSubscribe {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        since: Option<SpawnCursor>,
    },
    #[serde(rename = "supervisor.restart")]
    SupervisorRestart {
        module_id: String,
        /// Optional per-restart override of the module's drain budget, in ms.
        /// Absent: the module's configured `drain_timeout_ms` (or the daemon
        /// default) applies. `0` tears down without waiting: use it to restart a
        /// wedged module whose stuck in-flight request would never settle, so
        /// waiting for a drain would be pointless.
        /// Additive; older daemons silently ignore this field and use their
        /// configured/default drain budget, even when the sender requests `0`.
        /// Senders that require the override must use a daemon version that
        /// supports it (the CLI only sends it when a flag is passed).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        drain_timeout_ms: Option<u64>,
    },
    /// Blue/green restart: start a replacement beside the running process, keep
    /// routing to the running one until the replacement declares itself ready,
    /// then move new routes over and drain the old process.
    ///
    /// Refused before anything is spawned unless the module's daemon config
    /// declares `overlap: "safe"`: two processes on one single-writer store is
    /// a data hazard, so the default is exclusive. Answered once the swap has
    /// either cut over (the old process is still draining) or failed; a failure
    /// leaves the old process serving and undrained, and names the arm in
    /// `ErrorBody.detail`.
    ///
    /// A new op rather than a flag on `supervisor.restart`: a daemon that
    /// predates swap rejects an unknown op, whereas an unknown field could be
    /// dropped and turned into a plain restart.
    #[serde(rename = "supervisor.swap")]
    SupervisorSwap {
        module_id: String,
        /// How long the replacement may take to register and declare itself
        /// ready before the swap is abandoned. Absent: the daemon default.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ready_timeout_ms: Option<u64>,
    },
    #[serde(rename = "supervisor.reload")]
    SupervisorReload { module_id: String },
    #[serde(rename = "supervisor.rescan")]
    SupervisorRescan {
        /// Compute the reconciliation and return it WITHOUT applying it.
        ///
        /// Rescan retires any supervised module absent from the config, which
        /// stops live processes. Both halves of that decision are inspectable in
        /// advance -- the config is a file, the running set is `supervisor.list`
        /// -- but nothing reconstructs the diff for the operator, so it is read
        /// from the result table AFTER the retires have happened.
        ///
        /// A preview must be computed daemon-side rather than by a client, because
        /// a client would have to locate the daemon's config itself: two rules
        /// selecting one subject, agreeing until someone runs a daemon with a
        /// non-default config. A preview that can describe a different file than
        /// the operation reads is worse than none, because it is believed.
        ///
        /// Defaults to false so an existing client sending `{}` still executes,
        /// and is OMITTED when false so the bytes an existing client sends are
        /// unchanged. Serialising `preview:false` would have altered the request's
        /// wire form for every caller that never asked for a preview -- caught by
        /// the golden fixture, which is the whole reason that pin exists.
        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
        preview: bool,
    },
    /// Retire the retained exact-id reservation after its configuration entry has
    /// been removed. This is intentionally separate from rescan so deleting
    /// configuration never silently opens a protected module id to registration.
    #[serde(rename = "supervisor.release_reserved")]
    SupervisorReleaseReserved { module_id: String },
    #[serde(rename = "supervisor.set_enabled")]
    SupervisorSetEnabled { module_id: String, enabled: bool },
    #[serde(rename = "supervisor.health_probe")]
    SupervisorHealthProbe { module_id: String },
    #[serde(rename = "supervisor.health")]
    SupervisorHealth {},
    /// Enumerate the routes currently served by one supervised module, or every
    /// module when omitted.
    ///
    /// This privileged census is control-plane-only. It is deliberately not an
    /// MCP facade or agent-tool operation: callers holding the daemon control
    /// connection may inspect live route ownership, while agent-facing modules
    /// must not be able to address that surface at all.
    ///
    /// The daemon answers from its forwarding table under a read lock and never
    /// consults a module. That makes the read safe during a drain, when a module
    /// cannot be queried without recreating the hang/restart hazard that route
    /// status reads avoid.
    #[serde(rename = "supervisor.routes")]
    SupervisorRoutes {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        module_id: Option<String>,
    },
    /// Report source-tagged provenance for supervised modules, optionally narrowed
    /// to one module.
    #[serde(rename = "supervisor.provenance")]
    SupervisorProvenance {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        module_id: Option<String>,
    },
    /// Retained stderr for one module.
    ///
    /// A separate op rather than a field on `supervisor.list`: the tail is
    /// kilobytes per module and `list` renders every module, so carrying it in
    /// the snapshot would charge every status read for a payload almost no
    /// caller wants. Caps ride on the REQUEST so a caller wanting twenty lines
    /// and one wanting the whole ring need no separate fields anywhere.
    #[serde(rename = "supervisor.stderr_tail")]
    SupervisorStderrTail {
        module_id: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        max_lines: Option<u32>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        max_bytes: Option<u32>,
    },
    /// Retained terminal exits for one module.
    ///
    /// This stays separate from `supervisor.list`: a history grows with every
    /// incident, while the list is a current-state read most callers issue often.
    ///
    /// The read MUST stay off the supervisor command channel — it reads the
    /// module's shared ring directly. This is a requirement, not an
    /// optimisation: when the supervision task itself dies, every
    /// command-channel op returns `CommandClosed`, and that is precisely the
    /// moment an operator needs the exit history most. A history reachable only
    /// through the machinery whose death you are diagnosing is unreachable when
    /// it matters. Proven failure mode, not a hypothetical.
    #[serde(rename = "supervisor.terminals")]
    SupervisorTerminals { module_id: String },
}

/// subc's channel-0 response body for client control RPCs.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "op")]
pub enum ClientControlResponse {
    #[serde(rename = "server.describe")]
    ServerDescribe {
        protocol_ver: u8,
        subc_ops: Vec<String>,
        capabilities: Vec<String>,
        connected_clients: u64,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        counters: Option<serde_json::Value>,
        /// Git commit the daemon was built from, or "unavailable" when the
        /// build could not read it. The crate version cannot discriminate a
        /// skewed daemon/CLI pair (it moves per release, not per commit), so
        /// this is the identity a consumer compares against its own embedded
        /// commit to detect that it is talking to an older build than it was
        /// compiled with. Absent from daemons predating the field.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        build_git_sha: Option<String>,
        /// sha256 of the workspace Cargo.lock at build time, or "unavailable".
        /// Answers "which dependency set" where the commit answers "which
        /// source"; a commit match with a digest mismatch means a rebuild
        /// against edited dependencies. Absent from daemons predating the
        /// field.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        build_lock_digest: Option<String>,
        /// Daemon-evaluated capability requirements. Present when the configured
        /// fleet has declarations to evaluate, so operators can inspect an absent
        /// required capability without parsing daemon logs.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        capability_requirements: Vec<CapabilityRequirementStatus>,
        /// The daemon's machine id (`subc_protocol::MachineId`), the same value
        /// every module receives on `HELLO_ACK`. A name for this machine, never
        /// an authority: nothing may grant trust because two parties report the
        /// same value. Absent from daemons predating the field.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        machine_id: Option<String>,
    },
    #[serde(rename = "catalog.list")]
    CatalogList {
        generation: u64,
        modules: Vec<CatalogEntry>,
        subc_ops: Vec<String>,
    },
    #[serde(rename = "route.open")]
    RouteOpen {
        route_channel: u16,
        route_epoch: u32,
    },
    #[serde(rename = "route.poll")]
    RoutePoll {
        route_channel: u16,
        route_epoch: u32,
        status: Option<String>,
        live: Option<bool>,
    },
    #[serde(rename = "supervisor.list")]
    SupervisorList {
        generation: u64,
        modules: Vec<SupervisorEntry>,
    },
    #[serde(rename = "supervisor.spawn_snapshot")]
    SupervisorSpawnSnapshot {
        #[serde(flatten)]
        snapshot: SpawnSnapshot,
    },
    #[serde(rename = "supervisor.ack")]
    SupervisorAck { module_id: String, applied: bool },
    #[serde(rename = "supervisor.rescan")]
    SupervisorRescan {
        #[serde(flatten)]
        result: SupervisorRescanResult,
    },
    #[serde(rename = "supervisor.health_probe")]
    SupervisorHealthProbe {
        module_id: String,
        status: HealthStatus,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        detail: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        metrics: Option<serde_json::Value>,
    },
    #[serde(rename = "supervisor.health")]
    SupervisorHealth {
        generation: u64,
        modules: Vec<SupervisorHealthEntry>,
    },
    #[serde(rename = "supervisor.routes")]
    SupervisorRoutes { modules: Vec<SupervisorRouteModule> },
    #[serde(rename = "supervisor.provenance")]
    SupervisorProvenance {
        daemon: SupervisorDaemonProvenance,
        modules: Vec<SupervisorModuleProvenance>,
    },
    #[serde(rename = "supervisor.stderr_tail")]
    SupervisorStderrTail {
        module_id: String,
        #[serde(flatten)]
        tail: StderrTail,
    },
    #[serde(rename = "supervisor.terminals")]
    SupervisorTerminals {
        module_id: String,
        #[serde(flatten)]
        terminals: TerminalHistory,
    },
}

/// Daemon-originated channel-0 control push body.
///
/// A module cannot originate these pushes: subc creates them from its own
/// forwarding state and enqueues them directly to client connection sinks.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "op")]
pub enum ClientControlPush {
    #[serde(rename = "route.closing")]
    RouteClosing {
        module_id: String,
        /// Channel numbers identifying affected routes on this client's connection.
        /// Channels used by other clients or by the provider are not included.
        #[serde(default)]
        channels: Vec<u16>,
        reason: RouteCloseReason,
    },
    #[serde(rename = "route.closed")]
    RouteClosed {
        module_id: String,
        /// Channel numbers identifying affected routes on this client's connection.
        /// Channels used by other clients or by the provider are not included.
        #[serde(default)]
        channels: Vec<u16>,
        reason: RouteCloseReason,
        /// The exact result of the forwarding-quiescence wait for live routes.
        drained: bool,
        /// Pending route.bind relays forced down before that wait. They are not
        /// covered by `drained`, even when live routes quiesced.
        abandoned: u32,
        /// Subscription credits captured and excluded from this drain's wire predicate.
        #[serde(default)]
        excluded_subscriptions: u32,
        /// Whether subc will leave this module down until operator action.
        ///
        /// The claim covers daemon-owned recovery only. `None` is accepted only
        /// from daemons that predate this field; every current daemon emission is
        /// `Some`.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        terminal: Option<bool>,
    },
}

/// A daemon-incarnation-scoped position in the supervised spawn event stream.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SpawnCursor {
    pub daemon_incarnation: String,
    pub seq: u64,
}

/// One process present in an atomic supervisor spawn snapshot.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LiveSpawn {
    pub module_id: String,
    pub spawn_generation: u64,
    pub pid: u32,
    pub spawned_at_ms: u64,
}

/// Atomic live-process census and the cursor at which it was observed.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SpawnSnapshot {
    pub cursor: SpawnCursor,
    /// Maximum retained event count for this daemon.
    pub ring_bound: u64,
    pub live: Vec<LiveSpawn>,
}

/// Fact observed by the supervisor when a child process starts or exits.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum SpawnEventKind {
    Spawned,
    Exited,
}

/// One retained or live spawn event.
///
/// Exit events intentionally carry no disposition or reason because exit
/// classification is recorded separately; credential consumers revoke on every
/// exit regardless of the cause.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SpawnEvent {
    pub cursor: SpawnCursor,
    pub kind: SpawnEventKind,
    pub module_id: String,
    pub spawn_generation: u64,
    pub pid: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_signal: Option<i32>,
}

/// A module's retained stderr, oldest entry first.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct StderrTail {
    pub capture: StderrCaptureState,
    pub entries: Vec<StderrTailEntry>,
    /// Lines not present above: evicted by the ring, or held back by this
    /// request's own caps.
    ///
    /// Non-zero means the first entry is not the first line the module wrote. A
    /// reader hunting a cause needs that, or an absent explanation reads as a
    /// module that never gave one.
    ///
    /// Zero is skipped so the common complete-tail case stays compact.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub dropped_lines: u64,
}

/// Live routes served by one module.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorRouteModule {
    pub module_id: String,
    pub routes: Vec<SupervisorRoute>,
}

/// One live consumer route in a [`SupervisorRouteModule`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorRoute {
    pub consumer: SupervisorRouteConsumer,
    /// Milliseconds since the daemon bound this route.
    pub age_ms: u64,
    /// True once the endpoint began draining. Draining routes remain visible so
    /// a census does not misreport an already-closing route as live.
    pub draining: bool,
    /// WHY the endpoint is draining — the same reason vocabulary the
    /// route.closing push carries — present exactly when `draining` is true.
    /// Additive: older daemons omit it, and a census consumer must treat a
    /// draining route without a reason as draining-for-an-unstated-reason,
    /// never as not-draining.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub drain_reason: Option<RouteCloseReason>,
}

/// Source-tagged provenance for one supervised module.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorModuleProvenance {
    pub module_id: String,
    pub module_declared: ModuleDeclaredProvenance,
    pub daemon_observed: SupervisorObservedProcess,
}

/// A module's declared build metadata, if its HELLO manifest carried it.
#[derive(Debug, Clone, PartialEq)]
pub enum ModuleDeclaredProvenance {
    Reported {
        build: ManifestProvenance,
    },
    Unverifiable,
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

/// Process facts observed by the daemon for a supervised module.
///
/// Build claims remain under `module_declared`; mixing them here would imply the
/// daemon independently observed module-provided metadata.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorObservedProcess {
    /// The daemon's Windows delivery record for this process: `pipe` after the
    /// process whose PID was returned by spawn has read all nonce bytes; `env`
    /// when that delivery was not completed and SUBC_LAUNCH_NONCE was offered.
    /// This is returned under `daemon_observed.launch_nonce_source` by
    /// `supervisor.provenance`, and is absent when the daemon cannot report it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub launch_nonce_source: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pid: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spawned_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spawned_from: Option<PathBuf>,
    pub running_image: RunningImageAgreement,
}

/// Independent comparisons of configured path and running image at list time.
/// An absent verdict means the daemon predates this field, not agreement.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PendingReloadVerdict {
    pub path: ReloadPathAgreement,
    pub image: RunningImageAgreement,
}

/// Whether the running process was spawned from the currently configured program.
#[derive(Debug, Clone, PartialEq)]
pub enum ReloadPathAgreement {
    Match,
    Mismatch {
        configured: PathBuf,
        spawned_from: PathBuf,
    },
    Unavailable {
        reason: ReloadPathUnavailableReason,
    },
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

open_string_enum! {
    /// Why configured and spawned paths cannot be compared.
    ReloadPathUnavailableReason {
        NotRunning => "not_running",
        SpawnedPathUnavailable => "spawned_path_unavailable",
    }
}

/// Daemon provenance paired with its runtime process observation.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorDaemonProvenance {
    pub daemon_build: DaemonBuildProvenance,
    pub daemon_observed: DaemonObservedProcess,
}

/// Build metadata embedded in the daemon binary.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct DaemonBuildProvenance {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build_git_sha: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build_lock_digest: Option<String>,
}

/// Runtime process facts observed for the daemon itself.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct DaemonObservedProcess {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pid: Option<u32>,
    /// Wall time derived from suspend-inclusive elapsed time at each read. Clock
    /// correction can move it by the size of a clock step, and even without a
    /// step it may vary by about a second between reads. Do not equality-compare
    /// it. Use raw process start ticks for stable identity.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started_at_ms: Option<u64>,
    pub running_image: RunningImageAgreement,
}

/// Whether the executable currently running agrees with the spawned image.
#[derive(Debug, Clone, PartialEq)]
pub enum RunningImageAgreement {
    Match {
        evidence: RunningImageEvidence,
    },
    Mismatch {
        running: RunningImageEvidence,
        disk: RunningImageEvidence,
    },
    Unavailable {
        reason: RunningImageUnavailableReason,
    },
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

/// Platform-specific evidence used to compare a running image with its spawn path.
#[derive(Debug, Clone, PartialEq)]
pub enum RunningImageEvidence {
    LinuxProcSha256 {
        digest: String,
    },
    MacosSpawnInode {
        device: u64,
        inode: u64,
    },
    /// The spawn-time Windows file object, not verified executable content.
    WindowsSpawnFileId {
        volume_serial_number: u64,
        file_id: [u8; 16],
    },
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

open_string_enum! {
    /// Reasons why an executable identity could not be observed.
    RunningImageUnavailableReason {
        NotRunning => "not_running",
        UnsupportedPlatform => "unsupported_platform",
        RunningExecutableUnreadable => "running_executable_unreadable",
        RunningExecutablePermissionDenied => "running_executable_permission_denied",
        ProcessHidden => "process_hidden",
        SpawnedPathUnreadable => "spawned_path_unreadable",
        HashFailed => "hash_failed",
        ProcessIdentityUnconfirmed => "process_identity_unconfirmed",
    }
}

/// The identity tier the daemon can honestly report for a route consumer.
///
/// A caller that proved a live daemon-issued launch nonce is named `reserved`.
/// A direct key-holder has no such attestation, so it is reported as `direct`
/// with its connection counter instead of an invented module name.
#[derive(Debug, Clone, PartialEq)]
pub enum SupervisorRouteConsumer {
    Reserved {
        module_id: String,
    },
    Direct {
        connection_id: u64,
    },
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

/// Whether stderr is being captured for a module, and if not, why not.
///
/// A typed state rather than an empty-tail convention. "The module printed
/// nothing before dying" and "nobody was capturing" send an operator in opposite
/// directions, and rendering them alike is the defect this op exists to fix --
/// the same shape as a `detail -` that means both no-detail and never-probed.
#[derive(Debug, Clone, PartialEq)]
pub enum StderrCaptureState {
    /// A reader is attached, or was attached and saw clean EOF. An empty
    /// `entries` under this state means the module genuinely wrote nothing.
    Captured,
    /// Retained entries are valid, but the stderr reader ended before clean EOF.
    Incomplete { reason: String },
    /// No reader was attached. `entries` says nothing about what the module wrote.
    NotCaptured { reason: String },
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

#[derive(Debug, Clone, PartialEq)]
pub enum StderrTailEntry {
    Line {
        text: String,
        /// The line was cut at the per-line cap and `text` is a prefix.
        ///
        /// Carried as a field rather than left to a marker in `text` so a
        /// consumer can branch on it without string matching.
        truncated: bool,
        /// Wall-clock Unix milliseconds at which the daemon read this line off
        /// the module's pipe. `None` from a daemon that predates the field; a
        /// reader must then show no time rather than make one up.
        at_ms: Option<u64>,
    },
    /// The supervisor spawned a new process. Entries after this came from it.
    ///
    /// In-band because position is the information: which side of the restart a
    /// line falls on is unanswerable from a count.
    ProcessStart,
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
enum ModuleDeclaredProvenanceWire {
    Reported { build: ManifestProvenance },
    Unverifiable,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
enum RunningImageAgreementWire {
    Match {
        evidence: RunningImageEvidence,
    },
    Mismatch {
        running: RunningImageEvidence,
        disk: RunningImageEvidence,
    },
    Unavailable {
        reason: RunningImageUnavailableReason,
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
enum ReloadPathAgreementWire {
    Match,
    Mismatch {
        configured: PathBuf,
        spawned_from: PathBuf,
    },
    Unavailable {
        reason: ReloadPathUnavailableReason,
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "method", rename_all = "snake_case")]
enum RunningImageEvidenceWire {
    LinuxProcSha256 {
        digest: String,
    },
    MacosSpawnInode {
        device: u64,
        inode: u64,
    },
    WindowsSpawnFileId {
        volume_serial_number: u64,
        file_id: [u8; 16],
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum SupervisorRouteConsumerWire {
    Reserved { module_id: String },
    Direct { connection_id: u64 },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
enum StderrCaptureStateWire {
    Captured,
    Incomplete { reason: String },
    NotCaptured { reason: String },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
enum ChildResourceUsageWire {
    Measured(ChildResourceReading),
    Unavailable {
        reason: ChildResourceUnavailableReason,
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum StderrTailEntryWire {
    Line {
        text: String,
        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
        truncated: bool,
        // Omitted when absent so a reply without it is byte-identical to what
        // an older daemon sends. Older decoders ignore the member when present.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        at_ms: Option<u64>,
    },
    ProcessStart,
}

/// JSON values whose object members retain wire order at every depth.
#[derive(Debug, Clone, PartialEq)]
pub enum OrderedJsonValue {
    Null,
    Bool(bool),
    Number(serde_json::Number),
    String(String),
    Array(Vec<Self>),
    Object(OrderedJsonObject),
}

/// Ordered JSON members retained for an unknown tagged value.
#[derive(Debug, Clone, PartialEq)]
pub struct OrderedJsonObject(Vec<(String, OrderedJsonValue)>);

impl OrderedJsonObject {
    /// Returns the members in the order they appeared on the wire.
    pub fn as_entries(&self) -> &[(String, OrderedJsonValue)] {
        &self.0
    }

    fn into_value(self) -> serde_json::Value {
        serde_json::Value::Object(
            self.0
                .into_iter()
                .map(|(key, value)| (key, value.into_value()))
                .collect(),
        )
    }
}

impl OrderedJsonValue {
    fn into_value(self) -> serde_json::Value {
        match self {
            Self::Null => serde_json::Value::Null,
            Self::Bool(value) => serde_json::Value::Bool(value),
            Self::Number(value) => serde_json::Value::Number(value),
            Self::String(value) => serde_json::Value::String(value),
            Self::Array(values) => {
                serde_json::Value::Array(values.into_iter().map(Self::into_value).collect())
            }
            Self::Object(value) => value.into_value(),
        }
    }
}

impl Serialize for OrderedJsonValue {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Null => serializer.serialize_unit(),
            Self::Bool(value) => serializer.serialize_bool(*value),
            Self::Number(value) => value.serialize(serializer),
            Self::String(value) => serializer.serialize_str(value),
            Self::Array(values) => values.serialize(serializer),
            Self::Object(value) => value.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for OrderedJsonValue {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        struct OrderedValueVisitor;

        impl<'de> Visitor<'de> for OrderedValueVisitor {
            type Value = OrderedJsonValue;

            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                formatter.write_str("a JSON value with ordered object members")
            }

            fn visit_unit<E>(self) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::Null)
            }

            fn visit_none<E>(self) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::Null)
            }

            fn visit_some<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
            where
                D: Deserializer<'de>,
            {
                OrderedJsonValue::deserialize(deserializer)
            }

            fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::Bool(value))
            }

            fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::Number(value.into()))
            }

            fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::Number(value.into()))
            }

            fn visit_f64<E>(self, value: f64) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                serde_json::Number::from_f64(value)
                    .map(OrderedJsonValue::Number)
                    .ok_or_else(|| E::custom("non-finite JSON number"))
            }

            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::String(value.to_owned()))
            }

            fn visit_string<E>(self, value: String) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                Ok(OrderedJsonValue::String(value))
            }

            fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
            where
                A: SeqAccess<'de>,
            {
                let mut values = Vec::new();
                while let Some(value) = sequence.next_element()? {
                    values.push(value);
                }
                Ok(OrderedJsonValue::Array(values))
            }

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: MapAccess<'de>,
            {
                let mut entries = Vec::new();
                while let Some((key, value)) = map.next_entry()? {
                    entries.push((key, value));
                }
                Ok(OrderedJsonValue::Object(OrderedJsonObject(entries)))
            }
        }

        deserializer.deserialize_any(OrderedValueVisitor)
    }
}

impl Serialize for OrderedJsonObject {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut map = serializer.serialize_map(Some(self.0.len()))?;
        for (key, value) in &self.0 {
            map.serialize_entry(key, value)?;
        }
        map.end()
    }
}

impl<'de> Deserialize<'de> for OrderedJsonObject {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        struct OrderedObjectVisitor;

        impl<'de> Visitor<'de> for OrderedObjectVisitor {
            type Value = OrderedJsonObject;

            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                formatter.write_str("an object with ordered JSON members")
            }

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: MapAccess<'de>,
            {
                let mut entries = Vec::new();
                while let Some((key, value)) = map.next_entry()? {
                    entries.push((key, value));
                }
                Ok(OrderedJsonObject(entries))
            }
        }

        deserializer.deserialize_map(OrderedObjectVisitor)
    }
}

fn read_tagged<'de, D>(
    deserializer: D,
    field: &'static str,
) -> Result<(String, OrderedJsonObject), D::Error>
where
    D: Deserializer<'de>,
{
    let body = OrderedJsonObject::deserialize(deserializer)?;
    let mut tag = None;
    for (key, value) in body.as_entries() {
        if key != field {
            continue;
        }
        if tag.is_some() {
            return Err(D::Error::custom(format!(
                "tagged object has duplicate `{field}` field"
            )));
        }
        let OrderedJsonValue::String(value) = value else {
            return Err(D::Error::custom(format!(
                "tagged object has no string `{field}` field"
            )));
        };
        tag = Some(value);
    }
    let Some(tag) = tag else {
        return Err(D::Error::custom(format!(
            "tagged object has no string `{field}` field"
        )));
    };
    Ok((tag.to_string(), body))
}

fn read_ordered_tagged(
    value: OrderedJsonValue,
    field: &'static str,
) -> Result<(String, OrderedJsonObject), String> {
    let OrderedJsonValue::Object(body) = value else {
        return Err(format!("expected tagged object with `{field}` field"));
    };
    let mut tag = None;
    for (key, value) in body.as_entries() {
        if key != field {
            continue;
        }
        if tag.is_some() {
            return Err(format!("tagged object has duplicate `{field}` field"));
        }
        let OrderedJsonValue::String(value) = value else {
            return Err(format!("tagged object has no string `{field}` field"));
        };
        tag = Some(value);
    }
    let Some(tag) = tag else {
        return Err(format!("tagged object has no string `{field}` field"));
    };
    Ok((tag.to_string(), body))
}

fn ordered_field<'a>(body: &'a OrderedJsonObject, field: &str) -> Option<&'a OrderedJsonValue> {
    body.as_entries()
        .iter()
        .find_map(|(key, value)| (key == field).then_some(value))
}

fn ordered_string(body: &OrderedJsonObject, field: &str) -> Result<String, String> {
    match ordered_field(body, field) {
        Some(OrderedJsonValue::String(value)) => Ok(value.clone()),
        Some(_) => Err(format!("tagged object field `{field}` is not a string")),
        None => Err(format!("tagged object has no `{field}` field")),
    }
}

fn decode_running_image_evidence(value: OrderedJsonValue) -> Result<RunningImageEvidence, String> {
    let (tag, body) = read_ordered_tagged(value, "method")?;
    match tag.as_str() {
        "linux_proc_sha256" => Ok(RunningImageEvidence::LinuxProcSha256 {
            digest: ordered_string(&body, "digest")?,
        }),
        "macos_spawn_inode" => {
            let device = ordered_field(&body, "device")
                .and_then(|value| match value {
                    OrderedJsonValue::Number(number) => number.as_u64(),
                    _ => None,
                })
                .ok_or_else(|| "tagged object has no unsigned `device` field".to_string())?;
            let inode = ordered_field(&body, "inode")
                .and_then(|value| match value {
                    OrderedJsonValue::Number(number) => number.as_u64(),
                    _ => None,
                })
                .ok_or_else(|| "tagged object has no unsigned `inode` field".to_string())?;
            Ok(RunningImageEvidence::MacosSpawnInode { device, inode })
        }
        "windows_spawn_file_id" => {
            let wire: RunningImageEvidenceWire = serde_json::from_value(
                serde_json::to_value(&body).map_err(|error| error.to_string())?,
            )
            .map_err(|error| error.to_string())?;
            match wire {
                RunningImageEvidenceWire::WindowsSpawnFileId {
                    volume_serial_number,
                    file_id,
                } => Ok(RunningImageEvidence::WindowsSpawnFileId {
                    volume_serial_number,
                    file_id,
                }),
                _ => unreachable!(),
            }
        }
        _ => Ok(RunningImageEvidence::Unknown { tag, body }),
    }
}

impl Serialize for ModuleDeclaredProvenance {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Reported { build } => ModuleDeclaredProvenanceWire::Reported {
                build: build.clone(),
            }
            .serialize(serializer),
            Self::Unverifiable => ModuleDeclaredProvenanceWire::Unverifiable.serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for ModuleDeclaredProvenance {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "status")?;
        match tag.as_str() {
            "reported" => match serde_json::from_value(value.into_value())
                .map_err(D::Error::custom)?
            {
                ModuleDeclaredProvenanceWire::Reported { build } => Ok(Self::Reported { build }),
                ModuleDeclaredProvenanceWire::Unverifiable => unreachable!(),
            },
            "unverifiable" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    ModuleDeclaredProvenanceWire::Unverifiable => Ok(Self::Unverifiable),
                    ModuleDeclaredProvenanceWire::Reported { .. } => unreachable!(),
                }
            }
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

impl Serialize for RunningImageAgreement {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Match { evidence } => RunningImageAgreementWire::Match {
                evidence: evidence.clone(),
            }
            .serialize(serializer),
            Self::Mismatch { running, disk } => RunningImageAgreementWire::Mismatch {
                running: running.clone(),
                disk: disk.clone(),
            }
            .serialize(serializer),
            Self::Unavailable { reason } => RunningImageAgreementWire::Unavailable {
                reason: reason.clone(),
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl Serialize for ReloadPathAgreement {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Match => ReloadPathAgreementWire::Match.serialize(serializer),
            Self::Mismatch {
                configured,
                spawned_from,
            } => ReloadPathAgreementWire::Mismatch {
                configured: configured.clone(),
                spawned_from: spawned_from.clone(),
            }
            .serialize(serializer),
            Self::Unavailable { reason } => ReloadPathAgreementWire::Unavailable {
                reason: reason.clone(),
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for ReloadPathAgreement {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let (tag, body) = read_tagged(deserializer, "status")?;
        match tag.as_str() {
            "match" => Ok(Self::Match),
            "mismatch" => {
                match serde_json::from_value(body.into_value()).map_err(D::Error::custom)? {
                    ReloadPathAgreementWire::Mismatch {
                        configured,
                        spawned_from,
                    } => Ok(Self::Mismatch {
                        configured,
                        spawned_from,
                    }),
                    _ => unreachable!(),
                }
            }
            "unavailable" => match serde_json::from_value(body.into_value())
                .map_err(D::Error::custom)?
            {
                ReloadPathAgreementWire::Unavailable { reason } => Ok(Self::Unavailable { reason }),
                _ => unreachable!(),
            },
            _ => Ok(Self::Unknown { tag, body }),
        }
    }
}

impl<'de> Deserialize<'de> for RunningImageAgreement {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "status")?;
        match tag.as_str() {
            "match" => Ok(Self::Match {
                evidence: decode_running_image_evidence(
                    ordered_field(&value, "evidence")
                        .cloned()
                        .ok_or_else(|| D::Error::custom("tagged object has no `evidence` field"))?,
                )
                .map_err(D::Error::custom)?,
            }),
            "mismatch" => Ok(Self::Mismatch {
                running: decode_running_image_evidence(
                    ordered_field(&value, "running")
                        .cloned()
                        .ok_or_else(|| D::Error::custom("tagged object has no `running` field"))?,
                )
                .map_err(D::Error::custom)?,
                disk: decode_running_image_evidence(
                    ordered_field(&value, "disk")
                        .cloned()
                        .ok_or_else(|| D::Error::custom("tagged object has no `disk` field"))?,
                )
                .map_err(D::Error::custom)?,
            }),
            "unavailable" => Ok(Self::Unavailable {
                reason: serde_json::from_value(
                    ordered_field(&value, "reason")
                        .cloned()
                        .ok_or_else(|| D::Error::custom("tagged object has no `reason` field"))?
                        .into_value(),
                )
                .map_err(D::Error::custom)?,
            }),
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

impl Serialize for ChildResourceUsage {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Measured(reading) => {
                ChildResourceUsageWire::Measured(reading.clone()).serialize(serializer)
            }
            Self::Unavailable { reason } => ChildResourceUsageWire::Unavailable {
                reason: reason.clone(),
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for ChildResourceUsage {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let (tag, body) = read_tagged(deserializer, "status")?;
        match tag.as_str() {
            "measured" | "unavailable" => {
                match serde_json::from_value(body.into_value()).map_err(D::Error::custom)? {
                    ChildResourceUsageWire::Measured(reading) => Ok(Self::Measured(reading)),
                    ChildResourceUsageWire::Unavailable { reason } => {
                        Ok(Self::Unavailable { reason })
                    }
                }
            }
            _ => Ok(Self::Unknown { tag, body }),
        }
    }
}

impl Serialize for RunningImageEvidence {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::LinuxProcSha256 { digest } => RunningImageEvidenceWire::LinuxProcSha256 {
                digest: digest.clone(),
            }
            .serialize(serializer),
            Self::MacosSpawnInode { device, inode } => RunningImageEvidenceWire::MacosSpawnInode {
                device: *device,
                inode: *inode,
            }
            .serialize(serializer),
            Self::WindowsSpawnFileId {
                volume_serial_number,
                file_id,
            } => RunningImageEvidenceWire::WindowsSpawnFileId {
                volume_serial_number: *volume_serial_number,
                file_id: *file_id,
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for RunningImageEvidence {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "method")?;
        match tag.as_str() {
            "linux_proc_sha256" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    RunningImageEvidenceWire::LinuxProcSha256 { digest } => {
                        Ok(Self::LinuxProcSha256 { digest })
                    }
                    _ => unreachable!(),
                }
            }
            "macos_spawn_inode" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    RunningImageEvidenceWire::MacosSpawnInode { device, inode } => {
                        Ok(Self::MacosSpawnInode { device, inode })
                    }
                    _ => unreachable!(),
                }
            }
            "windows_spawn_file_id" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    RunningImageEvidenceWire::WindowsSpawnFileId {
                        volume_serial_number,
                        file_id,
                    } => Ok(Self::WindowsSpawnFileId {
                        volume_serial_number,
                        file_id,
                    }),
                    _ => unreachable!(),
                }
            }
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

impl Serialize for SupervisorRouteConsumer {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Reserved { module_id } => SupervisorRouteConsumerWire::Reserved {
                module_id: module_id.clone(),
            }
            .serialize(serializer),
            Self::Direct { connection_id } => SupervisorRouteConsumerWire::Direct {
                connection_id: *connection_id,
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for SupervisorRouteConsumer {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "kind")?;
        match tag.as_str() {
            "reserved" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    SupervisorRouteConsumerWire::Reserved { module_id } => {
                        Ok(Self::Reserved { module_id })
                    }
                    _ => unreachable!(),
                }
            }
            "direct" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    SupervisorRouteConsumerWire::Direct { connection_id } => {
                        Ok(Self::Direct { connection_id })
                    }
                    _ => unreachable!(),
                }
            }
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

impl Serialize for StderrCaptureState {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Captured => StderrCaptureStateWire::Captured.serialize(serializer),
            Self::Incomplete { reason } => StderrCaptureStateWire::Incomplete {
                reason: reason.clone(),
            }
            .serialize(serializer),
            Self::NotCaptured { reason } => StderrCaptureStateWire::NotCaptured {
                reason: reason.clone(),
            }
            .serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for StderrCaptureState {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "state")?;
        match tag.as_str() {
            "captured" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    StderrCaptureStateWire::Captured => Ok(Self::Captured),
                    _ => unreachable!(),
                }
            }
            "incomplete" => match serde_json::from_value(value.into_value())
                .map_err(D::Error::custom)?
            {
                StderrCaptureStateWire::Incomplete { reason } => Ok(Self::Incomplete { reason }),
                _ => unreachable!(),
            },
            "not_captured" => match serde_json::from_value(value.into_value())
                .map_err(D::Error::custom)?
            {
                StderrCaptureStateWire::NotCaptured { reason } => Ok(Self::NotCaptured { reason }),
                _ => unreachable!(),
            },
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

impl Serialize for StderrTailEntry {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match self {
            Self::Line {
                text,
                truncated,
                at_ms,
            } => StderrTailEntryWire::Line {
                text: text.clone(),
                truncated: *truncated,
                at_ms: *at_ms,
            }
            .serialize(serializer),
            Self::ProcessStart => StderrTailEntryWire::ProcessStart.serialize(serializer),
            Self::Unknown { body, .. } => body.serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for StderrTailEntry {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (tag, value) = read_tagged(deserializer, "kind")?;
        match tag.as_str() {
            "line" => match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                StderrTailEntryWire::Line {
                    text,
                    truncated,
                    at_ms,
                } => Ok(Self::Line {
                    text,
                    truncated,
                    at_ms,
                }),
                _ => unreachable!(),
            },
            "process_start" => {
                match serde_json::from_value(value.into_value()).map_err(D::Error::custom)? {
                    StderrTailEntryWire::ProcessStart => Ok(Self::ProcessStart),
                    _ => unreachable!(),
                }
            }
            _ => Ok(Self::Unknown { tag, body: value }),
        }
    }
}

fn is_zero_u64(value: &u64) -> bool {
    *value == 0
}

fn default_true() -> bool {
    true
}

/// Bounded terminal history for one module, oldest retained record first.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TerminalHistory {
    /// Unix milliseconds at the current daemon's start; entries may predate it.
    pub daemon_started_at_ms: u64,
    pub entries: Vec<TerminalEntry>,
    /// Exits evicted by the current daemon's ring, possibly recovered from its
    /// journal. Not a count of missing exits: expired journal totals are unknown.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub dropped: u64,
    /// Unparseable or incomplete lines across the shared journal, including
    /// lines whose module cannot be determined. Zero on older daemons.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub journal_skipped_lines: u64,
    /// Files that could not be read completely, excluding absent generations.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub journal_read_errors: u64,
    /// Failed journal appends across all modules in the current daemon.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub journal_write_failures: u64,
}

/// One terminal child exit and the supervisor action it selected.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TerminalEntry {
    /// Opaque identity of the daemon that observed this exit; absent on older
    /// daemons. Different tokens mean different lifetimes, not chronological order.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub daemon_incarnation: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_signal: Option<i32>,
    pub at_ms: u64,
    pub disposition: TerminalDisposition,
    /// Supervisor classification of this exit. Absent on daemons that predate
    /// the field; unknown future kinds remain readable instead of failing the
    /// enclosing terminal record.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_kind: Option<TerminalExitKind>,
    /// Why the supervisor chose this disposition, when the disposition alone
    /// does not say. A `failed` record carries the exhausted crash budget here
    /// (`crash budget exhausted: max_restarts=3 within window_secs=600`), which
    /// is the difference between an operator seeing "it failed" and seeing which
    /// limit stopped it. Prose for humans: render it, never parse it. Absent for
    /// ordinary dispositions and on daemons predating the field.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub disposition_detail: Option<String>,
}

/// Exit classification carried by supervisor history and census records.
///
/// This is an open string enum so future daemon variants degrade to a readable
/// unknown kind rather than making a consumer discard the enclosing record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TerminalExitKind {
    Clean,
    Crash,
    DeliberateSeverance,
    Unknown(String),
}

impl TerminalExitKind {
    fn wire_name(&self) -> &str {
        match self {
            Self::Clean => "clean",
            Self::Crash => "crash",
            Self::DeliberateSeverance => "deliberate_severance",
            Self::Unknown(value) => value,
        }
    }
}

impl Serialize for TerminalExitKind {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.serialize_str(self.wire_name())
    }
}

impl<'de> Deserialize<'de> for TerminalExitKind {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Ok(match value.as_str() {
            "clean" => Self::Clean,
            "crash" => Self::Crash,
            "deliberate_severance" => Self::DeliberateSeverance,
            _ => Self::Unknown(value),
        })
    }
}

open_string_enum! {
    /// The supervisor disposition selected after observing a terminal exit.
    TerminalDisposition {
        Stopped => "stopped",
        Disabled => "disabled",
        Failed => "failed",
        Restarting => "restarting",
        /// The child exited after the daemon had begun its own announced
        /// shutdown, whatever its exit code or signal. Such an exit is not
        /// a crash and is never followed by a respawn. Readers that predate
        /// this value decode it as `Unknown("daemon_shutdown")`.
        DaemonShutdown => "daemon_shutdown",
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum PollKind {
    Status,
    Liveness,
}

/// A registered module's catalog projection.
///
/// Use [`CatalogEntry::new`] and its `with_*` methods to construct entries so
/// future fields do not require changes to callers.
/// Struct literals and functional record updates are not supported outside this crate:
///
/// ```compile_fail,E0639
/// use subc_control::CatalogEntry;
/// let entry = CatalogEntry::new("provider", vec![], vec![]);
/// let entry = CatalogEntry { ready: false, ..entry };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[non_exhaustive]
pub struct CatalogEntry {
    pub module_id: String,
    /// Whether the registered module currently accepts new route binds.
    ///
    /// This is the module's EFFECTIVE readiness: its own declared readiness
    /// AND every one of its `need: required` capabilities having a registered
    /// provider. It is exactly the condition `route.open` checks, so a caller
    /// reading `false` here will be refused with `module_warming`; `not_ready`
    /// says why.
    ///
    /// Older daemons omit this field and are interpreted as ready. Daemons that
    /// predate `not_ready` report declared readiness only.
    #[serde(default = "default_true")]
    pub ready: bool,
    /// Why `ready` is false, in the same shape `route.open` puts in the
    /// `detail` of its `module_warming` refusal. Absent when the module is
    /// ready, and absent from daemons that predate the field.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub not_ready: Option<NotReadyReason>,
    /// The registered module's self-declared build version, projected from its
    /// manifest so a consumer can tell WHICH BUILD of a module it is talking
    /// to at connect time.
    ///
    /// Without this, a client compiled against a module's current source reads
    /// a contract that is true of the repository and false of the running
    /// process -- the types match, the JSON decodes, and the meaning has
    /// changed. That failure carries no error to notice; the version in the
    /// catalog turns a semantic skew into a log line at connect instead of a
    /// wrong sentence on a user's screen.
    ///
    /// Optional on the wire only because entries serialized by older daemons
    /// lack it: absent means "daemon predates the field", never "module has
    /// no version" (the manifest field is required at registration).
    ///
    /// The reading is ARMED BY OBSERVATION, not by this documentation: until
    /// a consumer has seen at least one populated entry from the daemon it is
    /// connected to, an all-None catalog is indistinguishable from an old
    /// daemon, and a client shipping the documented reading against it would
    /// hold a guarantee it does not have.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub module_version: Option<String>,
    pub roles: Vec<ProviderRole>,
    pub control_ops: Vec<String>,
    /// Static capability declarations from the registering module's manifest.
    ///
    /// Optional on the wire so consumers connected to a daemon that predates the
    /// capability grammar retain their existing catalog decoding behavior.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub capabilities: Option<CapabilityDeclarations>,
    /// Self-signal declarations mirrored verbatim from the registering module's
    /// manifest. The daemon relays these declarations without interpreting them.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub self_signals: Option<Vec<SelfSignalDeclaration>>,
    /// Event declarations mirrored verbatim from the registering module's manifest.
    /// Absent when the manifest omits `events`; an empty array records an affirmative
    /// declaration that the module publishes no events.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub events: Option<Vec<EventDeclaration>>,
}

impl CatalogEntry {
    /// Construct an entry with its required fields and wire-compatible defaults.
    #[must_use]
    pub fn new(
        module_id: impl Into<String>,
        roles: Vec<ProviderRole>,
        control_ops: Vec<String>,
    ) -> Self {
        Self {
            module_id: module_id.into(),
            ready: default_true(),
            not_ready: None,
            module_version: None,
            roles,
            control_ops,
            capabilities: None,
            self_signals: None,
            events: None,
        }
    }

    /// Set the module's effective readiness.
    #[must_use]
    pub fn with_ready(mut self, ready: bool) -> Self {
        self.ready = ready;
        self
    }

    /// Set or clear the reason the module is not ready.
    #[must_use]
    pub fn with_not_ready(mut self, not_ready: Option<NotReadyReason>) -> Self {
        self.not_ready = not_ready;
        self
    }

    /// Set or clear the module's self-declared version.
    #[must_use]
    pub fn with_module_version(mut self, module_version: Option<String>) -> Self {
        self.module_version = module_version;
        self
    }

    /// Set or clear the module's static capability declarations.
    #[must_use]
    pub fn with_capabilities(mut self, capabilities: Option<CapabilityDeclarations>) -> Self {
        self.capabilities = capabilities;
        self
    }

    /// Set or clear the module's self-signal declarations.
    #[must_use]
    pub fn with_self_signals(mut self, self_signals: Option<Vec<SelfSignalDeclaration>>) -> Self {
        self.self_signals = self_signals;
        self
    }

    /// Set or clear the event declarations mirrored from the module manifest.
    #[must_use]
    pub fn with_events(mut self, events: Option<Vec<EventDeclaration>>) -> Self {
        self.events = events;
        self
    }
}

/// Why a registered module is not accepting new route binds.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct NotReadyReason {
    /// `declared_not_ready` when the module itself said it is not ready, or
    /// `required_capability_unprovided` when a capability it declares
    /// `need: required` has no registered provider. Open vocabulary: a newer
    /// daemon may add reasons.
    pub reason: String,
    /// For `required_capability_unprovided`, the lexicographically first
    /// required capability that has no registered provider.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub capability: Option<String>,
}

impl NotReadyReason {
    pub const DECLARED_NOT_READY: &'static str = "declared_not_ready";
    pub const REQUIRED_CAPABILITY_UNPROVIDED: &'static str = "required_capability_unprovided";
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CapabilityRequirementStatus {
    pub consumer: String,
    pub capability: String,
    pub need: String,
    pub verdict: String,
    pub episode_seq: u64,
    pub config_satisfiable: bool,
    pub runtime_available: bool,
    pub detail: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SupervisorRescanResult {
    pub added: Vec<String>,
    pub removed: Vec<String>,
    pub changed_pending_reload: Vec<String>,
    /// Modules whose enabled flag differs between config and running state.
    ///
    /// Rescan calls `set_enabled` for these, so omitting them made the preview
    /// describe two of the three mutation classes it performs. A module changing
    /// only its enabled flag landed in no bucket at all -- not added, removed or
    /// changed, and deliberately not counted as unchanged either -- so the sole
    /// evidence was that the buckets no longer summed to the configured module
    /// count. A preview is consulted precisely when someone is being careful,
    /// which is the worst place to under-report.
    ///
    /// Empty is skipped so consumers written against the older shape keep
    /// parsing.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub enabled_changes: Vec<String>,
    pub unchanged: u32,
    /// True when this reconciliation was computed but NOT applied.
    ///
    /// Carried on the result rather than left to the caller's memory of what it
    /// asked for. A preview and an execution are otherwise byte-identical, so a
    /// reader who meets this output later -- in a log, a transcript, a pasted
    /// snippet -- cannot tell which one happened. Absent when false, so existing
    /// consumers see the shape they already parse.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub preview: bool,
    /// Config sections that changed but which rescan CANNOT apply, so the
    /// operator learns a daemon restart is required from the command they just
    /// ran rather than from the journal.
    ///
    /// The daemon has always detected this and logged a warning. A warning in a
    /// log is addressed to whoever is reading the log, and the person who just
    /// edited the config is by construction looking at the CLI instead: reported
    /// by an outside contributor after a module crash-looped through four
    /// respawns because a new top-level `storage` section was silently not
    /// applied, diagnosable only by journal archaeology.
    ///
    /// Names the SECTIONS rather than a boolean, because "something else
    /// changed" sends the operator back to diffing their own file -- which is
    /// the work the message exists to save.
    ///
    /// Empty is skipped, so consumers written against the older shape keep
    /// parsing.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub restart_required: Vec<String>,
    /// Required capabilities that a dry-run's resulting module set would leave
    /// unprovided. Rows are human-readable because the preview is an operator
    /// explanation, not a second manifest schema.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub capability_warnings: Vec<String>,
}

/// Which wire protocol a supervised module speaks to subc, as DECLARED in
/// daemon config. Never inferred from observed behaviour.
///
/// The distinction this exists to keep is between a module that should have
/// registered and has not yet, and one that never will. A `Subc` module that has
/// not registered is a subc module that is LATE -- it may be booting, it may be
/// wedged, and the supervisor's health probing and restart escalation are the
/// right response. A `None` module is a third-party process (the NATS server is
/// the first) that subc launches, supervises, and stops, and that is all: it
/// speaks no subc wire at all, so treating its silence as a fault would restart
/// a perfectly healthy process forever.
///
/// Inferring the difference from "has not registered within N seconds" would
/// collapse exactly the two cases that must stay apart, which is why this is a
/// declaration.
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ModuleProtocol {
    /// The module registers over channel 0, answers `health.check`, and can
    /// serve routes. Every module predating this field is one of these, which is
    /// why it is the default.
    #[default]
    Subc,
    /// The module speaks no subc wire. It is supervised as a process only.
    ///
    /// A clean exit (status 0) that the daemon did not request is restarted as
    /// a crash, counting against the restart budget, instead of being recorded
    /// as a stop. Such a module is usually a stock program that exits 0 on
    /// SIGTERM, so a stray outside signal would otherwise leave it down for
    /// good; a subc-wire module re-raises SIGTERM instead, so this rule is not
    /// needed for it.
    None,
}

/// A supervised module's current state and process observations.
///
/// Use [`SupervisorEntry::new`] and its `with_*` methods to construct entries so
/// future fields do not require changes to callers.
/// Struct literals and functional record updates are not supported outside this crate:
///
/// ```compile_fail,E0639
/// use subc_control::{SupervisorEntry, SupervisorHealthStatus};
/// let entry = SupervisorEntry::new("provider", "running", true, true, SupervisorHealthStatus::Unknown);
/// let entry = SupervisorEntry { resources: None, ..entry };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[non_exhaustive]
pub struct SupervisorEntry {
    /// A hidden previous instance blocks only this module's next spawn. This
    /// PID is not the supervised child and must not be presented as its identity.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub waiting_for_previous_pid: Option<u32>,
    pub module_id: String,
    pub state: String,
    pub enabled: bool,
    /// Whether this module is serving.
    ///
    /// For a `Subc` module: enabled, running, process alive, AND registered.
    /// For a `None` module the registration term is dropped, because a module
    /// that speaks no subc wire never registers and the daemon cannot assert
    /// more than "the process it launched is alive". READ IT WITH `protocol`:
    /// `live: true` means something weaker for a `None` module, and a renderer
    /// that prints it as a bare boolean for one is claiming more than the daemon
    /// knows.
    pub live: bool,
    /// The module's declared wire protocol. Absent on daemons predating the
    /// field, where every module was a subc module, so the default is exactly
    /// what those daemons meant.
    #[serde(default)]
    pub protocol: ModuleProtocol,
    /// Retained for one release for older status consumers: false on Unix, true
    /// on Windows. Non-wire modules receive no nonce on either platform.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub launch_nonce_env: Option<bool>,
    /// The daemon's Windows per-spawn delivery result: `pipe` after the spawned
    /// process read all bytes, otherwise `env` for the offered environment copy.
    /// Returned in `supervisor.list`, independently of the source the module
    /// declares in `manifest.provenance.launch_nonce_source` when registering.
    /// Processes that do not speak the wire protocol and stopped modules omit it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub launch_nonce_source: Option<String>,
    pub health: SupervisorHealthStatus,
    /// Computed from the stored launch spec and observed process at list time;
    /// None means an older daemon did not report this comparison.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pending_reload: Option<PendingReloadVerdict>,
    /// When the daemon last collected this module's health, as unix
    /// milliseconds. Absent means NEVER PROBED (a module inside its first probe
    /// window, whose `health` is therefore `Unknown` rather than good), not
    /// probed-long-ago. An old value and an absent one call for opposite
    /// readings, so do not render them alike.
    #[serde(default)]
    pub last_probe_ms: Option<u64>,
    /// Exit code of the module's most recent process exit, if the process has
    /// exited at least once. Survives respawn so a now-`running` module still
    /// reports what killed its previous incarnation.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_exit_code: Option<i32>,
    /// Terminating signal of the module's most recent process exit (Unix), if
    /// any. `Some(9)` = SIGKILL (OOM/jetsam/kill-on-drop), `Some(6)` = SIGABRT
    /// (often a panic-abort). Survives respawn.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_exit_signal: Option<i32>,
    /// Unix milliseconds when the most recent child exit was observed. Present
    /// even when the terminal ring is not queried, so existing list readers can
    /// order their latest observed exit against events they already received.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_exit_ms: Option<u64>,
    /// Classification of the most recent child exit. Absent on daemons that
    /// predate exit-kind reporting.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_exit_kind: Option<TerminalExitKind>,
    /// Replacement processes spawned for this module so far, against the budget
    /// that disables it.
    ///
    /// THIS IS THE COUNTER THAT ENDS A MODULE, and it is not the one beside it.
    /// `SupervisorHealthEntry::consecutive_failures` returns to zero on any
    /// successful probe, so a module can miss probes all day and read zero; this
    /// one only decreases when an operator restarts, reloads, or re-enables the
    /// module. Reaching the budget moves it to `Failed` and it stays there until
    /// somebody intervenes.
    ///
    /// So a module one restart from being disabled is indistinguishable from a
    /// freshly booted one unless this pair is read. Both are reported together
    /// because the count alone does not say how close it is.
    ///
    /// Absent from daemons predating the field, which is why it is optional
    /// rather than defaulted to zero: zero would assert a full budget.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_count: Option<u32>,
    /// Replacement processes this module is allowed before it is disabled. See
    /// `restart_count`; absent on daemons predating the field.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_restarts: Option<u32>,
    /// Replacement processes spawned over this module's entire supervisor lifetime.
    /// Unlike `restart_count`, this value is never reset by an operator action.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub lifetime_restarts: Option<u32>,
    /// Successful child spawns in this daemon incarnation. Zero means the
    /// module has not successfully spawned; every successful spawn increments
    /// the value exactly once.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spawn_generation: Option<u64>,
    /// The span `restart_count` is counted over, in seconds. The crash budget is
    /// a RATE, not a lifetime total: `restart_count` counts only the restarts
    /// inside the last `restart_window_secs`, and older ones no longer hold a
    /// slot. Without this field a reader cannot tell "2 of 3 crashes, ever" from
    /// "2 of 3 crashes in the last ten minutes", and those two call for opposite
    /// reactions.
    ///
    /// Absent on daemons predating the windowed budget, where the count really
    /// was a lifetime total.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_window_secs: Option<u64>,
    /// Effective drain budget for this module, in milliseconds. This is the
    /// resolved policy the running supervisor uses, not a config-file reread.
    /// Absent on older daemons.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub drain_timeout_ms: Option<u64>,
    /// Effective base delay before a crash restart, in milliseconds. Absent on
    /// older daemons.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_backoff_ms: Option<u64>,
    /// Effective maximum delay before a crash restart, in milliseconds. Absent
    /// on older daemons.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_max_backoff_ms: Option<u64>,
    /// Memory and cumulative CPU time of the module's process, read when this
    /// list was answered. Report only: the daemon keeps no history and acts on
    /// none of it.
    ///
    /// It describes the one process the supervisor spawned (its `pid`), not
    /// processes that one has started in turn, so a module that forks workers
    /// reports only its own share.
    ///
    /// Absent means the daemon predates the field. A daemon that has the field
    /// but could not read the process (not running, unsupported platform, read
    /// failed) says so with `Unavailable` and a reason, so neither case can be
    /// mistaken for a process using nothing.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resources: Option<ChildResourceUsage>,
}

impl SupervisorEntry {
    /// Construct an entry with its required fields and wire-compatible defaults.
    #[must_use]
    pub fn new(
        module_id: impl Into<String>,
        state: impl Into<String>,
        enabled: bool,
        live: bool,
        health: SupervisorHealthStatus,
    ) -> Self {
        Self {
            module_id: module_id.into(),
            state: state.into(),
            waiting_for_previous_pid: None,
            enabled,
            live,
            protocol: ModuleProtocol::default(),
            launch_nonce_env: None,
            launch_nonce_source: None,
            health,
            pending_reload: None,
            last_probe_ms: None,
            last_exit_code: None,
            last_exit_signal: None,
            last_exit_ms: None,
            last_exit_kind: None,
            restart_count: None,
            max_restarts: None,
            lifetime_restarts: None,
            spawn_generation: None,
            restart_window_secs: None,
            drain_timeout_ms: None,
            restart_backoff_ms: None,
            restart_max_backoff_ms: None,
            resources: None,
        }
    }

    /// Set the module's declared wire protocol.
    #[must_use]
    pub fn with_protocol(mut self, protocol: ModuleProtocol) -> Self {
        self.protocol = protocol;
        self
    }

    /// Set or clear the launch nonce environment indicator.
    #[must_use]
    pub fn with_launch_nonce_env(mut self, launch_nonce_env: Option<bool>) -> Self {
        self.launch_nonce_env = launch_nonce_env;
        self
    }

    pub fn with_launch_nonce_source(mut self, source: Option<String>) -> Self {
        self.launch_nonce_source = source;
        self
    }

    #[must_use]
    pub fn with_waiting_for_previous_pid(mut self, pid: Option<u32>) -> Self {
        self.waiting_for_previous_pid = pid;
        self
    }

    /// Set or clear the comparison between the launch spec and observed process.
    #[must_use]
    pub fn with_pending_reload(mut self, pending_reload: Option<PendingReloadVerdict>) -> Self {
        self.pending_reload = pending_reload;
        self
    }

    /// Set or clear the last health collection timestamp.
    #[must_use]
    pub fn with_last_probe_ms(mut self, last_probe_ms: Option<u64>) -> Self {
        self.last_probe_ms = last_probe_ms;
        self
    }

    /// Set or clear the most recent process exit code.
    #[must_use]
    pub fn with_last_exit_code(mut self, last_exit_code: Option<i32>) -> Self {
        self.last_exit_code = last_exit_code;
        self
    }

    /// Set or clear the most recent process exit signal.
    #[must_use]
    pub fn with_last_exit_signal(mut self, last_exit_signal: Option<i32>) -> Self {
        self.last_exit_signal = last_exit_signal;
        self
    }

    /// Set or clear the most recent process exit timestamp.
    #[must_use]
    pub fn with_last_exit_ms(mut self, last_exit_ms: Option<u64>) -> Self {
        self.last_exit_ms = last_exit_ms;
        self
    }

    /// Set or clear the most recent process exit classification.
    #[must_use]
    pub fn with_last_exit_kind(mut self, last_exit_kind: Option<TerminalExitKind>) -> Self {
        self.last_exit_kind = last_exit_kind;
        self
    }

    /// Set or clear the restart count for the current crash budget.
    #[must_use]
    pub fn with_restart_count(mut self, restart_count: Option<u32>) -> Self {
        self.restart_count = restart_count;
        self
    }

    /// Set or clear the maximum restarts allowed by the crash budget.
    #[must_use]
    pub fn with_max_restarts(mut self, max_restarts: Option<u32>) -> Self {
        self.max_restarts = max_restarts;
        self
    }

    /// Set or clear the lifetime restart count.
    #[must_use]
    pub fn with_lifetime_restarts(mut self, lifetime_restarts: Option<u32>) -> Self {
        self.lifetime_restarts = lifetime_restarts;
        self
    }

    /// Set or clear the successful spawn count for this daemon incarnation.
    #[must_use]
    pub fn with_spawn_generation(mut self, spawn_generation: Option<u64>) -> Self {
        self.spawn_generation = spawn_generation;
        self
    }

    /// Set or clear the time window of the crash budget.
    #[must_use]
    pub fn with_restart_window_secs(mut self, restart_window_secs: Option<u64>) -> Self {
        self.restart_window_secs = restart_window_secs;
        self
    }

    /// Set or clear the effective drain budget in milliseconds.
    #[must_use]
    pub fn with_drain_timeout_ms(mut self, drain_timeout_ms: Option<u64>) -> Self {
        self.drain_timeout_ms = drain_timeout_ms;
        self
    }

    /// Set or clear the effective base restart delay in milliseconds.
    #[must_use]
    pub fn with_restart_backoff_ms(mut self, restart_backoff_ms: Option<u64>) -> Self {
        self.restart_backoff_ms = restart_backoff_ms;
        self
    }

    /// Set or clear the effective maximum restart delay in milliseconds.
    #[must_use]
    pub fn with_restart_max_backoff_ms(mut self, restart_max_backoff_ms: Option<u64>) -> Self {
        self.restart_max_backoff_ms = restart_max_backoff_ms;
        self
    }

    /// Set or clear the observed resource usage of the supervised process.
    #[must_use]
    pub fn with_resources(mut self, resources: Option<ChildResourceUsage>) -> Self {
        self.resources = resources;
        self
    }
}

/// A module process's memory and CPU time as read at list time, or why none
/// could be read.
#[derive(Debug, Clone, PartialEq)]
pub enum ChildResourceUsage {
    Measured(ChildResourceReading),
    Unavailable {
        reason: ChildResourceUnavailableReason,
    },
    /// Future discriminator. `body` retains the complete ordered object; `tag`
    /// is its decoded discriminator projection.
    Unknown {
        tag: String,
        body: OrderedJsonObject,
    },
}

/// One reading of a module process's memory and CPU time.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ChildResourceReading {
    /// Memory in bytes, measured as `memory_kind` says. The figures differ
    /// by platform and are not comparable across kinds.
    pub memory_bytes: u64,
    pub memory_kind: ChildMemoryKind,
    /// Bytes swapped out, where the platform reports it per process (Linux).
    /// Absent means not reported, not zero.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub swap_bytes: Option<u64>,
    /// CPU time spent in user mode since the process started, in
    /// milliseconds. Cumulative, not a rate: a percentage needs two readings
    /// and the elapsed time between them.
    pub cpu_user_ms: u64,
    /// CPU time spent in the kernel on the process's behalf since it started,
    /// in milliseconds.
    pub cpu_system_ms: u64,
}

open_string_enum! {
    /// What `ChildResourceReading::memory_bytes` measures.
    ChildMemoryKind {
        /// macOS `phys_footprint`: memory the kernel charges to the process,
        /// the figure jetsam acts on. Unlike resident size it does not count
        /// pages an allocator has already released with `MADV_FREE`.
        PhysFootprint => "phys_footprint",
        /// Linux `VmRSS`: pages resident in RAM, shared file-backed pages
        /// included and swapped-out pages excluded.
        ResidentSet => "resident_set",
        /// Windows `WorkingSetSize`: resident pageable memory, including shared
        /// pages, not Unix RSS or private committed bytes.
        WindowsWorkingSet => "windows_working_set",
    }
}

open_string_enum! {
    /// Why a module process's resources could not be read.
    ChildResourceUnavailableReason {
        /// The module has no running process.
        NotRunning => "not_running",
        /// The daemon's platform has no per-process source.
        UnsupportedPlatform => "unsupported_platform",
        /// The process could not be read, typically because it exited while
        /// the list was being answered.
        Unreadable => "unreadable",
        /// The pid no longer names the process the supervisor spawned, so a
        /// reading would describe some other process.
        ProcessIdentityUnconfirmed => "process_identity_unconfirmed",
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum SupervisorHealthStatus {
    Ok,
    Degraded,
    Failing,
    Unresponsive,
    Unknown,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SupervisorHealthEntry {
    pub module_id: String,
    pub status: SupervisorHealthStatus,
    /// The module's own human-readable note on its state. Absent means the
    /// module said nothing, which is the ordinary shape for a healthy module and
    /// is NOT a claim that nothing is wrong. Never parse it: it is prose the
    /// module may reword freely, and `status` plus `metrics` are the machine
    /// surface.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
    /// The module's own metrics object, relayed opaquely. Absent means the module
    /// published none on this probe — either it reports no metrics at all, or the
    /// probe did not reach it — so absence cannot distinguish "nothing to report"
    /// from "nobody asked". Read `last_probe_ms` to tell those apart.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metrics: Option<serde_json::Value>,
    pub consecutive_failures: u32,
    /// Number of recurring health replies received after their daemon deadline.
    /// Each increment is evidence that the module remained alive despite a miss.
    #[serde(default)]
    pub late_answer_count: u64,
    /// End-to-end latency of the newest late reply, measured from probe start.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_late_answer_latency_ms: Option<u64>,
    /// The escalation the supervisor last took for this module (report, restart,
    /// alert). Absent means NO ACTION HAS EVER BEEN TAKEN, not that the last one
    /// succeeded — a module that has never misbehaved and one whose action record
    /// predates a daemon restart both present as absent.
    #[serde(default)]
    pub last_action: Option<String>,
    /// When `last_action` was taken, as unix milliseconds. Absent exactly when
    /// `last_action` is absent; the pair moves together.
    #[serde(default)]
    pub last_action_ms: Option<u64>,
    /// When the daemon last collected this entry, as unix milliseconds.
    ///
    /// `supervisor.health` answers from the supervisor's STORED record rather
    /// than probing, so every field above describes some moment in the past and
    /// nothing here said which. That matters most right after a restart, where
    /// the surface is used to confirm a deploy: a record collected before the
    /// restart reports the OLD process, reads as a failed deploy, and invites a
    /// redeploy of something that was already correct.
    ///
    /// `None` means never probed — distinct from probed-long-ago, and the reader
    /// must not collapse them. Absent on modules that advertise no health
    /// capability, which is why it is optional rather than defaulted to zero.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_probe_ms: Option<u64>,
}

#[cfg(test)]
mod tests {
    #[test]
    fn windows_running_image_evidence_round_trips_all_sixteen_id_bytes() {
        let evidence = super::RunningImageEvidence::WindowsSpawnFileId {
            volume_serial_number: u64::MAX,
            file_id: [
                0, 1, 2, 3, 4, 5, 6, 7, 128, 129, 130, 131, 252, 253, 254, 255,
            ],
        };
        let agreement = super::RunningImageAgreement::Match { evidence };
        let encoded = serde_json::to_string(&agreement).unwrap();
        assert!(encoded.contains("windows_spawn_file_id"));
        assert_eq!(
            serde_json::from_str::<super::RunningImageAgreement>(&encoded).unwrap(),
            agreement
        );
    }

    #[test]
    fn unix_running_image_evidence_keeps_original_wire_bytes() {
        assert_eq!(
            serde_json::to_string(&super::RunningImageEvidence::MacosSpawnInode {
                device: 7,
                inode: 11
            })
            .unwrap(),
            "{\"method\":\"macos_spawn_inode\",\"device\":7,\"inode\":11}"
        );
        assert_eq!(
            serde_json::to_string(&super::RunningImageEvidence::LinuxProcSha256 {
                digest: "abc".to_owned()
            })
            .unwrap(),
            "{\"method\":\"linux_proc_sha256\",\"digest\":\"abc\"}"
        );
    }
    use super::*;
    use subc_protocol::{BindIdentity, RouteTarget};

    #[test]
    fn legacy_terminal_decoder_ignores_deliberate_severance_kind() {
        let entry = TerminalEntry {
            daemon_incarnation: Some("daemon-before-restart".into()),
            exit_code: Some(1),
            exit_signal: None,
            at_ms: 1_700_000_000_123,
            disposition: TerminalDisposition::Restarting,
            exit_kind: Some(TerminalExitKind::DeliberateSeverance),
            disposition_detail: None,
        };
        let wire = serde_json::to_string(&entry).expect("terminal entry serializes");
        assert_eq!(
            serde_json::from_str::<serde_json::Value>(&wire).expect("terminal entry is JSON")
                ["exit_kind"],
            "deliberate_severance"
        );

        #[derive(serde::Deserialize)]
        struct LegacyTerminalEntry {
            exit_code: Option<i32>,
            exit_signal: Option<i32>,
            at_ms: u64,
            disposition: TerminalDisposition,
        }

        let decoded: LegacyTerminalEntry =
            serde_json::from_str(&wire).expect("legacy decoder keeps the terminal record");
        assert_eq!(decoded.exit_code, Some(1));
        assert_eq!(decoded.exit_signal, None);
        assert_eq!(decoded.at_ms, 1_700_000_000_123);
        assert_eq!(decoded.disposition, TerminalDisposition::Restarting);

        let future_wire = wire.replace("deliberate_severance", "future_exit_kind");
        let future: TerminalEntry =
            serde_json::from_str(&future_wire).expect("new decoder keeps a future terminal kind");
        assert_eq!(
            future.exit_kind,
            Some(TerminalExitKind::Unknown("future_exit_kind".to_string()))
        );
    }

    #[test]
    fn terminal_incarnation_is_optional_for_older_daemons() {
        let entry: TerminalEntry = serde_json::from_value(serde_json::json!({
            "at_ms": 123,
            "disposition": "stopped"
        }))
        .unwrap();
        let encoded = serde_json::to_value(&entry).unwrap();
        assert_eq!(
            (entry.daemon_incarnation, encoded.get("daemon_incarnation")),
            (None, None)
        );
    }

    #[test]
    fn route_poll_uses_kind_field() {
        let body = serde_json::to_value(ClientControlRequest::RoutePoll {
            route_channel: 7,
            route_epoch: 11,
            kind: PollKind::Status,
        })
        .unwrap();

        assert_eq!(body["op"], "route.poll");
        assert_eq!(body["route_epoch"], 11);
        assert_eq!(body["kind"], "status");
        assert!(body.get("op").is_some());
    }

    #[test]
    fn route_open_is_internally_tagged() {
        let request = ClientControlRequest::RouteOpen {
            target: RouteTarget::ToolProvider {
                module_id: "aft".to_string(),
            },
            identity: BindIdentity::new("/tmp/project", "opencode", "session-1"),
            consumer_identity: None,
            consumer_capabilities: None,
            role_versions: None,
            admission_facts: None,
            scope: None,
        };

        let body = serde_json::to_value(request).unwrap();
        assert_eq!(body["op"], "route.open");
        assert_eq!(body["target"]["kind"], "tool_provider");
        assert!(body.get("consumer_identity").is_none());
        assert!(body.get("consumer_capabilities").is_none());
        assert!(body.get("role_versions").is_none());
    }

    #[test]
    fn route_open_without_optional_fields_still_decodes() {
        let body = serde_json::json!({
            "op": "route.open",
            "target": { "kind": "tool_provider", "module_id": "aft" },
            "identity": {
                "project_root": "/tmp/project",
                "harness": "opencode",
                "session": "session-1"
            }
        });

        let decoded: ClientControlRequest = serde_json::from_value(body).unwrap();
        let ClientControlRequest::RouteOpen {
            consumer_identity,
            consumer_capabilities,
            role_versions,
            admission_facts,
            ..
        } = decoded
        else {
            panic!("decoded wrong request variant");
        };
        assert_eq!(consumer_identity, None);
        assert_eq!(consumer_capabilities, None);
        assert_eq!(role_versions, None);
        assert_eq!(admission_facts, None);
    }

    #[test]
    fn new_route_closed_decoder_defaults_fields_absent_from_old_daemon() {
        let old_wire = r#"{"op":"route.closed","module_id":"aft-tools","reason":"crash","drained":false,"abandoned":0}"#;
        let decoded: ClientControlPush = serde_json::from_str(old_wire).unwrap();
        match decoded {
            ClientControlPush::RouteClosed {
                excluded_subscriptions,
                terminal,
                ..
            } => {
                assert_eq!(excluded_subscriptions, 0);
                assert_eq!(terminal, None);
            }
            other => panic!("unexpected push: {other:?}"),
        }
        assert!(!serde_json::to_string(&decoded)
            .unwrap()
            .contains("terminal"));
    }

    #[test]
    fn old_route_closed_decoder_ignores_new_terminal_field() {
        #[derive(serde::Deserialize)]
        #[serde(tag = "op")]
        enum LegacyClientControlPush {
            #[serde(rename = "route.closed")]
            RouteClosed {
                module_id: String,
                reason: RouteCloseReason,
                drained: bool,
                abandoned: u32,
            },
        }

        let wire = r#"{"op":"route.closed","module_id":"aft-tools","reason":"crash","drained":false,"abandoned":0,"excluded_subscriptions":3,"terminal":true}"#;
        let decoded: LegacyClientControlPush = serde_json::from_str(wire).unwrap();
        match decoded {
            LegacyClientControlPush::RouteClosed {
                module_id,
                reason,
                drained,
                abandoned,
            } => {
                assert_eq!(module_id, "aft-tools");
                assert_eq!(reason, RouteCloseReason::Crash);
                assert!(!drained);
                assert_eq!(abandoned, 0);
            }
        }
    }

    #[test]
    fn supervisor_routes_is_a_control_plane_request() {
        let body = serde_json::json!({
            "op": "supervisor.routes",
            "module_id": "aft"
        });

        let request: ClientControlRequest = serde_json::from_value(body.clone()).unwrap();
        assert_eq!(serde_json::to_value(request).unwrap(), body);
    }

    #[test]
    fn diagnostic_string_enums_retain_unknown_wire_values() {
        let reason: RunningImageUnavailableReason =
            serde_json::from_str("\"future_reason\"").unwrap();
        let disposition: TerminalDisposition =
            serde_json::from_str("\"future_disposition\"").unwrap();

        assert_eq!(
            reason,
            RunningImageUnavailableReason::Unknown("future_reason".to_string())
        );
        assert_eq!(
            disposition,
            TerminalDisposition::Unknown("future_disposition".to_string())
        );
    }

    #[test]
    fn diagnostic_string_enums_preserve_existing_wire_names() {
        let names = [
            (RunningImageUnavailableReason::NotRunning, "not_running"),
            (
                RunningImageUnavailableReason::UnsupportedPlatform,
                "unsupported_platform",
            ),
            (
                RunningImageUnavailableReason::RunningExecutableUnreadable,
                "running_executable_unreadable",
            ),
            (
                RunningImageUnavailableReason::RunningExecutablePermissionDenied,
                "running_executable_permission_denied",
            ),
            (
                RunningImageUnavailableReason::ProcessHidden,
                "process_hidden",
            ),
            (
                RunningImageUnavailableReason::SpawnedPathUnreadable,
                "spawned_path_unreadable",
            ),
            (RunningImageUnavailableReason::HashFailed, "hash_failed"),
            (
                RunningImageUnavailableReason::ProcessIdentityUnconfirmed,
                "process_identity_unconfirmed",
            ),
        ];
        for (value, expected) in names {
            let wire = serde_json::to_string(&value).unwrap();
            assert_eq!(wire, format!("\"{expected}\""));
            let decoded: RunningImageUnavailableReason = serde_json::from_str(&wire).unwrap();
            assert_eq!(decoded, value);
        }

        for (value, expected) in [
            (TerminalDisposition::Stopped, "stopped"),
            (TerminalDisposition::Disabled, "disabled"),
            (TerminalDisposition::Failed, "failed"),
            (TerminalDisposition::Restarting, "restarting"),
            (TerminalDisposition::DaemonShutdown, "daemon_shutdown"),
        ] {
            let wire = serde_json::to_string(&value).unwrap();
            assert_eq!(wire, format!("\"{expected}\""));
            let decoded: TerminalDisposition = serde_json::from_str(&wire).unwrap();
            assert_eq!(decoded, value);
        }
    }

    #[test]
    fn supervisor_waiting_previous_pid_is_optional_and_round_trips() {
        let entry = SupervisorEntry::new(
            "waiting-module",
            "starting",
            true,
            false,
            SupervisorHealthStatus::Unknown,
        );
        let old_wire = serde_json::to_value(&entry).unwrap();
        assert!(old_wire.get("waiting_for_previous_pid").is_none());
        let decoded: SupervisorEntry = serde_json::from_value(old_wire).unwrap();
        assert_eq!(decoded.waiting_for_previous_pid, None);
        let entry = entry.with_waiting_for_previous_pid(Some(42));
        let wire = serde_json::to_value(&entry).unwrap();
        assert_eq!(wire["waiting_for_previous_pid"], 42);
        let decoded: SupervisorEntry = serde_json::from_value(wire).unwrap();
        assert_eq!(decoded.waiting_for_previous_pid, Some(42));
    }

    #[test]
    fn diagnostic_string_enums_reject_non_string_bodies() {
        assert!(serde_json::from_str::<RunningImageUnavailableReason>("42").is_err());
        assert!(serde_json::from_str::<TerminalDisposition>("{\"value\":\"failed\"}").is_err());
    }

    #[test]
    fn unknown_provenance_reason_does_not_discard_healthy_siblings() {
        let body = serde_json::json!({
            "op": "supervisor.provenance",
            "daemon": {
                "daemon_build": {},
                "daemon_observed": {
                    "running_image": {
                        "status": "unavailable",
                        "reason": "not_running"
                    }
                }
            },
            "modules": [
                {
                    "module_id": "future",
                    "module_declared": { "status": "unverifiable" },
                    "daemon_observed": {
                        "running_image": {
                            "status": "unavailable",
                            "reason": "future_reason"
                        }
                    }
                },
                {
                    "module_id": "healthy-a",
                    "module_declared": { "status": "unverifiable" },
                    "daemon_observed": {
                        "running_image": {
                            "status": "match",
                            "evidence": {
                                "method": "linux_proc_sha256",
                                "digest": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                            }
                        }
                    }
                },
                {
                    "module_id": "healthy-b",
                    "module_declared": { "status": "unverifiable" },
                    "daemon_observed": {
                        "running_image": {
                            "status": "unavailable",
                            "reason": "unsupported_platform"
                        }
                    }
                }
            ]
        });

        let decoded: ClientControlResponse = serde_json::from_value(body).unwrap();
        let ClientControlResponse::SupervisorProvenance { modules, .. } = decoded else {
            panic!("decoded wrong response variant");
        };
        assert_eq!(modules.len(), 3);
        assert_eq!(modules[0].module_id, "future");
        assert_eq!(
            modules[0].daemon_observed.running_image,
            RunningImageAgreement::Unavailable {
                reason: RunningImageUnavailableReason::Unknown("future_reason".to_string())
            }
        );
        assert_eq!(modules[1].module_id, "healthy-a");
        assert_eq!(modules[2].module_id, "healthy-b");
    }

    #[test]
    fn tagged_unknown_values_retain_tag_and_body() {
        macro_rules! assert_unknown_round_trip {
            ($ty:ident, $field:literal, $value:expr) => {
                let value = $value;
                let wire = serde_json::to_string(&value).unwrap();
                let decoded: $ty = serde_json::from_str(&wire).unwrap();
                match decoded {
                    $ty::Unknown { tag, body } => {
                        assert_eq!(tag, value[$field].as_str().unwrap());
                        assert_eq!(serde_json::to_value(&body).unwrap(), value);
                    }
                    _ => panic!("decoded known variant"),
                }
            };
        }

        assert_unknown_round_trip!(
            ModuleDeclaredProvenance,
            "status",
            serde_json::json!({"status": "future", "build": {"version": 7}})
        );
        assert_unknown_round_trip!(
            RunningImageAgreement,
            "status",
            serde_json::json!({"status": "future", "evidence": {"digest": "abc"}})
        );
        assert_unknown_round_trip!(
            RunningImageEvidence,
            "method",
            serde_json::json!({"method": "future", "digest": "abc"})
        );
        assert_unknown_round_trip!(
            SupervisorRouteConsumer,
            "kind",
            serde_json::json!({"kind": "future", "module_id": "m"})
        );
        assert_unknown_round_trip!(
            StderrCaptureState,
            "state",
            serde_json::json!({"state": "future", "reason": "because"})
        );
        assert_unknown_round_trip!(
            StderrTailEntry,
            "kind",
            serde_json::json!({"kind": "future", "text": "line"})
        );
        assert_unknown_round_trip!(
            ChildResourceUsage,
            "status",
            serde_json::json!({"status": "future", "memory_bytes": 1})
        );
    }

    #[test]
    fn child_resource_usage_round_trips_both_known_states() {
        let measured = ChildResourceUsage::Measured(ChildResourceReading {
            memory_bytes: 0,
            memory_kind: ChildMemoryKind::ResidentSet,
            swap_bytes: Some(0),
            cpu_user_ms: 0,
            cpu_system_ms: 0,
        });
        let wire = serde_json::to_value(&measured).unwrap();
        assert_eq!(
            wire,
            serde_json::json!({
                "status": "measured",
                "memory_bytes": 0,
                "memory_kind": "resident_set",
                "swap_bytes": 0,
                "cpu_user_ms": 0,
                "cpu_system_ms": 0
            })
        );
        assert_eq!(
            serde_json::from_value::<ChildResourceUsage>(wire).unwrap(),
            measured
        );

        let unavailable = ChildResourceUsage::Unavailable {
            reason: ChildResourceUnavailableReason::NotRunning,
        };
        let wire = serde_json::to_value(&unavailable).unwrap();
        assert_eq!(
            wire,
            serde_json::json!({"status": "unavailable", "reason": "not_running"})
        );
        assert_eq!(
            serde_json::from_value::<ChildResourceUsage>(wire).unwrap(),
            unavailable
        );
    }

    #[test]
    fn a_stderr_line_decodes_with_and_without_its_capture_time() {
        // A current daemon stamps each line; an older one sends no `at_ms`.
        // Both must decode, and the absent case must stay absent rather than
        // turn into a time nobody recorded.
        let stamped: StderrTailEntry = serde_json::from_str(
            r#"{"kind":"line","text":"boom","truncated":true,"at_ms":1789801440685}"#,
        )
        .unwrap();
        assert_eq!(
            stamped,
            StderrTailEntry::Line {
                text: "boom".to_string(),
                truncated: true,
                at_ms: Some(1_789_801_440_685),
            }
        );
        let unstamped: StderrTailEntry =
            serde_json::from_str(r#"{"kind":"line","text":"boom"}"#).unwrap();
        assert_eq!(
            unstamped,
            StderrTailEntry::Line {
                text: "boom".to_string(),
                truncated: false,
                at_ms: None,
            }
        );
        // Absent stays absent on the way out, so a reply without stamps is
        // exactly what an older daemon would have sent.
        assert_eq!(
            serde_json::to_string(&unstamped).unwrap(),
            r#"{"kind":"line","text":"boom"}"#
        );
        assert_eq!(
            serde_json::to_value(&stamped).unwrap()["at_ms"],
            serde_json::json!(1_789_801_440_685u64)
        );
    }

    #[test]
    fn tagged_unknown_values_round_trip_the_original_json() {
        let wire = r#"{"kind":"future_consumer","detail":{"z":1}}"#;
        let decoded: SupervisorRouteConsumer = serde_json::from_str(wire).unwrap();
        assert_eq!(serde_json::to_string(&decoded).unwrap(), wire);
    }

    #[test]
    fn tagged_unknown_values_round_trip_trailing_tag() {
        let route_wire = r#"{"detail":{"z":1},"kind":"future_consumer"}"#;
        let route: SupervisorRouteConsumer = serde_json::from_str(route_wire).unwrap();
        assert_eq!(serde_json::to_string(&route).unwrap(), route_wire);

        let stderr_wire = r#"{"reason":"because","state":"future_state"}"#;
        let stderr: StderrCaptureState = serde_json::from_str(stderr_wire).unwrap();
        assert_eq!(serde_json::to_string(&stderr).unwrap(), stderr_wire);
    }

    #[test]
    fn tagged_unknown_values_round_trip_middle_tag() {
        let route_wire = r#"{"a":1,"kind":"future_x","b":2}"#;
        let route: SupervisorRouteConsumer = serde_json::from_str(route_wire).unwrap();
        assert_eq!(serde_json::to_string(&route).unwrap(), route_wire);

        let stderr_wire = r#"{"a":1,"state":"future_state","b":2}"#;
        let stderr: StderrCaptureState = serde_json::from_str(stderr_wire).unwrap();
        assert_eq!(serde_json::to_string(&stderr).unwrap(), stderr_wire);
    }

    #[test]
    fn tagged_unknown_values_round_trip_deep_payload() {
        let route_wire = r#"{"a":{"n":[1,2]},"kind":"future_x","zz":"s","b":null}"#;
        let route: SupervisorRouteConsumer = serde_json::from_str(route_wire).unwrap();
        assert_eq!(serde_json::to_string(&route).unwrap(), route_wire);

        let stderr_wire = r#"{"a":{"n":[1,2]},"state":"future_state","zz":"s","b":null}"#;
        let stderr: StderrCaptureState = serde_json::from_str(stderr_wire).unwrap();
        assert_eq!(serde_json::to_string(&stderr).unwrap(), stderr_wire);
    }

    #[test]
    fn tagged_unknown_values_reject_non_object_bodies() {
        for wire in ["42", r#""future""#, "[]"] {
            assert!(serde_json::from_str::<SupervisorRouteConsumer>(wire).is_err());
            assert!(serde_json::from_str::<StderrCaptureState>(wire).is_err());
        }
    }

    #[test]
    fn duplicate_discriminators_reject_without_panicking() {
        assert_eq!(
            serde_json::from_str::<ModuleDeclaredProvenance>(r#"{"status":"unverifiable"}"#)
                .unwrap(),
            ModuleDeclaredProvenance::Unverifiable
        );
        match serde_json::from_str::<ModuleDeclaredProvenance>(r#"{"status":"future_thing"}"#)
            .unwrap()
        {
            ModuleDeclaredProvenance::Unknown { tag, .. } => assert_eq!(tag, "future_thing"),
            _ => panic!("future discriminator decoded as a known variant"),
        }

        let wires = [
            r#"{"status":"reported","status":"unverifiable"}"#,
            r#"{"status":"unverifiable","status":"reported"}"#,
            r#"{"status":"reported","build":{},"status":"unverifiable"}"#,
            r#"{"status":"unverifiable","build":{},"status":"reported"}"#,
        ];

        for wire in wires {
            let result =
                std::panic::catch_unwind(|| serde_json::from_str::<ModuleDeclaredProvenance>(wire));
            assert!(result.is_ok(), "duplicate discriminator panicked: {wire}");
            assert!(
                result.unwrap().is_err(),
                "duplicate discriminator decoded: {wire}"
            );
        }

        let wire = r#"{"state":"captured","state":"incomplete","reason":"x"}"#;
        let result = std::panic::catch_unwind(|| serde_json::from_str::<StderrCaptureState>(wire));
        assert!(result.is_ok(), "duplicate discriminator panicked: {wire}");
        assert!(
            result.unwrap().is_err(),
            "duplicate discriminator decoded: {wire}"
        );
    }

    #[test]
    fn nested_unknown_values_round_trip_without_normalizing_member_order() {
        let known_wire =
            r#"{"status":"match","evidence":{"method":"linux_proc_sha256","digest":"abc"}}"#;
        let known: RunningImageAgreement = serde_json::from_str(known_wire).unwrap();
        assert_eq!(serde_json::to_string(&known).unwrap(), known_wire);

        for wire in [
            r#"{"kind":"future_x","detail":{"zeta":1,"alpha":2}}"#,
            r#"{"kind":"future_x","d":{"b":{"zz":1,"aa":2}}}"#,
        ] {
            let decoded: SupervisorRouteConsumer = serde_json::from_str(wire).unwrap();
            assert_eq!(serde_json::to_string(&decoded).unwrap(), wire);
        }

        for wire in [
            r#"{"status":"match","evidence":{"method":"future_probe","zz":1,"aa":2}}"#,
            r#"{"status":"match","evidence":{"method":"future_probe","d":{"zz":1,"aa":2}}}"#,
        ] {
            let decoded: RunningImageAgreement = serde_json::from_str(wire).unwrap();
            assert_eq!(serde_json::to_string(&decoded).unwrap(), wire);
        }

        let wire = r#"{"status":"mismatch","running":{"detail":{"z":1},"method":"future_running"},"disk":{"method":"future_disk","detail":{"z":1}}}"#;
        let decoded: RunningImageAgreement = serde_json::from_str(wire).unwrap();
        assert_eq!(serde_json::to_string(&decoded).unwrap(), wire);

        let wire = r#"{"capture":{"state":"captured"},"entries":[{"detail":{"z":1,"a":2},"kind":"future_line"},{"kind":"future_restart","meta":{"b":{"zz":1,"aa":2}}}]}"#;
        let decoded: StderrTail = serde_json::from_str(wire).unwrap();
        assert_eq!(serde_json::to_string(&decoded).unwrap(), wire);
    }

    #[test]
    fn tagged_unknown_member_does_not_discard_known_siblings() {
        let body = serde_json::json!({
            "modules": [{
                "module_id": "target",
                "routes": [
                    {"consumer": {"kind": "future_consumer", "module_id": "m", "detail": {"retry": true}}, "age_ms": 0, "draining": false},
                    {"consumer": {"kind": "direct", "connection_id": 7}, "age_ms": 0, "draining": false}
                ]
            }]
        });
        let decoded: ClientControlResponse = serde_json::from_value(
            serde_json::json!({"op": "supervisor.routes", "modules": body["modules"]}),
        )
        .unwrap();
        let ClientControlResponse::SupervisorRoutes { modules } = decoded else {
            panic!("decoded wrong response variant");
        };
        assert_eq!(modules[0].routes.len(), 2);
        assert_eq!(
            modules[0].routes[1].consumer,
            SupervisorRouteConsumer::Direct { connection_id: 7 }
        );
    }
}

#[cfg(test)]
mod launch_nonce_redaction_tests {
    use super::*;

    const NONCE: &str = "nonce-f00dfeed1234abcd";

    fn identity() -> ConsumerIdentity {
        ConsumerIdentity {
            module_id: "wernicke".to_string(),
            launch_nonce: NONCE.to_string(),
        }
    }

    #[test]
    fn consumer_identity_debug_names_the_module_and_never_the_nonce() {
        let printed = format!("{:?}", identity());
        assert!(printed.contains("wernicke"), "{printed}");
        assert!(!printed.contains(NONCE), "launch nonce printed: {printed}");
    }

    #[test]
    fn route_open_request_debug_never_prints_the_nonce() {
        let request = ClientControlRequest::RouteOpen {
            target: subc_protocol::RouteTarget::ToolProvider {
                module_id: "broca".to_string(),
            },
            identity: subc_protocol::BindIdentity::new(
                PathBuf::from("/tmp/project"),
                "test".to_string(),
                "session".to_string(),
            ),
            consumer_identity: Some(identity()),
            consumer_capabilities: None,
            role_versions: None,
            admission_facts: None,
            scope: None,
        };
        let printed = format!("{request:?}");
        assert!(!printed.contains(NONCE), "launch nonce printed: {printed}");
    }
}