liminal_server/server/participant/

dispatch.rs

//! Participant transport-to-semantics dispatch boundary.
//!
//! This module contains no lifecycle rules. The shared protocol crate gates and
//! decodes inbound frames, while an injected semantic handler returns one typed
//! protocol value. The server then performs only generic-frame encoding.

use std::collections::BTreeSet;
use std::sync::{Arc, Weak};

use liminal::durability::DurableStore;
use liminal::protocol::Frame;
use liminal_protocol::lifecycle::{BindingTerminalAdmitError, ConnectionConversationTracking};
use liminal_protocol::wire::{
    BindingEpoch, ClientRequest, CodecError, ConnectionIncarnation, ConversationId,
    ObserverRecoveryHandshake, ParticipantId, ServerValue, ValidatedFrameLimit,
};

use crate::auth_pass::PassPrincipal;
use crate::server::mount::MountKind;

use super::dispatch_impact::DispatchImpact;
use super::transport::{
    ParticipantIngress, ParticipantSession, encode_server_value, gate_generic_frame,
    normalize_configured_frame_limit,
};
use super::{
    MarkerSettledPublication, ObserverPublicationTarget, ParticipantOfferedProgress,
    ParticipantPublication, ParticipantPublicationInbox, ParticipantPublicationRegistry,
};

/// One connection waiting on one settlement epoch, installed by its OWN refusal.
///
/// ⛔ This registry is the whole of §0.16 build obligation 3. The wake is
/// delivered "exactly to connections that received the refusal in this process
/// lifetime": a waiter can only be created from a
/// `ServerValue::MarkerSettlementBackpressure` leaving THIS connection's own
/// request, and the fire path walks waiters — never conversation membership,
/// never the publication registry's incarnation map. There is deliberately no
/// code path from a conversation id to a set of connections here, because that
/// is exactly the settlement-timing side channel the amendment outlaws.
///
/// A connection refused at the ENROLLMENT wrapper cannot appear: its refusal is
/// `EnrollmentSettlementBackpressure`, a different `ServerValue` variant with no
/// epoch, and the registration match does not admit it.
#[derive(Debug)]
struct MarkerSettlementWaiter {
    connection_incarnation: ConnectionIncarnation,
    refused_epoch: u64,
    target: ObserverPublicationTarget,
}

/// Told when a committed drain removes a head of one conversation's
/// immutable-candidate lane.
///
/// The connection runtime installs itself here so a connection fate parked on
/// candidate-lane `Precedence` is retried by the drain that removed the
/// refusal, on the thread that committed it, and never by a timer or by
/// whichever unrelated reply happens to come next. A lane drain is the only
/// event that can turn that refusal into an admission, so it is the only
/// event that retries.
pub trait CandidateLaneListener: Send + Sync + std::fmt::Debug {
    /// A drain of `conversation_id`'s candidate lane is durable and its
    /// resulting owner installed. Runs after the conversation lock is released.
    fn candidate_lane_drained(&self, conversation_id: ConversationId);
}

/// Connection-local semantic-conversation dispatch map (contract R-D1: the
/// connection's binding/interest/dispatch maps are bounded by the signed
/// `max_semantic_conversations_per_connection`).
///
/// One value lives in each connection process's state for the connection's
/// lifetime and is dropped with it. A conversation enters the map exactly
/// when a semantic operation for it COMMITS on this connection (the crate's
/// `ConnectionConversationCapacityCommit::newly_tracked` verdict) or when an
/// observer-recovery batch arms its refusal-only recipient; refusals and
/// replays leave the map untouched, exactly as the crate's stage-6 selector
/// leaves its counter unchanged. Growth is therefore bounded by the signed
/// limit the stage-6 selector enforces.
#[derive(Debug, Default)]
pub struct ParticipantConnectionConversations {
    tracked: BTreeSet<ConversationId>,
}

impl ParticipantConnectionConversations {
    /// Stage-6 tracking fact for one conversation on this connection.
    #[must_use]
    pub fn tracking(&self, conversation_id: ConversationId) -> ConnectionConversationTracking {
        if self.tracked.contains(&conversation_id) {
            ConnectionConversationTracking::AlreadyTracked
        } else {
            ConnectionConversationTracking::Untracked
        }
    }

    /// Current connection-conversation occupancy.
    #[must_use]
    pub fn occupied(&self) -> u64 {
        // `usize` fits `u64` on every supported target; if that ever stopped
        // holding, saturating at MAX fails CLOSED (capacity reads as full)
        // rather than silently under-counting occupancy.
        u64::try_from(self.tracked.len()).unwrap_or(u64::MAX)
    }

    /// Installs one conversation slot after a capacity-committing operation.
    pub fn track(&mut self, conversation_id: ConversationId) {
        self.tracked.insert(conversation_id);
    }

    /// Sorted tracked conversations (the observer-recovery preflight's
    /// current-occupancy input).
    #[must_use]
    pub fn tracked_conversations(&self) -> Vec<ConversationId> {
        self.tracked.iter().copied().collect()
    }
}

/// Connection-scoped authority facts supplied to participant semantics.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ParticipantConnectionContext {
    connection_incarnation: ConnectionIncarnation,
    mount: MountKind,
}

impl ParticipantConnectionContext {
    /// Captures the durably allocated incarnation of the receiving connection
    /// and the mount its admitting door stamped.
    ///
    /// Both arguments are server facts. `mount` in particular is supplied by
    /// the spawn path from its own knowledge of which door it is (design §10);
    /// it is a required argument rather than a defaulted field precisely so a
    /// new transport cannot acquire a mount attestation by forgetting to say
    /// which one it is.
    #[must_use]
    pub const fn new(connection_incarnation: ConnectionIncarnation, mount: MountKind) -> Self {
        Self {
            connection_incarnation,
            mount,
        }
    }

    /// Returns the durably allocated receiving-connection incarnation.
    #[must_use]
    pub const fn connection_incarnation(self) -> ConnectionIncarnation {
        self.connection_incarnation
    }

    /// Returns the mount the admitting door stamped on this connection.
    ///
    /// This is the mount attestation the consumer's door reads before stamping
    /// its own append. Nothing a client sends can move it: the value was fixed
    /// by the spawn path before the connection's first inbound byte was read.
    #[must_use]
    pub const fn mount(self) -> MountKind {
        self.mount
    }
}

/// Exact terminal classification preserved from a connection's close trigger.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ConnectionFateClass {
    /// A protocol-level clean Disconnect.
    CleanDisconnect,
    /// An orderly server `ForceClose`.
    ServerShutdown,
    /// EOF or transport loss without clean protocol evidence.
    ConnectionLost,
    /// A terminal protocol/decode refusal after participant binding.
    ProtocolError,
}

/// One durable bounded connection-fate intent delivered to participant semantics.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ConnectionFateWorkItem {
    /// Durable incarnation-stream Open sequence used by participant source rows.
    pub open_sequence: u64,
    /// Exact connection whose current Bound slots are eligible.
    pub connection_incarnation: ConnectionIncarnation,
    /// Preserved close classification.
    pub class: ConnectionFateClass,
    /// Canonical sorted tracked-conversation snapshot owned by the Open.
    pub tracked_conversations: Vec<ConversationId>,
}

/// Process-wide terminal participant-service latch.
#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)]
pub enum ParticipantServiceFatal {
    /// A durable Open landed but one listed conversation could not durably finish its fate.
    #[error(
        "connection-fate intent {open_sequence} is incomplete at conversation {conversation_id}"
    )]
    ConnectionFateIntentIncomplete {
        /// Durable incarnation-stream Open sequence.
        open_sequence: u64,
        /// Exact conversation whose non-idempotent completion failed.
        conversation_id: ConversationId,
    },
}

/// Why F8B boot recovery could not empty a restored conversation's
/// immutable-candidate lane (`docs/design/F8B-INTENT-DEADLOCK.md` §6.2
/// R-BOOT-VERDICT).
///
/// The discrimination is BY TYPE, for the same reason
/// [`ParticipantSemanticError::BindingTerminalAdmissionRefused`] carries
/// [`BindingTerminalAdmitError`]: a consumer deciding what a refused boot
/// means must not read it out of a formatted message.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BootDrainRefusal {
    /// The lane head is a pending binding terminal under an armed
    /// fenced-attach recovery block. The terminal drain refuses outright while
    /// a recovery block is armed, and the only consumer of a recovery block is
    /// a live fenced attach — which boot cannot perform. Such a store is not
    /// repairable by the boot drain, and this verdict is the honest answer
    /// rather than a repair.
    RecoveryArmed,
    /// Any other drain refusal: the head was reachable, the drain was
    /// attempted, and the protocol refused the transition.
    Shape,
}

/// Non-wire semantic service failure.
///
/// A failure is terminal to the connection attempt. It is deliberately not
/// convertible to [`ServerValue`], preventing the server from inventing a
/// lifecycle response when the protocol-owned transition did not produce one.
#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)]
pub enum ParticipantSemanticError {
    /// The complete semantic service is not installed.
    #[error("participant semantic service is unavailable")]
    Unavailable,
    /// Durable state or a protocol invariant prevented semantic completion.
    #[error("participant semantic service failed: {message}")]
    Internal {
        /// Diagnostic text for server logs; never placed on the participant wire.
        message: String,
    },
    /// The request's durable write ended with an outcome the store could not
    /// tell and the server could not settle by reading it back: it may have
    /// been applied. The dispatch answers a token-carrying request with its
    /// write-outcome-unknown value (`0x0128`) on an open connection.
    #[error("participant durable write outcome unknown: {message}")]
    WriteOutcomeUnknown {
        /// Diagnostic text for server logs; never placed on the participant wire.
        message: String,
    },
    /// A process-wide participant fatal has already latched.
    #[error(transparent)]
    ServiceFatal(ParticipantServiceFatal),
    /// A keyed binding-terminal candidate was refused, carrying the protocol's
    /// own reason rather than a formatted description of it.
    ///
    /// [`BindingTerminalAdmitError::Precedence`] is lane occupancy: the
    /// conversation's immutable-candidate lane already holds a terminal
    /// awaiting its drain. That is a designed structural boundary, and a
    /// caller deciding whether to park or to treat the refusal as corruption
    /// must be able to tell it from the five genuine authority defects by
    /// type.
    #[error("participant binding-terminal admission refused: {error:?}")]
    BindingTerminalAdmissionRefused {
        /// Exact protocol refusal reason.
        error: BindingTerminalAdmitError,
    },
    /// F8B R-BOOT-VERDICT: boot recovery could not empty a restored
    /// conversation's immutable-candidate lane, so the boot refuses HERE,
    /// naming the conversation and the shape, instead of starting and dying
    /// several collapses downstream on a retained `Open` it can never replay.
    #[error(
        "participant boot drain refused conversation {conversation_id} on lane head {candidate} \
         ({refusal:?}): {reason} — docs/design/F8B-INTENT-DEADLOCK.md §6.2 R-BOOT-VERDICT"
    )]
    BootDrainRefused {
        /// Conversation whose restored lane refused its drain.
        conversation_id: ConversationId,
        /// Typed reason, so a consumer never discriminates on a substring.
        refusal: BootDrainRefusal,
        /// Exact lane head that refused, rendered for the operator.
        candidate: String,
        /// The drain's own refusal text.
        reason: String,
    },
    /// F8B R-SEAL: the conversation is Closed — a Died-flavor drain erased its
    /// final enrollment token, so its log holds records, terminals and drain
    /// rows but no live identity can ever be reached through it again.
    ///
    /// Enrollment answers with this NAMED refusal rather than falling through
    /// to a fresh identity, which would silently re-open a conversation whose
    /// history has already ended. On the wire it rides the existing
    /// semantic-error framing; a protocol-native response value with its own
    /// discriminant is deferred to a protocol-version leg (§9.8).
    #[error(
        "participant enrollment refused: conversation {conversation_id} is sealed — its final \
         enrollment token was erased by a binding-terminal drain — \
         docs/design/F8B-INTENT-DEADLOCK.md §6.6 R-SEAL"
    )]
    ConversationSealed {
        /// Conversation whose closure refused the request.
        conversation_id: ConversationId,
    },
    /// CONTAINMENT: this one conversation's durable state cannot be loaded, so
    /// this one conversation is refused. The node starts, every other
    /// conversation is served, and the refusal NAMES its subject.
    ///
    /// Attribution is not decoration here, it is half the property. A node
    /// that contains an unloadable conversation without naming it boots clean
    /// and silently serves nothing on that conversation forever, which is
    /// worse than the crash it replaced, because the crash was the only thing
    /// telling anyone. The underlying failure travels as `reason` rather than
    /// as a wrapped error because it is already a rendered diagnostic by the
    /// time replay refuses it.
    #[error(
        "participant conversation {conversation_id} is unloadable and is refused on its own: \
         {reason}"
    )]
    ConversationUnloadable {
        /// Conversation whose durable state could not be loaded.
        conversation_id: ConversationId,
        /// The load failure's own text, as the operator needs to see it.
        reason: String,
    },
}

impl ParticipantSemanticError {
    /// Stable operator-facing class for this refusal.
    ///
    /// The rendered message is a diagnostic: it carries the subject and the
    /// detail, and it is allowed to move. This is the discriminant an operator
    /// surface and a log field can be read against without matching on a
    /// substring — which is what the containment record's consumers need, since
    /// the failure text a refused load carries ("expected value at line 1
    /// column 1") names no class at all on its own.
    #[must_use]
    pub const fn class(&self) -> &'static str {
        match self {
            Self::Unavailable => "unavailable",
            Self::Internal { .. } => "internal",
            Self::WriteOutcomeUnknown { .. } => "write_outcome_unknown",
            Self::ServiceFatal(_) => "service_fatal",
            Self::BindingTerminalAdmissionRefused { .. } => "binding_terminal_admission_refused",
            Self::BootDrainRefused { .. } => "boot_drain_refused",
            Self::ConversationSealed { .. } => "conversation_sealed",
            Self::ConversationUnloadable { .. } => "conversation_unloadable",
        }
    }
}

/// One semantic result paired with every dispatch effect durably installed by
/// the request before it returned.
///
/// The envelope deliberately owns the `Result`: a marker-drain prefix can
/// commit before a later retry fails, and that failure must not erase the
/// prefix's post-commit tell.
#[derive(Debug)]
pub struct ParticipantSemanticOutcome<T> {
    result: Result<T, ParticipantSemanticError>,
    impact: DispatchImpact,
}

impl<T> ParticipantSemanticOutcome<T> {
    /// Wraps a fixture or operation which installed no dispatch effect.
    #[must_use]
    pub const fn unchanged(result: Result<T, ParticipantSemanticError>) -> Self {
        Self {
            result,
            impact: DispatchImpact::Unchanged,
        }
    }

    /// Carries an operation result and its complete request accumulator.
    #[must_use]
    pub const fn new(result: Result<T, ParticipantSemanticError>, impact: DispatchImpact) -> Self {
        Self { result, impact }
    }

    pub(crate) fn into_parts(self) -> (Result<T, ParticipantSemanticError>, DispatchImpact) {
        (self.result, self.impact)
    }

    /// Returns the semantic result when an internal caller has no notification
    /// boundary. Production request dispatch uses the complete envelope; this
    /// projection exists for the trait's legacy direct-call entry point.
    pub(crate) fn into_result(self) -> Result<T, ParticipantSemanticError> {
        self.result
    }
}

/// One connection-fate result paired with every conversation impact committed
/// before the fate operation returned.
#[derive(Debug)]
pub struct ParticipantConnectionFateOutcome {
    result: Result<(), ParticipantSemanticError>,
    impacts: Vec<DispatchImpact>,
}

impl ParticipantConnectionFateOutcome {
    /// Wraps a fixture fate handler which committed no dispatch impact.
    #[must_use]
    pub const fn unchanged(result: Result<(), ParticipantSemanticError>) -> Self {
        Self {
            result,
            impacts: Vec::new(),
        }
    }

    /// Carries a fate result and every committed per-conversation impact.
    #[must_use]
    pub const fn new(
        result: Result<(), ParticipantSemanticError>,
        impacts: Vec<DispatchImpact>,
    ) -> Self {
        Self { result, impacts }
    }

    pub(crate) fn into_parts(self) -> (Result<(), ParticipantSemanticError>, Vec<DispatchImpact>) {
        (self.result, self.impacts)
    }

    pub(crate) fn into_result(self) -> Result<(), ParticipantSemanticError> {
        self.result
    }
}

/// Server-owned adapter from a decoded request to a protocol-owned value.
pub trait ParticipantSemanticHandler: core::fmt::Debug + Send + Sync {
    /// Applies one already authenticated and capability-gated request.
    ///
    /// `conversations` is the receiving connection's semantic-conversation
    /// dispatch map: the handler reads it for the crate's stage-6
    /// connection-conversation capacity facts and installs a slot exactly
    /// when an operation's capacity commit reports `newly_tracked`.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when no protocol value can be
    /// produced. The caller closes rather than fabricating a response.
    ///
    /// Production handlers override this with the signed
    /// `max_semantic_conversations_per_connection`; semantic-only fixtures own
    /// no publication conversations.
    ///
    /// Returns the latched fatal, when participant service must remain stopped.
    fn service_fatal(&self) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        Ok(None)
    }

    /// Atomically latches the post-Open fatal selected by Decision B.
    ///
    /// Implementations must preserve the first fatal and return it on every later call.
    /// The default exists only for semantic fixtures which own no durable intents.
    ///
    /// # Errors
    ///
    /// Returns a semantic service error when the fatal latch cannot be inspected or updated.
    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: ConversationId,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        Ok(ParticipantServiceFatal::ConnectionFateIntentIncomplete {
            open_sequence,
            conversation_id,
        })
    }

    /// Applies every matching participant binding named by one durable Open.
    ///
    /// The incarnation-stream lock is not held while this method runs. Each
    /// implementation serializes conversations independently and must return
    /// only after every source and immediately executable specific fate flushes.
    ///
    /// # Errors
    ///
    /// Returns a semantic failure without consuming the Open; startup or the
    /// live fatal path retains it for exact replay.
    fn handle_connection_fate(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        drop(work_item);
        Err(ParticipantSemanticError::Unavailable)
    }

    /// Applies connection fate while preserving every committed conversation's
    /// post-flush dispatch effects on both success and failure exits.
    fn handle_connection_fate_with_impact(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> ParticipantConnectionFateOutcome {
        ParticipantConnectionFateOutcome::unchanged(self.handle_connection_fate(work_item))
    }

    /// Drains every head of one conversation's immutable-candidate lane, as
    /// boot recovery does, and reports the drains on the outcome's impact.
    ///
    /// A clean stop calls this for each conversation a parked connection fate
    /// still waits on, so the fate can complete before the process exits
    /// instead of leaving its Open for the next boot to reconcile. The default
    /// owns no lane and answers [`ParticipantSemanticError::Unavailable`],
    /// which the stop reports by name.
    fn drain_candidate_lane_with_impact(
        &self,
        conversation_id: ConversationId,
    ) -> ParticipantSemanticOutcome<()> {
        let _ = conversation_id;
        ParticipantSemanticOutcome::unchanged(Err(ParticipantSemanticError::Unavailable))
    }

    /// Repairs every remaining binding owned by a prior server incarnation.
    ///
    /// Startup calls this after all retained Opens complete and before publishing
    /// the incarnation authority, scheduler, listener, or new admission.
    ///
    /// The default performs NO repair and reports success, so startup proceeds as
    /// if there were nothing left over from the prior incarnation. That is correct
    /// only for an implementor that holds no durable bindings. Any implementor with
    /// real restart semantics MUST override this — inheriting the default leaves
    /// stale bindings in place and says nothing about them.
    ///
    /// # Errors
    ///
    /// Returns a semantic failure while startup still owns all publication seams.
    fn repair_unclean_server_restart(
        &self,
        current_server_incarnation: u64,
    ) -> Result<(), ParticipantSemanticError> {
        let _ = current_server_incarnation;
        Ok(())
    }

    /// Reports whether any listed conversation currently contains a Bound slot
    /// owned by this exact connection. Terminal decode funnels use this query to
    /// distinguish bound-only `ProtocolError` from pre-auth/detached/internal paths.
    ///
    /// The default answers "unbound" unconditionally, without inspecting anything,
    /// so every caller is told this connection owns no Bound slot. That is correct
    /// only for an implementor that can never hold one. Any implementor with real
    /// binding semantics MUST override this — inheriting the default silently
    /// routes genuinely bound connections down the unbound funnel.
    ///
    /// # Errors
    ///
    /// Returns a semantic failure when exact bound authority cannot be inspected.
    fn connection_has_bound_participant(
        &self,
        connection_incarnation: ConnectionIncarnation,
        conversations: &[ConversationId],
    ) -> Result<bool, ParticipantSemanticError> {
        let _ = connection_incarnation;
        let _ = conversations;
        Ok(false)
    }

    fn publication_conversation_limit(&self) -> u64 {
        0
    }

    /// Resolves all live current bindings with pending durable obligations for
    /// one conversation. Production overrides this; semantic-only fixtures have
    /// no publication source.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when durable readiness cannot be resolved.
    fn ready_connection_incarnations(
        &self,
        _conversation_id: ConversationId,
    ) -> Result<Vec<ConnectionIncarnation>, ParticipantSemanticError> {
        Ok(Vec::new())
    }

    /// Selects the least durable recipient obligation for this incarnation,
    /// restarting from durable ack when `offered` names an older binding.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when the durable obligation owner is unavailable.
    fn next_publication(
        &self,
        _connection_incarnation: ConnectionIncarnation,
        _conversation_id: ConversationId,
        _offered: Option<ParticipantOfferedProgress>,
    ) -> Result<Option<ParticipantPublication>, ParticipantSemanticError> {
        Ok(None)
    }

    /// Checks that a held head still belongs to the exact current binding before
    /// it is offered after writable readiness.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when current binding authority cannot be read.
    fn publication_binding_is_current(
        &self,
        _conversation_id: ConversationId,
        _participant_id: ParticipantId,
        _binding_epoch: BindingEpoch,
    ) -> Result<bool, ParticipantSemanticError> {
        Ok(false)
    }

    /// Re-selects a held publication against current binding, cursor, debt, and
    /// outbox authority before its first offer. Semantic-only handlers retain
    /// the binding-only default; production overrides this with the full locked
    /// dispatch decision.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when current publication authority cannot be read.
    fn publication_is_current(
        &self,
        publication: &ParticipantPublication,
        offered: Option<ParticipantOfferedProgress>,
    ) -> Result<bool, ParticipantSemanticError> {
        if offered.is_some_and(|progress| progress.binding_epoch != publication.binding_epoch) {
            return Ok(false);
        }
        self.publication_binding_is_current(
            publication.conversation_id(),
            publication.participant_id,
            publication.binding_epoch,
        )
    }

    /// Records exact successful marker enqueue testimony. Non-marker offers are
    /// ignored by production after validating their current binding.
    ///
    /// # Errors
    ///
    /// Returns a semantic fault when exact offer testimony cannot be recorded.
    fn record_publication_offer(
        &self,
        _publication: &ParticipantPublication,
    ) -> Result<(), ParticipantSemanticError> {
        Ok(())
    }

    /// Applies observer recovery with the weak exact-live-connection target
    /// captured by the installed service. Semantic-only handlers delegate to
    /// their ordinary request path and do not own observer publication.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] under the same contract as
    /// [`Self::handle`].
    fn handle_observer_recovery(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ObserverRecoveryHandshake,
        target: Option<ObserverPublicationTarget>,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        drop(target);
        self.handle(
            context,
            conversations,
            ClientRequest::ObserverRecovery(request),
        )
    }

    /// Applies one request and preserves post-commit effects on every exit.
    ///
    /// Semantic-only fixtures default to an empty accumulator. Production
    /// overrides this boundary and returns operation-owned effects.
    fn handle_with_impact(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> ParticipantSemanticOutcome<ServerValue> {
        ParticipantSemanticOutcome::unchanged(self.handle(context, conversations, request))
    }

    /// Applies one decoded participant request to protocol-owned authority.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when durable or protocol authority
    /// cannot produce a truthful terminal value. The connection fails rather
    /// than fabricating a response.
    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError>;
}

/// Server-sealed participant activation token installed on a connection
/// supervisor.
///
/// The semantic handler and its durable store form one value so participant
/// capability activation cannot observe one without the other. The supervisor
/// uses the store to durably allocate connection incarnations before spawning a
/// connection process, and the process uses the handler only after that exact
/// incarnation has been carried into its state. The token atomically carries the
/// pair declared by server composition; it does not independently prove storage
/// namespace identity.
///
/// Construction and access are server-private. Until a complete production
/// lifecycle handler exists, external [`ConnectionServices`](crate::server::connection::ConnectionServices)
/// implementations cannot manufacture an activation token or advertise the
/// participant capability.
#[derive(Clone, Debug)]
pub struct InstalledParticipantService {
    handler: Arc<dyn ParticipantSemanticHandler>,
    durable_store: Arc<dyn DurableStore>,
    frame_limit: ValidatedFrameLimit,
    publication_registry: Arc<ParticipantPublicationRegistry>,
    /// Connections refused `MarkerSettlementBackpressure` in this process
    /// lifetime, by conversation. See [`MarkerSettlementWaiter`].
    settlement_waiters: Arc<
        std::sync::Mutex<std::collections::BTreeMap<ConversationId, Vec<MarkerSettlementWaiter>>>,
    >,
    /// The connection runtime told about every candidate-lane drain. Weak so
    /// the service never keeps the runtime that owns it alive.
    candidate_lane_listener: Arc<std::sync::Mutex<Option<Weak<dyn CandidateLaneListener>>>>,
}

impl InstalledParticipantService {
    /// Pairs a semantic handler, its declared durable store, and the raw
    /// configured participant wire-frame limit.
    ///
    /// Production construction happens exactly once, in the server's
    /// connection-services layer, from the deployment's `[participant]`
    /// configuration; tests construct it directly with fixture handlers.
    ///
    /// # Errors
    ///
    /// Returns the shared codec error when the configured limit is smaller than
    /// the protocol's minimum complete frame.
    pub(crate) fn new(
        handler: Arc<dyn ParticipantSemanticHandler>,
        durable_store: Arc<dyn DurableStore>,
        configured_wf: u64,
    ) -> Result<Self, CodecError> {
        Ok(Self {
            handler,
            durable_store,
            frame_limit: normalize_configured_frame_limit(configured_wf)?,
            publication_registry: Arc::new(ParticipantPublicationRegistry::default()),
            settlement_waiters: Arc::new(std::sync::Mutex::new(std::collections::BTreeMap::new())),
            candidate_lane_listener: Arc::new(std::sync::Mutex::new(None)),
        })
    }

    /// Installs the runtime told about candidate-lane drains, replacing any
    /// earlier one.
    pub(crate) fn install_candidate_lane_listener(
        &self,
        listener: Weak<dyn CandidateLaneListener>,
    ) -> Result<(), ParticipantSemanticError> {
        let mut installed = self.candidate_lane_listener.lock().map_err(|_| {
            ParticipantSemanticError::Internal {
                message: "candidate-lane listener slot is poisoned".to_owned(),
            }
        })?;
        *installed = Some(listener);
        drop(installed);
        Ok(())
    }

    /// Drains one conversation's candidate lane and tells the listener, as a
    /// publish's drain would.
    ///
    /// # Errors
    ///
    /// Returns the handler's drain refusal after telling every drain it
    /// committed first.
    pub(crate) fn drain_candidate_lane(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        let outcome = self
            .handler
            .drain_candidate_lane_with_impact(conversation_id);
        let (result, impact) = outcome.into_parts();
        self.notify_impact(&impact)?;
        result
    }

    /// Tells the installed listener that `conversation_id`'s lane drained.
    ///
    /// The listener is called with the slot lock released, because it retries
    /// durable connection fates.
    fn tell_candidate_lane_drained(
        &self,
        conversation_id: ConversationId,
    ) -> Result<(), ParticipantSemanticError> {
        let listener = self
            .candidate_lane_listener
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "candidate-lane listener slot is poisoned".to_owned(),
            })?
            .as_ref()
            .and_then(Weak::upgrade);
        if let Some(listener) = listener {
            listener.candidate_lane_drained(conversation_id);
        }
        Ok(())
    }

    /// Clones the durable store used by the installed participant service.
    #[must_use]
    pub(crate) fn durable_store(&self) -> Arc<dyn DurableStore> {
        Arc::clone(&self.durable_store)
    }

    /// Returns the normalized configured complete-frame limit advertised by
    /// this installed participant service.
    #[must_use]
    pub(crate) const fn frame_limit(&self) -> ValidatedFrameLimit {
        self.frame_limit
    }

    /// Returns the signed semantic-conversation allowance shared by publication
    /// readiness and connection-held encoded heads.
    #[must_use]
    pub(crate) fn publication_conversation_limit(&self) -> u64 {
        self.handler.publication_conversation_limit()
    }

    /// Creates the strongly connection-owned ready inbox at process spawn.
    #[must_use]
    pub(crate) fn new_publication_inbox(&self) -> ParticipantPublicationInbox {
        ParticipantPublicationInbox::new(self.handler.publication_conversation_limit())
    }

    /// Returns the shared weak publication registry.
    #[must_use]
    pub(crate) fn publication_registry(&self) -> &ParticipantPublicationRegistry {
        &self.publication_registry
    }

    /// Selects one exact durable publication through the installed production
    /// source.
    pub(crate) fn next_publication(
        &self,
        connection_incarnation: ConnectionIncarnation,
        conversation_id: ConversationId,
        offered: Option<ParticipantOfferedProgress>,
    ) -> Result<Option<ParticipantPublication>, ParticipantSemanticError> {
        self.handler
            .next_publication(connection_incarnation, conversation_id, offered)
    }

    pub(crate) fn publication_is_current(
        &self,
        publication: &ParticipantPublication,
        offered: Option<ParticipantOfferedProgress>,
    ) -> Result<bool, ParticipantSemanticError> {
        self.handler.publication_is_current(publication, offered)
    }

    pub(crate) fn record_publication_offer(
        &self,
        publication: &ParticipantPublication,
    ) -> Result<(), ParticipantSemanticError> {
        self.handler.record_publication_offer(publication)
    }

    /// Installs the waiter a `MarkerSettlementBackpressure` refusal earns.
    ///
    /// The match is the scope: only that ONE `ServerValue` variant registers,
    /// so an enrollment-wrapper refusal (which carries no epoch and no wake) and
    /// every other refusal on the same arm install nothing. Re-refusing the same
    /// conversation on the same connection REPLACES the waiter, because a
    /// connection waits on its most recent refusal and never on two epochs at
    /// once.
    fn register_settlement_waiter(
        &self,
        context: ParticipantConnectionContext,
        value: &ServerValue,
    ) -> Result<(), ParticipantSemanticError> {
        let ServerValue::MarkerSettlementBackpressure(refusal) = value else {
            return Ok(());
        };
        let (conversation_id, refused_epoch) = match *refusal {
            liminal_protocol::wire::MarkerSettlementBackpressure::CredentialAttach {
                conversation_id,
                refused_epoch,
            }
            | liminal_protocol::wire::MarkerSettlementBackpressure::Detach {
                conversation_id,
                refused_epoch,
            } => (conversation_id, refused_epoch),
        };
        let incarnation = context.connection_incarnation();
        let Some(target) = self
            .publication_registry
            .observer_target(incarnation)
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("settlement publication target failed: {error}"),
            })?
        else {
            return Ok(());
        };
        let mut waiters =
            self.settlement_waiters
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "settlement waiter registry is poisoned".to_owned(),
                })?;
        let entry = waiters.entry(conversation_id).or_default();
        entry.retain(|waiter| waiter.connection_incarnation != incarnation);
        entry.push(MarkerSettlementWaiter {
            connection_incarnation: incarnation,
            refused_epoch,
            target,
        });
        drop(waiters);
        Ok(())
    }

    /// Test hook onto [`Self::register_settlement_waiter`].
    ///
    /// The registry §0.16 obligation 3 rests on is private and is exercised in
    /// production only as a side effect of a refusal travelling out of `handle`.
    /// A pin that can only reach it that way measures the refusal path and the
    /// registry at once, and cannot tell which of the two is scoping the wake.
    /// These two hooks let the registry be measured ALONE.
    #[cfg(test)]
    pub(super) fn register_settlement_waiter_for_test(
        &self,
        context: ParticipantConnectionContext,
        value: &ServerValue,
    ) -> Result<(), ParticipantSemanticError> {
        self.register_settlement_waiter(context, value)
    }

    /// Test hook onto [`Self::fire_settlements`].
    #[cfg(test)]
    pub(super) fn fire_settlements_for_test(
        &self,
        conversation_id: ConversationId,
        settled_epochs: &[u64],
    ) -> Result<(), ParticipantSemanticError> {
        self.fire_settlements(conversation_id, settled_epochs)
    }

    /// Number of waiters currently installed for one conversation.
    #[cfg(test)]
    pub(super) fn settlement_waiter_count(&self, conversation_id: ConversationId) -> usize {
        self.settlement_waiters.lock().map_or(0, |waiters| {
            waiters.get(&conversation_id).map_or(0, Vec::len)
        })
    }

    /// Fires the settlement wake for every connection whose OWN refusal named
    /// this exact epoch, and for no one else.
    ///
    /// Epoch equality is load-bearing: it is what the stage-11 retry discipline
    /// matches on, and it is also what keeps a connection waiting on a later
    /// candidate from being told its own wait is over. Waiters are one-shot —
    /// fired or dead, they leave the registry.
    fn fire_settlements(
        &self,
        conversation_id: ConversationId,
        settled_epochs: &[u64],
    ) -> Result<(), ParticipantSemanticError> {
        if settled_epochs.is_empty() {
            return Ok(());
        }
        let mut waiters =
            self.settlement_waiters
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "settlement waiter registry is poisoned".to_owned(),
                })?;
        let Some(entry) = waiters.get_mut(&conversation_id) else {
            return Ok(());
        };
        let mut fired = Vec::new();
        entry.retain(|waiter| {
            if settled_epochs.contains(&waiter.refused_epoch) {
                fired.push((waiter.target.clone(), waiter.refused_epoch));
                false
            } else {
                true
            }
        });
        if entry.is_empty() {
            waiters.remove(&conversation_id);
        }
        drop(waiters);
        for (target, refused_epoch) in fired {
            target
                .publish_marker_settled(MarkerSettledPublication {
                    conversation_id,
                    refused_epoch,
                })
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("marker settled publication failed: {error}"),
                })?;
        }
        Ok(())
    }

    fn notify_impact(&self, impact: &DispatchImpact) -> Result<(), ParticipantSemanticError> {
        let Some(conversation_id) = impact.conversation_id() else {
            return Ok(());
        };
        if impact.candidate_lane_drained() {
            self.tell_candidate_lane_drained(conversation_id)?;
        }
        self.fire_settlements(conversation_id, impact.settled_epochs())?;
        for target in impact.target_union() {
            self.publication_registry
                .notify(
                    target.binding_epoch().connection_incarnation,
                    conversation_id,
                )
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("participant publication wake failed: {error}"),
                })?;
        }
        Ok(())
    }
}

impl ParticipantSemanticHandler for InstalledParticipantService {
    fn service_fatal(&self) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        self.handler.service_fatal()
    }

    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: ConversationId,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        self.handler
            .latch_connection_fate_intent_incomplete(open_sequence, conversation_id)
    }

    fn publication_conversation_limit(&self) -> u64 {
        self.handler.publication_conversation_limit()
    }

    fn handle_connection_fate(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        let outcome = self.handler.handle_connection_fate_with_impact(work_item);
        let (result, impacts) = outcome.into_parts();
        for impact in &impacts {
            self.notify_impact(impact)?;
        }
        result
    }

    fn handle_connection_fate_with_impact(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> ParticipantConnectionFateOutcome {
        let result = self.handle_connection_fate(work_item);
        ParticipantConnectionFateOutcome::unchanged(result)
    }

    fn drain_candidate_lane_with_impact(
        &self,
        conversation_id: ConversationId,
    ) -> ParticipantSemanticOutcome<()> {
        ParticipantSemanticOutcome::unchanged(self.drain_candidate_lane(conversation_id))
    }

    fn repair_unclean_server_restart(
        &self,
        current_server_incarnation: u64,
    ) -> Result<(), ParticipantSemanticError> {
        self.handler
            .repair_unclean_server_restart(current_server_incarnation)
    }

    fn connection_has_bound_participant(
        &self,
        connection_incarnation: ConnectionIncarnation,
        conversations: &[ConversationId],
    ) -> Result<bool, ParticipantSemanticError> {
        self.handler
            .connection_has_bound_participant(connection_incarnation, conversations)
    }

    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        if let ClientRequest::ObserverRecovery(request) = request {
            let target = self
                .publication_registry
                .observer_target(context.connection_incarnation())
                .map_err(|error| ParticipantSemanticError::Internal {
                    message: format!("observer publication target failed: {error}"),
                })?;
            return self
                .handler
                .handle_observer_recovery(context, conversations, request, target);
        }
        let outcome = self
            .handler
            .handle_with_impact(context, conversations, request);
        let (result, impact) = outcome.into_parts();
        // Fire before registering: a request that both drained and was refused
        // must not wake itself with its own drain, and within one request the
        // settlement it cleared is never the one it is now waiting on.
        self.notify_impact(&impact)?;
        if let Ok(value) = &result {
            self.register_settlement_waiter(context, value)?;
        }
        result
    }
}

/// Result of dispatching one generic frame through participant transport.
#[derive(Debug)]
pub enum ParticipantDispatch {
    /// The generic frame belongs to another protocol.
    NotParticipant,
    /// Exact encoded response selected by the shared gate or semantic handler.
    Respond(Frame),
    /// Exact crate-owned pre-semantic rejection, followed by connection close.
    RespondThenClose(Frame),
    /// No truthful participant response exists; the connection must fail closed.
    Fatal(ParticipantDispatchError),
}

/// Failure after a generic frame has entered participant dispatch.
#[derive(Debug, thiserror::Error)]
pub enum ParticipantDispatchError {
    /// The preserved generic frame could not represent a canonical participant frame.
    #[error("invalid generic participant frame")]
    InvalidGenericFrame,
    /// The semantic handler could not produce a protocol value.
    #[error(transparent)]
    Semantic(#[from] ParticipantSemanticError),
    /// The crate-produced value could not be encoded into the generic transport.
    #[error("failed to encode participant response: {0:?}")]
    Encode(CodecError),
}

fn pass_refusal(
    principal: Option<&PassPrincipal>,
    request: &liminal_protocol::wire::ClientRequest,
) -> Option<liminal_protocol::wire::TransportRejectionReason> {
    let principal = principal?;
    match request {
        liminal_protocol::wire::ClientRequest::Enrollment(request) if !principal.may_enroll => Some(liminal_protocol::wire::TransportRejectionReason::EnrollmentNotPermitted),
        liminal_protocol::wire::ClientRequest::Enrollment(request) if !principal.conversations.contains(&request.conversation_id) => Some(liminal_protocol::wire::TransportRejectionReason::EnrollmentConversationOutOfScope),
        liminal_protocol::wire::ClientRequest::CredentialAttach(request) if !principal.conversations.contains(&request.conversation_id) => Some(liminal_protocol::wire::TransportRejectionReason::CredentialAttachConversationOutOfScope),
        _ => None,
    }
}

/// The write-outcome-unknown value (`0x0128`) answering `request`, or `None`
/// for a request that carries no attempt token to be re-presented with.
const fn write_outcome_unknown(
    request: &liminal_protocol::wire::ClientRequest,
) -> Option<ServerValue> {
    use liminal_protocol::wire::{
        AttachEnvelope, ClientRequest, DetachEnvelope, EnrollmentEnvelope, LeaveEnvelope,
        RecordAdmissionEnvelope, WriteOutcomeUnknown,
    };
    Some(ServerValue::WriteOutcomeUnknown(match request {
        ClientRequest::Enrollment(request) => WriteOutcomeUnknown::Enrollment {
            request: EnrollmentEnvelope {
                conversation_id: request.conversation_id,
                enrollment_token: request.enrollment_token,
            },
        },
        ClientRequest::CredentialAttach(request) => WriteOutcomeUnknown::CredentialAttach {
            request: AttachEnvelope {
                conversation_id: request.conversation_id,
                participant_id: request.participant_id,
                capability_generation: request.capability_generation,
                attach_attempt_token: request.attach_attempt_token,
                accept_marker_delivery_seq: request.accept_marker_delivery_seq,
            },
        },
        ClientRequest::Detach(request) => WriteOutcomeUnknown::Detach {
            request: DetachEnvelope {
                conversation_id: request.conversation_id,
                participant_id: request.participant_id,
                capability_generation: request.capability_generation,
                detach_attempt_token: request.detach_attempt_token,
            },
        },
        ClientRequest::Leave(request) => WriteOutcomeUnknown::Leave {
            request: LeaveEnvelope {
                conversation_id: request.conversation_id,
                participant_id: request.participant_id,
                capability_generation: request.capability_generation,
                leave_attempt_token: request.leave_attempt_token,
            },
        },
        ClientRequest::RecordAdmission(request) => WriteOutcomeUnknown::RecordAdmission {
            request: RecordAdmissionEnvelope {
                conversation_id: request.conversation_id,
                participant_id: request.participant_id,
                capability_generation: request.capability_generation,
                record_admission_attempt_token: request.record_admission_attempt_token,
            },
        },
        _ => return None,
    }))
}

/// Gates, decodes, semantically applies, and encodes one participant frame.
///
/// Transport rejection values originate in `liminal-protocol`; semantic values
/// originate only in `handler`. No lifecycle outcome is constructed here.
#[must_use]
pub fn dispatch_generic_frame(
    frame: &Frame,
    authenticated: bool,
    session: ParticipantSession,
    context: ParticipantConnectionContext,
    pass_principal: Option<&PassPrincipal>,
    conversations: &mut ParticipantConnectionConversations,
    handler: &dyn ParticipantSemanticHandler,
) -> ParticipantDispatch {
    let (value, close_after_response) = match gate_generic_frame(frame, authenticated, session) {
        ParticipantIngress::NotParticipant => return ParticipantDispatch::NotParticipant,
        ParticipantIngress::Rejected(rejection) => {
            (ServerValue::ParticipantTransportRejected(rejection), true)
        }
        ParticipantIngress::InvalidGenericFrame => {
            return ParticipantDispatch::Fatal(ParticipantDispatchError::InvalidGenericFrame);
        }
        ParticipantIngress::Request(request) => {
            if let Some(reason) = pass_refusal(pass_principal, &request) {
                (
                    ServerValue::ParticipantTransportRejected(
                        liminal_protocol::wire::ParticipantTransportRejected { reason },
                    ),
                    false,
                )
            } else {
                let outcome_unknown = write_outcome_unknown(&request);
                match handler.handle(context, conversations, request) {
                    Ok(value) => (value, false),
                    Err(error) => match (&error, outcome_unknown) {
                        (
                            ParticipantSemanticError::WriteOutcomeUnknown { message },
                            Some(value),
                        ) => {
                            tracing::warn!(
                                %message,
                                "a participant write's outcome is unknown; the client is told so \
                                 and re-presents the request with its token"
                            );
                            (value, false)
                        }
                        _ => {
                            return ParticipantDispatch::Fatal(ParticipantDispatchError::Semantic(
                                error,
                            ));
                        }
                    },
                }
            }
        }
    };
    match encode_server_value(value) {
        Ok(frame) if close_after_response => ParticipantDispatch::RespondThenClose(frame),
        Ok(frame) => ParticipantDispatch::Respond(frame),
        Err(error) => ParticipantDispatch::Fatal(ParticipantDispatchError::Encode(error)),
    }
}