frame-conv 0.2.0

//! F-3b R1 — the pub/sub pattern at the pinned substrate: caller-supplied
//! `PublicationId`, substrate fan-out (A5), identity-function admission
//! token (A4), membership-linearized epochs, O(1) dedup.
//!
//! The crash-window and economics halves of R4 (subscriber/publisher kill,
//! gap/resync attack, zero-queue instrumentation, no-poll tripwire) land
//! with the R4 battery; this suite pins the pattern's typed contract.

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

mod support;

use std::error::Error;
use std::fs;
use std::time::Duration;

use frame_conv::{ConversationHandle, ConversationSeq, PublicationId, PublicationItem};
use serde::{Deserialize, Serialize};
use support::{FileStore, RunningServer, attachment, store_dir};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct Note {
    text: String,
    version: u32,
}

/// Positive-observation budget: comfortably above two of the substrate's
/// 5 s IO quanta.
const OBSERVE: Duration = Duration::from_secs(12);

/// Negative-watch budget: one full quantum plus slack — long enough that a
/// wrongly-delivered record would have surfaced.
const QUIET: Duration = Duration::from_secs(6);

/// One observed publication: identity, body, and record position.
type ObservedPublication = (PublicationId, Note, ConversationSeq);

/// Waits until the next PUBLICATION item (skipping lifecycle items, which
/// interleave legitimately), or returns `None` when the budget elapses
/// with no publication.
fn next_publication_only(
    handle: &mut ConversationHandle<FileStore>,
    budget: Duration,
) -> Result<Option<ObservedPublication>, Box<dyn Error>> {
    let started = std::time::Instant::now();
    while started.elapsed() < budget {
        match handle.next_publication::<Note>(budget)? {
            Some(PublicationItem::Publication { id, body, seq, .. }) => {
                return Ok(Some((id, body, seq)));
            }
            Some(PublicationItem::PeerJoined { .. } | PublicationItem::PeerDeparted { .. }) => {}
            Some(other) => {
                return Err(format!("unexpected non-publication item: {other:?}").into());
            }
            None => return Ok(None),
        }
    }
    Ok(None)
}

/// R1 — one admitted publication reaches every ACTIVE epoch exactly once,
/// with identical bytes and identity; the sender is excluded from its own
/// record; no epoch observes it twice.
#[test]
fn one_publication_reaches_every_active_epoch_exactly_once() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-broadcast")?;

    let (mut publisher, publisher_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();
    let (mut observer_a, _grant_a) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("observer-a.lpcr")),
    )?;
    let (mut observer_b, _grant_b) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("observer-b.lpcr")),
    )?;

    let id = PublicationId::from_bytes([3; 16]);
    let body = Note {
        text: "broadcast".to_owned(),
        version: 1,
    };
    let receipt = publisher.publish(id, &body)?;

    for (name, observer) in [("a", &mut observer_a), ("b", &mut observer_b)] {
        let (got_id, got_body, got_seq) = next_publication_only(observer, OBSERVE)?
            .ok_or_else(|| format!("observer {name} never received the publication"))?;
        assert_eq!(got_id, id);
        assert_eq!(got_body, body);
        assert_eq!(
            got_seq, receipt.seq,
            "the delivered record and the admission witness are one record"
        );
        // Exactly once: nothing further arrives for this epoch.
        let extra = next_publication_only(observer, QUIET)?;
        assert!(
            extra.is_none(),
            "observer {name} received a duplicate: {extra:?}"
        );
        assert_eq!(observer.anomaly_counters().duplicate_publications, 0);
    }

    // Sender excluded from its own record (proven S1 ordering rule).
    let own = next_publication_only(&mut publisher, QUIET)?;
    assert!(
        own.is_none(),
        "publisher {:?} received its own publication: {own:?}",
        publisher_grant.participant
    );

    server.shutdown()?;
    Ok(())
}

/// CHARACTERIZATION (observed 2026-07-23 at liminal-server 0.4.1,
/// contradicting amendment A4's factual premise — escalated to the
/// coordination seat the same day): republishing the same id with the
/// SAME bytes on a live binding admits a NEW record. The admission
/// attempt token is per-attempt correlation identity (its own wire doc),
/// NOT a durable idempotency key like Leave's; `AttemptTokenBodyConflict`
/// has no `RecordAdmission` arm at protocol 0.3.2, so an admission body
/// conflict is unrepresentable on the wire. Subscribers observe BOTH
/// records — each exactly once, same id, distinct seqs. This suite pins
/// the substrate truth so any future release that adds admission
/// idempotency trips it loudly.
#[test]
fn same_id_republish_admits_a_new_record_at_the_pinned_substrate() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-republish")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();
    let (mut observer, _observer_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("observer.lpcr")),
    )?;

    let id = PublicationId::from_bytes([4; 16]);
    let body = Note {
        text: "once".to_owned(),
        version: 1,
    };
    let first = publisher.publish(id, &body)?;
    let second = publisher.publish(id, &body)?;
    assert!(
        second.seq > first.seq,
        "observed 0.4.1 truth: the republish admits a NEW record \
         (first {first:?}, second {second:?})"
    );

    let (got_id, _got_body, got_seq) = next_publication_only(&mut observer, OBSERVE)?
        .ok_or("observer never received the first publication")?;
    assert_eq!(got_id, id);
    assert_eq!(got_seq, first.seq);
    let (again_id, _again_body, again_seq) = next_publication_only(&mut observer, OBSERVE)?
        .ok_or("observer never received the second record")?;
    assert_eq!(again_id, id, "same id, distinct record");
    assert_eq!(again_seq, second.seq);

    server.shutdown()?;
    Ok(())
}

/// CHARACTERIZATION (observed 2026-07-23, same escalation as above):
/// reusing an id for DIFFERENT bytes on a live binding ALSO admits — no
/// refusal of any kind exists on this path at the pinned substrate.
#[test]
fn same_id_different_bytes_also_admits_at_the_pinned_substrate() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-conflict")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;

    let id = PublicationId::from_bytes([5; 16]);
    let first = publisher.publish(
        id,
        &Note {
            text: "original".to_owned(),
            version: 1,
        },
    )?;
    let second = publisher.publish(
        id,
        &Note {
            text: "different".to_owned(),
            version: 2,
        },
    )?;
    assert!(
        second.seq > first.seq,
        "observed 0.4.1 truth: different bytes under a reused id admit as \
         a new record (first {first:?}, second {second:?})"
    );

    server.shutdown()?;
    Ok(())
}

/// R1 membership linearization, join half — a publication admitted BEFORE
/// an epoch's join is owed zero times to it (membership-forward), while a
/// publication admitted after the join arrives.
#[test]
fn pre_join_publication_is_absent_for_a_later_epoch() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-prejoin")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();

    let early = PublicationId::from_bytes([6; 16]);
    publisher.publish(
        early,
        &Note {
            text: "before the epoch".to_owned(),
            version: 1,
        },
    )?;

    let (mut late_joiner, _late_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("late.lpcr")),
    )?;
    let lively = PublicationId::from_bytes([7; 16]);
    let receipt = publisher.publish(
        lively,
        &Note {
            text: "within the epoch".to_owned(),
            version: 2,
        },
    )?;

    let (got_id, _body, got_seq) = next_publication_only(&mut late_joiner, OBSERVE)?
        .ok_or("late joiner never received the in-epoch publication")?;
    assert_eq!(
        got_id, lively,
        "the FIRST publication a late joiner observes is the in-epoch one; \
         the pre-join publication is owed zero times"
    );
    assert_eq!(got_seq, receipt.seq);

    server.shutdown()?;
    Ok(())
}

/// R1 membership linearization, leave half — a publication admitted AFTER
/// an epoch's leave is owed zero times to it, while a surviving epoch
/// still receives exactly once.
#[test]
fn post_leave_publication_owes_the_left_epoch_nothing() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-postleave")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();
    let (mut leaver, _leaver_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("leaver.lpcr")),
    )?;
    let (mut survivor, _survivor_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("survivor.lpcr")),
    )?;

    let outcome = leaver.leave()?;
    assert!(
        matches!(outcome, frame_conv::LeaveOutcome::Left { .. }),
        "leave must commit: {outcome:?}"
    );

    let id = PublicationId::from_bytes([8; 16]);
    let receipt = publisher.publish(
        id,
        &Note {
            text: "after the leave".to_owned(),
            version: 1,
        },
    )?;

    let (got_id, _body, got_seq) = next_publication_only(&mut survivor, OBSERVE)?
        .ok_or("surviving epoch never received the publication")?;
    assert_eq!(got_id, id);
    assert_eq!(got_seq, receipt.seq);

    let leaked = next_publication_only(&mut leaver, QUIET)?;
    assert!(
        leaked.is_none(),
        "the left epoch is owed nothing: {leaked:?}"
    );

    server.shutdown()?;
    Ok(())
}

/// R1 zero-subscriber economics, contract half — publish with no live
/// epochs is validation plus ONE admission outcome; the receipt is the
/// entire observable effect. (The instrumentation half — no queue, worker,
/// send, or timer — lands with the R4 battery.)
#[test]
fn zero_subscriber_publish_is_validation_plus_one_admission() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-zerosub")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let receipt = publisher.publish(
        PublicationId::from_bytes([9; 16]),
        &Note {
            text: "into the void".to_owned(),
            version: 1,
        },
    )?;
    assert!(receipt.seq.value() > 0);

    // Idle witness: nothing at the publisher moved besides the admission —
    // no queued item (its own record is never delivered back), no counter,
    // no anomaly. The zero-worker half is structural
    // (tests/no_poll_tripwire.rs: no thread::spawn in any production
    // source).
    let idle = publisher.next_publication::<Note>(Duration::from_millis(10))?;
    assert!(idle.is_none(), "zero-subscriber publish retained: {idle:?}");
    let counters = publisher.anomaly_counters();
    assert_eq!(counters.duplicate_publications, 0);
    assert_eq!(counters.gaps, 0);
    assert_eq!(counters.unexpected_frames, 0);

    server.shutdown()?;
    Ok(())
}

/// Builds the resume-hole scenario (the recovery probe's measured truth
/// at the FRAME surface): victim commits a cursor, optionally lets an
/// UNACKED foreign record land inside the window, publishes its own
/// record, dies; the witness places the far edge; the victim resumes.
/// Returns (resumed handle, own receipt seq, far receipt seq).
fn resume_over_own_hole(
    server: &RunningServer,
    stores: &std::path::Path,
    mid_window_foreign: bool,
) -> Result<
    (
        ConversationHandle<FileStore>,
        ConversationSeq,
        ConversationSeq,
    ),
    Box<dyn Error>,
> {
    let (mut witness, _w_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("witness.lpcr")),
    )?;
    let conversation = witness.conversation();
    let (mut victim, victim_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("victim.lpcr")),
    )?;

    witness.publish(
        PublicationId::from_bytes([21; 16]),
        &Note {
            text: "pre".to_owned(),
            version: 1,
        },
    )?;
    let (_, _, pre_seq) = next_publication_only(&mut victim, OBSERVE)?
        .ok_or("victim never received the pre-record")?;
    victim.commit_cursor(pre_seq)?;
    if mid_window_foreign {
        // An unacked foreign record between the cursor and the own record:
        // replay will DELIVER this one, so the hole sits mid-window.
        witness.publish(
            PublicationId::from_bytes([24; 16]),
            &Note {
                text: "mid".to_owned(),
                version: 1,
            },
        )?;
    }
    let own = victim.publish(
        PublicationId::from_bytes([22; 16]),
        &Note {
            text: "own".to_owned(),
            version: 1,
        },
    )?;
    let victim_state = fs::read(stores.join("victim.lpcr"))?;
    drop(victim);

    let far = witness.publish(
        PublicationId::from_bytes([23; 16]),
        &Note {
            text: "far".to_owned(),
            version: 1,
        },
    )?;

    let (resumed, _rotated) = ConversationHandle::resume(
        &attachment(server.endpoint()),
        &victim_grant,
        &victim_state,
        FileStore::new(stores.join("resumed.lpcr")),
    )?;
    Ok((resumed, own.seq, far.seq))
}

/// R4 gap/resync row, mid-window placement (probe cell B at the FRAME
/// surface): the substrate has no in-protocol drop boundary (S4 reality —
/// `pattern_backpressure.rs`), so the one real gap source is resume-replay
/// excluding the resumed participant's own record and its old binding's
/// death record. With a delivered foreign record on the near side, the
/// authorless hole IS accountable: the typed `Gap` is presented BEFORE the
/// post-gap item — no post-gap item is ever presented as contiguous.
#[test]
fn mid_window_resume_hole_surfaces_typed_gap_before_any_post_gap_item() -> Result<(), Box<dyn Error>>
{
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-gap-mid")?;
    let (mut resumed, own_seq, far_seq) = resume_over_own_hole(&server, &stores, true)?;

    // Replay first presents the delivered mid-window foreign record...
    let first = next_publication_only(&mut resumed, OBSERVE)?
        .ok_or("resumed victim never received the mid-window record")?;
    assert_eq!(first.0, PublicationId::from_bytes([24; 16]));
    // ...then the typed Gap, BEFORE the far edge (claim suspension
    // precedes every post-gap item).
    let second = resumed
        .next_publication::<Note>(OBSERVE)?
        .ok_or("the gap must follow the mid-window record")?;
    let PublicationItem::Gap { expected, observed } = second else {
        return Err(format!(
            "the item after the near neighbor must be the typed Gap; observed: {second:?}"
        )
        .into());
    };
    assert_eq!(
        expected.value(),
        own_seq.value(),
        "the hole begins at the excluded own record"
    );
    assert_eq!(
        observed, far_seq,
        "the far edge is the first position replay presents past the hole"
    );
    let third =
        next_publication_only(&mut resumed, OBSERVE)?.ok_or("the far edge must follow its gap")?;
    assert_eq!(third.0, PublicationId::from_bytes([23; 16]));
    assert_eq!(third.2, far_seq);
    assert_eq!(resumed.anomaly_counters().gaps, 1);

    server.shutdown()?;
    Ok(())
}

/// R4 gap/resync row, first-past-cursor placement (probe cell A at the
/// FRAME surface) — THE SWALLOW, pinned deliberately: when the excluded
/// own record is the first position past the committed cursor, baseline
/// seeding absorbs the hole and NO gap is minted; the resumed victim sees
/// the far edge as its contiguous first item. This is exactly why the R4
/// kill-during-submission contract is typed-possible-duplicate (the brief's
/// A4-supersession probe findings): a publisher crashed mid-admission can
/// resume and observe NOTHING that distinguishes its record from
/// never-admitted. Any future substrate that starts replaying own records
/// trips this pin loudly.
#[test]
fn first_past_cursor_resume_hole_is_swallowed_by_baseline_seeding() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-gap-first")?;
    let (mut resumed, _own_seq, far_seq) = resume_over_own_hole(&server, &stores, false)?;

    let first =
        next_publication_only(&mut resumed, OBSERVE)?.ok_or("resumed victim observed nothing")?;
    assert_eq!(
        first.0,
        PublicationId::from_bytes([23; 16]),
        "the far edge arrives as the FIRST item — the own-record hole is invisible"
    );
    assert_eq!(first.2, far_seq);
    assert_eq!(
        resumed.anomaly_counters().gaps,
        0,
        "no gap is minted for the swallowed first-past-cursor hole"
    );

    server.shutdown()?;
    Ok(())
}

/// R4 slow-subscriber row (the F-3a incident-shape port): a consumer
/// draining under SHORT caller wait quanta receives every publication
/// exactly once — changing only the quantum changes no protocol outcome
/// (assertion-8 at this pattern's surface).
#[test]
fn short_quantum_drain_is_complete_and_exactly_once() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("pubsub-quantum")?;

    let (mut publisher, _grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();
    let (mut consumer, _c_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("consumer.lpcr")),
    )?;

    let mut expected = Vec::new();
    for version in 1..=4_u32 {
        let mut bytes = [30; 16];
        bytes[15] = u8::try_from(version)?;
        let id = PublicationId::from_bytes(bytes);
        let receipt = publisher.publish(
            id,
            &Note {
                text: format!("q-{version}"),
                version,
            },
        )?;
        expected.push((id, receipt.seq));
    }

    // Drain under a 100ms wait quantum — far below the substrate's 5s IO
    // quantum; the elapsed short waits are benign quiet, never outcomes.
    let mut drained = Vec::new();
    let until = std::time::Instant::now() + OBSERVE;
    while drained.len() < expected.len() && std::time::Instant::now() < until {
        match consumer.next_publication::<Note>(Duration::from_millis(100))? {
            Some(PublicationItem::Publication { id, seq, .. }) => drained.push((id, seq)),
            Some(PublicationItem::PeerJoined { .. }) | None => {}
            Some(other) => {
                return Err(format!("unexpected item during drain: {other:?}").into());
            }
        }
    }
    assert_eq!(
        drained, expected,
        "the short-quantum drain is complete, ordered, exactly once"
    );
    assert_eq!(consumer.anomaly_counters().duplicate_publications, 0);

    server.shutdown()?;
    Ok(())
}