liminal-server 0.23.1

//! A REAL log refuses a stored fenced attach on cold replay on BOTH orders the
//! live core can write (Hermes's rulings of 18:22 and 19:05; measured on Dean
//! at 19:52 and 20:29; work-list row on main at 2610be6). Where it refuses
//! depends on the binding the marker source row recorded.
//!
//! The marker source row is written when marker M drains, and its canonical
//! bytes record the target's binding AT THE DRAIN. The fenced expectation
//! (`fenced_marker_source`) always recomputes `Detached(prior)`.
//!
//! - Option A, M drains while P is detached: measured, a real log CAN commit
//!   that drain. After a live `Detach(P)` the bound sender's FIRST record
//!   drains M with `target_binding: Detached(prior)`; only later records are
//!   refused with `ObserverBackpressure`. The source row then agrees with the
//!   expectation, the proof is minted, and the fenced attach reaches the
//!   frontier through `apply_live_fenced_attach` in liminal-protocol's
//!   `claim_frontier.rs`, which refuses with `Precedence(FencedRecovery)`
//!   because the sequence and order recovery blocks are never armed outside
//!   restore capsules (`ops_attach.rs:335` records the live arm as unreachable
//!   in production). Pinned by
//!   `a_stored_fenced_attach_is_refused_on_cold_replay_with_precedence_fenced_recovery_at_the_frontier_transition_after_the_mint`.
//! - Option B, M drains while P is bound, then `Detach(P)`: the source row
//!   records `Bound(prior)`, so marker-source validation refuses `MarkerBody`
//!   earlier, before the proof is minted. Pinned by
//!   `a_stored_fenced_attach_after_a_bound_drain_is_refused_on_cold_replay_with_marker_body_before_the_mint`.
//!
//! Row g is a RAW `Attached(Fenced)` row appended straight to the log, with a
//! proof built exactly as `tests_w1b_oracle26.rs` builds it.

use std::error::Error;
use std::sync::Arc;

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, open_ephemeral};
use liminal_protocol::lifecycle::{
    BindingState, FrontierBinding, LiveFrontierError, PrecedenceCondition,
};
use liminal_protocol::wire::{
    AttachAttemptToken, AttachSecret, BindingEpoch, ClientRequest, CloseCause,
    ConnectionIncarnation, CredentialAttachRequest, DetachAttemptToken, DetachRequest, EnrollBound,
    EnrollmentRequest, EnrollmentToken, Generation, LeaveAttemptToken, LeaveRequest,
    ObserverBackpressure, ParticipantAck, ParticipantId, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::fenced_attach_codec::{
    StoredDebtCompletion, StoredDetachedCredentialRecovery, StoredMarkerCursorProgress,
    StoredMarkerDelivery, StoredProofBinding, StoredRecoveryTerminal, StoredWideResourceVector,
};
use super::log::{
    DecodedStoredOperation, OperationLog, OperationSchemaPhase, READ_BATCH_SIZE,
    StoredAttachAllocation, StoredAttachModeV3, StoredBindingEpoch, StoredFencedAttachProof,
    StoredMarkerDrain, StoredOperation, StoredOperationV2, StoredOperationV3, StoredU128,
};
use super::log_v3::StoredComposedTerminalCause;
use super::marker_source::MarkerSourceRefusalReason;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::outbox_replay::RestoreError;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::state::{ConversationAuthority, StateError};
use super::tests::dispatch;
use super::tests_marker_ack_fixture::{attached_marker_fixture, marker_fixture_config};

const fn stored_wide(entries: u128, bytes: u128) -> StoredWideResourceVector {
    StoredWideResourceVector {
        entries: StoredU128(entries.to_be_bytes()),
        bytes: StoredU128(bytes.to_be_bytes()),
    }
}

/// What the live `Detach` of P left behind, read off the conversation owner
/// the production dispatch path installed.
struct DetachedTarget {
    participant_id: ParticipantId,
    prior_epoch: BindingEpoch,
    attach_secret: AttachSecret,
    cursor: u64,
    terminal_transaction_order: u64,
    terminal_delivery_seq: u64,
    next_order: u64,
    next_seq: u64,
    /// The physical log sequence the raw fenced row is appended at.
    fenced_row_sequence: u64,
}

/// The durable rows of one fixture, ready for row g.
struct FencedFixture {
    store: Arc<dyn DurableStore>,
    conversation_id: u64,
    marker_seq: u64,
    marker_source: u64,
    target: DetachedTarget,
}

/// A participant's live binding epoch and attach secret, read before the
/// detach so the detach is addressed to the epoch that exists.
fn bound_participant(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    participant_id: ParticipantId,
) -> Result<(BindingEpoch, AttachSecret), Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let owner = cell
        .lock()
        .map_err(|_| "fenced replay fixture owner lock was poisoned")?;
    let slot = owner
        .as_ref()
        .ok_or("fenced replay fixture conversation owner was absent")?
        .slots
        .get(&participant_id)
        .ok_or("fenced replay fixture target slot was absent")?;
    let BindingState::Bound(active) = slot.binding else {
        return Err(format!(
            "NOT ARMED: participant {participant_id} is not bound, so there is no live binding \
             to detach: {:?}",
            slot.binding
        )
        .into());
    };
    let found = (active.binding_epoch, slot.attach_secret);
    drop(owner);
    Ok(found)
}

/// Row f: the live `Detach` of P, through production dispatch.
fn detach(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    participant_id: ParticipantId,
    prior_epoch: BindingEpoch,
) -> Result<(), Box<dyn Error>> {
    let detached = dispatch(
        handler,
        prior_epoch.connection_incarnation,
        ClientRequest::Detach(DetachRequest {
            conversation_id,
            participant_id,
            capability_generation: prior_epoch.capability_generation,
            detach_attempt_token: DetachAttemptToken::new([0xE6; 16]),
        }),
    )?;
    if !matches!(detached, ServerValue::DetachCommitted(_)) {
        return Err(format!(
            "NOT ARMED: the live detach of the target did not commit: {detached:?}"
        )
        .into());
    }
    Ok(())
}

/// The detached state the dispatch path left installed for P, and the head
/// row g will be appended at.
fn read_detached_target(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    participant_id: ParticipantId,
    prior_epoch: BindingEpoch,
    attach_secret: AttachSecret,
) -> Result<DetachedTarget, Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let owner = cell
        .lock()
        .map_err(|_| "fenced replay fixture owner lock was poisoned")?;
    let authority = owner
        .as_ref()
        .ok_or("fenced replay fixture conversation owner was absent after the detach")?;
    let slot = authority
        .slots
        .get(&participant_id)
        .ok_or("fenced replay fixture target slot was absent after the detach")?;
    if slot.binding != BindingState::Detached {
        return Err(format!(
            "NOT ARMED: the committed detach left the target {:?}, not Detached, so the row \
             under test would need a composed terminal this fixture does not write",
            slot.binding
        )
        .into());
    }
    let terminal = slot
        .member
        .latest_terminal()
        .ok_or("the committed detach left the target no latest terminal")?;
    if terminal.binding_epoch() != prior_epoch
        || terminal.close_cause() != CloseCause::CleanDeregister
    {
        return Err(format!(
            "NOT ARMED: the target's latest terminal is not the clean detach of epoch \
             {prior_epoch:?}: {terminal:?}"
        )
        .into());
    }
    let (next_order, next_seq) = authority.next_position_for_test();
    let found = DetachedTarget {
        participant_id,
        prior_epoch,
        attach_secret,
        cursor: slot.member.cursor(),
        terminal_transaction_order: terminal.admission_order().transaction_order(),
        terminal_delivery_seq: terminal.delivery_seq(),
        next_order,
        next_seq,
        fenced_row_sequence: authority.next_log_sequence,
    };
    drop(owner);
    Ok(found)
}

/// Every `MarkerDrained` row in `[from, to)`, with its physical sequence.
fn marker_drains(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
    from: u64,
    to: u64,
) -> Result<Vec<(u64, StoredMarkerDrain)>, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let mut drains = Vec::new();
    for sequence in from..to {
        let decoded = block_on(log.read_at(sequence))??
            .ok_or_else(|| format!("operation log row {sequence} is missing"))?;
        match decoded.operation {
            DecodedStoredOperation::V2(StoredOperationV2::MarkerDrained { row })
            | DecodedStoredOperation::V3(StoredOperationV3::MarkerDrained { row }) => {
                drains.push((sequence, row));
            }
            DecodedStoredOperation::V2(_) | DecodedStoredOperation::V3(_) => {}
        }
    }
    Ok(drains)
}

/// The source row: the `MarkerDrained` row whose retained charge is M.
fn marker_source_sequence(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
    marker_seq: u64,
    log_head: u64,
) -> Result<u64, Box<dyn Error>> {
    marker_drains(store, conversation_id, 0, log_head)?
        .into_iter()
        .find(|(_, row)| row.retained_charge.delivery_seq == marker_seq)
        .map(|(sequence, _)| sequence)
        .ok_or_else(|| {
            format!("no MarkerDrained row below log head {log_head} drains marker {marker_seq}")
                .into()
        })
}

/// Option B's rows: exactly `attached_marker_fixture` (M drains while P is
/// bound), then the live `Detach(P)`.
fn bound_drain_fixture() -> Result<FencedFixture, Box<dyn Error>> {
    let fixture = attached_marker_fixture()?;
    let conversation_id = fixture.marker_delivery.conversation_id;
    let marker_seq = fixture.marker_delivery.delivery_seq;
    let participant_id = fixture.target_participant;
    let (prior_epoch, attach_secret) =
        bound_participant(&fixture.handler, conversation_id, participant_id)?;
    detach(
        &fixture.handler,
        conversation_id,
        participant_id,
        prior_epoch,
    )?;
    let target = read_detached_target(
        &fixture.handler,
        conversation_id,
        participant_id,
        prior_epoch,
        attach_secret,
    )?;
    let store = Arc::clone(&fixture.store);
    let marker_source = marker_source_sequence(
        &store,
        conversation_id,
        marker_seq,
        target.fenced_row_sequence,
    )?;
    drop(fixture);
    Ok(FencedFixture {
        store,
        conversation_id,
        marker_seq,
        marker_source,
        target,
    })
}

/// The fenced mode of row g. The proof is built exactly as
/// `tests_w1b_oracle26.rs` `fenced_mode` builds it; only the identities come
/// from the real log (P, its detached epoch, its committed detach terminal,
/// the marker and the marker-source row).
fn fenced_mode(fixture: &FencedFixture) -> Result<StoredAttachModeV3, Box<dyn Error>> {
    let conversation_id = fixture.conversation_id;
    let marker_seq = fixture.marker_seq;
    let target = &fixture.target;
    let prior: StoredBindingEpoch = target.prior_epoch.into();
    let marker_predecessor = marker_seq
        .checked_sub(1)
        .ok_or("marker predecessor underflow")?;
    let recovery = StoredDetachedCredentialRecovery {
        conversation_id,
        participant_id: target.participant_id,
        marker_delivery_seq: marker_seq,
        prior_binding_epoch: prior,
        resulting_floor: marker_predecessor,
        terminal: StoredRecoveryTerminal::Committed {
            binding: StoredProofBinding {
                conversation_id,
                participant_id: target.participant_id,
                binding_epoch: prior,
            },
            cause: StoredComposedTerminalCause::CleanDeregister,
            transaction_order: target.terminal_transaction_order,
            delivery_seq: target.terminal_delivery_seq,
        },
        progress: StoredMarkerCursorProgress {
            conversation_id,
            participant_id: target.participant_id,
            binding_epoch: prior,
            through_seq: marker_seq,
            marker_delivery_seq: marker_seq,
            delivery: StoredMarkerDelivery {
                participant_id: target.participant_id,
                binding_epoch: prior,
                marker_delivery_seq: marker_seq,
            },
        },
    };
    let proof = StoredFencedAttachProof::encode(
        &recovery,
        stored_wide(2, 2),
        marker_seq.checked_add(1).ok_or("fenced floor overflow")?,
        StoredDebtCompletion::ObserverProjection {
            debt: stored_wide(1, 1),
            through_seq: marker_seq
                .checked_add(2)
                .ok_or("projection boundary overflow")?,
        },
    )?;
    Ok(StoredAttachModeV3::Fenced {
        prior_binding_epoch: prior,
        marker_delivery_seq: marker_seq,
        marker_source_sequence: fixture.marker_source,
        proof,
        composed_terminal: None,
    })
}

/// Row g: the RAW `Attached(Fenced)` row, appended directly to the log at the
/// head the fixture left, as one routed commit the way a live commit writes
/// it (`OperationLog::append`), and read back through `read_at`, which reads
/// a routed row at and after the native head.
fn append_fenced_row(fixture: &FencedFixture) -> Result<(), Box<dyn Error>> {
    let target = &fixture.target;
    let next_generation = Generation::new(
        target
            .prior_epoch
            .capability_generation
            .get()
            .checked_add(1)
            .ok_or("fenced generation overflow")?,
    )
    .ok_or("fenced generation was zero")?;
    let new_epoch = BindingEpoch::new(target.prior_epoch.connection_incarnation, next_generation);
    let request = CredentialAttachRequest {
        conversation_id: fixture.conversation_id,
        participant_id: target.participant_id,
        capability_generation: target.prior_epoch.capability_generation,
        attach_secret: target.attach_secret,
        attach_attempt_token: AttachAttemptToken::new([0xE7; 16]),
        accept_marker_delivery_seq: Some(fixture.marker_seq),
    };
    let allocation = StoredAttachAllocation {
        binding_epoch: new_epoch.into(),
        attach_secret: [0xE8; 32],
        attached_order: target.next_order,
        attached_seq: target.next_seq,
        receipt_expires_at: StoredU128(1_u128.to_be_bytes()),
        provenance_expires_at: StoredU128(2_u128.to_be_bytes()),
        admitted_now_ms: 0,
    };
    let mode = fenced_mode(fixture)?;
    let log = OperationLog::new(Arc::clone(&fixture.store), fixture.conversation_id);
    block_on(log.append(
        &StoredOperation::Attached {
            request: (&request).into(),
            secret_verified: true,
            allocation,
            mode: Box::new(mode),
            event: Vec::new(),
        },
        target.fenced_row_sequence,
    ))??;
    let decoded = block_on(log.read_at(target.fenced_row_sequence))??
        .ok_or("the raw fenced Attached row is not durable at its sequence")?;
    if !matches!(
        decoded.operation,
        DecodedStoredOperation::V3(StoredOperation::Attached { ref mode, .. })
            if matches!(**mode, StoredAttachModeV3::Fenced { .. })
    ) {
        return Err("the row at the fenced sequence did not decode as a v3 fenced Attached".into());
    }
    Ok(())
}

/// Cold replay through the production full rebuild, as
/// `tests_one_pass_replay.rs` `one_pass` drives it, under the fixture's config.
///
/// Every live commit is one routed batch (`OperationBatch::commit`), so the
/// conversation's flat stream stays empty and its rows sit routed on
/// `checkpoint_conversation_route`. The rebuild reads the flat rows below the
/// native head and then the routed rows at and after it; the legacy
/// `replay_rewriting` walk reads the flat stream alone and replayed none of
/// these rows.
fn cold_replay(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Result<ConversationAuthority, RestoreError>, Box<dyn Error>> {
    let config = marker_fixture_config();
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let outbox_log = OutboxLog::new(Arc::clone(store), conversation_id);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    Ok(block_on(ConversationAuthority::replay_for_rebuild(
        conversation_id,
        &log,
        &outbox_log,
        &config,
        limits,
    ))?
    .map(|(authority, _, _, _)| authority))
}

/// CONTROL: the log through row f holds exactly the rows before g where the
/// commits wrote them, replays cleanly by the same path and config, consuming
/// exactly those rows, and a boot over it does not refuse the conversation.
/// Whatever refuses after g is appended is therefore g's own.
fn assert_control_replays_clean(fixture: &FencedFixture) -> Result<(), Box<dyn Error>> {
    let durable_head = durable_log_head(&fixture.store, fixture.conversation_id)?;
    assert_eq!(
        durable_head, fixture.target.fenced_row_sequence,
        "control: the durable rows, flat then routed, must end exactly where row g is appended"
    );
    let replayed = cold_replay(&fixture.store, fixture.conversation_id)?
        .map_err(|error| format!("CONTROL FAILED: the log before row g refused: {error}"))?;
    assert_eq!(
        replayed.next_log_sequence, fixture.target.fenced_row_sequence,
        "control: the clean replay must stop exactly where row g is appended"
    );
    drop(replayed);
    let booted = ProductionParticipantHandler::new_restored(
        Arc::clone(&fixture.store),
        marker_fixture_config(),
    )?;
    let unloadable = booted.unloadable_conversations();
    assert!(
        !unloadable.contains_key(&fixture.conversation_id),
        "control: a boot over the log before row g refused the conversation: {unloadable:?}"
    );
    Ok(())
}

/// Cold replay with row g refuses with `RestoreError::Semantic(StateError::
/// Invariant { message })` and exactly `expected`; a fresh boot lists the
/// conversation as unloadable with the same text and installs no owner.
fn assert_refused_on_replay_and_boot(
    fixture: &FencedFixture,
    expected: &str,
) -> Result<(), Box<dyn Error>> {
    let sequence = fixture.target.fenced_row_sequence;
    match cold_replay(&fixture.store, fixture.conversation_id)? {
        Err(RestoreError::Semantic(StateError::Invariant { message })) => assert_eq!(
            message, expected,
            "the fenced row at log sequence {sequence} was refused, but not with the pinned \
             refusal"
        ),
        Err(other) => {
            return Err(format!(
                "the fenced row at log sequence {sequence} was refused by something other than \
                 the pinned refusal `{expected}`: {other:?}"
            )
            .into());
        }
        Ok(_) => {
            return Err(format!(
                "THE PIN FLIPPED: the stored fenced attach at log sequence {sequence} replayed. \
                 Rewrite this test to pin the new behaviour; do not delete it."
            )
            .into());
        }
    }
    let booted = ProductionParticipantHandler::new_restored(
        Arc::clone(&fixture.store),
        marker_fixture_config(),
    )?;
    let unloadable = booted.unloadable_conversations();
    let reason = unloadable
        .get(&fixture.conversation_id)
        .ok_or("the boot did not list the fenced conversation as unloadable")?;
    assert!(
        reason.contains(expected),
        "the boot refused the fenced conversation, but not with `{expected}`: {reason}"
    );
    // DUR-020 (e5c6093d) and f8a21104: the first touch runs the
    // conversation's restore unit, and an unloadable conversation answers it
    // with its typed refusal carrying the cause, so no authority, partial or
    // whole, can be reached through it. An `Ok` cell must still hold none.
    match booted.cell(fixture.conversation_id) {
        Err(crate::server::participant::ParticipantSemanticError::ConversationUnloadable {
            conversation_id,
            reason,
        }) => {
            assert_eq!(
                conversation_id, fixture.conversation_id,
                "the first touch refused another conversation"
            );
            assert!(
                reason.contains(expected),
                "the first touch refused the fenced conversation, but not with `{expected}`: \
                 {reason}"
            );
        }
        Err(error) => {
            return Err(format!(
                "the first touch of the fenced conversation failed with something other than \
                 its typed refusal: {error:?}"
            )
            .into());
        }
        Ok(cell) => {
            let owner = cell
                .lock()
                .map_err(|_| "booted fenced conversation owner lock was poisoned")?;
            let installed = owner.is_some();
            drop(owner);
            assert!(
                !installed,
                "a partial authority was installed for the refused fenced conversation"
            );
        }
    }
    Ok(())
}

/// PINS that on a real log — the rows `attached_marker_fixture` writes
/// (marker M drains while P is BOUND), a live `Detach(P)`, then a raw
/// `Attached(Fenced)` row — cold replay refuses BEFORE the mint, at
/// marker-source validation, with `MarkerBody`.
///
/// The drain wrote the source row's canonical marker bytes with
/// `target_binding: Bound(prior)`; the fenced expectation recomputes
/// `Detached(prior)` (`fenced_marker_source` in liminal-protocol's
/// `claim_frontier.rs`), so `marker_source.rs` `validate_row` refuses
/// `MarkerBody` and `verify_fenced_mode` surfaces it as
/// `StateError::Invariant`. This order never reaches the mint, let alone the
/// frontier transition. The other order CAN be built on a real log (measured,
/// Dean 20:29: the bound sender's first record after a live `Detach(P)`
/// drains M with `target_binding: Detached(prior)`), and it gets past this
/// check to the frontier, where it is refused with `Precedence(FencedRecovery)`
/// after the mint; see
/// `a_stored_fenced_attach_is_refused_on_cold_replay_with_precedence_fenced_recovery_at_the_frontier_transition_after_the_mint`.
/// Work-list row: 2610be6. Live-arm counterpart: `ops_attach.rs:335`.
///
/// ⛔ THIS PIN FLIPS only if marker-source validation stops refusing a
/// source row whose drain-time binding differs from the recomputed
/// `Detached(prior)`; arming the recovery block alone does not flip it. At
/// that point it must be REWRITTEN to pin the new behaviour, never deleted.
#[test]
fn a_stored_fenced_attach_after_a_bound_drain_is_refused_on_cold_replay_with_marker_body_before_the_mint()
-> Result<(), Box<dyn Error>> {
    let fixture = bound_drain_fixture()?;
    if fixture.target.cursor >= fixture.marker_seq {
        return Err(format!(
            "NOT ARMED: P's anchor is acknowledged through the marker (cursor {}, marker {})",
            fixture.target.cursor, fixture.marker_seq
        )
        .into());
    }
    assert_control_replays_clean(&fixture)?;
    append_fenced_row(&fixture)?;
    let expected = format!(
        "fenced attach marker source was refused: {:?}",
        MarkerSourceRefusalReason::MarkerBody
    );
    assert_refused_on_replay_and_boot(&fixture, &expected)
}

/// The most records the bound sender may admit after the live `Detach(P)`
/// before the drain of M to P must have appeared. Measured on Dean at 20:29:
/// the FIRST record after the detach drains it (and the second is refused
/// with `ObserverBackpressure`). The bound guards the loop; it does not pin
/// the index.
const MAX_ADMISSIONS_BEFORE_DRAIN: u8 = 4;

/// One attached member: where it speaks from, who it is, and the capability
/// generation its attach minted.
#[derive(Clone, Copy)]
struct AttachedMember {
    connection: ConnectionIncarnation,
    participant_id: ParticipantId,
    generation: Generation,
}

/// S and P enrolled, attached, and through the transient-peer retirement that
/// `attempt_marker_fixture_with_attaches` drives, on a store this test owns.
struct BoundPair {
    handler: ProductionParticipantHandler,
    store: Arc<dyn DurableStore>,
    conversation_id: u64,
    sender: AttachedMember,
    target: AttachedMember,
}

fn enroll(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<EnrollBound, Box<dyn Error>> {
    let enrolled = dispatch(
        handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(bound) = enrolled else {
        return Err(format!("NOT ARMED: enrolment {token:#x} did not bind: {enrolled:?}").into());
    };
    Ok(bound)
}

/// Enrols and then credential-attaches one member, returning the generation
/// the attach minted so every later request is addressed to it.
fn enroll_and_attach(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<AttachedMember, Box<dyn Error>> {
    let enrolled = enroll(handler, conversation_id, connection, token)?;
    let attached = dispatch(
        handler,
        connection,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id,
            participant_id: enrolled.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: enrolled.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([0xB1; 16]),
            accept_marker_delivery_seq: None,
        }),
    )?;
    let ServerValue::AttachBound(bound) = attached else {
        return Err(format!(
            "NOT ARMED: participant {} did not attach: {attached:?}",
            enrolled.participant_id()
        )
        .into());
    };
    Ok(AttachedMember {
        connection,
        participant_id: enrolled.participant_id(),
        generation: bound.origin_binding_epoch().capability_generation,
    })
}

/// Acks `member` through the final endpoint of its live obligations index,
/// derived as `tests_marker_ack_fixture.rs` `live_ack_endpoint` derives it
/// (production's own derivation at `ops_acks.rs`). Nothing new outstanding
/// means no ack: re-acking the cursor would be refused as a regression.
fn ack_to_endpoint(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    member: AttachedMember,
) -> Result<(), Box<dyn Error>> {
    let (acknowledged_through, endpoint) = {
        let cell = handler.cell(conversation_id)?;
        let owner = cell
            .lock()
            .map_err(|_| "fenced replay fixture owner lock was poisoned")?;
        let authority = owner
            .as_ref()
            .ok_or("fenced replay fixture conversation owner was absent")?;
        let outbox = authority
            .outbox
            .as_ref()
            .ok_or("fenced replay fixture outbox is absent")?;
        let acknowledged_through = authority
            .slots
            .get(&member.participant_id)
            .ok_or("fenced replay fixture ack slot was absent")?
            .member
            .cursor();
        let (_, endpoint) = outbox
            .recipient_ack_obligations(member.participant_id, acknowledged_through)
            .map_err(|error| format!("ack obligations refused: {error:?}"))?;
        drop(owner);
        (acknowledged_through, endpoint)
    };
    if endpoint <= acknowledged_through {
        return Ok(());
    }
    let acked = dispatch(
        handler,
        member.connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id,
            participant_id: member.participant_id,
            capability_generation: member.generation,
            through_seq: endpoint,
        }),
    )?;
    if !matches!(acked, ServerValue::AckCommitted(_)) {
        return Err(format!(
            "NOT ARMED: participant {} did not ack through {endpoint}: {acked:?}",
            member.participant_id
        )
        .into());
    }
    Ok(())
}

/// The transient-peer retirement of `ack_marker_prefix`: a third member
/// enrols, both ack, it leaves, both ack.
fn retire_transient_peer(pair: &BoundPair) -> Result<(), Box<dyn Error>> {
    let handler = &pair.handler;
    let conversation_id = pair.conversation_id;
    let connection = ConnectionIncarnation::new(0xA7, 3);
    let third = enroll(handler, conversation_id, connection, 0xA0)?;
    for member in [pair.sender, pair.target] {
        ack_to_endpoint(handler, conversation_id, member)?;
    }
    let left = dispatch(
        handler,
        connection,
        ClientRequest::Leave(LeaveRequest {
            conversation_id,
            participant_id: third.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: third.attach_secret(),
            leave_attempt_token: LeaveAttemptToken::new([0xA6; 16]),
        }),
    )?;
    if !matches!(left, ServerValue::LeaveCommitted(_)) {
        return Err(format!("NOT ARMED: the transient peer did not leave: {left:?}").into());
    }
    for member in [pair.sender, pair.target] {
        ack_to_endpoint(handler, conversation_id, member)?;
    }
    Ok(())
}

/// S and P bound under `marker_fixture_config`, with the transient peer
/// retired: the arming `attached_marker_fixture` uses up to its drain.
fn bound_pair() -> Result<BoundPair, Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), marker_fixture_config())?;
    let conversation_id = 0xA7;
    let sender = enroll_and_attach(
        &handler,
        conversation_id,
        ConnectionIncarnation::new(0xA7, 1),
        0xA1,
    )?;
    let target = enroll_and_attach(
        &handler,
        conversation_id,
        ConnectionIncarnation::new(0xA7, 2),
        0xA2,
    )?;
    let pair = BoundPair {
        handler,
        store,
        conversation_id,
        sender,
        target,
    };
    retire_transient_peer(&pair)?;
    Ok(pair)
}

/// Every drain row in a range as `(sequence, target, delivery_seq)`, and the
/// first one charged to the target with its sequence.
type DrainScan = (Vec<(u64, u64, u64)>, Option<(u64, StoredMarkerDrain)>);

/// Every `MarkerDrained` row in `[from, durable head)` as `(sequence, target,
/// marker delivery_seq)`, and the first one charged to `target`, if any.
fn drains_since(
    pair: &BoundPair,
    target: ParticipantId,
    from: u64,
) -> Result<DrainScan, Box<dyn Error>> {
    let head = durable_log_head(&pair.store, pair.conversation_id)?;
    if head < from {
        return Err(format!(
            "NOT ARMED: the durable rows end at {head}, below the live head {from} the detach \
             left, so the scan does not read where the commits wrote"
        )
        .into());
    }
    let drains = marker_drains(&pair.store, pair.conversation_id, from, head)?;
    let listed = drains
        .iter()
        .map(|(sequence, row)| {
            (
                *sequence,
                row.retained_charge.participant_id,
                row.retained_charge.delivery_seq,
            )
        })
        .collect();
    let to_target = drains
        .into_iter()
        .find(|(_, row)| row.retained_charge.participant_id == target);
    Ok((listed, to_target))
}

/// S admits ordinary records exactly as the 20:29 arming did: after every
/// committed record S acks itself, and only then is the durable log scanned
/// over `[detach_head, durable head)`. Admissions stop at the first
/// `MarkerDrained` row charged to P, returned with its physical sequence.
///
/// A refused admission is scanned too, since a drain row may be written in
/// the same call that is then refused. A refusal with no drain to P in the
/// range, `ObserverBackpressure` included, means the order under test was
/// never built; the message lists every drain row in the range.
fn admit_until_drain_to_target(
    pair: &BoundPair,
    detach_head: u64,
) -> Result<(u64, StoredMarkerDrain), Box<dyn Error>> {
    let sender = pair.sender;
    let target = pair.target.participant_id;
    for admission in 1..=MAX_ADMISSIONS_BEFORE_DRAIN {
        let outcome = dispatch(
            &pair.handler,
            sender.connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: pair.conversation_id,
                participant_id: sender.participant_id,
                capability_generation: sender.generation,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(
                    [0xC0 | admission; 16],
                ),
                payload: vec![0xC0 | admission],
            }),
        )?;
        let committed = matches!(outcome, ServerValue::RecordCommitted(_));
        if committed {
            ack_to_endpoint(&pair.handler, pair.conversation_id, sender)?;
        }
        let (listed, to_target) = drains_since(pair, target, detach_head)?;
        if let Some(drain) = to_target {
            return Ok(drain);
        }
        if !committed {
            return Err(format!(
                "NOT ARMED: admission {admission} from the bound sender was refused before any \
                 marker drained to detached target {target}: {outcome:?}; MarkerDrained rows \
                 (sequence, target, delivery_seq) since log row {detach_head}: {listed:?}"
            )
            .into());
        }
    }
    Err(format!(
        "NOT ARMED: {MAX_ADMISSIONS_BEFORE_DRAIN} admissions from the bound sender committed and \
         none drained a marker to detached target {target}"
    )
    .into())
}

/// The operation log's DURABLE head, read from the store rather than from the
/// cached owner: the count of committed rows, each checked to sit at its own
/// sequence.
///
/// Rows are read where their commit put them, as
/// `tests_p0_64_live_extension.rs` `base_rows` reads them: the flat rows below
/// the native head, then the routed rows at and after it on the
/// conversation's route. Every live commit, the record admission that stages
/// a `MarkerDrained` row included, is one routed batch
/// (`OperationBatch::commit`), so a scan of the flat stream alone found no
/// row, and no drain, at all.
fn durable_log_head(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let route = checkpoint_conversation_route(conversation_id);
    let mut head = 0_u64;
    let mut phase = OperationSchemaPhase::V2Prefix;
    let mut routed = false;
    loop {
        let page = if routed {
            let Some(page) = block_on(log.read_rebuild_page(&route, head, phase))?? else {
                return Ok(head);
            };
            page
        } else {
            block_on(log.read_bounded_page(head, phase, READ_BATCH_SIZE))??
        };
        phase = page.next_phase;
        if page.rows.is_empty() {
            if routed {
                return Ok(head);
            }
            routed = true;
            continue;
        }
        for row in page.rows {
            if row.sequence != head {
                return Err(
                    format!("operation log expected row {head}, found {}", row.sequence).into(),
                );
            }
            head = head.checked_add(1).ok_or("operation log head overflow")?;
        }
    }
}

/// The drain row's canonical marker bytes must record the target DETACHED at
/// its prior epoch: the binding `fenced_marker_source` recomputes.
fn assert_drained_detached(
    row: &StoredMarkerDrain,
    prior_epoch: BindingEpoch,
) -> Result<(), Box<dyn Error>> {
    let marker = std::str::from_utf8(&row.marker)?;
    let detached = format!(
        "target_binding: {:?},",
        FrontierBinding::Detached(prior_epoch)
    );
    if !marker.contains(&detached) {
        return Err(format!(
            "NOT ARMED: the drain to the target does not record `{detached}`: {marker}"
        )
        .into());
    }
    Ok(())
}

/// Option A's rows: S and P bound, the transient peer retired, the live
/// `Detach(P)`, then S's admissions up to and including the first one that
/// drains M to the detached P.
fn detached_drain_fixture() -> Result<FencedFixture, Box<dyn Error>> {
    let pair = bound_pair()?;
    let conversation_id = pair.conversation_id;
    let participant_id = pair.target.participant_id;
    let (prior_epoch, attach_secret) =
        bound_participant(&pair.handler, conversation_id, participant_id)?;
    detach(&pair.handler, conversation_id, participant_id, prior_epoch)?;
    let detach_head = read_detached_target(
        &pair.handler,
        conversation_id,
        participant_id,
        prior_epoch,
        attach_secret,
    )?
    .fenced_row_sequence;
    let (marker_source, row) = admit_until_drain_to_target(&pair, detach_head)?;
    assert_drained_detached(&row, prior_epoch)?;
    let target = read_detached_target(
        &pair.handler,
        conversation_id,
        participant_id,
        prior_epoch,
        attach_secret,
    )?;
    let store = Arc::clone(&pair.store);
    drop(pair);
    Ok(FencedFixture {
        store,
        conversation_id,
        marker_seq: row.retained_charge.delivery_seq,
        marker_source,
        target,
    })
}

/// PINS that on a real log where marker M drained while P was DETACHED (option
/// A) — S and P bound, the transient peer retired, a live `Detach(P)`, S's
/// first record after it (which drains M with `target_binding:
/// Detached(prior)`), then a raw `Attached(Fenced)` row g — cold replay passes
/// marker-source validation, mints the proof, and is refused at the frontier
/// transition with `Precedence(FencedRecovery)`.
///
/// Measured (Dean, 20:29): a real log CAN commit a drain of a marker to a
/// detached target; the first record after the detach drains it, and only
/// later records are refused with `ObserverBackpressure`. The source row then
/// agrees with `fenced_marker_source`, so the refusal moves past `MarkerBody`
/// to the frontier: `apply_live_fenced_attach` in liminal-protocol's
/// `claim_frontier.rs` refuses with `Precedence(FencedRecovery)` because the
/// sequence and order recovery blocks are never armed outside restore
/// capsules, and `ops_attach.rs` surfaces it as `StateError::Invariant`.
/// Work-list row: 2610be6. Live-arm counterpart: `ops_attach.rs:335`. The
/// bound-drain order is refused earlier, with `MarkerBody`, by
/// `a_stored_fenced_attach_after_a_bound_drain_is_refused_on_cold_replay_with_marker_body_before_the_mint`.
///
/// ⛔ THIS PIN FLIPS once the recovery block is armed. At that point it must
/// be REWRITTEN to pin the new behaviour, never deleted. The refusal of the
/// record after the drain, which bounds how this order is built, is pinned by
/// `the_record_after_the_drain_to_a_detached_target_is_refused_with_observer_backpressure`.
#[test]
fn a_stored_fenced_attach_is_refused_on_cold_replay_with_precedence_fenced_recovery_at_the_frontier_transition_after_the_mint()
-> Result<(), Box<dyn Error>> {
    let fixture = detached_drain_fixture()?;
    if fixture.target.cursor >= fixture.marker_seq {
        return Err(format!(
            "NOT ARMED: P's anchor is acknowledged through the marker (cursor {}, marker {})",
            fixture.target.cursor, fixture.marker_seq
        )
        .into());
    }
    assert_control_replays_clean(&fixture)?;
    append_fenced_row(&fixture)?;
    let expected = format!(
        "attach frontier transition failed: {:?}",
        LiveFrontierError::Precedence(PrecedenceCondition::FencedRecovery)
    );
    assert_refused_on_replay_and_boot(&fixture, &expected)
}

/// PINS the refusal that bounds option A: once S's record has drained M to the
/// detached P, S's next record is refused on the wire with
/// `ObserverBackpressure::RecordAdmission`, because P's unacknowledged marker
/// holds the outbox (measured, Dean 20:29 and 20:41).
///
/// ⛔ THIS PIN FLIPS if a detached target's drained marker stops holding the
/// sender's admissions. The fenced-replay pins above are then built on a log
/// that may carry more drains, so rewrite this pin and recheck them; never
/// delete it.
#[test]
fn the_record_after_the_drain_to_a_detached_target_is_refused_with_observer_backpressure()
-> Result<(), Box<dyn Error>> {
    let pair = bound_pair()?;
    let conversation_id = pair.conversation_id;
    let participant_id = pair.target.participant_id;
    let (prior_epoch, attach_secret) =
        bound_participant(&pair.handler, conversation_id, participant_id)?;
    detach(&pair.handler, conversation_id, participant_id, prior_epoch)?;
    let detach_head = read_detached_target(
        &pair.handler,
        conversation_id,
        participant_id,
        prior_epoch,
        attach_secret,
    )?
    .fenced_row_sequence;
    let (_, drain) = admit_until_drain_to_target(&pair, detach_head)?;
    assert_drained_detached(&drain, prior_epoch)?;
    let sender = pair.sender;
    let outcome = dispatch(
        &pair.handler,
        sender.connection,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id,
            participant_id: sender.participant_id,
            capability_generation: sender.generation,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new([0xEE; 16]),
            payload: vec![0xEE],
        }),
    )?;
    assert!(
        matches!(
            outcome,
            ServerValue::ObserverBackpressure(ObserverBackpressure::RecordAdmission { .. })
        ),
        "the record after the drain to detached P was answered {outcome:?}"
    );
    Ok(())
}