import io


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)


p = 'crates/spt-daemon/tests/twohost_axes.rs'
s, c = load(p)

s = rep(s, '''//! | 4 | TIE | both write active with the SAME counter (1000), then both push. Within 15 s the higher node hex (the `(u64, &str)` order `sibling_outranks` uses) stays active, the lower reads dormant, both registries agree. |
''', '''//! | 4 | TIE | both write active with the SAME counter (1000), then both push. Within 15 s the higher node hex (the `(u64, &str)` order `sibling_outranks` uses) stays active, the lower reads dormant, both registries agree. |
//! | 6 | HANDOFF + DORMANT RESTRICTION (releases#349) | from cell 4's end (W = the tie winner, active; L dormant). **6a:** read over each side's live registry, the production restriction refuses L's send to a peer and to its own id on another node, and passes L's send pinned to W; W is unrestricted, and its handoff plan to L ships. W then sends a real `handoff` WanMessage to L over the wire: L answers `handoff`, L takes active, and within 15 s W reads dormant and both registries show exactly one active row, L's. After the swap the restriction and plan read the other way round on each side. **6b** (a SUSPENDED sibling, doyle's cell-6b ruling): W suspends; L hands back to W; W's intent reads active with a counter above L's, and L never reads dormant while W's row is not yet active with that counter, so L stays active until W advertises. The rig pins liveness, so W's "resume" is instant here; a real resume rides #351 field acceptance. |
''')

s = rep(s, '''const P_TIE_WRITTEN: u32 = 5;
const P_DONE: u32 = 6;''', '''const P_TIE_WRITTEN: u32 = 5;
const P_TIED: u32 = 6;
const P_SWAPPED: u32 = 7;
const P_W_SUSPENDED: u32 = 8;
const P_DONE: u32 = 9;''')

s = rep(s, '''use spt_net::net::registry::{resolve_across_visible, Instance, Resolution, Status, SubnetRegistry};''', '''use spt_net::net::registry::{resolve_across_visible, Instance, Resolution, Status, SubnetRegistry};
use spt_net::net::siblings::{dormant_send_refusal, plan_handoff, sibling_rows, HandoffPlan, SendHead, SiblingRow};
use spt_net::net::wanmsg::WanMessage;''')

s = rep(s, '''    /// The node a bare [`ID`] resolves to on this side: the production resolver''', '''    /// This side's SIBLING rows for [`ID`] from its live registry, through the
    /// production reader the send path uses.
    fn siblings(&self) -> Vec<SiblingRow> {
        let Some(reg) = self.registry.snapshot(&self.subnet) else {
            return Vec::new();
        };
        let regs: BTreeMap<String, SubnetRegistry> = [(self.subnet.clone(), reg)].into_iter().collect();
        sibling_rows(&regs, &self.own_hex, ID, Some(AXES_UID), |_, _| false)
    }

    /// The production dormant restriction for a send from THIS side's
    /// instance, judged on its local state and its live sibling rows.
    fn restriction(&self, head: SendHead<'_>) -> Option<String> {
        let own = match self.local().state {
            RestState::Active => Status::Active,
            RestState::Dormant => Status::Dormant,
            RestState::Suspended => Status::Suspended,
        };
        dormant_send_refusal(ID, own, &head, &self.own_hex, None, &self.siblings())
    }

    /// The production handoff plan from THIS side to `node`.
    fn handoff_plan_to(&self, node: &str) -> HandoffPlan {
        let own = match self.local().state {
            RestState::Active => Status::Active,
            RestState::Dormant => Status::Dormant,
            RestState::Suspended => Status::Suspended,
        };
        plan_handoff(Some(ID), Some(own), &format!("{ID}@{node}"), &self.own_hex, None, &self.siblings())
    }

    /// Ship a real `handoff` message to the peer's instance over the wire and
    /// return the receiver's answer.
    fn ship_handoff(&self, broker_name: &str, peer_addr: serde_json::Value, op: &str) -> spt_daemon::WanRequestOutcome {
        let mut brain = connect_retry_pump(broker_name);
        let conn = brain.net_dial(peer_addr, None).expect("dial the peer for the handoff");
        let msg = WanMessage {
            target: ID.to_string(),
            from: ID.to_string(),
            body: "over to you".to_string(),
            op_id: op.to_string(),
            sender_proven: Some(ID.to_string()),
            sender_origin: None,
            handoff: true,
        };
        spt_daemon::request_wan(&mut brain, &conn, &msg).expect("the handoff round-trip completed")
    }

    /// The node a bare [`ID`] resolves to on this side: the production resolver''')

s = rep(s, '''fn connect_retry(name: &str) -> Brain {''', '''/// A BOUNDED carrier for `request_wan` (releases#289), as in `twohost.rs`.
fn connect_retry_pump(name: &str) -> Brain {
    for _ in 0..300 {
        if let Ok(b) = Brain::cold_start_pump(
            name,
            1,
            Duration::from_secs(30),
            spt_daemon::brain::PumpTrace::Stderr,
        ) {
            return b;
        }
        thread::sleep(Duration::from_millis(10));
    }
    panic!("pump brain could not connect");
}

fn connect_retry(name: &str) -> Brain {''')

# The Side needs its broker name; carry it.
s = rep(s, '''struct Side {
    role: &'static str,''', '''struct Side {
    role: &'static str,
    broker_name: String,''')
s = rep(s, '''    let side = Side {
        role,
        subnet: rig.subnet.clone(),''', '''    let side = Side {
        role,
        broker_name: broker_name.clone(),
        subnet: rig.subnet.clone(),''')

# Both roles: cell 6 after the tie.
old_tail = '''    // ── Cell 4 — TIE.
    // [int->REQ-ACTIVATION-COUNTER]
    tie_cell(&side, &rig);

    side.barrier(P_DONE);
    linger(&side, &stop);
}
'''
new_tail = '''    // ── Cell 4 — TIE.
    // [int->REQ-ACTIVATION-COUNTER]
    tie_cell(&side, &rig);

    // ── Cell 6 — HANDOFF + DORMANT RESTRICTION.
    // [int->REQ-INSTANCE-HANDOFF] [int->REQ-DORMANT-SEND-RESTRICTION]
    handoff_cell(&side, &rig);

    side.barrier(P_DONE);
    linger(&side, &stop);
}
'''
assert s.count(old_tail) == 2, s.count(old_tail)
s = s.replace(old_tail, new_tail)

s = s.rstrip('\n') + '''

/// Cell 6, run by both roles from cell 4's end: the tie WINNER is active and
/// the loser dormant.
fn handoff_cell(side: &Side, rig: &Rig) {
    let winner = rig.tie_winner();
    let i_win = side.own_hex == winner;
    let loser = if i_win { side.peer_hex.clone() } else { side.own_hex.clone() };
    let peer_addr = rig.peer_broker_addr(side.role);
    side.barrier(P_TIED);

    // ── 6a — the restriction over the live registry, then a handoff to a
    // DORMANT sibling.
    if i_win {
        assert_eq!(side.restriction(SendHead::Local("someone-else")), None, "an active instance is unrestricted");
        assert_eq!(
            side.handoff_plan_to(&loser),
            HandoffPlan::Ship { pin: format!("{ID}@{loser}") },
            "the active instance may hand off to its dormant sibling"
        );
        let t0 = Instant::now();
        let answer = side.ship_handoff(&side.broker_name, peer_addr.clone(), "axes-handoff-1");
        assert_eq!(answer, spt_daemon::WanRequestOutcome::HandoffTaken, "the dormant sibling took active");
        wait_for("cell 6a: the old active reads Dormant and the sibling is the sole active", CONVERGE, || {
            side.local().state == RestState::Dormant && side.sole_active_is(&loser)
        });
        println!("TWOHOST_AXES METRIC handoff_converged_ms={}", ms(t0.elapsed()));
        let why = side
            .restriction(SendHead::Local("someone-else"))
            .expect("now dormant, a send to a peer is refused");
        assert!(why.starts_with("SEND_REFUSED_DORMANT:someone-else"), "{why}");
        assert_eq!(side.restriction(SendHead::Pinned { id: ID, node: &loser }), None, "the active sibling stays reachable");
        assert!(
            matches!(side.handoff_plan_to(&loser), HandoffPlan::Refuse(why) if why.contains("only the active instance")),
            "a dormant instance cannot hand off"
        );
    } else {
        let why = side
            .restriction(SendHead::Local("someone-else"))
            .expect("a dormant instance's send to a peer is refused");
        assert!(why.starts_with("SEND_REFUSED_DORMANT:someone-else"), "{why}");
        assert!(side.restriction(SendHead::Shell("term-1")).is_some(), "and so is its shell traffic");
        assert_eq!(
            side.restriction(SendHead::Pinned { id: ID, node: &winner }),
            None,
            "a dormant instance may message its active sibling"
        );
        assert_eq!(side.restriction(SendHead::Local(ID)), None, "a bare self-send stays open (R4-9)");
        wait_for("cell 6a: the handoff made this instance the sole active", CONVERGE, || {
            side.local().state == RestState::Active && side.sole_active_is(&side.own_hex)
        });
    }
    side.barrier(P_SWAPPED);

    // ── 6b — a handoff to a SUSPENDED sibling: the old winner suspends, and
    // the new active hands back to it.
    if i_win {
        daemon_rest_event(ID, RestEvent::Suspend, None).expect("suspend the old active's seat");
        assert_eq!(side.local().state, RestState::Suspended);
        request_advertise_now();
    } else {
        wait_for("cell 6b: the sibling reads Suspended here", CONVERGE, || {
            side.peer_status() == Some(Status::Suspended)
        });
    }
    side.barrier(P_W_SUSPENDED);
    if i_win {
        wait_for("cell 6b: the handoff woke this suspended instance ACTIVE, above the sender's counter", CONVERGE, || {
            let local = side.local();
            local.state == RestState::Active
                && side.peer_row().is_some_and(|r| local.activation > r.activation)
                && side.sole_active_is(&side.own_hex)
        });
    } else {
        assert_eq!(
            side.handoff_plan_to(&winner),
            HandoffPlan::Ship { pin: format!("{ID}@{winner}") },
            "a suspended sibling can take a handoff"
        );
        let answer = side.ship_handoff(&side.broker_name, peer_addr, "axes-handoff-2");
        assert_eq!(answer, spt_daemon::WanRequestOutcome::HandoffTaken, "the suspended sibling took active");
        wait_for("cell 6b: this instance goes dormant only once the sibling advertises active", CONVERGE, || {
            let local = side.local();
            if local.state == RestState::Dormant {
                let sibling = side.row_of(&winner).expect("the sibling's row");
                assert!(
                    sibling.status == Status::Active && sibling.activation > local.activation,
                    "the sender went dormant before the sibling advertised active: {sibling:?}"
                );
            }
            local.state == RestState::Dormant && side.sole_active_is(&winner)
        });
    }
}
'''
save(p, s, c)
print('rig ok')