liminal_server/server/connection/

services.rs

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, RwLock, mpsc};
use std::time::Instant;

use haematite::{CommitPolicy, Database, DatabaseConfig, EventStore, Retention};
use liminal::channel::{
    ChannelConfig, ChannelDelivery, ChannelHandle, ChannelMode, ChannelSupervisor, Schema,
};
use liminal::conversation::{
    ConversationSupervisor, CrashPolicy, EchoBehaviour, ParticipantBehaviour,
};
use liminal::durability::bridge::block_on;
#[cfg(any(test, feature = "test-support"))]
use liminal::durability::open_ephemeral;
use liminal::durability::{
    CountingStore, DedupCache, DedupDecision, DurableStore, EphemeralRoot, EphemeralSweepCounts,
    HaematiteStore, ProcessingReceipt, StoreCountsHandle,
};
use liminal::protocol::{MessageEnvelope, ProtocolError, SchemaId as ProtocolSchemaId};
use std::num::NonZeroU64;

use super::channel_registry::{
    ChannelAccessError, ChannelBuildError, ChannelConfigField, ChannelDescriptor, ChannelOrigin,
    ChannelRegistration, ChannelRegistryError, ChannelState, ChannelStatus, MAX_CHANNELS_KEY,
    Registered, STATE_ACTIVE, STATE_QUIESCED, UNRECORDED_QUIESCE_REASON,
};
use super::conversation::{ConnectionConversation, LiminalConversationResource};
use super::services_cluster::build_channel_cluster;
use super::services_schema::{ChannelSchema, resolve_channel_schema, resolve_schema_bytes};
use super::worker_front_door::WorkerFrontDoorServices;
use crate::ServerError;
use crate::config::types::{ClusterConfig, ParticipantConfig, ServerConfig, ServiceProfile};
use crate::config::validation::required_persistence_path;
use crate::health::reissue::OperatorCredentialReissuer;
use crate::health::unloadable::UnloadableConversationRecord;
use crate::server::participant::{
    InstalledParticipantService, ParticipantWarm, ProductionParticipantHandler,
};
use crate::server::self_streams::events::PublishOrigin;
use crate::server::self_streams::sink::OriginPublication;
use crate::server::signals::ServerSignals;

pub use super::services_cluster::ChannelCluster;

/// Registry of custom conversation responders, keyed by conversation subject.
///
/// A registered [`ParticipantBehaviour`] becomes the participant for any
/// conversation opened on its subject; subjects with no entry fall back to the
/// built-in [`EchoBehaviour`].
type ResponderRegistry = HashMap<String, Arc<dyn ParticipantBehaviour>>;

/// Marker for resources retained by a connection process until unsubscribe.
pub trait SubscriptionResource: std::fmt::Debug + Send {
    /// Releases the library subscription resource.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal library reports an unsubscribe failure.
    fn unsubscribe(self: Box<Self>) -> Result<(), ServerError>;

    /// Attempts to pull the next delivered envelope from the wrapped library
    /// subscription without blocking.
    ///
    /// Returns `None` when the subscriber inbox is empty (or momentarily
    /// unavailable): the connection process is the delivery pump, so a transient
    /// empty read is simply "nothing to deliver this slice", never an error.
    fn try_next(&mut self) -> Option<liminal::envelope::Envelope>;

    /// Non-consuming availability query for the post-arm race barrier.
    fn has_pending(&self) -> bool;

    /// Whether an overflow has marked this subscription for shedding (§5). The
    /// delivery pump sheds an overflowed subscription with a typed error frame.
    /// Defaulted to `false`: a resource with no bounded inbox never overflows.
    fn is_overflowed(&self) -> bool {
        false
    }

    /// The library subscriber's pid, the id a recording publish names it by.
    /// Defaulted to `None`: a resource with no library subscriber is never a
    /// recipient.
    fn subscriber_pid(&self) -> Option<u64> {
        None
    }
}

/// Library subscription resource owned by a single connection process.
#[derive(Debug)]
pub struct ConnectionSubscription {
    id: u64,
    /// Channel this subscription was opened on, carried for DIAGNOSTICS only —
    /// the shed warning has to name the channel or an operator reading it learns
    /// that something was shed without learning what. Empty when unset (the
    /// scheduler-free unit fixtures, which construct through `new` and never log).
    channel: String,
    /// Client-chosen application stream the server delivers this subscription's
    /// messages on (echoed on `SubscribeAck`, carried on every `Deliver`). Set by
    /// the connection process from the `Subscribe` frame before the subscription is
    /// stored; `0` only while momentarily unset during construction.
    stream_id: u32,
    selected_schema: ProtocolSchemaId,
    resource: Box<dyn SubscriptionResource>,
    issued: super::state::IssuedDeliveries,
}

impl ConnectionSubscription {
    /// Creates an owned subscription resource for one connection process.
    #[must_use]
    pub fn new(
        id: u64,
        selected_schema: ProtocolSchemaId,
        resource: Box<dyn SubscriptionResource>,
    ) -> Self {
        Self {
            id,
            channel: String::new(),
            stream_id: 0,
            selected_schema,
            resource,
            issued: super::state::IssuedDeliveries::new(0),
        }
    }

    /// Records the channel this subscription was opened on (diagnostics).
    ///
    /// Additive setter rather than a fourth `new` parameter: `new` is public API
    /// and its existing callers keep compiling byte-for-byte.
    pub(super) fn set_channel(&mut self, channel: String) {
        self.channel = channel;
    }

    /// The channel this subscription was opened on, or `""` when unset.
    #[must_use]
    pub(super) fn channel(&self) -> &str {
        &self.channel
    }

    /// Returns the protocol subscription id.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }

    /// Records the client-chosen delivery stream id for this subscription.
    pub(super) const fn set_stream_id(&mut self, stream_id: u32) {
        self.stream_id = stream_id;
    }

    /// Returns the client-chosen application stream id deliveries ride on.
    #[must_use]
    pub(super) const fn stream_id(&self) -> u32 {
        self.stream_id
    }

    pub(super) fn set_pressure_ledger_depth(&mut self, limit: usize) {
        self.issued = super::state::IssuedDeliveries::new(limit);
    }

    pub(super) fn issued_delivery(&mut self, sequence: u64) {
        self.issued.record(sequence);
    }

    pub(super) fn match_pressure(&mut self, reference: &str, retain: bool) -> bool {
        self.issued.matches(reference, retain)
    }

    /// Returns the schema selected for this subscription stream.
    #[must_use]
    pub const fn selected_schema(&self) -> ProtocolSchemaId {
        self.selected_schema
    }

    /// The library subscriber's pid, when the subscription has one.
    #[must_use]
    pub fn subscriber_pid(&self) -> Option<u64> {
        self.resource.subscriber_pid()
    }

    /// Attempts to pull the next delivered envelope without blocking.
    pub(super) fn try_next(&mut self) -> Option<liminal::envelope::Envelope> {
        self.resource.try_next()
    }

    pub(super) fn has_pending(&self) -> bool {
        self.resource.has_pending()
    }

    /// Whether this subscription has been shed by an inbox overflow (§5).
    pub(super) fn is_overflowed(&self) -> bool {
        self.resource.is_overflowed()
    }

    pub(super) fn unsubscribe(self) -> Result<(), ServerError> {
        self.resource.unsubscribe()
    }
}

/// Outcome of a server publish.
///
/// Carries the assigned message id plus a genuine delivery ack (`delivered` = the
/// message was accepted by at least one live subscriber on this publish, after any
/// dedup-on-delivery suppression).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PublishOutcome {
    /// Transport-scoped message id assigned to the accepted publish.
    ///
    /// TRANSPORT-SCOPED, NOT A DURABLE POSITION (G1,
    /// `docs/design/wire-durability-gaps-20260831.md`): this is a
    /// process-local counter (`next_message_id`, reset to 1 on every
    /// construction), so it restarts from 1 with the server and two servers
    /// mint the same ids. It is monotonic only within one service instance's
    /// lifetime and identifies the publish on THIS connection's wire, nothing
    /// more. The durable position of a publish on a durable channel is the
    /// store-assigned sequence the library now surfaces as
    /// `ChannelDelivery::durable_position`; putting that on the wire ack is
    /// the A1 resume lane's frame decision, not a reinterpretation of this
    /// field.
    pub message_id: u64,
    /// Whether the message was genuinely delivered to a subscriber. `false` means
    /// the publish was accepted but reached no subscriber (empty channel) or was
    /// a duplicate suppressed by dedup-on-delivery.
    pub delivered: bool,
    /// Store-assigned position of the row this outcome answers for, when the
    /// publish is known to have one. Unlike `message_id` this IS a durable
    /// position; the connection process puts it on the wire ack only for a
    /// connection that negotiated `PUBLISH_ACK_POSITION_VERSION`.
    pub durable_position: Option<u64>,
    /// Whom the publish reached, the console tap's word for it (SRV-012
    /// R10): a subscriber, no subscriber, only subscribers that refused it,
    /// or nobody because dedup suppressed it.
    pub reach: Reach,
}

/// Whom a publish reached (SRV-012 R10).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Reach {
    /// At least one live subscriber took it.
    Delivered,
    /// The channel had no matching subscriber.
    NoSubscriber,
    /// Every matching subscriber refused it.
    NoneAccepted,
    /// A dedup-suppressed re-publish, sent to nobody.
    Suppressed,
}

impl Reach {
    /// The reach of a library delivery.
    #[must_use]
    pub const fn of(delivery: &ChannelDelivery) -> Self {
        if delivery.is_delivered() {
            Self::Delivered
        } else if delivery.matched_count() == 0 {
            Self::NoSubscriber
        } else {
            Self::NoneAccepted
        }
    }
}

/// Which channel operation is asking the roster for admission.
///
/// `Copy`, so the hot path allocates nothing to ask. The contrast is
/// [`ServerError::UnsupportedOperation`]'s owned `operation: String`: that one is
/// built on a refusal path and pays for its allocation once per refusal, while
/// this one is consulted on every publish and every subscribe frame that reaches
/// the connection process.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChannelOperation {
    /// A `Publish` frame is asking.
    Publish,
    /// A `Subscribe` frame is asking.
    Subscribe,
}

/// Operations that adapt wire frames to liminal library calls.
pub trait ConnectionServices: std::fmt::Debug + Send + Sync {
    /// Returns the complete participant service installed on this adapter.
    ///
    /// `None` keeps participant capability disabled even when the adapter owns a
    /// durable store for unrelated channel traffic. The returned token is
    /// server-sealed and atomically carries declared semantics plus durability,
    /// making a handler-without-store activation impossible by construction.
    fn participant_service(&self) -> Option<InstalledParticipantService> {
        None
    }

    /// Delegates a publish request to the liminal library.
    ///
    /// `connection` is the publishing connection's id (its process pid, as
    /// `apply_frame` knows it), carried as the publish origin so the console
    /// tap attributes the message to it (SRV-012). `idempotency_key`, when `Some`, drives dedup-on-delivery: a re-publish with
    /// the same key is delivered to subscribers at most once. The returned
    /// [`PublishOutcome`] carries the genuine delivery ack.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal publish operation fails.
    fn publish(
        &self,
        connection: u64,
        channel: &str,
        envelope: &MessageEnvelope,
        idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError>;

    /// Delegates a subscribe request to the liminal library.
    ///
    /// `install`, when `Some`, carries the connection's §5 shared inbox byte
    /// budget, per-inbox fairness cap, and R3 wake notifier. The implementation
    /// MUST install it on the subscription's inbox BEFORE the registration is
    /// published to the channel actor (i.e. before any envelope can be
    /// delivered), so no envelope is ever admitted uncharged, past the depth
    /// cap, or without a wake. Implementations with no real inbox (test
    /// stand-ins, capability-scoped profiles that refuse subscribe) may ignore
    /// it.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal subscribe operation fails.
    fn subscribe(
        &self,
        channel: &str,
        accepted_schemas: &[ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError>;

    /// Delegates unsubscribe to the liminal library.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal unsubscribe operation fails.
    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError>;

    /// Delegates conversation open to the liminal library.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal conversation open operation fails.
    fn open_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError>;

    /// Delegates a conversation message to the liminal library.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal conversation message operation fails.
    fn conversation_message(
        &self,
        conversation: &ConnectionConversation,
        envelope: &MessageEnvelope,
        op_id: Option<u64>,
    ) -> Result<(), ServerError>;

    /// Delegates conversation close to the liminal library.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the liminal conversation close operation fails.
    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError>;

    /// Flushes the shared durable store after channel publications have settled.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the shared store flush operation fails.
    fn flush_durable_state(&self) -> Result<(), ServerError>;

    /// Stops the participant warm and records the clean stop's capacity
    /// snapshot (DUR-020 R3, R5). Called once, by the shutdown sequence,
    /// before the final flush; an adapter with no participant does nothing.
    fn stop_participant_warm(&self) {}

    /// The listener is bound: whatever the start reads that no admission
    /// needs before it is judged runs from here, behind the listening door.
    /// An adapter with nothing to read does nothing. Answers the thread it
    /// started, which the listener leaves to run and a test may join.
    fn listening(self: Arc<Self>) -> Option<std::thread::JoinHandle<()>> {
        None
    }

    /// Whether this adapter backs ordinary channel and conversation operations.
    ///
    /// The default is `true` — the full-service adapter serves publish, subscribe,
    /// and conversation frames, so full mode is byte-for-byte unchanged. The
    /// capability-scoped worker front door overrides this to `false`, letting
    /// [`super::apply`] reject the channel/conversation frames it short-circuits on
    /// empty connection state (`Unsubscribe`, `ConversationMessage`,
    /// `ConversationClose`) with a typed error frame instead of silently swallowing
    /// an operation for a resource that could never have been created in this
    /// profile. Frames that always reach a service method (`Publish`, `Subscribe`,
    /// `ConversationOpen`) are rejected by the front door's own method bodies and do
    /// not consult this flag.
    fn supports_channel_operations(&self) -> bool {
        true
    }

    /// Whether `channel` admits `operation` right now.
    ///
    /// Consulted by the connection process BEFORE the operation is delegated, so
    /// a roster refusal is typed at the moment of the decision, by the component
    /// that made it, from the value it decided on. The alternative — classifying
    /// an opaque failure afterwards by re-reading the roster in the error arm —
    /// cannot tell "the roster refused this" from "something else failed while
    /// the roster happened to change", and would put a confident wrong cause on
    /// the wire.
    ///
    /// It does NOT replace the adapter's own inner check. Admission here is the
    /// caller's guard; a service method is public and callable without a frame,
    /// so it keeps its own.
    ///
    /// The default ADMITS. An adapter with no roster has nothing to say here and
    /// its refusals travel as service errors exactly as they do today; only the
    /// roster-owning adapter overrides this. A default body is also what keeps
    /// this addition inside "minor": a method added without one breaks every
    /// downstream implementor of a public trait.
    ///
    /// # Errors
    /// Returns [`ChannelAccessError`] when the roster refuses the operation.
    fn admit_channel(
        &self,
        operation: ChannelOperation,
        channel: &str,
    ) -> Result<(), ChannelAccessError> {
        let _ = (operation, channel);
        Ok(())
    }
}

/// Default adapter from server wire frames to liminal channel/conversation APIs.
#[derive(Debug)]
pub struct LiminalConnectionServices {
    /// The channel roster, keyed by channel name.
    ///
    /// Behind an [`RwLock`] over `Arc` values so a reader clones one pointer out
    /// under the guard and works with it after the guard is released: no roster
    /// read is ever held across a library call, and an entry handed out can
    /// outlive a concurrent mutation of the map itself.
    channels: RwLock<HashMap<String, Arc<ConfiguredChannel>>>,
    /// The operator-declared bound on RUNTIME-registered channels
    /// (`limits.max_channels`), carried verbatim from config.
    ///
    /// `None` is "the operator declared no bound", which refuses every runtime
    /// registration rather than admitting an unbounded one — the roster's
    /// aggregate size is otherwise unbounded the moment a registration API
    /// exists. Boot-configured channels are never counted against it: they are
    /// the bound the operator already wrote, in the file they wrote it in.
    max_channels: Option<usize>,
    cluster: ChannelCluster,
    durable_store: Arc<dyn DurableStore>,
    /// Reads the counters of every store these services built: the durable
    /// store and, when it is a separate store, the ephemeral dedup store.
    store_counts: StoreCountsHandle,
    ephemeral_sweep: Option<EphemeralSweepCounts>,
    signals: ServerSignals,
    /// Complete participant service, installed only when semantic lifecycle
    /// handling and its durable aggregate store are both ready.
    participant_service: Option<InstalledParticipantService>,
    /// The production handler's refused-load record, captured at construction.
    ///
    /// [`InstalledParticipantService`] holds its handler behind
    /// `dyn ParticipantSemanticHandler`, so the concrete production handler is
    /// no longer reachable once it is installed; the record is taken here, in
    /// the one place the concrete handler exists, and carried out to the
    /// startup path that publishes it onto the health endpoint. `None` is "no
    /// participant is configured", which the operator surface reports as such
    /// rather than as an empty refusal set.
    unloadable_conversations: Option<UnloadableConversationRecord>,
    /// The production handler as the A7 operator credential-re-issue authority
    /// (§0.18), captured at construction for exactly the reason above: the
    /// concrete handler is unreachable once erased behind
    /// `dyn ParticipantSemanticHandler`, so the operator surface's handle on it
    /// has to be taken in the one place it still exists. `None` is "no
    /// participant is configured", which the route reports as such.
    credential_reissuer: Option<Arc<dyn OperatorCredentialReissuer>>,
    /// The participant warm's stop handle (DUR-020 R4, R5), captured at
    /// construction for the same reason as the two above. The shutdown flush
    /// stops the warm first, so no conversation begins restoring after the
    /// server has begun to stop. `None` is "no participant is configured".
    participant_warm: Option<ParticipantWarm>,
    /// Dedup-on-delivery cache for publishes to **durable** channels. Keyed by
    /// the per-message idempotency key carried on the publish frame; a duplicate
    /// key is suppressed before fan-out so a subscriber receives it at most once.
    ///
    /// The namespace is stable across restarts by design (A1 §5): on a durable
    /// channel the MESSAGE survives a crash too, so a surviving receipt is
    /// truthful.
    dedup: DedupCache,
    /// Dedup-on-delivery cache for publishes to **ephemeral** channels
    /// (A1 §5, graft §0.2 — the panel's sharpest crash hole).
    ///
    /// Identical machinery, different namespace: this one carries a per-boot
    /// incarnation id, so a receipt and the message it vouches for share a
    /// lifetime. Without it, `ephemeral + keyed publish -> Defer (buffered in
    /// memory) -> complete_receipt (persisted) -> crash before the consumer
    /// pops` loses the message while the receipt survives to suppress the
    /// producer's retry — silent permanent loss of something the producer was
    /// told was admitted. With it, the post-crash retry re-claims and
    /// re-delivers, which is exactly ephemeral's at-least-once contract.
    ///
    /// This changes no dedup API. It changes only the namespace string the
    /// server composes.
    dedup_ephemeral: DedupCache,
    /// Durable-channel idempotency keys whose publish outcome is not yet
    /// known, held until their row is looked up by its key. A cache of the
    /// durable in-flight claims, rebuilt from them when the services start.
    held_keys: unresolved::HeldKeys,
    /// The store history whose in-flight claims are still to be resolved:
    /// some until the resolve has run, behind the listening door or at the
    /// first keyed durable admission, whichever comes first (`unresolved_start.rs`).
    claims_start: Mutex<Option<StoreHistory>>,
    conversation_supervisor: Arc<ConversationSupervisor>,
    /// Registered custom conversation responders, keyed by conversation subject.
    ///
    /// When a conversation is opened (`open_conversation`), the subject is looked
    /// up here: a registered [`ParticipantBehaviour`] becomes the conversation's
    /// participant; with no registration the conversation falls back to the
    /// built-in [`EchoBehaviour`], preserving the original echo semantics exactly.
    /// This is the seam aion #13 plugs a remote worker responder into. Interior
    /// mutability is required because the services are shared behind `&self`.
    responders: Mutex<ResponderRegistry>,
    next_message_id: AtomicU64,
    next_subscription_id: AtomicU64,
    /// SRV-012: the console's held topology and tap.
    console: Arc<crate::console::ConsoleHub>,
}

impl LiminalConnectionServices {
    /// Builds library-backed services from validated server configuration.
    ///
    /// Durable-mode channels are backed by a shared haematite event store so
    /// their publishes are persisted and survive the graceful-shutdown flush;
    /// ephemeral channels carry no store.
    ///
    /// Full-only: a config selecting the worker-front-door profile is rejected at
    /// entry (before any store is built), so this constructor can never build full
    /// services for a profile that forbids them. Profile-aware callers go through
    /// [`build_connection_services`] instead.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the config selects a non-full profile or a
    /// configured channel cannot be initialized.
    pub fn from_config(config: &ServerConfig) -> Result<Self, ServerError> {
        require_full_profile(config)?;
        let path = required_persistence_path(config)?;
        let history = StoreHistory::of(path);
        let subsystems = RootedProductionSubsystems::new(path)?;
        let store = subsystems.durable_store(path)?;
        Self::from_config_with_store_via(config, store, history, &subsystems)
    }

    /// Builds services over a caller-provided durable store.
    ///
    /// Used by tests that need to inspect persisted state through the same store
    /// handle the durable channels write to.
    ///
    /// Full-only: rejects a worker-front-door profile at entry, exactly like
    /// [`Self::from_config`].
    ///
    /// # Errors
    /// Returns [`ServerError`] when the config selects a non-full profile or a
    /// configured channel cannot be initialized.
    pub fn from_config_with_store(
        config: &ServerConfig,
        durable_store: Arc<dyn DurableStore>,
    ) -> Result<Self, ServerError> {
        require_full_profile(config)?;
        #[cfg(any(test, feature = "test-support"))]
        if config.persistence_path.is_none() {
            return Self::from_config_with_store_via(
                config,
                durable_store,
                StoreHistory::Recorded,
                &ProductionSubsystems,
            );
        }
        let subsystems = RootedProductionSubsystems::new(required_persistence_path(config)?)?;
        Self::from_config_with_store_via(config, durable_store, StoreHistory::Recorded, &subsystems)
    }

    /// [`Self::from_config_with_store`] with the subsystem factory injected.
    ///
    /// The channel supervisor and conversation supervisor are constructed ONLY
    /// through `subsystems` — there is no direct constructor call in this body —
    /// so a factory that records as a side effect of constructing cannot have its
    /// recording omitted (§9 D2 seam census, record-by-construction). No profile
    /// check here: the caller (the public wrapper or the profile dispatch in
    /// [`build_connection_services`]) has already established the full profile.
    /// `history` says whether `durable_store` can already hold in-flight
    /// claims for the services to look up before they judge a publish.
    fn from_config_with_store_via(
        config: &ServerConfig,
        durable_store: Arc<dyn DurableStore>,
        history: StoreHistory,
        subsystems: &dyn SubsystemFactory,
    ) -> Result<Self, ServerError> {
        // Build ONE shared channel supervisor for the whole server. When a
        // [cluster] section is present it is distribution-enabled, so every
        // channel actor and subscriber shares the clustered scheduler the cluster
        // attaches its process-group transport to (SRV-005, Constraint B).
        let cluster = subsystems.channel_cluster(config.cluster.as_ref())?;
        let mut channels = HashMap::new();
        for channel in &config.channels {
            // Resolve the channel's real JSON Schema (loaded from `schema_ref`
            // during config validation) or the permissive empty schema when the
            // channel declared none. The protocol schema id advertised at
            // subscribe time is derived from the SAME schema bytes so an SDK
            // deriving ids from schema bytes converges on it.
            let resolved = resolve_channel_schema(channel);
            let configured = build_configured_channel(
                &channel.name,
                resolved,
                channel.durable,
                ChannelOrigin::BootConfigured,
                &durable_store,
                cluster.supervisor(),
            )
            .map_err(|error| ServerError::ConfigValidation {
                message: error.boot_message(&channel.name),
            })?;
            channels.insert(channel.name.clone(), Arc::new(configured));
        }
        let conversation_supervisor = subsystems.conversation_supervisor()?;
        let dedup = history.delivery_dedup(Arc::clone(&durable_store));
        // A1 §5: the ephemeral namespace is scoped to THIS incarnation, so an
        // ephemeral receipt cannot outlive the in-memory message it vouches for.
        // No earlier process used it, so it starts empty.
        let ephemeral_store = ephemeral_dedup_store(&durable_store, history, subsystems);
        let store_counts = combined_store_counts(&durable_store, &ephemeral_store);
        let dedup_ephemeral = DedupCache::new_empty(ephemeral_store, ephemeral_dedup_namespace());
        // Production participant activation (LP gap closure, Part B): the
        // deployment's [participant] section installs the ONE production
        // semantic handler, sealed together with the same durable store the
        // conversation logs live in, under the configured wire-frame limit.
        // No section, no service — the capability bit stays off and the
        // connection path is byte-identical to the pre-activation build.
        let signals = ServerSignals::default();
        let installed_participant = config
            .participant
            .as_ref()
            .map(|participant| install_participant(&durable_store, participant, &signals))
            .transpose()?;
        let device = config
            .cluster
            .as_ref()
            .map_or("local", |cluster| cluster.node_name.as_str());
        let console = Arc::new(console_hub(
            device,
            &channels,
            config.persistence_path.is_some(),
        )?);
        signals.install_console(Arc::clone(&console))?;
        cluster.observer().set_console(Arc::clone(&console))?;
        let (participant_service, unloadable_conversations, credential_reissuer, participant_warm) =
            installed_participant.map_or((None, None, None, None), |installed| {
                (
                    Some(installed.service),
                    Some(installed.unloadable_conversations),
                    Some(installed.credential_reissuer),
                    Some(installed.warm),
                )
            });
        let services = Self {
            channels: RwLock::new(channels),
            max_channels: config.limits.max_channels,
            cluster,
            durable_store,
            store_counts,
            ephemeral_sweep: subsystems.ephemeral_sweep_counts(),
            signals,
            participant_service,
            unloadable_conversations,
            credential_reissuer,
            participant_warm,
            dedup,
            dedup_ephemeral,
            held_keys: unresolved::HeldKeys::default(),
            // Every durable in-flight claim is looked up by its key before the
            // services judge a keyed durable publish: behind the listening
            // door, or by that publish itself when it comes first.
            claims_start: Mutex::new(Some(history)),
            conversation_supervisor,
            responders: Mutex::new(HashMap::new()),
            next_message_id: AtomicU64::new(1),
            next_subscription_id: AtomicU64::new(1),
            console,
        };
        Ok(services)
    }

    /// Builds services with no configured channels.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the conversation supervisor scheduler cannot start.
    #[cfg(any(test, feature = "test-support"))]
    pub fn empty() -> Result<Self, ServerError> {
        let conversation_supervisor = ProductionSubsystems.conversation_supervisor()?;
        let store =
            open_ephemeral(DEFAULT_SHARD_COUNT).map_err(|error| ServerError::ConfigValidation {
                message: format!("failed to open test durable store: {error}"),
            })?;
        let durable_store: Arc<dyn DurableStore> = Arc::new(CountingStore::new(store));
        let store_counts = combined_store_counts(&durable_store, &durable_store);
        let dedup = StoreHistory::Fresh.delivery_dedup(Arc::clone(&durable_store));
        // A1 §5: the ephemeral namespace is scoped to THIS incarnation, so an
        // ephemeral receipt cannot outlive the in-memory message it vouches for.
        let dedup_ephemeral =
            DedupCache::new_empty(Arc::clone(&durable_store), ephemeral_dedup_namespace());
        let signals = ServerSignals::default();
        let console = Arc::new(console_hub("local", &HashMap::new(), false)?);
        signals.install_console(Arc::clone(&console))?;
        let cluster = build_channel_cluster(None)?;
        cluster.observer().set_console(Arc::clone(&console))?;
        Ok(Self {
            channels: RwLock::new(HashMap::new()),
            // No config, so no declared bound: this builder serves tests and
            // callers with no channels at all, and a registration against it
            // refuses `CapNotConfigured` exactly as an undeclared deployment's
            // would.
            max_channels: None,
            cluster,
            durable_store,
            store_counts,
            ephemeral_sweep: None,
            signals,
            participant_service: None,
            unloadable_conversations: None,
            credential_reissuer: None,
            participant_warm: None,
            dedup,
            dedup_ephemeral,
            held_keys: unresolved::HeldKeys::default(),
            claims_start: Mutex::new(None),
            conversation_supervisor,
            responders: Mutex::new(HashMap::new()),
            next_message_id: AtomicU64::new(1),
            next_subscription_id: AtomicU64::new(1),
            console,
        })
    }

    /// The retained counts from the root's one startup sweep.
    #[must_use]
    pub const fn ephemeral_sweep_counts(&self) -> Option<EphemeralSweepCounts> {
        self.ephemeral_sweep
    }

    /// The shared channel supervisor + cluster resolver backing this service.
    ///
    /// The server runtime uses this to attach the cluster to the channel
    /// supervisor's clustered scheduler (SRV-005).
    #[must_use]
    pub const fn channel_cluster(&self) -> &ChannelCluster {
        &self.cluster
    }

    /// Returns the shared durable store backing this service's durable channels.
    #[must_use]
    pub fn durable_store(&self) -> Arc<dyn DurableStore> {
        Arc::clone(&self.durable_store)
    }

    /// The console's held topology and tap (SRV-012).
    #[must_use]
    pub fn console(&self) -> &crate::console::ConsoleHub {
        &self.console
    }

    /// Applies a topology change; a poisoned model is counted by the hub
    /// (`topology_emit_failures`) and logged, never recovered.
    fn console_changed(&self, change: crate::console::TopologyChange) {
        self.signals.topology(change);
    }

    /// Conversation subjects with a registered responder, sorted. Read from
    /// the in-memory registry; touches no store.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the responder registry lock is poisoned.
    pub fn responder_subjects(&self) -> Result<Vec<String>, ServerError> {
        let responders = self.lock_responders()?;
        let mut subjects: Vec<String> = responders.keys().cloned().collect();
        drop(responders);
        subjects.sort();
        Ok(subjects)
    }

    /// Restores every participant conversation the warm has not yet reached,
    /// on the caller's thread, and answers how many refused (DUR-020 R4).
    /// The store-count journeys call it so their windows do not race the
    /// warm thread; the server itself never waits for the warm.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the pending-restore lock is poisoned.
    pub fn finish_participant_warm(&self) -> Result<usize, ServerError> {
        self.participant_warm.as_ref().map_or(Ok(0), |warm| {
            warm.finish()
                .map_err(|error| ServerError::ParticipantStartupRestore {
                    message: error.to_string(),
                })
        })
    }

    /// Reads the store work these services have done since they were built,
    /// across every store they built.
    ///
    /// Empty (every snapshot zero) when the durable store was supplied by the
    /// caller and does not count.
    #[must_use]
    pub fn store_counts(&self) -> StoreCountsHandle {
        self.store_counts.clone()
    }

    /// The production participant handler's refused-load record, when a
    /// participant is configured.
    ///
    /// The server's startup path publishes this onto the health endpoint so
    /// `GET /unloadable-conversations` answers from the same record the
    /// handler writes. `None` means no participant is configured at all, which
    /// the surface reports as a distinct state from "nothing was refused".
    #[must_use]
    pub fn unloadable_conversation_record(&self) -> Option<UnloadableConversationRecord> {
        self.unloadable_conversations.clone()
    }

    /// The A7 operator credential-re-issue authority, when a participant is
    /// configured (§0.18).
    ///
    /// The startup path publishes this onto the health endpoint so
    /// `POST /operator/credential-reissue` reaches the SAME serialized
    /// participant-state point every wire operation does. `None` means no
    /// participant is configured, which the route reports as a distinct state
    /// from an unknown identity.
    #[must_use]
    pub fn credential_reissuer(&self) -> Option<Arc<dyn OperatorCredentialReissuer>> {
        self.credential_reissuer.clone()
    }

    /// Installs a complete participant bundle in full-service supervisor tests.
    ///
    /// Production full services intentionally stay disabled until a concrete
    /// lifecycle handler exists. This consuming test builder exercises the real
    /// supervisor activation path without allowing an already-shared adapter to
    /// change capability posture.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn with_participant_service(
        mut self,
        participant_service: InstalledParticipantService,
    ) -> Self {
        self.participant_service = Some(participant_service);
        self
    }

    /// Returns the conversation supervisor backing supervised conversations.
    ///
    /// Tests use this to reach the underlying beamr scheduler so they can spawn
    /// or terminate participant processes and exercise crash detection.
    #[must_use]
    pub fn conversation_supervisor(&self) -> Arc<ConversationSupervisor> {
        Arc::clone(&self.conversation_supervisor)
    }

    /// Registers a custom conversation responder for a routing `subject`.
    ///
    /// When a conversation is later opened with this exact `subject`, its
    /// participant runs `behaviour` instead of the built-in [`EchoBehaviour`].
    /// The responder is spawned and supervised identically to the echo
    /// participant — a real linked beamr process with the same crash-detection
    /// semantics — so this exposes the responder seam without changing how
    /// participants run. Registering a subject that already has a responder
    /// replaces it; the previous behaviour is returned.
    ///
    /// This is the liminal-side seam aion #13 plugs a remote worker into: it
    /// registers a responder that forwards each request to the worker and routes
    /// the worker's reply back through the conversation. Subjects with no
    /// registration keep echoing, so existing callers are unaffected.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the responder registry lock is poisoned.
    pub fn register_responder(
        &self,
        subject: impl Into<String>,
        behaviour: Arc<dyn ParticipantBehaviour>,
    ) -> Result<Option<Arc<dyn ParticipantBehaviour>>, ServerError> {
        let subject = subject.into();
        let mut responders = self.lock_responders()?;
        let previous = responders.insert(subject.clone(), behaviour);
        drop(responders);
        self.console_changed(crate::console::TopologyChange::Service {
            subject,
            present: true,
        });
        Ok(previous)
    }

    /// Removes the custom responder registered for `subject`, if any.
    ///
    /// After removal the subject reverts to the built-in [`EchoBehaviour`] on the
    /// next [`Self::open_conversation`]. Returns the removed behaviour when one
    /// was registered.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the responder registry lock is poisoned.
    pub fn unregister_responder(
        &self,
        subject: &str,
    ) -> Result<Option<Arc<dyn ParticipantBehaviour>>, ServerError> {
        let mut responders = self.lock_responders()?;
        let removed = responders.remove(subject);
        drop(responders);
        if removed.is_some() {
            self.console_changed(crate::console::TopologyChange::Service {
                subject: subject.to_owned(),
                present: false,
            });
        }
        Ok(removed)
    }

    /// Resolves the responder behaviour for `subject`: the registered custom
    /// responder when present, otherwise the built-in [`EchoBehaviour`].
    ///
    /// This is the single routing decision behind the seam — registered-or-echo —
    /// so the fallback is identical to the original hard-wired echo path.
    fn responder_for(&self, subject: &str) -> Result<Arc<dyn ParticipantBehaviour>, ServerError> {
        let responders = self.lock_responders()?;
        Ok(responders.get(subject).map_or_else(
            || Arc::new(EchoBehaviour) as Arc<dyn ParticipantBehaviour>,
            Arc::clone,
        ))
    }

    /// Locks the responder registry, mapping a poisoned lock to a [`ServerError`]
    /// rather than panicking (the workspace denies `unwrap`/`expect`/`panic`).
    fn lock_responders(&self) -> Result<std::sync::MutexGuard<'_, ResponderRegistry>, ServerError> {
        self.responders
            .lock()
            .map_err(|_poisoned| ServerError::ListenerAccept {
                message: "responder registry lock poisoned".to_owned(),
            })
    }

    /// Takes the channel roster's read guard, mapping a poisoned lock to a
    /// [`ServerError`] rather than panicking (the workspace denies
    /// `unwrap`/`expect`/`panic`), exactly as [`Self::lock_responders`] does for
    /// the responder registry.
    ///
    /// Every caller clones the `Arc` it needs out of the returned guard and drops
    /// the guard before doing anything else: the roster lock is never held across
    /// a call into the liminal library.
    fn read_channels(
        &self,
    ) -> Result<std::sync::RwLockReadGuard<'_, HashMap<String, Arc<ConfiguredChannel>>>, ServerError>
    {
        self.channels
            .read()
            .map_err(|_poisoned| ServerError::ListenerAccept {
                message: "channel roster lock poisoned".to_owned(),
            })
    }

    /// Subscribes to a configured channel and returns the raw library
    /// subscription handle so a test can drain the subscriber inbox directly and
    /// observe exactly which messages reached a subscriber.
    #[cfg(test)]
    pub(in crate::server) fn channel_handle_for_test(
        &self,
        channel: &str,
    ) -> Result<ChannelHandle, ServerError> {
        self.read_channels()?
            .get(channel)
            .map(|configured| configured.handle.clone())
            .ok_or_else(|| ServerError::ListenerAccept {
                message: format!("channel '{channel}' is not configured"),
            })
    }

    #[cfg(test)]
    pub(crate) fn subscribe_handle_for_test(
        &self,
        channel: &str,
    ) -> Result<liminal::channel::SubscriptionHandle, ServerError> {
        self.subscribe_handle_for_test_with_install(channel, None)
    }

    /// [`Self::subscribe_handle_for_test`] carrying an explicit install, so a
    /// test can put a subscriber under a REAL §5 regime (byte budget, fairness
    /// cap) and observe what a publish the bus dropped does to the dedup
    /// receipt.
    #[cfg(test)]
    pub(crate) fn subscribe_handle_for_test_with_install(
        &self,
        channel: &str,
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<liminal::channel::SubscriptionHandle, ServerError> {
        let channels = self.read_channels()?;
        let configured =
            channels
                .get(channel)
                .map(Arc::clone)
                .ok_or_else(|| ServerError::ListenerAccept {
                    message: format!("channel '{channel}' is not configured"),
                })?;
        drop(channels);
        install
            .map_or_else(
                || configured.handle.subscribe(),
                |install| configured.handle.subscribe_with_install(install),
            )
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("liminal subscribe failed for channel '{channel}': {error}"),
            })
    }

    /// Claims the delivery right for an idempotency key.
    ///
    /// Returns `Ok(true)` when this is the first publish for the key (the caller
    /// may deliver), and `Ok(false)` when the key was already claimed/completed (a
    /// duplicate the caller must suppress). The dedup cache is driven synchronously
    /// over the in-memory haematite store via the durable bridge.
    fn claim_delivery(cache: &DedupCache, key: &str) -> Result<bool, ServerError> {
        let decision = block_on(cache.claim_or_get(key, dedup_timestamp_millis()))
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("dedup bridge failed for key '{key}': {error}"),
            })?
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("dedup claim failed for key '{key}': {error}"),
            })?;
        Ok(matches!(decision, DedupDecision::Claimed))
    }

    /// The dedup namespace a publish to `channel` belongs in (A1 §5).
    ///
    /// Durability decides, because durability is what determines whether a
    /// receipt can outlive the message it vouches for. A durable channel's
    /// message survives a crash, so its receipt may; an ephemeral channel's
    /// does not, so its receipt must not.
    const fn dedup_for(&self, channel: &ConfiguredChannel) -> &DedupCache {
        match channel.handle.config().mode {
            ChannelMode::Durable => &self.dedup,
            ChannelMode::Ephemeral => &self.dedup_ephemeral,
        }
    }

    /// Releases a dangling in-flight dedup claim after a failed delivery.
    ///
    /// Best-effort: a release failure cannot mask the original publish error, so
    /// this returns nothing and logs at `error` level instead of surfacing. It is
    /// never silent — the leak (a permanently suppressed key) must be observable.
    /// `release_claim` itself never clobbers a stored receipt, so calling it on the
    /// failure path is safe even if a concurrent completion raced ahead.
    fn release_claim(cache: &DedupCache, key: &str) {
        match block_on(cache.release_claim(key)) {
            Ok(Ok(())) => {}
            Ok(Err(error)) => {
                tracing::error!(
                    idempotency_key = key,
                    %error,
                    "failed to release dedup claim after publish failure; key may stay suppressed"
                );
            }
            Err(error) => {
                tracing::error!(
                    idempotency_key = key,
                    %error,
                    "dedup release bridge failed after publish failure; key may stay suppressed"
                );
            }
        }
    }

    /// The next connection-level message id.
    fn next_message_id(&self) -> u64 {
        self.next_message_id.fetch_add(1, Ordering::Relaxed)
    }

    /// The outcome of a publish the channel took, recorded as accepted with
    /// its genuine subscriber deliveries. The delivered count (0 for an empty
    /// channel) is the same signal the delivery ack is derived from.
    fn delivered(&self, delivery: &ChannelDelivery) -> PublishOutcome {
        crate::metrics::publish_accepted();
        let delivered_count = u64::try_from(delivery.delivered_count()).unwrap_or(u64::MAX);
        crate::metrics::deliveries_recorded(delivered_count);
        PublishOutcome {
            message_id: self.next_message_id(),
            delivered: delivery.is_delivered(),
            durable_position: delivery.durable_position(),
            reach: Reach::of(delivery),
        }
    }
}

#[path = "deferred_store.rs"]
mod deferred_store;
#[path = "unresolved.rs"]
mod unresolved;
#[path = "unresolved_start.rs"]
mod unresolved_start;
#[cfg(test)]
#[path = "unresolved_start_tests.rs"]
mod unresolved_start_tests;

use deferred_store::{DeferredStore, ThrowawayStoreOpener};
use unresolved_start::StoreHistory;

/// The runtime channel-registration surface.
///
/// # Reachability: a test seam today, not a production capability
///
/// Production channels come exclusively from boot configuration; no production
/// code path calls `register_channel` (caller census at `c381f01`, G2 in
/// `docs/design/wire-durability-gaps-20260831.md` — the only callers are
/// `services_registry_tests.rs` and `apply_admission_tests.rs`). The method is
/// kept because its semantics — cap gating, typed identity refusals,
/// idempotent re-registration — are the contract a future runtime admission
/// path must satisfy, and the tests below pin that contract now. Wiring an
/// actual runtime admission route (an operator endpoint, a wire frame) is a
/// product decision that has NOT been taken; nothing here implies it. Reading
/// this method as an existing operator capability is the false-affordance
/// mistake this paragraph exists to prevent.
///
/// Inherent methods, not trait methods, and deliberately so: this is
/// authority-moving vocabulary that belongs to the ONE adapter owning a channel
/// roster. Putting `register`/`quiesce` on the public [`ConnectionServices`]
/// trait would break every external implementor and would hand
/// register/quiesce words to a profile that serves no channels at all.
///
/// The same seam as the existing runtime-mutation API on this type
/// (`register_responder`/`unregister_responder`): `&self`, interior mutability,
/// typed `Result`.
impl LiminalConnectionServices {
    /// Registers `spec` on the live roster.
    ///
    /// Test seam today: see the reachability note on this `impl` block — no
    /// production path reaches this method.
    ///
    /// Idempotent when an entry of that name already has an IDENTICAL
    /// configuration — mode, protocol schema id, and schema document, all three
    /// — and refuses typed, naming the first differing field, otherwise. An
    /// identical registration against a boot-configured entry answers
    /// [`Registered::AlreadyIdentical`] and leaves its origin alone: flipping it
    /// would make the entry lie about its restart fate and would move it into
    /// the counted population without a channel having been created.
    ///
    /// The cap is consulted first, in two steps with different reach. An absent
    /// `limits.max_channels` refuses EVERY call, identical or not, before the
    /// roster is read at all: a deployment that has declared no bound has not
    /// said what it admits, and unbounded-by-default is not a bound. A REACHED
    /// cap refuses only a call that would create an entry — it gates the insert,
    /// which is the population it bounds. An identical re-registration inserts
    /// nothing and so is answered `AlreadyIdentical` even at a full roster;
    /// refusing it would break idempotency at exactly the boundary a projector
    /// re-projecting its record crosses.
    ///
    /// # Errors
    /// Returns [`ChannelRegistryError`] when no cap is configured, the cap is
    /// reached, the name exists with a different configuration, the schema bytes
    /// do not parse or compile, durable initialization over the shared store
    /// fails, or the roster lock is poisoned.
    pub fn register_channel(
        &self,
        spec: &ChannelRegistration,
    ) -> Result<Registered, ChannelRegistryError> {
        // The cap is consulted before the roster is touched: an undeclared bound
        // refuses every runtime registration, identical or not, because the
        // deployment has not said what it admits.
        let Some(limit) = self.max_channels else {
            return Err(ChannelRegistryError::CapNotConfigured {
                cap: MAX_CHANNELS_KEY,
            });
        };
        // Schema resolution is pure (a JSON parse and a digest), so it runs with
        // no lock held and its result serves BOTH the identity comparison and
        // the construction below.
        let resolved = resolve_schema_bytes(spec.schema_bytes.as_deref()).map_err(|error| {
            ChannelRegistryError::SchemaRejected {
                name: spec.name.clone(),
                message: error.to_string(),
            }
        })?;

        // Fast paths, under a READ lock: an already-identical registration
        // builds nothing, and a full roster refuses before paying for a
        // construction it would discard. Neither is the authority — the write
        // lock below re-decides both, because the roster can move in between.
        let (existing, registered_count) = {
            let channels = self.read_roster()?;
            let existing = channels.get(&spec.name).map(Arc::clone);
            let registered_count = runtime_registered_count(&channels);
            drop(channels);
            (existing, registered_count)
        };
        if let Some(existing) = existing {
            return compare_registration(&existing, spec, &resolved);
        }
        if registered_count >= limit {
            return Err(ChannelRegistryError::CapReached {
                cap: MAX_CHANNELS_KEY,
                limit,
            });
        }

        // Construction runs with NO lock held. A durable channel recovers its
        // per-partition sequence counters from the store here, which is O(stream
        // length) in store reads; holding the roster lock across it would put a
        // slow store walk on every connection's publish and subscribe path.
        let configured = build_configured_channel(
            &spec.name,
            resolved,
            spec.durable,
            ChannelOrigin::RuntimeRegistered,
            &self.durable_store,
            self.cluster.supervisor(),
        )
        .map_err(|error| error.into_registry_error(&spec.name))?;

        // The authoritative decision: identity, cap, and insert under ONE write
        // lock, so the count that admitted the entry and the insert that added
        // it cannot be separated by a concurrent registration.
        let mut channels = self.write_roster()?;
        let raced = channels.get(&spec.name).map(Arc::clone);
        let registered_count = runtime_registered_count(&channels);
        if raced.is_none() && registered_count < limit {
            channels.insert(spec.name.clone(), Arc::new(configured));
            drop(channels);
            self.console_changed(crate::console::TopologyChange::Channel {
                name: spec.name.clone(),
                life: line_life(spec.durable),
                state: crate::console::records::LineState::Live,
            });
            return Ok(Registered::Created);
        }
        drop(channels);
        if let Some(raced) = raced {
            // A racer registered this name while the channel above was being
            // built. The built entry is dropped unused: it owns no actor (the
            // actor is spawned lazily on first use) and its durable
            // construction only READ the store, so discarding it changes
            // nothing an observer could see.
            return compare_registration(&raced, spec, &schema_of(&configured));
        }
        Err(ChannelRegistryError::CapReached {
            cap: MAX_CHANNELS_KEY,
            limit,
        })
    }

    /// Moves `name` from active to quiesced with a named `reason`. ONE-WAY.
    ///
    /// New publishes and new subscribes are refused afterwards, carrying the
    /// reason. Existing subscriptions are UNTOUCHED: nothing revokes a
    /// subscription handle, the actor's subscriber list is not walked, no EXIT
    /// is sent, and the channel actor keeps running. Quiesce is a roster-level
    /// admission decision, not an actor command.
    ///
    /// Re-quiescing under the IDENTICAL reason is `Ok(())`; a DIFFERENT reason
    /// refuses, carrying the reason already on record.
    ///
    /// The return does NOT mean "no new subscriber can appear". A subscribe that
    /// has already passed admission completes and gets its stream — the
    /// linearisation point is the admission read, not the subscribe's
    /// completion. A consumer that needs "nobody is attached" must observe
    /// attachment directly.
    ///
    /// # Errors
    /// Returns [`ChannelRegistryError`] when the name is not registered, is
    /// already quiesced under a different reason, or the roster lock is
    /// poisoned.
    pub fn quiesce_channel(
        &self,
        name: &str,
        reason: impl Into<String>,
    ) -> Result<(), ChannelRegistryError> {
        let reason = reason.into();
        // A READ lock: the state machine lives on the ENTRY, not in the map, so
        // the operation the design most wants to be safe never blocks a reader.
        let configured = {
            let channels = self.read_roster()?;
            channels.get(name).map(Arc::clone).ok_or_else(|| {
                ChannelRegistryError::NotRegistered {
                    name: name.to_owned(),
                }
            })?
        };
        configured.quiesce(name, &reason)?;
        self.console_changed(crate::console::TopologyChange::Channel {
            name: name.to_owned(),
            life: line_life(configured.durable),
            state: crate::console::records::LineState::Quiesced,
        });
        Ok(())
    }

    /// Cheap typed probe: one roster read plus one atomic load.
    ///
    /// Touches no actor and therefore CANNOT spawn one. Every handle accessor
    /// that could answer a question about a channel's activity routes through
    /// the lazy-spawn path, so a probe built on one would materialise the actor
    /// of the idle channel it was asked about — turning a read into a side
    /// effect. This reads the roster entry's own recorded fields and nothing
    /// else.
    ///
    /// # Errors
    /// Returns [`ChannelRegistryError::RosterUnavailable`] only.
    pub fn channel_status(&self, name: &str) -> Result<ChannelStatus, ChannelRegistryError> {
        let configured = {
            let channels = self.read_roster()?;
            channels.get(name).map(Arc::clone)
        };
        Ok(
            configured.map_or(ChannelStatus::NotRegistered, |configured| {
                status_of(&configured)
            }),
        )
    }

    /// Every roster entry with its status, sorted by name, from ONE roster
    /// read (SRV-012 R7: the console's topology snapshot).
    ///
    /// # Errors
    /// Returns [`ChannelRegistryError::RosterUnavailable`] when the roster
    /// lock is poisoned.
    pub fn channel_census(&self) -> Result<Vec<(String, ChannelStatus)>, ChannelRegistryError> {
        let mut census: Vec<(String, ChannelStatus)> = self
            .read_roster()?
            .iter()
            .map(|(name, configured)| (name.clone(), status_of(configured)))
            .collect();
        census.sort_by(|left, right| left.0.cmp(&right.0));
        Ok(census)
    }

    /// The whole roster: one minimal descriptor per entry, sorted by name.
    ///
    /// The census companion to [`Self::channel_status`]. A by-name probe answers
    /// about a name the caller already suspects; only an enumeration can reveal
    /// a name the caller does not know to ask about, and a verification sweep
    /// with no population denominator is an instrument shape this estate
    /// forbids. Touches no actor, under the same constraint as the probe.
    ///
    /// # Errors
    /// Returns [`ChannelRegistryError::RosterUnavailable`] only.
    pub fn registered_channels(&self) -> Result<Vec<ChannelDescriptor>, ChannelRegistryError> {
        let entries: Vec<(String, Arc<ConfiguredChannel>)> = {
            let channels = self.read_roster()?;
            channels
                .iter()
                .map(|(name, configured)| (name.clone(), Arc::clone(configured)))
                .collect()
        };
        let mut descriptors: Vec<ChannelDescriptor> = entries
            .into_iter()
            .map(|(name, configured)| ChannelDescriptor {
                name,
                origin: configured.origin,
                state: configured.state(),
            })
            .collect();
        descriptors.sort_by(|left, right| left.name.cmp(&right.name));
        Ok(descriptors)
    }

    /// The roster admission funnel: the ONE place a channel operation's
    /// permission is decided.
    ///
    /// Returns the admitted entry, so the caller works from the value the
    /// decision was made on rather than reading the roster a second time and
    /// risking a different answer. This read is the LINEARISATION POINT for the
    /// quiesce race: everything the caller does with the returned entry happens
    /// outside the lock and is not re-checked, which is exactly why a quiesce
    /// that commits after this returns does not stop the operation it admitted.
    ///
    /// # Errors
    /// Returns [`ChannelAccessError`] when the channel is absent, quiesced, or
    /// the roster lock is poisoned.
    fn admit_channel(&self, channel: &str) -> Result<Arc<ConfiguredChannel>, ChannelAccessError> {
        let configured = {
            let channels = self.channels.read().map_err(|_poisoned| {
                ChannelAccessError::RosterUnavailable {
                    message: ROSTER_POISONED.to_owned(),
                }
            })?;
            channels.get(channel).map(Arc::clone)
        };
        let configured = configured.ok_or_else(|| ChannelAccessError::NotRegistered {
            name: channel.to_owned(),
        })?;
        if configured.state.load(Ordering::Acquire) == STATE_QUIESCED {
            return Err(ChannelAccessError::Quiesced {
                name: channel.to_owned(),
                reason: configured.recorded_quiesce_reason(),
            });
        }
        Ok(configured)
    }

    /// Takes the roster's READ guard for the registration surface, mapping a
    /// poisoned lock to a typed [`ChannelRegistryError`].
    ///
    /// Separate from [`Self::read_channels`] because the two surfaces answer
    /// different callers with different error types; recovering one from the
    /// other would mean inspecting a message.
    fn read_roster(
        &self,
    ) -> Result<
        std::sync::RwLockReadGuard<'_, HashMap<String, Arc<ConfiguredChannel>>>,
        ChannelRegistryError,
    > {
        #[cfg(test)]
        ROSTER_READS.with(|reads| reads.set(reads.get() + 1));
        self.channels
            .read()
            .map_err(|_poisoned| ChannelRegistryError::RosterUnavailable {
                message: ROSTER_POISONED.to_owned(),
            })
    }

    /// Takes the roster's WRITE guard — held only for the check-and-insert that
    /// must be atomic, never across a call into the liminal library.
    fn write_roster(
        &self,
    ) -> Result<
        std::sync::RwLockWriteGuard<'_, HashMap<String, Arc<ConfiguredChannel>>>,
        ChannelRegistryError,
    > {
        self.channels
            .write()
            .map_err(|_poisoned| ChannelRegistryError::RosterUnavailable {
                message: ROSTER_POISONED.to_owned(),
            })
    }

    /// Gives back the tap slot a subscribe to `channel` reserved at
    /// admission, taking its tap channel off the roster when no tap stands
    /// behind it (SRV-012 R6).
    fn give_back_tap(&self, channel: &str) {
        self.console.release_tap(channel, |subject| {
            crate::console::ServerPublisher::retire_console_channel(self, subject)
        });
    }

    /// The trait's subscribe, from the console guard to the subscription;
    /// [`ConnectionServices::subscribe`] gives back the tap reservation when
    /// this fails.
    fn open_subscription(
        &self,
        channel: &str,
        accepted_schemas: &[ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        // The frame path refused this already at admission; this guard is for
        // a caller holding the trait directly.
        refuse_console_tap_target(channel).map_err(|error| access_to_server_error(&error))?;
        // As in `publish`: the admission funnel clones the entry's `Arc` out
        // under the read guard and the guard is released before schema
        // negotiation or the actor round-trip below. That release is what makes
        // this the linearisation point — a quiesce landing after it does not
        // stop the subscription this call is already building.
        let configured = self
            .admit_channel(channel)
            .map_err(|error| access_to_server_error(&error))?;
        let selected_schema = if accepted_schemas.is_empty() {
            configured.protocol_schema
        } else {
            liminal::protocol::negotiate_schema(configured.protocol_schema, accepted_schemas)
                .map_err(|error| server_error_from_protocol(&error))?
        };
        // `subscribe_with_install` installs the §5 budget/fairness cap and the R3
        // wake notifier on the inbox at construction — strictly before the
        // registration is published to the channel actor — so there is no window
        // in which a publish can land uncharged or without a wake.
        let subscription = install
            .map_or_else(
                || configured.handle.subscribe(),
                |install| configured.handle.subscribe_with_install(install),
            )
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("liminal subscribe failed for channel '{channel}': {error}"),
            })?;
        let id = self.next_subscription_id.fetch_add(1, Ordering::Relaxed);
        let mut connection_subscription = ConnectionSubscription::new(
            id,
            selected_schema,
            Box::new(LiminalSubscriptionResource { subscription }),
        );
        connection_subscription.set_channel(channel.to_owned());
        // SRV-012 R6: the console tap attaches and detaches through the
        // supervisor's observer, which hears every subscribe and every
        // departure once (ServerObserver).
        Ok(connection_subscription)
    }

    /// Refuses a subscribe that would start one more live tap than
    /// [`crate::console::MAX_LIVE_TAPS`] (SRV-012 R6); a subscriber to a tap
    /// already live is not a new tap.
    ///
    /// A poisoned tap table does not refuse the subscribe (R8: a console
    /// bookkeeping failure never fails the client operation it watches); the
    /// attach that follows meets the poison and counts it.
    ///
    /// # Errors
    /// Returns [`ChannelAccessError::ConsoleTapLimit`] naming the channel.
    fn refuse_console_tap_limit(&self, channel: &str) -> Result<(), ChannelAccessError> {
        match self.console.reserve_tap(channel) {
            Ok(false) => Err(ChannelAccessError::ConsoleTapLimit {
                name: channel.to_owned(),
                limit: crate::console::MAX_LIVE_TAPS,
            }),
            Ok(true) | Err(_) => Ok(()),
        }
    }
}

impl crate::console::ServerPublisher for LiminalConnectionServices {
    /// The server's own record on the one publish path, with a server origin:
    /// attributed to `liminal.<device>`, and never counted as a connection's
    /// work by the cost tick.
    fn publish_console(
        &self,
        subject: &str,
        payload: &[u8],
    ) -> Result<(), crate::console::ConsoleError> {
        let envelope = MessageEnvelope::new(
            ProtocolSchemaId::new([0_u8; ProtocolSchemaId::WIRE_LEN]),
            liminal::protocol::CausalContext::independent(),
            payload.to_vec(),
        );
        self.publish_with_origin(
            subject,
            &envelope,
            None,
            PublishOrigin::Server {
                device: self.console.device(),
            },
        )
        .map(|_published| ())
        .map_err(|source| crate::console::ConsoleError::Server {
            subject: subject.to_owned(),
            source,
        })
    }

    /// Takes a `console.tap.<target>` channel off the roster (SRV-012 R6).
    /// Only a console subject is ever removed here.
    fn retire_console_channel(&self, subject: &str) -> Result<(), crate::console::ConsoleError> {
        if !super::channel_registry::is_console_subject(subject) {
            return Ok(());
        }
        let mut channels =
            self.write_roster()
                .map_err(|error| crate::console::ConsoleError::Server {
                    subject: subject.to_owned(),
                    source: ServerError::ListenerAccept {
                        message: error.to_string(),
                    },
                })?;
        channels.remove(subject);
        drop(channels);
        Ok(())
    }
}

/// The production participant service and the operator surfaces taken from
/// its one live handler before it is erased behind
/// `dyn ParticipantSemanticHandler`.
struct InstalledParticipant {
    service: InstalledParticipantService,
    unloadable_conversations: UnloadableConversationRecord,
    credential_reissuer: Arc<dyn OperatorCredentialReissuer>,
    warm: ParticipantWarm,
}

/// Production participant activation (LP gap closure, Part B): the
/// deployment's [participant] section installs the ONE production semantic
/// handler, sealed together with the same durable store the conversation logs
/// live in, under the configured wire-frame limit.
///
/// # Errors
/// Returns [`ServerError::ParticipantStartupRestore`] when the handler cannot
/// be built or wired, and [`ServerError::ConfigValidation`] when the
/// wire-frame limit is below the protocol's minimum participant frame.
fn install_participant(
    durable_store: &Arc<dyn DurableStore>,
    participant: &ParticipantConfig,
    signals: &ServerSignals,
) -> Result<InstalledParticipant, ServerError> {
    let handler = ProductionParticipantHandler::new(Arc::clone(durable_store), *participant)
        .map_err(|error| ServerError::ParticipantStartupRestore {
            message: error.to_string(),
        })?;
    // Taken BEFORE the handler is erased behind
    // `dyn ParticipantSemanticHandler`. Construction restores no
    // conversation (DUR-020), so the record fills as conversations restore,
    // on first touch or in the warm, and is complete once the warm has
    // finished.
    handler.install_signals(signals.clone()).map_err(|error| {
        ServerError::ParticipantStartupRestore {
            message: error.to_string(),
        }
    })?;
    let unloadable_conversations = handler.unloadable_record();
    let handler = Arc::new(handler);
    let warm = handler
        .install_warm()
        .map_err(|error| ServerError::ParticipantStartupRestore {
            message: error.to_string(),
        })?;
    // A7 (§0.18): taken here for the same reason, and from the same one live
    // handler, so the operator surface and the participant wire cannot end up
    // talking to two different owners.
    let credential_reissuer: Arc<dyn OperatorCredentialReissuer> =
        Arc::<ProductionParticipantHandler>::clone(&handler);
    let service = InstalledParticipantService::new(
        handler,
        Arc::clone(durable_store),
        participant.wire_frame_limit,
    )
    .map_err(|error| ServerError::ConfigValidation {
        message: format!(
            "participant.wire_frame_limit: {} is below the protocol's minimum \
             complete participant frame ({error:?})",
            participant.wire_frame_limit
        ),
    })?;
    Ok(InstalledParticipant {
        service,
        unloadable_conversations,
        credential_reissuer,
        warm,
    })
}

/// The console hub for `device`, its model built from the roster as it
/// stands (responders, connections and peers arrive as changes).
///
/// # Errors
/// Returns [`ServerError::ListenerAccept`] when the device is blank or the
/// wall clock cannot be read.
fn console_hub(
    device: &str,
    channels: &HashMap<String, Arc<ConfiguredChannel>>,
    depot: bool,
) -> Result<crate::console::ConsoleHub, ServerError> {
    let this = crate::console::ThisDevice {
        id: device.to_owned(),
        liminal: env!("CARGO_PKG_VERSION").to_owned(),
        started: Instant::now(),
        depot,
    };
    let lines = channels.iter().map(|(name, configured)| {
        let state = match configured.state() {
            ChannelState::Active => crate::console::records::LineState::Live,
            ChannelState::Quiesced { .. } => crate::console::records::LineState::Quiesced,
        };
        (name.clone(), line_life(configured.durable), state)
    });
    let model = crate::console::TopologyModel::new(this, [], lines, [], []);
    crate::console::wall::WallBase::now()
        .and_then(|wall| crate::console::ConsoleHub::new(device, model, wall))
        .map_err(|error| ServerError::ListenerAccept {
            message: format!("console hub: {error}"),
        })
}

/// A roster entry's status as [`LiminalConnectionServices::channel_status`]
/// answers it.
fn status_of(configured: &ConfiguredChannel) -> ChannelStatus {
    let mode = configured.handle.config().mode;
    match configured.state() {
        ChannelState::Active => ChannelStatus::Active {
            origin: configured.origin,
            mode,
            schema: configured.protocol_schema,
        },
        ChannelState::Quiesced { reason } => ChannelStatus::Quiesced {
            reason,
            origin: configured.origin,
            mode,
        },
    }
}

#[cfg(test)]
thread_local! {
    /// Roster read locks taken on this thread. Test-only.
    static ROSTER_READS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}

/// Roster read locks taken on the calling thread so far. Test-only.
#[cfg(test)]
pub(crate) fn roster_reads() -> u64 {
    ROSTER_READS.with(std::cell::Cell::get)
}

/// A channel's line life: durable channels keep, ephemeral ones pass.
const fn line_life(durable: bool) -> crate::console::records::LineLife {
    if durable {
        crate::console::records::LineLife::Kept
    } else {
        crate::console::records::LineLife::Passing
    }
}

/// The diagnostic carried when the roster lock is poisoned. One string, shared
/// by every surface that reports it, so the wording cannot drift between them.
const ROSTER_POISONED: &str = "channel roster lock poisoned";

/// How many roster entries the registration cap counts.
///
/// Runtime-registered entries ONLY. Boot-configured channels are the operator's
/// own authored bound — they are in the file the operator wrote — and counting
/// them would make one number mean two different things depending on how the
/// deployment was configured. The origin never flips, so this population is
/// well defined over time.
fn runtime_registered_count(channels: &HashMap<String, Arc<ConfiguredChannel>>) -> usize {
    channels
        .values()
        .filter(|configured| configured.origin == ChannelOrigin::RuntimeRegistered)
        .count()
}

/// The three-field identity comparison behind idempotent-if-identical
/// registration.
///
/// All three must match; the FIRST mismatch is reported by name. The name itself
/// is not compared — it is the roster key, a precondition of the comparison
/// rather than a member of it — and neither are the supervisor or the durable
/// store, which are one server-wide instance each and cannot differ between two
/// registrations in one process.
///
/// The schema is compared as BOTH its protocol id and its parsed document, on
/// purpose. The id is a 64-bit non-cryptographic digest, so the document guards
/// against a collision accepting a different schema as identical; and the two
/// fail in opposite directions, because two byte sequences that parse to the
/// same document but differ in whitespace produce different ids — which must
/// refuse, since the id is what every future subscriber negotiates.
///
/// The channel's own `Schema` is not compared: it is not `PartialEq`, and its
/// identifier is a fresh value per construction, so comparing it would refuse
/// every idempotent re-registration.
fn compare_registration(
    existing: &ConfiguredChannel,
    spec: &ChannelRegistration,
    resolved: &ChannelSchema,
) -> Result<Registered, ChannelRegistryError> {
    let requested_mode = if spec.durable {
        ChannelMode::Durable
    } else {
        ChannelMode::Ephemeral
    };
    let config = existing.handle.config();
    let mismatch = if config.mode == requested_mode {
        if existing.protocol_schema == resolved.protocol_id {
            if *config.schema.definition() == resolved.document {
                None
            } else {
                Some(ChannelConfigField::SchemaDocument)
            }
        } else {
            Some(ChannelConfigField::SchemaId)
        }
    } else {
        Some(ChannelConfigField::Mode)
    };
    mismatch.map_or(Ok(Registered::AlreadyIdentical), |field| {
        Err(ChannelRegistryError::AlreadyRegistered {
            name: spec.name.clone(),
            field,
        })
    })
}

/// Recovers the resolved schema of an already-built entry, so the write-lock
/// re-check compares the SAME three fields against the same values the fast path
/// would have.
fn schema_of(configured: &ConfiguredChannel) -> ChannelSchema {
    ChannelSchema {
        document: configured.handle.config().schema.definition().clone(),
        protocol_id: configured.protocol_schema,
    }
}

/// Builds one roster entry: the channel's validation engine, its library handle
/// (durable or ephemeral), and the protocol schema id advertised at subscribe
/// time.
///
/// This is the SOLE place a [`ConfiguredChannel`] is constructed. Keeping
/// construction in one function is what makes "every channel on this server is
/// the same kind of object" structural rather than a promise: there is no second
/// body a channel could be built by, so no channel can drift into a different
/// shape. The boot loop in [`LiminalConnectionServices::from_config_with_store_via`]
/// is its caller.
///
/// `resolved` is consumed because its JSON Schema document is moved into the
/// channel's [`Schema`]; the protocol id is carried onto the entry unchanged.
/// `origin` is stamped here and never written again.
///
/// The failure is typed ([`ChannelBuildError`]) rather than a [`ServerError`] so
/// BOTH callers can render it without inspecting a message: the boot loop maps
/// it back to the exact `ConfigValidation` strings it has always produced, and
/// `register_channel` maps it to the registry's own typed variants.
///
/// # Errors
/// Returns [`ChannelBuildError`] when the JSON Schema document does not compile
/// or durable initialization over the shared store fails.
fn build_configured_channel(
    name: &str,
    resolved: ChannelSchema,
    durable: bool,
    origin: ChannelOrigin,
    durable_store: &Arc<dyn DurableStore>,
    supervisor: &ChannelSupervisor,
) -> Result<ConfiguredChannel, ChannelBuildError> {
    let schema =
        Schema::new(resolved.document).map_err(|error| ChannelBuildError::SchemaRejected {
            message: error.to_string(),
        })?;
    let channel_config = if durable {
        ChannelConfig::new(name.to_owned(), schema, ChannelMode::Durable)
    } else {
        ChannelConfig::new(name.to_owned(), schema, ChannelMode::Ephemeral)
    };
    let handle = if durable {
        ChannelHandle::new_durable_with_supervisor(
            channel_config,
            Arc::clone(durable_store),
            supervisor.clone(),
        )
        .map_err(|error| ChannelBuildError::DurableInitFailed {
            message: error.to_string(),
        })?
    } else {
        ChannelHandle::with_supervisor(channel_config, supervisor.clone())
    };
    Ok(ConfiguredChannel {
        handle,
        protocol_schema: resolved.protocol_id,
        durable,
        origin,
        state: AtomicU8::new(STATE_ACTIVE),
        quiesce_reason: OnceLock::new(),
    })
}

/// Returns the current epoch-millis timestamp used as the dedup entry anchor.
///
/// A clock error before the Unix epoch yields `0`: the timestamp is only a TTL
/// anchor for the in-memory cache and a zero anchor never breaks the at-most-once
/// claim semantics, so this avoids surfacing a clock fault on the publish path.
fn dedup_timestamp_millis() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
        .unwrap_or(0)
}

/// Default shard count for an on-disk durable store.
///
/// Haematite routes keys across this many single-threaded shard actors; a small
/// power of two gives parallelism across cursors/streams without spawning an
/// actor per core. The value is fixed (haematite has no silent default) and not
/// yet surfaced in server config.
const DEFAULT_SHARD_COUNT: usize = 8;

/// Shard count of the throwaway store an incarnation's ephemeral dedup
/// namespace lives in: it holds one namespace, and dies with the process.
const THROWAWAY_SHARD_COUNT: usize = 1;

/// Namespace prefix for the dedup-on-delivery cache streams. Keeps delivery dedup
/// keys from colliding with any other haematite streams in the shared store.
///
/// This is the **durable** namespace (A1 §5): stable across restarts, because a
/// durable channel's message outlives a restart too.
const DELIVERY_DEDUP_NAMESPACE: &str = "liminal:delivery-dedup";

/// The **ephemeral** dedup namespace for this server incarnation (A1 §5,
/// graft §0.2).
///
/// Scoped to the process incarnation so an ephemeral channel's receipt shares a
/// lifetime with the in-memory message it vouches for. Composed from the
/// process id and the boot instant, which is enough to make two incarnations
/// disagree; nothing reads the components back, so no meaning is attached to
/// them beyond distinctness.
///
/// The namespace lives in [`ephemeral_dedup_store`], never in a store that
/// outlives the process, so no incarnation's records are left behind for a
/// later one to read past.
fn ephemeral_dedup_namespace() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let boot_nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |since| since.as_nanos());
    format!(
        "{DELIVERY_DEDUP_NAMESPACE}:incarnation:{}-{boot_nanos}",
        std::process::id()
    )
}

/// One reader over the counters of `durable_store` and, when it is a separate
/// store, `ephemeral_store`; a store that is the same `Arc` is read once.
fn combined_store_counts(
    durable_store: &Arc<dyn DurableStore>,
    ephemeral_store: &Arc<dyn DurableStore>,
) -> StoreCountsHandle {
    let durable = durable_store.store_counts().unwrap_or_default();
    if Arc::ptr_eq(durable_store, ephemeral_store) {
        return durable;
    }
    durable.with(&ephemeral_store.store_counts().unwrap_or_default())
}

/// The store an incarnation's ephemeral dedup namespace is written to: the
/// services' own store when it was created empty for this process and dies
/// with it, otherwise a store of its own that does, opened by the namespace's
/// first write rather than at start.
///
/// Written into a persistent store, every incarnation's namespace would stay
/// there for good, under the durable namespace's prefix, with nothing that
/// ever reads it again to prune it.
fn ephemeral_dedup_store(
    durable_store: &Arc<dyn DurableStore>,
    history: StoreHistory,
    subsystems: &dyn SubsystemFactory,
) -> Arc<dyn DurableStore> {
    match (history, durable_store) {
        #[cfg(any(test, feature = "test-support"))]
        (StoreHistory::Fresh, store) => Arc::clone(store),
        (StoreHistory::Created | StoreHistory::Recorded, _) => Arc::new(CountingStore::new(
            DeferredStore::new(subsystems.throwaway_store()),
        )),
    }
}

/// Beamr-scheduler-owning subsystems the full-service construction path builds
/// beyond the connection supervisor's own scheduler. The §9 D2 seam census counts
/// these: the worker-front-door profile must construct NONE of them. Test-gated:
/// this is the recording vocabulary of the gate's instrument, not production state.
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(super) enum SchedulerSubsystem {
    /// The shared channel supervisor (its own beamr scheduler).
    ChannelSupervisor,
    /// The conversation supervisor (its own beamr scheduler).
    ConversationSupervisor,
    /// The haematite store's database (its shard-actor scheduler).
    HaematiteStore,
}

/// Constructor seam for every scheduler-owning subsystem (`SchedulerSubsystem`)
/// the profile-aware construction path can create.
///
/// These methods are the ONLY route through which [`build_connection_services`]
/// and the [`LiminalConnectionServices`] config constructors reach
/// `build_channel_cluster`, `ConversationSupervisor::new`, and the durable-store
/// constructors — no direct constructor call exists in those bodies. The §9 D2
/// gate therefore injects a factory that records as a side effect of
/// constructing (the D3 store-seam ownership move applied to schedulers): a
/// recording cannot be omitted without also failing to construct the subsystem,
/// which closes the "hand-placed census call beside the constructor" gap where a
/// future subsystem could be built without its courtesy call.
pub(super) trait SubsystemFactory {
    /// Constructs the shared channel supervisor + cluster resolver (a beamr
    /// scheduler).
    ///
    /// # Errors
    /// Returns [`ServerError`] when the channel supervisor scheduler cannot start.
    fn channel_cluster(
        &self,
        cluster_config: Option<&ClusterConfig>,
    ) -> Result<ChannelCluster, ServerError>;

    /// Constructs the conversation supervisor (a beamr scheduler).
    ///
    /// # Errors
    /// Returns [`ServerError`] when the conversation supervisor scheduler cannot
    /// start.
    fn conversation_supervisor(&self) -> Result<Arc<ConversationSupervisor>, ServerError>;

    /// Constructs the persistent durable store under the explicit directory.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the store cannot be opened.
    fn durable_store(&self, persistence_path: &Path) -> Result<Arc<dyn DurableStore>, ServerError>;

    /// The opener of the throwaway store an incarnation's ephemeral dedup
    /// namespace lives in when the durable store outlives the process. The
    /// store is opened by the namespace's first write, never by this call.
    fn throwaway_store(&self) -> ThrowawayStoreOpener;

    fn ephemeral_sweep_counts(&self) -> Option<EphemeralSweepCounts> {
        None
    }
}

/// Test construction factory for callers that own their fixture directories.
#[cfg(any(test, feature = "test-support"))]
pub(super) struct ProductionSubsystems;

#[cfg(any(test, feature = "test-support"))]
impl SubsystemFactory for ProductionSubsystems {
    fn channel_cluster(
        &self,
        cluster_config: Option<&ClusterConfig>,
    ) -> Result<ChannelCluster, ServerError> {
        build_channel_cluster(cluster_config)
    }

    fn conversation_supervisor(&self) -> Result<Arc<ConversationSupervisor>, ServerError> {
        Ok(Arc::new(ConversationSupervisor::new().map_err(
            |error| ServerError::ConfigValidation {
                message: format!("failed to start conversation supervisor: {error}"),
            },
        )?))
    }

    fn durable_store(&self, persistence_path: &Path) -> Result<Arc<dyn DurableStore>, ServerError> {
        build_durable_store(persistence_path)
    }

    fn throwaway_store(&self) -> ThrowawayStoreOpener {
        Box::new(|| open_ephemeral(THROWAWAY_SHARD_COUNT))
    }
}

struct RootedProductionSubsystems {
    root: Arc<EphemeralRoot>,
}

impl RootedProductionSubsystems {
    fn new(path: &Path) -> Result<Self, ServerError> {
        let root = EphemeralRoot::open(path).map_err(|error| ServerError::ConfigValidation {
            message: error.to_string(),
        })?;
        Ok(Self { root })
    }
}

impl SubsystemFactory for RootedProductionSubsystems {
    fn channel_cluster(
        &self,
        config: Option<&ClusterConfig>,
    ) -> Result<ChannelCluster, ServerError> {
        build_channel_cluster(config)
    }

    fn conversation_supervisor(&self) -> Result<Arc<ConversationSupervisor>, ServerError> {
        Ok(Arc::new(ConversationSupervisor::new().map_err(
            |error| ServerError::ConfigValidation {
                message: format!("failed to start conversation supervisor: {error}"),
            },
        )?))
    }

    fn durable_store(&self, path: &Path) -> Result<Arc<dyn DurableStore>, ServerError> {
        build_durable_store(path)
    }

    fn throwaway_store(&self) -> ThrowawayStoreOpener {
        let root = Arc::clone(&self.root);
        Box::new(move || root.open_store(THROWAWAY_SHARD_COUNT))
    }

    fn ephemeral_sweep_counts(&self) -> Option<EphemeralSweepCounts> {
        Some(self.root.sweep_counts())
    }
}

/// Test-only recording implementation of [`SubsystemFactory`] for the §9 D2
/// construction gate. Lives here (not in a test module's private scope) so the
/// supervisor-level gate test reuses the same instrument.
#[cfg(test)]
#[allow(clippy::expect_used)]
pub(super) mod subsystem_census {
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Mutex};

    use liminal::conversation::ConversationSupervisor;
    use liminal::durability::{DurableStore, open_ephemeral_rooted};

    use super::{
        ChannelCluster, ProductionSubsystems, SchedulerSubsystem, SubsystemFactory,
        THROWAWAY_SHARD_COUNT, ThrowawayStoreOpener, build_durable_store,
    };
    use crate::ServerError;
    use crate::config::types::ClusterConfig;

    /// Records each [`SchedulerSubsystem`] AS A SIDE EFFECT of constructing it,
    /// then hands back the production-constructed subsystem (with ephemeral
    /// stores rooted in an isolated directory, so the fs half of the gate is a
    /// real negative assertion). Because [`SubsystemFactory`] is the only route
    /// the profile-aware construction path has to these constructors, a
    /// recording cannot be omitted without also failing to construct — the
    /// record-by-construction guarantee.
    pub struct RecordingSubsystems {
        census: Mutex<Vec<SchedulerSubsystem>>,
        ephemeral_root: PathBuf,
    }

    impl RecordingSubsystems {
        /// A recording factory whose ephemeral stores live under `ephemeral_root`.
        pub fn rooted(ephemeral_root: &Path) -> Self {
            Self {
                census: Mutex::new(Vec::new()),
                ephemeral_root: ephemeral_root.to_path_buf(),
            }
        }

        /// The recorded construction census, sorted for order-independent
        /// comparison.
        pub fn recorded(&self) -> Vec<SchedulerSubsystem> {
            let mut recorded = self
                .census
                .lock()
                .expect("subsystem census lock is never poisoned in tests")
                .clone();
            recorded.sort();
            recorded
        }

        fn record(&self, subsystem: SchedulerSubsystem) {
            self.census
                .lock()
                .expect("subsystem census lock is never poisoned in tests")
                .push(subsystem);
        }
    }

    impl SubsystemFactory for RecordingSubsystems {
        fn channel_cluster(
            &self,
            cluster_config: Option<&ClusterConfig>,
        ) -> Result<ChannelCluster, ServerError> {
            let cluster = ProductionSubsystems.channel_cluster(cluster_config)?;
            self.record(SchedulerSubsystem::ChannelSupervisor);
            Ok(cluster)
        }

        fn conversation_supervisor(&self) -> Result<Arc<ConversationSupervisor>, ServerError> {
            let supervisor = ProductionSubsystems.conversation_supervisor()?;
            self.record(SchedulerSubsystem::ConversationSupervisor);
            Ok(supervisor)
        }

        fn durable_store(
            &self,
            persistence_path: &Path,
        ) -> Result<Arc<dyn DurableStore>, ServerError> {
            let store = build_durable_store(persistence_path)?;
            self.record(SchedulerSubsystem::HaematiteStore);
            Ok(store)
        }

        fn throwaway_store(&self) -> ThrowawayStoreOpener {
            let root = self.ephemeral_root.clone();
            Box::new(move || open_ephemeral_rooted(&root, THROWAWAY_SHARD_COUNT))
        }
    }
}

/// Full-only constructor guard: rejects a config whose profile is not `Full`.
///
/// [`LiminalConnectionServices`]' config-based constructors call this at entry so
/// the full service stack can never be built for a worker-front-door config —
/// profile enforcement holds on every public construction path, not only the
/// file-loading pipeline.
fn require_full_profile(config: &ServerConfig) -> Result<(), ServerError> {
    match config.services.profile()? {
        ServiceProfile::Full => Ok(()),
        ServiceProfile::WorkerFrontDoor => Err(ServerError::ConfigValidation {
            message: format!(
                "services.profile: \"{}\" cannot construct the full LiminalConnectionServices; \
                 build profile-selected services via build_connection_services",
                ServiceProfile::WORKER_FRONT_DOOR
            ),
        }),
    }
}

/// Builds the connection-services adapter selected by `config`'s service profile.
///
/// `Full` builds [`LiminalConnectionServices`] (channels, conversations, durable
/// store, dedup cache) exactly as today. `WorkerFrontDoor` builds
/// [`WorkerFrontDoorServices`], which constructs none of that machinery. This is the
/// single profile-dispatch authority: [`super::supervisor::ConnectionSupervisor`]'s
/// config constructor and the standalone runtime's worker arm both route through it
/// (the runtime's full arm stays on the explicit
/// [`LiminalConnectionServices::from_config`] path because it also needs the shared
/// channel cluster, which the trait object does not expose — that path is guarded
/// full-only at entry).
///
/// # Errors
/// Returns [`ServerError`] when the selected adapter cannot be constructed, the
/// configured profile value is not recognised, or the worker-front-door profile is
/// combined with full-only config fields.
pub fn build_connection_services(
    config: &ServerConfig,
) -> Result<Arc<dyn ConnectionServices>, ServerError> {
    match config.services.profile()? {
        ServiceProfile::Full => {
            let subsystems = RootedProductionSubsystems::new(required_persistence_path(config)?)?;
            build_connection_services_via(config, &subsystems)
        }
        ServiceProfile::WorkerFrontDoor => build_worker_front_door(config),
    }
}

/// [`build_connection_services`] with the subsystem factory injected — the §9 D2
/// gate seam, both halves at once.
///
/// Every scheduler-owning subsystem the `Full` branch creates is constructed
/// through `subsystems` and nowhere else, so a recording factory observes exactly
/// what was built (thread half, record-by-construction); the gate's factory also
/// roots its ephemeral stores in an isolated directory, so "the root stays empty
/// on the worker branch" is a real negative assertion (fs half, the D3 pattern).
/// The `WorkerFrontDoor` branch never touches the factory — a regression that gave
/// the front door any subsystem would both record in the census and land a store
/// directory in the injected root.
///
/// The worker branch re-runs the cross-field checks here, not only in file-loading
/// validation, so a directly-constructed config cannot smuggle full-only machinery
/// past the profile.
pub(super) fn build_connection_services_via(
    config: &ServerConfig,
    subsystems: &dyn SubsystemFactory,
) -> Result<Arc<dyn ConnectionServices>, ServerError> {
    match config.services.profile()? {
        ServiceProfile::Full => {
            let path = required_persistence_path(config)?;
            let history = StoreHistory::of(path);
            let store = subsystems.durable_store(path)?;
            Ok(Arc::new(
                LiminalConnectionServices::from_config_with_store_via(
                    config, store, history, subsystems,
                )?,
            ))
        }
        ServiceProfile::WorkerFrontDoor => build_worker_front_door(config),
    }
}

fn build_worker_front_door(
    config: &ServerConfig,
) -> Result<Arc<dyn ConnectionServices>, ServerError> {
    let errors = crate::config::validation::worker_front_door_field_errors(config);
    if !errors.is_empty() {
        return Err(ServerError::ConfigValidation {
            message: errors.join("; "),
        });
    }
    Ok(Arc::new(WorkerFrontDoorServices::new()))
}

/// Opens the persistent durable store under the configured directory.
pub(crate) fn open_persistent_store(
    persistence_path: &Path,
) -> Result<Arc<dyn DurableStore>, ServerError> {
    build_durable_store(persistence_path)
}

/// Every append is permanent through its WAL frames and one device flush
/// before it returns under either commit policy; what the policy decides is
/// when the shard folds its staged appends into a new tree version, the fold
/// that writes the leaf and every internal node above it as fresh files. Per
/// append, one keyed publish left 8 new node files (82,137 bytes) on the J3
/// store (NODE-FILES-PER-POST, check 15 on Dean at 68a08b35, 10 October
/// 2026), and the live spine reached 2,856,796 files that way. Buffered at
/// 1 MiB, Waffles' number (10 October 2026 11:19): the threshold haematite's
/// own `LeafLog` runs on, and the most an unclean restart replays from staged
/// frames per shard. The store's own reads merge the buffer, so a publish
/// reads back at once; a clean stop commits what is staged; no policy carries
/// a timer.
const SPINE_COMMIT_THRESHOLD_BYTES: NonZeroU64 = match NonZeroU64::new(1 << 20) {
    Some(threshold) => threshold,
    None => unreachable!(),
};

/// The commit policy the spine's event store opens under.
const SPINE_COMMIT_POLICY: CommitPolicy = CommitPolicy::buffered(SPINE_COMMIT_THRESHOLD_BYTES);

fn build_durable_store(path: &Path) -> Result<Arc<dyn DurableStore>, ServerError> {
    let data_dir = durable_data_dir(path);
    let database = open_or_create_database(&data_dir)?;
    let event_store = EventStore::with_commit_policy(database, SPINE_COMMIT_POLICY);
    Ok(Arc::new(CountingStore::new(HaematiteStore::new(Arc::new(
        event_store,
    )))))
}

/// The directory the persistent database under `persistence_path` lives in.
pub(super) fn durable_data_dir(persistence_path: &Path) -> PathBuf {
    persistence_path.join("durability")
}

/// Whether a haematite database already exists at `data_dir`: its
/// `config.json` is written when it is created.
pub(super) fn database_exists(data_dir: &Path) -> bool {
    database_config_file(data_dir).exists()
}

/// What creating a haematite database leaves in its directory. A shard's
/// directory appears beside them the first time anything opens the shard, so
/// a database holding nothing else has never been read or written.
const UNTOUCHED_DATABASE_ENTRIES: [&str; 2] = ["config.json", "writer.lock"];

/// Whether the database at `data_dir` holds only what its creation wrote.
/// Anything this cannot read, or does not recognise, answers `false`: the
/// database is then taken to hold what an earlier server wrote.
pub(super) fn database_untouched(data_dir: &Path) -> bool {
    std::fs::read_dir(data_dir).is_ok_and(|entries| {
        entries.into_iter().all(|entry| {
            entry.is_ok_and(|entry| {
                entry
                    .file_name()
                    .to_str()
                    .is_some_and(|name| UNTOUCHED_DATABASE_ENTRIES.contains(&name))
            })
        })
    })
}

fn database_config_file(data_dir: &Path) -> PathBuf {
    data_dir.join("config.json")
}

/// Every commit writes new copies of the nodes it changes; the durable store
/// deletes the copies its committed root no longer reaches while it runs, so it
/// holds its live records and not every version ever written. A store created
/// under `KeepAll` is moved onto this retention by its next open.
pub(super) const DURABLE_RETENTION: Retention = Retention::reclaim(0);

/// Opens an existing haematite database at `data_dir`, or creates one.
fn open_or_create_database(data_dir: &Path) -> Result<Database, ServerError> {
    let config_file = database_config_file(data_dir);
    let result = if config_file.exists() {
        adopt_missing_config_keys(data_dir, &config_file)?;
        Database::open_with_retention(
            data_dir,
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
                // DUR-021 R1: no pass starts and no step runs until the
                // listeners are bound and runtime.rs releases it.
                reclaim_start: haematite::ReclaimStart::Held,
            },
            DURABLE_RETENTION,
        )
    } else {
        Database::create(
            DatabaseConfig {
                data_dir: data_dir.to_path_buf(),
                shard_count: DEFAULT_SHARD_COUNT,
                distributed: None,
                executor_threads: None,
                // haematite 0.8.3 requires this field and refuses `None` at
                // validation; `Unlimited` is the crate's explicit spelling of the
                // pre-budget (0.8.1) behaviour — the behaviour-preserving choice
                // for a dependency hop. A real byte ceiling is a deployment
                // decision for its own measured lane. A store whose `config.json`
                // predates the field is migrated by the open arm above
                // (`adopt_missing_config_keys`) — same value, written once, on
                // upgrade.
                node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
                retention: DURABLE_RETENTION,
            },
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
                // DUR-021 R1: no pass starts and no step runs until the
                // listeners are bound and runtime.rs releases it.
                reclaim_start: haematite::ReclaimStart::Held,
            },
        )
    };
    let database = result.map_err(|error| ServerError::ConfigValidation {
        message: format!(
            "failed to open durable store at {}: {error}",
            data_dir.display()
        ),
    })?;
    liminal::durability::record_store_opened();
    super::reclaim_placement::arm_watch(&database);
    Ok(database)
}

/// The stored keys this crate's create arm always writes, with the value it
/// writes, for a `config.json` written before haematite had the key.
///
/// - `node_cache_budget`: haematite 0.8.3 refuses a store without it
///   (`MissingNodeCacheBudget`); `"unlimited"` is the pre-budget behaviour.
/// - `retention`: `"keep_all"` is the 0.14.1 behaviour, written in so the
///   store's stored retention is explicit before the open. The open itself is
///   [`Database::open_with_retention`] under [`DURABLE_RETENTION`], which then
///   records reclaim in place of it, so an adopted store reclaims from that
///   same open onward (spine reclaim, Hermes's ruling of 6 October 2026:
///   `past_roots` 0).
const ADOPTED_CONFIG_KEYS: [(&str, &str); 2] = [
    ("node_cache_budget", "unlimited"),
    ("retention", "keep_all"),
];

/// Writes the values of [`ADOPTED_CONFIG_KEYS`] into a `config.json` that
/// lacks any of them, so a store created before haematite had the key opens on
/// upgrade with no operator hand. The open that follows then moves its
/// retention onto [`DURABLE_RETENTION`].
///
/// This crate IS the caller that owns these decisions, and it already makes
/// them on every create, so answering them here, once, on upgrade, changes
/// nothing about how the store runs and takes the file off the operator's
/// desk (the store migrates itself or it doesn't ship).
///
/// Touches the file in exactly one case: a readable JSON object missing one of
/// those keys. Anything else — unreadable bytes, a non-object, an explicit
/// value however wrong — is left byte-identical so [`Database::open`] surfaces
/// haematite's own sharpest refusal instead of this function guessing over
/// it. The write is atomic and durable (tmp file beside the target, fsync,
/// rename, directory fsync): a crash mid-migration leaves the original bytes,
/// and the next boot migrates again.
fn adopt_missing_config_keys(data_dir: &Path, config_file: &Path) -> Result<(), ServerError> {
    let Ok(bytes) = std::fs::read(config_file) else {
        return Ok(());
    };
    let Ok(mut config) = serde_json::from_slice::<serde_json::Value>(&bytes) else {
        return Ok(());
    };
    let Some(object) = config.as_object_mut() else {
        return Ok(());
    };
    let mut adopted = Vec::new();
    for (key, value) in ADOPTED_CONFIG_KEYS {
        if !object.contains_key(key) {
            object.insert(key.to_owned(), serde_json::Value::String(value.to_owned()));
            adopted.push(key);
        }
    }
    if adopted.is_empty() {
        return Ok(());
    }
    let keys = adopted.join(", ");
    let migrated =
        serde_json::to_vec_pretty(&config).map_err(|error| ServerError::ConfigValidation {
            message: format!(
                "durable store config at {} predates `{keys}` and its migration could not be \
                 serialized: {error}",
                config_file.display()
            ),
        })?;
    let write_error = |step: &str, error: std::io::Error| ServerError::ConfigValidation {
        message: format!(
            "durable store config at {} predates `{keys}` and migrating it failed at {step}: \
             {error}",
            config_file.display()
        ),
    };
    let tmp_file = data_dir.join("config.json.migrate-tmp");
    let mut tmp = std::fs::File::create(&tmp_file)
        .map_err(|error| write_error("creating the temporary file", error))?;
    std::io::Write::write_all(&mut tmp, &migrated)
        .map_err(|error| write_error("writing the temporary file", error))?;
    tmp.sync_all()
        .map_err(|error| write_error("syncing the temporary file", error))?;
    drop(tmp);
    std::fs::rename(&tmp_file, config_file)
        .map_err(|error| write_error("renaming over config.json", error))?;
    std::fs::File::open(data_dir)
        .and_then(|dir| dir.sync_all())
        .map_err(|error| write_error("syncing the data directory", error))?;
    tracing::warn!(
        config = %config_file.display(),
        keys = %keys,
        "durable store config predates these keys -- wrote the create-arm values (the \
         behaviour-preserving ones) so the store opens with no operator action"
    );
    Ok(())
}

pub(super) fn reserved_namespace_refusal(channel: &str) -> Option<ServerError> {
    if channel.starts_with("liminal.") {
        crate::metrics::self_stream_reserved_publication();
        Some(ServerError::ReservedNamespace {
            channel: channel.to_owned(),
        })
    } else {
        None
    }
}

impl LiminalConnectionServices {
    pub(in crate::server) const fn signals(&self) -> &ServerSignals {
        &self.signals
    }

    pub(in crate::server) fn self_stream_head(&self, name: &str) -> Result<u64, ServerError> {
        let channels = self.read_channels()?;
        let channel = channels
            .get(name)
            .ok_or_else(|| ServerError::ListenerAccept {
                message: format!("self stream '{name}' is absent after append"),
            })?;
        let heads =
            channel
                .handle
                .durable_positions()
                .map_err(|error| ServerError::ListenerAccept {
                    message: format!("self stream '{name}' head is unavailable: {error}"),
                })?;
        drop(channels);
        match heads.as_slice() {
            [head] => Ok(*head),
            _ => Err(ServerError::ListenerAccept {
                message: format!("self stream '{name}' must have exactly one durable partition"),
            }),
        }
    }

    pub(in crate::server) fn publish_self_stream_batch(
        &self,
        payloads: &[Vec<u8>],
        publisher: &str,
    ) -> Result<u64, ServerError> {
        if publisher.trim().is_empty() || payloads.is_empty() {
            return Err(ServerError::ConfigValidation {
                message: "self-stream batch requires a publisher and rows".to_owned(),
            });
        }
        let configured = self
            .admit_channel("liminal.events")
            .map_err(|error| access_to_server_error(&error))?;
        let deliveries = configured
            .handle
            .publish_batch_with_delivery(
                payloads,
                &liminal::envelope::PublisherId::new(publisher),
                None,
            )
            .map_err(|error| match error {
                liminal::error::LiminalError::PublishUnresolved { append } => {
                    ServerError::PublishUnresolved {
                        channel: "liminal.events".to_owned(),
                        idempotency_key: None,
                        sequence: Some(append.sequence()),
                        message: append.to_string(),
                    }
                }
                error => ServerError::ListenerAccept {
                    message: format!("self-stream receipt batch failed: {error}"),
                },
            })?;
        if deliveries.len() != payloads.len()
            || deliveries
                .iter()
                .any(|delivery| delivery.durable_position().is_none())
        {
            return Err(ServerError::ListenerAccept {
                message: "self-stream receipt batch was refused before append".to_owned(),
            });
        }
        for pair in deliveries.windows(2) {
            if pair[0]
                .durable_position()
                .and_then(|position| position.checked_add(1))
                != pair[1].durable_position()
            {
                return Err(ServerError::ListenerAccept {
                    message: "self-stream receipt batch positions are not contiguous".to_owned(),
                });
            }
        }
        for delivery in &deliveries {
            self.delivered(delivery);
        }
        deliveries
            .last()
            .and_then(ChannelDelivery::durable_position)
            .and_then(|position| position.checked_add(1))
            .ok_or_else(|| ServerError::ListenerAccept {
                message: "self-stream receipt batch head is unavailable".to_owned(),
            })
    }

    pub(in crate::server) fn checkpoint_self_stream_heads(
        &self,
        heads: [u64; 2],
    ) -> Result<(), ServerError> {
        for (name, head) in ["liminal.events:0", "liminal.cost:0"]
            .into_iter()
            .zip(heads)
        {
            block_on(liminal::durability::head_checkpoint::advance_head(
                self.durable_store.as_ref(),
                name,
                head,
            ))
            .map_err(|error| ServerError::ShutdownFlush {
                message: format!("self-stream head checkpoint bridge for '{name}' failed: {error}"),
            })?
            .map_err(|error| ServerError::ShutdownFlush {
                message: format!("self-stream head checkpoint for '{name}' failed: {error}"),
            })?;
        }
        Ok(())
    }

    pub(in crate::server) fn install_self_stream_channels(&self) -> Result<(), ServerError> {
        let names = ["liminal.cost", "liminal.events"];
        {
            let channels = self.read_channels()?;
            for name in names {
                if channels.contains_key(name) {
                    return Err(ServerError::ConfigValidation {
                        message: format!("self-stream channel '{name}' is already configured"),
                    });
                }
            }
        }
        let mut built = Vec::with_capacity(names.len());
        for name in names {
            let schema =
                resolve_schema_bytes(None).map_err(|error| ServerError::ConfigValidation {
                    message: format!("self-stream channel '{name}' schema failed: {error}"),
                })?;
            let channel = build_configured_channel(
                name,
                schema,
                true,
                ChannelOrigin::BootConfigured,
                &self.durable_store,
                self.cluster.supervisor(),
            )
            .map_err(|error| ServerError::ConfigValidation {
                message: error.boot_message(name),
            })?;
            built.push((name.to_owned(), Arc::new(channel)));
        }
        let mut channels =
            self.channels
                .write()
                .map_err(|error| ServerError::ConfigValidation {
                    message: format!("self-stream channel roster lock poisoned: {error}"),
                })?;
        for (name, _) in &built {
            if channels.contains_key(name) {
                return Err(ServerError::ConfigValidation {
                    message: format!("self-stream channel '{name}' was concurrently configured"),
                });
            }
        }
        channels.extend(built);
        drop(channels);
        // SRV-012 R7: every roster channel but `console.*` is a line, and
        // these join after the hub's model was built from the roster.
        for name in names {
            self.console_changed(crate::console::TopologyChange::Channel {
                name: name.to_owned(),
                life: line_life(true),
                state: crate::console::records::LineState::Live,
            });
        }
        Ok(())
    }

    /// SRV-012 R4: registers `console.topology` as an ephemeral,
    /// server-owned channel (never a store write). A channel of that name
    /// already configured (ephemeral, since config refuses a durable one) is
    /// kept as it is.
    pub(in crate::server) fn install_console_channels(&self) -> Result<(), ServerError> {
        self.register_console_channel(crate::console::TOPOLOGY_SUBJECT)
            .map_err(|error| ServerError::ConfigValidation {
                message: format!("console channel registration failed: {error}"),
            })
    }

    /// Puts the ephemeral console channel `name` on the roster unless it is
    /// there.
    fn register_console_channel(&self, name: &str) -> Result<(), ChannelRegistryError> {
        if self.read_roster()?.contains_key(name) {
            return Ok(());
        }
        let schema = resolve_schema_bytes(None).map_err(|error| {
            ChannelRegistryError::RosterUnavailable {
                message: format!("console channel '{name}' schema failed: {error}"),
            }
        })?;
        let channel = build_configured_channel(
            name,
            schema,
            false,
            ChannelOrigin::BootConfigured,
            &self.durable_store,
            self.cluster.supervisor(),
        )
        .map_err(|error| error.into_registry_error(name))?;
        self.write_roster()?
            .entry(name.to_owned())
            .or_insert_with(|| Arc::new(channel));
        Ok(())
    }

    /// Publishes on an ordinary channel for an application embedding the server.
    ///
    /// The live and durable publisher identity is liminal.<host>. Reserved self
    /// streams and console channels retain the same refusal as connection publishes.
    /// An idempotency key has the same delivery semantics as a connection publish.
    ///
    /// # Errors
    ///
    /// Returns a configuration error for an empty or whitespace-only host name,
    /// a namespace refusal for a reserved channel, or the existing admission,
    /// durability, and publication error.
    pub fn publish_embedded(
        &self,
        host: &str,
        channel: &str,
        envelope: &MessageEnvelope,
        idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        self.publish_with_origin(
            channel,
            envelope,
            idempotency_key,
            PublishOrigin::Host { name: host },
        )
        .map(|published| published.outcome)
    }

    pub(in crate::server) fn publish_with_origin(
        &self,
        channel: &str,
        envelope: &MessageEnvelope,
        idempotency_key: Option<&str>,
        origin: PublishOrigin<'_>,
    ) -> Result<OriginPublication, ServerError> {
        validate_publication_origin(channel, idempotency_key, origin)?;
        // SRV-012 to_stations: only while a console tap is live does the
        // fan-out record whom the message reached, in a buffer the tap lends
        // and takes back; otherwise the publish is the plain one.
        let mut recipients = self.signals.lend_recipients();
        // Lane brief ruling 6: a tapped message is first seen as the server
        // receives it, before admission; the clock is read only while a tap
        // lent a record.
        let received_at = recipients.is_some().then(Instant::now);
        let result = (|| {
            // The roster read holds its guard only long enough to clone the entry's
            // `Arc` out; everything below — the dedup bridge and the publish into the
            // channel actor — runs with no roster lock held.
            //
            // This inner admission STAYS even once the connection process consults
            // the roster ahead of delegating: this method is public and callable
            // without going through a frame at all, and a guard that only exists in
            // the caller is not a guard.
            let configured = self
                .admit_channel(channel)
                .map_err(|error| access_to_server_error(&error))?;

            // Dedup-on-delivery: a publish carrying an idempotency key is delivered to
            // subscribers AT MOST ONCE across re-publishes of the same key. Only a
            // fresh `Claimed` decision proceeds to fan-out; a `Completed`/`InFlight`
            // decision is a duplicate and is suppressed (no second delivery), which is
            // the at-most-once guarantee the aion outbox relies on.
            // A1 §5: the namespace is chosen by the channel's DURABILITY, so an
            // ephemeral receipt cannot outlive the message it vouches for.
            // A keyed publish to a durable channel claims its key WITH the channel
            // as evidence and holds the claim across an unresolved append
            // (`unresolved.rs`); the ephemeral namespace keeps the claim below.
            if let (Some(key), ChannelMode::Durable) =
                (idempotency_key, configured.handle.config().mode)
            {
                return self
                    .publish_durable_keyed(
                        &configured,
                        channel,
                        (envelope, origin),
                        key,
                        recipients.as_mut(),
                    )
                    .map(OriginPublication::connection);
            }
            let dedup = self.dedup_for(&configured);
            if let Some(key) = idempotency_key {
                if !Self::claim_delivery(dedup, key)? {
                    // A dedup-suppressed re-publish is still an accepted publish (it is
                    // assigned a message id), but it reaches no subscriber, so it counts
                    // toward publishes and not deliveries.
                    crate::metrics::publish_accepted();
                    return Ok(OriginPublication::connection(PublishOutcome {
                        message_id: self.next_message_id(),
                        delivered: false,
                        durable_position: None,
                        reach: Reach::Suppressed,
                    }));
                }
            }

            let delivery = fan_out(&configured, envelope, origin, recipients.as_mut());
            let delivery = match delivery {
                Ok(delivery) => delivery,
                // An unkeyed durable publish whose append outcome is not known is
                // answered as unresolved, never as a definite failure a producer
                // would retry into a second row.
                Err(liminal::error::LiminalError::PublishUnresolved { append }) => {
                    return Err(ServerError::PublishUnresolved {
                        channel: channel.to_owned(),
                        idempotency_key: None,
                        sequence: Some(append.sequence()),
                        message: append.to_string(),
                    });
                }
                Err(error) => {
                    // The claim above appended an `InFlight` entry but the delivery
                    // failed before `complete_receipt` could run. Release the claim so
                    // the key is re-claimable; otherwise every re-publish would see
                    // `InFlight` and be suppressed forever. Best-effort: surface the
                    // ORIGINAL publish error regardless, but never swallow a release
                    // failure silently (it leaves the leak intact).
                    if let Some(key) = idempotency_key {
                        Self::release_claim(dedup, key);
                    }
                    return Err(ServerError::ListenerAccept {
                        message: format!("liminal publish failed for channel '{channel}': {error}"),
                    });
                }
            };

            // A1 §5, the two-way dedup mapping. Every publish outcome maps to
            // EXACTLY ONE of complete/release — the same shape as the C1 contract:
            //
            // * **Defer (or Accept) => `complete_receipt`.** The message is
            //   admitted and bus-held, so the key is done: a producer retrying it
            //   hits `Completed` and there is no duplicate delivery and no InFlight
            //   leak. This is the path that makes "re-publishing with the same key
            //   is safe and is suppressed while the bus owns the message" true.
            // * **Reject => `release_claim`.** Aggregation makes a producer-visible
            //   Reject mean delivered-to-zero, so the correct dedup state is "as if
            //   never published": tombstone it so a deliberate retry can re-claim.
            //   `release_claim` never clobbers a receipt, so the at-most-once guard
            //   survives a concurrent completion racing ahead.
            //
            // Mixed accept/reject reports Defer and so completes — consistent with
            // "at least one delivery happened".
            if let Some(key) = idempotency_key {
                if delivery.is_admitted() {
                    block_on(dedup.complete_receipt(key, ProcessingReceipt::new(Vec::new())))
                        .map_err(|error| ServerError::ListenerAccept {
                            message: format!(
                                "dedup receipt bridge failed for key '{key}': {error}"
                            ),
                        })?
                        .map_err(|error| ServerError::ListenerAccept {
                            message: format!("dedup receipt write failed for key '{key}': {error}"),
                        })?;
                } else {
                    Self::release_claim(dedup, key);
                }
            }

            let self_row_at = delivery.durable_position().filter(|_| origin.is_server());
            Ok(OriginPublication {
                outcome: self.delivered(&delivery),
                self_row_appended: self_row_at.is_some(),
                self_row_at,
            })
        })();
        self.signals.published(
            channel,
            origin,
            envelope,
            &result,
            recipients.as_ref().zip(received_at),
        );
        if let Some(buffer) = recipients {
            self.signals.give_back(buffer);
        }
        result
    }
}

/// The library publish of an unkeyed message, recording what it saw in
/// `recipients` when the tap lent a record.
fn fan_out(
    configured: &ConfiguredChannel,
    envelope: &MessageEnvelope,
    origin: PublishOrigin<'_>,
    recipients: Option<&mut liminal::channel::PublishRecord>,
) -> Result<liminal::channel::ChannelDelivery, liminal::error::LiminalError> {
    let publisher = || publisher_for_origin(origin);
    recipients.map_or_else(
        || {
            configured
                .handle
                .publish_with_delivery(&envelope.payload, publisher(), None)
        },
        |recipients| {
            configured.handle.publish_recording(
                &envelope.payload,
                publisher(),
                None,
                None,
                recipients,
            )
        },
    )
}

fn validate_publication_origin(
    channel: &str,
    idempotency_key: Option<&str>,
    origin: PublishOrigin<'_>,
) -> Result<(), ServerError> {
    if !origin.is_server() {
        refuse_console_namespace(channel).map_err(|error| access_to_server_error(&error))?;
        if let Some(error) = reserved_namespace_refusal(channel) {
            return Err(error);
        }
    }
    if let PublishOrigin::Host { name } = origin
        && name.trim().is_empty()
    {
        return Err(ServerError::ConfigValidation {
            message: "embedded publication requires a nonempty host name".to_owned(),
        });
    }
    if let PublishOrigin::Server { device } = origin {
        if device.trim().is_empty() || idempotency_key.is_some() {
            return Err(ServerError::ConfigValidation {
                message: "server-origin publication requires a device and no idempotency key"
                    .to_owned(),
            });
        }
    }
    Ok(())
}

fn publisher_for_origin(origin: PublishOrigin<'_>) -> liminal::envelope::PublisherId {
    match origin {
        PublishOrigin::Connection { .. } => liminal::envelope::PublisherId::default(),
        PublishOrigin::Host { name } => {
            liminal::envelope::PublisherId::new(format!("liminal.{name}"))
        }
        PublishOrigin::Server { device } => {
            liminal::envelope::PublisherId::new(format!("liminal.{device}"))
        }
    }
}

impl ConnectionServices for LiminalConnectionServices {
    fn participant_service(&self) -> Option<InstalledParticipantService> {
        self.participant_service.clone()
    }

    /// The roster's admission decision, exposed across the trait boundary as the
    /// permission alone.
    ///
    /// Two methods share the name `admit_channel` on this type: this trait
    /// method, and the private inherent funnel it delegates to. They are the
    /// same decision at two different boundaries — the funnel returns the ENTRY
    /// the decision was made on, because its in-crate callers go on to publish
    /// or subscribe through it, and this one returns `()`, because a caller
    /// outside the crate is asking whether it MAY, not for the thing itself.
    /// Handing the entry out here would make [`ConfiguredChannel`] public
    /// surface and hand a frame-level caller a channel handle it has no business
    /// holding. Rust resolves an inherent method ahead of a trait method of the
    /// same name, so the funnel's existing callers — and the `Self::` call below
    /// — reach the funnel, not this method; the trait form is reached only
    /// through a `dyn ConnectionServices`, which is exactly the caller it exists
    /// for. `an_admitted_operation_the_service_then_refuses_stays_undifferentiated`
    /// is the instrument that would catch this delegation turning into a
    /// recursion.
    ///
    /// `operation` is where the answers part: absence and quiesce refuse
    /// publishes and subscribes alike, but the console namespace refuses only
    /// a connection's publish (a console reads `console.*` as an ordinary
    /// subscriber, and only the server may write there), and a tap target
    /// refuses only a subscribe. Both are checked before the roster, so a
    /// client is told the same thing whether or not the subject is registered
    /// yet.
    fn admit_channel(
        &self,
        operation: ChannelOperation,
        channel: &str,
    ) -> Result<(), ChannelAccessError> {
        match operation {
            ChannelOperation::Publish => refuse_console_namespace(channel)?,
            ChannelOperation::Subscribe => {
                refuse_console_tap_target(channel)?;
                self.refuse_console_tap_limit(channel)?;
                // SRV-012 R4/R6: a tap channel exists while it has a
                // subscriber, and only while the console streams run.
                let admitted = if self.signals.console_enabled()
                    && super::channel_registry::console_tap_target(channel).is_some()
                {
                    self.register_console_channel(channel).map_err(|error| {
                        ChannelAccessError::RosterUnavailable {
                            message: error.to_string(),
                        }
                    })
                } else {
                    Ok(())
                }
                .and_then(|()| Self::admit_channel(self, channel).map(|_| ()));
                if admitted.is_err() {
                    self.give_back_tap(channel);
                }
                return admitted;
            }
        }
        Self::admit_channel(self, channel)?;
        Ok(())
    }

    fn publish(
        &self,
        connection: u64,
        channel: &str,
        envelope: &MessageEnvelope,
        idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        // The frame path refused this already at admission. This guard is for
        // a caller holding the trait directly: a guard that only exists in the
        // caller is not a guard.
        refuse_console_namespace(channel).map_err(|error| access_to_server_error(&error))?;
        self.publish_with_origin(
            channel,
            envelope,
            idempotency_key,
            PublishOrigin::Connection { id: connection },
        )
        .map(|published| published.outcome)
    }

    fn subscribe(
        &self,
        channel: &str,
        accepted_schemas: &[ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        let subscribed = self.open_subscription(channel, accepted_schemas, install);
        if subscribed.is_err() {
            // SRV-012 R6: a tap slot reserved at admission for a subscribe
            // that did not happen is given back, and a tap channel it
            // registered with no tap behind it leaves the roster.
            self.give_back_tap(channel);
        }
        subscribed
    }

    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError> {
        subscription.unsubscribe()
    }

    fn open_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        // Spawn a REAL participant process (a beamr `NativeHandler` running the
        // resolved responder behaviour) on the conversation supervisor's
        // scheduler, and a supervised conversation actor linked to it. The actor
        // FORWARDS each conversation message to the participant, which genuinely
        // processes it and delivers a reply back. The actor traps the
        // participant's EXIT (a beamr process link), so killing it fires a
        // structural, microsecond-scale crash signal.
        //
        // The responder is chosen by `subject`: a custom responder registered via
        // `register_responder` for this subject, or the built-in `EchoBehaviour`
        // when none is registered. Either way it runs as the SAME supervised,
        // linked participant process — the seam changes WHO responds, not HOW the
        // participant is spawned or supervised.
        let behaviour = self.responder_for(subject)?;
        let (actor, participant) = self
            .conversation_supervisor
            .spawn_with_participant(behaviour, None, ChannelMode::Ephemeral, CrashPolicy::Fail)
            .map_err(|error| ServerError::ListenerAccept {
                message: format!(
                    "failed to spawn supervised conversation {conversation_id} ('{subject}'): {error}"
                ),
            })?;

        // Drive boot to completion so the beamr link to the participant exists
        // before any message is forwarded (link-before-forward), mirroring the
        // ROUTING-004 dispatch pattern.
        actor.pid().map_err(|error| ServerError::ListenerAccept {
            message: format!(
                "failed to boot supervised conversation {conversation_id} ('{subject}'): {error}"
            ),
        })?;

        // Register the structural EXIT notifier BEFORE returning, so a crash that
        // fires the instant a message reaches the participant is never missed.
        // The notifier is woken by the actor's trapped-EXIT handler (event
        // driven), and a crash that already landed is replayed immediately.
        let (exit_tx, exit_rx) = mpsc::sync_channel::<Instant>(1);
        actor
            .notify_on_participant_exit(participant, exit_tx)
            .map_err(|error| ServerError::ListenerAccept {
                message: format!(
                    "failed to arm crash detection for conversation {conversation_id}: {error}"
                ),
            })?;

        Ok(ConnectionConversation::new(
            Box::new(LiminalConversationResource::new(
                actor,
                participant,
                exit_rx,
            )),
            subject.to_owned(),
        ))
    }

    fn conversation_message(
        &self,
        conversation: &ConnectionConversation,
        envelope: &MessageEnvelope,
        op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        conversation.message(envelope, op_id)
    }

    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError> {
        conversation.close()
    }

    fn listening(self: Arc<Self>) -> Option<std::thread::JoinHandle<()>> {
        self.resolve_claims_behind_listener()
    }

    fn stop_participant_warm(&self) {
        // DUR-020 R5: the warm stops at its next conversation boundary and
        // the stop does not wait for it; whatever it had not restored
        // restores on the next start. Then the clean stop records the
        // capacity snapshot (R3), and the final flush makes it durable.
        if let Some(warm) = &self.participant_warm {
            warm.stop();
            warm.record_clean_stop();
        }
    }

    fn flush_durable_state(&self) -> Result<(), ServerError> {
        // A flush observer publishes to a durable channel, so no channel append lock can be held.
        // Separate ephemeral dedup state dies with this incarnation; shared dedup state is covered here.
        self.signals.with_flush_barrier(|| {
            block_on(self.durable_store.flush())
                .map_err(|error| ServerError::ShutdownFlush {
                    message: format!("shared durable store flush bridge failed: {error}"),
                })?
                .map_err(|error| ServerError::ShutdownFlush {
                    message: format!("shared durable store flush failed: {error}"),
                })?;
            Ok(())
        })?;
        Ok(())
    }
}

/// One roster entry: a channel's library handle, the protocol schema id
/// advertised for it, where it came from, and its admission state.
#[derive(Debug)]
pub(super) struct ConfiguredChannel {
    handle: ChannelHandle,
    protocol_schema: ProtocolSchemaId,
    /// Durable (stored) or ephemeral, as built; the console's line life.
    durable: bool,
    /// Boot-configured or runtime-registered. Written once at construction and
    /// NEVER flipped: it is the only field that predicts what a restart does to
    /// this entry, and it defines the population the registration cap counts.
    origin: ChannelOrigin,
    /// [`STATE_ACTIVE`] or [`STATE_QUIESCED`]. Written at most once, by one
    /// `compare_exchange`, so the transition is one-way and exactly one caller
    /// can perform it.
    state: AtomicU8,
    /// The quiesce cause, set STRICTLY BEFORE `state` flips so any reader that
    /// observes [`STATE_QUIESCED`] can read it. Written once (the `OnceLock`
    /// enforces that structurally), which is why a re-quiesce under a different
    /// reason must refuse rather than silently keep or replace one of them.
    quiesce_reason: OnceLock<String>,
}

impl ConfiguredChannel {
    /// The entry's admission state as a value.
    ///
    /// One `Acquire` load and, when quiesced, one read of the already-written
    /// reason. Touches no actor: the whole point of recording state on the entry
    /// is that observing it cannot spawn the thing being observed.
    fn state(&self) -> ChannelState {
        if self.state.load(Ordering::Acquire) == STATE_QUIESCED {
            ChannelState::Quiesced {
                reason: self.recorded_quiesce_reason(),
            }
        } else {
            ChannelState::Active
        }
    }

    /// The recorded quiesce reason, for a caller that has already observed
    /// [`STATE_QUIESCED`] with an `Acquire` load.
    fn recorded_quiesce_reason(&self) -> String {
        self.quiesce_reason
            .get()
            .map_or_else(|| UNRECORDED_QUIESCE_REASON.to_owned(), Clone::clone)
    }

    /// Moves this entry from active to quiesced, recording `reason` first.
    ///
    /// The reason is written to the `OnceLock` BEFORE the `Release`
    /// `compare_exchange` that flips the state, so no reader can observe
    /// `QUIESCED` without being able to read why. Re-quiescing under the
    /// IDENTICAL reason is `Ok(())` — the caller's intent already holds — and a
    /// DIFFERENT reason refuses, because the recorded one cannot be replaced
    /// without losing a cause and cannot be kept without lying to the caller.
    fn quiesce(&self, name: &str, reason: &str) -> Result<(), ChannelRegistryError> {
        // Whoever wins the `OnceLock` writes the cause on record; every other
        // caller — a re-quiesce or a concurrent racer — is judged against THAT
        // reason, never against whether it happened to perform the flip itself.
        // Reporting success for having won the CAS would tell a racer whose
        // reason lost that its reason took effect.
        let recorded = match self.quiesce_reason.set(reason.to_owned()) {
            Ok(()) => reason.to_owned(),
            Err(_rejected) => self.recorded_quiesce_reason(),
        };
        // The reason is now readable, so the flip may become visible. The CAS
        // result is deliberately unused: the state is `QUIESCED` afterwards
        // whether this call or a racer performed the transition, and the answer
        // to the caller is governed by the recorded reason above.
        let _flipped_here = self.state.compare_exchange(
            STATE_ACTIVE,
            STATE_QUIESCED,
            Ordering::Release,
            Ordering::Acquire,
        );
        if recorded == reason {
            return Ok(());
        }
        Err(ChannelRegistryError::AlreadyQuiesced {
            name: name.to_owned(),
            reason: recorded,
        })
    }
}

#[derive(Debug)]
struct LiminalSubscriptionResource {
    subscription: liminal::channel::SubscriptionHandle,
}

impl SubscriptionResource for LiminalSubscriptionResource {
    fn unsubscribe(self: Box<Self>) -> Result<(), ServerError> {
        drop(self.subscription);
        Ok(())
    }

    fn is_overflowed(&self) -> bool {
        self.subscription.is_overflowed()
    }

    fn subscriber_pid(&self) -> Option<u64> {
        Some(self.subscription.subscriber_pid())
    }

    fn has_pending(&self) -> bool {
        self.subscription.has_pending()
    }

    fn try_next(&mut self) -> Option<liminal::envelope::Envelope> {
        match self.subscription.try_next() {
            Ok(envelope) => envelope,
            Err(error) => {
                // A poisoned inbox lock is PERMANENT, not transient: once poisoned it
                // stays poisoned, so every future `try_next` also returns `Err` and this
                // subscription goes silent for the rest of its life — no further
                // deliveries, not "held for the next slice". Poisoning requires a panic
                // while the lock is held, which the workspace lints forbid
                // (no unwrap/expect/panic), so this is an accepted low-probability
                // failure rather than a recoverable one. We keep the connection alive (a
                // single permanently-silent subscription is less harmful than tearing
                // down every other subscription and stream the connection multiplexes)
                // but log loudly so the silence is diagnosable. The log cannot storm:
                // it can only fire after the one panic that poisoned the lock.
                tracing::error!(
                    %error,
                    "subscription inbox lock is poisoned; this subscription is now \
                     permanently silent and will deliver no further messages"
                );
                None
            }
        }
    }
}

/// Renders an admission refusal as the service error the trait's callers expect.
///
/// The two refusals that already existed keep their EXACT bytes: an absent
/// channel is still `channel '<name>' is not configured` and a poisoned roster
/// is still `channel roster lock poisoned`, so nothing that reads these messages
/// today sees a change. Discrimination lives in the reason code
/// ([`ChannelAccessError::reason_code`]), which the connection process carries to
/// the wire — this rendering is the degraded path for a caller that reaches the
/// service directly and has no code to carry.
///
/// `Quiesced` has no predecessor string to preserve; it carries the refusal's own
/// message, reason included.
///
/// The connection process renders its admission refusals through this SAME
/// function rather than through [`ChannelAccessError`]'s own `Display`. Two
/// consequences, both wanted. The bytes on the wire for an absent channel stay
/// exactly what they have always been, which is the whole of the semver promise
/// this lane makes. And an operation refused AT admission and one refused after
/// admission by the service produce the identical message, differing only in the
/// reason code — so the code really is the only discriminator, instead of the
/// message quietly becoming a second one.
pub(super) fn access_to_server_error(error: &ChannelAccessError) -> ServerError {
    let message = match error {
        ChannelAccessError::NotRegistered { name } => {
            format!("channel '{name}' is not configured")
        }
        ChannelAccessError::RosterUnavailable { message } => message.clone(),
        ChannelAccessError::Quiesced { .. }
        | ChannelAccessError::ConsoleNamespace { .. }
        | ChannelAccessError::ConsoleTapTarget { .. }
        | ChannelAccessError::ConsoleTapLimit { .. } => error.to_string(),
    };
    ServerError::ListenerAccept { message }
}

/// Refuses a subscribe to `console.tap.<target>` whose target is empty or
/// under `console.`: the tap never taps the console's own records.
///
/// # Errors
/// Returns [`ChannelAccessError::ConsoleTapTarget`] naming the channel.
fn refuse_console_tap_target(channel: &str) -> Result<(), ChannelAccessError> {
    use super::channel_registry::{console_tap_target, is_console_subject};
    match console_tap_target(channel) {
        Some(target) if target.is_empty() || is_console_subject(target) => {
            Err(ChannelAccessError::ConsoleTapTarget {
                name: channel.to_owned(),
            })
        }
        _ => Ok(()),
    }
}

/// Refuses a connection's publish under `console.`, the server's own namespace.
///
/// # Errors
/// Returns [`ChannelAccessError::ConsoleNamespace`] naming the channel.
fn refuse_console_namespace(channel: &str) -> Result<(), ChannelAccessError> {
    if super::channel_registry::is_console_subject(channel) {
        return Err(ChannelAccessError::ConsoleNamespace {
            name: channel.to_owned(),
        });
    }
    Ok(())
}

pub(super) fn server_error_from_protocol(error: &ProtocolError) -> ServerError {
    ServerError::ListenerAccept {
        message: format!("protocol operation failed: {error}"),
    }
}

/// The three pinned registration tests. A CHILD module of `services` because
/// test 2 must hold the exact entry the admission funnel handed out.
#[cfg(test)]
#[path = "services_registry_tests.rs"]
mod services_registry_tests;

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod durable_store_tests {
    use liminal::durability::open_ephemeral_rooted;

    use super::subsystem_census::RecordingSubsystems;
    use super::{
        ConnectionServices, DEFAULT_SHARD_COUNT, LiminalConnectionServices, SchedulerSubsystem,
        build_connection_services, build_connection_services_via, build_durable_store,
    };
    use crate::ServerError;
    use crate::config::types::{LimitsConfig, ServerConfig, ServicesConfig};

    /// Counts directory entries under `root`, for the empty/one-dir assertions
    /// on an injected ephemeral root.
    fn entry_count(root: &std::path::Path) -> usize {
        std::fs::read_dir(root)
            .expect("ephemeral root is readable")
            .count()
    }

    /// A minimal channel-free config with the given service `profile`. No channels,
    /// routing, persistence, or cluster — the shape both profiles accept (the full
    /// profile simply builds an empty channel set; the worker-front-door profile
    /// requires exactly this shape).
    fn config_with_profile(profile: &str) -> ServerConfig {
        ServerConfig {
            listen_address: "127.0.0.1:0".parse().expect("valid socket addr"),
            health_listen_address: "127.0.0.1:1".parse().expect("valid socket addr"),
            drain_timeout_ms: 30_000,
            channels: Vec::new(),
            routing_rules: Vec::new(),
            persistence_path: None,
            cluster: None,
            auth: None,
            services: ServicesConfig {
                profile: profile.to_owned(),
                ..ServicesConfig::default()
            },
            limits: LimitsConfig::default(),
            participant: None,
            websocket: None,
        }
    }

    /// §9 D2 front-door construction gate (fs half): building the worker-front-door
    /// services creates NO haematite store and NO temp dir, while the full profile
    /// over an equally-rooted factory DOES create exactly one store directory.
    ///
    /// The injected root is the only place an ephemeral store directory can appear
    /// (the recording factory roots its stores there), so "the root stays empty on
    /// the front-door branch" is a real negative assertion: a regression that gave
    /// the front door a store would land its directory here and fail this test. The
    /// full-profile arm is the positive control proving the seam genuinely
    /// constructs a store when the profile asks for one.
    #[test]
    fn worker_front_door_builds_no_store_and_no_temp_dir() {
        let front_door_root = tempfile::tempdir().expect("test can create an ephemeral root");
        let full_root = tempfile::tempdir().expect("test can create an ephemeral root");

        let front_door_subsystems = RecordingSubsystems::rooted(front_door_root.path());
        let front_door: std::sync::Arc<dyn ConnectionServices> = build_connection_services_via(
            &config_with_profile("worker-front-door"),
            &front_door_subsystems,
        )
        .expect("worker-front-door services build");
        assert!(
            !front_door.supports_channel_operations(),
            "the worker front door serves no channel operations"
        );
        assert_eq!(
            entry_count(front_door_root.path()),
            0,
            "the worker front door creates no ephemeral store directory (no haematite, no temp dir)"
        );

        let full_subsystems = RecordingSubsystems::rooted(full_root.path());
        let mut config = config_with_profile("full");
        config.persistence_path = Some(full_root.path().to_path_buf());
        let full =
            build_connection_services_via(&config, &full_subsystems).expect("full services build");
        assert!(
            full.supports_channel_operations(),
            "full mode serves channel operations"
        );
        assert_eq!(
            entry_count(full_root.path()),
            1,
            "full mode builds its persistent database in the explicit directory"
        );

        drop(front_door);
        drop(full);
    }

    /// §9 D2 front-door construction gate (thread half — record-by-construction
    /// census): the worker profile constructs NO channel-supervisor,
    /// conversation-supervisor, or haematite scheduler, while the SAME instrument
    /// over the full profile records all three — the positive control proving the
    /// census detects the extra schedulers, so an empty census on the worker branch
    /// is a real observation, not a decoration.
    ///
    /// The instrument's boundary: recording happens INSIDE the [`SubsystemFactory`]
    /// methods that are the profile-aware path's only route to these constructors,
    /// so a recording cannot be silently omitted — a future subsystem added to this
    /// path either goes through the factory (and is recorded by construction) or
    /// bypasses it, which is a code-review-visible structural violation of the
    /// factory seam, not a silently-missing side call. The connection supervisor's
    /// own scheduler is the shared baseline of both profiles and is asserted at the
    /// supervisor level (`supervisor::tests`); an OS-level thread census upgrades
    /// this when the beamr composition lane's scheduler-inventory API (currently on
    /// their branch, not yet consumable from liminal) lands.
    #[test]
    fn worker_profile_census_is_empty_and_full_profile_records_all_schedulers() {
        let worker_root = tempfile::tempdir().expect("test can create an ephemeral root");
        let full_root = tempfile::tempdir().expect("test can create an ephemeral root");

        let worker_subsystems = RecordingSubsystems::rooted(worker_root.path());
        let front_door = build_connection_services_via(
            &config_with_profile("worker-front-door"),
            &worker_subsystems,
        )
        .expect("worker-front-door services build");
        assert_eq!(
            worker_subsystems.recorded(),
            Vec::<SchedulerSubsystem>::new(),
            "the worker front door constructs no scheduler-owning subsystem"
        );

        let full_subsystems = RecordingSubsystems::rooted(full_root.path());
        let mut config = config_with_profile("full");
        config.persistence_path = Some(full_root.path().to_path_buf());
        let full =
            build_connection_services_via(&config, &full_subsystems).expect("full services build");
        assert_eq!(
            full_subsystems.recorded(),
            vec![
                SchedulerSubsystem::ChannelSupervisor,
                SchedulerSubsystem::ConversationSupervisor,
                SchedulerSubsystem::HaematiteStore,
            ],
            "the full profile constructs every scheduler-owning subsystem, once each — \
             the positive control proving the census instrument detects them"
        );

        drop(front_door);
        drop(full);
    }

    /// MAJOR-1 regression: the full-only constructors reject a worker-front-door
    /// config with a typed `ConfigValidation` error AT ENTRY — no full service can
    /// be created through any public config-based constructor under that profile.
    #[test]
    fn full_only_constructors_reject_worker_profile() {
        let config = config_with_profile("worker-front-door");

        let from_config = LiminalConnectionServices::from_config(&config);
        assert!(
            matches!(from_config, Err(ServerError::ConfigValidation { .. })),
            "from_config must reject a worker-front-door profile with ConfigValidation, got {from_config:?}"
        );

        let root = tempfile::tempdir().expect("test can create an ephemeral root");
        let store = open_ephemeral_rooted(root.path(), DEFAULT_SHARD_COUNT)
            .expect("test store for the rejection check builds");
        let from_config_with_store =
            LiminalConnectionServices::from_config_with_store(&config, std::sync::Arc::new(store));
        assert!(
            matches!(
                from_config_with_store,
                Err(ServerError::ConfigValidation { .. })
            ),
            "from_config_with_store must reject a worker-front-door profile with ConfigValidation"
        );
    }

    /// MAJOR-1 regression: the profile-aware factory itself re-runs the
    /// worker-front-door cross-field checks, so a directly-constructed config (one
    /// that never passed file-loading validation) combining the worker profile with
    /// full-only machinery is refused with the same typed `ConfigValidation` errors.
    #[test]
    fn build_connection_services_rejects_worker_profile_with_full_only_fields() {
        let mut config = config_with_profile("worker-front-door");
        config.channels = vec![crate::config::types::ChannelDef {
            name: "orders".to_owned(),
            schema_ref: None,
            durable: false,
            loaded_schema: None,
        }];
        config.persistence_path = Some(std::path::PathBuf::from("/tmp"));

        let result = build_connection_services(&config);
        let Err(ServerError::ConfigValidation { message }) = result else {
            panic!("expected ConfigValidation for worker profile with full-only fields");
        };
        assert!(message.contains("builds no channels"), "got: {message}");
        assert!(
            message.contains("builds no durable store"),
            "got: {message}"
        );
    }

    /// The required-path constructor creates the persistent database in its
    /// configured directory without populating the separate throwaway root.
    #[test]
    fn persistent_store_uses_configured_path_and_creates_no_temp_dir() {
        let home = tempfile::tempdir().expect("test can create a temp dir");
        let ephemeral_root = tempfile::tempdir().expect("test can create an ephemeral root");

        let store = build_durable_store(home.path()).expect("persistent store builds");

        assert!(
            home.path().join("durability").join("config.json").exists(),
            "the persistent database is created under the configured path"
        );
        assert_eq!(
            entry_count(ephemeral_root.path()),
            0,
            "the persistent branch creates no ephemeral guard directory"
        );

        drop(store);
    }

    #[test]
    fn missing_path_refuses_before_any_store_is_opened() -> Result<(), Box<dyn std::error::Error>> {
        let root = tempfile::tempdir()?;
        let factory = RecordingSubsystems::rooted(root.path());
        let result = build_connection_services_via(&config_with_profile("full"), &factory);
        assert!(
            matches!(result, Err(ServerError::ConfigValidation { message }) if message.contains("no store is opened"))
        );
        assert_eq!(entry_count(root.path()), 0);
        assert!(factory.recorded().is_empty());
        Ok(())
    }

    /// The store migrates itself: a `config.json` written before haematite
    /// required `node_cache_budget` opens on the new server with no operator
    /// hand, and the file afterwards carries the key — spelled exactly as the
    /// create arm spells it. The pre-field store is fabricated by creating a
    /// current one and deleting the key, which is byte-faithful to what a
    /// 0.8.1-era store looks like to the parser: the key absent, everything
    /// else in place.
    #[test]
    fn a_store_whose_config_predates_the_budget_field_opens_with_no_operator_hand() {
        let home = tempfile::tempdir().expect("test can create a temp dir");
        let store = build_durable_store(home.path()).expect("persistent store builds");
        drop(store);

        let config_file = home.path().join("durability").join("config.json");
        let bytes = std::fs::read(&config_file).expect("config.json is readable");
        let mut config: serde_json::Value =
            serde_json::from_slice(&bytes).expect("config.json parses");
        config
            .as_object_mut()
            .expect("config.json is an object")
            .remove("node_cache_budget")
            .expect("the created config carries the budget to remove");
        std::fs::write(
            &config_file,
            serde_json::to_vec_pretty(&config).expect("serializes"),
        )
        .expect("test can age the config");

        let reopened = build_durable_store(home.path())
            .expect("a pre-budget store opens with no operator action");
        drop(reopened);

        let after = std::fs::read(&config_file).expect("config.json is readable after");
        let migrated: serde_json::Value =
            serde_json::from_slice(&after).expect("migrated config parses");
        assert_eq!(
            migrated.get("node_cache_budget"),
            Some(&serde_json::Value::String("unlimited".to_owned())),
            "the migrated file carries the budget the create arm chooses"
        );
    }

    /// The migration touches only the one migratable case. A corrupt
    /// `config.json` still refuses — through haematite's own parse refusal,
    /// which is sharper than anything this crate could say — and the bytes are
    /// left exactly as found, so the operator diagnoses the file that failed,
    /// not one the migration half-rewrote.
    #[test]
    fn a_corrupt_config_still_refuses_and_is_left_untouched() {
        let home = tempfile::tempdir().expect("test can create a temp dir");
        let data_dir = home.path().join("durability");
        std::fs::create_dir_all(&data_dir).expect("test can create the data dir");
        let config_file = data_dir.join("config.json");
        let garbage = b"{ not json at all";
        std::fs::write(&config_file, garbage).expect("test can write garbage");

        let refused = build_durable_store(home.path());
        assert!(refused.is_err(), "a corrupt config refuses to open");
        assert_eq!(
            std::fs::read(&config_file)
                .expect("config.json is still readable")
                .as_slice(),
            garbage,
            "the refusing file is left byte-identical for the operator to read"
        );
    }

    /// A config already carrying the budget is not rewritten: reopening leaves
    /// the file byte-identical, so the migration cannot churn mtimes, reorder
    /// keys, or otherwise touch a store that needs nothing.
    #[test]
    fn a_config_already_carrying_the_budget_is_left_byte_identical() {
        let home = tempfile::tempdir().expect("test can create a temp dir");
        let store = build_durable_store(home.path()).expect("persistent store builds");
        drop(store);

        let config_file = home.path().join("durability").join("config.json");
        let before = std::fs::read(&config_file).expect("config.json is readable");
        let reopened = build_durable_store(home.path()).expect("a current store reopens");
        drop(reopened);
        assert_eq!(
            std::fs::read(&config_file).expect("config.json is readable after"),
            before,
            "a config that already carries the budget is left byte-identical"
        );
    }

    /// A durable store created under `KeepAll` is moved onto reclaim by the
    /// server's next open: its `config.json` names `reclaim` afterwards, and the
    /// copies its old commits left are gone once reclaim has run.
    #[test]
    fn a_keep_all_durable_store_reclaims_after_the_next_open() {
        let dir = tempfile::tempdir().expect("test can create a store dir");
        let data_dir = dir.path().join("durability");
        let database = haematite::Database::create(
            haematite::DatabaseConfig {
                data_dir: data_dir.clone(),
                shard_count: 1,
                distributed: None,
                executor_threads: None,
                node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
                retention: haematite::Retention::KeepAll,
            },
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
                reclaim_start: haematite::ReclaimStart::AtOpen,
            },
        )
        .expect("a keep-all store is created");
        for round in 0_u32..200 {
            let mut value = round.to_be_bytes().to_vec();
            value.resize(1_024, 0x5a);
            database.put(b"key".to_vec(), value).expect("put");
            database.commit().expect("commit");
        }
        drop(database);
        let grown =
            haematite::db::vacuum::stats(&haematite::db::vacuum::Options::new(data_dir.clone()))
                .expect("stats of the grown store")
                .totals
                .enumerated_bytes;

        let reopened = super::open_or_create_database(&data_dir).expect("the server opens it");
        // The server holds reclaim at open and releases it once its listeners
        // bind (DUR-021 R1); the test releases it the same way.
        reopened
            .release_reclaim()
            .expect("the held reclaim is released");
        reopened.quiesce_reclaim().expect("reclaim runs to rest");
        drop(reopened);

        let config: serde_json::Value = serde_json::from_slice(
            &std::fs::read(data_dir.join("config.json")).expect("config is readable"),
        )
        .expect("config is JSON");
        assert_eq!(
            config["retention"],
            serde_json::to_value(super::DURABLE_RETENTION).expect("the retention serializes")
        );
        let report = haematite::db::vacuum::stats(&haematite::db::vacuum::Options::new(data_dir))
            .expect("stats of the reclaimed store");
        assert!(
            report.totals.enumerated_bytes <= report.totals.marked_bytes.saturating_mul(3),
            "the reopened store holds about its live set: enumerated {} B, marked {} B",
            report.totals.enumerated_bytes,
            report.totals.marked_bytes
        );
        assert!(
            report.totals.enumerated_bytes.saturating_mul(10) < grown,
            "the old copies are gone: {} B now against {grown} B grown",
            report.totals.enumerated_bytes
        );
    }

    /// `config.json` exactly as a store grown by an engine that never reclaimed
    /// carries it: `reclaim` with `past_roots` 0 is already written, so the
    /// open has no retention to adopt (it does move the format to version 3),
    /// and every superseded copy is still on disk.
    /// `DATA_DIR` stands for the one field the fixture rewrites.
    const STORED_RECLAIM_CONFIG: &str = r#"{
  "chunking_policy": {
    "chunking": "bytes-v2",
    "internal_target_bytes": 49152,
    "leaf_target_bytes": 65536
  },
  "data_dir": "DATA_DIR",
  "distributed": null,
  "executor_threads": null,
  "format_version": 2,
  "node_cache_budget": "unlimited",
  "retention": {
    "reclaim": {
      "past_roots": 0
    }
  },
  "shard_count": 8
}"#;

    /// A store whose `config.json` already names reclaim, but whose history was
    /// grown with nothing reclaimed, is opened by the server and gives its old
    /// copies back with no write and no quiesce. A store with no reclaim
    /// record owes a startup pass, and a shard runs it on its first turn. The
    /// seed holds both shapes the pass must free: one row overwritten many
    /// times, and rows written and then deleted, as a prune leaves them.
    #[test]
    fn a_stored_reclaim_store_grown_without_reclaim_gives_its_old_copies_back_on_open() {
        let dir = tempfile::tempdir().expect("test can create a store dir");
        let data_dir = dir.path().join("durability");
        let database = haematite::Database::create(
            haematite::DatabaseConfig {
                data_dir: data_dir.clone(),
                shard_count: 8,
                distributed: None,
                executor_threads: None,
                node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
                retention: haematite::Retention::KeepAll,
            },
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
                reclaim_start: haematite::ReclaimStart::AtOpen,
            },
        )
        .expect("a keep-all store is created");
        for round in 0_u32..200 {
            let mut value = round.to_be_bytes().to_vec();
            value.resize(1_024, 0x5a);
            database.put(b"key".to_vec(), value).expect("put");
            database.commit().expect("commit");
        }
        for row in 0_u32..200 {
            let key = [b"pruned-".as_slice(), &row.to_be_bytes()].concat();
            database.put(key, vec![0xa5; 1_024]).expect("put");
            database.commit().expect("commit");
        }
        for row in 0_u32..200 {
            let key = [b"pruned-".as_slice(), &row.to_be_bytes()].concat();
            database.delete(key).expect("delete");
        }
        database.commit().expect("commit");
        drop(database);
        let grown_report =
            haematite::db::vacuum::stats(&haematite::db::vacuum::Options::new(data_dir.clone()))
                .expect("stats of the grown store");
        let grown = grown_report.totals;
        assert!(
            grown.unmarked_nodes > 0,
            "the seed leaves copies no root reaches"
        );
        let mut owed: std::collections::BTreeSet<usize> = grown_report
            .shards
            .iter()
            .filter(|shard| shard.nodes.unmarked_nodes > 0)
            .map(|shard| shard.shard_id)
            .collect();

        let config_file = data_dir.join("config.json");
        let seeded = STORED_RECLAIM_CONFIG.replace(
            "DATA_DIR",
            data_dir.to_str().expect("the test dir is UTF-8"),
        );
        std::fs::write(&config_file, &seeded).expect("config.json is written");

        let reopened = super::open_or_create_database(&data_dir).expect("the server opens it");
        let watch = reopened
            .watch_reclaim_passes(64)
            .expect("every shard accepts a pass watch");
        // Released as the server releases it once its listeners bind (DUR-021 R1).
        reopened
            .release_reclaim()
            .expect("the held reclaim is released");
        let mut unmarked_by_shard = std::collections::BTreeMap::new();
        while !owed.is_empty() {
            let report = watch
                .recv_timeout(std::time::Duration::from_secs(30))
                .expect("each pass the open owes ends without a write or a quiesce");
            assert_eq!(
                report.outcome,
                haematite::ReclaimPassOutcome::Passed,
                "the pass the open owed on shard {} finished",
                report.shard_id
            );
            if owed.remove(&report.shard_id) {
                unmarked_by_shard.insert(report.shard_id, report.unmarked_bytes);
            }
        }
        drop(reopened);

        let after = haematite::db::vacuum::stats(&haematite::db::vacuum::Options::new(data_dir))
            .expect("stats of the reclaimed store")
            .totals;
        eprintln!(
            "grown {} B, after {} B, {} unreachable nodes seeded, unmarked bytes by shard {unmarked_by_shard:?}",
            grown.enumerated_bytes, after.enumerated_bytes, grown.unmarked_nodes
        );
        assert!(
            after.enumerated_bytes.saturating_mul(10) < grown.enumerated_bytes,
            "the old copies are gone: {} B now against {} B grown",
            after.enumerated_bytes,
            grown.enumerated_bytes
        );
        let config: serde_json::Value = serde_json::from_slice(
            &std::fs::read(&config_file).expect("config.json is readable after"),
        )
        .expect("config is JSON");
        assert_eq!(
            config["retention"],
            serde_json::to_value(super::DURABLE_RETENTION).expect("the retention serializes"),
            "config.json names reclaim after the open"
        );
    }
}