liminal-rs 0.16.3

//! Public channel surface: [`ChannelConfig`], [`ChannelMode`], and the
//! cloneable [`ChannelHandle`] that drives a REAL supervised beamr channel
//! actor (LIM-002).
//!
//! The handle is a thin, synchronous-looking facade over the process-backed
//! actor: every operation enqueues a typed command onto the actor's mailbox and
//! blocks on a per-command reply (the haematite `ShardHandle` pattern). It owns
//! no subscriber state and performs no fan-out itself — the actor process does.
//!
//! `ChannelHandle::new` stays infallible (existing call-sites depend on it): the
//! actor is spawned lazily on first use and the spawn result is memoised, so a
//! scheduler failure surfaces as a `LiminalError` from the first operation
//! rather than as a panic.

use std::sync::atomic::AtomicU32;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::Value;

use crate::causal::CausalContext;
use crate::channel::actor::{ChannelActorCore, PendingPublish, predicate_from};
use crate::channel::admission::{
    ChannelPressureConfig, defer_after_append, defer_delay, watermark_reached,
};
use crate::channel::durable_publish::{Outgoing, RuntimeDurable};
use crate::channel::observer::ClusterObserver;
use crate::channel::record::PublishRecord;
use crate::channel::schema::{Schema, SchemaId, SchemaValidationError};
use crate::channel::subscription::{InboxInstall, SubscriptionHandle, SubscriptionPredicate};
use crate::channel::supervisor::{ChannelSupervisor, shared_supervisor};
use crate::durability::bridge::block_on;
use crate::durability::{DurableChannel, DurableStore, recover_durable_channel};
use crate::envelope::{Envelope, PublisherId};
use crate::error::LiminalError;
use crate::pressure::PressureSignal;

/// Single-partition count used to back a flat runtime channel with durable storage.
const RUNTIME_DURABLE_PARTITIONS: usize = 1;

/// Genuine delivery ack returned by [`ChannelHandle::publish_with_delivery`].
///
/// Distinct from backpressure: it reports whether a published message was
/// actually received by a subscriber, not whether it was admitted to the bus.
///
/// `Copy` is preserved from published `liminal-rs 0.9.1`. A1 added two fields;
/// the one that briefly cost `Copy` was `signal: PressureSignal`, which is four
/// `usize`s and now derives `Copy` itself (`durable_position: Option<u64>` and
/// `pressure_config: ChannelPressureConfig` were already `Copy`). Nothing about
/// this type wants move semantics, and taking `Copy` away from a published type
/// breaks callers silently — `let a = delivery; use(delivery);` simply stops
/// compiling.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ChannelDelivery {
    delivery_id: Option<crate::protocol::DeliveredId>,
    delivered_count: usize,
    matched_count: usize,
    durable_position: Option<u64>,
    signal: PressureSignal,
    pressure_config: ChannelPressureConfig,
}

impl ChannelDelivery {
    /// The exact live identity supplied by the publication's transport.
    #[must_use]
    pub const fn delivery_id(&self) -> Option<crate::protocol::DeliveredId> {
        self.delivery_id
    }

    /// Number of local subscribers the message was genuinely delivered to.
    #[must_use]
    pub const fn delivered_count(&self) -> usize {
        self.delivered_count
    }

    /// Number of local subscribers whose predicate matched the message, whether
    /// or not their inbox took it. With [`Self::delivered_count`] it tells
    /// "nobody was subscribed" (0 matched) from "subscribers were there and
    /// none took it" (matched, 0 delivered). Zero when the publish was refused
    /// before any fan-out (a durable watermark Reject).
    #[must_use]
    pub const fn matched_count(&self) -> usize {
        self.matched_count
    }

    /// Whether the message was accepted by at least one subscriber.
    #[must_use]
    pub const fn is_delivered(&self) -> bool {
        self.delivered_count > 0
    }

    /// The store-assigned append sequence of this publish on a durable
    /// channel, `None` on a memory channel.
    ///
    /// This is the durable position G1 of
    /// `docs/design/wire-durability-gaps-20260831.md` names: haematite's
    /// append assigns it, and before this field the caller discarded it, so
    /// no client of a durable channel could ever learn where its publish
    /// landed. Any resume/ack protocol needs exactly this value.
    ///
    /// Assigned is not flushed: the position is real the moment the append
    /// returns, but the store flushes on graceful shutdown only (G4's flush
    /// half), so a crash can still lose the tail. The position tells you
    /// WHERE the publish sits, not that it survives every failure mode.
    #[must_use]
    pub const fn durable_position(&self) -> Option<u64> {
        self.durable_position
    }

    /// The A1 backpressure signal for this publish
    /// (`docs/design/A1-DEFER-SEMANTICS.md` §6: "`ChannelDelivery` gains a
    /// `pressure()` accessor (additive)").
    ///
    /// Orthogonal to [`Self::is_delivered`]. Delivery answers "did a subscriber
    /// receive it"; pressure answers "should you slow down". A publish to a
    /// channel with no subscribers is `Accept` and delivered to nobody; a
    /// publish every consumer buffered is `Defer` and delivered to everybody.
    #[must_use]
    pub const fn pressure(&self) -> &PressureSignal {
        &self.signal
    }

    /// Whether the bus took custody of the message (A1 §7): `Accept` or
    /// `Defer`.
    ///
    /// The accessor an aion worker's fire-and-forget completion policy reads,
    /// deliberately distinct from [`Self::is_delivered`] ("a subscriber
    /// genuinely received it"). A Deferred message is admitted — the bus holds
    /// it and the consumer's next pop is its redelivery — so a producer that
    /// re-publishes it with a fresh key double-delivers. `false` means Reject:
    /// shed, delivered to nobody, and the producer owns the retry.
    #[must_use]
    pub const fn is_admitted(&self) -> bool {
        !matches!(self.signal, PressureSignal::Reject { .. })
    }

    /// The producer's advisory pacing hint (A1 §3):
    /// `base + (max − base) × buffer_fill_fraction`, from this channel's
    /// configured delay policy. `None` on Accept — an accepted publish is not
    /// paced.
    ///
    /// The bus never blocks a producer on this and never lies about it:
    /// ignoring the hint leads to Reject on an ephemeral channel, and to
    /// growing hints and eventually the pre-append watermark Reject on a
    /// durable one.
    #[must_use]
    pub fn defer_delay(&self) -> Option<Duration> {
        defer_delay(&self.signal, self.pressure_config)
    }
}

/// Defines whether a channel is memory-only or durable across restarts.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChannelMode {
    /// In-memory channel mode with no persistence overhead.
    Ephemeral,
    /// Durable channel mode reserved for future haematite-backed storage.
    Durable,
}

/// Compatibility alias for the channel-owned schema definition.
pub type SchemaRef = Schema;

/// Required configuration for creating a typed channel.
///
/// **SEMVER — this type takes a declared break in the next publish wave.** A1
/// added the `pressure` field, and adding a field to a struct with public
/// fields breaks every out-of-crate struct-literal construction
/// (`ChannelConfig { name, schema, mode }` no longer compiles). There is no
/// version of "carry a per-channel pressure policy on the config" that avoids
/// it, so it is declared rather than hidden — see
/// `docs/design/A1-DEFER-WIRING-RECORD-2026-09-01.md` §6.
///
/// `#[non_exhaustive]` is added in the SAME wave, deliberately: the break is
/// already being paid, and it buys the property that no future field costs one
/// again. Out-of-crate construction goes through [`Self::new`] plus the
/// `with_*` builders, which is what every construction site in this workspace
/// already does. Reading a field is unaffected.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct ChannelConfig {
    /// Explicit channel name.
    pub name: String,
    /// Explicit schema for validating published payloads.
    pub schema: Schema,
    /// Explicit durability mode for the channel.
    pub mode: ChannelMode,
    /// A1 pressure policy for this channel: the Defer delay-hint band and the
    /// durable pre-append hard watermark. Defaulted by [`Self::new`], so no
    /// existing construction site changes; overridden with
    /// [`Self::with_pressure`].
    pub pressure: ChannelPressureConfig,
}

impl ChannelConfig {
    /// Creates channel configuration from its required fields, with the
    /// default A1 pressure policy.
    #[must_use]
    pub const fn new(name: String, schema: Schema, mode: ChannelMode) -> Self {
        Self {
            name,
            schema,
            mode,
            pressure: ChannelPressureConfig::DEFAULT,
        }
    }

    /// Replaces this channel's A1 pressure policy.
    #[must_use]
    pub const fn with_pressure(mut self, pressure: ChannelPressureConfig) -> Self {
        self.pressure = pressure;
        self
    }
}

/// A lazily-spawned, supervised channel actor shared by every clone of a handle.
///
/// `supervisor` is stored as a `Result` so [`ChannelHandle::new`] can stay
/// infallible: a scheduler-start failure is captured here and surfaced as a
/// `LiminalError` the first time the actor is actually used.
struct ChannelActorState {
    /// The channel's name, which the actor reports a departed subscriber under.
    name: String,
    supervisor: Result<ChannelSupervisor, String>,
    core: OnceLock<Result<Arc<ChannelActorCore>, String>>,
    restarts: AtomicU32,
}

impl ChannelActorState {
    const fn new(name: String, supervisor: Result<ChannelSupervisor, String>) -> Self {
        Self {
            name,
            supervisor,
            core: OnceLock::new(),
            restarts: AtomicU32::new(0),
        }
    }

    fn supervisor(&self) -> Result<&ChannelSupervisor, LiminalError> {
        self.supervisor
            .as_ref()
            .map_err(|message| LiminalError::PublishFailed {
                message: format!("channel supervisor unavailable: {message}"),
            })
    }

    /// The installed cluster observer, if this channel runs on a clustered
    /// supervisor (SRV-005). Returns `None` for non-clustered channels.
    fn observer(&self) -> Option<Arc<dyn ClusterObserver>> {
        self.supervisor
            .as_ref()
            .ok()
            .and_then(|supervisor| supervisor.observer().cloned())
    }

    /// Returns the live actor core, spawning it (and any restart) on demand.
    fn core(&self, schema: &Schema) -> Result<Arc<ChannelActorCore>, LiminalError> {
        let supervisor = self.supervisor()?;
        let stored = self.core.get_or_init(|| {
            supervisor
                .spawn_channel(&self.name, schema.clone())
                .map_err(|error| error.to_string())
        });
        let core = stored
            .as_ref()
            .map_err(|message| LiminalError::PublishFailed {
                message: format!("channel actor unavailable: {message}"),
            })?;
        // Restart on a dead pid (R4) before returning the core for use.
        supervisor.ensure_running(core, &self.restarts)?;
        Ok(Arc::clone(core))
    }
}

impl std::fmt::Debug for ChannelActorState {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ChannelActorState")
            .field("supervisor", &self.supervisor)
            .finish_non_exhaustive()
    }
}

/// Cloneable handle for interacting with a channel actor process.
#[derive(Clone, Debug)]
pub struct ChannelHandle {
    delivery_identity: Arc<OnceLock<Arc<dyn super::PublicationIdentity>>>,
    pub(super) config: ChannelConfig,
    actor: Arc<ChannelActorState>,
    pub(super) durable: Option<Arc<Mutex<RuntimeDurable>>>,
    /// The same store `durable` was built over, kept beside it so A1 §4's
    /// host-side refill can read the log WITHOUT taking the `DurableChannel`
    /// mutex the publish path holds. A refill runs on a consumer's thread while
    /// producers are publishing; making it contend for the append lock would
    /// let a slow reader stall every writer, which is the opposite of what
    /// backpressure is for. `Some` exactly when `durable` is.
    durable_store: Option<Arc<dyn DurableStore>>,
}

impl ChannelHandle {
    /// Installs a transport allocator before publishing or exposing this handle.
    ///
    /// # Errors
    /// An allocator was already installed on this shared channel handle.
    pub fn install_publication_identity(
        &self,
        source: Arc<dyn super::PublicationIdentity>,
    ) -> Result<(), LiminalError> {
        self.delivery_identity
            .set(source)
            .map_err(|_| LiminalError::PublishFailed {
                message: "publication identity already installed".into(),
            })
    }

    /// Creates an ephemeral handle backed by a real supervised channel actor on
    /// the shared default supervisor.
    ///
    /// The actor process is spawned lazily on first use; a scheduler failure is
    /// surfaced as a [`LiminalError`] from the first operation, not a panic.
    #[must_use]
    pub fn new(config: ChannelConfig) -> Self {
        let supervisor = shared_supervisor().map_err(|error| error.to_string());
        let name = config.name.clone();
        Self {
            delivery_identity: Arc::new(OnceLock::new()),
            config,
            actor: Arc::new(ChannelActorState::new(name, supervisor)),
            durable: None,
            durable_store: None,
        }
    }

    /// Creates an ephemeral handle bound to an explicit `supervisor` (isolation
    /// for the registry and tests).
    #[must_use]
    pub fn with_supervisor(config: ChannelConfig, supervisor: ChannelSupervisor) -> Self {
        let name = config.name.clone();
        Self {
            delivery_identity: Arc::new(OnceLock::new()),
            config,
            actor: Arc::new(ChannelActorState::new(name, Ok(supervisor))),
            durable: None,
            durable_store: None,
        }
    }

    /// Creates a durable handle that persists every accepted publish to `store`
    /// before fanning it out to subscribers.
    ///
    /// Construction reads the store: every partition's next-sequence counter is
    /// recovered from the existing stream head, so a handle rebuilt over a
    /// previously used store resumes appending where the log left off instead
    /// of conflicting at sequence zero. A fresh store recovers to zero, which is
    /// identical to cold start. This makes construction O(stream length) in
    /// store reads and adds store read errors to the failure modes below.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError::PublishFailed`] when the durable channel cannot be
    /// initialized over `store`, including when recovering the per-partition
    /// sequence counters from the store fails.
    pub fn new_durable(
        config: ChannelConfig,
        store: Arc<dyn DurableStore>,
    ) -> Result<Self, LiminalError> {
        let durable = recover_durable(&config.name, Arc::clone(&store))?;
        let supervisor = shared_supervisor()?;
        let name = config.name.clone();
        Ok(Self {
            delivery_identity: Arc::new(OnceLock::new()),
            config,
            actor: Arc::new(ChannelActorState::new(name, Ok(supervisor))),
            durable: Some(Arc::new(Mutex::new(RuntimeDurable::new(durable)))),
            durable_store: Some(store),
        })
    }

    /// Creates a durable handle bound to an explicit `supervisor`.
    ///
    /// Used by the standalone server so every channel — durable or ephemeral —
    /// shares ONE (optionally clustered) supervisor and thus one scheduler, which
    /// is the precondition for cross-node delivery (SRV-005): a subscriber pid
    /// joined to a channel's distributed process group must live on the same
    /// scheduler that owns the distribution links.
    ///
    /// Like [`Self::new_durable`], construction recovers each partition's
    /// next-sequence counter from the store, so restarting over an existing
    /// persistence path resumes the log instead of conflicting at sequence zero.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError::PublishFailed`] when the durable channel cannot be
    /// initialized over `store`, including when recovering the per-partition
    /// sequence counters from the store fails.
    pub fn new_durable_with_supervisor(
        config: ChannelConfig,
        store: Arc<dyn DurableStore>,
        supervisor: ChannelSupervisor,
    ) -> Result<Self, LiminalError> {
        let durable = recover_durable(&config.name, Arc::clone(&store))?;
        let name = config.name.clone();
        Ok(Self {
            delivery_identity: Arc::new(OnceLock::new()),
            config,
            actor: Arc::new(ChannelActorState::new(name, Ok(supervisor))),
            durable: Some(Arc::new(Mutex::new(RuntimeDurable::new(durable)))),
            durable_store: Some(store),
        })
    }

    /// Returns the channel configuration used to create this handle.
    #[must_use]
    pub const fn config(&self) -> &ChannelConfig {
        &self.config
    }

    /// Publishes a payload to the channel with the default publisher identity.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel cannot accept the payload or the schema rejects it.
    pub fn publish<Payload>(&self, payload: Payload) -> Result<(), LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        self.publish_with_context(payload, PublisherId::default(), None)
    }

    /// Publishes a payload with an explicit publisher identity.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel cannot accept the payload or the schema rejects it.
    pub fn publish_from<Payload>(
        &self,
        publisher_id: impl Into<PublisherId>,
        payload: Payload,
    ) -> Result<(), LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        self.publish_with_context(payload, publisher_id.into(), None)
    }

    /// Publishes a payload with explicit publisher and causal metadata.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel cannot accept the payload or the schema rejects it.
    pub fn publish_with_context<Payload>(
        &self,
        payload: Payload,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
    ) -> Result<(), LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        self.publish_with_delivery(payload, publisher_id, causal_context)
            .map(|_delivery| ())
    }

    /// Publishes a payload and reports a genuine delivery ack.
    ///
    /// Returns a [`ChannelDelivery`] whose `delivered_count` is the number of
    /// local subscribers the message was actually delivered to. A caller that
    /// needs to know the message was ACCEPTED by a subscriber (not merely
    /// buffered/published) inspects [`ChannelDelivery::is_delivered`]. This is the
    /// channel-library half of the 13-L1 delivery-ack signal; the publish-without-
    /// delivery methods stay unchanged for existing callers.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel cannot accept the payload or the schema rejects it.
    pub fn publish_with_delivery<Payload>(
        &self,
        payload: Payload,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
    ) -> Result<ChannelDelivery, LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        self.publish_with_key(payload, publisher_id, causal_context, None, None)
    }

    /// Publishes as [`ChannelHandle::publish_with_delivery`] does (and, with
    /// an `idempotency_key`, as
    /// [`ChannelHandle::publish_with_idempotency_key`] does), and fills
    /// `record` with what it saw: the subscriber pid of every local
    /// subscriber whose inbox queued the message, with the instant it did
    /// (exactly [`ChannelDelivery::delivered_count`] of them), and, on a
    /// durable channel, the instant its append completed. A subscriber whose
    /// predicate matched and whose inbox refused the message is counted in
    /// [`ChannelDelivery::matched_count`] and is not a recipient.
    ///
    /// `record` is cleared first and its allocation is reused, so a caller
    /// that keeps one record allocates only while it grows to the channel's
    /// widest fan-out. A publish that fails, or is refused at the watermark,
    /// names no recipient and hands its allocation back, with one exception:
    /// a publish whose channel actor took the command and then never answered
    /// (it stopped, or the wait for it ended) leaves no allocation, because
    /// the one it lent went with that actor. A durable publish whose append
    /// committed before it failed (it is then
    /// [`LiminalError::PublishUnresolved`]) still says when the append
    /// completed: the row is stored. Any other failure leaves the record
    /// empty.
    ///
    /// # Errors
    ///
    /// Returns the errors of [`ChannelHandle::publish_with_delivery`].
    pub fn publish_recording<Payload>(
        &self,
        payload: Payload,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
        idempotency_key: Option<&str>,
        record: &mut PublishRecord,
    ) -> Result<ChannelDelivery, LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        record.clear();
        self.publish_with_key(
            payload,
            publisher_id,
            causal_context,
            idempotency_key,
            Some(record),
        )
    }

    pub(super) fn publish_with_key<Payload>(
        &self,
        payload: Payload,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
        idempotency_key: Option<&str>,
        mut record: Option<&mut PublishRecord>,
    ) -> Result<ChannelDelivery, LiminalError>
    where
        Payload: AsRef<[u8]>,
    {
        let core = self.core()?;
        let delivery_id = self
            .delivery_identity
            .get()
            .map(|source| source.next())
            .transpose()?;
        // G4 (docs/design/wire-durability-gaps-20260831.md): the schema/closed
        // gate runs BEFORE the durable bridge, so a payload the channel would
        // refuse never burns a store sequence. The actor's validation inside
        // `publish` below stays authoritative; a concurrent schema evolve
        // narrows the remaining pre-validated-then-refused window to that race.
        core.pre_publish_gate(payload.as_ref())?;
        // A1 §4 (graft §0.3), the PRE-APPEND HARD WATERMARK. A durable channel
        // owes the producer Defer-not-Reject after an append, which means the
        // only honest Reject it can ever emit is one taken before anything is
        // persisted. Without this check a producer that ignores delay hints
        // grows the log without bound, because every post-append outcome is by
        // construction an admission.
        //
        // Cheap and coarse on purpose: a channel-aggregate occupancy read with
        // no predicate evaluation, so it may refuse a message whose target
        // subscriber is fast. The design documents that imprecision and defers
        // the precise per-cursor bound to v2, once live consumer cursors exist.
        if let Some(mut rejection) = self.watermark_rejection(&core)? {
            rejection.delivery_id = delivery_id;
            return Ok(rejection);
        }
        // Durable channels persist the message to the store BEFORE acknowledging
        // the publish (and before fanning out): a published message that was not
        // durably recorded would be lost on shutdown, which CN7 forbids. (The
        // store is flushed on graceful shutdown only — an unflushed append can
        // still be lost to a crash; that bound is G4's flush half, named in the
        // gaps docket, not silently absorbed here.)
        //
        // AND THE FAN-OUT IS ENQUEUED UNDER THE SAME LOCK (round 2). See
        // `persist_and_enqueue` for why: the position and the queue slot are
        // taken together, so the actor's FIFO command queue receives durable
        // publishes in durable-position order. Everything §4 does with the
        // replay cursor — a single `u64` meaning "the contiguous prefix this
        // subscriber has been offered" — is false without it.
        let (durable_position, pending, committed) = match self.durable.as_ref() {
            Some(durable) => {
                let (position, pending, committed) = self.persist_and_enqueue(
                    &core,
                    durable,
                    Outgoing {
                        delivery_id,
                        payload: payload.as_ref(),
                        publisher_id,
                        causal_context,
                        idempotency_key,
                        record: record.as_deref_mut(),
                    },
                )?;
                (Some(position), pending, Some(committed))
            }
            None => (
                None,
                core.enqueue_identified(
                    payload.as_ref().to_vec(),
                    publisher_id,
                    causal_context,
                    None,
                    record.as_deref_mut().map(|record| &mut record.recipients),
                    delivery_id,
                )?,
                None,
            ),
        };
        // The wait happens with the append lock released: the ordering is
        // already fixed by the enqueue above, and holding it across the actor
        // rendezvous would put a slow subscriber's notifier in front of the
        // next publisher's append for no ordering gain at all. A durable row is
        // stored by now, so a failed rendezvous is an unresolved publish, never
        // a definite failure that would license a second append.
        let mut outcome = core
            .await_publish(pending)
            .map_err(|error| match committed {
                Some(row) => row.unresolved(error),
                None => error,
            })?;
        // SRV-005: hand the normalised envelope to the cluster observer so it can
        // fan the message out to remote subscribers. Local fan-out already
        // happened inside `core.publish`; this is purely the cross-node leg and
        // is a no-op when no observer is installed (non-clustered channels).
        if let Some(observer) = self.actor.observer() {
            observer.on_publish(&self.config.name, &outcome.envelope);
        }
        // A1 §4: "after append, durable channels Defer, never Reject." The
        // message is at its sequence position and every replay consumer sees it
        // exactly once regardless of what the live buffers did — the live shed
        // marked its subscriber lagging and auto-catch-up covers the gap.
        // Emitting Reject here would say "delivered to nobody, retry", which is
        // false of a message that is in the log AND on its way to a consumer.
        //
        // WITH ONE EXCEPTION, AND IT IS A MEASURED ONE (round 3). That
        // justification holds for the A1 §4 PACING shed, which arms
        // auto-catch-up (`admit_inner` sets `state.lagging` on a durable
        // Reject). It is FALSE for the two §5 memory-safety doors: neither
        // `FairnessTripped` nor `BudgetExceeded` sets `lagging` or bumps
        // `shed_generation`, so no refill is ever due, and on the connection
        // path the subscription is torn down outright before one could run —
        // the server's delivery pump sheds an `is_overflowed()` subscription
        // with a typed `SubscribeError` and releases it, and `try_next`, the
        // ONLY driver of `refill_if_lagging`, is never called on it again.
        // MEASURED, M1, gate-logs/a1-defer/m1-measurement-section5-recovery.log:
        // after the shedding pressure was fully freed and the consumer polled
        // on, the pacing door recovered 3/3 of its shed rows and the two §5
        // doors recovered 0/4 and 0/13.
        //
        // So when EVERY matching subscriber dropped the envelope through §5,
        // the rewrite is suppressed and the producer reads the Reject the
        // aggregate actually resolved. That answer is about LIVE CUSTODY, not
        // about the log: the row IS appended and a deliberate `replay_from`
        // read still finds it, but no subscription will ever be offered it
        // again (a fresh subscription seeds its replay cursor at the CURRENT
        // head), so telling the producer "admitted" would be the §7 custody lie
        // F2 closed on ephemeral channels, arriving by a different door.
        //
        // A §5 drop BESIDE a subscriber that took it, or beside one the pacing
        // door shed, is not this case: `every_match_dropped` is false and the
        // Defer stands, because a consumer will still get the message.
        //
        // SENTINEL-SHAPES: this makes the two channel modes give the same shape
        // of answer to the same situation — an everybody-dropped publish is a
        // Reject on ephemeral and on durable alike.
        let signal = if durable_position.is_some() && !outcome.every_match_dropped {
            defer_after_append(outcome.signal)
        } else {
            outcome.signal
        };
        if let (Some(record), Some(filled)) = (record, outcome.recipients.take()) {
            record.recipients = filled;
        }
        Ok(ChannelDelivery {
            delivery_id,
            delivered_count: outcome.delivered_count,
            matched_count: outcome.matched_count,
            durable_position,
            signal,
            pressure_config: self.config.pressure,
        })
    }

    /// Awaits the fan-out of a row whose uncertain append resolved as
    /// committed, and hands it to the cluster observer as any publish is.
    pub(super) fn finish_recovered(
        &self,
        core: &ChannelActorCore,
        pending: PendingPublish,
    ) -> Result<(), LiminalError> {
        let outcome = core.await_publish(pending)?;
        if let Some(observer) = self.actor.observer() {
            observer.on_publish(&self.config.name, &outcome.envelope);
        }
        Ok(())
    }

    /// Appends and flushes a durable batch once, then delivers each row in its stored order.
    ///
    /// An empty batch does no work. The single-record publication path is unchanged.
    /// Flush observers must hand publication off without re-entering this channel's append lock.
    ///
    /// # Errors
    /// Returns an error for an ephemeral channel, an empty publisher, invalid payloads or failed storage.
    /// A committed row whose delivery is unconfirmed returns [`LiminalError::PublishUnresolved`].
    pub fn publish_batch_with_delivery(
        &self,
        payloads: &[Vec<u8>],
        publisher: &PublisherId,
        causal: Option<&CausalContext>,
    ) -> Result<Vec<ChannelDelivery>, LiminalError> {
        if payloads.is_empty() {
            return Ok(Vec::new());
        }
        if publisher.as_str().trim().is_empty() {
            return Err(LiminalError::PublishFailed {
                message: "durable batch publisher is empty".to_owned(),
            });
        }
        let durable = self
            .durable
            .as_ref()
            .ok_or_else(|| LiminalError::PublishFailed {
                message: "batch publication requires a durable channel".to_owned(),
            })?;
        let core = self.core()?;
        for payload in payloads {
            core.pre_publish_gate(payload)?;
        }
        if let Some(rejection) = self.watermark_rejection(&core)? {
            return Ok(vec![rejection; payloads.len()]);
        }
        let queued = self.persist_batch_and_enqueue(&core, durable, payloads, publisher, causal)?;
        let mut deliveries = Vec::with_capacity(queued.len());
        for (pending, mut proof) in queued {
            let position = proof.sequence();
            let outcome = core.await_publish(pending).map_err(|error| {
                proof.set_cause(format!("batch committed; live delivery failed: {error}"));
                LiminalError::PublishUnresolved { append: proof }
            })?;
            if let Some(observer) = self.actor.observer() {
                observer.on_publish(&self.config.name, &outcome.envelope);
            }
            let signal = if outcome.every_match_dropped {
                outcome.signal
            } else {
                defer_after_append(outcome.signal)
            };
            deliveries.push(ChannelDelivery {
                delivery_id: outcome.envelope.delivery_id,
                delivered_count: outcome.delivered_count,
                matched_count: outcome.matched_count,
                durable_position: Some(position),
                signal,
                pressure_config: self.config.pressure,
            });
        }
        Ok(deliveries)
    }

    /// The pre-append hard watermark verdict for a durable channel (A1 §4).
    ///
    /// `Some(delivery)` is an honest Reject: **nothing was appended**, nothing
    /// was fanned out, `durable_position` is `None`, and `delivered_count` is
    /// zero — so the caller's `Reject ⇒ delivered to nobody` reading holds and
    /// the server's dedup mapping releases the claim exactly as if the publish
    /// had never happened (§5). `None` means proceed.
    ///
    /// Always `None` on a memory channel: there is no log to bound, and an
    /// ephemeral channel's Reject is the per-message shed the fan-out produces.
    fn watermark_rejection(
        &self,
        core: &Arc<ChannelActorCore>,
    ) -> Result<Option<ChannelDelivery>, LiminalError> {
        if self.durable.is_none() {
            return Ok(None);
        }
        let (total_queued, total_bound) = core.live_buffer_occupancy()?;
        if !watermark_reached(
            total_queued,
            total_bound,
            self.config.pressure.durable_reject_watermark_percent,
        ) {
            return Ok(None);
        }
        Ok(Some(ChannelDelivery {
            delivery_id: None,
            delivered_count: 0,
            matched_count: 0,
            durable_position: None,
            signal: PressureSignal::reject(total_queued, total_bound, total_queued, total_bound),
            pressure_config: self.config.pressure,
        }))
    }

    /// The durable partition stream key and its current head — the point a new
    /// subscription joins the log at (A1 §4).
    ///
    /// Liminal backs a runtime channel with a single partition
    /// ([`RUNTIME_DURABLE_PARTITIONS`]), so partition 0 is the whole log.
    fn durable_join_point(
        durable: &Arc<Mutex<RuntimeDurable>>,
    ) -> Result<(String, u64), LiminalError> {
        let channel = durable
            .lock()
            .map_err(|error| LiminalError::PublishFailed {
                message: format!("durable channel state unavailable: {error}"),
            })?;
        let stream_key = channel.channel.stream_key_for(0);
        let head = channel.channel.next_expected_sequence(0).unwrap_or(0);
        drop(channel);
        Ok((stream_key, head))
    }

    /// The store this durable channel was built over.
    fn durable_store(&self) -> Result<Arc<dyn DurableStore>, LiminalError> {
        self.durable_store
            .as_ref()
            .map(Arc::clone)
            .ok_or_else(|| LiminalError::PublishFailed {
                message: format!(
                    "channel '{}' has no durable store to replay from",
                    self.config.name
                ),
            })
    }

    /// Returns the schema version currently owned by the channel actor.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel actor cannot be read.
    pub fn current_schema_id(&self) -> Result<SchemaId, LiminalError> {
        self.core()?.schema_id()
    }

    /// Evolves the channel schema by adding a defaulted field without disconnecting subscribers.
    ///
    /// # Errors
    ///
    /// Returns [`SchemaValidationError`] when the schema cannot be evolved.
    pub fn evolve_schema_add_field(
        &self,
        name: impl Into<String>,
        field_schema: Value,
        default: Value,
    ) -> Result<SchemaId, SchemaValidationError> {
        let core = self
            .core()
            .map_err(|error| SchemaValidationError::InvalidSchema {
                message: error.to_string(),
            })?;
        core.evolve(name.into(), field_schema, default)
    }

    /// Subscribes to the channel, receiving every published message.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when a subscription cannot be created.
    pub fn subscribe(&self) -> Result<SubscriptionHandle, LiminalError> {
        self.subscribe_inner(None, None, None)
    }

    /// Subscribes with a server-connection [`InboxInstall`]: the §5 shared byte
    /// budget, per-inbox fairness cap, and R3 wake notifier are installed on the
    /// inbox AT CONSTRUCTION — strictly before the registration is published to
    /// the channel actor — so no envelope can be admitted uncharged, past the
    /// depth cap, or without a wake (the pre-install window is structurally
    /// closed, not merely narrowed).
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when a subscription cannot be created.
    pub fn subscribe_with_install(
        &self,
        install: InboxInstall,
    ) -> Result<SubscriptionHandle, LiminalError> {
        self.subscribe_inner(None, Some(install), None)
    }

    /// Subscribes with a delivery predicate: only messages for which `predicate`
    /// returns `true` are delivered to this subscriber. The predicate is owned
    /// and evaluated by the actor process (R3).
    ///
    /// # Clustering
    ///
    /// The predicate filters **local-node publishes only**. Under clustering
    /// (SRV-005), messages published on a remote node are delivered to this
    /// subscriber *ungated* — the predicate is a non-serializable closure and is
    /// not propagated across the wire, so remote nodes cannot evaluate it. If you
    /// need filtering to hold for cross-node traffic, filter again on receipt
    /// rather than relying on this predicate alone.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when a subscription cannot be created.
    pub fn subscribe_filtered<F>(&self, predicate: F) -> Result<SubscriptionHandle, LiminalError>
    where
        F: Fn(&Envelope) -> bool + Send + Sync + 'static,
    {
        self.subscribe_inner(Some(predicate_from(predicate)), None, None)
    }

    /// Opens a stable durable consumer at its subscription-owned checkpoint.
    /// Ephemeral channels retain the same bounded live subscription.
    ///
    /// # Errors
    /// Subscription construction, cursor recovery, or replay failed.
    pub fn subscribe_consumer(
        &self,
        consumer: &str,
        install: InboxInstall,
    ) -> Result<SubscriptionHandle, LiminalError> {
        self.subscribe_inner(None, Some(install), Some(consumer))
    }

    fn subscribe_inner(
        &self,
        predicate: Option<SubscriptionPredicate>,
        install: Option<InboxInstall>,
        consumer: Option<&str>,
    ) -> Result<SubscriptionHandle, LiminalError> {
        let core = self.core()?;
        let (handle, registration) =
            SubscriptionHandle::spawn(&core.scheduler()?, predicate.clone(), install)?;
        // A1 §4 auto-catch-up: attach the durable log this subscription can
        // converge from, and seed its replay cursor at the head it is JOINING
        // AT — both strictly before the registration reaches the actor, so the
        // first publish this subscription can possibly see is measured against
        // the right origin. A cursor left at zero would replay the entire
        // history on the first shed.
        if let Some(durable) = self.durable.as_ref() {
            let (stream_key, head) = Self::durable_join_point(durable)?;
            handle.seed_replay_cursor(head);
            handle.attach_durable_refill(
                self.durable_store()?,
                stream_key,
                self.config.schema.id(),
                predicate,
            );
            if let Some(consumer) = consumer {
                handle.resume_consumer(consumer)?;
            }
        }
        let pid = registration.pid();
        core.subscribe(registration)?;
        if consumer.is_some() {
            handle.refill_if_lagging()?;
        }
        // SRV-005: tell the cluster a local subscriber joined this channel so it
        // can advertise the subscription to peers via its process group.
        if let Some(observer) = self.actor.observer() {
            observer.on_subscribe(&self.config.name, pid);
        }
        Ok(handle)
    }

    /// Unsubscribes the subscriber owning `subscription` by its process pid.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the unsubscribe command fails.
    pub fn unsubscribe(&self, subscription: &SubscriptionHandle) -> Result<(), LiminalError> {
        let pid = subscription.pid();
        self.core()?.unsubscribe(pid)?;
        // SRV-005: tell the cluster the local subscriber left so it can withdraw
        // the subscription from its process group.
        if let Some(observer) = self.actor.observer() {
            observer.on_unsubscribe(&self.config.name, pid);
        }
        Ok(())
    }

    /// Flushes buffered durable channel state to the backing store before shutdown.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel actor cannot be inspected or
    /// when the durable store flush fails.
    pub fn flush(&self) -> Result<(), LiminalError> {
        // Confirm the actor is reachable (and restart it if needed) before flush.
        drop(self.core()?);
        let Some(durable) = self.durable.as_ref() else {
            return Ok(());
        };
        let flush_result = {
            let channel = durable
                .lock()
                .map_err(|error| LiminalError::PublishFailed {
                    message: format!("durable channel state unavailable: {error}"),
                })?;
            block_on(channel.channel.flush_store())
        };
        flush_result
            .map_err(|error| LiminalError::PublishFailed {
                message: format!(
                    "durable flush bridge for channel '{}' failed: {error}",
                    self.config.name
                ),
            })?
            .map_err(|error| LiminalError::PublishFailed {
                message: format!(
                    "durable flush for channel '{}' failed: {error}",
                    self.config.name
                ),
            })?;
        Ok(())
    }

    /// Returns the number of currently-active subscribers on the channel actor.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the actor cannot service the query.
    pub fn subscriber_count(&self) -> Result<usize, LiminalError> {
        Ok(self.core()?.list_subscribers()?.len())
    }

    /// Closes the channel gracefully, stopping the actor process.
    ///
    /// # Errors
    ///
    /// Returns a [`LiminalError`] when the channel cannot be shut down.
    pub fn close(&self) -> Result<(), LiminalError> {
        self.core()?.close()
    }

    /// Whether this channel's actor has been spawned — **without spawning it**.
    ///
    /// This is the one accessor on the handle that observes the lazy-spawn
    /// state instead of consuming it. Every other accessor
    /// ([`subscriber_count`](Self::subscriber_count),
    /// [`close`](Self::close)) routes through the private `core()`, which
    /// spawns the actor on demand, so reading "is there an actor?" through any
    /// of them destroys the very property being read. A handle that spawns
    /// lazily has a genuine, observable idle/live state, and a consumer that
    /// needs to know it — a registry reporting which of its channels are inert,
    /// a cost accountant bounding what mere configuration costs — otherwise has
    /// to choose between not knowing and changing the answer by asking.
    ///
    /// It reads the memoised spawn slot and nothing else: no supervisor call,
    /// no schema clone, no `ensure_running`. That last exclusion is the point
    /// of the restriction — `ensure_running` RESTARTS a dead actor, so an
    /// implementation routed through it would repair the state rather than
    /// report it.
    ///
    /// # What `false` and `true` mean
    ///
    /// The slot holds `Result<Arc<ChannelActorCore>, String>`: a spawn that
    /// failed is memoised as `Err` and, because the slot is one-shot, is never
    /// retried — every later operation on this handle fails with that stored
    /// message. So a failed spawn ATTEMPT leaves no actor and can never acquire
    /// one, and this method answers `false` for it, exactly as it does for a
    /// channel nobody has touched. `false` therefore means "no actor core
    /// exists" and deliberately does not distinguish "never attempted" from
    /// "attempted and permanently failed" — neither has an actor.
    ///
    /// `true` means the actor core exists, not that its process is currently
    /// alive: an actor whose pid has died stays `true` until the next real
    /// operation restarts it through the supervisor. Liveness is a stronger
    /// question than this method asks, and answering it would require running
    /// the repair path this method exists to avoid.
    #[must_use]
    pub fn is_actor_spawned(&self) -> bool {
        self.actor.core.get().is_some_and(Result::is_ok)
    }

    pub(super) fn core(&self) -> Result<Arc<ChannelActorCore>, LiminalError> {
        self.actor.core(&self.config.schema)
    }

    /// The channel actor's current beamr pid, ensuring it is running first.
    /// Test-only: lets restart tests crash the exact actor process.
    #[cfg(test)]
    pub(crate) fn actor_pid(&self) -> Result<u64, LiminalError> {
        let core = self.core()?;
        core.current_pid()?
            .ok_or_else(|| LiminalError::DeliveryFailed {
                message: "channel actor has no live pid".to_owned(),
            })
    }

    /// The scheduler the channel actor and its subscribers run on (test-only).
    #[cfg(test)]
    pub(crate) fn scheduler(&self) -> Result<Arc<beamr::scheduler::Scheduler>, LiminalError> {
        self.core()?.scheduler()
    }
}

/// Reconstructs a durable channel over `store`, deriving each partition's next
/// sequence from the persisted log so a restart over an existing stream appends at
/// the tail rather than colliding with the occupied head (H2 / ledger G1). On a
/// fresh store, recovery yields zeroed sequences — identical to a first-boot
/// channel — so this is the single durable-construction path for both cold start
/// and restart.
///
/// The async recovery is driven to completion through the synchronous
/// [`block_on`] bridge, exactly as the durable publish/flush paths do: the
/// haematite store completes on its first poll, so no executor is needed.
fn recover_durable(
    channel_name: &str,
    store: Arc<dyn DurableStore>,
) -> Result<DurableChannel, LiminalError> {
    block_on(recover_durable_channel(
        channel_name.to_owned(),
        RUNTIME_DURABLE_PARTITIONS,
        store,
    ))
    .map_err(|error| LiminalError::PublishFailed {
        message: format!("durable recovery bridge for channel '{channel_name}' failed: {error}"),
    })?
    .map_err(|error| LiminalError::PublishFailed {
        message: format!("failed to recover durable channel '{channel_name}': {error}"),
    })
}

/// Returns the current epoch milliseconds, saturating to zero before the epoch.
pub(super) fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| {
            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
        })
}