liminal-server 0.23.1

use crate::ServerError;

use liminal::durability::StoreCounts;

use serde_json::Value;

use std::error::Error;
use std::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use super::cost::{CostWake, CostWindow, RecordError, SelfWork, ServerMetadata};
use super::events::{PublishOrigin, ServerEvent};
use super::publisher::{CostSink, PublisherStop, run_cost_publisher};

use tokio::sync::{Notify, mpsc};

fn copy_fixture(from: &Path, to: &Path) -> Result<(), Box<dyn Error>> {
    fs::create_dir_all(to)?;
    for entry in fs::read_dir(from)? {
        let entry = entry?;
        let destination = to.join(entry.file_name());
        if entry.file_type()?.is_dir() {
            copy_fixture(&entry.path(), &destination)?;
        } else {
            fs::copy(entry.path(), destination)?;
        }
    }
    Ok(())
}

/// Name of the shared self-streams seed fixture.
const SELF_STREAMS_FIXTURE: &str = "self-streams";

/// Name of the shared clean-shutdown head fixture.
const SELF_HEADS_FIXTURE: &str = "self-heads";

/// The file in the head fixture that records the heads its build measured.
const HEADS_FILE: &str = "heads.json";

/// This test binary's fixture directory and the key of this build, derived
/// independently as `test_fixture`'s module doc lays them out:
/// `target/<profile>/test-fixtures/<binary>/<blake3 of the binary's bytes>`.
/// A cache keyed any other way leaves nothing here (R3b review finding 16).
fn r3_binary_fixtures() -> Result<(PathBuf, String), Box<dyn Error>> {
    let exe = std::env::current_exe()?;
    let name = exe.file_name().ok_or("the test binary has no file name")?;
    let profile = exe
        .parent()
        .and_then(Path::parent)
        .ok_or("the test binary has no target directory above it")?;
    Ok((
        profile.join("test-fixtures").join(name),
        crate::test_fixture::content_hash(&exe)?,
    ))
}

/// Writes `bytes` to a new file at `path` and sets its modification time to
/// `stamp`, answering the file's length and modification time.
fn stamped_binary(
    path: &Path,
    bytes: &[u8],
    stamp: std::time::SystemTime,
) -> Result<(u64, std::time::SystemTime), Box<dyn Error>> {
    let mut file = File::create(path)?;
    file.write_all(bytes)?;
    file.set_modified(stamp)?;
    drop(file);
    let metadata = fs::metadata(path)?;
    Ok((metadata.len(), metadata.modified()?))
}

/// Row 16's two pins, kept out of the unit suite by the default filter in
/// `.config/nextest.toml` and run as their own leg with nothing else
/// building fixtures beside them:
/// `cargo nextest run -p liminal-server --lib --ignore-default-filter -E
/// 'test(/::ceiling_fixture::/)' -j 1`. Each builds both real caches' fixtures,
/// so its time is a ceiling leg's, recorded on the work list's slow-tests row,
/// not a unit test's.
mod ceiling_fixture {
    use std::error::Error;
    use std::fs::{self, OpenOptions};

    use super::{
        SELF_HEADS_FIXTURE, SELF_STREAMS_FIXTURE, clean_self_stream_fixture, r3_binary_fixtures,
        stamped_binary, store_fixture,
    };

    /// Pin for R3b review finding 16, through the path both caches call: two
    /// test binaries with equal length and modification time but different
    /// bytes get different keys, and each cache's fixture is built under this
    /// build's key, the blake3 hash of the binary's bytes.
    ///
    /// The ad hoc key was `<length>-<modification nanos>`, which both binaries
    /// answered, so a rebuilt binary could be handed a seed an older build
    /// wrote. A cache that keys its fixture its own way, rather than through
    /// `test_fixture::fixture_copy`, leaves no complete fixture under this key,
    /// and this goes red.
    #[test]
    fn r3_fixture_key_tells_equal_metadata_apart() -> Result<(), Box<dyn Error>> {
        let root = tempfile::tempdir()?;
        let stamp = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
        let first = root.path().join("generation-a");
        let second = root.path().join("generation-b");
        let first_metadata = stamped_binary(&first, b"liminal-test-binary-a", stamp)?;
        let second_metadata = stamped_binary(&second, b"liminal-test-binary-b", stamp)?;
        assert_eq!(
            first_metadata, second_metadata,
            "positive control failed: the two binaries do not share length and modification time"
        );
        assert_ne!(
            fs::read(&first)?,
            fs::read(&second)?,
            "positive control failed: the two binaries have the same bytes"
        );
        let first_key = crate::test_fixture::content_hash(&first)?;
        let second_key = crate::test_fixture::content_hash(&second)?;
        assert_ne!(
            first_key, second_key,
            "R3b review finding 16: key {first_key} == {second_key}, expected distinct for binaries \
             with different bytes and equal length and modification time"
        );
        drop(store_fixture()?);
        drop(clean_self_stream_fixture(&std::time::Instant::now())?);
        let (binary, key) = r3_binary_fixtures()?;
        for name in [SELF_STREAMS_FIXTURE, SELF_HEADS_FIXTURE] {
            let complete = binary.join(&key).join(name).join("complete");
            assert!(
                complete.exists(),
                "R3b review finding 16: the {name} cache built no fixture under this build's key: \
                 {} is absent",
                complete.display()
            );
        }
        Ok(())
    }

    /// Pin for R3b review finding 16, through the path both caches call: a key
    /// another process is reading survives a newer build's cleanup, and the
    /// newer build's fixture is copied out.
    ///
    /// The reader holds the shared lock on `<key>/key.lock`, the reader protocol
    /// of `test_fixture`. The ad hoc cleanup removed the held key without taking
    /// any lock; `fixture_copy`, which both caches here call, removes another key
    /// only by taking its lock exclusively without waiting, so a held key stays.
    /// A key no one holds is removed, the positive control that the cleanup ran.
    /// The fixture is named for this process, so this call always builds it.
    #[test]
    fn r3_cleanup_spares_a_key_being_read() -> Result<(), Box<dyn Error>> {
        use rustix::fs::{FlockOperation, flock};
        let (binary, _) = r3_binary_fixtures()?;
        let process = std::process::id();
        let held = binary.join(format!("r3-held-key-{process}"));
        let unheld = binary.join(format!("r3-unheld-key-{process}"));
        for key in [&held, &unheld] {
            fs::create_dir_all(key.join("data"))?;
            fs::write(key.join("data").join("row"), b"old")?;
        }
        let reader = OpenOptions::new()
            .create(true)
            .truncate(false)
            .write(true)
            .open(held.join("key.lock"))?;
        flock(&reader, FlockOperation::LockShared)?;
        let name = format!("r3-cleanup-{process}");
        let copied = crate::test_fixture::fixture_copy(&name, |data| {
            fs::write(data.join("row"), b"new")?;
            Ok(())
        });
        let survived = held.join("data").join("row").exists();
        let unheld_removed = !unheld.exists();
        flock(&reader, FlockOperation::Unlock)?;
        drop(reader);
        fs::remove_dir_all(&held)?;
        let (_, key) = r3_binary_fixtures()?;
        fs::remove_dir_all(binary.join(key).join(&name))?;
        let copy = copied?;
        assert_eq!(
            fs::read(copy.path().join("row"))?,
            b"new",
            "the newer build's fixture was not copied out"
        );
        assert!(
            unheld_removed,
            "positive control failed: the build's cleanup left {}, which no one held",
            unheld.display()
        );
        assert!(
            survived,
            "R3b review finding 16: old key removed while held: cleanup removed {} while a reader \
             held its shared key lock",
            held.display()
        );
        Ok(())
    }
}

type StoreFixture = (
    tempfile::TempDir,
    Arc<dyn liminal::durability::DurableStore>,
);

fn store_fixture() -> Result<StoreFixture, Box<dyn Error>> {
    use crate::server::connection::services::{LiminalConnectionServices, open_persistent_store};
    let copy = crate::test_fixture::fixture_copy(SELF_STREAMS_FIXTURE, |data| {
        let store = open_persistent_store(data)?;
        let services =
            LiminalConnectionServices::from_config_with_store(&wire_config()?, Arc::clone(&store))?;
        services.install_self_stream_channels()?;
        liminal::durability::bridge::block_on(store.flush())??;
        drop(services);
        drop(store);
        Ok(())
    })?;
    assert_eq!(
        crate::test_fixture::fixture_builds(SELF_STREAMS_FIXTURE)?,
        1
    );
    let store = open_persistent_store(copy.path())?;
    Ok((copy, store))
}

pub(in crate::server) struct R3Fixture {
    pub(in crate::server) services:
        Arc<crate::server::connection::services::LiminalConnectionServices>,
    pub(in crate::server) store: Arc<dyn liminal::durability::DurableStore>,
    root: tempfile::TempDir,
}

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

impl R3Fixture {
    pub(in crate::server) fn new() -> Result<Self, Box<dyn Error>> {
        let fixture = Self::with_config(&wire_config()?)?;
        fixture.services.install_self_stream_channels()?;
        Ok(fixture)
    }

    fn with_config(config: &crate::config::types::ServerConfig) -> Result<Self, Box<dyn Error>> {
        use crate::server::connection::services::LiminalConnectionServices;
        let (root, store) = store_fixture()?;
        let services = Arc::new(LiminalConnectionServices::from_config_with_store(
            config,
            Arc::clone(&store),
        )?);
        Ok(Self {
            services,
            store,
            root,
        })
    }
}

fn write_raw_frame(
    client: &mut crate::server::connection::LoopbackClientEnd,
    frame: &liminal::protocol::Frame,
) -> Result<(), Box<dyn Error>> {
    let mut bytes = vec![0; liminal::protocol::encoded_len(frame)?];
    let length = liminal::protocol::encode(frame, &mut bytes)?;
    let mut remaining = bytes.get(..length).ok_or("invalid encoded frame length")?;
    while !remaining.is_empty() {
        let written = client.write_timeout(remaining, None)?;
        if written == 0 {
            return Err("raw client closed while writing".into());
        }
        remaining = remaining.get(written..).ok_or("invalid raw write length")?;
    }
    Ok(())
}

fn read_raw_frame(
    client: &mut crate::server::connection::LoopbackClientEnd,
    buffer: &mut Vec<u8>,
) -> Result<liminal::protocol::Frame, Box<dyn Error>> {
    loop {
        match liminal::protocol::decode(buffer) {
            Ok((frame, consumed)) => {
                buffer.drain(..consumed);
                return Ok(frame);
            }
            Err(
                liminal::protocol::ProtocolError::IncompleteHeader { .. }
                | liminal::protocol::ProtocolError::TruncatedPayload { .. },
            ) => {
                let mut chunk = [0; 8192];
                let read = client.read_timeout(&mut chunk, None)?;
                if read == 0 {
                    return Err("raw client closed before a frame arrived".into());
                }
                buffer.extend_from_slice(chunk.get(..read).ok_or("invalid raw read length")?);
            }
            Err(error) => return Err(error.into()),
        }
    }
}

fn wire_config() -> Result<crate::config::types::ServerConfig, Box<dyn Error>> {
    use crate::config::types::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 999_999,
        channels: vec![ChannelDef {
            name: "pressure.source".to_owned(),
            schema_ref: None,
            durable: false,
            loaded_schema: None,
        }],
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    })
}

struct WireFixture {
    services: Arc<crate::server::connection::services::LiminalConnectionServices>,
    supervisor: crate::server::connection::ConnectionSupervisor,
    sink: Arc<super::sink::SelfStreams>,
    wake: Arc<CostWake>,
    incarnation: u64,
    root: tempfile::TempDir,
}

struct PressurePeer {
    client: crate::server::connection::LoopbackClientEnd,
    buffer: Vec<u8>,
}

impl PressurePeer {
    async fn connect(fixture: &WireFixture) -> Result<Self, Box<dyn Error>> {
        use liminal::protocol::{Frame, ProtocolVersion};
        let (client, server) = crate::server::connection::LoopbackDuplex::bounded(256 * 1024);
        fixture.supervisor.spawn_loopback_connection(server)?;
        let mut peer = Self {
            client,
            buffer: Vec::new(),
        };
        peer.send(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        })?;
        assert!(matches!(peer.read()?, Frame::ConnectAck { .. }));
        fixture.sink.event_completed().await;
        Ok(peer)
    }

    fn send(&mut self, frame: &liminal::protocol::Frame) -> Result<(), Box<dyn Error>> {
        write_raw_frame(&mut self.client, frame)
    }

    fn read(&mut self) -> Result<liminal::protocol::Frame, Box<dyn Error>> {
        read_raw_frame(&mut self.client, &mut self.buffer)
    }

    fn barrier(&mut self) -> Result<(), Box<dyn Error>> {
        use liminal::protocol::Frame;
        self.send(&Frame::new_ping(0)?)?;
        assert!(matches!(self.read()?, Frame::Pong { .. }));
        Ok(())
    }

    fn subscribe(&mut self, channel: &str) -> Result<u64, Box<dyn Error>> {
        use liminal::protocol::Frame;
        self.send(&Frame::Subscribe {
            flags: 0,
            stream_id: 7,
            channel: channel.to_owned(),
            accepted_schemas: Vec::new(),
            max_in_flight: 16,
        })?;
        match self.read()? {
            Frame::SubscribeAck {
                stream_id: 7,
                subscription_id,
                ..
            } => Ok(subscription_id),
            frame => Err(format!("expected subscription acknowledgement, got {frame:?}").into()),
        }
    }

    fn accept(&mut self, spelling: &str) -> Result<(), Box<dyn Error>> {
        self.send(&liminal::protocol::Frame::Accept {
            flags: 0,
            stream_id: 7,
            referenced_message_id: liminal::protocol::MessageId::new(spelling),
        })?;
        self.barrier()
    }

    fn delivery(&mut self) -> Result<u64, Box<dyn Error>> {
        match self.read()? {
            liminal::protocol::Frame::Deliver {
                stream_id: 7,
                delivery_seq,
                ..
            } => Ok(delivery_seq),
            frame => {
                Err(format!("expected delivery on the pressure subscription, got {frame:?}").into())
            }
        }
    }
}

fn pressure_metric(name: &str) -> Result<u64, Box<dyn Error>> {
    use liminal::metrics::{MetricValue, global_registry};
    let registry = global_registry().ok_or("server metrics registry is absent")?;
    let snapshot = registry.snapshot();
    let Some(metric) = snapshot.metrics().iter().find(|metric| metric.name == name) else {
        return Ok(0);
    };
    assert!(metric.labels.is_empty());
    match metric.value {
        MetricValue::Counter(value) => Ok(value),
        _ => Err("pressure metric is not a counter".into()),
    }
}

fn self_stream_publication_envelope() -> liminal::protocol::MessageEnvelope {
    liminal::protocol::MessageEnvelope::new(
        liminal::protocol::SchemaId::from_schema_bytes(b"{}"),
        liminal::protocol::CausalContext::default(),
        b"{}".to_vec(),
    )
}

#[test]
fn reserved_namespace_direct_refusal_has_no_store_work() -> Result<(), Box<dyn Error>> {
    use super::events::PublishOrigin;
    crate::metrics::init();
    let fixture = WireFixture::new()?;
    fixture.sink.drain()?;
    let before = fixture.services.store_counts().snapshot();
    let refusals = pressure_metric("liminal_self_streams_reserved_publications_total")?;
    let envelope = self_stream_publication_envelope();
    for channel in ["liminal.events", "liminal.cost", "liminal.unconfigured"] {
        let error = fixture
            .services
            .publish_with_origin(
                channel,
                &envelope,
                Some("forged"),
                PublishOrigin::Connection { id: 7 },
            )
            .err()
            .ok_or("connection publication forged a reserved self record")?;
        assert!(
            matches!(&error, ServerError::ReservedNamespace { channel: refused } if refused == channel)
        );
        assert_eq!(
            error.to_string(),
            format!(
                "SELF_STREAM_NAMESPACE_REFUSED: connection publication to reserved namespace: {channel}"
            )
        );
        assert_eq!(fixture.services.store_counts().snapshot(), before);
    }
    assert_eq!(
        pressure_metric("liminal_self_streams_reserved_publications_total")? - refusals,
        3
    );
    fixture.supervisor.shutdown();
    Ok(())
}

#[tokio::test]
async fn reserved_namespace_wire_refusal_is_named_and_has_no_store_work()
-> Result<(), Box<dyn Error>> {
    use liminal::protocol::Frame;
    let fixture = WireFixture::new()?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    fixture.sink.drain()?;
    let before = fixture.services.store_counts().snapshot();
    for channel in ["liminal.events", "liminal.cost", "liminal.unconfigured"] {
        peer.send(&Frame::new_publish_with_idempotency_key(
            8,
            channel,
            self_stream_publication_envelope(),
            "forged",
        )?)?;
        let Frame::PublishError {
            stream_id,
            reason_code,
            message,
            ..
        } = peer.read()?
        else {
            return Err("reserved publication did not answer PublishError".into());
        };
        assert_eq!(stream_id, 8);
        assert_eq!(reason_code, 0xFFFF);
        assert_eq!(message.as_deref(), Some(format!("SELF_STREAM_NAMESPACE_REFUSED: connection publication to reserved namespace: {channel}").as_str()));
        fixture.sink.drain()?;
        assert_eq!(fixture.services.store_counts().snapshot(), before);
    }
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn reserved_namespace_server_and_ordinary_origins_remain_admitted() -> Result<(), Box<dyn Error>> {
    use super::events::PublishOrigin;
    let fixture = WireFixture::new()?;
    publish_pressure_source(&fixture)?;
    let before = fixture.services.store_counts().snapshot();
    let result = fixture.services.publish_with_origin(
        "liminal.events",
        &self_stream_publication_envelope(),
        None,
        PublishOrigin::Server { device: "local" },
    )?;
    assert!(result.self_row_appended);
    let store = fixture.services.durable_store();
    let row = liminal::durability::bridge::block_on(store.read_at("liminal.events:0", 0))??
        .ok_or("admitted server origin wrote no durable row")?;
    let stored = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
    assert_eq!(stored.publisher_id, "liminal.local");
    assert_eq!(stored.payload, b"{}");
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(delta.append_calls, 1);
    assert_eq!(delta.append_rows, 1);
    fixture.supervisor.shutdown();
    Ok(())
}

fn publish_pressure_source(fixture: &WireFixture) -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::ConnectionServices;
    let envelope = liminal::protocol::MessageEnvelope::new(
        liminal::protocol::SchemaId::from_schema_bytes(b"{}"),
        liminal::protocol::CausalContext::default(),
        b"{}".to_vec(),
    );
    fixture
        .services
        .publish(7, "pressure.source", &envelope, None)?;
    Ok(())
}

#[tokio::test]
async fn external_publication_before_a_self_append_keeps_its_cost_tick()
-> Result<(), Box<dyn Error>> {
    use std::future::{Future, poll_fn};
    use std::pin::pin;
    use std::task::Poll;
    let fixture = WireFixture::new()?;
    publish_pressure_source(&fixture)?;
    fixture.sink.event(ServerEvent::Pressure {
        connection: 1,
        response: "observed",
    });
    let observed = {
        let mut tick = pin!(fixture.wake.next_tick(|duration| {
            assert_eq!(duration, std::time::Duration::from_secs(10));
            std::future::ready(())
        }));
        poll_fn(|context| Poll::Ready(tick.as_mut().poll(context))).await
    };
    fixture.supervisor.shutdown();
    assert!(
        matches!(observed, Poll::Ready(Ok(()))),
        "self append hid an externally armed cost window: {observed:?}"
    );
    Ok(())
}

#[tokio::test]
async fn unknown_pressure_is_counted_without_a_durable_event() -> Result<(), Box<dyn Error>> {
    use liminal::protocol::{Frame, MessageId};
    crate::metrics::init();
    let fixture = WireFixture::new()?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    let before = fixture.services.store_counts().snapshot();
    let unmatched = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    for frame in [
        Frame::Accept {
            flags: 0,
            stream_id: 7,
            referenced_message_id: MessageId::new("missing"),
        },
        Frame::Defer {
            flags: 0,
            stream_id: 7,
            referenced_message_id: MessageId::new("missing"),
            reason: None,
        },
        Frame::Reject {
            flags: 0,
            stream_id: 7,
            referenced_message_id: MessageId::new("missing"),
            reason: None,
        },
    ] {
        peer.send(&frame)?;
    }
    peer.barrier()?;
    let after = fixture.services.store_counts().snapshot();
    let unmatched_after = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    drop(peer);
    fixture.supervisor.shutdown();
    assert_eq!(after.since(&before).append_rows, 0);
    assert_eq!(unmatched_after - unmatched, 3);
    Ok(())
}

#[tokio::test]
async fn matched_pressure_requires_the_issued_sequence_canonical_spelling()
-> Result<(), Box<dyn Error>> {
    crate::metrics::init();
    let fixture = WireFixture::new()?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    peer.subscribe("pressure.source")?;
    publish_pressure_source(&fixture)?;
    let sequence = peer.delivery()?;
    let mut sink = Arc::clone(&fixture.sink);
    CostSink::before_stop(&mut sink)?;
    let before = fixture.services.store_counts().snapshot();
    let unmatched = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    for spelling in [
        format!("0{sequence}"),
        format!("+{sequence}"),
        format!(" {sequence}"),
    ] {
        peer.accept(&spelling)?;
    }
    let noncanonical = fixture.services.store_counts().snapshot();
    peer.accept(&sequence.to_string())?;
    CostSink::before_stop(&mut sink)?;
    let matched = fixture.services.store_counts().snapshot();
    peer.accept(&sequence.to_string())?;
    CostSink::before_stop(&mut sink)?;
    let duplicate = fixture.services.store_counts().snapshot();
    let unmatched_after = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    drop(peer);
    fixture.supervisor.shutdown();
    assert_eq!(noncanonical.since(&before).append_rows, 0);
    assert_eq!(matched.since(&noncanonical).append_rows, 1);
    assert_eq!(duplicate.since(&matched).append_rows, 0);
    assert_eq!(unmatched_after - unmatched, 4);
    Ok(())
}

#[tokio::test]
async fn matched_defer_retains_reject_consumes_and_unsubscribe_forgets_pressure()
-> Result<(), Box<dyn Error>> {
    use liminal::protocol::{Frame, MessageId};
    crate::metrics::init();
    let fixture = WireFixture::new()?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    let subscription_id = peer.subscribe("pressure.source")?;
    publish_pressure_source(&fixture)?;
    let sequence = peer.delivery()?.to_string();
    let events_head = fixture.services.self_stream_head("liminal.events")?;
    let before = fixture.services.store_counts().snapshot();
    let unmatched = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    peer.send(&Frame::Defer {
        flags: 0,
        stream_id: 8,
        referenced_message_id: MessageId::new(&sequence),
        reason: None,
    })?;
    peer.barrier()?;
    fixture.sink.drain()?;
    assert_eq!(
        fixture
            .services
            .store_counts()
            .snapshot()
            .since(&before)
            .append_rows,
        0
    );
    for expected_rows in 1..=2 {
        peer.send(&Frame::Defer {
            flags: 0,
            stream_id: 7,
            referenced_message_id: MessageId::new(&sequence),
            reason: None,
        })?;
        peer.barrier()?;
        fixture.sink.drain()?;
        assert_eq!(
            fixture
                .services
                .store_counts()
                .snapshot()
                .since(&before)
                .append_rows,
            expected_rows
        );
    }
    for _ in 0..2 {
        peer.send(&Frame::Reject {
            flags: 0,
            stream_id: 7,
            referenced_message_id: MessageId::new(&sequence),
            reason: None,
        })?;
        peer.barrier()?;
        fixture.sink.drain()?;
        assert_eq!(
            fixture
                .services
                .store_counts()
                .snapshot()
                .since(&before)
                .append_rows,
            3
        );
    }
    publish_pressure_source(&fixture)?;
    let pending = peer.delivery()?.to_string();
    peer.send(&Frame::Unsubscribe {
        flags: 0,
        stream_id: 7,
        subscription_id,
    })?;
    peer.barrier()?;
    peer.accept(&pending)?;
    fixture.sink.drain()?;
    assert_eq!(
        fixture
            .services
            .store_counts()
            .snapshot()
            .since(&before)
            .append_rows,
        3
    );
    assert_eq!(
        pressure_metric("liminal_self_streams_unmatched_pressure_total")? - unmatched,
        3
    );
    assert_stored_pressure_responses(&fixture, events_head)?;
    drop(peer);
    fixture.supervisor.shutdown();
    Ok(())
}

fn assert_stored_pressure_responses(
    fixture: &WireFixture,
    events_head: u64,
) -> Result<(), Box<dyn Error>> {
    let store = fixture.services.durable_store();
    for (offset, response) in [(0, "defer"), (1, "defer"), (2, "reject")] {
        let position = events_head + offset;
        let row =
            liminal::durability::bridge::block_on(store.read_at("liminal.events:0", position))??
                .ok_or("matched pressure event was not durably appended")?;
        let stored = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        let event: Value = serde_json::from_slice(&stored.payload)?;
        assert_eq!(event["details"]["response"], response, "{event}");
    }
    Ok(())
}

#[tokio::test]
async fn self_stream_pressure_acknowledgement_is_suppressed_and_idle() -> Result<(), Box<dyn Error>>
{
    crate::metrics::init();
    let fixture = WireFixture::new()?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    peer.subscribe("liminal.events")?;
    fixture.sink.event(ServerEvent::Pressure {
        connection: 1,
        response: "seed",
    });
    let sequence = peer.delivery()?;
    publish_initial_cost_window_for_test(&fixture).await?;
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&fixture.wake),
        Arc::clone(&fixture.sink),
        crate::server::shutdown::ShutdownHandle::new(),
    )?;
    fixture.wake.waiting().await;
    let before = fixture.services.store_counts().snapshot();
    let wakes = fixture.wake.notifications();
    let suppressed = pressure_metric("liminal_self_streams_suppressed_pressure_total")?;
    peer.accept(&sequence.to_string())?;
    for _ in 0..3 {
        peer.barrier()?;
    }
    let after = fixture.services.store_counts().snapshot();
    let wakes_after = fixture.wake.notifications();
    let pending = fixture.wake.timer_pending();
    let suppressed_after = pressure_metric("liminal_self_streams_suppressed_pressure_total")?;
    publisher.finish()?;
    drop(peer);
    fixture.supervisor.shutdown();
    assert_eq!(after, before);
    assert_eq!(wakes_after, wakes);
    assert!(!pending);
    assert_eq!(suppressed_after - suppressed, 1);
    Ok(())
}

#[tokio::test]
async fn never_acking_peer_keeps_receiving_and_evicted_pressure_is_unmatched()
-> Result<(), Box<dyn Error>> {
    crate::metrics::init();
    let limits = crate::config::LimitsConfig {
        max_subscription_inbox_depth: 2,
        ..crate::config::LimitsConfig::default()
    };
    let fixture = WireFixture::with_limits(limits, |store| store)?;
    let mut peer = PressurePeer::connect(&fixture).await?;
    peer.subscribe("pressure.source")?;
    let evictions = pressure_metric("liminal_self_streams_pressure_ledger_evictions_total")?;
    for expected in 1..=6 {
        publish_pressure_source(&fixture)?;
        assert_eq!(peer.delivery()?, expected);
    }
    let before = fixture.services.store_counts().snapshot();
    let unmatched = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    peer.accept("1")?;
    let after = fixture.services.store_counts().snapshot();
    let unmatched_after = pressure_metric("liminal_self_streams_unmatched_pressure_total")?;
    let evictions_after = pressure_metric("liminal_self_streams_pressure_ledger_evictions_total")?;
    drop(peer);
    fixture.supervisor.shutdown();
    assert_eq!(after.since(&before).append_rows, 0);
    assert_eq!(unmatched_after - unmatched, 1);
    assert_eq!(evictions_after - evictions, 4);
    Ok(())
}

impl WireFixture {
    fn new() -> Result<Self, Box<dyn Error>> {
        Self::with_store(|store| store)
    }

    fn with_store(
        prepare: impl FnOnce(
            Arc<dyn liminal::durability::DurableStore>,
        ) -> Arc<dyn liminal::durability::DurableStore>,
    ) -> Result<Self, Box<dyn Error>> {
        Self::with_limits(crate::config::LimitsConfig::default(), prepare)
    }

    fn with_limits(
        limits: crate::config::LimitsConfig,
        prepare: impl FnOnce(
            Arc<dyn liminal::durability::DurableStore>,
        ) -> Arc<dyn liminal::durability::DurableStore>,
    ) -> Result<Self, Box<dyn Error>> {
        use crate::server::connection::ConnectionSupervisor;
        use crate::server::connection::services::LiminalConnectionServices;
        use crate::server::shutdown::ShutdownHandle;
        let mut config = wire_config()?;
        config.limits = limits;
        config.participant = Some(crate::server::participant::test_participant_config());
        let (root, store) = store_fixture()?;
        let services = Arc::new(LiminalConnectionServices::from_config_with_store(
            &config,
            prepare(store),
        )?);
        let shutdown = ShutdownHandle::new();
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            services.clone(),
            None,
            None,
            config.limits,
            shutdown.clone(),
        )?;
        let incarnation = supervisor
            .current_server_incarnation()?
            .ok_or("missing server incarnation")?;
        let sink =
            crate::server::runtime::bind_self_streams(&services, &supervisor, &config, &shutdown)?
                .ok_or("missing self-stream binding")?;
        let wake = services.signals().wake();
        Ok(Self {
            services,
            supervisor,
            sink,
            wake,
            incarnation,
            root,
        })
    }
}

/// The first cost row's self rows depend on how the drainer took the
/// Connect rows (`subscribe_counting_connect_batches`). Opening the window
/// after a told signal that the Connect batch was committed would not fit
/// here: the pin asserts the publisher is idle (`wake.waiting()`) before the
/// connection exists, so its first window opens before Connect (Hermes, 7 Oct
/// 12:15 and 12:20).
#[tokio::test(start_paused = true)]
async fn ordinary_wire_delivers_self_records_and_idle_has_no_store_work()
-> Result<(), Box<dyn Error>> {
    use crate::server::connection::LoopbackDuplex;
    use crate::server::connection::services::ConnectionServices;
    use liminal::protocol::Frame;
    let WireFixture {
        services,
        supervisor,
        mut sink,
        wake,
        incarnation,
        root,
    } = WireFixture::new()?;
    let stop = Arc::new(PublisherStop::default());
    let runner_stop = Arc::clone(&stop);
    let runner_wake = Arc::clone(&wake);
    let events = Arc::clone(&sink);
    let task = tokio::spawn(async move {
        run_cost_publisher(
            &runner_wake,
            &runner_stop,
            (StoreCounts::default(), SelfWork::default()),
            tokio::time::sleep,
            &mut sink,
        )
        .await
    });
    wake.waiting().await;
    let (mut client, server) = LoopbackDuplex::bounded(256 * 1024);
    let connection = supervisor.spawn_loopback_connection(server)?;
    let mut buffer = Vec::new();
    let connect_batches = subscribe_counting_connect_batches(&mut client, &mut buffer, &events)?;
    let source = liminal::protocol::MessageEnvelope::new(
        liminal::protocol::SchemaId::from_schema_bytes(b"{}"),
        liminal::protocol::CausalContext::default(),
        b"{}".to_vec(),
    );
    services.publish(7, "pressure.source", &source, None)?;
    let Frame::Deliver {
        stream_id: 3,
        delivery_seq: issued,
        ..
    } = read_raw_frame(&mut client, &mut buffer)?
    else {
        return Err("ordinary source publication was not delivered".into());
    };
    write_raw_frame(
        &mut client,
        &Frame::Accept {
            flags: 0,
            stream_id: 3,
            referenced_message_id: liminal::protocol::MessageId::new(issued.to_string()),
        },
    )?;
    let Frame::Deliver {
        stream_id: 1,
        delivery_seq: event_sequence,
        envelope,
        ..
    } = read_raw_frame(&mut client, &mut buffer)?
    else {
        return Err("ordinary subscription did not deliver the pressure event".into());
    };
    let normalizer = liminal::channel::Schema::new(serde_json::json!({}))?;
    assert_pressure_wire_payload(
        &envelope.payload,
        connection.pid(),
        incarnation,
        &normalizer,
    )?;
    acknowledge_self_record_wire(&mut client, &mut buffer, 1, event_sequence)?;
    wake.timer_started().await;
    tokio::time::advance(std::time::Duration::from_secs(10)).await;
    wake.waiting().await;
    let Frame::Deliver {
        stream_id: 2,
        delivery_seq: cost_sequence,
        envelope,
        ..
    } = read_raw_frame(&mut client, &mut buffer)?
    else {
        return Err("ordinary subscription did not deliver the cost record".into());
    };
    assert_cost_wire_payload(&envelope.payload, &normalizer, connect_batches)?;
    acknowledge_self_record_wire(&mut client, &mut buffer, 2, cost_sequence)?;
    assert!(!wake.armed());
    assert!(!wake.timer_pending());
    let counts = services.store_counts();
    let before_idle = counts.snapshot();
    tokio::time::advance(std::time::Duration::from_secs(25)).await;
    assert_eq!(counts.snapshot(), before_idle);
    assert_eq!(client.readable_bytes(), 0);
    assert!(!wake.armed());
    assert!(!wake.timer_pending());
    stop.request();
    task.await??;
    drop(client);
    supervisor.shutdown();
    drop(supervisor);
    drop(services);
    drop(root);
    Ok(())
}

/// Subscribes as [`subscribe_self_record_wire`] does, and answers how many
/// self batches the Connect rows took. The drainer may take the Connect-time
/// Flush row and `connection_open` as one batch, and a batch adds its own Flush
/// row. Which happened is read from the drainer's own batch count, after the
/// Connect rows are drained, never from the cost row.
fn subscribe_counting_connect_batches(
    client: &mut crate::server::connection::LoopbackClientEnd,
    buffer: &mut Vec<u8>,
    events: &super::sink::SelfStreams,
) -> Result<u64, Box<dyn Error>> {
    let before_connect = events.own_work_for_test()?.flushes;
    subscribe_self_record_wire(client, buffer, events)?;
    let connect_batches = events.own_work_for_test()?.flushes - before_connect;
    assert!(
        connect_batches <= 1,
        "Connect wrote {connect_batches} self batches, more than its one"
    );
    Ok(connect_batches)
}

fn subscribe_self_record_wire(
    client: &mut crate::server::connection::LoopbackClientEnd,
    buffer: &mut Vec<u8>,
    events: &super::sink::SelfStreams,
) -> Result<(), Box<dyn Error>> {
    use liminal::protocol::{Frame, ProtocolVersion};
    write_raw_frame(
        client,
        &Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        },
    )?;
    assert!(matches!(
        read_raw_frame(client, buffer)?,
        Frame::ConnectAck { .. }
    ));
    events.drain()?;
    for (stream_id, channel) in [
        (1, "liminal.events"),
        (2, "liminal.cost"),
        (3, "pressure.source"),
    ] {
        write_raw_frame(
            client,
            &Frame::Subscribe {
                flags: 0,
                stream_id,
                channel: channel.to_owned(),
                accepted_schemas: Vec::new(),
                max_in_flight: 16,
            },
        )?;
        assert!(matches!(
            read_raw_frame(client, buffer)?,
            Frame::SubscribeAck { .. }
        ));
    }
    Ok(())
}

fn acknowledge_self_record_wire(
    client: &mut crate::server::connection::LoopbackClientEnd,
    buffer: &mut Vec<u8>,
    stream_id: u32,
    sequence: u64,
) -> Result<(), Box<dyn Error>> {
    use liminal::protocol::Frame;
    write_raw_frame(
        client,
        &Frame::Accept {
            flags: 0,
            stream_id,
            referenced_message_id: liminal::protocol::MessageId::new(sequence.to_string()),
        },
    )?;
    write_raw_frame(client, &Frame::new_ping(0)?)?;
    assert!(matches!(
        read_raw_frame(client, buffer)?,
        Frame::Pong { .. }
    ));
    Ok(())
}

/// The first cost row: three self rows and no own flush when the drainer
/// appended the Connect rows apart, four and one when it batched them.
fn assert_cost_wire_payload(
    payload: &[u8],
    normalizer: &liminal::channel::Schema,
    connect_batches: u64,
) -> Result<(), Box<dyn Error>> {
    let cost: Value = serde_json::from_slice(payload)?;
    assert_eq!(payload, normalizer.validate_and_apply_defaults(payload)?);
    assert_eq!(cost["v"], 1);
    assert_eq!(cost["previous_snapshot"]["append_rows"], 0);
    assert_eq!(cost["self_rows"], 3 + connect_batches);
    assert_eq!(cost["self_flushes"], connect_batches);
    Ok(())
}

fn assert_pressure_wire_payload(
    payload: &[u8],
    connection: u64,
    incarnation: u64,
    normalizer: &liminal::channel::Schema,
) -> Result<(), Box<dyn Error>> {
    let event: Value = serde_json::from_slice(payload)?;
    let metadata = ServerMetadata::new("local".to_owned(), incarnation)?;
    let expected_event = ServerEvent::Pressure {
        connection,
        response: "accept",
    }
    .encode(
        &metadata,
        event["server_time_ms"]
            .as_u64()
            .ok_or("wire event has no server time")?,
    )?;
    assert_eq!(
        payload,
        normalizer.validate_and_apply_defaults(expected_event)?
    );
    assert_eq!(event["v"], 1);
    assert_eq!(event["details"]["kind"], "pressure");
    assert_eq!(event["details"]["connection"], connection);
    assert_eq!(event["details"]["response"], "accept");
    assert_eq!(event["server_incarnation"], incarnation);
    assert_eq!(event["publisher"], "liminal.local");
    Ok(())
}

#[derive(Default)]
struct ObservedSink {
    rows: AtomicU64,
    snapshots: AtomicU64,
    own_rows: AtomicU64,
    records: Mutex<Vec<Vec<u8>>>,
    published: Notify,
}

impl CostSink for Arc<ObservedSink> {
    fn snapshot(&self) -> Result<(StoreCounts, SelfWork), ServerError> {
        self.snapshots.fetch_add(1, Ordering::Relaxed);
        Ok((
            StoreCounts {
                append_rows: self.rows.load(Ordering::Acquire),
                ..StoreCounts::default()
            },
            SelfWork {
                rows: self.own_rows.load(Ordering::Acquire),
                ..SelfWork::default()
            },
        ))
    }

    fn publish(&mut self, record: &super::cost::PreparedCost) -> Result<(), ServerError> {
        let metadata = ServerMetadata::new("local".to_owned(), 7).map_err(test_sink_error)?;
        let bytes = record.encode(&metadata, 123).map_err(test_sink_error)?;
        self.records.lock().map_err(test_sink_error)?.push(bytes);
        self.rows.fetch_add(1, Ordering::AcqRel);
        self.own_rows.fetch_add(1, Ordering::AcqRel);
        self.published.notify_one();
        Ok(())
    }
}

fn test_sink_error(error: impl std::fmt::Display) -> ServerError {
    ServerError::ListenerAccept {
        message: format!("self-stream test sink failed: {error}"),
    }
}

#[test]
fn server_origin_receipts_count_watermark_shedding_and_admitted_rows() -> Result<(), Box<dyn Error>>
{
    use liminal::channel::{ConnectionInboxBudget, InboxInstall};
    use liminal::pressure::ConsumerCapacity;
    use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};

    let fixture = R3Fixture::new()?;
    let services = &fixture.services;
    let counts = services.store_counts();
    let envelope = MessageEnvelope::new(
        SchemaId::from_schema_bytes(b"{}"),
        CausalContext::default(),
        br#"{"n":1}"#.to_vec(),
    );
    let origin = PublishOrigin::Server { device: "local" };
    let watermark = services.subscribe_handle_for_test_with_install(
        "liminal.cost",
        Some(InboxInstall {
            budget: ConnectionInboxBudget::new(usize::MAX),
            depth_cap: usize::MAX,
            notifier: None,
            capacity: Some(ConsumerCapacity::new(2, 2)?),
        }),
    )?;
    for _ in 0..4 {
        let published = services.publish_with_origin("liminal.cost", &envelope, None, origin)?;
        assert!(published.self_row_appended);
    }
    let before = counts.snapshot();
    let rejected = services.publish_with_origin("liminal.cost", &envelope, None, origin)?;
    assert!(!rejected.self_row_appended);
    assert!(!rejected.outcome.delivered);
    assert_eq!(counts.snapshot().since(&before).append_rows, 0);
    assert_eq!(counts.snapshot().since(&before).flushes, 0);

    let shed = services.subscribe_handle_for_test_with_install(
        "liminal.events",
        Some(InboxInstall {
            budget: ConnectionInboxBudget::new(1),
            depth_cap: usize::MAX,
            notifier: None,
            capacity: Some(ConsumerCapacity::new(64, 64)?),
        }),
    )?;
    let before = counts.snapshot();
    let rejected = services.publish_with_origin("liminal.events", &envelope, None, origin)?;
    assert!(shed.is_overflowed());
    assert!(rejected.self_row_appended);
    assert!(!rejected.outcome.delivered);
    assert_eq!(counts.snapshot().since(&before).append_rows, 1);
    assert_eq!(counts.snapshot().since(&before).flushes, 0);
    drop(shed);

    let before = counts.snapshot();
    let admitted = services.publish_with_origin("liminal.events", &envelope, None, origin)?;
    assert!(admitted.self_row_appended);
    assert_eq!(counts.snapshot().since(&before).append_rows, 1);
    assert_eq!(counts.snapshot().since(&before).flushes, 0);
    assert!(!services.signals().wake().armed());
    drop(watermark);
    Ok(())
}

#[tokio::test(start_paused = true)]
async fn publisher_loop_is_idle_honest_and_stop_publishes_the_pending_window()
-> Result<(), Box<dyn Error>> {
    let wake = Arc::new(CostWake::default());
    let stop = Arc::new(PublisherStop::default());
    let sink = Arc::new(ObservedSink::default());
    let runner_wake = Arc::clone(&wake);
    let runner_stop = Arc::clone(&stop);
    let mut runner_sink = Arc::clone(&sink);
    let task = tokio::spawn(async move {
        run_cost_publisher(
            &runner_wake,
            &runner_stop,
            (StoreCounts::default(), SelfWork::default()),
            tokio::time::sleep,
            &mut runner_sink,
        )
        .await
    });
    wake.waiting().await;
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 0);
    sink.rows.store(4, Ordering::Release);
    tokio::time::advance(std::time::Duration::from_secs(25)).await;
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 0);
    assert!(sink.records.lock().map_err(test_sink_error)?.is_empty());
    assert!(!wake.timer_pending());

    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.timer_started().await;
    tokio::time::advance(std::time::Duration::from_secs(10)).await;
    sink.published.notified().await;
    wake.waiting().await;
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 1);
    {
        let records = sink.records.lock().map_err(test_sink_error)?;
        assert_eq!(records.len(), 1);
        let record: Value = serde_json::from_slice(&records[0])?;
        drop(records);
        assert_eq!(record["counts"]["append_rows"], 4);
        assert_eq!(record["snapshot"]["append_rows"], 4);
    }
    wake.arm(&PublishOrigin::Server { device: "local" });
    tokio::time::advance(std::time::Duration::from_secs(25)).await;
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 1);
    assert_eq!(sink.records.lock().map_err(test_sink_error)?.len(), 1);
    assert!(!wake.timer_pending());

    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.timer_started().await;
    assert!(wake.timer_pending());
    stop.request();
    task.await??;
    assert!(!wake.timer_pending());
    assert!(!wake.armed());
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 2);
    let records = sink.records.lock().map_err(test_sink_error)?;
    assert_eq!(records.len(), 2);
    let final_record: Value = serde_json::from_slice(&records[1])?;
    drop(records);
    assert_eq!(final_record["previous_snapshot"]["append_rows"], 4);
    assert_eq!(final_record["counts"]["append_rows"], 1);
    assert_eq!(final_record["self_rows"], 1);
    Ok(())
}

#[tokio::test(start_paused = true)]
async fn publisher_stop_before_start_needs_no_timer() -> Result<(), Box<dyn Error>> {
    let wake = CostWake::default();
    let stop = PublisherStop::default();
    stop.request();
    let mut sink = Arc::new(ObservedSink::default());
    run_cost_publisher(
        &wake,
        &stop,
        (StoreCounts::default(), SelfWork::default()),
        tokio::time::sleep,
        &mut sink,
    )
    .await?;
    assert!(!wake.timer_pending());
    assert_eq!(sink.snapshots.load(Ordering::Acquire), 1);
    assert!(sink.records.lock().map_err(test_sink_error)?.is_empty());
    Ok(())
}

#[tokio::test(start_paused = true)]
async fn publisher_stop_publishes_the_pending_cost_window_without_waiting_for_a_timer()
-> Result<(), Box<dyn Error>> {
    let wake = Arc::new(CostWake::default());
    let stop = Arc::new(PublisherStop::default());
    let sink = Arc::new(ObservedSink::default());
    let runner_wake = Arc::clone(&wake);
    let runner_stop = Arc::clone(&stop);
    let mut runner_sink = Arc::clone(&sink);
    let task = tokio::spawn(async move {
        run_cost_publisher(
            &runner_wake,
            &runner_stop,
            (StoreCounts::default(), SelfWork::default()),
            tokio::time::sleep,
            &mut runner_sink,
        )
        .await
    });
    wake.waiting().await;
    sink.rows.store(4, Ordering::Release);
    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.timer_started().await;
    stop.request();
    task.await??;
    let records = sink.records.lock().map_err(test_sink_error)?;
    assert_eq!(
        records.len(),
        1,
        "shutdown discarded its pending cost window"
    );
    let record: Value = serde_json::from_slice(&records[0])?;
    drop(records);
    assert_eq!(record["counts"]["append_rows"], 4);
    assert_eq!(record["previous_snapshot"]["append_rows"], 0);
    assert!(!wake.timer_pending());
    Ok(())
}

#[tokio::test(start_paused = true)]
async fn first_cost_record_includes_startup_work() -> Result<(), Box<dyn Error>> {
    let wake = Arc::new(CostWake::default());
    let stop = Arc::new(PublisherStop::default());
    let sink = Arc::new(ObservedSink::default());
    sink.rows.store(3, Ordering::Release);
    let runner_wake = Arc::clone(&wake);
    let runner_stop = Arc::clone(&stop);
    let mut runner_sink = Arc::clone(&sink);
    let task = tokio::spawn(async move {
        run_cost_publisher(
            &runner_wake,
            &runner_stop,
            (StoreCounts::default(), SelfWork::default()),
            tokio::time::sleep,
            &mut runner_sink,
        )
        .await
    });
    wake.waiting().await;
    sink.rows.store(6, Ordering::Release);
    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.timer_started().await;
    tokio::time::advance(std::time::Duration::from_secs(10)).await;
    sink.published.notified().await;
    wake.waiting().await;
    stop.request();
    task.await??;
    let records = sink.records.lock().map_err(test_sink_error)?;
    assert_eq!(records.len(), 2);
    let record: Value = serde_json::from_slice(&records[0])?;
    let final_record: Value = serde_json::from_slice(&records[1])?;
    drop(records);
    assert_eq!(record["counts"]["append_rows"], 6);
    assert_eq!(record["previous_snapshot"]["append_rows"], 0);
    assert_eq!(final_record["counts"]["append_rows"], 1);
    assert_eq!(final_record["previous_snapshot"]["append_rows"], 6);
    assert_eq!(final_record["self_rows"], 1);
    Ok(())
}

#[test]
fn failed_publication_keeps_all_twenty_two_counter_deltas() -> Result<(), Box<dyn Error>> {
    let earlier = StoreCounts::default();
    let later = StoreCounts {
        apply_routed_batch_calls: 19,
        apply_routed_batch_items: 20,
        get_routed_calls: 21,
        get_routed_rows: 22,
        append_calls: 1,
        append_rows: 2,
        append_bytes: 3,
        append_all_calls: 4,
        append_all_rows: 5,
        append_all_bytes: 6,
        read_from_calls: 7,
        read_from_rows: 8,
        read_at_calls: 9,
        read_at_rows: 10,
        cas_calls: 11,
        read_value_calls: 12,
        read_stream_next_seq_calls: 18,
        scan_page_calls: 13,
        scan_page_rows: 14,
        flushes: 15,
        truncate_calls: 16,
        truncate_rows: 17,
    };
    let mut window = CostWindow::new(earlier, SelfWork::default());
    let own = SelfWork {
        rows: 2,
        flushes: 1,
        ..SelfWork::default()
    };
    let first = window.prepare(later, own).ok_or("missing first delta")?;
    assert_eq!(first.delta(), later);
    let retry = window.prepare(later, own).ok_or("missing retry delta")?;
    assert_eq!(retry.delta(), later);
    window.published(&retry);
    assert!(window.prepare(later, own).is_none());
    Ok(())
}

#[test]
fn relaxed_call_and_row_snapshots_are_reported_as_taken() -> Result<(), Box<dyn Error>> {
    let mut window = CostWindow::new(StoreCounts::default(), SelfWork::default());
    let call = StoreCounts {
        append_calls: 1,
        ..StoreCounts::default()
    };
    let call_record = window
        .prepare(call, SelfWork::default())
        .ok_or("missing call")?;
    assert_eq!(call_record.delta().append_calls, 1);
    assert_eq!(call_record.delta().append_rows, 0);
    window.published(&call_record);
    let rows = StoreCounts {
        append_rows: 1,
        append_bytes: 8,
        ..call
    };
    let row_record = window
        .prepare(rows, SelfWork::default())
        .ok_or("missing row")?;
    assert_eq!(row_record.delta().append_calls, 0);
    assert_eq!(row_record.delta().append_rows, 1);
    assert_eq!(row_record.delta().append_bytes, 8);
    Ok(())
}

#[test]
fn cost_json_preserves_snapshots_own_work_and_canonical_order() -> Result<(), Box<dyn Error>> {
    let metadata = ServerMetadata::new("local".to_owned(), 7)?;
    let before = StoreCounts {
        append_rows: 4,
        flushes: 2,
        read_stream_next_seq_calls: 2,
        ..StoreCounts::default()
    };
    let after = StoreCounts {
        append_rows: 9,
        flushes: 5,
        // Each operation-log walk captures one head. This fixture's two earlier
        // walks and five total walks leave three single-head walks in the window.
        read_stream_next_seq_calls: 5,
        ..before
    };
    let window = CostWindow::new(
        before,
        SelfWork {
            rows: 1,
            flushes: 1,
            ..SelfWork::default()
        },
    );
    let record = window
        .prepare(
            after,
            SelfWork {
                rows: 3,
                flushes: 2,
                ..SelfWork::default()
            },
        )
        .ok_or("missing cost record")?;
    let bytes = record.encode(&metadata, 123)?;
    let json: Value = serde_json::from_slice(&bytes)?;
    assert_eq!(json["counts"]["append_rows"], 5);
    assert_eq!(json["counts"]["flushes"], 3);
    assert_eq!(json["counts"]["read_stream_next_seq_calls"], 3);
    eprintln!(
        "COST_HEAD_DERIVATION one head per walk * (5 total - 2 baseline walks) = 3 window reads"
    );
    assert_eq!(json["previous_snapshot"]["read_stream_next_seq_calls"], 2);
    assert_eq!(json["snapshot"]["read_stream_next_seq_calls"], 5);
    assert_eq!(json["previous_snapshot"]["append_rows"], 4);
    assert_eq!(json["snapshot"]["append_rows"], 9);
    assert_eq!(json["self_rows"], 2);
    assert_eq!(json["self_flushes"], 1);
    assert_eq!(json["self_prunes"], 0);
    assert_eq!(json["self_rows_pruned"], 0);
    assert_eq!(json["publisher"], "liminal.local");
    assert_eq!(json["server_incarnation"], 7);
    assert_eq!(json["server_time_ms"], 123);
    assert!(bytes.starts_with(br#"{"v":1,"build_commit":"#));
    let text = std::str::from_utf8(&bytes)?;
    assert!(text.contains(r#""counts":{"append_all_bytes":0,"append_all_calls":0,"append_all_rows":0,"append_bytes":0,"append_calls":0,"append_rows":5,"apply_routed_batch_calls":0,"apply_routed_batch_items":0,"cas_calls":0,"flushes":3,"get_routed_calls":0,"get_routed_rows":0,"read_at_calls":0,"read_at_rows":0,"read_from_calls":0,"read_from_rows":0,"read_stream_next_seq_calls":3,"read_value_calls":0,"scan_page_calls":0,"scan_page_rows":0,"truncate_calls":0,"truncate_rows":0}"#));
    let keys = [
        "build_commit",
        "counts",
        "crate_version",
        "device",
        "previous_snapshot",
        "publisher",
        "self_flushes",
        "self_prune_failures",
        "self_prunes",
        "self_rows",
        "self_rows_pruned",
        "server_incarnation",
        "server_time_ms",
        "snapshot",
    ];
    let mut previous = 0;
    for key in keys {
        let position = text
            .find(&format!("\"{key}\":"))
            .ok_or("missing root key")?;
        assert!(position > previous);
        previous = position;
    }
    Ok(())
}

#[test]
fn every_server_event_has_its_context_and_server_time() -> Result<(), Box<dyn Error>> {
    let metadata = ServerMetadata::new("local".to_owned(), 7)?;
    let events = [
        ServerEvent::Lifecycle {
            conversation: 3,
            generation: 4,
            transition: "enrolled",
        },
        ServerEvent::Pressure {
            connection: 5,
            response: "backpressure",
        },
        ServerEvent::ConnectionOpened { connection: 6 },
        ServerEvent::ConnectionClosed { connection: 6 },
    ];
    let expected = [
        r#""details":{"conversation":3,"generation":4,"kind":"lifecycle","transition":"enrolled"}"#,
        r#""details":{"connection":5,"kind":"pressure","response":"backpressure"}"#,
        r#""details":{"connection":6,"kind":"connection_open"}"#,
        r#""details":{"connection":6,"kind":"connection_close"}"#,
    ];
    for (event, expected_details) in events.into_iter().zip(expected) {
        let bytes = event.encode(&metadata, 123)?;
        let text = std::str::from_utf8(&bytes)?;
        assert!(text.starts_with(r#"{"v":1,"build_commit":"#));
        assert!(text.contains(expected_details));
        let json: Value = serde_json::from_slice(&bytes)?;
        assert_eq!(json["server_incarnation"], 7);
        assert_eq!(json["server_time_ms"], 123);
    }
    Ok(())
}

#[test]
fn record_boundaries_reject_empty_descriptions() -> Result<(), Box<dyn Error>> {
    assert!(matches!(
        ServerMetadata::new(" ".to_owned(), 7),
        Err(RecordError::EmptyDevice)
    ));
    let metadata = ServerMetadata::new("local".to_owned(), 7)?;
    let lifecycle = ServerEvent::Lifecycle {
        conversation: 3,
        generation: 4,
        transition: "",
    };
    assert!(matches!(
        lifecycle.encode(&metadata, 123),
        Err(RecordError::EmptyLifecycleTransition)
    ));
    let pressure = ServerEvent::Pressure {
        connection: 5,
        response: " ",
    };
    assert!(matches!(
        pressure.encode(&metadata, 123),
        Err(RecordError::EmptyPressureResponse)
    ));
    Ok(())
}

#[tokio::test(start_paused = true)]
async fn only_external_work_arms_one_tick_then_returns_to_timer_free_wait()
-> Result<(), Box<dyn Error>> {
    let wake = Arc::new(CostWake::default());
    let runner = Arc::clone(&wake);
    let (ticks, mut received) = mpsc::unbounded_channel();
    let task = tokio::spawn(async move {
        loop {
            runner.next_tick(tokio::time::sleep).await?;
            if ticks.send(()).is_err() {
                break;
            }
        }
        Ok::<(), ServerError>(())
    });
    wake.waiting().await;
    assert!(!wake.armed());
    assert!(!wake.timer_pending());
    tokio::time::advance(std::time::Duration::from_secs(25)).await;
    assert!(matches!(
        received.try_recv(),
        Err(mpsc::error::TryRecvError::Empty)
    ));
    wake.arm(&PublishOrigin::Server { device: "local" });
    assert!(!wake.armed());
    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.arm(&PublishOrigin::Connection { id: 7 });
    wake.timer_started().await;
    assert!(wake.armed());
    assert!(wake.timer_pending());
    tokio::time::advance(std::time::Duration::from_secs(10)).await;
    received.recv().await.ok_or("tick task ended")?;
    wake.waiting().await;
    assert!(!wake.armed());
    assert!(!wake.timer_pending());
    tokio::time::advance(std::time::Duration::from_secs(25)).await;
    assert!(matches!(
        received.try_recv(),
        Err(mpsc::error::TryRecvError::Empty)
    ));
    task.abort();
    let exit = task.await;
    assert!(matches!(exit, Err(error) if error.is_cancelled()));
    Ok(())
}

#[test]
fn self_stream_binding_installs_signals_without_participant() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::ConnectionSupervisor;
    use crate::server::shutdown::ShutdownHandle;
    let config = wire_config()?;
    let fixture = R3Fixture::with_config(&config)?;
    let services = &fixture.services;
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    assert!(supervisor.current_server_incarnation()?.is_none());
    crate::metrics::init();
    let log = crate::test_log::CapturedLog::default();
    let binding = log.capture(|| {
        crate::server::runtime::bind_self_streams(services, &supervisor, &config, &shutdown)
    });
    assert!(binding?.is_none());
    assert_eq!(
        log.contents()
            .matches("SELF_STREAMS_REQUIRE_PARTICIPANT")
            .count(),
        1
    );
    assert_eq!(disabled_self_streams_counter()?, 1);
    assert!(
        supervisor.signals_installed(),
        "shared signals were skipped without participant configuration"
    );
    supervisor.shutdown();
    Ok(())
}

/// SRV-012 R4: without a participant the console streams are off, said once
/// in the log and once on the counter, and a console subscribe answers
/// `CHANNEL_NOT_REGISTERED`.
#[test]
fn console_streams_require_a_participant() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::ConnectionSupervisor;
    use crate::server::connection::services::{ChannelOperation, ConnectionServices};
    use crate::server::shutdown::ShutdownHandle;
    use liminal_protocol::reason_code::CHANNEL_NOT_REGISTERED_CODE;
    let config = wire_config()?;
    let fixture = R3Fixture::with_config(&config)?;
    let services = &fixture.services;
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    crate::metrics::init();
    let log = crate::test_log::CapturedLog::default();
    let binding = log.capture(|| {
        crate::server::runtime::bind_self_streams(services, &supervisor, &config, &shutdown)
    });
    assert!(binding?.is_none());
    assert_eq!(
        log.contents()
            .matches("CONSOLE_STREAMS_REQUIRE_PARTICIPANT")
            .count(),
        1
    );
    assert_eq!(disabled_console_streams_counter()?, 1);
    assert!(!services.signals().console_enabled());
    // The inherent `admit_channel` shadows the trait's; the frame path asks
    // through the trait.
    let refusal = ConnectionServices::admit_channel(
        services.as_ref(),
        ChannelOperation::Subscribe,
        crate::console::TOPOLOGY_SUBJECT,
    )
    .err()
    .ok_or("a console subscribe without a participant must be refused")?;
    assert_eq!(refusal.reason_code(), CHANNEL_NOT_REGISTERED_CODE);
    supervisor.shutdown();
    Ok(())
}

#[derive(Clone, Copy, Debug)]
enum ShutdownStoreCall {
    AppendStarted,
    AppendCompleted,
    Cost(u64),
    Event(u64),
    Flush,
}

#[derive(Debug)]
struct CommitThenErrorStore {
    inner: Arc<dyn liminal::durability::DurableStore>,
    fail_cost: std::sync::atomic::AtomicBool,
    commit_cost: bool,
    partial_batch: bool,
    fail_flush: std::sync::atomic::AtomicBool,
    pause_cost: Option<Arc<AppendPause>>,
    calls: Option<Arc<Mutex<Vec<ShutdownStoreCall>>>>,
    observer_boundary: Option<Arc<ObserverBoundary>>,
}

impl CommitThenErrorStore {
    fn record_call(
        &self,
        call: ShutdownStoreCall,
    ) -> Result<(), liminal::durability::DurabilityError> {
        if let Some(calls) = &self.calls {
            calls.lock().map_err(append_pause_error)?.push(call);
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl liminal::durability::DurableStore for CommitThenErrorStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        key: &str,
        payload: Vec<u8>,
        expected: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        let call = match key {
            "liminal.cost:0" => Some(ShutdownStoreCall::Cost(expected)),
            "liminal.events:0" => Some(ShutdownStoreCall::Event(expected)),
            _ => None,
        };
        self.record_call(ShutdownStoreCall::AppendStarted)?;
        if key == "liminal.events:0" {
            if let Some(boundary) = &self.observer_boundary {
                if std::thread::current().id() == boundary.caller {
                    boundary.inline_appends.fetch_add(1, Ordering::Relaxed);
                    return Err(append_pause_error("SELF_STREAM_OBSERVER_INLINE_APPEND"));
                }
            }
        }
        if key == "liminal.cost:0" {
            if let Some(pause) = &self.pause_cost {
                pause.wait_for_release()?;
            }
        }
        if key == "liminal.cost:0" && self.fail_cost.swap(false, Ordering::SeqCst) {
            if self.commit_cost {
                self.inner.append(key, payload, expected).await?;
            }
            return Err(liminal::durability::DurabilityError::StoreError(
                haematite::ApiError::Storage(haematite::DatabaseError::IoError(
                    std::io::Error::other("committed self append failed to answer"),
                )),
            ));
        }
        let result = self.inner.append(key, payload, expected).await;
        if result.is_ok() {
            self.record_call(ShutdownStoreCall::AppendCompleted)?;
            if key == "liminal.events:0" {
                if let Some(boundary) = &self.observer_boundary {
                    boundary.appended.send(()).map_err(append_pause_error)?;
                }
            }
            if let Some(call) = call {
                self.record_call(call)?;
            }
        }
        result
    }

    async fn append_all(
        &self,
        key: &str,
        payloads: Vec<Vec<u8>>,
        expected: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        self.record_call(ShutdownStoreCall::AppendStarted)?;
        if key == "liminal.cost:0" && self.partial_batch {
            let first = payloads
                .into_iter()
                .next()
                .ok_or_else(|| append_pause_error("partial batch has no row"))?;
            self.inner.append(key, first, expected).await?;
            return Err(append_pause_error(
                "partial receipt batch append failed to answer",
            ));
        }

        if key == "liminal.cost:0" && self.fail_cost.swap(false, Ordering::SeqCst) {
            if self.commit_cost {
                self.inner.append_all(key, payloads, expected).await?;
            }
            return Err(liminal::durability::DurabilityError::StoreError(
                haematite::ApiError::Storage(haematite::DatabaseError::IoError(
                    std::io::Error::other("receipt batch append failed to answer"),
                )),
            ));
        }
        let result = self.inner.append_all(key, payloads, expected).await;
        if result.is_ok() {
            self.record_call(ShutdownStoreCall::AppendCompleted)?;
        }
        result
    }

    async fn read_from(
        &self,
        key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<liminal::durability::StoredEntry>, liminal::durability::DurabilityError> {
        self.inner.read_from(key, offset, limit).await
    }

    async fn read_at(
        &self,
        key: &str,
        sequence: u64,
    ) -> Result<Option<liminal::durability::StoredEntry>, liminal::durability::DurabilityError>
    {
        self.inner.read_at(key, sequence).await
    }

    async fn cas(
        &self,
        key: &str,
        old: u64,
        new: u64,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.cas(key, old, new).await
    }

    async fn read_stream_next_seq(
        &self,
        stream_key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(
        &self,
        key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<liminal::durability::ScanCursor>,
        limit: usize,
    ) -> Result<liminal::durability::ScanPage, liminal::durability::DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), liminal::durability::DurabilityError> {
        if self.fail_flush.swap(false, Ordering::SeqCst) {
            return Err(append_pause_error("receipt batch flush failed"));
        }
        if let Some(calls) = &self.calls {
            calls
                .lock()
                .map_err(append_pause_error)?
                .push(ShutdownStoreCall::Flush);
        }
        self.inner.flush().await
    }

    fn store_counts(&self) -> Option<liminal::durability::StoreCountsHandle> {
        self.inner.store_counts()
    }
}

#[test]
fn own_row_is_counted_after_committed_unresolved_append() -> Result<(), Box<dyn Error>> {
    unresolved_self_row_attribution(true, 1)
}

#[test]
fn absent_unresolved_self_row_is_not_attributed() -> Result<(), Box<dyn Error>> {
    unresolved_self_row_attribution(false, 0)
}

fn unresolved_self_row_attribution(
    commit_cost: bool,
    expected_rows: u64,
) -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::LiminalConnectionServices;
    use crate::server::self_streams::sink::SelfStreams;
    use crate::server::shutdown::ShutdownHandle;
    use liminal::durability::bridge::block_on;
    let fixture = R3Fixture::new()?;
    let faults = Arc::new(CommitThenErrorStore {
        inner: Arc::clone(&fixture.store),
        fail_cost: std::sync::atomic::AtomicBool::new(true),
        commit_cost,
        partial_batch: false,
        fail_flush: std::sync::atomic::AtomicBool::new(false),
        pause_cost: None,
        calls: None,
        observer_boundary: None,
    });
    let services = Arc::new(LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        faults,
    )?);
    services.install_self_stream_channels()?;
    let mut sink = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 91)?,
        &services,
        services.signals().wake(),
        ShutdownHandle::new(),
        crate::config::types::SelfStreamsConfig::default(),
    ));
    let counts = services.store_counts();
    let before = counts.snapshot();
    let record = CostWindow::new(StoreCounts::default(), SelfWork::default())
        .prepare(before, SelfWork::default())
        .ok_or("fixture produced no startup counts")?;
    let result = CostSink::publish(&mut sink, &record);
    if commit_cost {
        assert!(
            result.is_ok(),
            "confirmed self row was treated as fatal: {result:?}"
        );
    } else {
        assert!(matches!(result, Err(ServerError::ListenerAccept { .. })));
    }
    let after = counts.snapshot();
    let own = sink.own_work_for_test()?;
    assert_eq!(after.since(&before).append_rows, expected_rows);
    assert_eq!(after.since(&before).flushes, 0);
    assert_eq!(
        own.rows, expected_rows,
        "committed unresolved self row was not attributed"
    );
    assert_eq!(after.since(&before).read_at_calls, 1);
    assert_eq!(after.since(&before).read_at_rows, expected_rows);
    assert_eq!(after.since(&before).read_from_calls, 0);
    let row = block_on(fixture.store.read_at("liminal.cost:0", 0))??;
    assert_eq!(u64::from(row.is_some()), expected_rows);
    if let Some(row) = row {
        assert_eq!(row.sequence, 0);
    }
    assert!(block_on(fixture.store.read_at("liminal.cost:0", 1))??.is_none());
    Ok(())
}

fn disabled_self_streams_counter() -> Result<u64, Box<dyn Error>> {
    use liminal::metrics::{MetricValue, global_registry};
    let registry = global_registry().ok_or("server metrics registry is absent")?;
    let snapshot = registry.snapshot();
    let metric = snapshot
        .metrics()
        .iter()
        .find(|metric| metric.name == "liminal_self_streams_disabled_total")
        .ok_or("liminal_self_streams_disabled_total is absent")?;
    assert!(
        metric.labels.is_empty(),
        "disabled self streams counter must be unlabelled"
    );
    match metric.value {
        MetricValue::Counter(value) => Ok(value),
        MetricValue::Gauge(_) | MetricValue::Histogram(_) => {
            Err("disabled self streams metric is not a counter".into())
        }
    }
}

fn disabled_console_streams_counter() -> Result<u64, Box<dyn Error>> {
    use liminal::metrics::{MetricValue, global_registry};
    let registry = global_registry().ok_or("server metrics registry is absent")?;
    let snapshot = registry.snapshot();
    let metric = snapshot
        .metrics()
        .iter()
        .find(|metric| metric.name == "liminal_console_streams_disabled_total")
        .ok_or("liminal_console_streams_disabled_total is absent")?;
    assert!(
        metric.labels.is_empty(),
        "disabled console streams counter must be unlabelled"
    );
    match metric.value {
        MetricValue::Counter(value) => Ok(value),
        MetricValue::Gauge(_) | MetricValue::Histogram(_) => {
            Err("disabled console streams metric is not a counter".into())
        }
    }
}

#[test]
fn participant_startup_keeps_self_streams_disabled_counter_at_zero() -> Result<(), Box<dyn Error>> {
    crate::metrics::init();
    let log = crate::test_log::CapturedLog::default();
    let fixture = log.capture(WireFixture::new)?;
    assert_eq!(disabled_self_streams_counter()?, 0);
    assert!(!log.contents().contains("SELF_STREAMS_REQUIRE_PARTICIPANT"));
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn self_stream_build_commit_is_the_build_stamp() {
    assert_eq!(super::cost::build_commit(), env!("LIMINAL_BUILD_COMMIT"));
}

#[path = "../../../build.rs"]
mod build_script;

#[path = "build_metadata_emission_tests.rs"]
mod build_metadata_emission_tests;

fn metadata_repository() -> Result<tempfile::TempDir, Box<dyn Error>> {
    let fixture = tempfile::tempdir()?;
    fs::write(
        fixture.path().join("Cargo.toml"),
        "[workspace]\nmembers = []\n",
    )?;
    fixture_git(fixture.path(), &["init", "-b", "fixture"])?;
    fixture_git(
        fixture.path(),
        &[
            "-c",
            "user.name=Fixture",
            "-c",
            "user.email=fixture@example.invalid",
            "commit",
            "--allow-empty",
            "-m",
            "first",
        ],
    )?;
    Ok(fixture)
}

#[test]
fn build_metadata_ignores_consumer_git_for_workspace_archive() -> Result<(), Box<dyn Error>> {
    let consumer = metadata_repository()?;
    let workspace = consumer.path().join("archive");
    let manifest = workspace.join("crates/server");
    fs::create_dir_all(&manifest)?;
    fs::write(workspace.join("Cargo.toml"), "[workspace]\nmembers = []\n")?;
    fs::write(manifest.join("Cargo.toml"), "[package]\nname = 'fixture'\n")?;
    let metadata = build_script::build_metadata(&manifest);
    assert_eq!(metadata.commit, "unknown");
    assert!(
        metadata
            .warning
            .as_deref()
            .is_some_and(|warning| warning.contains("workspace has no git metadata"))
    );
    assert!(metadata.watched.is_empty());
    Ok(())
}

#[test]
fn build_metadata_broken_or_empty_git_does_not_refuse_build() -> Result<(), Box<dyn Error>> {
    let fixture = tempfile::tempdir()?;
    fs::write(
        fixture.path().join("Cargo.toml"),
        "[workspace]\nmembers = []\n",
    )?;
    fixture_git(fixture.path(), &["init", "-b", "fixture"])?;
    let empty = build_script::build_metadata(fixture.path());
    assert_eq!(empty.commit, "unknown");
    assert!(
        empty
            .warning
            .as_deref()
            .is_some_and(|warning| warning.contains("git command failed"))
    );
    fs::remove_dir_all(fixture.path().join(".git"))?;
    fs::write(fixture.path().join(".git"), "gitdir: absent")?;
    let broken = build_script::build_metadata(fixture.path());
    assert_eq!(broken.commit, "unknown");
    assert!(
        broken
            .warning
            .as_deref()
            .is_some_and(|warning| warning.contains("git command failed"))
    );
    Ok(())
}

#[test]
fn build_metadata_watches_only_head_and_current_reference() -> Result<(), Box<dyn Error>> {
    let fixture = metadata_repository()?;
    let repository = fs::canonicalize(fixture.path())?;
    let first = build_script::build_metadata(&repository);
    assert_ne!(first.commit, "unknown");
    assert!(first.warning.is_none());
    assert_eq!(
        first.watched,
        vec![
            repository.join(".git/HEAD"),
            repository.join(".git/refs/heads/fixture")
        ]
    );
    fixture_git(
        &repository,
        &[
            "-c",
            "user.name=Fixture",
            "-c",
            "user.email=fixture@example.invalid",
            "commit",
            "--allow-empty",
            "-m",
            "second",
        ],
    )?;
    let second = build_script::build_metadata(&repository);
    assert_ne!(first.commit, second.commit);
    fixture_git(&repository, &["checkout", "--detach"])?;
    let detached = build_script::build_metadata(&repository);
    assert_eq!(detached.commit, second.commit);
    assert_eq!(detached.watched, vec![repository.join(".git/HEAD")]);
    let stamped = super::cost::build_commit();
    assert_eq!(stamped, env!("LIMINAL_BUILD_COMMIT"));
    let worktree = fixture.path().join("worktree");
    fixture_git(
        &repository,
        &[
            "worktree",
            "add",
            "--detach",
            worktree.to_str().ok_or("worktree is not UTF-8")?,
        ],
    )?;
    fs::write(worktree.join("Cargo.toml"), "[workspace]\nmembers = []\n")?;
    let linked = build_script::build_metadata(&worktree);
    assert_eq!(linked.commit, second.commit);
    assert_eq!(linked.watched.len(), 1);
    assert!(linked.watched[0].starts_with(repository.join(".git/worktrees")));
    fixture_git(
        &repository,
        &[
            "worktree",
            "remove",
            "--force",
            worktree.to_str().ok_or("worktree is not UTF-8")?,
        ],
    )?;
    Ok(())
}

fn fixture_git(directory: &Path, arguments: &[&str]) -> Result<(), Box<dyn Error>> {
    let output = std::process::Command::new("git")
        .current_dir(directory)
        .args(arguments)
        .output()?;
    if !output.status.success() {
        return Err(format!(
            "fixture git failed: {}",
            String::from_utf8_lossy(&output.stderr)
        )
        .into());
    }
    Ok(())
}

// SRV-012 R2: R3's banked proof, linked unchanged.
include!("../../../../../docs/design/evidence/SRV-011-R3-prune-bound-proof.rs");

/// The services, their connection supervisor and the self-stream sink.
type BoundedSelfStreams = (
    Arc<crate::server::connection::services::LiminalConnectionServices>,
    crate::server::connection::ConnectionSupervisor,
    Arc<super::sink::SelfStreams>,
);

/// Services with self streams bound at `events_rows` events and `cost_rows`
/// cost records, over `persistence_path` when given.
fn bounded_self_streams(
    events_rows: u64,
    cost_rows: u64,
    persistence_path: Option<&Path>,
) -> Result<BoundedSelfStreams, Box<dyn Error>> {
    use crate::config::types::SelfStreamsConfig;
    use crate::server::connection::ConnectionSupervisor;
    use crate::server::connection::services::LiminalConnectionServices;
    use crate::server::shutdown::ShutdownHandle;

    let mut config = wire_config()?;
    config.participant = Some(crate::server::participant::test_participant_config());
    config.persistence_path = persistence_path.map(Path::to_path_buf);
    config.services.self_streams = SelfStreamsConfig {
        cost_rows,
        events_rows,
    };
    let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    let sink =
        crate::server::runtime::bind_self_streams(&services, &supervisor, &config, &shutdown)?
            .ok_or("missing self-stream publisher")?;
    Ok((services, supervisor, sink))
}

fn opened_events(
    sink: &super::sink::SelfStreams,
    from: u64,
    count: u64,
) -> Result<(), Box<dyn Error>> {
    for connection in from..from + count {
        sink.event(ServerEvent::ConnectionOpened { connection });
        sink.drain()?;
    }
    Ok(())
}

fn retained_events(
    services: &crate::server::connection::services::LiminalConnectionServices,
) -> Result<Vec<liminal::durability::StoredEntry>, Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    // ast-grep-allow: no-read-from-start: the prune pins count every retained row
    let rows = block_on(
        services
            .durable_store()
            .read_from("liminal.events:0", 0, 1_000),
    )??;
    Ok(rows)
}

/// R2, keepalive-honest: below the bound the sink appends and never prunes.
/// With `events_rows` = 100, 99 events append 99 rows (the control that the
/// sink ran) while `truncate_calls` and `self_prunes` stay 0.
#[test]
fn self_stream_prune_is_idle_below_the_bound() -> Result<(), Box<dyn Error>> {
    let persistence_directory = crate::test_persistence::directory()?;
    let (services, supervisor, sink) =
        bounded_self_streams(100, 100, Some(persistence_directory.path()))?;
    let before = services.store_counts().snapshot();
    opened_events(&sink, 0, 99)?;
    let moved = services.store_counts().snapshot().since(&before);
    assert!(moved.append_calls + moved.append_all_calls >= 99);
    assert_eq!(moved.truncate_calls, 0);
    let (_, own) = sink.snapshot()?;
    assert_eq!((own.prunes, own.rows_pruned), (0, 0));
    assert_eq!(retained_events(&services)?.len(), 99);
    supervisor.shutdown();
    Ok(())
}

/// R2: a prune keeps B - B / 8 rows, so it runs once per B / 8 + 1 appends.
/// With B = 64 (keep 56): 64 events, no truncate; the 65th (sequence 64)
/// cuts below 65 - 56 = 9, pruning 9 and leaving 56 from sequence 9; eight
/// more leave 64 and no truncate; the next (sequence 73) cuts below 18.
#[test]
fn self_stream_prune_is_batched_by_an_eighth() -> Result<(), Box<dyn Error>> {
    let persistence_directory = crate::test_persistence::directory()?;
    let (services, supervisor, sink) =
        bounded_self_streams(64, 64, Some(persistence_directory.path()))?;
    let before = services.store_counts().snapshot();
    let truncates = || {
        services
            .store_counts()
            .snapshot()
            .since(&before)
            .truncate_calls
    };

    opened_events(&sink, 0, 64)?;
    assert_eq!(truncates(), 0);
    opened_events(&sink, 64, 1)?;
    assert_eq!(truncates(), 1);
    let (_, own) = sink.snapshot()?;
    assert_eq!(own.rows_pruned, 9);
    let retained = retained_events(&services)?;
    assert_eq!(retained.len(), 56);
    assert_eq!(retained.first().ok_or("no retained events")?.sequence, 9);

    opened_events(&sink, 65, 8)?;
    assert_eq!(truncates(), 1);
    assert_eq!(retained_events(&services)?.len(), 64);
    opened_events(&sink, 73, 1)?;
    assert_eq!(truncates(), 2);
    assert_eq!(retained_events(&services)?.len(), 56);
    supervisor.shutdown();
    Ok(())
}

/// R2 after a restart: the process learns nothing from the store, so the
/// first append past the bound prunes once (truncate is idempotent below the
/// head). B = 5: six rows, stop, reopen the same store, one more row.
#[test]
fn self_stream_prune_after_restart() -> Result<(), Box<dyn Error>> {
    let root = tempfile::tempdir()?;
    {
        let (_services, supervisor, sink) = bounded_self_streams(5, 5, Some(root.path()))?;
        opened_events(&sink, 0, 6)?;
        supervisor.shutdown();
    }
    let (services, supervisor, sink) = bounded_self_streams(5, 5, Some(root.path()))?;
    let before = services.store_counts().snapshot();
    opened_events(&sink, 6, 1)?;
    assert!(retained_events(&services)?.len() <= 5);
    assert_eq!(
        services
            .store_counts()
            .snapshot()
            .since(&before)
            .truncate_calls,
        1
    );
    supervisor.shutdown();
    Ok(())
}

/// R2: a store that cannot truncate. The failure is counted in the next
/// cost record's `self_prune_failures`, the event that triggered it was
/// appended, and the next append tries again.
#[test]
fn a_failed_self_stream_prune_is_counted_and_retried() -> Result<(), Box<dyn Error>> {
    use crate::config::types::SelfStreamsConfig;
    use crate::server::connection::services::LiminalConnectionServices;
    use crate::server::self_streams::sink::SelfStreams;
    use crate::server::shutdown::ShutdownHandle;

    let fixture = R3Fixture::new()?;
    let refusing = Arc::new(liminal::durability::CountingStore::new(
        CommitThenErrorStore {
            inner: Arc::clone(&fixture.store),
            fail_cost: std::sync::atomic::AtomicBool::new(false),
            commit_cost: false,
            partial_batch: false,
            fail_flush: std::sync::atomic::AtomicBool::new(false),
            pause_cost: None,
            calls: None,
            observer_boundary: None,
        },
    ));
    let services = Arc::new(LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        refusing,
    )?);
    services.install_self_stream_channels()?;
    let shutdown = ShutdownHandle::new();
    let sink = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 92)?,
        &services,
        services.signals().wake(),
        shutdown.clone(),
        SelfStreamsConfig {
            cost_rows: 1,
            events_rows: 1,
        },
    ));
    let before = services.store_counts().snapshot();
    let truncates = || {
        services
            .store_counts()
            .snapshot()
            .since(&before)
            .truncate_calls
    };

    opened_events(&sink, 0, 1)?;
    assert_eq!(truncates(), 0);
    opened_events(&sink, 1, 1)?;
    assert_eq!(truncates(), 1);
    assert_eq!(
        retained_events(&services)?.len(),
        2,
        "the triggering append stands"
    );
    assert!(
        !shutdown.is_initiated(),
        "a prune failure never fails the append"
    );

    let (counts, own) = CostSink::snapshot(&sink)?;
    let record = CostWindow::new(StoreCounts::default(), SelfWork::default())
        .prepare(counts, own)
        .ok_or("missing prune failure")?;
    let json: Value =
        serde_json::from_slice(&record.encode(&ServerMetadata::new("local".to_owned(), 1)?, 1)?)?;
    assert_eq!(json["self_prune_failures"], 1);
    assert_eq!(json["self_prunes"], 0);

    opened_events(&sink, 2, 1)?;
    assert_eq!(truncates(), 2, "the next append tries again");
    Ok(())
}

#[test]
fn counted_read_without_a_publication_arms_one_cost_tick() -> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let fixture = WireFixture::new()?;
    let before = fixture.services.store_counts().snapshot();
    assert!(!fixture.wake.armed());
    assert_eq!(fixture.wake.notifications(), 0);
    let store = fixture.services.durable_store();
    assert!(block_on(store.read_value("absent-counter-key"))??.is_none());
    assert!(block_on(store.read_value("absent-counter-key"))??.is_none());
    let delta = fixture.services.store_counts().snapshot().since(&before);
    let armed = fixture.wake.armed();
    let notifications = fixture.wake.notifications();
    fixture.supervisor.shutdown();
    assert_eq!(delta.read_value_calls, 2);
    assert_eq!(delta.append_calls, 0);
    assert_eq!(delta.append_all_calls, 0);
    assert!(armed, "counted reads did not arm the cost publisher");
    assert_eq!(notifications, 1);
    Ok(())
}

/// How long a pin waits to be told the paused append was entered. It is
/// generous, as in the churn e2e: a told wait never sleeps, and a tight
/// deadline on a loaded machine fails for the wrong reason.
const PAUSE_ENTRY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);

/// Releases the pause when dropped, so a pin that fails before its own
/// release never leaves the paused worker parked behind it.
struct ReleasedOnDrop(Arc<AppendPause>);

impl Drop for ReleasedOnDrop {
    fn drop(&mut self) {
        if let Err(error) = self.0.release() {
            tracing::error!(%error, "test append pause was not released on drop");
        }
    }
}

#[derive(Debug, Default)]
struct AppendPause {
    state: Mutex<(bool, bool)>,
    changed: std::sync::Condvar,
}

impl AppendPause {
    fn wait_for_release(&self) -> Result<(), liminal::durability::DurabilityError> {
        let mut state = self.state.lock().map_err(append_pause_error)?;
        state.0 = true;
        self.changed.notify_all();
        while !state.1 {
            state = self.changed.wait(state).map_err(append_pause_error)?;
        }
        drop(state);
        Ok(())
    }

    /// Waits for the paused append to be entered, never past
    /// [`PAUSE_ENTRY_DEADLINE`]: a pin that waits on it fails by name rather
    /// than hanging.
    fn entered(&self) -> Result<(), Box<dyn Error>> {
        let entered = self
            .changed
            .wait_timeout_while(
                self.state.lock().map_err(append_pause_error)?,
                PAUSE_ENTRY_DEADLINE,
                |state| !state.0,
            )
            .map_err(append_pause_error)?
            .0
            .0;
        if entered {
            Ok(())
        } else {
            Err(format!("the paused append was not entered within {PAUSE_ENTRY_DEADLINE:?}").into())
        }
    }

    fn release(&self) -> Result<(), Box<dyn Error>> {
        let mut state = self.state.lock().map_err(append_pause_error)?;
        state.1 = true;
        self.changed.notify_all();
        drop(state);
        Ok(())
    }
}

fn append_pause_error(error: impl std::fmt::Display) -> liminal::durability::DurabilityError {
    liminal::durability::DurabilityError::ConfigError(format!(
        "test append pause lock failed: {error}"
    ))
}

#[tokio::test]
async fn shutdown_fate_and_final_cost_rows_survive_reopen_after_the_final_flush()
-> Result<(), Box<dyn Error>> {
    let calls = Arc::new(Mutex::new(Vec::new()));
    let store_calls = Arc::clone(&calls);
    let fixture = WireFixture::with_store(move |inner| {
        Arc::new(liminal::durability::CountingStore::new(
            CommitThenErrorStore {
                inner,
                fail_cost: std::sync::atomic::AtomicBool::new(false),
                commit_cost: false,
                partial_batch: false,
                fail_flush: std::sync::atomic::AtomicBool::new(false),
                pause_cost: None,
                calls: Some(store_calls),
                observer_boundary: None,
            },
        ))
    })?;
    let peer = PressurePeer::connect(&fixture).await?;
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&fixture.wake),
        Arc::clone(&fixture.sink),
        crate::server::shutdown::ShutdownHandle::new(),
    )?;
    fixture.wake.waiting().await;
    // Every queued self row is appended before the trace is cleared, so no
    // append can lose its start to the clear and keep its completion
    // (work-list row, 7 Oct 12:15).
    fixture.sink.drain_all()?;
    calls.lock().map_err(test_sink_error)?.clear();
    let store = fixture.services.durable_store();
    assert!(store.read_value("pending-before-stop").await?.is_none());
    drop(store);
    let mut listener =
        crate::server::listener::ServerListener::bind(&wire_config()?, fixture.supervisor.clone())?;
    crate::server::shutdown::run_shutdown_sequence_before_flush(
        &mut listener,
        None,
        &fixture.supervisor,
        std::time::Duration::ZERO,
        || publisher.finish(),
    )?;
    let trace = calls.lock().map_err(test_sink_error)?.clone();
    let cost_position = assert_shutdown_flush_order(&trace)?;
    let WireFixture {
        services,
        supervisor,
        sink,
        wake,
        root,
        ..
    } = fixture;
    drop(peer);
    drop(listener);
    drop(publisher);
    drop(supervisor);
    drop(services);
    drop(sink);
    drop(wake);
    let reopened = crate::server::connection::services::open_persistent_store(root.path())?;
    let mut close_position = None;
    for (position, call) in trace.iter().enumerate() {
        let (channel, sequence) = match call {
            ShutdownStoreCall::Cost(sequence) => ("liminal.cost:0", *sequence),
            ShutdownStoreCall::Event(sequence) => ("liminal.events:0", *sequence),
            ShutdownStoreCall::AppendStarted
            | ShutdownStoreCall::AppendCompleted
            | ShutdownStoreCall::Flush => continue,
        };
        let row = reopened
            .read_at(channel, sequence)
            .await?
            .ok_or("shutdown row did not survive reopening")?;
        let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        let record: Value = serde_json::from_slice(&envelope.payload)?;
        if channel == "liminal.cost:0" {
            assert!(
                record["counts"]["read_value_calls"]
                    .as_u64()
                    .ok_or("final cost has no read count")?
                    > 0
            );
        }
        if record["details"]["kind"] == "connection_close" {
            close_position = Some(position);
        }
    }
    assert!(close_position.ok_or("shutdown emitted no connection-close fate")? < cost_position);
    drop(reopened);
    Ok(())
}

fn assert_shutdown_flush_order(trace: &[ShutdownStoreCall]) -> Result<usize, Box<dyn Error>> {
    let cost_position = trace
        .iter()
        .position(|call| matches!(call, ShutdownStoreCall::Cost(_)))
        .ok_or("shutdown emitted no final cost append")?;
    let flush_position = trace
        .iter()
        .position(|call| matches!(call, ShutdownStoreCall::Flush))
        .ok_or("shutdown issued no final flush")?;
    assert!(
        cost_position < flush_position,
        "publisher completed after the final flush: {trace:?}"
    );
    assert_eq!(
        trace
            .iter()
            .filter(|call| matches!(call, ShutdownStoreCall::Flush))
            .count(),
        1
    );
    let mut pending = 0;
    for (position, call) in trace.iter().enumerate() {
        match call {
            ShutdownStoreCall::AppendStarted => {
                assert!(
                    position < flush_position,
                    "append started after flush: {trace:?}"
                );
                pending += 1;
            }
            ShutdownStoreCall::AppendCompleted => {
                assert!(
                    position < flush_position,
                    "append completed after flush: {trace:?}"
                );
                pending -= 1;
            }
            ShutdownStoreCall::Flush => assert_eq!(pending, 0, "flush overtook append: {trace:?}"),
            ShutdownStoreCall::Cost(_) | ShutdownStoreCall::Event(_) => {}
        }
    }
    Ok(cost_position)
}

#[test]
fn shutdown_flush_uses_one_shared_store_operation() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::ConnectionServices;
    let fixture = R3Fixture::new()?;
    let before = fixture.services.store_counts().snapshot();
    fixture.services.flush_durable_state()?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(delta.flushes, 1, "shared store flushed once per channel");
    Ok(())
}

fn publish_receipt_batch_for_test(
    channel: &liminal::channel::ChannelHandle,
    payloads: &[Vec<u8>],
) -> Result<Vec<liminal::channel::ChannelDelivery>, Box<dyn Error>> {
    Ok(channel.publish_batch_with_delivery(
        payloads,
        &liminal::envelope::PublisherId::new("batch-probe"),
        None,
    )?)
}

#[test]
fn durable_receipt_batch_is_one_append_one_flush_and_ordered_delivery() -> Result<(), Box<dyn Error>>
{
    use liminal::durability::bridge::block_on;
    let fixture = R3Fixture::new()?;
    let subscriber = fixture
        .services
        .subscribe_handle_for_test("liminal.events")?;
    let channel = fixture.services.channel_handle_for_test("liminal.events")?;
    let payloads = vec![
        br#"{"generation":1}"#.to_vec(),
        br#"{"generation":2}"#.to_vec(),
        br#"{"generation":3}"#.to_vec(),
    ];
    let before = fixture.services.store_counts().snapshot();
    let deliveries = publish_receipt_batch_for_test(&channel, &payloads)?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(delta.append_calls, 0);
    assert_eq!(delta.append_all_calls, 1);
    assert_eq!(delta.append_rows, 0);
    assert_eq!(delta.append_all_rows, 3);
    assert_eq!(delta.flushes, 1);
    assert_eq!(deliveries.len(), payloads.len());
    for (index, (delivery, payload)) in deliveries.iter().zip(&payloads).enumerate() {
        let sequence = u64::try_from(index)?;
        assert_eq!(delivery.durable_position(), Some(sequence));
        let live = subscriber
            .try_next()?
            .ok_or("batch receipt was not delivered")?;
        assert_eq!(live.payload, *payload);
        let stored = block_on(fixture.store.read_at("liminal.events:0", sequence))??
            .ok_or("batch receipt was not stored")?;
        assert_eq!(
            liminal::durability::MessageEnvelope::deserialize(&stored.payload)?.payload,
            *payload
        );
    }
    assert!(subscriber.try_next()?.is_none());
    Ok(())
}

#[test]
fn committed_receipt_batch_recovers_every_row_without_reappending() -> Result<(), Box<dyn Error>> {
    receipt_batch_recovery(true)
}

#[test]
fn absent_receipt_batch_does_not_deliver_or_resubmit_its_rows() -> Result<(), Box<dyn Error>> {
    let panics = PanicCapture::install();
    let (finished, exit) = std::sync::mpsc::channel();
    receipt_batch_recovery_watched(false, Some(finished))?;
    let thread = exit.recv()?;
    assert_ne!(thread, std::thread::current().id());
    let captured = panics.snapshot()?;
    assert!(
        captured.is_empty(),
        "channel scheduler panicked: {captured:#?}"
    );
    Ok(())
}

fn receipt_batch_recovery(committed: bool) -> Result<(), Box<dyn Error>> {
    receipt_batch_recovery_watched(committed, None)
}

fn receipt_batch_recovery_watched(
    committed: bool,
    finished: Option<std::sync::mpsc::Sender<std::thread::ThreadId>>,
) -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::LiminalConnectionServices;
    let mut fixture = R3Fixture::new()?;
    let faults = Arc::new(liminal::durability::CountingStore::new(
        CommitThenErrorStore {
            inner: Arc::clone(&fixture.store),
            fail_cost: std::sync::atomic::AtomicBool::new(true),
            commit_cost: committed,
            partial_batch: false,
            fail_flush: std::sync::atomic::AtomicBool::new(false),
            pause_cost: None,
            calls: None,
            observer_boundary: None,
        },
    ));
    let services = Arc::new(LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        faults.clone(),
    )?);
    services.install_self_stream_channels()?;
    fixture.services = services;
    fixture.store = faults;
    let channel = fixture.services.channel_handle_for_test("liminal.cost")?;
    let subscriber = if let Some(finished) = finished {
        channel.subscribe_filtered(move |_| {
            CHANNEL_THREAD_EXIT.with(|slot| {
                if slot.borrow().is_none() {
                    *slot.borrow_mut() = Some(ChannelThreadExit(finished.clone()));
                }
            });
            true
        })?
    } else {
        fixture.services.subscribe_handle_for_test("liminal.cost")?
    };
    let payloads = vec![
        br#"{"generation":1}"#.to_vec(),
        br#"{"generation":2}"#.to_vec(),
        br#"{"generation":3}"#.to_vec(),
    ];
    let failed = publish_receipt_batch_for_test(&channel, &payloads);
    assert!(
        matches!(failed, Err(error) if matches!(error.downcast_ref::<liminal::LiminalError>(), Some(liminal::LiminalError::PublishUnresolved { .. })))
    );
    assert!(subscriber.try_next()?.is_none());
    let before = fixture.services.store_counts().snapshot();
    let next = channel.publish_with_delivery(
        b"{}",
        liminal::envelope::PublisherId::new("batch-probe"),
        None,
    )?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(delta.append_all_calls, 0, "recovery resubmitted the batch");
    assert_eq!(delta.append_calls, 1);
    assert_eq!(delta.flushes, u64::from(committed));
    let stored = if committed { 3 } else { 0 };
    assert_eq!(next.durable_position(), Some(stored));
    if committed {
        for payload in &payloads {
            assert_eq!(
                subscriber
                    .try_next()?
                    .ok_or("recovered batch row was not delivered")?
                    .payload,
                *payload
            );
        }
    }
    assert_eq!(
        subscriber
            .try_next()?
            .ok_or("following row was not delivered")?
            .payload,
        b"{}"
    );
    assert!(subscriber.try_next()?.is_none());
    assert_eq!(channel.durable_positions()?, vec![stored + 1]);
    Ok(())
}

#[test]
fn receipt_batch_validates_every_payload_before_any_store_operation() -> Result<(), Box<dyn Error>>
{
    let fixture = R3Fixture::new()?;
    let channel = fixture.services.channel_handle_for_test("liminal.events")?;
    let before = fixture.services.store_counts().snapshot();
    assert!(publish_receipt_batch_for_test(&channel, &[])?.is_empty());
    let payloads = vec![b"{}".to_vec(), b"invalid JSON".to_vec()];
    assert!(publish_receipt_batch_for_test(&channel, &payloads).is_err());
    assert_eq!(fixture.services.store_counts().snapshot(), before);
    Ok(())
}

#[test]
fn receipt_batch_checkpoint_crossing_reads_no_prefix_on_reconstruction()
-> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::LiminalConnectionServices;
    use liminal::durability::bridge::block_on;
    let fixture = R3Fixture::new()?;
    let channel = fixture.services.channel_handle_for_test("liminal.events")?;
    let count = liminal::durability::HEAD_CHECKPOINT_EVERY + 1;
    let payloads = vec![b"{}".to_vec(); usize::try_from(count)?];
    publish_receipt_batch_for_test(&channel, &payloads)?;
    let key = liminal::durability::head_checkpoint_key("liminal.events:0");
    assert_eq!(block_on(fixture.store.read_value(&key))??, Some(count));
    let before = fixture.services.store_counts().snapshot();
    let rebuilt = LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        Arc::clone(&fixture.store),
    )?;
    rebuilt.install_self_stream_channels()?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(
        delta.read_from_rows, 0,
        "batch reconstruction reread the committed prefix"
    );
    assert_eq!(
        rebuilt
            .channel_handle_for_test("liminal.events")?
            .durable_positions()?,
        vec![count]
    );
    Ok(())
}

#[derive(Debug)]
struct ReentryProbe {
    services: std::sync::Weak<crate::server::connection::services::LiminalConnectionServices>,
    wake: Arc<CostWake>,
    failure: std::sync::OnceLock<String>,
}

impl liminal::durability::StoreCountObserver for ReentryProbe {
    fn changed(&self) {
        liminal::durability::StoreCountObserver::changed(self.wake.as_ref());
    }

    fn flushed(&self, store: &str, batch_rows: Option<u64>) {
        let result = (|| -> Result<(), ServerError> {
            let services = self
                .services
                .upgrade()
                .ok_or_else(|| test_sink_error("flush services owner is absent"))?;
            let events = services.channel_handle_for_test("liminal.events")?;
            // Mutex formatting observes contention with try_lock before the callback enters publication.
            let state = format!("{events:?}");
            let locked = state.contains("durable: Some(Mutex { data: <locked>")
                || state.contains("durable: Some(Mutex { data: \"<locked>\"");
            assert!(!locked, "SELF_STREAM_FLUSH_REENTRY_WOULD_BLOCK");
            liminal::durability::StoreCountObserver::flushed(self.wake.as_ref(), store, batch_rows);
            Ok(())
        })();
        if let Err(error) = result {
            let retained = self.failure.set(error.to_string());
            assert!(retained.is_ok(), "flush probe failed more than once");
        }
    }
}

#[test]
fn shared_shutdown_flush_can_synchronously_publish_its_flush_event() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::ConnectionServices;
    use crate::server::self_streams::sink::SelfStreams;
    let fixture = R3Fixture::new()?;
    let wake = fixture.services.signals().wake();
    let sink = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 91)?,
        &fixture.services,
        Arc::clone(&wake),
        crate::server::shutdown::ShutdownHandle::new(),
        crate::config::types::SelfStreamsConfig::default(),
    ));
    fixture.services.signals().install(Arc::clone(&sink))?;
    wake.install_source(&sink)?;
    let counts = fixture.services.store_counts();
    let names = ["durable", "ephemeral-dedup"];
    let probe = Arc::new(ReentryProbe {
        services: Arc::downgrade(&fixture.services),
        wake,
        failure: std::sync::OnceLock::new(),
    });
    counts.install_observer(&names[..counts.store_count()], probe.clone())?;
    let before = counts.snapshot();
    if std::env::var_os("SELF_STREAM_REENTRY_RED_CONTROL").is_some() {
        fixture
            .services
            .channel_handle_for_test("liminal.events")?
            .flush()?;
    } else {
        fixture.services.flush_durable_state()?;
    }
    let delta = counts.snapshot().since(&before);
    assert_eq!(delta.flushes, 1);
    assert_eq!(delta.append_rows, 1);
    assert_eq!(CostSink::snapshot(&sink)?.1.rows, 1);
    assert!(
        probe.failure.get().is_none(),
        "flush probe failed: {:?}",
        probe.failure
    );
    Ok(())
}

#[test]
fn external_read_during_self_append_remains_in_the_next_acknowledged_window()
-> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let pause = Arc::new(AppendPause::default());
    let store_pause = Arc::clone(&pause);
    let fixture = WireFixture::with_store(move |store| {
        Arc::new(CommitThenErrorStore {
            inner: store,
            fail_cost: std::sync::atomic::AtomicBool::new(false),
            commit_cost: false,
            partial_batch: false,
            fail_flush: std::sync::atomic::AtomicBool::new(false),
            pause_cost: Some(store_pause),
            calls: None,
            observer_boundary: None,
        })
    })?;
    let mut window = CostWindow::new(StoreCounts::default(), SelfWork::default());
    let before = fixture.services.store_counts().snapshot();
    let record = window
        .prepare(before, SelfWork::default())
        .ok_or("fixture has no counted startup work")?;
    let mut sink = Arc::clone(&fixture.sink);
    let task = std::thread::spawn(move || {
        CostSink::publish(&mut sink, &record)?;
        Ok::<_, ServerError>(record)
    });
    pause.entered()?;
    let store = fixture.services.durable_store();
    let read = block_on(store.read_at("external-missing:0", 91));
    let released = pause.release();
    let published = task.join().map_err(|_| "cost append worker panicked")?;
    released?;
    assert!(read??.is_none());
    let record = published?;
    window.published(&record);
    let (after, own) = CostSink::snapshot(&fixture.sink)?;
    let next = window
        .prepare(after, own)
        .ok_or("concurrent work disappeared")?;
    assert_eq!(next.delta().read_at_calls, 1);
    assert_eq!(next.delta().read_at_rows, 0);
    assert_eq!(next.delta().append_rows, 1);
    assert_eq!(own.rows, 1);
    let mut sink = Arc::clone(&fixture.sink);
    CostSink::publish(&mut sink, &next)?;
    window.published(&next);
    let stored = block_on(store.read_at("liminal.cost:0", 1))??
        .ok_or("next cost window was not appended")?;
    let envelope = liminal::durability::MessageEnvelope::deserialize(&stored.payload)?;
    let acknowledged: Value = serde_json::from_slice(&envelope.payload)?;
    assert_eq!(acknowledged["counts"]["read_at_calls"], 1);
    assert_eq!(
        acknowledged["previous_snapshot"]["read_at_calls"],
        before.read_at_calls
    );
    assert!(window.prepare(after, own).is_none());
    fixture.supervisor.shutdown();
    Ok(())
}

type TickFuture<'a> =
    std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), ServerError>> + 'a>>;

/// Polls `tick` once with a waker that does nothing: whether the tick is
/// due now, answered without a runtime and without waiting. A tick that has
/// fired is spent, and polling it again answers that by name.
fn tick_now(tick: &mut Option<TickFuture<'_>>) -> std::task::Poll<Result<(), ServerError>> {
    let Some(future) = tick.as_mut() else {
        return std::task::Poll::Ready(Err(super::sink::SelfStreams::failure(
            "the cost tick had already fired",
        )));
    };
    let mut context = std::task::Context::from_waker(std::task::Waker::noop());
    let polled = std::future::Future::poll(future.as_mut(), &mut context);
    if polled.is_ready() {
        *tick = None;
    }
    polled
}

/// How many due ticks it takes to spend the work behind an idle wake.
/// One due tick spends all of it, since the tick swaps the external flag
/// and `Notify` keeps at most one permit. With the drainer idle, nothing
/// adds more, so the second tick waits.
const TICKS_TO_SPEND_SETUP: usize = 2;

/// A real cost tick with a timer that is due at once, polled until the
/// work already behind it (fixture setup, earlier rows) is spent and it
/// waits. The drainer is drained first, so no self row is still on its way.
fn idle_tick<'a>(
    wake: &'a CostWake,
    sink: &super::sink::SelfStreams,
) -> Result<Option<TickFuture<'a>>, Box<dyn Error>> {
    sink.drain_all()?;
    for _ in 0..TICKS_TO_SPEND_SETUP {
        let mut tick: Option<TickFuture<'_>> =
            Some(Box::pin(wake.next_tick(|_| std::future::ready(()))));
        match tick_now(&mut tick) {
            std::task::Poll::Pending => return Ok(tick),
            std::task::Poll::Ready(result) => result?,
        }
    }
    Err(format!("the cost tick never settled idle after {TICKS_TO_SPEND_SETUP} due ticks").into())
}

/// R3b F2: an external read that lands before a self append is still
/// external work. The cost row appended after it must not take the read
/// into a baseline and swallow the tick; a cost row alone (the control)
/// must not wake it. (A server event is not the control: it arms the tick
/// on purpose, for the connection that caused it.)
#[test]
fn a_self_append_after_external_work_keeps_its_cost_tick() -> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let fixture = WireFixture::new()?;
    let row = || {
        CostWindow::new(StoreCounts::default(), SelfWork::default())
            .prepare(
                fixture.services.store_counts().snapshot(),
                SelfWork::default(),
            )
            .ok_or("fixture has no counted startup work")
    };
    let mut publisher = Arc::clone(&fixture.sink);
    let mut tick = idle_tick(&fixture.wake, &fixture.sink)?;
    CostSink::publish(&mut publisher, &row()?)?;
    let alone = tick_now(&mut tick);
    let store = fixture.services.durable_store();
    let read = block_on(store.read_at("external-missing:0", 91));
    CostSink::publish(&mut publisher, &row()?)?;
    let after_external = tick_now(&mut tick);
    drop(tick);
    fixture.supervisor.shutdown();
    assert!(read??.is_none());
    assert!(
        alone.is_pending(),
        "a self append alone woke the cost tick: {alone:?}"
    );
    assert!(
        matches!(after_external, std::task::Poll::Ready(Ok(()))),
        "a self append after an external read swallowed its cost tick: {after_external:?}"
    );
    Ok(())
}

/// R3b, Hermes's third condition: store work made on another thread while
/// the drainer is inside its own append is external and wakes the tick. A
/// self-work mark held process-wide, rather than by the drainer's thread,
/// fails this.
#[test]
fn external_work_while_the_drainer_appends_wakes_the_cost_tick() -> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let pause = Arc::new(AppendPause::default());
    let store_pause = Arc::clone(&pause);
    let fixture = WireFixture::with_store(move |store| {
        Arc::new(CommitThenErrorStore {
            inner: store,
            fail_cost: std::sync::atomic::AtomicBool::new(false),
            commit_cost: false,
            partial_batch: false,
            fail_flush: std::sync::atomic::AtomicBool::new(false),
            pause_cost: Some(store_pause),
            calls: None,
            observer_boundary: None,
        })
    })?;
    let mut tick = idle_tick(&fixture.wake, &fixture.sink)?;
    let record = CostWindow::new(StoreCounts::default(), SelfWork::default())
        .prepare(
            fixture.services.store_counts().snapshot(),
            SelfWork::default(),
        )
        .ok_or("fixture has no counted startup work")?;
    let mut sink = Arc::clone(&fixture.sink);
    let task = std::thread::spawn(move || CostSink::publish(&mut sink, &record));
    let _released = ReleasedOnDrop(Arc::clone(&pause));
    pause.entered()?;
    let store = fixture.services.durable_store();
    let read = block_on(store.read_at("external-missing:0", 91));
    let released = pause.release();
    let published = task.join().map_err(|_| "cost append worker panicked")?;
    let after = tick_now(&mut tick);
    drop(tick);
    fixture.supervisor.shutdown();
    released?;
    published?;
    assert!(read??.is_none());
    assert!(
        matches!(after, std::task::Poll::Ready(Ok(()))),
        "an external read made while the drainer appended was taken for self work: {after:?}"
    );
    Ok(())
}

#[path = "tests/self_prune.rs"]
mod self_prune;

#[test]
fn counted_flush_emits_one_named_event_without_invented_batch_rows() -> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let fixture = WireFixture::new()?;
    let counts = fixture.services.store_counts();
    let before = counts.snapshot();
    let store = fixture.services.durable_store();
    block_on(store.flush())??;
    fixture.sink.drain()?;
    let after = counts.snapshot();
    assert_eq!(after.since(&before).flushes, 1);
    assert_eq!(after.since(&before).append_rows, 1);
    let row =
        block_on(store.read_at("liminal.events:0", 0))??.ok_or("counted flush emitted no event")?;
    let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
    let event: Value = serde_json::from_slice(&envelope.payload)?;
    assert_eq!(event["v"], 1);
    assert_eq!(event["details"]["kind"], "flush");
    assert_eq!(event["details"]["store"], "durable");
    assert!(event["details"].get("batch_rows").is_none());
    assert!(block_on(store.read_at("liminal.events:0", 1))??.is_none());
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn self_records_generate_at_most_one_observer_wake_per_record() -> Result<(), Box<dyn Error>> {
    let fixture = WireFixture::new()?;
    let before = fixture.services.store_counts().snapshot();
    let notifications = fixture.wake.notifications();
    let records = 16;
    for connection in 0..records {
        fixture
            .sink
            .event(ServerEvent::ConnectionOpened { connection });
    }
    let mut sink = Arc::clone(&fixture.sink);
    CostSink::before_stop(&mut sink)?;
    let (after, own) = CostSink::snapshot(&sink)?;
    let delta = after.since(&before);
    assert_eq!(
        delta.append_rows + delta.append_all_rows,
        records + delta.flushes
    );
    assert_eq!(own.rows, records + delta.flushes);
    assert_eq!(own.flushes, delta.flushes);
    assert!(fixture.wake.notifications() - notifications <= own.rows);
    fixture.supervisor.shutdown();
    Ok(())
}

type KeepalivePeer = (
    crate::server::connection::WebSocketListener,
    tungstenite::WebSocket<std::net::TcpStream>,
    u64,
);

fn self_stream_keepalive_peer(fixture: &WireFixture) -> Result<KeepalivePeer, Box<dyn Error>> {
    use liminal::protocol::{Frame, ProtocolVersion};
    let listener = crate::server::connection::WebSocketListener::bind(
        &crate::config::types::WebSocketConfig {
            listen_address: "127.0.0.1:0".parse()?,
            path: "/liminal".to_owned(),
            allowed_origins: Vec::new(),
            ping_interval_ms: Some(10),
        },
        fixture.supervisor.clone(),
    )?;
    let before = fixture.supervisor.active_connection_pids();
    let stream = std::net::TcpStream::connect(listener.local_addr())?;
    let (mut socket, _) =
        tungstenite::client::client(format!("ws://{}/liminal", listener.local_addr()), stream)?;
    let connect = Frame::Connect {
        flags: 0,
        min_version: ProtocolVersion::new(1, 0),
        max_version: ProtocolVersion::new(1, 0),
        auth_token: Vec::new(),
    };
    let mut bytes = vec![0; liminal::protocol::encoded_len(&connect)?];
    let length = liminal::protocol::encode(&connect, &mut bytes)?;
    bytes.truncate(length);
    socket.send(tungstenite::Message::Binary(bytes.into()))?;
    let tungstenite::Message::Binary(bytes) = socket.read()? else {
        return Err("keepalive peer received no connection acknowledgement".into());
    };
    let (ack, consumed) = liminal::protocol::decode(&bytes)?;
    assert_eq!(consumed, bytes.len());
    assert!(matches!(ack, Frame::ConnectAck { .. }));
    let pid = fixture
        .supervisor
        .active_connection_pids()
        .into_iter()
        .find(|pid| !before.contains(pid))
        .ok_or("keepalive peer was not registered")?;
    Ok((listener, socket, pid))
}

#[tokio::test]
async fn real_keepalive_slices_grow_without_cost_wakes_or_publications()
-> Result<(), Box<dyn Error>> {
    let fixture = WireFixture::new()?;
    let (listener, mut socket, pid) = self_stream_keepalive_peer(&fixture)?;
    fixture.sink.event_completed().await;
    publish_initial_cost_window_for_test(&fixture).await?;
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&fixture.wake),
        Arc::clone(&fixture.sink),
        crate::server::shutdown::ShutdownHandle::new(),
    )?;
    fixture.wake.waiting().await;
    let counts_before = fixture.services.store_counts().snapshot();
    let wakes_before = fixture.wake.notifications();
    let slices_before = fixture.supervisor.slice_count(pid);
    let mut pings = 0;
    for _ in 0..3 {
        assert!(matches!(socket.read()?, tungstenite::Message::Ping(_)));
        let serviced = fixture.supervisor.observe_next_slice(pid);
        socket.flush()?;
        serviced.recv()?;
        pings += 1;
    }
    let slices_after = fixture.supervisor.slice_count(pid);
    let wakes_after = fixture.wake.notifications();
    let counts_after = fixture.services.store_counts().snapshot();
    let timer_pending = fixture.wake.timer_pending();
    publisher.finish()?;
    socket.close(None)?;
    listener.shutdown()?;
    fixture.supervisor.shutdown();
    assert_eq!(pings, 3);
    assert!(slices_after > slices_before);
    assert_eq!(wakes_after, wakes_before);
    assert_eq!(counts_after, counts_before);
    assert!(!timer_pending);
    Ok(())
}

#[derive(Debug)]
struct ObserverBoundary {
    caller: std::thread::ThreadId,
    inline_appends: AtomicU64,
    appended: std::sync::mpsc::SyncSender<()>,
}

#[test]
fn flushed_observer_hands_off_without_appending_on_its_caller() -> Result<(), Box<dyn Error>> {
    use super::sink::SelfStreams;
    use crate::server::connection::services::LiminalConnectionServices;
    use liminal::durability::{StoreCountObserver, bridge::block_on};
    let fixture = R3Fixture::new()?;
    let (appended, completed) = std::sync::mpsc::sync_channel(1);
    let boundary = Arc::new(ObserverBoundary {
        caller: std::thread::current().id(),
        inline_appends: AtomicU64::new(0),
        appended,
    });
    let store = Arc::new(CommitThenErrorStore {
        inner: Arc::clone(&fixture.store),
        fail_cost: std::sync::atomic::AtomicBool::new(false),
        commit_cost: false,
        partial_batch: false,
        fail_flush: std::sync::atomic::AtomicBool::new(false),
        pause_cost: None,
        calls: None,
        observer_boundary: Some(Arc::clone(&boundary)),
    });
    let services = Arc::new(LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        store,
    )?);
    services.install_self_stream_channels()?;
    let wake = services.signals().wake();
    let streams = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 1)?,
        &services,
        Arc::clone(&wake),
        crate::server::shutdown::ShutdownHandle::new(),
        crate::config::types::SelfStreamsConfig::default(),
    ));
    wake.install_source(&streams)?;
    wake.flushed("durable", None);
    assert_eq!(
        boundary.inline_appends.load(Ordering::Relaxed),
        0,
        "SELF_STREAM_OBSERVER_INLINE_APPEND: observer attempted durable IO on its caller"
    );
    completed.recv()?;
    let row = block_on(fixture.store.read_at("liminal.events:0", 0))??
        .ok_or("dedicated drainer did not append the flush event")?;
    let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
    let record: Value = serde_json::from_slice(&envelope.payload)?;
    assert_eq!(record["details"]["kind"], "flush");
    assert_eq!(record["details"]["store"], "durable");
    Ok(())
}

struct AppendRelease(Arc<AppendPause>);

impl Drop for AppendRelease {
    fn drop(&mut self) {
        let mut state = match self.0.state.lock() {
            Ok(state) => state,
            Err(error) => {
                tracing::error!(%error, "append release cleanup recovered a poisoned lock");
                error.into_inner()
            }
        };
        state.1 = true;
        self.0.changed.notify_all();
        drop(state);
    }
}

#[test]
fn self_record_queue_stays_bounded_and_acknowledges_only_after_append() -> Result<(), Box<dyn Error>>
{
    use super::sink::SelfStreams;
    use crate::server::connection::services::LiminalConnectionServices;
    crate::metrics::init();
    let failures = pressure_metric("liminal_self_stream_append_failures_total")?;
    let fixture = R3Fixture::new()?;
    let pause = Arc::new(AppendPause::default());
    let release = AppendRelease(Arc::clone(&pause));
    let calls = Arc::new(Mutex::new(Vec::new()));
    let store = Arc::new(CommitThenErrorStore {
        inner: Arc::clone(&fixture.store),
        fail_cost: std::sync::atomic::AtomicBool::new(false),
        commit_cost: false,
        partial_batch: false,
        fail_flush: std::sync::atomic::AtomicBool::new(false),
        pause_cost: Some(Arc::clone(&pause)),
        calls: Some(Arc::clone(&calls)),
        observer_boundary: None,
    });
    let services = Arc::new(LiminalConnectionServices::from_config_with_store(
        &wire_config()?,
        store,
    )?);
    services.install_self_stream_channels()?;
    let shutdown = crate::server::shutdown::ShutdownHandle::new();
    let sink = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 1)?,
        &services,
        services.signals().wake(),
        shutdown.clone(),
        crate::config::types::SelfStreamsConfig::default(),
    ));
    let record = CostWindow::new(StoreCounts::default(), SelfWork::default())
        .prepare(services.store_counts().snapshot(), SelfWork::default())
        .ok_or("fixture produced no startup counts")?;
    let mut writer = Arc::clone(&sink);
    let publisher = std::thread::spawn(move || CostSink::publish(&mut writer, &record));
    pause.entered()?;
    assert_eq!(sink.queue_measurements()?, (0, 1024, None));
    assert_eq!(sink.own_work_for_test()?.rows, 0);
    for connection in 0..1024 {
        sink.event(ServerEvent::ConnectionOpened { connection });
    }
    assert_eq!(sink.queue_measurements()?, (1024, 1024, None));
    sink.event(ServerEvent::ConnectionOpened { connection: 1024 });
    assert_eq!(sink.queue_measurements()?, (1024, 1024, None));
    assert_eq!(
        pressure_metric("liminal_self_stream_append_failures_total")? - failures,
        1
    );
    drop(release);
    let result = publisher.join().map_err(|_| "cost publisher panicked")?;
    assert!(
        matches!(result, Err(ServerError::ListenerAccept { message })
        if message.contains("SELF_STREAM_QUEUE_OVERFLOW"))
    );
    services.signals().install(Arc::clone(&sink))?;
    let flushed = crate::server::connection::services::ConnectionServices::flush_durable_state(
        services.as_ref(),
    );
    assert!(matches!(flushed, Err(ServerError::ListenerAccept { .. })));
    assert_eq!(
        calls
            .lock()
            .map_err(test_sink_error)?
            .iter()
            .filter(|call| matches!(call, ShutdownStoreCall::Flush))
            .count(),
        1,
        "fatal record queue prevented the final backing-store flush"
    );
    assert_shutdown_flush_order(&calls.lock().map_err(test_sink_error)?.clone())?;
    shutdown.wait();
    drop(sink);
    Ok(())
}

#[test]
fn self_record_handoff_has_no_allocations() -> Result<(), Box<dyn Error>> {
    use super::queue::{QueuedRecord, RecordPayload, RecordQueue};
    let queue = RecordQueue::new(crate::server::shutdown::ShutdownHandle::new());
    let records: Vec<_> = (0..1024)
        .map(|sequence| QueuedRecord {
            sequence,
            channel: "liminal.events",
            payload: RecordPayload::Single(b"{}".to_vec()),
            connection_opened_count: 0,
        })
        .collect();
    let mut result = Ok(());
    let allocations = allocation_counter::measure(|| {
        result = records
            .into_iter()
            .try_for_each(|record| queue.enqueue(record).map(|_| ()));
    });
    result?;
    assert_eq!(allocations.count_total, 0, "record handoff allocated");
    assert_eq!(allocations.bytes_total, 0, "record handoff allocated bytes");
    assert_eq!(queue.measurements()?, (1024, 1024, None));
    let metadata = ServerMetadata::new("local".to_owned(), 1)?;
    let event = ServerEvent::ConnectionOpened { connection: 1 };
    let mut encoded = Ok(Vec::new());
    let encoding_allocations = allocation_counter::measure(|| {
        encoded = event.encode(&metadata, 1);
    });
    encoded?;
    eprintln!(
        "ENCODER_ALLOCATION_COUNT={} ENCODER_ALLOCATION_BYTES={}",
        encoding_allocations.count_total, encoding_allocations.bytes_total
    );
    Ok(())
}

#[test]
fn public_durable_publish_keeps_its_store_cost() -> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    use liminal::durability::{CountingStore, DurableChannel, MessageEnvelope};
    let store = Arc::new(CountingStore::new(ReceiptMemoryStore::default()));
    let mut channel = DurableChannel::new("receipt.cost", 1, store.clone())?;
    let envelope = MessageEnvelope {
        payload: b"{}".to_vec(),
        causal_context: None,
        timestamp: 0,
        publisher_id: "cost-probe".to_owned(),
        idempotency_key: Some("cost-key".to_owned()),
    };
    let before = store.snapshot();
    for position in 0..100 {
        assert_eq!(block_on(channel.publish(&envelope))??, position);
    }
    let counts = store.snapshot().since(&before);
    eprintln!("PUBLIC_PUBLISH_STORE_COUNTS counts={counts:?}");
    // 5800 = 100 rows of 58, derived from `MessageEnvelope::serialize`
    // (liminal durability/channel/codec.rs): magic 4 + payload `{}` (8 + 2) +
    // causal ABSENT 1 + timestamp 8 + "cost-probe" (8 + 10) + key PRESENT 1 +
    // "cost-key" (8 + 8).
    assert_eq!(
        counts,
        StoreCounts {
            append_calls: 100,
            append_rows: 100,
            append_bytes: 5800,
            ..StoreCounts::default()
        }
    );
    Ok(())
}

#[test]
fn receipt_batch_flush_observer_can_enqueue_while_its_channel_lock_is_held()
-> Result<(), Box<dyn Error>> {
    use liminal::durability::bridge::block_on;
    let fixture = WireFixture::new()?;
    let channel = fixture.services.channel_handle_for_test("liminal.events")?;
    let payloads = vec![
        br#"{"generation":1}"#.to_vec(),
        br#"{"generation":2}"#.to_vec(),
    ];
    let before = fixture.services.store_counts().snapshot();
    let delivered = publish_receipt_batch_for_test(&channel, &payloads)?;
    fixture.sink.drain()?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(delivered.len(), 2);
    assert_eq!(delta.append_all_calls, 1);
    assert_eq!(delta.append_all_rows, 2);
    assert_eq!(delta.append_rows, 1);
    assert_eq!(delta.flushes, 1);
    assert_eq!(fixture.sink.own_work_for_test()?.rows, 1);
    let store = fixture.services.durable_store();
    for (sequence, expected) in payloads.iter().enumerate() {
        let row = block_on(store.read_at("liminal.events:0", u64::try_from(sequence)?))??
            .ok_or("batch receipt is absent")?;
        assert_eq!(
            liminal::durability::MessageEnvelope::deserialize(&row.payload)?.payload,
            *expected
        );
    }
    let row = block_on(store.read_at("liminal.events:0", 2))??
        .ok_or("batch flush observer was not drained")?;
    let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
    let event: Value = serde_json::from_slice(&envelope.payload)?;
    assert_eq!(event["details"]["kind"], "flush");
    assert!(block_on(store.read_at("liminal.events:0", 3))??.is_none());
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn ordinary_shutdown_receives_one_completion_signal_and_no_completion_failure()
-> Result<(), Box<dyn Error>> {
    let fixture = WireFixture::new()?;
    let shutdown = crate::server::shutdown::ShutdownHandle::new();
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&fixture.wake),
        Arc::clone(&fixture.sink),
        shutdown.clone(),
    )?;
    publisher.finish()?;
    assert_eq!(publisher.completion_counts(), (1, 0, 0));
    publisher.finish()?;
    assert_eq!(publisher.completion_counts(), (1, 0, 0));
    assert!(!shutdown.is_initiated());
    let (pending, _, acknowledged) = fixture.sink.queue_measurements()?;
    assert_eq!(pending, 0);
    assert!(acknowledged.is_some());
    fixture.supervisor.shutdown();
    Ok(())
}

enum OwnerPanicPayload {
    OwnedText,
    BorrowedText,
    Opaque,
}

struct PanickingOwnerSink {
    payload: OwnerPanicPayload,
    drops: Arc<AtomicU64>,
    snapshots: Arc<AtomicU64>,
    publications: Arc<AtomicU64>,
}

impl CostSink for PanickingOwnerSink {
    fn before_stop(&mut self) -> Result<(), ServerError> {
        let payload: Box<dyn std::any::Any + Send> = match self.payload {
            OwnerPanicPayload::OwnedText => Box::new("seeded owner panic".to_owned()),
            OwnerPanicPayload::BorrowedText => Box::new("seeded owner panic"),
            OwnerPanicPayload::Opaque => Box::new(7_u64),
        };
        std::panic::resume_unwind(payload)
    }

    fn snapshot(&self) -> Result<(StoreCounts, SelfWork), ServerError> {
        self.snapshots.fetch_add(1, Ordering::Relaxed);
        Ok((StoreCounts::default(), SelfWork::default()))
    }

    fn publish(&mut self, record: &super::cost::PreparedCost) -> Result<(), ServerError> {
        self.publications.fetch_add(1, Ordering::Relaxed);
        Err(test_sink_error(format!(
            "owner panic unexpectedly published {record:?}"
        )))
    }
}

impl Drop for PanickingOwnerSink {
    fn drop(&mut self) {
        self.drops.fetch_add(1, Ordering::Release);
    }
}

#[test]
fn owner_panic_signals_fatal_before_join_and_completes_after_owned_sink_drop()
-> Result<(), Box<dyn Error>> {
    for (payload, message) in [
        (OwnerPanicPayload::OwnedText, "seeded owner panic"),
        (OwnerPanicPayload::BorrowedText, "seeded owner panic"),
        (OwnerPanicPayload::Opaque, "non-string panic"),
    ] {
        let drops = Arc::new(AtomicU64::new(0));
        let snapshots = Arc::new(AtomicU64::new(0));
        let publications = Arc::new(AtomicU64::new(0));
        let shutdown = crate::server::shutdown::ShutdownHandle::new();
        let mut publisher = super::owner::PublisherRuntime::start(
            Arc::new(CostWake::default()),
            PanickingOwnerSink {
                payload,
                drops: Arc::clone(&drops),
                snapshots: Arc::clone(&snapshots),
                publications: Arc::clone(&publications),
            },
            shutdown.clone(),
        )?;
        publisher.request_stop_for_test();
        shutdown.wait();
        assert!(shutdown.is_initiated());
        let error = publisher
            .finish()
            .err()
            .ok_or("owner panic was reported as a successful join")?;
        assert_eq!(
            error.to_string(),
            format!("listener accept failed: self-stream join panic failed: {message}")
        );
        assert_eq!(publisher.completion_counts(), (1, 1, 0));
        assert_eq!(drops.load(Ordering::Acquire), 1);
        assert_eq!(snapshots.load(Ordering::Acquire), 0);
        assert_eq!(publications.load(Ordering::Acquire), 0);
        let repeated = publisher
            .finish()
            .err()
            .ok_or("repeated finish swallowed the owner panic")?;
        assert_eq!(repeated.to_string(), error.to_string());
        assert_eq!(publisher.completion_counts(), (1, 1, 0));
    }
    Ok(())
}

struct FailingOwnerSink {
    error: Option<ServerError>,
    drops: Arc<AtomicU64>,
}

impl CostSink for FailingOwnerSink {
    fn before_stop(&mut self) -> Result<(), ServerError> {
        Err(self
            .error
            .take()
            .ok_or_else(|| test_sink_error("owner failure was requested twice"))?)
    }

    fn snapshot(&self) -> Result<(StoreCounts, SelfWork), ServerError> {
        Err(test_sink_error(
            "failing owner unexpectedly took a snapshot",
        ))
    }

    fn publish(&mut self, record: &super::cost::PreparedCost) -> Result<(), ServerError> {
        Err(test_sink_error(format!(
            "failing owner unexpectedly published {record:?}"
        )))
    }
}

impl Drop for FailingOwnerSink {
    fn drop(&mut self) {
        self.drops.fetch_add(1, Ordering::Release);
    }
}

fn prove_owner_error_repeat(error: ServerError) -> Result<(), Box<dyn Error>> {
    let original_variant = std::mem::discriminant(&error);
    let original_display = error.to_string();
    let drops = Arc::new(AtomicU64::new(0));
    let shutdown = crate::server::shutdown::ShutdownHandle::new();
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::new(CostWake::default()),
        FailingOwnerSink {
            error: Some(error),
            drops: Arc::clone(&drops),
        },
        shutdown.clone(),
    )?;
    publisher.request_stop_for_test();
    shutdown.wait();
    assert!(shutdown.is_initiated());
    let first = publisher
        .finish()
        .err()
        .ok_or("first finish swallowed the owner error")?;
    assert_eq!(std::mem::discriminant(&first), original_variant);
    assert_eq!(first.to_string(), original_display);
    assert_eq!(drops.load(Ordering::Acquire), 1);
    let repeated = publisher
        .finish()
        .err()
        .ok_or("repeated finish swallowed the owner error")?;
    assert!(matches!(
        &repeated,
        ServerError::SelfStreamPublisherAlreadyFinished { first_error }
            if first_error == &original_display
    ));
    assert_eq!(
        repeated.to_string(),
        format!("self-stream publisher already finished with an error: {original_display}")
    );
    assert_eq!(publisher.completion_counts(), (1, 1, 0));
    let third = publisher
        .finish()
        .err()
        .ok_or("third finish swallowed the owner error")?;
    assert_eq!(third.to_string(), repeated.to_string());
    assert_eq!(publisher.completion_counts(), (1, 1, 0));
    assert_eq!(drops.load(Ordering::Acquire), 1);
    Ok(())
}

#[test]
fn owner_listener_error_keeps_first_variant_and_names_repeat() -> Result<(), Box<dyn Error>> {
    prove_owner_error_repeat(ServerError::ListenerAccept {
        message: "seeded owner failure".to_owned(),
    })
}

#[test]
fn owner_config_error_keeps_first_variant_and_names_repeat() -> Result<(), Box<dyn Error>> {
    prove_owner_error_repeat(ServerError::ConfigValidation {
        message: "seeded owner failure".to_owned(),
    })
}

#[test]
fn stalled_store_hits_one_owned_bound_and_shutdown_is_named_unclean() -> Result<(), Box<dyn Error>>
{
    use crate::server::shutdown::{ShutdownHandle, run_shutdown_sequence_before_flush};
    let pause = Arc::new(AppendPause::default());
    let store_pause = Arc::clone(&pause);
    let calls = Arc::new(Mutex::new(Vec::new()));
    let store_calls = Arc::clone(&calls);
    let fixture = WireFixture::with_store(move |inner| {
        Arc::new(CommitThenErrorStore {
            inner,
            fail_cost: std::sync::atomic::AtomicBool::new(true),
            commit_cost: false,
            partial_batch: false,
            fail_flush: std::sync::atomic::AtomicBool::new(false),
            pause_cost: Some(store_pause),
            calls: Some(store_calls),
            observer_boundary: None,
        })
    })?;
    calls.lock().map_err(test_sink_error)?.clear();
    let shutdown = ShutdownHandle::new();
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&fixture.wake),
        Arc::clone(&fixture.sink),
        shutdown.clone(),
    )?;
    let release = AppendRelease(Arc::clone(&pause));
    publisher.request_stop_for_test();
    pause.entered()?;
    let error = publisher
        .finish_with_bound(std::time::Duration::ZERO)
        .err()
        .ok_or("stalled store unexpectedly acknowledged its final cost")?;
    assert!(
        error
            .to_string()
            .contains("self-stream owned durable drain bound failed")
    );
    assert_eq!(publisher.completion_counts(), (0, 1, 1));
    assert!(shutdown.is_initiated());
    assert!(
        fixture
            .sink
            .enqueue_event_for_test(ServerEvent::ConnectionOpened { connection: 1 })
            .is_err(),
        "owned drain abandonment still admits new durable records"
    );
    let log = HeadLog(Arc::new(Mutex::new(Vec::new())));
    let writer = log.clone();
    let subscriber = tracing_subscriber::fmt()
        .without_time()
        .with_ansi(false)
        .with_max_level(tracing::Level::INFO)
        .with_writer(move || writer.clone())
        .finish();
    let mut listener =
        crate::server::listener::ServerListener::bind(&wire_config()?, fixture.supervisor.clone())?;
    let result = tracing::subscriber::with_default(subscriber, || {
        run_shutdown_sequence_before_flush(
            &mut listener,
            None,
            &fixture.supervisor,
            std::time::Duration::ZERO,
            || publisher.finish_with_bound(std::time::Duration::ZERO),
        )
    });
    let failure = result.err().ok_or("stalled shutdown was reported clean")?;
    assert!(
        failure
            .to_string()
            .contains("self-stream owned durable drain bound failed")
    );
    assert_eq!(publisher.completion_counts(), (0, 1, 1));
    let output = String::from_utf8(log.0.lock().map_err(test_sink_error)?.clone())?;
    assert!(output.lines().any(|line| line.contains("shutdown sequence complete") && line.contains("clean=false")));
    assert!(output.contains("self-stream owned durable drain bound failed"));
    let trace = calls.lock().map_err(test_sink_error)?.clone();
    assert_eq!(
        trace
            .iter()
            .filter(|call| matches!(call, ShutdownStoreCall::AppendStarted))
            .count(),
        1
    );
    assert_eq!(
        trace
            .iter()
            .filter(|call| matches!(
                call,
                ShutdownStoreCall::AppendCompleted
                    | ShutdownStoreCall::Cost(_)
                    | ShutdownStoreCall::Flush
            ))
            .count(),
        0
    );
    drop(release);
    assert!(publisher.join_completed_for_test().is_err());
    fixture.sink.join_drainer_for_test()?;
    Ok(())
}

#[test]
fn single_event_backlog_is_one_batch_and_preserves_the_cost_boundary() -> Result<(), Box<dyn Error>>
{
    use super::queue::{QueuedRecord, RecordPayload, RecordQueue};
    let queue = RecordQueue::new(crate::server::shutdown::ShutdownHandle::new());
    for value in 0..4 {
        queue.enqueue(QueuedRecord {
            sequence: 0,
            channel: "liminal.events",
            payload: RecordPayload::Single(vec![value]),
            connection_opened_count: 0,
        })?;
    }
    queue.enqueue(QueuedRecord {
        sequence: 0,
        channel: "liminal.cost",
        payload: RecordPayload::Single(vec![9]),
        connection_opened_count: 0,
    })?;
    let batch = queue.next()?.ok_or("event backlog disappeared")?;
    let RecordPayload::Batch(payloads) = &batch.payload else {
        return Err("event backlog still requires separate durable operations".into());
    };
    assert_eq!(payloads, &vec![vec![0], vec![1], vec![2], vec![3]]);
    assert_eq!(batch.sequence, 3);
    assert_eq!(batch.channel, "liminal.events");
    assert_eq!(queue.measurements()?.0, 1);
    queue.appended(&batch, 4)?;
    let cost = queue.next()?.ok_or("cost boundary disappeared")?;
    assert_eq!(cost.sequence, 4);
    assert_eq!(cost.channel, "liminal.cost");
    assert!(matches!(cost.payload, RecordPayload::Single(_)));
    queue.appended(&cost, 1)?;
    assert_eq!(queue.own_work()?.rows, 5);
    assert_eq!(queue.own_work()?.flushes, 1);
    Ok(())
}

fn head_proof_stage(started: &std::time::Instant, stage: &str) {
    eprintln!(
        "HEAD_PROOF_STAGE stage={stage} elapsed_ns={}",
        started.elapsed().as_nanos()
    );
}

type CleanHeadFixture = (tempfile::TempDir, [u64; 2], u64);

/// A private copy of the clean-shutdown head fixture, built once per test
/// binary build, with the heads its build measured and how many times it was
/// built.
fn clean_self_stream_fixture(
    started: &std::time::Instant,
) -> Result<CleanHeadFixture, Box<dyn Error>> {
    let copy = crate::test_fixture::fixture_copy(SELF_HEADS_FIXTURE, |data| {
        let (source, heads) = build_clean_self_stream_fixture(started)?;
        copy_fixture(source.path(), data)?;
        fs::write(data.join(HEADS_FILE), serde_json::to_vec(&heads)?)?;
        source.close()?;
        Ok(())
    })?;
    let heads_path = copy.path().join(HEADS_FILE);
    let heads: [u64; 2] = serde_json::from_slice(&fs::read(&heads_path)?)?;
    fs::remove_file(heads_path)?;
    let builds = crate::test_fixture::fixture_builds(SELF_HEADS_FIXTURE)?;
    head_proof_stage(started, "clean_fixture_copied");
    Ok((copy, heads, builds))
}

fn build_clean_self_stream_fixture(
    started: &std::time::Instant,
) -> Result<(tempfile::TempDir, [u64; 2]), Box<dyn Error>> {
    use super::sink::SelfStreams;
    use crate::server::connection::services::ConnectionServices;
    use liminal::durability::{bridge::block_on, head_checkpoint::head_checkpoint_key};
    let R3Fixture {
        services,
        store,
        root,
    } = R3Fixture::new()?;
    head_proof_stage(started, "fixture_opened");
    let wake = services.signals().wake();
    let streams = Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 1)?,
        &services,
        Arc::clone(&wake),
        crate::server::shutdown::ShutdownHandle::new(),
        crate::config::types::SelfStreamsConfig::default(),
    ));
    services.signals().install(Arc::clone(&streams))?;
    wake.install_source(&streams)?;
    let counts = services.store_counts();
    let names = ["durable", "ephemeral-dedup"];
    counts.install_observer(
        &names[..counts.store_count()],
        Arc::clone(&wake) as Arc<dyn liminal::durability::StoreCountObserver>,
    )?;
    eprintln!(
        "HEAD_PROOF_COUNTS phase=before_event counts={:?}",
        counts.snapshot()
    );
    head_proof_stage(started, "streams_installed");
    streams.event(ServerEvent::ConnectionOpened { connection: 1 });
    streams.drain()?;
    head_proof_stage(started, "event_drained");
    eprintln!(
        "HEAD_PROOF_COUNTS phase=event_drained counts={:?}",
        counts.snapshot()
    );
    let mut publisher = super::owner::PublisherRuntime::start(
        Arc::clone(&wake),
        Arc::clone(&streams),
        crate::server::shutdown::ShutdownHandle::new(),
    )?;
    head_proof_stage(started, "publisher_started");
    publisher.finish()?;
    head_proof_stage(started, "publisher_finished");
    eprintln!(
        "HEAD_PROOF_COUNTS phase=publisher_finished counts={:?}",
        counts.snapshot()
    );
    services.flush_durable_state()?;
    head_proof_stage(started, "durable_state_flushed");
    eprintln!(
        "HEAD_PROOF_COUNTS phase=durable_state_flushed counts={:?}",
        counts.snapshot()
    );
    let mut heads = [0; 2];
    for (index, name) in ["liminal.events", "liminal.cost"].into_iter().enumerate() {
        let positions = services
            .channel_handle_for_test(name)?
            .durable_positions()?;
        heads[index] = *positions.first().ok_or("self stream has no partition")?;
        let head = block_on(store.read_value(&head_checkpoint_key(&format!("{name}:0"))))??;
        assert_eq!(
            head,
            Some(heads[index]),
            "clean shutdown did not record {name} head"
        );
    }
    head_proof_stage(started, "heads_verified");
    drop(publisher);
    drop(streams);
    drop(wake);
    drop(services);
    drop(store);
    head_proof_stage(started, "fixture_closed");
    Ok((root, heads))
}

#[derive(Clone)]
struct HeadLog(Arc<Mutex<Vec<u8>>>);

impl std::io::Write for HeadLog {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .map_err(|error| std::io::Error::other(error.to_string()))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

fn read_clean_heads(
    root: &Path,
    expected: [u64; 2],
) -> Result<(StoreCounts, String), Box<dyn Error>> {
    use crate::server::connection::services::{LiminalConnectionServices, open_persistent_store};
    let log = HeadLog(Arc::new(Mutex::new(Vec::new())));
    let writer = log.clone();
    let subscriber = tracing_subscriber::fmt()
        .without_time()
        .with_ansi(false)
        .with_max_level(tracing::Level::INFO)
        .with_writer(move || writer.clone())
        .finish();
    let counts = tracing::subscriber::with_default(
        subscriber,
        || -> Result<StoreCounts, Box<dyn Error>> {
            let store = open_persistent_store(root)?;
            let services =
                LiminalConnectionServices::from_config_with_store(&wire_config()?, store)?;
            let before = services.store_counts().snapshot();
            services.install_self_stream_channels()?;
            let delta = services.store_counts().snapshot().since(&before);
            for (index, name) in ["liminal.events", "liminal.cost"].into_iter().enumerate() {
                assert_eq!(
                    services
                        .channel_handle_for_test(name)?
                        .durable_positions()?,
                    vec![expected[index]]
                );
            }
            Ok(delta)
        },
    )?;
    let text = String::from_utf8(log.0.lock().map_err(test_sink_error)?.clone())?;
    Ok((counts, text))
}

#[test]
fn clean_shutdown_records_both_self_heads_and_deleted_head_control_replays()
-> Result<(), Box<dyn Error>> {
    use liminal::durability::head_checkpoint::head_checkpoint_key;
    let started = std::time::Instant::now();
    head_proof_stage(&started, "case_entered");
    let (root, heads, builds) = clean_self_stream_fixture(&started)?;
    assert_eq!(heads, [2, 1]);
    let (missing, repeated_heads, repeated_builds) = clean_self_stream_fixture(&started)?;
    assert_eq!(repeated_heads, heads);
    eprintln!("HEAD_PROOF_BUILD_COUNTS first={builds} repeated={repeated_builds}");
    assert_eq!(
        [builds, repeated_builds],
        [1, 1],
        "the clean head fixture was rebuilt"
    );
    head_proof_stage(&started, "deleted_control_copied");
    let (clean, log) = read_clean_heads(root.path(), heads)?;
    head_proof_stage(&started, "clean_heads_read");
    eprintln!("HEAD_PROOF_COUNTS phase=clean_read counts={clean:?}");
    assert_eq!(clean.read_value_calls, 2);
    assert_eq!(clean.read_at_calls, 2);
    assert_eq!(clean.read_at_rows, 2);
    assert_eq!(clean.read_from_calls, 0);
    assert_eq!(
        clean.read_from_rows, 0,
        "clean shutdown replayed the self streams' prefix"
    );
    assert_eq!(clean.cas_calls, 0);
    assert!(!log.contains("no head is recorded"), "{log}");
    assert!(!log.contains("rebuilt_head"), "{log}");
    let database = haematite::Database::open(
        missing.path().join("durability"),
        haematite::RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: haematite::ReclaimStart::AtOpen,
        },
    )?;
    head_proof_stage(&started, "delete_database_opened");
    for name in ["liminal.events:0", "liminal.cost:0"] {
        database.delete(head_checkpoint_key(name).into_bytes())?;
    }
    database.commit()?;
    head_proof_stage(&started, "head_deletion_committed");
    drop(database);
    head_proof_stage(&started, "delete_database_closed");
    let (deleted, log) = read_clean_heads(missing.path(), heads)?;
    head_proof_stage(&started, "deleted_heads_read");
    eprintln!("HEAD_PROOF_COUNTS phase=deleted_read counts={deleted:?}");
    assert_eq!(deleted.read_value_calls, 4);
    assert_eq!(deleted.read_at_calls, 0);
    assert_eq!(deleted.read_from_rows, heads.into_iter().sum::<u64>());
    assert_eq!(deleted.cas_calls, 2);
    assert_eq!(
        log.matches("durable channel head recorded for a stream under its first checkpoint")
            .count(),
        2,
        "{log}"
    );
    assert!(!log.contains("WARN"), "{log}");
    assert!(
        !log.contains("durable channel head record refused"),
        "{log}"
    );
    drop(missing);
    drop(root);
    head_proof_stage(&started, "roots_removed");
    Ok(())
}

#[test]
fn lifecycle_receipts_share_one_append_all_one_flush_and_commit_order() -> Result<(), Box<dyn Error>>
{
    use liminal::durability::bridge::block_on;
    let fixture = WireFixture::new()?;
    let subscriber = fixture
        .services
        .subscribe_handle_for_test("liminal.events")?;
    let before = fixture.services.store_counts().snapshot();
    fixture
        .services
        .signals()
        .lifecycle_batch(12, [(1, "enrolled"), (2, "attached"), (3, "quit")])?;
    fixture.sink.drain()?;
    let delta = fixture.services.store_counts().snapshot().since(&before);
    assert_eq!(
        delta.append_all_calls, 1,
        "one operation's receipts were separate appends"
    );
    assert_eq!(delta.append_all_rows, 3);
    assert_eq!(
        delta.append_rows, 1,
        "only the flush observer is a separate append"
    );
    assert_eq!(delta.flushes, 1);
    let work = fixture.sink.own_work_for_test()?;
    assert_eq!(work.rows, 4);
    assert_eq!(work.flushes, 1);
    let store = fixture.services.durable_store();
    for sequence in 0..3 {
        let delivered = subscriber
            .try_next()?
            .ok_or("lifecycle receipt was not delivered")?;
        let live: Value = serde_json::from_slice(&delivered.payload)?;
        assert_eq!(live["details"]["conversation"], 12);
        assert_eq!(live["details"]["generation"], sequence + 1);
        let row = block_on(store.read_at("liminal.events:0", sequence))??
            .ok_or("lifecycle receipt was not stored")?;
        let stored = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        assert_eq!(serde_json::from_slice::<Value>(&stored.payload)?, live);
    }
    let flush = subscriber
        .try_next()?
        .ok_or("batch flush event was not delivered")?;
    let value: Value = serde_json::from_slice(&flush.payload)?;
    assert_eq!(value["details"]["kind"], "flush");
    assert!(subscriber.try_next()?.is_none());
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn acknowledged_cost_windows_attribute_batch_flush_once_and_cost_row_later()
-> Result<(), Box<dyn Error>> {
    let fixture = WireFixture::new()?;
    let mut sink = Arc::clone(&fixture.sink);
    let (initial, own) = CostSink::snapshot(&sink)?;
    let mut window = CostWindow::new(initial, own);
    fixture
        .services
        .signals()
        .lifecycle_batch(7, [(1, "enrolled"), (2, "attached"), (3, "quit")])?;
    for (index, expected) in [(4, 1), (1, 0)].into_iter().enumerate() {
        let (snapshot, work) = CostSink::snapshot(&sink)?;
        let record = window
            .prepare(snapshot, work)
            .ok_or("pending cost work disappeared")?;
        CostSink::publish(&mut sink, &record)?;
        window.published(&record);
        let position = u64::try_from(index)?;
        let store = fixture.services.durable_store();
        let row =
            liminal::durability::bridge::block_on(store.read_at("liminal.cost:0", position))??
                .ok_or("acknowledged cost row was not stored")?;
        let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        let cost: Value = serde_json::from_slice(&envelope.payload)?;
        assert_eq!(cost["self_rows"], expected.0);
        assert_eq!(cost["self_flushes"], expected.1);
    }
    assert_eq!(
        fixture
            .services
            .store_counts()
            .snapshot()
            .since(&initial)
            .flushes,
        1
    );
    fixture.supervisor.shutdown();
    Ok(())
}

#[test]
fn receipt_batch_flush_failure_keeps_rows_and_recovers_without_reappend()
-> Result<(), Box<dyn Error>> {
    receipt_batch_fault_boundary(false)
}

#[test]
fn partial_receipt_batch_is_named_and_never_delivered_or_reappended() -> Result<(), Box<dyn Error>>
{
    receipt_batch_fault_boundary(true)
}

fn receipt_batch_fault_boundary(partial: bool) -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::LiminalConnectionServices;
    let fixture = R3Fixture::new()?;
    let faults = Arc::new(liminal::durability::CountingStore::new(
        CommitThenErrorStore {
            inner: Arc::clone(&fixture.store),
            fail_cost: std::sync::atomic::AtomicBool::new(false),
            commit_cost: false,
            partial_batch: partial,
            fail_flush: std::sync::atomic::AtomicBool::new(!partial),
            pause_cost: None,
            calls: None,
            observer_boundary: None,
        },
    ));
    let services = LiminalConnectionServices::from_config_with_store(&wire_config()?, faults)?;
    services.install_self_stream_channels()?;
    let channel = services.channel_handle_for_test("liminal.cost")?;
    let subscriber = services.subscribe_handle_for_test("liminal.cost")?;
    let payloads = vec![b"{}".to_vec(); 3];
    let error = publish_receipt_batch_for_test(&channel, &payloads)
        .err()
        .ok_or("faulted batch succeeded")?;
    assert!(matches!(
        error.downcast_ref::<liminal::LiminalError>(),
        Some(liminal::LiminalError::PublishUnresolved { .. })
    ));
    assert!(subscriber.try_next()?.is_none());
    let before = services.store_counts().snapshot();
    let next = channel.publish_with_delivery(
        b"{}",
        liminal::envelope::PublisherId::new("following"),
        None,
    );
    let delta = services.store_counts().snapshot().since(&before);
    assert_eq!(delta.append_all_calls, 0);
    assert_eq!(delta.read_at_calls, 3);
    assert_eq!(delta.read_at_rows, if partial { 1 } else { 3 });
    assert_eq!(delta.read_from_calls, 0);
    if partial {
        let failure = next
            .err()
            .ok_or("partially stored batch allowed a following append")?;
        assert!(
            failure
                .to_string()
                .contains("partial durable batch: 1/3 reserved rows match"),
            "{failure}"
        );
        assert_eq!(delta.append_calls, 0);
        assert_eq!(delta.flushes, 0);
    } else {
        assert_eq!(next?.durable_position(), Some(3));
        assert_eq!(delta.append_calls, 1);
        assert_eq!(delta.flushes, 1);
        for _ in 0..4 {
            assert_eq!(
                subscriber
                    .try_next()?
                    .ok_or("recovered batch delivery is absent")?
                    .payload,
                b"{}"
            );
        }
    }
    assert!(subscriber.try_next()?.is_none());
    Ok(())
}

#[test]
fn receipt_batch_rejects_blank_publisher_and_ephemeral_before_store_work()
-> Result<(), Box<dyn Error>> {
    let fixture = R3Fixture::new()?;
    let durable = fixture.services.channel_handle_for_test("liminal.events")?;
    let ephemeral = fixture
        .services
        .channel_handle_for_test("pressure.source")?;
    let before = fixture.services.store_counts().snapshot();
    let payloads = vec![b"{}".to_vec()];
    for publisher in ["", " \t"] {
        assert!(
            durable
                .publish_batch_with_delivery(
                    &payloads,
                    &liminal::envelope::PublisherId::new(publisher),
                    None
                )
                .is_err()
        );
    }
    assert!(publish_receipt_batch_for_test(&ephemeral, &payloads).is_err());
    assert_eq!(fixture.services.store_counts().snapshot(), before);
    Ok(())
}

#[test]
fn receipt_batch_preserves_publisher_and_causal_parent_in_live_and_stored_rows()
-> Result<(), Box<dyn Error>> {
    let fixture = R3Fixture::new()?;
    let channel = fixture.services.channel_handle_for_test("liminal.events")?;
    let subscriber = fixture
        .services
        .subscribe_handle_for_test("liminal.events")?;
    let publisher = liminal::envelope::PublisherId::new("batch-parent-proof");
    let parent = liminal::causal::MessageId::new();
    let causal = liminal::causal::CausalContext::child_of(parent);
    let payloads = vec![b"{}".to_vec(); 3];
    let deliveries = channel.publish_batch_with_delivery(&payloads, &publisher, Some(&causal))?;
    for (index, delivery) in deliveries.into_iter().enumerate() {
        let position = u64::try_from(index)?;
        assert_eq!(delivery.durable_position(), Some(position));
        let live = subscriber
            .try_next()?
            .ok_or("causal batch was not delivered")?;
        assert_eq!(live.publisher_id, publisher);
        assert_eq!(live.causal_context.as_ref(), Some(&causal));
        let row = liminal::durability::bridge::block_on(
            fixture.store.read_at("liminal.events:0", position),
        )??
        .ok_or("causal batch was not stored")?;
        let stored = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        assert_eq!(stored.publisher_id, publisher.as_str());
        assert_eq!(
            stored
                .causal_context
                .ok_or("stored causal parent is absent")?
                .parent_id,
            Some(parent.as_uuid().to_string())
        );
    }
    assert!(subscriber.try_next()?.is_none());
    Ok(())
}

#[path = "tests/drainer.rs"]
mod drainer;

type CapturedPanicHook = dyn Fn(&std::panic::PanicHookInfo<'_>) + Send + Sync + 'static;

struct PanicCapture {
    previous: Arc<CapturedPanicHook>,
    panics: Arc<Mutex<Vec<String>>>,
}

impl PanicCapture {
    fn install() -> Self {
        let previous: Arc<CapturedPanicHook> = Arc::from(std::panic::take_hook());
        let original = Arc::clone(&previous);
        let panics = Arc::new(Mutex::new(Vec::new()));
        let captured = Arc::clone(&panics);
        std::panic::set_hook(Box::new(move |info| {
            let backtrace = std::backtrace::Backtrace::force_capture();
            match captured.lock() {
                Ok(mut panics) => panics.push(format!("{info}\n{backtrace}")),
                Err(error) => eprintln!("panic capture lock poisoned: {error}"),
            }
            original(info);
        }));
        Self { previous, panics }
    }

    fn snapshot(&self) -> Result<Vec<String>, Box<dyn Error>> {
        Ok(self.panics.lock().map_err(test_sink_error)?.clone())
    }
}

impl Drop for PanicCapture {
    fn drop(&mut self) {
        if !std::thread::panicking() {
            let previous = Arc::clone(&self.previous);
            std::panic::set_hook(Box::new(move |info| previous(info)));
        }
    }
}

struct ChannelThreadExit(std::sync::mpsc::Sender<std::thread::ThreadId>);

impl Drop for ChannelThreadExit {
    fn drop(&mut self) {
        if let Err(error) = self.0.send(std::thread::current().id()) {
            eprintln!("channel thread exit watch failed: {error}");
        }
    }
}

thread_local! {
    static CHANNEL_THREAD_EXIT: std::cell::RefCell<Option<ChannelThreadExit>> =
        const { std::cell::RefCell::new(None) };
}

#[derive(Debug, Default)]
struct ReceiptMemoryState {
    streams: std::collections::BTreeMap<
        String,
        std::collections::BTreeMap<u64, liminal::durability::StoredEntry>,
    >,
    values: std::collections::BTreeMap<String, u64>,
}

#[derive(Debug, Default)]
struct ReceiptMemoryStore(Mutex<ReceiptMemoryState>);

fn receipt_memory_error(error: impl std::fmt::Display) -> liminal::durability::DurabilityError {
    liminal::durability::DurabilityError::ConfigError(format!(
        "receipt memory store failed: {error}"
    ))
}

#[async_trait::async_trait]
impl liminal::durability::DurableStore for ReceiptMemoryStore {
    async fn apply_routed_batch(
        &self,
        _: &str,
        _: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        Err(liminal::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn get_routed(
        &self,
        _: &str,
        _: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        Err(liminal::durability::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn append(
        &self,
        key: &str,
        payload: Vec<u8>,
        expected: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        let next = self.append_all(key, vec![payload], expected).await?;
        next.checked_sub(1)
            .ok_or_else(|| receipt_memory_error("assigned sequence underflow"))
    }

    async fn append_all(
        &self,
        key: &str,
        payloads: Vec<Vec<u8>>,
        expected: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        let mut state = self.0.lock().map_err(receipt_memory_error)?;
        let actual = u64::try_from(
            state
                .streams
                .get(key)
                .map_or(0, std::collections::BTreeMap::len),
        )
        .map_err(receipt_memory_error)?;
        if actual != expected {
            return Err(liminal::durability::DurabilityError::SequenceConflict {
                expected,
                actual,
            });
        }
        let count = u64::try_from(payloads.len()).map_err(receipt_memory_error)?;
        let next = actual
            .checked_add(count)
            .ok_or_else(|| receipt_memory_error("sequence overflow"))?;
        if payloads.is_empty() {
            return Ok(next);
        }
        let stream = state.streams.entry(key.to_owned()).or_default();
        for (sequence, payload) in (actual..next).zip(payloads) {
            stream.insert(
                sequence,
                liminal::durability::StoredEntry {
                    payload,
                    sequence,
                    timestamp: 0,
                },
            );
        }
        drop(state);
        Ok(next)
    }

    async fn read_from(
        &self,
        key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<liminal::durability::StoredEntry>, liminal::durability::DurabilityError> {
        let state = self.0.lock().map_err(receipt_memory_error)?;
        let rows = state.streams.get(key).map_or_else(Vec::new, |stream| {
            stream
                .range(offset..)
                .take(limit)
                .map(|(_, entry)| entry.clone())
                .collect()
        });
        drop(state);
        Ok(rows)
    }

    async fn read_at(
        &self,
        key: &str,
        sequence: u64,
    ) -> Result<Option<liminal::durability::StoredEntry>, liminal::durability::DurabilityError>
    {
        let state = self.0.lock().map_err(receipt_memory_error)?;
        let row = state
            .streams
            .get(key)
            .and_then(|stream| stream.get(&sequence))
            .cloned();
        drop(state);
        Ok(row)
    }

    async fn cas(
        &self,
        key: &str,
        old: u64,
        new: u64,
    ) -> Result<(), liminal::durability::DurabilityError> {
        let mut state = self.0.lock().map_err(receipt_memory_error)?;
        let stored = state.values.get(key).copied().unwrap_or(0);
        if stored != old {
            return Err(liminal::durability::DurabilityError::CursorRegression {
                stored,
                attempted: old,
            });
        }
        state.values.insert(key.to_owned(), new);
        drop(state);
        Ok(())
    }

    async fn read_stream_next_seq(
        &self,
        stream_key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        let state = self.0.lock().map_err(receipt_memory_error)?;
        let head = state
            .streams
            .get(stream_key)
            .filter(|stream| !stream.is_empty())
            .map(|stream| u64::try_from(stream.len()).map_err(receipt_memory_error))
            .transpose();
        drop(state);
        head
    }

    async fn read_value(
        &self,
        key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        let state = self.0.lock().map_err(receipt_memory_error)?;
        let value = state.values.get(key).copied();
        drop(state);
        Ok(value)
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<liminal::durability::ScanCursor>,
        limit: usize,
    ) -> Result<liminal::durability::ScanPage, liminal::durability::DurabilityError> {
        if limit == 0
            || cursor
                .as_ref()
                .is_some_and(|cursor| !cursor.stream_key.starts_with(prefix))
        {
            return Err(receipt_memory_error("invalid scan limit or cursor prefix"));
        }
        let start = cursor
            .as_ref()
            .map_or(prefix, |cursor| cursor.stream_key.as_str());
        let state = self.0.lock().map_err(receipt_memory_error)?;
        let mut entries = Vec::with_capacity(limit);
        let mut next = None;
        for (key, stream) in state.streams.range(start.to_owned()..) {
            if !key.starts_with(prefix) {
                break;
            }
            let offset = cursor
                .as_ref()
                .filter(|cursor| &cursor.stream_key == key)
                .map_or(0, |cursor| cursor.next_sequence);
            for (sequence, entry) in stream.range(offset..) {
                entries.push(entry.clone());
                if entries.len() == limit {
                    next = Some(liminal::durability::ScanCursor {
                        stream_key: key.clone(),
                        next_sequence: sequence
                            .checked_add(1)
                            .ok_or_else(|| receipt_memory_error("scan sequence overflow"))?,
                    });
                    break;
                }
            }
            if next.is_some() {
                break;
            }
        }
        drop(state);
        Ok(liminal::durability::ScanPage { entries, next })
    }

    async fn flush(&self) -> Result<(), liminal::durability::DurabilityError> {
        Ok(())
    }
}

/// Hermes, 7 Oct 11:53, ruling 1 of docs/design/SELF-STREAMS-COST-WAKE-2026-10-07.md:
/// only the server's own writer, the self-record drainer, does self work. A
/// flush made anywhere else is outside work: it arms one tick, and the next
/// cost row carries it. (This pin expected the opposite under ruling
/// 8234a7e, while a counts baseline decided what was self work.)
#[tokio::test]
async fn a_flush_off_the_servers_own_writer_is_outside_work_for_the_next_cost_window()
-> Result<(), Box<dyn Error>> {
    use std::future::{Future, poll_fn};
    use std::pin::pin;
    use std::task::Poll;
    let mut raw_store = None;
    let fixture = WireFixture::with_store(|store| {
        raw_store = Some(Arc::clone(&store));
        store
    })?;
    let raw_store = raw_store.ok_or("raw fixture store is absent")?;
    publish_pressure_source(&fixture)?;
    let first_tick = {
        let mut tick = pin!(fixture.wake.next_tick(|duration| {
            assert_eq!(duration, std::time::Duration::from_secs(10));
            std::future::ready(())
        }));
        poll_fn(|context| Poll::Ready(tick.as_mut().poll(context))).await
    };
    assert!(matches!(first_tick, Poll::Ready(Ok(()))));
    let mut sink = Arc::clone(&fixture.sink);
    let mut window = CostWindow::new(StoreCounts::default(), SelfWork::default());
    let (snapshot, own) = sink.snapshot()?;
    let first = window
        .prepare(snapshot, own)
        .ok_or("first cost window is absent")?;
    CostSink::publish(&mut sink, &first)?;
    window.published(&first);
    sink.before_stop()?;
    let before_flush = fixture.services.store_counts().snapshot().flushes;
    fixture.services.durable_store().flush().await?;
    sink.before_stop()?;
    let external_work = fixture.wake.external_work();
    let mut ready_ticks = 0;
    for _ in 0..2 {
        let observed = {
            let mut tick = pin!(fixture.wake.next_tick(|duration| {
                assert_eq!(duration, std::time::Duration::from_secs(10));
                std::future::ready(())
            }));
            poll_fn(|context| Poll::Ready(tick.as_mut().poll(context))).await
        };
        match observed {
            Poll::Ready(result) => {
                result?;
                ready_ticks += 1;
                let (snapshot, own) = sink.snapshot()?;
                let record = window
                    .prepare(snapshot, own)
                    .ok_or("armed cost window is absent")?;
                CostSink::publish(&mut sink, &record)?;
                window.published(&record);
                sink.before_stop()?;
            }
            Poll::Pending => {}
        }
    }
    let mut cost_rows = 0;
    for sequence in 0..3 {
        if raw_store
            .read_at("liminal.cost:0", sequence)
            .await?
            .is_some()
        {
            cost_rows += 1;
        }
    }
    let flushes = fixture.services.store_counts().snapshot().flushes - before_flush;
    fixture.supervisor.shutdown();
    assert_eq!(flushes, 1, "own store flush must exercise one real flush");
    assert_eq!(
        (external_work, ready_ticks, cost_rows),
        (true, 1, 2),
        "a flush off the server's own writer was not published as outside work"
    );
    Ok(())
}

#[tokio::test]
async fn head_read_arms_one_cost_tick_and_returns_to_idle() -> Result<(), Box<dyn Error>> {
    use liminal::durability::{CountingStore, DurableStore};

    let wake = Arc::new(CostWake::default());
    let store = CountingStore::new(ReceiptMemoryStore::default());
    store
        .handle()
        .install_observer(&["durable"], wake.clone())?;
    assert!(!ready_cost_tick_for_test(&wake).await?);
    assert_eq!(store.read_stream_next_seq("head-read").await?, None);
    assert_eq!(
        store.snapshot(),
        StoreCounts {
            read_stream_next_seq_calls: 1,
            ..StoreCounts::default()
        }
    );
    assert!(ready_cost_tick_for_test(&wake).await?);
    assert!(!ready_cost_tick_for_test(&wake).await?);
    Ok(())
}

async fn ready_cost_tick_for_test(wake: &CostWake) -> Result<bool, ServerError> {
    use std::future::{Future, poll_fn};
    use std::pin::pin;
    use std::task::Poll;
    let mut tick = pin!(wake.next_tick(|duration| {
        assert_eq!(duration, std::time::Duration::from_secs(10));
        std::future::ready(())
    }));
    match poll_fn(|context| Poll::Ready(tick.as_mut().poll(context))).await {
        Poll::Ready(result) => result.map(|()| true),
        Poll::Pending => Ok(false),
    }
}

async fn publish_initial_cost_window_for_test(fixture: &WireFixture) -> Result<(), Box<dyn Error>> {
    assert!(ready_cost_tick_for_test(&fixture.wake).await?);
    let mut sink = Arc::clone(&fixture.sink);
    let (snapshot, own) = sink.snapshot()?;
    let record = CostWindow::new(StoreCounts::default(), SelfWork::default())
        .prepare(snapshot, own)
        .ok_or("initial connection cost window is absent")?;
    CostSink::publish(&mut sink, &record)?;
    sink.before_stop()?;
    Ok(())
}

#[tokio::test]
async fn participant_lifecycle_commit_arms_the_next_cost_window() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::ConnectionServices;
    use crate::server::mount::MountKind;
    use crate::server::participant::{
        ParticipantConnectionContext, ParticipantConnectionConversations,
        ParticipantSemanticHandler,
    };
    use liminal_protocol::wire::{
        ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, ServerValue,
    };
    let mut raw_store = None;
    let fixture = WireFixture::with_store(|store| {
        raw_store = Some(Arc::clone(&store));
        store
    })?;
    let raw_store = raw_store.ok_or("raw fixture store is absent")?;
    let participant = fixture
        .services
        .participant_service()
        .ok_or("participant service is absent")?;
    let mut sink = Arc::clone(&fixture.sink);
    let mut window = CostWindow::new(StoreCounts::default(), SelfWork::default());
    let (snapshot, own) = sink.snapshot()?;
    let first = window
        .prepare(snapshot, own)
        .ok_or("startup cost window is absent")?;
    CostSink::publish(&mut sink, &first)?;
    window.published(&first);
    sink.before_stop()?;
    assert!(!fixture.wake.external_work());
    let response = participant.handle(
        ParticipantConnectionContext::new(
            ConnectionIncarnation::new(fixture.incarnation, 1),
            MountKind::Tcp,
        ),
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: 7,
            enrollment_token: EnrollmentToken::new([1; 16]),
        }),
    )?;
    assert!(matches!(response, ServerValue::EnrollBound(_)));
    sink.before_stop()?;
    let external_publication = fixture.wake.external_work();
    let ready = ready_cost_tick_for_test(&fixture.wake).await?;
    if ready {
        let (snapshot, own) = sink.snapshot()?;
        let record = window
            .prepare(snapshot, own)
            .ok_or("participant cost window is absent")?;
        CostSink::publish(&mut sink, &record)?;
        window.published(&record);
        sink.before_stop()?;
    }
    let cost_row = raw_store.read_at("liminal.cost:0", 1).await?.is_some();
    let mut enrolled = false;
    for sequence in 0..8 {
        let Some(row) = raw_store.read_at("liminal.events:0", sequence).await? else {
            break;
        };
        let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        let event: Value = serde_json::from_slice(&envelope.payload)?;
        enrolled |= event["details"]["kind"] == "lifecycle"
            && event["details"]["conversation"] == 7
            && event["details"]["transition"] == "enrolled";
    }
    fixture.supervisor.shutdown();
    assert!(
        enrolled,
        "participant commit stored no enrolled lifecycle receipt"
    );
    assert_eq!(
        (ready, cost_row),
        (true, true),
        "participant lifecycle commit lost its cost window (external_publication={external_publication})"
    );
    Ok(())
}

#[test]
fn receipt_batch_retention_reaches_the_bound_including_its_flush_row() -> Result<(), Box<dyn Error>>
{
    use liminal::durability::bridge::block_on;

    let persistence_directory = crate::test_persistence::directory()?;
    let (services, supervisor, mut sink) =
        bounded_self_streams(5, 3, Some(persistence_directory.path()))?;
    let before = services.store_counts().snapshot();
    sink.lifecycle_batch(7, (0..6).map(|generation| (generation, "enrolled")))?;
    sink.before_stop()?;
    let store = services.durable_store();
    let rows = block_on(store.read_from("liminal.events:0", 0, 6))??;
    assert_eq!(rows.len(), 5);
    assert_eq!(rows.first().ok_or("no retained batch events")?.sequence, 2);
    let mut flush_rows = 0;
    for row in &rows {
        let envelope = liminal::durability::MessageEnvelope::deserialize(&row.payload)?;
        let record: Value = serde_json::from_slice(&envelope.payload)?;
        if record["details"]["kind"] == "flush" {
            assert_eq!(record["details"]["store"], "durable");
            flush_rows += 1;
        }
    }
    assert_eq!(flush_rows, 1);
    let (counts, own) = sink.snapshot()?;
    let measured = counts.since(&before);
    assert_eq!((own.prunes, own.rows_pruned, own.prune_failures), (2, 2, 0));
    assert_eq!((measured.truncate_calls, measured.truncate_rows), (2, 2));
    assert_eq!(own.prunes, measured.truncate_calls);
    assert_eq!(own.rows_pruned, measured.truncate_rows);
    supervisor.shutdown();
    Ok(())
}

/// How long a wire pin waits to be told what a connection wrote. It is
/// generous: a told wait never sleeps, and a tight deadline on a loaded
/// machine fails for the wrong reason.
const FRAME_WITNESS_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);

/// R3b F8, ruling 2a: an ack the server never sent is named by the Pong it
/// wrote after the request, never waited on.
#[test]
fn withheld_acks_are_named_by_the_pong_written_after_them() -> Result<(), Box<dyn Error>> {
    use liminal::protocol::{Frame, FrameType, ProtocolVersion};
    let fixture = WireFixture::new()?;
    let mut peer = witnessed_peer(&fixture)?;
    let connect_ack_withheld = expect_frame(&mut peer, FrameType::ConnectAck);
    let connect_sent = send_frame(
        &mut peer,
        &Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        },
    )
    .map_err(|error| error.to_string());
    let connect_ack = expect_frame(&mut peer, FrameType::ConnectAck);
    let subscribe_ack_withheld = expect_frame(&mut peer, FrameType::SubscribeAck);
    let subscribe_sent = send_frame(
        &mut peer,
        &Frame::Subscribe {
            flags: 0,
            stream_id: 1,
            channel: "pressure.source".to_owned(),
            accepted_schemas: Vec::new(),
            max_in_flight: 16,
        },
    )
    .map_err(|error| error.to_string());
    let subscribe_ack = expect_frame(&mut peer, FrameType::SubscribeAck);
    drop(peer);
    fixture.supervisor.shutdown();
    assert_eq!(
        connect_ack_withheld.err().as_deref(),
        Some("ConnectAck absent although the Pong after it was written")
    );
    assert!(connect_sent.is_ok(), "{connect_sent:?}");
    assert!(connect_ack.is_ok(), "{connect_ack:?}");
    assert_eq!(
        subscribe_ack_withheld.err().as_deref(),
        Some("SubscribeAck absent although the Pong after it was written")
    );
    assert!(subscribe_sent.is_ok(), "{subscribe_sent:?}");
    assert!(subscribe_ack.is_ok(), "{subscribe_ack:?}");
    Ok(())
}

/// R3b F8, ruling 2a: a delivery the server never wrote is named by the Pong
/// written after the publish answered that the inbox queued it.
#[test]
fn a_withheld_deliver_is_named_by_the_pong_written_after_it() -> Result<(), Box<dyn Error>> {
    use crate::server::connection::services::ConnectionServices;
    use liminal::protocol::{Frame, FrameType, ProtocolVersion};
    let fixture = WireFixture::new()?;
    let mut peer = witnessed_peer(&fixture)?;
    send_frame(
        &mut peer,
        &Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        },
    )?;
    let connect_ack = expect_frame(&mut peer, FrameType::ConnectAck);
    send_frame(
        &mut peer,
        &Frame::Subscribe {
            flags: 0,
            stream_id: 1,
            channel: "pressure.source".to_owned(),
            accepted_schemas: Vec::new(),
            max_in_flight: 16,
        },
    )?;
    let subscribe_ack = expect_frame(&mut peer, FrameType::SubscribeAck);
    let nothing_published = expect_frame(&mut peer, FrameType::Deliver);
    let published = fixture.services.publish(
        7,
        "pressure.source",
        &self_stream_publication_envelope(),
        None,
    )?;
    let deliver = expect_frame(&mut peer, FrameType::Deliver);
    drop(peer);
    fixture.supervisor.shutdown();
    assert!(connect_ack.is_ok(), "{connect_ack:?}");
    assert!(subscribe_ack.is_ok(), "{subscribe_ack:?}");
    assert_eq!(
        nothing_published.err().as_deref(),
        Some("Deliver absent although the Pong after it was written")
    );
    assert!(
        published.delivered,
        "the publish reached no inbox: {published:?}"
    );
    assert!(deliver.is_ok(), "{deliver:?}");
    Ok(())
}

/// A loopback peer whose frames are expected by kind (R3b F8): it is told
/// what its connection wrote (ruling 2a), so it never parks on a read.
struct WitnessedPeer {
    client: crate::server::connection::LoopbackClientEnd,
    buffer: Vec<u8>,
    written: std::sync::mpsc::Receiver<Vec<liminal::protocol::FrameType>>,
}

fn witnessed_peer(fixture: &WireFixture) -> Result<WitnessedPeer, Box<dyn Error>> {
    let (client, server) = crate::server::connection::LoopbackDuplex::bounded(256 * 1024);
    let connection = fixture.supervisor.spawn_loopback_connection(server)?;
    Ok(WitnessedPeer {
        client,
        buffer: Vec::new(),
        written: connection.observe_frames_written(),
    })
}

fn send_frame(
    peer: &mut WitnessedPeer,
    frame: &liminal::protocol::Frame,
) -> Result<(), Box<dyn Error>> {
    write_raw_frame(&mut peer.client, frame)
}

/// The first `expected` frame the connection has written. A Ping goes out
/// after whatever the pin sent, and the pin waits to be told the Pong was
/// written. Replies are queued in order, so anything owed before the Pong
/// is written by then. Only the bytes already there are read, so an absent
/// frame is named, never waited on.
fn expect_frame(
    peer: &mut WitnessedPeer,
    expected: liminal::protocol::FrameType,
) -> Result<liminal::protocol::Frame, String> {
    use liminal::protocol::{Frame, FrameType};
    let ping = Frame::new_ping(0).map_err(|error| error.to_string())?;
    write_raw_frame(&mut peer.client, &ping).map_err(|error| error.to_string())?;
    loop {
        let kinds = peer
            .written
            .recv_timeout(FRAME_WITNESS_DEADLINE)
            .map_err(|error| {
                format!(
                    "the connection did not report the Pong for {expected:?} within \
                     {FRAME_WITNESS_DEADLINE:?}: {error}"
                )
            })?;
        if kinds.contains(&FrameType::Pong) {
            break;
        }
    }
    let readable = peer.client.readable_bytes();
    let mut chunk = vec![0; readable];
    let read = peer
        .client
        .read_timeout(&mut chunk, Some(std::time::Duration::ZERO))
        .map_err(|error| error.to_string())?;
    peer.buffer
        .extend_from_slice(chunk.get(..read).ok_or("invalid raw read length")?);
    let mut found = None;
    while !peer.buffer.is_empty() {
        let (frame, consumed) =
            liminal::protocol::decode(&peer.buffer).map_err(|error| error.to_string())?;
        peer.buffer.drain(..consumed);
        if found.is_none() && frame.frame_type() == expected {
            found = Some(frame);
        }
    }
    found.ok_or_else(|| format!("{expected:?} absent although the Pong after it was written"))
}

#[path = "embedded_publish_tests.rs"]
mod embedded_publish_tests;