liminal-server 0.22.0

//! Inbound-frame application: maps a decoded client frame to a [`FrameAction`]
//! (a response to enqueue, silence, or a close) by delegating to the liminal
//! library through [`ConnectionServices`]. Split out of [`super::process`] so the
//! connection handler there stays focused on socket IO and the slice pump.

use std::collections::BTreeMap;

use liminal::protocol::{
    Frame, MessageEnvelope, PUBLISH_ACK_POSITION_VERSION, PUBLISH_DELIVERED_FLAG, ProtocolError,
    ProtocolVersion, SchemaId as ProtocolSchemaId, WorkerRegisterOutcome, WorkerRegistration,
    negotiate_version,
};
use liminal_protocol::reason_code::PUBLISH_UNRESOLVED_CODE;
use liminal_protocol::wire::PARTICIPANT_FRAME_TYPE;

use super::link_route::{link_message_frame_response, link_open_frame_response};
use super::services::{
    ChannelOperation, ConnectionServices, access_to_server_error, reserved_namespace_refusal,
};
use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::{ConnectionRuntime, PushBinding, PushReplyDisposition};
use super::worker_route::Request;
use crate::ServerError;
use crate::auth_lys::routes::{RouteClassification, classify_participant_route};
use crate::auth_lys::{
    DeliberateDecision, HotAction, RefusalCause, SubjectBinding, SubjectBindingError,
};
use crate::auth_membership::{ConnectionMembership, MembershipAction};
use crate::auth_pass::{PassVerifier, VerifiedHolder, VerifiedPass};
use crate::auth_revocation::RevocationRefusal;
use crate::config::types::ServiceProfile;
use crate::console::FlowRole;
use crate::server::participant::{
    ConnectionFateClass, PARTICIPANT_CAPABILITY_BIT, ParticipantConnectionContext,
    ParticipantDispatch, ParticipantIngress, ParticipantSemanticHandler, constant_time_eq,
    dispatch_generic_frame, encode_server_value, gate_generic_frame,
};
use liminal_protocol::credential::WireCredential;
use liminal_protocol::wire::{
    ClientRequest, ConversationId, ParticipantTransportRejected, ServerValue,
    TransportRejectionReason,
};

const SERVER_ERROR_CODE: u16 = 0xFFFF;
/// Protocol versions the server negotiates: 1.0, and the version whose ack
/// carries the durable position. A client offering only 1.0 still selects 1.0.
const SUPPORTED_PROTOCOLS: [ProtocolVersion; 2] =
    [ProtocolVersion::new(1, 0), PUBLISH_ACK_POSITION_VERSION];

pub(super) fn apply_frame(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    frame: Frame,
) -> FrameAction {
    // Auth gate. When a token is configured, a connection must clear the `Connect`
    // handshake before any application frame is honoured. The token check itself
    // lives in `connect_response`, reached only via a `Connect` frame, so gating
    // solely there would let a client that skips `Connect` and sends
    // `Publish`/`Subscribe`/`WorkerRegister` straight through to the fan-out — this
    // per-frame gate closes that bypass, and the observability-tap arm below sits
    // behind it too. Frames without an application side effect are still admitted
    // pre-auth: `Connect` (which authenticates), `Disconnect` (a clean bow-out),
    // `Ping` (liveness), and every server-to-client/unknown frame the server ignores
    // unconditionally. Any other frame from an unauthenticated peer tears the
    // connection down with a silent `Close`. With no token configured this gate is
    // inert and the pre-auth behaviour is byte-identical.
    if rejects_unauthenticated(runtime, state, &frame) {
        return FrameAction::Close;
    }
    match frame {
        Frame::Connect {
            min_version,
            max_version,
            auth_token,
            ..
        } => connect_once(pid, runtime, state, min_version, max_version, &auth_token),
        Frame::Disconnect { .. } => {
            FrameAction::CloseWithFate(ConnectionFateClass::CleanDisconnect)
        }
        Frame::Ping { .. } => FrameAction::Respond(Frame::Pong { flags: 0 }),
        Frame::Publish {
            stream_id,
            channel,
            envelope,
            idempotency_key,
            ..
        } => publish_frame_response(
            pid,
            runtime,
            state,
            stream_id,
            &channel,
            &envelope,
            idempotency_key.as_deref(),
        ),
        Frame::Subscribe {
            stream_id,
            channel,
            accepted_schemas: schemas,
            max_in_flight: window,
            ..
        } => subscribe_frame_response(pid, runtime, state, stream_id, &channel, &schemas, window),
        Frame::Unsubscribe {
            stream_id,
            subscription_id,
            ..
        } => unsubscribe_frame_response(pid, runtime, state, stream_id, subscription_id),
        Frame::ConversationOpen {
            stream_id,
            conversation_id,
            subject,
            ..
        } => conversation_open(pid, runtime, state, stream_id, conversation_id, &subject),
        Frame::ConversationMessage {
            flags,
            stream_id,
            conversation_id,
            envelope,
        } => {
            let request = Request::new(flags, stream_id, conversation_id);
            super::service_tap::conversation_message(pid, runtime, state, request, &envelope)
        }
        Frame::ConversationClose {
            stream_id,
            conversation_id,
            ..
        } => conversation_close(runtime.services(), state, stream_id, conversation_id),
        Frame::PushReply {
            correlation_id,
            payload,
            ..
        } => push_reply_response(pid, runtime, state, correlation_id, payload),
        Frame::WorkerRegister { registration, .. } => {
            worker_register_frame_response(pid, runtime, state, registration)
        }
        // ACCESS-002 R1: a link is a machine's grant. Opening one end is
        // deliberate and asks Lys live; a crossing is decided from both
        // machines' passes before any effect.
        Frame::LinkOpen {
            stream_id, offer, ..
        } => link_open_frame_response(pid, runtime, state, stream_id, &offer),
        Frame::LinkMessage {
            stream_id,
            subject,
            sealed,
            ..
        } => link_message_frame_response(pid, runtime, state, stream_id, &subject, &sealed),
        frame @ Frame::Unknown {
            type_id: PARTICIPANT_FRAME_TYPE,
            ..
        } => participant_frame_response(runtime, state, &frame),
        // Client backpressure signals ride the subscription delivery machinery, so
        // they are application frames: gated by auth (they are NOT on the pre-auth
        // allowlist) and refused by a services profile that serves no
        // subscriptions.
        frame @ (Frame::Accept { .. } | Frame::Defer { .. } | Frame::Reject { .. }) => {
            apply_pressure_frame(pid, runtime, state, &frame)
        }
        // `Push`/`Deliver`/`WorkerRegisterAck` are server-to-client only; a client
        // must never originate one. Ignore these (and any stray/unknown inbound
        // frame) rather than treating them as fatal so a confused or malicious
        // client cannot tear the connection down with a stray frame.
        frame @ (Frame::Push { .. }
        | Frame::Deliver { .. }
        | Frame::WorkerRegisterAck { .. }
        | Frame::LinkOpenAck { .. }
        | Frame::DisconnectReason { .. }
        | Frame::Unknown { .. }
        | Frame::ConnectAck { .. }
        | Frame::ConnectError { .. }
        | Frame::SubscribeAck { .. }
        | Frame::SubscribeError { .. }
        | Frame::PublishAck { .. }
        | Frame::PublishError { .. }
        | Frame::ConversationError { .. }
        | Frame::Pong { .. }) => ignore_stray_inbound_frame(&frame),
    }
}

pub(super) fn rejects_unauthenticated(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    frame: &Frame,
) -> bool {
    // ACCESS-005: a revocation policy gates even an open-access server, so a
    // peer that skips `Connect` cannot reach an effect unbound.
    (runtime.auth_token().is_some()
        || runtime.pass_verifier().is_some()
        || runtime.revocation().is_some())
        && !state.authenticated
        && !permitted_before_auth(frame)
}

/// `PushReply` is decided as `serve` on the subject the push was made on (the
/// access matrix's `push_binding`): the subject is read back from the reply
/// slot, decided through the connection's Lys session, and a refusal settles
/// the slot as [`ServerError::PushReplyRefused`](crate::ServerError::PushReplyRefused)
/// so the pusher learns it by type. A push made with no subject is refused the
/// same way on a Lys-admitted connection: the binding is recorded at push time
/// or not at all. A connection without a Lys session keeps the V1 path.
fn push_reply_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    correlation_id: u64,
    payload: Vec<u8>,
) -> FrameAction {
    let refusal = match runtime.push_binding(pid, correlation_id) {
        PushBinding::Unbound => None,
        PushBinding::Subjectless => state.lys_session.is_some().then(|| {
            "the push carries no subject binding; a Lys-admitted connection's reply is decided on the pushed subject".to_owned()
        }),
        // ACCESS-006 R3: a reply is an outbound service effect on a protected
        // channel, so it needs post there beside the serve grant.
        PushBinding::Subject(subject) => lys_refusal(runtime, state, &subject, HotAction::Serve)
            .or_else(|| membership_refusal(state, &subject, MembershipAction::Post)),
    };
    if let Some(reason) = refusal {
        tracing::warn!(
            pid,
            correlation_id,
            reason = reason.as_str(),
            "refused PushReply on the subject its push was made on"
        );
        log_push_disposition(
            pid,
            correlation_id,
            runtime.refuse_push(pid, correlation_id, reason),
        );
        return FrameAction::NoResponse;
    }
    // Resolving the one-shot slot completes the round trip without another frame.
    log_push_disposition(
        pid,
        correlation_id,
        runtime.resolve_push(pid, correlation_id, payload),
    );
    FrameAction::NoResponse
}

fn log_push_disposition(pid: u64, correlation_id: u64, disposition: PushReplyDisposition) {
    match disposition {
        PushReplyDisposition::Resolved => {}
        PushReplyDisposition::SenderMismatch => tracing::warn!(
            pid,
            correlation_id,
            reply_disposition = "sender_mismatch",
            "discarded PushReply from a connection other than its recipient"
        ),
        PushReplyDisposition::UnknownCorrelation => tracing::debug!(
            pid,
            correlation_id,
            reply_disposition = "unknown_correlation",
            "discarded PushReply for a missing slot"
        ),
        PushReplyDisposition::ReceiverDropped => tracing::debug!(
            pid,
            correlation_id,
            reply_disposition = "receiver_dropped",
            "discarded PushReply after its receiver was abandoned"
        ),
    }
}

fn apply_pressure_frame(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    frame: &Frame,
) -> FrameAction {
    let (stream, reference, response) = match frame {
        Frame::Accept {
            stream_id,
            referenced_message_id,
            ..
        } => (*stream_id, referenced_message_id.as_str(), "accept"),
        Frame::Defer {
            stream_id,
            referenced_message_id,
            ..
        } => (*stream_id, referenced_message_id.as_str(), "defer"),
        Frame::Reject {
            stream_id,
            referenced_message_id,
            ..
        } => (*stream_id, referenced_message_id.as_str(), "reject"),
        other => return ignore_stray_inbound_frame(other),
    };
    pressure_frame_response(pid, runtime, state, stream, reference, response)
}

fn pressure_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    reference: &str,
    response: &'static str,
) -> FrameAction {
    let action = pressure_response(runtime.services(), stream_id);
    if !runtime.services().supports_channel_operations() {
        return action;
    }
    // Accept/Defer/Reject are decided as `subscribe` on the subject of the
    // subscription they signal for (the matrix's `subscription_binding`),
    // before the ledger is touched. The signal has no refusal frame: a refused
    // signal has no effect, is logged with its refusal and counted.
    let subjects: Vec<String> = state
        .subscriptions
        .values()
        .filter(|subscription| subscription.stream_id() == stream_id)
        .map(|subscription| subscription.channel().to_owned())
        .collect();
    for subject in &subjects {
        if let Some(reason) = lys_refusal(runtime, state, subject, HotAction::Subscribe)
            .or_else(|| membership_refusal(state, subject, MembershipAction::Read))
        {
            tracing::warn!(
                pid,
                stream_id,
                response,
                reason = reason.as_str(),
                "refused backpressure signal on its subscription's subject"
            );
            crate::metrics::pressure_refused();
            return action;
        }
    }
    let matched = state.subscriptions.values_mut().find_map(|subscription| {
        if subscription.stream_id() == stream_id
            && subscription.match_pressure(reference, response == "defer")
        {
            Some(subscription)
        } else {
            None
        }
    });
    if let Some(subscription) = matched {
        if matches!(subscription.channel(), "liminal.events" | "liminal.cost") {
            crate::metrics::self_stream_pressure_suppressed();
        } else {
            runtime.pressure(pid, response);
        }
    } else {
        crate::metrics::self_stream_pressure_unmatched();
    }
    action
}

/// The catch-all's body: warns with the frame-kind name, then keeps the
/// no-teardown `NoResponse` semantics.
fn ignore_stray_inbound_frame(frame: &Frame) -> FrameAction {
    warn_ignored_inbound_frame(ignored_inbound_frame_kind(frame));
    FrameAction::NoResponse
}

/// Names the frame variant for the ignored-inbound warn. Kind names only — no
/// payload is Debug-formatted, so a stray frame cannot spray opaque bytes into
/// the log.
const fn ignored_inbound_frame_kind(frame: &Frame) -> &'static str {
    match frame {
        Frame::Push { .. } => "push",
        Frame::Deliver { .. } => "deliver",
        Frame::WorkerRegisterAck { .. } => "worker_register_ack",
        Frame::LinkOpenAck { .. } => "link_open_ack",
        Frame::DisconnectReason { .. } => "disconnect_reason",
        Frame::Unknown { .. } => "unknown",
        Frame::ConnectAck { .. } => "connect_ack",
        Frame::ConnectError { .. } => "connect_error",
        Frame::SubscribeAck { .. } => "subscribe_ack",
        Frame::SubscribeError { .. } => "subscribe_error",
        Frame::PublishAck { .. } => "publish_ack",
        Frame::PublishError { .. } => "publish_error",
        Frame::ConversationError { .. } => "conversation_error",
        Frame::Pong { .. } => "pong",
        // Unreachable from the ignored arm (every other kind has its own
        // dispatch above); named honestly rather than panicking.
        _ => "other",
    }
}

/// Logs the ignored stray inbound frame. The no-teardown semantics are the
/// caller's; this is visibility only.
fn warn_ignored_inbound_frame(kind: &'static str) {
    tracing::warn!(
        frame_kind = kind,
        "ignored stray inbound frame; server-to-client-only or unknown frame kinds get no response and no teardown"
    );
}

/// Names a broken activation invariant at the point it closes the connection.
///
/// A complete participant service and a durable connection incarnation are one
/// invariant; either alone is internal configuration drift, not a client fault.
/// Closing is correct — the alternative is fabricating a lifecycle outcome — but
/// closing SILENTLY left an operator watching connections drop at the handshake
/// with nothing to read, while the sibling semantic-failure arm in the same
/// match already warned.
fn warn_partial_activation(stage: &'static str, service_installed: bool) {
    tracing::warn!(
        stage,
        service_installed,
        connection_incarnation_present = !service_installed,
        "closing on partial participant activation; a complete participant service \
         and a durable connection incarnation are one invariant and exactly one is \
         present (internal configuration drift, not a client fault)"
    );
}

/// Applies the core protocol's single-use connection handshake.
fn connect_once(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    min_version: ProtocolVersion,
    max_version: ProtocolVersion,
    auth_token: &[u8],
) -> FrameAction {
    // The core lifecycle excludes `Connect` from its Active state; accepting a
    // second one would silently renegotiate capability/session state in place.
    if state.authenticated {
        FrameAction::Close
    } else {
        connect_response(pid, runtime, state, min_version, max_version, auth_token)
    }
}

/// Applies the shared participant transport gate before semantic dispatch.
///
/// Semantic request handling is installed by the participant lifecycle service;
/// this boundary already guarantees that malformed, unauthenticated, and
/// capability-missing frames receive the crate's exact typed rejection.
fn participant_frame_response(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    frame: &Frame,
) -> FrameAction {
    match (runtime.participant_service(), state.connection_incarnation) {
        (Some(service), Some(connection_incarnation)) => {
            if let Some(refused) = participant_lys_refusal(runtime, state, service, frame) {
                return refused;
            }
            match dispatch_generic_frame(
                frame,
                state.authenticated,
                state.participant_session,
                // The mount fact is the CONNECTION's, stamped by its spawn path
                // before any inbound byte was read (design §10). It is read off
                // the connection state here — never derived from `frame`, whose
                // contents are exactly the thing a client controls.
                ParticipantConnectionContext::new(connection_incarnation, state.mount),
                state.pass_principal.as_ref(),
                &mut state.participant_conversations,
                service,
            ) {
                ParticipantDispatch::NotParticipant => FrameAction::NoResponse,
                ParticipantDispatch::Respond(response) => FrameAction::Respond(response),
                ParticipantDispatch::RespondThenClose(response) => {
                    FrameAction::RespondThenClose(response)
                }
                ParticipantDispatch::Fatal(error) => {
                    tracing::warn!(%error, "participant request failed closed");
                    FrameAction::Close
                }
            }
        }
        (None, None) => participant_frame_without_service(state, frame),
        // A complete service and a durable connection incarnation are one
        // activation invariant. Any mismatch is internal configuration drift;
        // close without fabricating a lifecycle outcome.
        (Some(_), None) | (None, Some(_)) => {
            warn_partial_activation(
                "participant frame dispatch",
                runtime.participant_service().is_some(),
            );
            FrameAction::Close
        }
    }
}

/// The participant routes' access decision for a connection admitted on a Lys
/// pass (the matrix's `participant_binding`): the decoded request is decided,
/// on every conversation it names, as the action its route is classified
/// with, before dispatch sees it. A refusal answers the participant transport
/// rejection `AccessRefused` carrying the refusal, without closing. `None`
/// means proceed: no Lys session, no decoded request, or every decision made.
fn participant_lys_refusal(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    service: &dyn ParticipantSemanticHandler,
    frame: &Frame,
) -> Option<FrameAction> {
    // A connection bound under a revocation policy (ACCESS-005 R4) is
    // decided here too, whether it was admitted on a Lys pass or not.
    if state.lys_session.is_none() && state.revocation.is_none() {
        return None;
    }
    let ParticipantIngress::Request(request) =
        gate_generic_frame(frame, state.authenticated, state.participant_session)
    else {
        return None;
    };
    let refusal = participant_access_refusal(runtime, state, service, &request)?;
    tracing::warn!(
        route = ?request.discriminant(),
        reason = refusal.as_str(),
        "refused participant request on its conversation's subject"
    );
    let rejected = ServerValue::ParticipantTransportRejected(ParticipantTransportRejected {
        reason: TransportRejectionReason::AccessRefused { refusal },
    });
    match encode_server_value(rejected) {
        Ok(frame) => Some(FrameAction::Respond(frame)),
        Err(error) => {
            tracing::warn!(
                error = ?error,
                "a participant access refusal could not be encoded; closing"
            );
            Some(FrameAction::Close)
        }
    }
}

/// The decision behind [`participant_lys_refusal`]: `Some` is the first
/// refusal over the request's conversations, in route order; `None` is every
/// decision made. A conversation no authority maps to a subject is refused by
/// name: a subject is never guessed from the numeric id.
fn participant_access_refusal(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    service: &dyn ParticipantSemanticHandler,
    request: &ClientRequest,
) -> Option<String> {
    let action = match classify_participant_route(request.discriminant()) {
        RouteClassification::Hot { action, .. } | RouteClassification::Cleanup { action, .. } => {
            action
        }
        RouteClassification::Admission
        | RouteClassification::Control
        | RouteClassification::ServerOnly
        | RouteClassification::ParticipantEnvelope
        | RouteClassification::Deliberate { .. }
        | RouteClassification::Crossing { .. }
        | RouteClassification::Unknown => {
            return Some(format!(
                "participant route {:?} has no hot or cleanup class; refused",
                request.discriminant()
            ));
        }
    };
    participant_conversations(request)
        .into_iter()
        .find_map(|conversation_id| {
            // ACCESS-006 R1: the subject comes from the handler's own
            // authority, else from the membership mapping the operator's
            // product schema supplies; never a guess from the id. A mapped
            // conversation is then decided on its pass rights and its
            // channel membership both.
            let subject = service.conversation_subject(conversation_id).or_else(|| {
                runtime
                    .membership_gate()
                    .and_then(|gate| gate.channel_for_conversation(conversation_id))
                    .map(str::to_owned)
            });
            subject.map_or_else(
                || {
                    Some(format!(
                        "no subject mapping for conversation {conversation_id}; a Lys-admitted connection's participant request is decided on its conversation's subject"
                    ))
                },
                |subject| {
                    lys_refusal(runtime, state, &subject, action).or_else(|| {
                        membership_refusal(state, &subject, MembershipAction::for_hot(action))
                    })
                },
            )
        })
}

/// Every conversation a participant request names, in request order. A
/// recovery batch names one per refusal it carries, each decided before any
/// effect.
fn participant_conversations(request: &ClientRequest) -> Vec<ConversationId> {
    match request {
        ClientRequest::Enrollment(request) => vec![request.conversation_id],
        ClientRequest::CredentialAttach(request) => vec![request.conversation_id],
        ClientRequest::Detach(request) => vec![request.conversation_id],
        ClientRequest::ParticipantAck(request) => vec![request.conversation_id],
        ClientRequest::Leave(request) => vec![request.conversation_id],
        ClientRequest::MarkerAck(request) => vec![request.conversation_id],
        ClientRequest::RecordAdmission(request) => vec![request.conversation_id],
        ClientRequest::SettlementQuery(request) => vec![request.conversation_id],
        ClientRequest::ObserverRecovery(request) => request
            .observer_refusals
            .iter()
            .map(|refusal| refusal.conversation_id)
            .collect(),
    }
}

/// Applies only the crate-owned pre-semantic gate while participant capability
/// is disabled. A decoded request cannot occur because the session is missing.
fn participant_frame_without_service(state: &ConnectionProcessState, frame: &Frame) -> FrameAction {
    match gate_generic_frame(frame, state.authenticated, state.participant_session) {
        ParticipantIngress::NotParticipant => FrameAction::NoResponse,
        // A decoded request is unreachable while the capability is deliberately
        // not advertised. Close if configuration drift ever makes it reachable;
        // silently consuming an authorized lifecycle operation would make a
        // retry indistinguishable from a lost commit. A locally unrepresentable
        // generic frame is likewise terminal.
        ParticipantIngress::Request(_) | ParticipantIngress::InvalidGenericFrame => {
            FrameAction::Close
        }
        ParticipantIngress::Rejected(rejection) => {
            let reason = format!("{:?}", rejection.reason);
            match encode_server_value(
                liminal_protocol::wire::ServerValue::ParticipantTransportRejected(rejection),
            ) {
                Ok(response) => FrameAction::RespondThenClose(response),
                Err(error) => {
                    // The client gets a bare close instead of the typed rejection
                    // it was owed, so neither the encode failure nor the reason it
                    // was rejected exists anywhere unless logged here.
                    tracing::warn!(
                        rejection_reason = %reason,
                        ?error,
                        "participant transport rejection could not be encoded; closing \
                         without telling the client why"
                    );
                    FrameAction::Close
                }
            }
        }
    }
}

/// Client backpressure signals (`Accept`/`Defer`/`Reject`). Full mode consumes
/// them with no wire response, exactly as before. The worker front door serves no
/// subscriptions — no delivery can exist for a client to signal pressure on — so
/// the frame is rejected on the subscription error channel (`SubscribeError`, the
/// same closest-honest-fit vocabulary as the unsubscribe rejection).
fn pressure_response(services: &dyn ConnectionServices, stream_id: u32) -> FrameAction {
    if services.supports_channel_operations() {
        FrameAction::NoResponse
    } else {
        FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(unsupported_by_front_door("backpressure signaling")),
        })
    }
}

/// Renders the worker-front-door refusal text for `operation` from the typed
/// [`ServerError::UnsupportedOperation`] variant, so every front-door rejection —
/// whether raised in the services adapter or short-circuited here — speaks the one
/// error vocabulary.
fn unsupported_by_front_door(operation: &str) -> String {
    ServerError::UnsupportedOperation {
        operation: operation.to_owned(),
        profile: ServiceProfile::WORKER_FRONT_DOOR,
    }
    .to_string()
}

/// Associates a worker registration with this connection and invokes the
/// configured connection notifier.
///
/// The notifier is consulted FIRST: only after the application accepts (or when
/// no notifier is configured) is the registration stored on the connection
/// record, so the close-path `on_worker_unregistered` fires for exactly the
/// connections the application accepted — a rejected worker leaves no record and
/// triggers no later deregistration. The ack is synchronous: a notifier error
/// yields a `Rejected` ack carrying the reason so the worker never believes it is
/// registered after the application declined it. With no notifier configured the
/// registration is accepted unconditionally, keeping liminal usable standalone.
/// A `WorkerRegister` frame; an accepted one makes this connection a worker,
/// whose flows are line stops from now on (SRV-012 R7).
fn worker_register_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    registration: WorkerRegistration,
) -> FrameAction {
    // ACCESS-004 R2: a connection that proved who it is registers under
    // that name and no other.
    // R3: an operator who requires a holder refuses a connection that
    // proved none.
    match state.verified_holder.as_ref() {
        Some(holder) if !holder.is_named(&registration.identity) => {
            return worker_register_rejected(REGISTRATION_NOT_HOLDER.to_owned());
        }
        None if runtime
            .pass_verifier()
            .is_some_and(PassVerifier::requires_verified_worker) =>
        {
            return worker_register_rejected(REGISTRATION_WITHOUT_HOLDER.to_owned());
        }
        Some(_) | None => {}
    }
    // ACCESS-001 R2, `WorkerRegister` = Hot(Serve, WorkerBinding): every
    // namespace the registration explicitly serves is decided from the
    // admitted pass before the registration reaches the notifier or the
    // record. The first namespace the pass lacks refuses the whole frame.
    for namespace in &registration.namespaces {
        if let Some(refused) = lys_decision(runtime, state, namespace, HotAction::Serve, |reason| {
            Frame::WorkerRegisterAck {
                flags: 0,
                outcome: WorkerRegisterOutcome::Rejected { reason },
            }
        }) {
            return refused;
        }
    }
    // SRV-028 R4: a service process registers its own subjects only, and
    // its first accepted registration makes its service live.
    let process = state.service_process.clone();
    if let Some(process) = process.as_ref()
        && !process.serves(&registration.task_queue)
    {
        return worker_register_rejected(
            "service_process_scope: a service process registers its own subjects only".to_owned(),
        );
    }
    let action =
        worker_register_response(pid, runtime, registration, state.verified_holder.as_ref());
    if let Some(process) = process
        && matches!(
            action,
            FrameAction::Respond(Frame::WorkerRegisterAck {
                outcome: WorkerRegisterOutcome::Accepted,
                ..
            })
        )
        && let Some(lifecycle) = runtime.lifecycle()
        && let Err(error) =
            lifecycle.mark_live(&process.service, unix_seconds_now().saturating_mul(1000))
    {
        tracing::warn!(service = %process.service, %error, "a live service could not be recorded");
    }
    if matches!(
        action,
        FrameAction::Respond(Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        })
    ) {
        state.worker_flows.get_or_insert_with(BTreeMap::new);
    }
    action
}

fn worker_register_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    registration: WorkerRegistration,
    holder: Option<&VerifiedHolder>,
) -> FrameAction {
    if let Some(notifier) = runtime.notifier() {
        if let Err(error) = notifier.on_worker_registered_by(pid, &registration, holder) {
            return worker_register_rejected(error.to_string());
        }
    }
    // Store only after acceptance. A poisoned-registry error here means the
    // accepted registration cannot be tracked for deregistration, so reject the
    // worker (and undo the application-side registration) rather than leave a
    // silent, never-deregistered association.
    if let Err(error) = runtime.set_registration(pid, registration) {
        if let Some(notifier) = runtime.notifier() {
            notifier.on_worker_unregistered(pid);
        }
        return worker_register_rejected(error.to_string());
    }
    FrameAction::Respond(Frame::WorkerRegisterAck {
        flags: 0,
        outcome: WorkerRegisterOutcome::Accepted,
    })
}

/// The `Rejected` reason for a registration whose name is not the verified
/// holder's. It names neither, so no holder bytes reach the worker's log.
const REGISTRATION_NOT_HOLDER: &str = "registration identity is not the verified holder";

/// The `Rejected` reason for a registration from a connection that proved no
/// holder, when the operator requires one.
const REGISTRATION_WITHOUT_HOLDER: &str = "worker registration requires a verified holder";

const fn worker_register_rejected(reason: String) -> FrameAction {
    FrameAction::Respond(Frame::WorkerRegisterAck {
        flags: 0,
        outcome: WorkerRegisterOutcome::Rejected { reason },
    })
}

/// Whether `frame` may be processed before the connection clears the auth gate.
///
/// Default-deny: only the handshake (`Connect`), a clean disconnect, a liveness
/// `Ping`, and the frames the server ignores unconditionally (server-to-client and
/// unknown kinds, which return `NoResponse` regardless) are admitted. Every
/// application frame — publish, subscribe, unsubscribe, conversation, worker-register,
/// push-reply, and the `Accept`/`Defer`/`Reject` backpressure signals (which act on
/// subscription delivery state) — is held back until a successful `Connect`, so a
/// newly added frame is gated by default until it is explicitly listed here.
const fn permitted_before_auth(frame: &Frame) -> bool {
    matches!(
        frame,
        Frame::Connect { .. }
            | Frame::Disconnect { .. }
            | Frame::Ping { .. }
            | Frame::Push { .. }
            | Frame::Deliver { .. }
            | Frame::WorkerRegisterAck { .. }
            | Frame::LinkOpenAck { .. }
            | Frame::DisconnectReason { .. }
            | Frame::Unknown { .. }
            | Frame::ConnectAck { .. }
            | Frame::ConnectError { .. }
            | Frame::SubscribeAck { .. }
            | Frame::SubscribeError { .. }
            | Frame::PublishAck { .. }
            | Frame::PublishError { .. }
            | Frame::ConversationError { .. }
            | Frame::Pong { .. }
    )
}

/// Whether the connection may use `channel`, decided from the holder its
/// `Connect` verified (ACCESS-004 R4): a holder may use only channels under
/// its pass's live prefix, and a connection that proved no holder is not
/// restricted here. Rights from a Lys pass wait for ACCESS-001 R2.
/// `ConnectError` for a Lys pass record the admitter refused, naming the
/// refusal and never the pass bytes.
fn lys_connect_refused(reason: &str) -> FrameAction {
    let error = ProtocolError::AuthenticationFailure {
        message: Some(format!("lys pass admission refused: {reason}")),
    };
    FrameAction::RespondThenClose(Frame::ConnectError {
        flags: 0,
        reason_code: error.reason_code(),
        message: error.message().map(str::to_owned),
    })
}

/// Seconds since the Unix epoch, the `now` every Lys decision is made at.
pub(super) fn unix_seconds_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_secs())
}

/// Decides one hot action on `channel` from the connection's admitted Lys
/// pass, offline. `None` lets the operation proceed: either the pass allows
/// it, or the connection holds no Lys pass and the V1 rules apply instead.
/// `Some` is the refusal frame in the one shape: the four-field refusal as
/// the message, under the route's refused reason code.
fn lys_decision(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    channel: &str,
    action: HotAction,
    refused: impl FnOnce(String) -> Frame,
) -> Option<FrameAction> {
    lys_refusal(runtime, state, channel, action)
        .or_else(|| membership_refusal(state, channel, MembershipAction::for_hot(action)))
        .map(|message| FrameAction::Respond(refused(message)))
}

/// ACCESS-006 R1, R3: a membership-protected channel's admission, decided on
/// the connection's verified subject after (and in addition to) its pass
/// rights, before any effect. `None` proceeds: the channel is not protected,
/// or the membership allows the act. `Some` is the refusal's stable name and
/// words, naming no resource and no holder.
fn membership_refusal(
    state: &ConnectionProcessState,
    channel: &str,
    action: MembershipAction,
) -> Option<String> {
    let refusal = state.membership.check(channel, action).err()?;
    tracing::warn!(
        refusal = refusal.name(),
        action = action.as_str(),
        "refused a membership-protected act"
    );
    Some(refusal.to_string())
}

/// The decision behind [`lys_decision`], for the routes that carry no refusal
/// frame (`PushReply`, `Accept`/`Defer`/`Reject`): `None` is the same
/// "proceed" as there; `Some` is the refusal message in the one shape.
fn lys_refusal(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    channel: &str,
    action: HotAction,
) -> Option<String> {
    // SRV-028 R4: a service process serves its own subjects and does
    // nothing else on this connection.
    if let Some(process) = state.service_process.as_ref()
        && !(action == HotAction::Serve && process.serves(channel))
    {
        return Some(
            "service_process_scope: a service process serves its own subjects and nothing else"
                .to_owned(),
        );
    }
    // ACCESS-005 R4: a revoked right is refused first, by name, from the
    // connection's own bound rights; no Lys call is made here.
    if let Some(grants) = state.revocation.as_ref()
        && let Err(refusal) = grants.allows(action, channel)
    {
        return Some(refusal.name().to_owned());
    }
    let session = state.lys_session.as_ref()?;
    let Some((_, network_id)) = runtime.lys_admitter() else {
        return Some("no lys admitter is installed".to_owned());
    };
    let subject = match SubjectBinding::new(
        network_id.to_owned(),
        channel.to_owned(),
        channel.to_owned(),
    ) {
        Ok(subject) => subject,
        Err(SubjectBindingError(member)) => {
            return Some(format!("subject binding refused: {member}"));
        }
    };
    let now = unix_seconds_now();
    let error = match session.decide(&subject, action, now) {
        Ok(_receipt) => return None,
        Err(error) => error,
    };
    // SRV-028: a service placed after this pass was issued is not among its
    // rights, so the offline refusal is for want of a placement and not of a
    // grant: Lys is asked live, once per such act, until the pass renews.
    let placed_after_pass = error.cause == RefusalCause::MissingGrant
        && session.issued_at().is_some_and(|issued| {
            runtime.lifecycle().is_some_and(|lifecycle| {
                lifecycle.placed_since(channel, issued.saturating_mul(1000))
            })
        });
    if placed_after_pass
        && let Ok(DeliberateDecision::Outright(_receipt)) =
            session.decide_live(subject.resource(), action.into(), now)
    {
        return None;
    }
    Some(serde_json::to_string(&error.refusal).unwrap_or_else(|_| error.cause.to_string()))
}

fn pass_allows_channel(state: &ConnectionProcessState, channel: &str) -> bool {
    state
        .verified_holder
        .as_ref()
        .is_none_or(|holder| holder.allows_channel(channel))
}

fn publish_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    channel: &str,
    envelope: &MessageEnvelope,
    idempotency_key: Option<&str>,
) -> FrameAction {
    if let Some(refused) = lys_decision(runtime, state, channel, HotAction::Publish, |message| {
        Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: 0x0202,
            message: Some(message),
        }
    }) {
        return refused;
    }
    if !pass_allows_channel(state, channel) {
        return FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: 0x0202,
            message: Some(format!("publish refused for channel {channel}")),
        });
    }
    // SRV-012 R1: the console guard comes first, so a notifier is never
    // offered a console subject's payload.
    if crate::console::tappable(channel)
        && runtime.notifier_channel_publish(pid, channel, &envelope.payload)
    {
        FrameAction::NoResponse
    } else {
        let action = publish_response(
            pid,
            runtime.services(),
            state.negotiated_version,
            stream_id,
            channel,
            envelope,
            idempotency_key,
        );
        flowed(pid, runtime, state, channel, FlowRole::Publishes, action)
    }
}

/// SRV-012 R7: a registered worker's first acknowledged publish or subscribe
/// on `channel` tells the topology which way connection `pid` uses it, then
/// `action` is answered unchanged. A repeat, or any other connection's, reads
/// connection state only: no allocation and no console lock.
fn flowed(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    channel: &str,
    role: FlowRole,
    action: FrameAction,
) -> FrameAction {
    let bit = flow_bit(role);
    if let Some(flows) = state.worker_flows.as_mut()
        && flows.get(channel).is_none_or(|known| known & bit == 0)
        && matches!(
            action,
            FrameAction::Respond(Frame::PublishAck { .. } | Frame::SubscribeAck { .. })
        )
    {
        *flows.entry(channel.to_owned()).or_default() |= bit;
        runtime.flow(pid, channel, role, true);
    }
    action
}

/// The bit a [`FlowRole`] holds in [`ConnectionProcessState::flows`].
const fn flow_bit(role: FlowRole) -> u8 {
    match role {
        FlowRole::Publishes => 1,
        FlowRole::Subscribes => 2,
    }
}

fn subscribe_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    channel: &str,
    accepted_schemas: &[ProtocolSchemaId],
    max_in_flight: u32,
) -> FrameAction {
    if let Some(refused) = lys_decision(runtime, state, channel, HotAction::Subscribe, |message| {
        Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: 0x0201,
            message: Some(message),
        }
    }) {
        return refused;
    }
    if pass_allows_channel(state, channel) {
        let action = subscribe_response(
            pid,
            runtime,
            state,
            stream_id,
            channel,
            accepted_schemas,
            max_in_flight,
        );
        flowed(pid, runtime, state, channel, FlowRole::Subscribes, action)
    } else {
        FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: 0x0201,
            message: Some(format!("subscribe refused for channel {channel}")),
        })
    }
}

/// The `ConnectError` closing a connection whose credential failed its check.
fn authentication_failed(message: String) -> FrameAction {
    let error = ProtocolError::AuthenticationFailure {
        message: Some(message),
    };
    FrameAction::RespondThenClose(Frame::ConnectError {
        flags: 0,
        reason_code: error.reason_code(),
        message: error.message().map(str::to_owned),
    })
}

/// Admits the connection's credential before version negotiation, so an
/// unauthenticated peer learns nothing about the negotiated protocol: the
/// configured bearer token, a Lys pass record through the installed admitter
/// (offline, or refused by name), or a V1 pass through the verifier; then
/// the ACCESS-005 revocation fence, which refuses a connection with no bound
/// pass by name. When no token is configured the server is open and the
/// bearer check is skipped, keeping the handshake byte-identical to the
/// pre-auth behaviour. `Err` is the frame action that answers a refusal.
fn admit_credential(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    (min_version, max_version): (ProtocolVersion, ProtocolVersion),
    auth_token: &[u8],
) -> Result<(), Box<FrameAction>> {
    // SRV-028 R4: a service process's own credential admits it to serve
    // its subjects and nothing else; it is never the bearer.
    if let Some(grant) = runtime
        .lifecycle()
        .and_then(|lifecycle| lifecycle.admit_process(auth_token))
    {
        state.service_process = Some(grant);
        return Ok(());
    }
    let bearer_matches = runtime
        .auth_token()
        .is_some_and(|expected| constant_time_eq(expected, auth_token));
    if !bearer_matches {
        if let Ok(WireCredential::LysPass(pass)) = WireCredential::classify(auth_token) {
            // A Lys pass record: admitted by the installed admitter, offline,
            // or refused by name. It never falls through to the V1 checker.
            let Some((admitter, _)) = runtime.lys_admitter() else {
                return Err(Box::new(lys_connect_refused(
                    "no lys admitter is installed",
                )));
            };
            match admitter.admit(pass, unix_seconds_now()) {
                Ok(session) => {
                    // ACCESS-004 R1's second producer: the Lys holder is the
                    // connection's one proof of who it is, so a worker
                    // registers under that name and no other (R2).
                    state.verified_holder =
                        Some(VerifiedHolder::from_lys_session(session.as_ref()));
                    state.lys_session = Some(std::sync::Arc::from(session));
                }
                Err(refusal) => return Err(Box::new(lys_connect_refused(&refusal.reason))),
            }
        } else if let Some(verifier) = runtime.pass_verifier() {
            match verifier.verify(auth_token) {
                Ok(VerifiedPass { principal, holder }) => {
                    // ACCESS-005 R3: under a revocation policy the binding is
                    // fenced before any attach is announced.
                    revocation_connect(
                        pid,
                        runtime,
                        state,
                        (min_version, max_version),
                        auth_token,
                        holder.participant(),
                    )?;
                    runtime.pass_attached(pid, state.mount, holder.participant());
                    state.pass_principal = Some(principal);
                    state.verified_holder = Some(holder);
                }
                Err(failure) => {
                    return Err(Box::new(authentication_failed(format!(
                        "connection pass {} check failed",
                        failure.check_name()
                    ))));
                }
            }
        } else if runtime.auth_token().is_some() {
            return Err(Box::new(authentication_failed(
                "connection authentication token rejected".to_owned(),
            )));
        }
    }
    // ACCESS-005 R1/R3: a Lys-pass connection is fenced here, and a bearer or
    // open-access one, which carries no bound pass, is refused by name.
    if runtime.revocation().is_some() && state.revocation.is_none() {
        let holder = state
            .verified_holder
            .as_ref()
            .map(|holder| holder.participant().to_vec());
        holder.map_or_else(
            || {
                Err(Box::new(revocation_connect_refused(
                    RevocationRefusal::BindingMissing,
                )))
            },
            |holder| {
                revocation_connect(
                    pid,
                    runtime,
                    state,
                    (min_version, max_version),
                    auth_token,
                    &holder,
                )
            },
        )?;
    }
    Ok(())
}

fn connect_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    min_version: ProtocolVersion,
    max_version: ProtocolVersion,
    auth_token: &[u8],
) -> FrameAction {
    if let Err(refused) =
        admit_credential(pid, runtime, state, (min_version, max_version), auth_token)
    {
        return *refused;
    }
    // ACCESS-006 R1: membership is bound to the subject this same
    // verification proved; under a scoped-guest mapping a Lys holder must
    // hold a current grant in the workspace before anything is admitted.
    let membership = ConnectionMembership::bind(
        runtime.membership_gate().cloned(),
        pid,
        state.lys_session.as_deref(),
    );
    if state.lys_session.is_some()
        && let Err(refusal) = membership.admit_connect()
    {
        return lys_connect_refused(refusal.name());
    }
    state.membership = membership;
    let selected_version =
        match negotiate_version(min_version, max_version, supported_protocols(runtime)) {
            Ok(selected_version) => selected_version,
            Err(error) => {
                return FrameAction::RespondThenClose(Frame::ConnectError {
                    flags: 0,
                    reason_code: error.reason_code(),
                    message: error.message().map(str::to_owned),
                });
            }
        };
    let (participant_session, capabilities) =
        match (runtime.participant_service(), state.connection_incarnation) {
            (Some(service), Some(_)) => {
                let mut participant_session =
                    crate::server::participant::ParticipantSession::default();
                participant_session.negotiate_v1(service.frame_limit());
                (participant_session, PARTICIPANT_CAPABILITY_BIT)
            }
            // A generic durable channel store alone does not activate the participant
            // protocol. Both service and incarnation are absent.
            (None, None) => (crate::server::participant::ParticipantSession::default(), 0),
            // Fail closed on impossible partial activation without inventing a
            // participant response.
            (Some(_), None) | (None, Some(_)) => {
                warn_partial_activation(
                    "connection handshake",
                    runtime.participant_service().is_some(),
                );
                return FrameAction::Close;
            }
        };

    // A pass-stamped connection announces its attach exactly once, HERE —
    // after authentication, version negotiation, and participant posture have
    // all admitted it — never on the verify alone, so a pass that verified but
    // failed a later check announces nothing and no detach can follow an
    // attach that never fired. The runtime stores the principal on the
    // connection record so the one close funnel fires the matching detach. A
    // principal the runtime cannot record is a connection whose detach could
    // never fire, so the handshake fails closed rather than attach silently.
    if let Some(principal) = state.pass_principal.as_ref() {
        if let Err(error) = runtime.attach_pass_principal(pid, principal) {
            tracing::warn!(
                %error,
                connection_pid = pid,
                "pass-stamped connection could not record its principal; closing"
            );
            return FrameAction::Close;
        }
    }

    // Publish handshake state only after authentication, version negotiation,
    // and participant posture have all succeeded. Every rejection above leaves
    // the pre-handshake state untouched.
    state.authenticated = true;
    state.negotiated_version = Some(selected_version);
    state.participant_session = participant_session;
    FrameAction::Respond(Frame::ConnectAck {
        flags: 0,
        selected_version,
        capabilities,
    })
}

pub(super) fn publish_response(
    pid: u64,
    services: &dyn ConnectionServices,
    negotiated_version: Option<ProtocolVersion>,
    stream_id: u32,
    channel: &str,
    envelope: &MessageEnvelope,
    idempotency_key: Option<&str>,
) -> FrameAction {
    if let Some(error) = reserved_namespace_refusal(channel) {
        return FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(error.to_string()),
        });
    }
    // Admission BEFORE delegation. A roster refusal is decided here, by the
    // component holding the roster, and stamped with its own reason code at the
    // moment of the decision — so the code cannot disagree with the read that
    // produced it. Nothing is published against a refused channel: the service is
    // not called at all.
    if let Err(refusal) = services.admit_channel(ChannelOperation::Publish, channel) {
        return FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: refusal.reason_code(),
            message: Some(access_to_server_error(&refusal).to_string()),
        });
    }
    match services.publish(pid, channel, envelope, idempotency_key) {
        Ok(outcome) => FrameAction::Respond(Frame::PublishAck {
            // Set the genuine-delivery flag only when the publish was accepted by
            // at least one subscriber. The ack is always sent on success (the
            // backpressure contract is unchanged); the flag bit is the additive
            // delivery-ack signal the caller can observe.
            flags: if outcome.delivered {
                PUBLISH_DELIVERED_FLAG
            } else {
                0
            },
            stream_id,
            message_id: outcome.message_id,
            // The position rides the ack only on a connection that negotiated the
            // version carrying it; a 1.0 peer keeps the older ack, byte for byte.
            durable_position: outcome.durable_position.filter(|_| {
                negotiated_version.is_some_and(|version| version >= PUBLISH_ACK_POSITION_VERSION)
            }),
        }),
        // Admitted, then refused by the service. This keeps the undifferentiated
        // code and today's exact bytes even when the cause WAS the roster — a
        // quiesce that landed in the window between the admission above and this
        // call surfaces here, not as `0x0102`. Degraded, never lying: a roster
        // code is a reliable positive signal and never a complete one, and
        // recovering the "real" cause by a second roster read is the exact lie
        // the probe-before form exists to make impossible. An unresolved publish
        // alone carries its own code, chosen by the error's TYPE, never its text.
        Err(error) => FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: if matches!(error, ServerError::PublishUnresolved { .. }) {
                PUBLISH_UNRESOLVED_CODE
            } else {
                SERVER_ERROR_CODE
            },
            message: Some(error.to_string()),
        }),
    }
}

fn subscribe_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    channel: &str,
    accepted_schemas: &[ProtocolSchemaId],
    max_in_flight: u32,
) -> FrameAction {
    let limits = runtime.limits();
    let max_subscriptions = limits.max_subscriptions_per_connection;
    // §5 `max_subscriptions_per_connection`: refuse a subscription past the cap
    // with the subscription error channel (the same vocabulary an unsupported or
    // failed subscribe uses), before touching the services adapter so an over-cap
    // subscribe never spawns a subscriber process.
    if state.subscriptions.len() >= max_subscriptions {
        return FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(
                ServerError::ConnectionCapReached {
                    operation: "subscribe".to_owned(),
                    cap: "max_subscriptions_per_connection",
                    limit: max_subscriptions,
                }
                .to_string(),
            ),
        });
    }
    // Admission BEFORE delegation, on the same grounds as `publish_response`.
    //
    // It sits AFTER the §5 per-connection subscription cap and BEFORE the inbox
    // budget is built. After the cap, because the cap is a statement about THIS
    // connection and the roster is a statement about the server: a connection
    // already at its cap is refused for that reason today, and moving the roster
    // ahead of it would change which of two true refusals an over-cap subscribe
    // to an absent channel receives. Before the budget, because
    // `get_or_insert_with` below MINTS the connection's shared inbox pool on
    // first use, and a refused frame must leave connection state exactly as it
    // found it.
    let services = runtime.services();
    if let Err(refusal) = services.admit_channel(ChannelOperation::Subscribe, channel) {
        return FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: refusal.reason_code(),
            message: Some(access_to_server_error(&refusal).to_string()),
        });
    }
    // R3 + §5: build the install spec CARRIED INTO subscribe, so the shared
    // connection byte budget, per-inbox fairness cap, and wake notifier are
    // installed on the inbox at construction — strictly BEFORE the registration
    // is published to the channel actor. There is no pre-install window: no
    // envelope can be admitted uncharged, past the depth cap, or without a wake.
    // The budget is created lazily on the first subscribe and shared across
    // every later subscription, so the 4 MiB is one connection-scoped pool. The
    // notifier captures the CONNECTION scheduler's `READY` enqueue handle here
    // (§1.2(2), Vesper advisory 3); under the busy loop the wake is redundant
    // (the every-slice pump still drains the inbox), so a missing waker —
    // scheduler-free unit tests — is harmless. PARK-FLIP: this is the wake that
    // replaces the delivery pump's every-slice assumption.
    let budget = state
        .inbox_budget
        .get_or_insert_with(|| {
            liminal::channel::ConnectionInboxBudget::new(limits.max_connection_inbox_bytes)
        })
        .clone();
    let notifier: Option<liminal::channel::InboxNotifier> = runtime.ready_waker(pid).map(|waker| {
        std::sync::Arc::new(move || {
            waker.fire();
        }) as liminal::channel::InboxNotifier
    });
    // A1 §2: the `Subscribe` frame's `max_in_flight` has been on the wire and
    // decoded (and non-zero-validated) all along, and was consumed nowhere.
    // THIS is where it becomes a bound. `max_buffer_depth` stays bus policy —
    // the client declares what it can process concurrently, not how much the
    // bus may hold for it — so it takes the library default rather than a
    // client-supplied number.
    //
    // A zero or unrepresentable value falls back to the library default rather
    // than refusing: the frame decoder already rejects zero, so this arm is
    // unreachable from a well-formed wire, and a defaulted window is a safer
    // answer than a window of zero (which would Reject every publish).
    //
    // AND IT IS CLAMPED. The declaration is a CLIENT's, and until it is clamped
    // nothing stops a client declaring a window so large that the A1 Reject
    // band sits ABOVE this connection's §5 depth cap — at which point the
    // terminal fairness shed fires first and a slow consumer loses its whole
    // subscription instead of pacing one message. `Frame::validate` refuses
    // only zero, so the wire cannot be relied on for this.
    //
    // `clamp_capacity_to_depth_cap` derives the bound from the two numbers it
    // is given, so the invariant follows the config: it is a control, not a
    // note. Bus policy (the buffer band) wins over the client's declaration,
    // which is §2's split.
    let capacity = usize::try_from(max_in_flight)
        .ok()
        .filter(|window| *window > 0)
        .map(|window| liminal::pressure::ConsumerCapacity {
            max_in_flight: window,
            max_buffer_depth: liminal::channel::DEFAULT_MAX_BUFFER_DEPTH,
        })
        .map(|declared| {
            liminal::channel::clamp_capacity_to_depth_cap(
                declared,
                limits.max_subscription_inbox_depth,
            )
        });
    let install = liminal::channel::InboxInstall {
        budget,
        depth_cap: limits.max_subscription_inbox_depth,
        notifier,
        capacity,
    };
    match services.subscribe(channel, accepted_schemas, Some(install)) {
        Ok(mut subscription) => {
            // Record the client-chosen delivery stream so the pump can address
            // every `Deliver` to the stream the client is reading this subscription
            // on.
            subscription.set_stream_id(stream_id);
            // ACCESS-006 R3: the delivery pump's release check reads the
            // channel the subscriber was admitted on, so the connection stamps
            // it from the frame rather than trusting a services impl to.
            subscription.set_channel(channel.to_owned());
            if let Some(subscriber) = subscription.subscriber_pid() {
                runtime.receiver(pid, subscriber);
            }
            subscription.set_pressure_ledger_depth(limits.max_subscription_inbox_depth);
            let subscription_id = subscription.id();
            let selected_schema = subscription.selected_schema();
            state.subscriptions.insert(subscription_id, subscription);
            FrameAction::Respond(Frame::SubscribeAck {
                flags: 0,
                stream_id,
                subscription_id,
                selected_schema,
            })
        }
        // Admitted, then refused by the service — the same degraded arm as
        // `publish_response`'s, on the same grounds.
        Err(error) => FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(error.to_string()),
        }),
    }
}

/// An `Unsubscribe` frame; a worker's last subscription to a line leaving
/// takes its subscribe stop off that line (SRV-012 R7).
fn unsubscribe_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    subscription_id: u64,
) -> FrameAction {
    let line = state
        .subscriptions
        .get(&subscription_id)
        .filter(|_| state.worker_flows.is_some())
        .map(|subscription| subscription.channel().to_owned());
    let action = unsubscribe_response(runtime.services(), state, stream_id, subscription_id);
    let bit = flow_bit(FlowRole::Subscribes);
    if let Some(line) = line
        && !state
            .subscriptions
            .values()
            .any(|subscription| subscription.channel() == line)
        && let Some(flows) = state.worker_flows.as_mut()
        && let Some(known) = flows.get_mut(&line)
        && *known & bit != 0
    {
        *known &= !bit;
        if *known == 0 {
            flows.remove(&line);
        }
        runtime.flow(pid, &line, FlowRole::Subscribes, false);
    }
    action
}

fn unsubscribe_response(
    services: &dyn ConnectionServices,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    subscription_id: u64,
) -> FrameAction {
    if let Some(subscription) = state.subscriptions.remove(&subscription_id) {
        // Drop the delivery-sequence counter and any held-back frame with the
        // subscription so a re-subscribe that reuses the id restarts clean at 1 and
        // never flushes a stale delivery.
        state.delivery_seqs.remove(&subscription_id);
        state.held_deliveries.remove(&subscription_id);
        // The wire stays silent about this failure on purpose. The server warns,
        // so the fault is not lost, but the client-visible contract for
        // `Unsubscribe` is removal from THIS connection's state — which the
        // `remove` calls above have already completed unconditionally. The
        // adapter-side teardown is the server's own bookkeeping; reporting its
        // failure would tell the client its subscription is still live when, as
        // far as this connection is concerned, it is not.
        if let Err(error) = services.unsubscribe(subscription) {
            tracing::warn!(subscription_id, %error, "liminal unsubscribe failed");
        }
        return FrameAction::NoResponse;
    }
    // No such subscription on this connection. Full mode keeps the idempotent no-op
    // (unsubscribing an unknown id is harmless). A capability-scoped adapter that
    // serves no channels never issued a subscription, so this frame targets an
    // operation the profile does not support: reject it explicitly rather than
    // silently accept it. There is no dedicated `UnsubscribeError` frame in the
    // vocabulary, so the closest honest fit — `SubscribeError`, the subscription
    // error channel — carries the rejection.
    if services.supports_channel_operations() {
        FrameAction::NoResponse
    } else {
        FrameAction::Respond(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(unsupported_by_front_door("unsubscribe")),
        })
    }
}

// Every `ConversationError` this function emits rides the REQUEST'S stream id
// (Sol round 2): a client that opened on stream 17 must see its typed refusal on
// stream 17 — a refusal on a hard-coded stream would leave the client's actual
// stream silent, turning the fail-closed refusals into a silent-timeout shape.
fn conversation_open(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    conversation_id: u64,
    subject: &str,
) -> FrameAction {
    // ACCESS-001 R2, `ConversationOpen` = Hot(Call, FrameSubject): decided
    // from the admitted pass before any duplicate, cap or actor question, so a
    // holder without `call` on the subject learns nothing else about it.
    if let Some(refused) = lys_decision(runtime, state, subject, HotAction::Call, |message| {
        Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(message),
        }
    }) {
        return refused;
    }
    // Duplicate open of a LIVE conversation id is refused fail-closed (review
    // ruling, round 1 item 3): an insert-replace would drop the old resource
    // without close/finalize (violating the D4 every-teardown-path-finalizes
    // discipline) and leave its pending/tombstone reply entries to FIFO-correlate
    // against the NEW instance's replies — cross-instance mis-correlation, the
    // corruption class the tombstone ruling exists to kill. Nothing sanctions
    // re-open-as-replace: the SDK explicitly guards against re-sending
    // `ConversationOpen` for an open id and calls a duplicate a defect
    // (`liminal-sdk/src/remote/tcp/connection.rs:48`). The remedy is named in the
    // error: close the conversation first, or use a fresh id.
    if state.conversations.contains_key(&conversation_id) {
        return conversation_error(
            stream_id,
            conversation_id,
            "conversation id is already open on this connection; close it before \
             reopening or use a fresh id",
        );
    }
    let max_conversations = runtime.limits().max_conversations_per_connection;
    // §5 `max_conversations_per_connection`: refuse opening past the cap before
    // constructing any supervised conversation actor. Duplicate ids were refused
    // above, so every admission here is a distinct live conversation.
    if state.conversations.len() >= max_conversations {
        return conversation_error(
            stream_id,
            conversation_id,
            &ServerError::ConnectionCapReached {
                operation: "conversation open".to_owned(),
                cap: "max_conversations_per_connection",
                limit: max_conversations,
            }
            .to_string(),
        );
    }
    match runtime
        .services()
        .open_conversation(conversation_id, subject)
    {
        Ok(conversation) => {
            // R1(vi)(a): install the reply-availability notifier PERMANENTLY at
            // conversation open (normative — not per-message). It fires the
            // connection's `READY` marker on the reply queue's empty→non-empty edge
            // and on terminal actor error, capturing the CONNECTION scheduler's
            // enqueue handle here at install. Under the busy loop the wake is
            // redundant (the slice polls pending conversations every slice); it is
            // the park-flip wake source. The notifier is cleared by the
            // conversation core at close/finalize, so a marker never fires after
            // teardown.
            let reply_notifier: std::sync::Arc<dyn Fn() + Send + Sync> =
                match runtime.ready_waker(pid) {
                    Some(waker) => std::sync::Arc::new(move || {
                        waker.fire();
                    }),
                    None => std::sync::Arc::new(|| {}),
                };
            // Ownership registration is mandatory even without a wake handle: the
            // busy-loop connection still polls the tagged reply queue every slice.
            conversation.register_reply_notifier(reply_notifier);
            state.conversations.insert(conversation_id, conversation);
            FrameAction::NoResponse
        }
        Err(error) => FrameAction::Respond(Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code: SERVER_ERROR_CODE,
            message: Some(error.to_string()),
        }),
    }
}

fn route_conversation_message(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    request: Request,
    subject: &str,
    envelope: &MessageEnvelope,
) -> Option<FrameAction> {
    let Request {
        flags,
        stream_id,
        conversation_id,
    } = request;
    // ACCESS-003 R1: a conversation on a service subject carries requests to
    // that service, decided and served on the admitted session; the message
    // never reaches the channel.
    // SRV-028 R3: a lifecycle verb's subject carries requests to the
    // services registry, decided live as deploy_service.
    if super::lifecycle_route::is_lifecycle(subject) {
        return Some(super::lifecycle_route::lifecycle_frame_response(
            runtime,
            state,
            flags,
            stream_id,
            conversation_id,
            subject,
            envelope,
        ));
    }
    if crate::services::broker::serves_subject(subject) {
        return Some(super::broker_route::broker_frame_response(
            runtime, state, request, subject, envelope,
        ));
    }
    if crate::services::service_of(subject).is_some() {
        return Some(super::service_route::service_frame_response(
            runtime,
            state,
            flags,
            stream_id,
            conversation_id,
            subject,
            envelope,
        ));
    }
    None
}

#[path = "apply/service_message.rs"]
mod service_message;
pub(super) use service_message::conversation_message;

fn conversation_error(stream_id: u32, conversation_id: u64, message: &str) -> FrameAction {
    FrameAction::Respond(Frame::ConversationError {
        flags: 0,
        stream_id,
        conversation_id,
        reason_code: SERVER_ERROR_CODE,
        message: Some(message.to_owned()),
    })
}

fn conversation_close(
    services: &dyn ConnectionServices,
    state: &mut ConnectionProcessState,
    stream_id: u32,
    conversation_id: u64,
) -> FrameAction {
    if let Some(conversation) = state.conversations.remove(&conversation_id) {
        // §1.2(3b) close sweep — one of only TWO sanctioned tombstone-reclamation
        // triggers (review round 1 item 2): synchronously remove EVERY pending and
        // tombstone entry for this conversation, IN the close transaction, before
        // the actor close/finalize below and before any same-id reopen can be
        // applied from the same buffer. Without this, a tombstone-only
        // conversation's entries linger to connection teardown (the lazy
        // service_pending_replies sweep only visits conversations with PENDING
        // entries) and could FIFO-correlate against a reopened id's replies.
        state.pending_replies.remove_conversation(conversation_id);
        if let Err(error) = services.close_conversation(conversation) {
            tracing::warn!(conversation_id, %error, "liminal conversation close failed");
        }
        return FrameAction::NoResponse;
    }
    // No such conversation on this connection. Full mode keeps the idempotent
    // no-op; the worker front door serves no conversations, so a close targets an
    // unsupported operation — reject it explicitly rather than swallow it.
    if services.supports_channel_operations() {
        FrameAction::NoResponse
    } else {
        conversation_error(
            stream_id,
            conversation_id,
            &unsupported_by_front_door("conversation close"),
        )
    }
}

#[cfg(test)]
#[path = "process_tests.rs"]
mod tests;

/// ACCESS-004: a worker's registration bound to the holder its `Connect`
/// verified.
#[cfg(test)]
#[path = "apply_worker_holder_tests.rs"]
mod worker_holder_tests;

/// ACCESS-001 R2: every hot wire route decided by its class from the admitted
/// Lys pass before its effect.
#[cfg(test)]
#[path = "apply_lys_routes_tests.rs"]
mod lys_routes_tests;

/// The wire leg's own instruments. A separate module from `tests` above because
/// those exercise the frame surface against a recording stand-in, and these have
/// to run against a REAL roster: the claim is that the typed code survives the
/// trip from the roster to the response frame, and a stand-in that answers the
/// admission question itself would be asserting on its own fixture.
#[cfg(test)]
#[path = "apply_admission_tests.rs"]
mod admission_tests;

/// A1 §2's capacity declaration, observed where the wire hands it over. Kept
/// out of `admission_tests` because it needs a spy over a real adapter rather
/// than the real adapter alone: the claim is about the `InboxInstall` the
/// server BUILDS, which nothing else on the path reports.
#[cfg(test)]
#[path = "apply_capacity_tests.rs"]
mod capacity_tests;

/// The publish reply's two additions: the unresolved publish's own reason code
/// and the durable position on the ack of a connection that negotiated it.
#[cfg(test)]
#[path = "apply_publish_outcome_tests.rs"]
mod publish_outcome_tests;

/// ACCESS-005: the protocol versions a server under a revocation policy
/// negotiates — the two it always does, plus the reason-bearing close's.
const REVOCATION_PROTOCOLS: [ProtocolVersion; 3] = [
    ProtocolVersion::new(1, 0),
    PUBLISH_ACK_POSITION_VERSION,
    liminal::protocol::REVOCATION_CLOSE_VERSION,
];

/// The versions this server negotiates: unchanged without a revocation
/// policy, so every legacy and private session keeps its bytes.
fn supported_protocols(runtime: &ConnectionRuntime) -> &'static [ProtocolVersion] {
    if runtime.revocation().is_some() {
        &REVOCATION_PROTOCOLS
    } else {
        &SUPPORTED_PROTOCOLS
    }
}

/// `ConnectError` for a revocation refusal (ACCESS-005 R5): reason `0x0006`
/// (`AuthenticationFailure`) and the refusal's stable name as the whole
/// message, never any pass, holder, live-prefix or grant-chain bytes.
fn revocation_connect_refused(refusal: crate::auth_revocation::RevocationRefusal) -> FrameAction {
    let error = ProtocolError::AuthenticationFailure {
        message: Some(refusal.name().to_owned()),
    };
    FrameAction::RespondThenClose(Frame::ConnectError {
        flags: 0,
        reason_code: error.reason_code(),
        message: error.message().map(str::to_owned),
    })
}

/// The revocation fence at `Connect` (ACCESS-005 R3, R5): the peer must be
/// able to negotiate the reason-bearing close, then the authority verifies
/// the binding for `credential` held by `holder` and commits it in one
/// ordered transition. Nothing is attached or announced before this passes,
/// and the committed binding is set once on the connection. `Ok` without a
/// revocation policy, untouched.
fn revocation_connect(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    (min_version, max_version): (ProtocolVersion, ProtocolVersion),
    credential: &[u8],
    holder: &[u8],
) -> Result<(), Box<FrameAction>> {
    let Some(authority) = runtime.revocation() else {
        return Ok(());
    };
    if state.revocation.is_some() {
        // A second Connect never replaces the verified binding.
        return Err(Box::new(FrameAction::Close));
    }
    let closes_with_reason = negotiate_version(min_version, max_version, &REVOCATION_PROTOCOLS)
        .is_ok_and(|selected| selected >= liminal::protocol::REVOCATION_CLOSE_VERSION);
    if !closes_with_reason {
        return Err(Box::new(revocation_connect_refused(
            RevocationRefusal::ProtocolUnsupported,
        )));
    }
    let grants = authority
        .admit(pid, credential, holder)
        .map_err(|refusal| Box::new(revocation_connect_refused(refusal)))?;
    state.revocation_swept = grants.changes();
    state.revocation = Some(grants);
    Ok(())
}

/// The connection's own revocation sweep (ACCESS-005 R4): every subscription
/// and conversation whose right was withdrawn is retired through its
/// existing cleanup path and answered with a named terminal outcome. Runs
/// only after the rights changed; returns the frames to enqueue.
pub(super) fn retire_withdrawn_rights(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
) -> Vec<Frame> {
    let Some(changes) = state
        .revocation
        .as_ref()
        .map(crate::auth_revocation::ConnectionGrants::changes)
    else {
        return Vec::new();
    };
    if changes == state.revocation_swept {
        return Vec::new();
    }
    let Some(grants) = state.revocation.clone() else {
        return Vec::new();
    };
    state.revocation_swept = changes;
    let services = runtime.services();
    let withdrawn: Vec<(u64, u32)> = state
        .subscriptions
        .iter()
        .filter(|(_, subscription)| !grants.permits(HotAction::Subscribe, subscription.channel()))
        .map(|(id, subscription)| (*id, subscription.stream_id()))
        .collect();
    let mut frames = Vec::with_capacity(withdrawn.len());
    for (subscription_id, stream_id) in withdrawn {
        unsubscribe_response(services, state, stream_id, subscription_id);
        frames.push(Frame::SubscribeError {
            flags: 0,
            stream_id,
            reason_code: 0x0201,
            message: Some(RevocationRefusal::GrantRevoked.name().to_owned()),
        });
    }
    let calls: Vec<u64> = state
        .conversations
        .iter()
        .filter(|(_, conversation)| !grants.permits(HotAction::Call, conversation.subject()))
        .map(|(id, _)| *id)
        .collect();
    for conversation_id in calls {
        // The conversation's stream is not retained per conversation, so its
        // terminal outcome is this named record and the existing close path.
        tracing::warn!(
            connection_pid = grants.pid(),
            conversation_id,
            reason = RevocationRefusal::GrantRevoked.name(),
            "conversation retired: its call right was revoked"
        );
        conversation_close(services, state, 0, conversation_id);
    }
    frames
}

/// ACCESS-005: the revocation fence at `Connect`, the hot refusals and the
/// withdrawn-rights sweep, against a real authority.
#[cfg(test)]
#[path = "apply_revocation_tests.rs"]
mod revocation_tests;

/// ACCESS-006 R1, R3: every protected admission and final data release
/// decided on the connection's verified subject.
#[cfg(test)]
#[path = "apply_membership_tests.rs"]
mod membership_tests;