import io, os

SP = os.path.dirname(os.path.abspath(__file__))


def load(p):
    raw = io.open(p, encoding='utf-8', newline='').read()
    return raw.replace('\r\n', '\n'), '\r\n' in raw


def save(p, s, crlf):
    if crlf:
        s = s.replace('\n', '\r\n')
    io.open(p, 'w', encoding='utf-8', newline='').write(s)


def rep(s, old, new):
    assert s.count(old) == 1, (old[:90], s.count(old))
    return s.replace(old, new)


# ---- wansend.rs tests
p = 'crates/spt/src/wansend.rs'
s, c = load(p)
block = io.open(os.path.join(SP, 'w3_tests_wansend.rs'), encoding='utf-8').read().replace('\r\n', '\n')
anchor = '''    // [unit->REQ-SEND-RESOLVES-BEFORE-LOCAL] FAIL-OPEN (condition 3): with no
    // registry answer'''
s = rep(s, anchor, block.lstrip('\n') + '\n' + anchor)
save(p, s, c)

# ---- resting.rs tests
p = 'crates/spt-daemon/src/resting.rs'
s, c = load(p)
anchor = '''    // [unit->REQ-EFFECTIVE-INSTANCE-STATE] [unit->REQ-ACTIVATION-COUNTER]
    // (doyle pre-gate #2 + #6) the sibling demotion reads the EFFECTIVE state'''
new = '''    // [unit->REQ-INSTANCE-HANDOFF] [unit->REQ-ACTIVATION-TRIGGERS] a SAME-id
    // message carrying `--handoff` is a Wake (trigger 3), where without the flag
    // it is a sibling's normal message. A different id steals with or without
    // the flag, so the flag grants nothing new. The window and empty-sender
    // rules are the plain classifier's.
    #[test]
    fn handoff_trigger_wakes_a_same_id_sibling() {
        use spt_store::spool::{WINDOW_ACTIVE_ONLY, WINDOW_DEFAULT};
        assert_eq!(handoff_trigger("ling", "ling", WINDOW_DEFAULT, true), Some(RestEvent::Wake));
        assert_eq!(
            handoff_trigger("ling", "ling", WINDOW_DEFAULT, false),
            Some(RestEvent::SiblingMessage)
        );
        assert_eq!(handoff_trigger("ling", "doyle", WINDOW_DEFAULT, true), Some(RestEvent::Wake));
        assert_eq!(handoff_trigger("ling", "doyle", WINDOW_DEFAULT, false), Some(RestEvent::Wake));
        assert_eq!(handoff_trigger("ling", "ling", WINDOW_ACTIVE_ONLY, true), None);
        assert_eq!(handoff_trigger("ling", "", WINDOW_DEFAULT, true), None);
    }

    // [unit->REQ-INSTANCE-HANDOFF] what the receiver reports as TAKEN: the
    // stored intent is active. A suspended sibling's wake writes that intent
    // at once and resumes later (ADR-0033), so the intent is the fact the
    // handoff made. A resting intent, or no record at all, is not taken.
    #[test]
    fn handoff_taken_reads_the_intent_the_wake_wrote() {
        crate::test_home::with_home(|_| {
            let p = spt_store::perch::resolve_perch_path("taker", spt_store::perch::ParentHint::Infer);
            assert!(!handoff_taken(&p), "no record");
            std::fs::create_dir_all(&p).unwrap();
            let mut rec = info::InfoJson::new("taker", "t", 4_000_000_000, "sid", "live_agent");
            rec.status = Some(spt_store::liveness::STATUS_OFFLINE.to_string());
            info::write_info(&p, &rec).unwrap();
            write_rest(&p, RestState::Suspended, 1).unwrap();
            assert!(!handoff_taken(&p));
            write_rest(&p, RestState::Dormant, 1).unwrap();
            assert!(!handoff_taken(&p));
            write_rest(&p, RestState::Active, 1).unwrap();
            assert!(handoff_taken(&p), "a cold instance woken active is taken before it resumes");
        });
    }

'''
s = rep(s, anchor, new + anchor)
save(p, s, c)

# ---- wan.rs tests
p = 'crates/spt-daemon/src/wan.rs'
s, c = load(p)
anchor = '''    /// The spooled rows as `(from, body)` — what a drain will hand the agent.
    fn spooled_rows(perch: &Path) -> Vec<(String, String)> {'''
new = '''    // [unit->REQ-INSTANCE-HANDOFF] the RECEIVER decides a handoff (doyle
    // condition 3). A same-id handoff wakes a DORMANT or SUSPENDED sibling
    // ACTIVE and answers `handoff`. An already-active target moves nothing and
    // still answers `handoff`, because it IS active after admission. The same
    // message without the flag stays a sibling's normal message. A different
    // id's flagged message steals as any message does but is not a handoff. A
    // handoff to a perch this node does not hold is the plain `no_perch` path.
    #[test]
    fn receiver_decides_a_handoff() {
        use crate::resting::{read_rest, write_rest, RestState};
        crate::test_home::with_home(|_| {
            let owlery = spt_store::perch::owlery_dir();
            let registry = crate::registryhost::RegistryHost::new("handoff-node");
            let msg = |target: &str, sender: &str, handoff: bool, op: &str| WanMessage {
                target: target.to_string(),
                from: sender.to_string(),
                body: "yours".to_string(),
                op_id: op.to_string(),
                sender_proven: Some(sender.to_string()),
                sender_origin: None,
                handoff,
            };
            let answer = |m: &WanMessage| {
                let (outcome, taken) = receive_wan_reply(m, "deadbeefcafef00d", &owlery, &registry);
                outcome.reply_token(taken)
            };

            for (id, from) in [("hand-d", RestState::Dormant), ("hand-s", RestState::Suspended)] {
                let p = stranger_admitting_receiver(id, id);
                write_rest(&p, from, 1).unwrap();
                assert_eq!(answer(&msg(id, id, true, &format!("{id}:1"))), OUTCOME_HANDOFF, "{id}");
                assert_eq!(read_rest(&p).unwrap().state, RestState::Active, "{id} takes active");
            }

            let a = stranger_admitting_receiver("hand-a", "hand-a");
            write_rest(&a, RestState::Active, 1).unwrap();
            assert_eq!(answer(&msg("hand-a", "hand-a", true, "a:1")), OUTCOME_HANDOFF);
            assert_eq!(read_rest(&a).unwrap().state, RestState::Active);

            let n = stranger_admitting_receiver("hand-n", "hand-n");
            write_rest(&n, RestState::Dormant, 1).unwrap();
            assert_eq!(answer(&msg("hand-n", "hand-n", false, "n:1")), "spooled");
            assert_eq!(read_rest(&n).unwrap().state, RestState::Dormant, "no flag, no wake");

            let x = stranger_admitting_receiver("hand-x", "stranger");
            write_rest(&x, RestState::Dormant, 1).unwrap();
            assert_eq!(answer(&msg("hand-x", "stranger", true, "x:1")), "spooled");
            assert_eq!(read_rest(&x).unwrap().state, RestState::Active, "a different id steals anyway");

            assert_eq!(answer(&msg("hand-gone", "hand-gone", true, "g:1")), "no_perch");
        });
    }

    // [unit->REQ-INSTANCE-HANDOFF] the `handoff` token needs custody AND active:
    // a refusal, an absence or a replay answers with its own token whatever the
    // flag said. The sender maps the token back, and it counts as custody.
    #[test]
    fn handoff_reply_token_needs_custody_and_active() {
        assert_eq!(WanOutcome::DeliveredTcp.reply_token(true), OUTCOME_HANDOFF);
        assert_eq!(WanOutcome::DeliveredInject.reply_token(true), OUTCOME_HANDOFF);
        assert_eq!(WanOutcome::Spooled.reply_token(true), OUTCOME_HANDOFF);
        for o in [WanOutcome::Duplicate, WanOutcome::Refused, WanOutcome::NoPerch] {
            assert_eq!(o.reply_token(true), o.token());
        }
        assert_eq!(WanOutcome::DeliveredTcp.reply_token(false), "delivered");
        assert_eq!(WanRequestOutcome::from_token(OUTCOME_HANDOFF), WanRequestOutcome::HandoffTaken);
        assert!(WanRequestOutcome::HandoffTaken.took_custody());
    }

'''
s = rep(s, anchor, new + anchor)
save(p, s, c)

# ---- cli.rs: factor the handoff tail + tests
p = 'crates/spt/src/cli.rs'
s, c = load(p)
s = rep(s, '''                return match crate::wansend::handoff_report(&outcome, &target) {
                    Ok(line) => {
                        ok_line(line);
                        SendVerdict::Delivered
                    }
                    Err(line) => {
                        eprintln!("{line}");
                        SendVerdict::Refused { code: 1 }
                    }
                };
            }''', '''                return finish_handoff(&outcome, &target);
            }''')
s = rep(s, '''fn admit_message_trigger(target: &str, sender: &str, window: &str) {''', '''/// A handoff's verdict, printed (REQ-INSTANCE-HANDOFF). It reads the receiver's
/// answer and WRITES NOTHING: the sender's rest state moves only when the new
/// active's advertisement reaches this node and outranks it (doyle condition 4),
/// so there is exactly one writer of that transition.
// [impl->REQ-INSTANCE-HANDOFF]
fn finish_handoff(outcome: &crate::wansend::WanSendOutcome, target: &str) -> SendVerdict {
    match crate::wansend::handoff_report(outcome, target) {
        Ok(line) => {
            ok_line(line);
            SendVerdict::Delivered
        }
        Err(line) => {
            eprintln!("{line}");
            SendVerdict::Refused { code: 1 }
        }
    }
}

fn admit_message_trigger(target: &str, sender: &str, window: &str) {''')
anchor = '''    // [unit->REQ-ACTIVATION-TRIGGERS] the RING admission point (doyle-accepted'''
new = '''    /// A perch for `id` whose effective state is `state`, with NO process tie:
    /// a dead pid and a daemon status, so neither the ancestry leg nor the pid
    /// probe can make it anyone's session.
    fn seed_instance(id: &str, state: spt_daemon::resting::RestState) -> std::path::PathBuf {
        use spt_daemon::resting::RestState;
        let p = perch::resolve_perch_path(id, perch::ParentHint::Infer);
        std::fs::create_dir_all(&p).unwrap();
        let mut rec = spt_store::info::InfoJson::new(id, "t", 4_000_000_000, &format!("sid-{id}"), "live_agent");
        rec.status = Some(
            match state {
                RestState::Active => spt_store::liveness::STATUS_ONLINE,
                RestState::Dormant => spt_store::liveness::STATUS_UNBOUND,
                RestState::Suspended => spt_store::liveness::STATUS_OFFLINE,
            }
            .to_string(),
        );
        spt_store::info::write_info(&p, &rec).unwrap();
        spt_daemon::resting::write_rest(&p, state, 1).unwrap();
        p
    }

    /// Clear the env legs of `detect_self_id` for the guard's life, so a test
    /// can assert that nothing proves a sender.
    struct NoSessionEnv(Vec<(&'static str, Option<String>)>);
    impl NoSessionEnv {
        fn new() -> Self {
            let saved: Vec<_> = ["SPT_AGENT_ID", "SPT_ENDPOINT_ID", "OWL_SESSION_ID"]
                .into_iter()
                .map(|k| (k, std::env::var(k).ok()))
                .collect();
            for (k, _) in &saved {
                std::env::remove_var(k);
            }
            NoSessionEnv(saved)
        }
    }
    impl Drop for NoSessionEnv {
        fn drop(&mut self) {
            for (k, v) in self.0.drain(..) {
                if let Some(v) = v {
                    std::env::set_var(k, v);
                }
            }
        }
    }

    // [unit->REQ-DORMANT-SEND-RESTRICTION] doyle condition 1: shortform runs
    // `cmd_send_verdict` in the `api state` process under the HARNESS's
    // environment, where the session lookup may prove nobody. The restriction
    // keys on the author the auth gate proved (`proven_author`), so a DORMANT
    // author's shortform to a peer is refused with no session env at all, and
    // nothing is spooled. Control: the same send with no proven author is not
    // held, and it reaches the peer's spool. That proves the refusal came from
    // the proven author and from nothing in the environment.
    #[test]
    fn a_dormant_authors_shortform_is_refused_without_a_session_env() {
        use spt_daemon::resting::RestState;
        let home = crate::testutil::isolated_home();
        let _ = &home;
        let _env = NoSessionEnv::new();
        seed_instance("ling", RestState::Dormant);
        let dean = seed_instance("dean", RestState::Suspended);
        assert_eq!(roster::detect_self_id(), None, "precondition: nothing proves a sender");
        let send = |author: Option<&str>| {
            cmd_send_verdict(
                "dean".into(),
                Some("ling".into()),
                false,
                false,
                false,
                false,
                false,
                None,
                false,
                None,
                false,
                Some("hi".into()),
                SendExtras {
                    proven_author: author.map(str::to_string),
                    ..Default::default()
                },
            )
        };
        let spooled = || spt_store::spool::peek_all_at(&dean).map(|r| r.len()).unwrap_or(0);
        assert_eq!(send(Some("ling")), SendVerdict::Refused { code: 1 });
        assert_eq!(spooled(), 0, "a refusal leaves nothing behind");
        assert_eq!(send(None), SendVerdict::Delivered, "control: no proven author, no restriction");
        assert_eq!(spooled(), 1);
    }

    // [unit->REQ-INSTANCE-HANDOFF] doyle condition 4, NO TWO WRITERS: whatever
    // the receiver answers, the sender's verdict writes nothing to its own
    // perch. Its record is byte-identical after a `handoff` reply, before any
    // push arrives, and after every other answer too. And a handoff from an
    // instance that is not active is refused before any byte moves.
    #[test]
    fn a_handoff_verdict_never_writes_the_senders_state() {
        use crate::wansend::WanSendOutcome;
        use spt_daemon::resting::RestState;
        let home = crate::testutil::isolated_home();
        let _ = &home;
        let _env = NoSessionEnv::new();
        let p = seed_instance("ling", RestState::Active);
        let before = std::fs::read(p.join("info.json")).unwrap();
        let node = || "fafa".to_string();
        for outcome in [
            WanSendOutcome::Sent { node: node(), how: "handoff" },
            WanSendOutcome::Sent { node: node(), how: "delivered" },
            WanSendOutcome::Refused { node: node() },
            WanSendOutcome::PeerSilent { node: node() },
        ] {
            let _ = finish_handoff(&outcome, "ling@desk");
            assert_eq!(std::fs::read(p.join("info.json")).unwrap(), before, "{outcome:?}");
        }
        assert_eq!(
            finish_handoff(&WanSendOutcome::Sent { node: node(), how: "handoff" }, "ling@desk"),
            SendVerdict::Delivered
        );

        seed_instance("oak", RestState::Dormant);
        assert!(matches!(
            crate::wansend::handoff_plan(Some("oak"), "oak@desk"),
            crate::wansend::HandoffPlan::Refuse(why) if why.contains("only the active instance may hand off")
        ));
    }

'''
s = rep(s, anchor, new + anchor)
save(p, s, c)
print('tests ok')