liminal-protocol 0.9.2

//! Marker-anchor ledger reconcile: the conversation-6 residue and the
//! 2026-08-28 field incident, executed.
//!
//! An anchor is DERIVED only while its marker record survives retention AND its
//! owning participant is still active with a cursor short of the marker. The
//! STORED side lives in closure accounting and is retired only by an
//! acknowledgement (or, since `67d780e`, an ordinary ack crossing the marker).
//! Participant erasure or record retirement therefore zeroes the derived side
//! with NO log row left that could ever retire the stored side — and because
//! cold replay rebuilds state by re-running rows, every boot rebuilds the same
//! split and every admission faults on the `MarkerAnchorAccounting`
//! cross-check, forever. That is the 2026-08-07 conversation-6 wedge: 36
//! continuing warns across a boot that healed its crossing-ack sibling.
//!
//! ⛔ The OPPOSITE split — derived AHEAD of stored — was pinned here as an
//! impossible history that reconciliation must never touch. The field proved
//! that claim false. On Tom's estate, conversation 1 / `delivery_seq` 187 has sat
//! in `{ derived: 1, stored: 0 }` since 2026-08-08, refusing every admission
//! with `Projection(MarkerAnchorAccounting { derived: 1, stored: 0 })`, and the
//! replay-retry path is what mints it: the replay census under-derives at the
//! committed row, the retry retires stored to zero, the row commits, and later
//! rows restore the participant with a cursor short of the marker. Nothing in
//! the log can ever put that anchor back. So the reconcile is now
//! CENSUS-AUTHORITATIVE in both directions.
//!
//! These units pin its four obligations: retire a stored orphan, stay
//! idempotent, leave an in-step ledger alone, and re-mint a census the
//! accounting lost — with the arithmetic bound (`ClosureAccounting::try_new`)
//! refusing a repair it cannot hold and leaving state unchanged.

#![allow(clippy::expect_used, clippy::panic)]

use alloc::{vec, vec::Vec};

use crate::algebra::{ResourceVector, WideResourceVector};
use crate::outcome::CandidatePhase;
use crate::wire::{
    AttachSecret, BindingEpoch, ConnectionIncarnation, Generation, RecordAdmission,
    RecordAdmissionAttemptToken,
};

use super::{
    super::{
        ActiveBinding, AdmissionOrder, BindingState, CapacityCounter, ClosureAccounting,
        ClosureState, ConnectionConversationTracking, EnrollmentFingerprint, LiveMember,
        LiveMemberRestore, OrderClaims, OrderHigh, OrderLedger, PresentedIdentity,
        RecoverySequenceReserve, SequenceClaims, SequenceLedger,
        claim_frontier::{
            BindingTerminalOwner, ClaimFrontiers, ClaimFrontiersRestore, FrontierBinding,
            FrontierParticipant, MarkerProvenance, MovableOrderClaim, MovableSequenceClaim,
            OrderClaimFrontierRestore, OrderDirectOwner, RetainedCausalRecord,
            RetainedCausalRecordKind, SequenceClaimFrontierRestore, SequenceDirectOwner,
            SequenceProductRangesRestore, TerminalProductRangeRestore,
        },
    },
    OrdinaryProjectionError, OrdinaryProjectionLimits, RetainedRecordCharge,
    live_frontier::{LiveFrontierOwner, MarkerAnchorReconcile},
    record_admission::{
        RecordAdmissionDecision, RecordAdmissionFault, RecordAdmissionPrestate,
        apply_record_admission,
    },
};

const CONVERSATION_ID: u64 = 6;
const P0: u64 = 0;
const MARKER_SEQ: u64 = 12;

fn epoch() -> BindingEpoch {
    BindingEpoch::new(
        ConnectionIncarnation::new(6, 0),
        Generation::new(1).expect("test generation is nonzero"),
    )
}

/// One active participant, one retained compaction marker at `MARKER_SEQ`, the
/// participant's cursor and the stored anchor count parameterized — so a
/// single validated shape can hold the marker as unaccepted (cursor short of
/// the marker), accepted (cursor at it), or orphaned (accepted by cursor,
/// never retired from accounting).
fn owner_with(cursor: u64, stored_anchors: u64) -> LiveFrontierOwner {
    owner_with_ledgers(cursor, stored_anchors, stored_anchors)
}

/// The same shape with the two marker ledgers separated, because the field
/// incident separates them: conversation 1 kept its marker CREDIT (the record
/// survives retention and its owner is active) and lost only the ANCHOR. A
/// fixture that ties the two together cannot express the state that wedged the
/// estate, and cannot exercise the arithmetic bound either — a re-mint is
/// admissible exactly while `anchors <= credits`.
fn owner_with_ledgers(cursor: u64, stored_credits: u64, stored_anchors: u64) -> LiveFrontierOwner {
    let binding_epoch = epoch();
    let terminal_owner = BindingTerminalOwner {
        participant_index: P0,
        binding_epoch,
    };
    let exit_seq = MARKER_SEQ + 1;
    let terminal_seq = exit_seq + 1;
    let product_seq = terminal_seq + 1;
    let sequence = SequenceLedger::try_new(
        MARKER_SEQ,
        SequenceClaims::new(1, 1, 0, RecoverySequenceReserve::None),
    )
    .expect("reconcile fixture sequence ledger is valid");
    let order = OrderLedger::try_new(
        OrderHigh::Allocated(0),
        OrderClaims::new(1, 1, false, false).expect("reconcile fixture order claims are valid"),
    )
    .expect("reconcile fixture order ledger is valid");
    let frontiers = ClaimFrontiers::restore(
        ClaimFrontiersRestore {
            conversation_id: CONVERSATION_ID,
            active_identities: vec![FrontierParticipant::new(
                P0,
                cursor,
                FrontierBinding::Bound(binding_epoch),
            )],
            identity_slot_limit: 1,
            retained_floor: u128::from(MARKER_SEQ),
            retained_record_limit: 1,
            retained_records: vec![RetainedCausalRecord {
                delivery_seq: MARKER_SEQ,
                admission_order: AdmissionOrder::new(0, CandidatePhase::CompactionMarker, P0),
                kind: RetainedCausalRecordKind::CompactionMarker {
                    participant_index: P0,
                    provenance: MarkerProvenance::NonProductM,
                },
            }],
            active_marker_anchors: vec![MARKER_SEQ],
            historical_marker_deliveries: vec![],
            historical_causal_facts: vec![],
            sequence: SequenceClaimFrontierRestore {
                movable_claims: vec![
                    MovableSequenceClaim {
                        delivery_seq: exit_seq,
                        owner: SequenceDirectOwner::MembershipExit {
                            participant_index: P0,
                        },
                    },
                    MovableSequenceClaim {
                        delivery_seq: terminal_seq,
                        owner: SequenceDirectOwner::BindingTerminal(terminal_owner),
                    },
                ],
                immutable_candidates: vec![],
                products: SequenceProductRangesRestore {
                    live_times_terminal: vec![TerminalProductRangeRestore {
                        start: product_seq,
                        length: 1,
                        terminal: terminal_owner,
                    }],
                    ..SequenceProductRangesRestore::default()
                },
                recovery: None,
            },
            order: OrderClaimFrontierRestore {
                movable_claims: vec![
                    MovableOrderClaim {
                        transaction_order: 1,
                        owner: OrderDirectOwner::ActiveBindingTerminal(terminal_owner),
                    },
                    MovableOrderClaim {
                        transaction_order: 2,
                        owner: OrderDirectOwner::MembershipExit {
                            participant_index: P0,
                        },
                    },
                ],
                immutable_candidates: vec![],
                recovery: None,
            },
            recovery_marker_delivery_seq: None,
        },
        sequence,
        order,
    )
    .expect("reconcile fixture restores from the canonical cold shape");
    let accounting = ClosureAccounting::try_new(
        ClosureState::Clear,
        stored_credits,
        stored_anchors,
        0,
        0,
        ResourceVector::default(),
        WideResourceVector::default(),
        ResourceVector::new(16, 1024),
        0,
        2,
    )
    .expect("reconcile fixture accounting is valid");
    LiveFrontierOwner::from_test_parts(frontiers, accounting, vec![], 1)
}

#[test]
fn reconcile_retires_the_orphaned_anchor_and_is_idempotent() {
    // Cursor AT the marker: the derived census counts nothing, yet the stored
    // accounting still carries the anchor — the wedge state as it survives in
    // durable rows, where no replayed ack can ever retire it.
    let mut owner = owner_with(MARKER_SEQ, 1);
    assert_eq!(owner.frontiers().unaccepted_marker_anchor_count(), 0);
    assert_eq!(owner.closure_accounting().marker_anchors(), 1);

    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::RetiredOrphans(1)
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 0);

    // In-step now; a second reconcile must retire nothing.
    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::InStep
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 0);
}

#[test]
fn reconcile_leaves_an_in_step_unaccepted_marker_alone() {
    // Cursor short of the marker: derived == stored == 1 — a live, healthy
    // delivered marker awaiting acceptance. Nothing may move.
    let mut owner = owner_with(MARKER_SEQ - 1, 1);
    assert_eq!(owner.frontiers().unaccepted_marker_anchor_count(), 1);

    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::InStep
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 1);
}

/// ⛔ THE FLIPPED PIN. This unit stood as
/// `reconcile_never_touches_the_derived_ahead_direction` and asserted the
/// derived-ahead split was "a corruption to fault on loudly, not ours to paper
/// over". The 2026-08-28 field incident is the history that proved the
/// direction reachable and the pin wrong: conversation 1 on Tom's estate has
/// held `{ derived: 1, stored: 0 }` since 2026-08-08, minted by this crate's
/// own one-way replay retry, refusing every admission and answering none. The
/// census is the authority; a stored ledger that has lost an anchor the census
/// still derives is repaired here, in the same shape and under the same
/// arithmetic bound as the retirement direction.
#[test]
fn reconcile_reminds_the_derived_ahead_direction() {
    // Derived 1, stored 0, credit intact: the exact field shape.
    let mut owner = owner_with_ledgers(MARKER_SEQ - 1, 1, 0);
    assert_eq!(owner.frontiers().unaccepted_marker_anchor_count(), 1);
    assert_eq!(owner.closure_accounting().marker_capacity_credits(), 1);
    assert_eq!(owner.closure_accounting().marker_anchors(), 0);

    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::RemintedDeficit(1)
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 1);

    // In step now; a second reconcile must mint nothing.
    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::InStep
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 1);
}

/// The re-mint is bounded by the same arithmetic that bounds the retirement:
/// `ClosureAccounting::try_new` refuses anchors above credits, and a refused
/// repair leaves the state exactly as it found it. Without a credit there is
/// no slot the anchor could belong to, so the split is a different corruption
/// and this reconcile declines to invent one.
#[test]
fn reconcile_refuses_a_remint_the_accounting_arithmetic_cannot_hold() {
    let mut owner = owner_with_ledgers(MARKER_SEQ - 1, 0, 0);
    assert_eq!(owner.frontiers().unaccepted_marker_anchor_count(), 1);
    assert_eq!(owner.closure_accounting().marker_capacity_credits(), 0);

    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::Refused {
            census: 1,
            stored: 0
        }
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 0);
}

// ---------------------------------------------------------------------------
// The wedge, end to end: a real admission refused before the heal and
// committed after it.
// ---------------------------------------------------------------------------

const UNIT: u64 = 10;
const WEDGE_MARKER_SEQ: u64 = 1;
const WEDGE_ORDINARY_SEQ: u64 = 2;

type Fingerprint = [u8; 32];

fn wedge_records() -> Vec<RetainedCausalRecord> {
    vec![
        RetainedCausalRecord {
            delivery_seq: WEDGE_MARKER_SEQ,
            admission_order: AdmissionOrder::new(0, CandidatePhase::CompactionMarker, P0),
            kind: RetainedCausalRecordKind::CompactionMarker {
                participant_index: P0,
                provenance: MarkerProvenance::NonProductM,
            },
        },
        RetainedCausalRecord {
            delivery_seq: WEDGE_ORDINARY_SEQ,
            admission_order: AdmissionOrder::new(1, CandidatePhase::OrdinaryRecord, P0),
            kind: RetainedCausalRecordKind::OrdinaryRecord {
                participant_index: P0,
            },
        },
    ]
}

fn wedge_charges() -> Vec<RetainedRecordCharge> {
    wedge_records()
        .iter()
        .map(|record| {
            RetainedRecordCharge::new(
                record.delivery_seq,
                record.admission_order,
                ResourceVector::new(1, UNIT),
            )
        })
        .collect()
}

/// The field shape at the frontier: one retained compaction marker the active
/// participant has NOT yet crossed (cursor 0, marker at 1), so the admission
/// projection derives exactly one unaccepted anchor.
fn wedge_frontiers(binding_epoch: BindingEpoch) -> ClaimFrontiers {
    let terminal = BindingTerminalOwner {
        participant_index: P0,
        binding_epoch,
    };
    ClaimFrontiers::restore(
        ClaimFrontiersRestore {
            conversation_id: CONVERSATION_ID,
            active_identities: vec![FrontierParticipant::new(
                P0,
                0,
                FrontierBinding::Bound(binding_epoch),
            )],
            identity_slot_limit: 1,
            retained_floor: u128::from(WEDGE_MARKER_SEQ),
            retained_record_limit: 8,
            retained_records: wedge_records(),
            active_marker_anchors: vec![WEDGE_MARKER_SEQ],
            historical_marker_deliveries: vec![],
            historical_causal_facts: vec![],
            sequence: SequenceClaimFrontierRestore {
                movable_claims: vec![
                    MovableSequenceClaim {
                        delivery_seq: 3,
                        owner: SequenceDirectOwner::MembershipExit {
                            participant_index: P0,
                        },
                    },
                    MovableSequenceClaim {
                        delivery_seq: 4,
                        owner: SequenceDirectOwner::BindingTerminal(terminal),
                    },
                ],
                immutable_candidates: vec![],
                products: SequenceProductRangesRestore {
                    live_times_terminal: vec![TerminalProductRangeRestore {
                        start: 5,
                        length: 1,
                        terminal,
                    }],
                    ..SequenceProductRangesRestore::default()
                },
                recovery: None,
            },
            order: OrderClaimFrontierRestore {
                movable_claims: vec![
                    MovableOrderClaim {
                        transaction_order: 2,
                        owner: OrderDirectOwner::ActiveBindingTerminal(terminal),
                    },
                    MovableOrderClaim {
                        transaction_order: 3,
                        owner: OrderDirectOwner::MembershipExit {
                            participant_index: P0,
                        },
                    },
                ],
                immutable_candidates: vec![],
                recovery: None,
            },
            recovery_marker_delivery_seq: None,
        },
        SequenceLedger::try_new(
            WEDGE_ORDINARY_SEQ,
            SequenceClaims::new(1, 1, 0, RecoverySequenceReserve::None),
        )
        .expect("wedge fixture sequence ledger is valid"),
        OrderLedger::try_new(
            OrderHigh::Allocated(1),
            OrderClaims::new(1, 1, false, false).expect("wedge fixture order claims are valid"),
        )
        .expect("wedge fixture order ledger is valid"),
    )
    .expect("wedge fixture restores from the canonical cold shape")
}

fn wedge_accounting(anchors: u64) -> ClosureAccounting {
    ClosureAccounting::try_new(
        ClosureState::Clear,
        1,
        anchors,
        0,
        0,
        ResourceVector::default(),
        WideResourceVector::new(2, u128::from(2 * UNIT)),
        ResourceVector::new(12, 12 * UNIT),
        0,
        2,
    )
    .expect("wedge fixture accounting is valid")
}

fn wedge_member(generation: Generation) -> LiveMember<Fingerprint> {
    LiveMember::restore(LiveMemberRestore {
        participant_id: P0,
        conversation_id: CONVERSATION_ID,
        generation,
        attach_secret: AttachSecret::new([0xA5; 32]),
        cursor: 0,
        enrollment_fingerprint: EnrollmentFingerprint::new([0xE1; 32]),
        latest_terminal: None,
    })
    .expect("wedge fixture member has consistent identity history")
}

fn wedge_request(generation: Generation) -> RecordAdmission {
    RecordAdmission {
        conversation_id: CONVERSATION_ID,
        participant_id: P0,
        capability_generation: generation,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([0xA7; 16]),
        payload: vec![0xD2; 4],
    }
}

fn wedge_limits() -> OrdinaryProjectionLimits {
    OrdinaryProjectionLimits::new(
        ResourceVector::new(1, UNIT),
        ResourceVector::new(2, 2 * UNIT),
        ResourceVector::new(2, 2 * UNIT),
    )
}

/// Runs one ordinary admission over the wedge shape at the supplied stored
/// anchor count and reports what the total selector decided.
fn wedge_admission(stored_anchors: u64) -> Result<(), (u64, u64)> {
    let generation = Generation::new(1).expect("test generation is nonzero");
    let binding_epoch = BindingEpoch::new(ConnectionIncarnation::new(6, 0), generation);
    let member = wedge_member(generation);
    let binding = BindingState::Bound(ActiveBinding {
        participant_id: P0,
        conversation_id: CONVERSATION_ID,
        binding_epoch,
    });
    let prestate = RecordAdmissionPrestate::new(
        wedge_request(generation),
        PresentedIdentity::<Fingerprint, Fingerprint, Fingerprint>::Live(&member),
        &binding,
        binding_epoch,
        ConnectionConversationTracking::AlreadyTracked,
        CapacityCounter::try_new(4, 1).expect("wedge fixture connection capacity is valid"),
        wedge_accounting(stored_anchors),
        ResourceVector::new(1, UNIT),
        wedge_frontiers(binding_epoch),
        wedge_charges(),
        2,
        wedge_limits(),
    );
    match apply_record_admission(prestate, ResourceVector::new(1, UNIT)) {
        RecordAdmissionDecision::Commit(_) => Ok(()),
        RecordAdmissionDecision::Fault(failure) => match failure.fault() {
            RecordAdmissionFault::Projection(OrdinaryProjectionError::MarkerAnchorAccounting {
                derived,
                stored,
            }) => Err((*derived, *stored)),
            other => panic!("wedge fixture faulted on an unrelated reason: {other:?}"),
        },
        other => panic!("wedge fixture did not reach a commit-or-fault decision: {other:?}"),
    }
}

/// The incident, end to end. The stored ledger has lost the anchor the census
/// still derives; every admission faults on the cross-check
/// (`ordinary_record_projection.rs:1001` @ `b4ef9fb`). Reconciling toward the
/// census re-mints it, and the SAME admission then commits — which is the whole
/// claim the heal makes, and the reason the estate settles.
///
/// The credit ledger is the safety proof, not a coincidence: the credit
/// cross-check at `ordinary_record_projection.rs:983` @ `b4ef9fb` runs BEFORE
/// the anchor one, and the anchor list is validated as a duplicate-free subset
/// of the credits at `:992-998`. So any state that reaches the anchor fault has
/// already proven `derived_anchors <= derived_credits == stored_credits`, and
/// the re-mint can never push anchors past credits.
#[test]
fn the_wedged_admission_commits_once_the_reconcile_reminds_the_lost_anchor() {
    // Red pole: the field state refuses.
    assert_eq!(wedge_admission(0), Err((1, 0)));

    // The heal, on exactly the same shape.
    let binding_epoch = BindingEpoch::new(
        ConnectionIncarnation::new(6, 0),
        Generation::new(1).expect("test generation is nonzero"),
    );
    let mut owner = LiveFrontierOwner::from_test_parts(
        wedge_frontiers(binding_epoch),
        wedge_accounting(0),
        wedge_charges(),
        8,
    );
    assert_eq!(
        owner.reconcile_marker_anchor_ledgers(),
        MarkerAnchorReconcile::RemintedDeficit(1)
    );
    assert_eq!(owner.closure_accounting().marker_anchors(), 1);

    // Green pole: the healed ledger admits.
    assert_eq!(
        wedge_admission(owner.closure_accounting().marker_anchors()),
        Ok(())
    );
}