liminal-server 0.21.1

//! ACCESS-001 R2 at every hot wire route: a connection admitted on a Lys pass
//! record has each `Call` and `Serve` route decided from that pass, offline,
//! before the route's effect, and refused by name on the route's own error
//! frame; the routes without a refusal frame (`PushReply`, the backpressure
//! signals) refuse by having no effect, the pusher reading the refusal by
//! type. The routes' classes are the ones `classify_wire_route` pins.

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use ed25519_dalek::SigningKey;
use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, MessageId, ProtocolVersion, SchemaId,
    WorkerRegisterOutcome, WorkerRegistration,
};
use liminal_protocol::credential::WireCredential;

use liminal_protocol::wire::{
    ClientRequest, Generation, ObserverRecoveryHandshake, ObserverRefusal, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue,
};

use super::{apply_frame, participant_access_refusal};
use crate::ServerError;
use crate::auth_lys::session::ConsumerAdmitter;
use crate::auth_lys::{
    AccessAction, AccessDecisionError, AccessRefusal, AccessResource, DeliberateDecision,
    GrantReceipt, HotAction, LysDeliberateConsumer, LysPassConsumer, RefusalCause, ResourceKind,
    SubjectBinding,
};
use crate::auth_pass::{PassVerifier, VerifiedHolder};
use crate::link::{LinkAction, LinkHost, LinkRights};
use crate::link_handshake::{LinkHandshake, LocalMachine, NOISE_OVERHEAD};
use crate::server::connection::conversation::{ConnectionConversation, ConversationResource};
use crate::server::connection::notifier::ConnectionNotifier;
use crate::server::connection::services::{
    ConnectionServices, ConnectionSubscription, PublishOutcome, SubscriptionResource,
};
use crate::server::connection::state::{ConnectionProcessState, FrameAction};
use crate::server::connection::supervisor::ConnectionRuntime;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
    ParticipantSemanticHandler,
};
use liminal::protocol::{LinkEnd, LinkOffer, LinkOpenOutcome};
use lys_core::keys::Ed25519Identity;
use zeroize::Zeroizing;

const PID: u64 = 11;
const STREAM: u32 = 3;
const CONVERSATION: u64 = 9;
const SUBSCRIPTION: u64 = 5;
const ISSUED: u64 = 7;
const REGISTRY_SEED: [u8; 32] = [42; 32];

type Failure = Box<dyn std::error::Error>;

/// A consumer standing where `lys-pass` will: one fixture pass, and a grant
/// table the test moves between routes so each decision is shown to be its own.
#[derive(Debug, Default)]
struct Grants {
    held: Mutex<Vec<(String, HotAction)>>,
    /// Link grants by subject (ACCESS-002 R1), decided offline like the hot
    /// table.
    links: Mutex<Vec<(String, LinkAction)>>,
    /// The link ends Lys answers outright, live; every other open is refused.
    opens: Mutex<Vec<(String, AccessAction)>>,
    /// When the fixture pass was issued, in Unix seconds, when it says.
    issued_at: Option<u64>,
}

impl Grants {
    fn grant_link(&self, subject: &str, action: LinkAction) -> Result<(), Failure> {
        self.links
            .lock()
            .map_err(|_| "the link table was poisoned")?
            .push((subject.to_owned(), action));
        Ok(())
    }

    fn allow_open(&self, link: &str, end: AccessAction) -> Result<(), Failure> {
        self.opens
            .lock()
            .map_err(|_| "the open table was poisoned")?
            .push((link.to_owned(), end));
        Ok(())
    }

    fn grant(&self, subject: &str, action: HotAction) -> Result<(), Failure> {
        self.held
            .lock()
            .map_err(|_| "the grant table was poisoned")?
            .push((subject.to_owned(), action));
        Ok(())
    }

    fn revoke(&self, subject: &str, action: HotAction) -> Result<(), Failure> {
        self.held
            .lock()
            .map_err(|_| "the grant table was poisoned")?
            .retain(|(held, held_action)| !(held == subject && *held_action == action));
        Ok(())
    }
}

impl LysPassConsumer for Grants {
    type Pass = ();
    type Holder = str;
    type AdmissionError = std::io::Error;

    fn verify_offline(
        &self,
        bytes: &[u8],
        audience: &str,
        _now: u64,
    ) -> Result<Self::Pass, Self::AdmissionError> {
        if bytes != b"fixture-pass" || audience != "liminal" {
            return Err(std::io::Error::other("signature"));
        }
        Ok(())
    }

    fn verified_holder<'p>(&self, pass: &'p Self::Pass) -> &'p Self::Holder {
        let () = pass;
        "fixture-holder"
    }

    fn verified_issued_at(&self, pass: &Self::Pass) -> Option<u64> {
        let () = pass;
        self.issued_at
    }

    fn decide_hot(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: HotAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        let held = self.held.lock().is_ok_and(|held| {
            held.iter()
                .any(|(id, held_action)| id == &subject.resource().id && *held_action == action)
        });
        if held {
            return Ok(GrantReceipt {
                grant: format!("grant-{}-{action:?}", subject.resource().id),
            });
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                AccessResource {
                    kind: ResourceKind::Subject,
                    id: subject.resource().id.clone(),
                },
                action.into(),
                format!("https://lys.example/ask/{}", subject.subject()),
            ),
        })
    }

    fn decide_link(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: LinkAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        let held = self.links.lock().is_ok_and(|held| {
            held.iter()
                .any(|(id, held_action)| id == &subject.resource().id && *held_action == action)
        });
        if held {
            return Ok(GrantReceipt {
                grant: format!("link-{}-{action:?}", subject.resource().id),
            });
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                AccessResource {
                    kind: ResourceKind::Subject,
                    id: subject.resource().id.clone(),
                },
                action.access_action(),
                format!("https://lys.example/ask/{}", subject.subject()),
            ),
        })
    }
}

impl LysDeliberateConsumer for Grants {
    fn decide_live(
        &self,
        pass: &Self::Pass,
        resource: &AccessResource,
        action: AccessAction,
        _now: u64,
    ) -> Result<DeliberateDecision, AccessDecisionError> {
        let () = pass;
        let allowed = self.opens.lock().is_ok_and(|opens| {
            opens
                .iter()
                .any(|(id, end)| id == &resource.id && *end == action)
        });
        if allowed {
            return Ok(DeliberateDecision::Outright(GrantReceipt {
                grant: format!("open-{}-{action:?}", resource.id),
            }));
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                resource.clone(),
                action,
                format!("https://lys.example/ask/{}", resource.id),
            ),
        })
    }
}

/// A conversation that answers nothing: the routes under test stop before it.
#[derive(Debug)]
struct Quiet;

impl ConversationResource for Quiet {
    fn message(&self, _envelope: &MessageEnvelope, _op_id: Option<u64>) -> Result<(), ServerError> {
        Ok(())
    }

    fn participant_pids(&self) -> Vec<u64> {
        Vec::new()
    }

    fn has_detected_crash(&self) -> bool {
        false
    }

    fn await_crash(&self, _timeout: Duration) -> Option<Instant> {
        None
    }

    fn receive_reply(&self, _timeout: Duration) -> Result<MessageEnvelope, ServerError> {
        Err(ServerError::ConfigValidation {
            message: "the quiet conversation never replies".to_owned(),
        })
    }

    fn close(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn finalize(self: Box<Self>) {}
}

/// Services recording every effect that reached them: a refused route leaves
/// no trace here.
#[derive(Debug, Default)]
struct Door {
    opens: Mutex<Vec<(u64, String)>>,
    messages: Mutex<usize>,
}

impl Door {
    fn opens(&self) -> Result<Vec<(u64, String)>, Failure> {
        Ok(self
            .opens
            .lock()
            .map_err(|_| "the open recorder was poisoned")?
            .clone())
    }

    fn messages(&self) -> Result<usize, Failure> {
        Ok(*self
            .messages
            .lock()
            .map_err(|_| "the message recorder was poisoned")?)
    }
}

fn unreached(route: &str) -> ServerError {
    ServerError::ConfigValidation {
        message: format!("{route} is not a route under test"),
    }
}

impl ConnectionServices for Door {
    fn publish(
        &self,
        _connection: u64,
        _channel: &str,
        _envelope: &MessageEnvelope,
        _idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        Err(unreached("publish"))
    }

    fn subscribe(
        &self,
        _channel: &str,
        _accepted_schemas: &[SchemaId],
        _install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        Err(unreached("subscribe"))
    }

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

    fn open_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        self.opens
            .lock()
            .map_err(|_| unreached("a poisoned open recorder"))?
            .push((conversation_id, subject.to_owned()));
        Ok(ConnectionConversation::new(
            Box::new(Quiet),
            subject.to_owned(),
        ))
    }

    fn conversation_message(
        &self,
        _conversation: &ConnectionConversation,
        _envelope: &MessageEnvelope,
        _op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        *self
            .messages
            .lock()
            .map_err(|_| unreached("a poisoned message recorder"))? += 1;
        Ok(())
    }

    fn close_conversation(&self, _conversation: ConnectionConversation) -> Result<(), ServerError> {
        Ok(())
    }

    fn flush_durable_state(&self) -> Result<(), ServerError> {
        Ok(())
    }
}

#[derive(Debug, Default)]
struct Silent;

impl ConnectionNotifier for Silent {
    fn on_worker_registered(
        &self,
        _pid: u64,
        _registration: &WorkerRegistration,
    ) -> Result<(), ServerError> {
        Ok(())
    }

    fn on_worker_registered_by(
        &self,
        _pid: u64,
        _registration: &WorkerRegistration,
        _holder: Option<&VerifiedHolder>,
    ) -> Result<(), ServerError> {
        Ok(())
    }

    fn on_worker_unregistered(&self, _pid: u64) {}
}

/// A connected process whose pass is `grants`, over `door`.
fn admitted(
    grants: &Arc<Grants>,
    door: &Arc<Door>,
) -> Result<(ConnectionRuntime, ConnectionProcessState), Failure> {
    let registry = SigningKey::from_bytes(&REGISTRY_SEED).verifying_key();
    let verifier = PassVerifier::new(registry.as_bytes(), 30)
        .map_err(|error| format!("verifier: {error:?}"))?;
    let runtime = ConnectionRuntime::for_tests_with_pass(
        Arc::clone(door) as Arc<dyn ConnectionServices>,
        Arc::new(Silent) as Arc<dyn ConnectionNotifier>,
        None,
        verifier,
    );
    runtime
        .install_lys_admitter(
            Arc::new(ConsumerAdmitter::new(Arc::clone(grants))),
            "network-a".to_owned(),
        )
        .map_err(|_| "a lys admitter was already installed")?;
    runtime.register_for_tests(PID)?;
    let mut state = ConnectionProcessState::default();
    let version = ProtocolVersion::new(1, 0);
    let action = apply_frame(
        PID,
        &runtime,
        &mut state,
        Frame::Connect {
            flags: 0,
            min_version: version,
            max_version: version,
            auth_token: WireCredential::lys_record(b"fixture-pass"),
        },
    );
    match action {
        FrameAction::Respond(Frame::ConnectAck { .. }) => Ok((runtime, state)),
        other => Err(format!("the Lys record was not admitted: {other:?}").into()),
    }
}

fn open(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    subject: &str,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationOpen {
            flags: 0,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            subject: subject.to_owned(),
        },
    )
}

fn message(runtime: &ConnectionRuntime, state: &mut ConnectionProcessState) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationMessage {
            flags: 0,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope: MessageEnvelope::new(
                SchemaId::new([1; SchemaId::WIRE_LEN]),
                CausalContext::independent(),
                b"hello".to_vec(),
            ),
        },
    )
}

/// The refusal a conversation route answered on its own error frame, naming
/// the subject and the action it lacked.
fn conversation_refusal(action: FrameAction) -> Result<String, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationError {
            conversation_id: CONVERSATION,
            stream_id: STREAM,
            message: Some(message),
            ..
        }) => Ok(message),
        other => Err(format!("the route was not refused on its conversation: {other:?}").into()),
    }
}

#[test]
fn a_conversation_opens_only_with_call_on_its_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;

    let refusal = conversation_refusal(open(&runtime, &mut state, "orders"))?;
    assert!(
        refusal.contains(r#""action":"call""#) && refusal.contains(r#""id":"orders""#),
        "the refusal names the subject and the action by name: {refusal}"
    );
    assert_eq!(door.opens()?, Vec::new(), "a refused open spawns nothing");
    assert!(!state.conversations.contains_key(&CONVERSATION));

    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        open(&runtime, &mut state, "orders"),
        FrameAction::NoResponse
    ));
    assert_eq!(door.opens()?, vec![(CONVERSATION, "orders".to_owned())]);
    assert!(state.conversations.contains_key(&CONVERSATION));
    Ok(())
}

#[test]
fn a_conversation_message_is_decided_on_the_open_subject_each_time() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        open(&runtime, &mut state, "orders"),
        FrameAction::NoResponse
    ));

    // The grant goes between open and message: the message is its own
    // decision on the conversation's retained subject, never the open's.
    grants.revoke("orders", HotAction::Call)?;
    let refusal = conversation_refusal(message(&runtime, &mut state))?;
    assert!(
        refusal.contains(r#""action":"call""#) && refusal.contains(r#""id":"orders""#),
        "{refusal}"
    );
    assert_eq!(door.messages()?, 0, "a refused message is never forwarded");

    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        message(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert_eq!(door.messages()?, 1);
    Ok(())
}

#[test]
fn a_worker_registration_is_decided_on_every_namespace_it_serves() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    grants.grant("orders", HotAction::Serve)?;
    let register = |state: &mut ConnectionProcessState| {
        apply_frame(
            PID,
            &runtime,
            state,
            Frame::WorkerRegister {
                flags: 0,
                registration: WorkerRegistration {
                    namespaces: vec!["orders".to_owned(), "invoices".to_owned()],
                    task_queue: "findings".to_owned(),
                    node: None,
                    activity_types: Vec::new(),
                    identity: "fixture-holder".to_owned(),
                    activities: Vec::new(),
                },
            },
        )
    };

    let FrameAction::Respond(Frame::WorkerRegisterAck {
        outcome: WorkerRegisterOutcome::Rejected { reason },
        ..
    }) = register(&mut state)
    else {
        return Err("a worker serving a namespace its pass lacks was not refused".into());
    };
    assert!(
        reason.contains(r#""action":"serve""#) && reason.contains(r#""id":"invoices""#),
        "the refusal names the namespace the pass lacks: {reason}"
    );
    assert_eq!(
        runtime.stored_registration_for_tests(PID)?,
        None,
        "a refused registration is never stored"
    );

    grants.grant("invoices", HotAction::Serve)?;
    assert!(matches!(
        register(&mut state),
        FrameAction::Respond(Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        })
    ));
    Ok(())
}

/// A subscription inbox that never delivers: the signal under test is decided
/// before anything is pulled from it.
#[derive(Debug)]
struct Empty;

impl SubscriptionResource for Empty {
    fn unsubscribe(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn try_next(&mut self) -> Option<liminal::envelope::Envelope> {
        None
    }

    fn has_pending(&self) -> bool {
        false
    }
}

/// Installs one subscription on `subject`, delivering on `STREAM`, with one
/// issued delivery `ISSUED` in its pressure ledger for a signal to reference.
fn subscribed(state: &mut ConnectionProcessState, subject: &str) {
    let mut subscription = ConnectionSubscription::new(
        SUBSCRIPTION,
        SchemaId::new([1; SchemaId::WIRE_LEN]),
        Box::new(Empty),
    );
    subscription.set_channel(subject.to_owned());
    subscription.set_stream_id(STREAM);
    subscription.set_pressure_ledger_depth(4);
    subscription.issued_delivery(ISSUED);
    state.subscriptions.insert(SUBSCRIPTION, subscription);
}

/// Whether the subscription's ledger still holds `ISSUED`: a granted signal
/// consumes it, a refused one leaves it.
fn ledger_holds_issued(state: &mut ConnectionProcessState) -> Result<bool, Failure> {
    let subscription = state
        .subscriptions
        .get_mut(&SUBSCRIPTION)
        .ok_or("the subscription under test is gone")?;
    Ok(subscription.match_pressure(&ISSUED.to_string(), true))
}

fn accept(runtime: &ConnectionRuntime, state: &mut ConnectionProcessState) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::Accept {
            flags: 0,
            stream_id: STREAM,
            referenced_message_id: MessageId::new(ISSUED.to_string()),
        },
    )
}

#[test]
fn a_backpressure_signal_is_decided_on_its_subscriptions_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    subscribed(&mut state, "orders");

    // No refusal frame exists for a signal: the refusal is that nothing
    // happens, and the ledger entry the signal referenced is still there.
    assert!(matches!(
        accept(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert!(
        ledger_holds_issued(&mut state)?,
        "a refused signal consumes nothing from the ledger"
    );

    grants.grant("orders", HotAction::Subscribe)?;
    assert!(matches!(
        accept(&runtime, &mut state),
        FrameAction::NoResponse
    ));
    assert!(
        !ledger_holds_issued(&mut state)?,
        "a granted signal is matched against the ledger and consumes its entry"
    );
    Ok(())
}

fn reply(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    correlation_id: u64,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::PushReply {
            flags: 0,
            stream_id: STREAM,
            correlation_id,
            payload: b"answer".to_vec(),
        },
    )
}

#[test]
fn a_push_reply_is_decided_on_the_subject_its_push_was_made_on() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;

    // The subject is recorded at push time; the reply is refused on it and the
    // pusher reads the refusal by type, with the subject and action named.
    let awaiter = runtime.push_for_tests(PID, Some("orders"))?;
    assert!(matches!(
        reply(&runtime, &mut state, awaiter.correlation_id()),
        FrameAction::NoResponse
    ));
    match awaiter.receive(Duration::from_millis(50)) {
        Err(ServerError::PushReplyRefused {
            correlation_id,
            reason,
        }) => {
            assert_eq!(correlation_id, awaiter.correlation_id());
            assert!(
                reason.contains(r#""action":"serve""#) && reason.contains(r#""id":"orders""#),
                "the refusal names the pushed subject and the action it lacked: {reason}"
            );
        }
        other => return Err(format!("the refused reply was not read by type: {other:?}").into()),
    }

    // A push made with no subject cannot be decided: refused by name, so the
    // pusher learns the binding was never recorded rather than waiting.
    let unbound = runtime.push_for_tests(PID, None)?;
    assert!(matches!(
        reply(&runtime, &mut state, unbound.correlation_id()),
        FrameAction::NoResponse
    ));
    match unbound.receive(Duration::from_millis(50)) {
        Err(ServerError::PushReplyRefused { reason, .. }) => {
            assert!(reason.contains("no subject binding"), "{reason}");
        }
        other => return Err(format!("the unbound reply was not refused: {other:?}").into()),
    }

    grants.grant("orders", HotAction::Serve)?;
    let granted = runtime.push_for_tests(PID, Some("orders"))?;
    assert!(matches!(
        reply(&runtime, &mut state, granted.correlation_id()),
        FrameAction::NoResponse
    ));
    assert_eq!(
        granted.receive(Duration::from_millis(50))?,
        b"answer".to_vec(),
        "a granted reply is delivered byte-exact"
    );
    Ok(())
}

const MAPPED: u64 = 70;
const UNMAPPED: u64 = 71;

/// A participant service that places conversation `MAPPED` under `orders` and
/// nothing else; its dispatch is never reached by a decided-and-refused
/// request, so `handle` is an error.
#[derive(Debug)]
struct Mapped;

impl ParticipantSemanticHandler for Mapped {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Internal {
            message: "the participant handler is not a route under test".to_owned(),
        })
    }

    fn conversation_subject(&self, conversation_id: u64) -> Option<String> {
        (conversation_id == MAPPED).then(|| "orders".to_owned())
    }
}

fn admission(conversation_id: u64) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id,
        participant_id: 2,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([0xA7; 16]),
        payload: vec![1, 2, 3],
    })
}

#[test]
fn a_participant_request_is_decided_on_its_conversations_subject() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, state) = admitted(&grants, &door)?;

    // A conversation no authority places under a subject is refused by name,
    // never decided on a subject guessed from its number.
    let unmapped = participant_access_refusal(&runtime, &state, &Mapped, &admission(UNMAPPED))
        .ok_or("an unmapped conversation was not refused")?;
    assert!(
        unmapped.contains("no subject mapping for conversation 71"),
        "{unmapped}"
    );

    // A mapped conversation is decided with the route's action on its subject.
    let refusal = participant_access_refusal(&runtime, &state, &Mapped, &admission(MAPPED))
        .ok_or("a record admission without publish on orders was not refused")?;
    assert!(
        refusal.contains(r#""action":"publish""#) && refusal.contains(r#""id":"orders""#),
        "the refusal names the conversation's subject and the route's action: {refusal}"
    );
    grants.grant("orders", HotAction::Publish)?;
    assert_eq!(
        participant_access_refusal(&runtime, &state, &Mapped, &admission(MAPPED)),
        None,
        "a granted admission proceeds to dispatch"
    );

    // A recovery batch is decided on every conversation it names before any
    // effect: one unmapped conversation refuses the whole batch.
    let batch = ClientRequest::ObserverRecovery(ObserverRecoveryHandshake {
        observer_refusals: vec![
            ObserverRefusal {
                conversation_id: MAPPED,
                refused_epoch: 1,
            },
            ObserverRefusal {
                conversation_id: UNMAPPED,
                refused_epoch: 1,
            },
        ],
    });
    grants.grant("orders", HotAction::Subscribe)?;
    let batch_refusal = participant_access_refusal(&runtime, &state, &Mapped, &batch)
        .ok_or("a batch naming an unmapped conversation was not refused")?;
    assert!(batch_refusal.contains("conversation 71"), "{batch_refusal}");
    Ok(())
}

/// This liminal's own end of a crossing, as a table (ACCESS-002 R1).
#[derive(Debug)]
struct HostRights {
    import: &'static [&'static str],
}

impl LinkRights for HostRights {
    fn holds(
        &self,
        action: LinkAction,
        subject: &SubjectBinding,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let reached =
            action == LinkAction::Import && self.import.contains(&subject.resource().id.as_str());
        if reached {
            return Ok(GrantReceipt {
                grant: format!("host-import-{}", subject.resource().id),
            });
        }
        Err(AccessDecisionError {
            cause: RefusalCause::MissingGrant,
            refusal: AccessRefusal::new(
                subject.resource().clone(),
                action.access_action(),
                "https://lys.example/ask/host".to_owned(),
            ),
        })
    }
}

const LINK_STREAM: u32 = 4;

fn machine(seed: u8) -> LocalMachine {
    LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([seed; 32])))
}

/// The remote machine's offer to export over `link_id`, with its first
/// handshake message; the initiator is returned to finish the handshake.
fn export_offer(
    remote: &LocalMachine,
    host: &LocalMachine,
    link_id: &str,
) -> Result<(LinkHandshake, LinkOffer), Failure> {
    let mut initiator = LinkHandshake::initiator(remote, host.machine())?;
    let mut first = vec![0_u8; NOISE_OVERHEAD];
    let written = initiator.write(&mut first)?;
    first.truncate(written);
    Ok((
        initiator,
        LinkOffer {
            link_id: link_id.to_owned(),
            end: LinkEnd::Export,
            machine: *remote.machine().ed25519_public(),
            handshake: first,
        },
    ))
}

fn link_open(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    offer: LinkOffer,
) -> Result<LinkOpenOutcome, Failure> {
    match apply_frame(
        PID,
        runtime,
        state,
        Frame::LinkOpen {
            flags: 0,
            stream_id: LINK_STREAM,
            offer,
        },
    ) {
        FrameAction::Respond(Frame::LinkOpenAck {
            stream_id: LINK_STREAM,
            outcome,
            ..
        }) => Ok(outcome),
        other => Err(format!("the open was not acknowledged on its stream: {other:?}").into()),
    }
}

fn link_message(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    subject: &str,
    sealed: Vec<u8>,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::LinkMessage {
            flags: 0,
            stream_id: LINK_STREAM,
            subject: subject.to_owned(),
            sealed,
        },
    )
}

/// The refusal a crossing answered under the link's own reason code.
fn crossing_refusal(action: FrameAction) -> Result<String, Failure> {
    match action {
        FrameAction::Respond(Frame::PublishError {
            stream_id: LINK_STREAM,
            reason_code: crate::server::connection::link_route::LINK_CROSSING_REFUSED_CODE,
            message: Some(message),
            ..
        }) => Ok(message),
        other => Err(format!("the crossing was not refused by name: {other:?}").into()),
    }
}

/// ACCESS-002 R1 on the wire: without `[auth.lys.machine]` no link opens
/// here, by name; with it, an offer is refused before Lys is asked when the
/// offered key is not the holder's recorded key, refused by Lys's own answer
/// when the end's grant is missing, and keyed when Lys allows it outright.
#[test]
fn a_link_opens_on_the_holders_recorded_key_when_lys_allows_the_end() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let remote = machine(1);
    let host = machine(2);
    let (_, offer) = export_offer(&remote, &host, "link-7")?;
    let LinkOpenOutcome::Refused { reason } = link_open(&runtime, &mut state, offer.clone())?
    else {
        return Err("a liminal without a machine identity opened a link".into());
    };
    assert!(reason.contains("[auth.lys.machine]"), "{reason}");

    runtime
        .install_link_host(Arc::new(LinkHost::new(
            machine(2),
            Box::new(HostRights {
                import: &["orders"],
            }),
            [("fixture-holder".to_owned(), remote.machine())].into(),
        )))
        .map_err(|_| "a link host was already installed")?;
    // A key that is not the recorded one is refused before Lys is asked.
    let (_, stranger) = export_offer(&machine(3), &host, "link-7")?;
    let LinkOpenOutcome::Refused { reason } = link_open(&runtime, &mut state, stranger)? else {
        return Err("a stranger's key opened a link".into());
    };
    assert!(reason.contains("recorded key"), "{reason}");
    // The recorded key, but Lys refuses the end: the four-field refusal.
    let LinkOpenOutcome::Refused { reason } = link_open(&runtime, &mut state, offer)? else {
        return Err("an end Lys refused opened a link".into());
    };
    assert!(reason.contains("\"link_export\""), "{reason}");
    assert!(reason.contains("liminal.link:link-7"), "{reason}");
    // Lys allows the end outright: the handshake completes on both sides.
    grants.allow_open("link-7", AccessAction::LinkExport)?;
    let (mut initiator, offer) = export_offer(&remote, &host, "link-7")?;
    let LinkOpenOutcome::Opened { handshake } = link_open(&runtime, &mut state, offer)? else {
        return Err("an allowed end did not open".into());
    };
    initiator.read(&handshake)?;
    assert!(initiator.is_finished());
    assert!(state.links.contains_key(&LINK_STREAM));
    Ok(())
}

/// A message crosses only when the exporting machine's pass holds
/// `link_export` and this machine's holds `link_import` on the subject; a
/// frame the link's transport did not seal is refused before any decision.
#[test]
fn a_crossing_is_decided_on_both_passes_before_it_is_published() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let remote = machine(1);
    runtime
        .install_link_host(Arc::new(LinkHost::new(
            machine(2),
            Box::new(HostRights {
                import: &["orders"],
            }),
            [("fixture-holder".to_owned(), remote.machine())].into(),
        )))
        .map_err(|_| "a link host was already installed")?;

    // No stream holds a link yet: refused by name.
    let refused = crossing_refusal(link_message(&runtime, &mut state, "orders", vec![1, 2, 3]))?;
    assert!(refused.contains("no link is open on stream"), "{refused}");

    grants.allow_open("link-7", AccessAction::LinkExport)?;
    let (mut initiator, offer) = export_offer(&remote, &machine(2), "link-7")?;
    let LinkOpenOutcome::Opened { handshake } = link_open(&runtime, &mut state, offer)? else {
        return Err("the link did not open".into());
    };
    initiator.read(&handshake)?;
    let mut transport = initiator.finish()?;
    let envelope = MessageEnvelope::new(
        SchemaId::new([1; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        b"crossing".to_vec(),
    );
    let plain = envelope.to_wire_bytes()?;
    let seal = |transport: &mut crate::link_handshake::LinkTransport| -> Result<Vec<u8>, Failure> {
        let mut sealed = vec![0_u8; plain.len() + NOISE_OVERHEAD];
        let written = transport.seal(&plain, &mut sealed)?;
        sealed.truncate(written);
        Ok(sealed)
    };

    // The exporting machine holds no link_export on orders: refused at export.
    let refused = crossing_refusal(link_message(
        &runtime,
        &mut state,
        "orders",
        seal(&mut transport)?,
    ))?;
    assert!(
        refused.starts_with("crossing refused at export:"),
        "{refused}"
    );
    assert!(refused.contains("\"link_export\""), "{refused}");

    // Export granted, but this machine imports only orders: invoices refuse at import.
    grants.grant_link("orders", LinkAction::Export)?;
    grants.grant_link("invoices", LinkAction::Export)?;
    let refused = crossing_refusal(link_message(
        &runtime,
        &mut state,
        "invoices",
        seal(&mut transport)?,
    ))?;
    assert!(
        refused.starts_with("crossing refused at import:"),
        "{refused}"
    );

    // Both ends hold orders: the crossing reaches the door's publish.
    let crossed = link_message(&runtime, &mut state, "orders", seal(&mut transport)?);
    match crossed {
        FrameAction::Respond(Frame::PublishError {
            reason_code,
            message: Some(message),
            ..
        }) => {
            assert_ne!(
                reason_code,
                crate::server::connection::link_route::LINK_CROSSING_REFUSED_CODE
            );
            assert!(
                message.contains("publish is not a route under test"),
                "{message}"
            );
        }
        other => return Err(format!("the crossing did not reach publish: {other:?}").into()),
    }

    // Bytes the transport did not seal are refused before any decision.
    let refused = crossing_refusal(link_message(&runtime, &mut state, "orders", vec![0; 40]))?;
    assert!(refused.starts_with("link frame refused:"), "{refused}");
    Ok(())
}

/// ACCESS-004 R1 and R2 on a Lys pass: the admitted holder is the
/// connection's verified holder, so a worker registers under that name and
/// is refused by name under any other, with nothing stored.
#[test]
fn a_lys_holder_registers_under_its_own_name_and_no_other() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let holder = state
        .verified_holder
        .as_ref()
        .ok_or("a Lys Connect left no verified holder")?;
    assert_eq!(holder.participant(), b"fixture-holder");
    let register = |state: &mut ConnectionProcessState, identity: &str| {
        apply_frame(
            PID,
            &runtime,
            state,
            Frame::WorkerRegister {
                flags: 0,
                registration: WorkerRegistration {
                    namespaces: Vec::new(),
                    task_queue: "findings".to_owned(),
                    node: None,
                    activity_types: Vec::new(),
                    identity: identity.to_owned(),
                    activities: Vec::new(),
                },
            },
        )
    };

    let FrameAction::Respond(Frame::WorkerRegisterAck {
        outcome: WorkerRegisterOutcome::Rejected { reason },
        ..
    }) = register(&mut state, "archie")
    else {
        return Err("a Lys holder registered under another name".into());
    };
    assert_eq!(reason, "registration identity is not the verified holder");
    assert!(!reason.contains("fixture-holder"), "{reason}");
    assert_eq!(runtime.stored_registration_for_tests(PID)?, None);

    assert!(matches!(
        register(&mut state, "fixture-holder"),
        FrameAction::Respond(Frame::WorkerRegisterAck {
            outcome: WorkerRegisterOutcome::Accepted,
            ..
        })
    ));
    Ok(())
}

// ---- ACCESS-003 R1: the service request route ----

/// Lys's HTTP as the service bindings see it: every POST answered 200 and
/// recorded by route, so a seat start is proven at the request Lys received.
#[derive(Debug, Default)]
struct ServiceLys {
    posts: Mutex<Vec<String>>,
}

impl crate::services::lys_http::LysHttp for ServiceLys {
    fn get(
        &self,
        _bearer: &str,
        path: &str,
    ) -> Result<crate::services::lys_http::LysAnswer, crate::services::ServiceEffectError> {
        Err(crate::services::ServiceEffectError(format!(
            "fixture has no answer for GET {path}"
        )))
    }

    fn post(
        &self,
        _bearer: &str,
        path: &str,
        _body: &serde_json::Value,
    ) -> Result<crate::services::lys_http::LysAnswer, crate::services::ServiceEffectError> {
        self.posts
            .lock()
            .map_err(|_| crate::services::ServiceEffectError("poisoned post recorder".to_owned()))?
            .push(path.to_owned());
        Ok(crate::services::lys_http::LysAnswer {
            status: 200,
            body: serde_json::json!({}),
        })
    }
}

/// The publish path behind the key sender: records every subject published
/// on, so a refused key send is proven to have published nothing.
#[derive(Debug, Default)]
struct ServicePublishes {
    published: Mutex<Vec<String>>,
}

impl crate::services::keys::ServicePublisher for ServicePublishes {
    fn publish(
        &self,
        subject: &str,
        _envelope: &MessageEnvelope,
        _operation: &str,
    ) -> Result<(), crate::services::ServiceEffectError> {
        self.published
            .lock()
            .map_err(|_| {
                crate::services::ServiceEffectError("poisoned publish recorder".to_owned())
            })?
            .push(subject.to_owned());
        Ok(())
    }
}

fn install_services(
    runtime: &ConnectionRuntime,
) -> Result<(Arc<ServiceLys>, Arc<ServicePublishes>), Failure> {
    let lys = Arc::new(ServiceLys::default());
    let sink = Arc::new(ServicePublishes::default());
    let http: Arc<dyn crate::services::lys_http::LysHttp> = lys.clone();
    let publisher: Arc<dyn crate::services::keys::ServicePublisher> = sink.clone();
    let bindings = crate::services::production::ServiceBindings::new(
        &http,
        "machine-bearer",
        "mac-1".to_owned(),
        &publisher,
    );
    runtime
        .install_service_bindings(Arc::new(bindings))
        .map_err(|_| "service bindings were already installed")?;
    Ok((lys, sink))
}

fn service_message(
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    body: &str,
) -> FrameAction {
    apply_frame(
        PID,
        runtime,
        state,
        Frame::ConversationMessage {
            flags: liminal::protocol::CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope: MessageEnvelope::new(
                SchemaId::new([3; SchemaId::WIRE_LEN]),
                CausalContext::independent(),
                body.as_bytes().to_vec(),
            ),
        },
    )
}

/// The JSON body a service answered on its conversation.
fn service_reply(action: FrameAction) -> Result<serde_json::Value, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationMessage {
            flags: liminal::protocol::CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: CONVERSATION,
            envelope,
        }) => Ok(serde_json::from_slice(&envelope.payload)?),
        other => Err(format!("the service did not reply on its conversation: {other:?}").into()),
    }
}

/// The refusal a service request answered, carrying the service reason code.
fn service_refusal(action: FrameAction) -> Result<String, Failure> {
    match action {
        FrameAction::Respond(Frame::ConversationError {
            conversation_id: CONVERSATION,
            stream_id: STREAM,
            reason_code: crate::server::connection::service_route::SERVICE_REFUSED_CODE,
            message: Some(message),
            ..
        }) => Ok(message),
        other => Err(
            format!("the service request was not refused on its conversation: {other:?}").into(),
        ),
    }
}

#[test]
fn a_seat_starts_on_an_outright_call_grant_through_the_service_route() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    grants.grant("seats.archie", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    let (lys, publishes) = install_services(&runtime)?;
    assert!(matches!(
        open(&runtime, &mut state, "seats.archie"),
        FrameAction::NoResponse
    ));

    let reply = service_reply(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-1","payload":{}}"#,
    ))?;
    assert_eq!(reply["outcome"], "acted", "the seat start acted: {reply}");
    assert!(
        reply["grant"].is_string(),
        "the reply names the grant: {reply}"
    );
    assert_eq!(
        *lys.posts.lock().map_err(|_| "poisoned")?,
        vec!["agents/archie/start-command".to_owned()],
        "the seat was started through Lys once, on the configured machine"
    );
    assert!(
        publishes
            .published
            .lock()
            .map_err(|_| "poisoned")?
            .is_empty()
    );
    assert_eq!(door.messages()?, 0, "the request never reached the channel");
    Ok(())
}

#[test]
fn a_session_key_send_is_decided_live_and_sends_no_key_when_refused() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    // `call` on the subject opens the conversation; the send itself is the
    // deliberate decision, which this pass has no live answer for.
    grants.grant("sessions.s1.keys", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    let (lys, publishes) = install_services(&runtime)?;
    assert!(matches!(
        open(&runtime, &mut state, "sessions.s1.keys"),
        FrameAction::NoResponse
    ));

    let refusal = service_refusal(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-2","payload":{"keys":"sealed"}}"#,
    ))?;
    assert!(
        refusal.contains(r#""id":"sessions.s1.keys""#) && refusal.contains(r#""action":"call""#),
        "the refusal names the resource and the action: {refusal}"
    );
    assert!(
        publishes
            .published
            .lock()
            .map_err(|_| "poisoned")?
            .is_empty(),
        "no key was published on a refused send"
    );
    assert!(lys.posts.lock().map_err(|_| "poisoned")?.is_empty());
    assert_eq!(door.messages()?, 0);

    grants.allow_open("sessions.s1.keys", AccessAction::Call)?;
    let reply = service_reply(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-3","payload":{"keys":"sealed"}}"#,
    ))?;
    assert_eq!(
        reply["outcome"], "acted",
        "an outright live answer sends once: {reply}"
    );
    assert_eq!(
        *publishes.published.lock().map_err(|_| "poisoned")?,
        vec!["sessions.s1.keys".to_owned()]
    );
    Ok(())
}

#[test]
fn a_service_request_without_bindings_or_operation_is_refused_by_name() -> Result<(), Failure> {
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    grants.grant("alarms", HotAction::Call)?;
    let (runtime, mut state) = admitted(&grants, &door)?;
    assert!(matches!(
        open(&runtime, &mut state, "alarms"),
        FrameAction::NoResponse
    ));
    let refusal = service_refusal(service_message(
        &runtime,
        &mut state,
        r#"{"operation":"op-4"}"#,
    ))?;
    assert!(
        refusal.contains("no service bindings are installed"),
        "refused by name before any decision: {refusal}"
    );

    install_services(&runtime)?;
    let refusal = service_refusal(service_message(&runtime, &mut state, r#"{"payload":{}}"#))?;
    assert!(
        refusal.contains("names no operation id"),
        "a request without an operation does nothing, by name: {refusal}"
    );
    assert_eq!(door.messages()?, 0);
    Ok(())
}

/// SRV-027 R2 (acceptance 8): the Lys `Call` decision precedes the routed
/// push. With a worker registered on the subject, a message under a pass
/// that lacks `Call` is refused by the existing decision before the directory
/// is read: 0 lookups, 0 pushes, 0 channel messages. With `Call` granted the
/// message takes the routed path (1 lookup, still 0 channel messages); this
/// fixture's runtime has no scheduler, so the push is refused by name there
/// and that refusal reaches the caller on its conversation.
#[test]
fn lys_call_decision_precedes_the_push() -> Result<(), Failure> {
    const WORKER: u64 = 12;
    let grants = Arc::new(Grants::default());
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    runtime.register_for_tests(WORKER)?;
    runtime.set_registration(
        WORKER,
        WorkerRegistration {
            namespaces: Vec::new(),
            task_queue: "orders".to_owned(),
            node: None,
            activity_types: Vec::new(),
            identity: "orders-worker".to_owned(),
            activities: Vec::new(),
        },
    )?;
    grants.grant("orders", HotAction::Call)?;
    assert!(matches!(
        open(&runtime, &mut state, "orders"),
        FrameAction::NoResponse
    ));

    grants.revoke("orders", HotAction::Call)?;
    let refusal = conversation_refusal(message(&runtime, &mut state))?;
    assert!(
        refusal.contains(r#""action":"call""#) && refusal.contains(r#""id":"orders""#),
        "{refusal}"
    );
    assert_eq!(
        runtime.directory_lookups(),
        0,
        "the directory is not read before the Call decision"
    );
    assert_eq!(runtime.pending_push_count(), 0, "nothing was pushed");
    assert_eq!(door.messages()?, 0, "a refused message reaches no channel");

    grants.grant("orders", HotAction::Call)?;
    let routed = message(&runtime, &mut state);
    assert_eq!(
        runtime.directory_lookups(),
        1,
        "the decision passed and the directory was read once"
    );
    assert_eq!(
        door.messages()?,
        0,
        "a served subject never reaches the channel"
    );
    let words = conversation_refusal(routed)?;
    assert!(
        words.contains("cannot route to connection process 12"),
        "the fixture has no scheduler, so the push is refused by name: {words}"
    );
    Ok(())
}

/// Places every service, refusing nothing.
#[derive(Debug)]
struct Placed;

impl crate::services::lifecycle::Placer for Placed {
    fn place(
        &self,
        _operation: &str,
        _child: (&str, &str),
        _parent: (&str, &str),
    ) -> Result<(), String> {
        Ok(())
    }
}

/// A registry of the `jobs` template with each named service placed at its
/// millisecond.
fn jobs_placed(
    dir: &std::path::Path,
    placed: &[(&str, u64)],
) -> Result<Arc<crate::services::lifecycle::Lifecycle>, Failure> {
    let lifecycle = crate::services::lifecycle::Lifecycle::open(
        &[crate::config::types::ServiceTemplateConfig {
            name: "jobs".to_owned(),
            command: std::path::PathBuf::from("/bin/sh"),
            args: vec!["-c".to_owned(), "exit 0".to_owned()],
            subjects: vec!["jobs".to_owned()],
            env: Vec::new(),
            stop_timeout_ms: 2_000,
        }],
        dir.to_path_buf(),
        "net".to_owned(),
        Some(Arc::new(Placed)),
    )
    .map_err(|error| format!("the registry opens: {error:?}"))?;
    for (name, at_ms) in placed {
        lifecycle
            .create("jobs", name, "a", &format!("op-{name}"), *at_ms)
            .map_err(|error| format!("{error:?}"))?;
    }
    Ok(lifecycle)
}

#[test]
fn a_call_on_a_service_placed_after_the_pass_is_asked_of_lys_live() -> Result<(), Failure> {
    // the pass was issued at 1,000 s; jobs.late was placed after it, jobs.early before
    let grants = Arc::new(Grants {
        issued_at: Some(1_000),
        ..Grants::default()
    });
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let dir = tempfile::tempdir()?;
    let lifecycle = jobs_placed(dir.path(), &[("early", 999_999), ("late", 1_000_000)])?;
    assert!(runtime.install_lifecycle_for_tests(lifecycle));
    for subject in ["jobs.early", "jobs.late", "elsewhere"] {
        grants.allow_open(subject, HotAction::Call.into())?;
    }

    // placed before the pass: its offline refusal stands, Lys is not asked
    let refused = conversation_refusal(open(&runtime, &mut state, "jobs.early"))?;
    assert!(refused.contains("jobs.early"), "{refused}");
    // no service holds the subject: refused offline as before
    let refused = conversation_refusal(open(&runtime, &mut state, "elsewhere"))?;
    assert!(refused.contains("elsewhere"), "{refused}");
    assert!(door.opens()?.is_empty());
    // placed after the pass: Lys answers outright, and the conversation opens
    let opened = open(&runtime, &mut state, "jobs.late");
    assert!(
        !matches!(
            opened,
            FrameAction::Respond(Frame::ConversationError { .. })
        ),
        "{opened:?}"
    );
    assert_eq!(door.opens()?, vec![(CONVERSATION, "jobs.late".to_owned())]);
    Ok(())
}

#[test]
fn a_late_service_lys_refuses_live_is_refused_with_the_offline_words() -> Result<(), Failure> {
    let grants = Arc::new(Grants {
        issued_at: Some(1_000),
        ..Grants::default()
    });
    let door = Arc::new(Door::default());
    let (runtime, mut state) = admitted(&grants, &door)?;
    let dir = tempfile::tempdir()?;
    let lifecycle = jobs_placed(dir.path(), &[("late", 1_000_000)])?;
    assert!(runtime.install_lifecycle_for_tests(lifecycle));
    let refused = conversation_refusal(open(&runtime, &mut state, "jobs.late"))?;
    assert!(refused.contains("jobs.late"), "{refused}");
    assert!(door.opens()?.is_empty());
    Ok(())
}