import io, sys

p = 'crates/spt-daemon/tests/wan_reply_bound.rs'
b = open(p, 'rb').read()


def crlf(s):
    # This file's committed blob is pure LF (git normalizes it); the worktree
    # copy is mixed, so anchor and append in LF.
    return s.encode()


def sub(old, new, label):
    global b
    o, n = crlf(old), crlf(new)
    assert b.count(o) == 1, (label, b.count(o))
    b = b.replace(o, n)


sub('''//!    at the call site and a comment there has already failed to stop one wrong
//!    wiring.
''',
    '''//!    at the call site and a comment there has already failed to stop one wrong
//!    wiring.
//! 3. THE FIVE SIBLING VERBS (commit 2) - fork, redeem, answer, knock and
//!    presence run those same two arms against the same silent peer. One cell
//!    each way rather than ten: the RIG is the expensive part and the per-verb
//!    fact is a single assertion, so a table keeps the cost of a sixth verb at
//!    one line instead of one file.
''', 'header')

sub('''use spt_daemon::{request_wan, WanRequestOutcome};
''',
    '''use spt_daemon::{
    request_answer, request_fork, request_knock, request_presence, request_redeem, request_wan,
    AnswerRequestOutcome, ForkRequestOutcome, KnockRequestOutcome, PresenceRequestOutcome,
    RedeemRequestOutcome, WanRequestOutcome,
};
''', 'imports')

TAIL = '''
/// The five sibling verbs, against the same silent peer.
///
/// Each pass answers one question: does THIS verb end its wait with ITS OWN
/// expiry outcome? The `request_wan` arm above proves the shape works; this
/// proves the shape was APPLIED and not merely copied past - a verb whose loop
/// was left unbounded hangs here and takes the pass to the kill bound.
// [int->REQ-WAN-REPLY-BOUND]
// [int->REQ-FORK-CONTROL-SURFACE]
// [int->REQ-KNOCK-VERBS]
// [int->REQ-KNOCK-REDEEM-WIRE]
// [int->REQ-KNOCK-ANSWER-RECEIPT]
// [int->REQ-UNLISTED-PRESENCE-PROBE]
#[test]
fn every_sibling_verb_ends_a_silent_peers_wait_with_its_own_outcome() {
    let _home = init_home();
    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = TempDir::new().unwrap();
    let (a_name, b_name) = (unique_name(), unique_name());
    let _a = net_broker(&a_name, dir.path());
    let _b = net_broker(&b_name, dir.path());

    let mut b_probe = connect_retry(&b_name);
    let b_addr = b_probe.net_status().expect("status").addr;

    // One verb per pass, each on its OWN bounded carrier and its own dial: a
    // shared carrier would let one verb spend the next one's budget, and the
    // elapsed assertion would stop meaning anything.
    for verb in ["fork", "redeem", "answer", "knock", "presence"] {
        let a_name = a_name.clone();
        let (tx, rx) = std::sync::mpsc::channel();
        thread::spawn(move || {
            let mut setup = connect_retry(&a_name);
            let conn = setup
                .net_dial(b_addr, Some(MintedOp::new(Minter::Cli, op())))
                .expect("dial a live-but-unserving peer");
            drop(setup);

            let mut a = connect_retry_pump(&a_name, SHORT_BUDGET);
            let bounded = a.reply_read_deadline().is_some();
            let started = Instant::now();
            // Rendered to ONE word rather than carried as five different types:
            // what this asserts is "its own expiry outcome", and a word says
            // that across five families without five channels.
            let verdict = match verb {
                "fork" => {
                    let rec = spt_net::net::forkmsg::ForkRecord::new("op", "src", "new", "sub");
                    match request_fork(&mut a, conn.conn_id, &rec) {
                        Ok(ForkRequestOutcome::PeerSilent) => "peer-silent".to_string(),
                        other => format!("{other:?}"),
                    }
                }
                "redeem" => {
                    let rec = spt_net::net::redeemmsg::RedeemRecord::new("id", "code", "me", false);
                    match request_redeem(&mut a, conn.conn_id, &rec) {
                        Ok(RedeemRequestOutcome::PeerSilent) => "peer-silent".to_string(),
                        other => format!("{other:?}"),
                    }
                }
                "answer" => {
                    let rec = spt_net::net::answermsg::AnswerRecord::new("knock-id", "me", true);
                    match request_answer(&mut a, conn.conn_id, &rec) {
                        Ok(AnswerRequestOutcome::PeerSilent) => "peer-silent".to_string(),
                        other => format!("{other:?}"),
                    }
                }
                "knock" => {
                    let rec = spt_net::net::knockmsg::KnockRecord::new(
                        "id",
                        "me",
                        "them",
                        vec!["MSG".to_string()],
                        false,
                        false,
                    );
                    match request_knock(&mut a, conn.conn_id, &rec) {
                        Ok(KnockRequestOutcome::PeerSilent) => "peer-silent".to_string(),
                        other => format!("{other:?}"),
                    }
                }
                _ => {
                    let rec = spt_net::net::presencemsg::PresenceRecord::new("id", "me", "them");
                    match request_presence(&mut a, conn.conn_id, &rec) {
                        Ok(PresenceRequestOutcome::PeerSilent) => "peer-silent".to_string(),
                        other => format!("{other:?}"),
                    }
                }
            };
            let _ = tx.send((bounded, started.elapsed(), verdict));
        });

        let (bounded, elapsed, verdict) = rx.recv_timeout(CALLER_TIMEOUT).unwrap_or_else(|_| {
            panic!("WITNESSED PRE-FIX RED: request_{verb} STILL BLOCKED after {CALLER_TIMEOUT:?}")
        });
        assert!(bounded, "PRECONDITION ({verb}): the carrier must be the bounded one");
        assert_eq!(
            verdict, "peer-silent",
            "request_{verb} must report a peer that said NOTHING as its own expiry outcome"
        );
        assert!(
            elapsed >= SHORT_BUDGET / 2,
            "({verb}) the return must be the BUDGET firing, not an instant failure: {elapsed:?}"
        );
        assert!(
            elapsed < CALLER_TIMEOUT,
            "({verb}) and it must END, not merely be reported: {elapsed:?}"
        );
    }
}

/// The enforcement arm for the same five. No peer is needed: the refusal lands
/// before any wire I/O, and it NAMES the verb - an operator with five call
/// sites cannot act on a refusal that will not say which one is mis-wired.
// [int->REQ-WAN-REPLY-BOUND]
#[test]
fn every_sibling_verb_refuses_an_unbounded_carrier_by_name() {
    let _home = init_home();
    let _g = LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = TempDir::new().unwrap();
    let (a_name, b_name) = (unique_name(), unique_name());
    let _a = net_broker(&a_name, dir.path());
    let _b = net_broker(&b_name, dir.path());

    let mut b_probe = connect_retry(&b_name);
    let b_addr = b_probe.net_status().expect("status").addr;

    let mut a = connect_retry(&a_name);
    assert!(
        a.reply_read_deadline().is_none(),
        "PRECONDITION: a cold-start brain is the UNBOUNDED carrier"
    );
    let conn = a
        .net_dial(b_addr, Some(MintedOp::new(Minter::Cli, op())))
        .expect("dial");

    let fork = spt_net::net::forkmsg::ForkRecord::new("op", "src", "new", "sub");
    let redeem = spt_net::net::redeemmsg::RedeemRecord::new("id", "code", "me", false);
    let answer = spt_net::net::answermsg::AnswerRecord::new("knock-id", "me", true);
    let knock = spt_net::net::knockmsg::KnockRecord::new(
        "id",
        "me",
        "them",
        vec!["MSG".to_string()],
        false,
        false,
    );
    let presence = spt_net::net::presencemsg::PresenceRecord::new("id", "me", "them");

    // The word each refusal must carry, beside the constructor that fixes it.
    let refusals = [
        ("fork", request_fork(&mut a, conn.conn_id, &fork).err()),
        (
            "redemption",
            request_redeem(&mut a, conn.conn_id, &redeem).err(),
        ),
        (
            "answer receipt",
            request_answer(&mut a, conn.conn_id, &answer).err(),
        ),
        ("knock", request_knock(&mut a, conn.conn_id, &knock).err()),
        (
            "presence probe",
            request_presence(&mut a, conn.conn_id, &presence).err(),
        ),
    ];
    for (word, err) in refusals {
        let text = err
            .unwrap_or_else(|| panic!("the {word} verb must REFUSE an unbounded carrier"))
            .to_string();
        assert!(
            text.contains("cold_start_pump"),
            "the {word} refusal must name the constructor that fixes it: {text:?}"
        );
        assert!(
            text.contains(word),
            "and it must name the VERB it refused: {text:?}"
        );
    }
}
'''

b = b + crlf(TAIL)
open(p, 'wb').write(b)
print('int cells appended')

# no collapsed continuations, no over-long lines among what we just added
lines = open(p, 'rb').read().split(b'\r\n')
long = [(i + 1, len(l)) for i, l in enumerate(lines) if len(l) > 100]
print('lines over 100 cols:', long)
print('backslash-continuations present:', sum(1 for l in lines if l.rstrip().endswith(bytes([92]))))