//! Two-host cross-node SERVING (WEBSERVE W1, releases#249) on the REAL rig —
//! the `int` cells for `REQ-WEB-CROSS-NODE-PROXY` that only two machines can
//! witness: a file registered on one host, fetched byte-equal through the
//! OTHER host's own loopback listener; a `WEB` deny on the owner answering
//! 403 naming the surface; `Range` answering the owner's 206.
//!
//! **Env-gated exactly like `twohost.rs`: runs only with `SPT_TWO_HOST=1` + a
//! role** — CI never sets it on the ordinary suite jobs, so this is a silent
//! no-op there. Same env table (`SPT_TWO_HOST_ROLE`, `_SECRET`, `_PEER_IP`,
//! `_SUBNET`, `_PORT_A`, `_PORT_B`, `_WAIT_SECS`), same derived identities.
//! `SPT_TWO_HOST_RENDEZVOUS_SECS` independently bounds the probe's pre-peer
//! wait (default 300); `_WAIT_SECS` still bounds actual ceremony/convergence.
//!
//! ## Pairing of the cells
//!
//! Role B (the OWNER) is ONE `#[test]`: it serves for the rig deadline with
//! two files registered — `open.md` by endpoint `w1-open`, `locked.md` by
//! endpoint `w1-locked` — and an ENDPOINT-scoped `WEB` deny for A's node on
//! `w1-locked`. One serving fixture, because the owner-side serve resolves
//! every store from the process-global `SPT_HOME` and a second B cell in the
//! same process would share it. Role A (the REQUESTER) is THREE `#[test]`s,
//! each its own listener, broker and home, so the three claims are three
//! separate runs (the gate's "sibling intact is WITNESSED" discipline):
//!
//! | role A cell | asserts against B |
//! |---|---|
//! | `two_host_web_fetch_role_a` | `/<b>/f/open.md` = the source bytes, 200 |
//! | `two_host_web_range_role_a` | `Range: bytes=0-3` → 206 + `Content-Range` + 4 bytes |
//! | `two_host_web_deny_role_a`  | `/<b>/f/locked.md` → 403, body names `WEB`, no sender |
//! | `two_host_web_helper_role_a` | a user's quoted path on B, registered by B on its own user's authority and pulled here byte-equal (WEBSERVE W2, releases#272/#17) |
//!
//! The helper cell is the one claim in this file that travels A-ward as well:
//! role B ORIGINATES a `user-msg` to an endpoint on A, and role A GROWS a
//! dispatch loop to receive it. Both are new to their roles and each is
//! explained where it is built. It lives on two hosts rather than on the
//! one-box rig for a measured reason — see that cell's own header.
//!
//! Ports: the rig's broker ports + [`PORT_OFFSET`] so this bin never sits on
//! the ladder's sockets; each A cell adds its own offset so the cells can run
//! as separate processes at once. Every A cell polls until B is up (a 502 is
//! "not yet", within the rig deadline), so start order does not matter.
//!
//! ## Status
//!
//! AUTHORED against the mechanized one-box rig (`crates/spt/tests/
//! webserve_cross_node_e2e.rs`, which climbs the same claims over two real
//! daemons); the two-box climb is the gate's field leg.

use std::io::{Read, Write as _};
use std::net::{IpAddr, SocketAddr, TcpStream, UdpSocket};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use sha2::{Digest, Sha256};

use spt_daemon::brain::Brain;
use spt_daemon::broker::Broker;
use spt_daemon::dispatch::{run_dispatch_loop_observed, DispatchPaths};
use spt_daemon::nethost::{NetConfig, NetHost};
use spt_daemon::registryhost::RegistryHost;
use spt_net::net::endpoint::{addr_for_node_hex, BindScope, LocalDiscovery, RelayPolicy};
use spt_net::net::pairing::totp::TotpSeed;
use spt_net::net::wanmsg::WanMessage;
use spt_proto::identity::Identity;
use spt_store::access::{AccessRule, AccessStore, OriginQualifier, Provenance, RuleDecision, Subject};
use spt_store::epoch::EpochSource;
use spt_store::perch;
use spt_store::roster::{RosterEntry, RosterStore};
use spt_store::serving::ServingRegistry;
use spt_store::subnet::SubnetStore;
use tempfile::TempDir;

#[path = "support/serving.rs"]
mod serving_fixture;
use serving_fixture::ServingFixture;

/// Wall-clock now (epoch ms) — the brain-IPC clock argument.
fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// Keeps this bin off the ladder's broker ports when both run on a box.
const PORT_OFFSET: u16 = 20;
/// The label A's roster carries for B (what the URL and the 502 name).
const B_LABEL: &str = "rig-web-b";
/// The requester cells role B waits to have served.
///
/// FIVE, and the arithmetic is written down because a wrong guess here fails in
/// two different directions: W1's three (fetch, range, deny), plus WEBSERVE W2's
/// two — the `ServeFor` B answers on its user's authority, and the helper cell's
/// fetch of what it registered. Over-counting makes role B time out LOUD with a
/// count in the message; under-counting lets B stop serving while a requester
/// cell is still asking, which is the failure that reads as a network problem.
/// Loud is the direction to be wrong in.
/// How many requester cells role B expects to serve before it is done. The
/// DEFAULT is the full A side; `SPT_TWO_HOST_EXPECT_CELLS` narrows it so a
/// FILTERED A run gives B a clean exit instead of a 240 s timeout.
///
/// Measured 2026-09-08: running A with `-E 'test(two_host_web_helper_role_a)'`
/// left B waiting for five cells that could not arrive, so B timed out and the
/// raw read `1 timed out` — a red that is entirely an artifact of the filter.
/// The next person to narrow A would have read it as a failure of the arm they
/// were testing, which is the worst kind of false red: it points at the thing
/// under test.
const A_CELLS: usize = 5;
/// WEBSERVE W2 (releases#272, #17's remote half). The Gateway on the OWNER
/// whose USER quotes a path, the endpoint on the REQUESTER that the words were
/// addressed to, and the file itself. Gateway-typed is load-bearing: WAN
/// ingress re-stamps an arriving `user-msg` down to a plain `msg` unless the
/// origin endpoint is advertised as a Gateway (REQ-MSG-6).
const HELPER_GATEWAY: &str = "w2-helper-gateway";
const HELPER_AUDIENCE: &str = "w2-helper-audience";
const HELPER_MSG_ID: &str = "BCDFGH23";
/// The requester-cell offsets. NAMED because each one is now a distinct NODE
/// IDENTITY as well as a distinct port (see `Rig::id_a_for`), and role B has to
/// name the same cell the A side does — a bare literal on both sides of an
/// identity is how two call sites drift apart without a compiler noticing.
const FETCH_OFFSET: u16 = 0;
const RANGE_OFFSET: u16 = 1;
const DENY_OFFSET: u16 = 2;
/// The requester-cell offset the helper cell uses — B dials this one A-ward.
const HELPER_OFFSET: u16 = 3;
/// The INBOUND PROBE's offset. Clear of the four requester cells (0..=3) and
/// inside the same `port_a + offset` range the rig already uses, so whatever
/// rule lets the ceremony's ports through covers this one too — a probe on a
/// port outside that range would vouch for a path nothing else takes.
const PROBE_OFFSET: u16 = 9;
/// B's SEND window, started at the first beacon — not A's, which listens for the
/// separate rendezvous budget. Ten seconds against the ceremony's 900: the point
/// of the probe is that a blocked link is answered in the time it takes to
/// notice, not in the time it takes to give up.
const PROBE_WINDOW: Duration = Duration::from_secs(10);
const PROBE_MAGIC: &[u8] = b"SPT-TWOHOST-INBOUND-PROBE-v1";
/// How long A keeps acking after its verdict is already decided. Long enough to
/// cover several of B's 500 ms probes, so a single dropped ack cannot make the
/// two halves disagree about a link they both just used.
const PROBE_ACK_WINDOW: Duration = Duration::from_secs(2);
/// A's liveness beacon, A→B. Distinct from the probe and the ack because a
/// listener that cannot tell them apart is a clock that never starts.
const PROBE_BEACON: &[u8] = b"SPT-TWOHOST-INBOUND-BEACON-v1";
/// A's receipt, A→B, and B's fast green.
const PROBE_ACK: &[u8] = b"SPT-TWOHOST-INBOUND-ACK-v1";
const HELPER_BYTES: &[u8] = b"# the file the user pointed at\n";

const OPEN_ENDPOINT: &str = "w1-open";
const LOCKED_ENDPOINT: &str = "w1-locked";
const OPEN_BYTES: &[u8] = b"# two-host open\nserved from role b\n";
const LOCKED_BYTES: &[u8] = b"# two-host locked\n";

struct Rig {
    subnet: String,
    peer_ip: IpAddr,
    port_a: u16,
    port_b: u16,
    wait: Duration,
    rendezvous: Duration,
    /// The shared rig secret, kept so each A cell can derive its OWN node key.
    secret: String,
    id_b: Identity,
    totp: TotpSeed,
}

impl Rig {
    fn from_env(role: &str) -> Option<Rig> {
        if std::env::var("SPT_TWO_HOST").as_deref() != Ok("1") {
            return None;
        }
        if std::env::var("SPT_TWO_HOST_ROLE").ok()?.to_ascii_lowercase() != role {
            return None;
        }
        let secret = std::env::var("SPT_TWO_HOST_SECRET")
            .expect("SPT_TWO_HOST_SECRET must be set (same value on both hosts)");
        let peer_ip: IpAddr = std::env::var("SPT_TWO_HOST_PEER_IP")
            .expect("SPT_TWO_HOST_PEER_IP must be set (the other host's IP)")
            .parse()
            .expect("SPT_TWO_HOST_PEER_IP parses as an IP address");
        let mut t20 = [0u8; 20];
        t20.copy_from_slice(&derive32(&secret, "subnet-totp")[..20]);
        Some(Rig {
            subnet: std::env::var("SPT_TWO_HOST_SUBNET").unwrap_or_else(|_| "twohost".into()),
            peer_ip,
            port_a: env_or("SPT_TWO_HOST_PORT_A", 7460) + PORT_OFFSET,
            port_b: env_or("SPT_TWO_HOST_PORT_B", 7461) + PORT_OFFSET,
            wait: Duration::from_secs(env_or("SPT_TWO_HOST_WAIT_SECS", 300)),
            // [impl->REQ-GOLDEN-CI-LANE]
            // Scheduler/start-up skew is not a ceremony/convergence budget.
            rendezvous: Duration::from_secs(env_or("SPT_TWO_HOST_RENDEZVOUS_SECS", 300)),
            id_b: Identity::from_seed(&derive32(&secret, "node-b")),
            secret,
            totp: TotpSeed::from_bytes(t20),
        })
    }

    /// EVERY A CELL GETS ITS OWN NODE IDENTITY, keyed by its port offset.
    ///
    /// Measured 2026-09-08 (hertz, on doyle's read): with ONE `id_a` shared
    /// across four cells on four ports, the helper cell took 63.371 s and B
    /// printed SIX `A not ready` lines; with B having no prior sighting of that
    /// id, the same cell took 3.379 s and printed ONE. iroh keys path state by
    /// NODE ID, not by address, so the earlier cells — which dial B from their
    /// own ports and exit — leave B holding a path for `id_a` at a port that is
    /// already dead. When B later dials that same id for the helper, it
    /// handshakes the stale path until liveness expiry: six 10 s
    /// `quic_op_timeout` bounds plus the rig's 2 s sleeps, ~60 s of nothing.
    ///
    /// A per-cell key removes the collision by construction rather than by
    /// ordering, so a future cell needing B->A inbound at another offset cannot
    /// reintroduce it. Production is not implicated — daemons do not share a key
    /// across live processes, which is exactly why this stayed invisible.
    fn id_a_for(&self, offset: u16) -> Identity {
        Identity::from_seed(&derive32(&self.secret, &format!("node-a{offset}")))
    }

    fn a_hex_for(&self, offset: u16) -> String {
        self.id_a_for(offset).public_key().to_hex()
    }
    fn b_hex(&self) -> String {
        self.id_b.public_key().to_hex()
    }

    /// A requester cell's broker address, by its port offset. WEBSERVE W2 is the
    /// first claim here that travels A-ward: every W1 cell dials B, so the rig
    /// had no reason to name A until the helper's message had to arrive.
    fn a_addr(&self, offset: u16) -> serde_json::Value {
        let addr = addr_for_node_hex(&self.a_hex_for(offset))
            .expect("derived node hex is a valid key")
            .with_ip_addr(SocketAddr::new(self.peer_ip, self.port_a + offset));
        serde_json::to_value(&addr).expect("EndpointAddr serializes")
    }

    /// B's broker address as the brain-IPC JSON value: derived node id + the
    /// static ip:port (no relay, no discovery — one direct path).
    fn b_addr(&self) -> serde_json::Value {
        let addr = addr_for_node_hex(&self.b_hex())
            .expect("derived node hex is a valid key")
            .with_ip_addr(SocketAddr::new(self.peer_ip, self.port_b));
        serde_json::to_value(&addr).expect("EndpointAddr serializes")
    }
}

fn derive32(secret: &str, tag: &str) -> [u8; 32] {
    let mut h = Sha256::new();
    h.update(secret.as_bytes());
    h.update(b":");
    h.update(tag.as_bytes());
    h.finalize().into()
}

fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}

/// Bind this host's broker on a fixed rig port.
fn rig_broker(name: &str, identity: Identity, port: u16, dir: &Path) -> Arc<Broker> {
    let host = rig_net_host(identity, port);
    let broker = Broker::bind_in_with_net(name, dir.join("effects.log"), Some(host)).expect("bind broker");
    let serve = Arc::clone(&broker);
    thread::spawn(move || {
        let _ = serve.serve();
    });
    broker
}

fn rig_net_host(identity: Identity, port: u16) -> NetHost {
    NetHost::start(NetConfig {
        identity,
        relay: RelayPolicy::Disabled,
        discovery: LocalDiscovery::Off,
        scope: BindScope::Port(port),
        membership: None,
    })
    .expect("net host start")
}

/// One plain HTTP/1.1 request against a loopback port.
fn http(port: u16, method: &str, path: &str, extra: &[(&str, &str)]) -> (u16, String, Vec<u8>) {
    let mut stream = TcpStream::connect(("127.0.0.1", port)).expect("connect serving listener");
    stream.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
    let mut request = format!("{method} {path} HTTP/1.1\r\nHost: localhost:{port}\r\nConnection: close\r\n");
    for (name, value) in extra {
        request.push_str(&format!("{name}: {value}\r\n"));
    }
    request.push_str("\r\n");
    stream.write_all(request.as_bytes()).unwrap();
    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).expect("read serving response");
    let split = raw.windows(4).position(|w| w == b"\r\n\r\n").expect("HTTP header boundary");
    let headers = String::from_utf8(raw[..split].to_vec()).unwrap().to_ascii_lowercase();
    let status = headers.split_whitespace().nth(1).unwrap().parse().unwrap();
    (status, headers, raw[split + 4..].to_vec())
}

fn header(headers: &str, name: &str) -> Option<String> {
    headers
        .lines()
        .find_map(|line| line.strip_prefix(&format!("{name}: ")))
        .map(|v| v.trim().to_owned())
}

/// A perch RECORD — what `perch_exists` reads and what the owner-side conjunct
/// scans. A record, not a brought-up endpoint: no claim in this file is about
/// endpoint bringup. (Measured on the one-box rig: a bare DIRECTORY answers
/// `NO_PERCH`, because `perch_exists` reads `info.json`.)
fn perch_record(home: &Path, id: &str, state: &str) {
    let perch = home.join("owlery").join(id);
    std::fs::create_dir_all(&perch).expect("perch dir");
    spt_store::info::write_info(
        &perch,
        &spt_store::info::InfoJson::new(id, "0", 0, &format!("sid-{id}"), state),
    )
    .expect("perch record");
}

/// The `user-msg` envelope both ends of the helper claim are about: ONE body,
/// carrying ONE id, quoting ONE absolute path.
fn helper_envelope(quoted: &str) -> String {
    spt_proto::event::compose_typed_event(
        spt_proto::event::EVENT_TYPE_USER_MSG,
        &[
            ("from", HELPER_GATEWAY),
            (spt_proto::event::EVENT_ATTR_MSG_ID, HELPER_MSG_ID),
        ],
        &format!("have a look at {quoted} when you get a moment"),
    )
}

// ══ THE INBOUND PROBE ═══════════════════════════════════════════════════════
// A golden run once spent 900 s on each half discovering that B's datagrams
// never reached A: every QUIC dial died on the 10 s bound — 75 of them, cadenced
// at exactly 12.00 s, never one ADMIT.
//
// TWO LAYERS were behind it and EITHER ALONE produces the same silence, which is
// why the first fix looked wrong when it changed nothing:
//   1. the host firewall — BlockInbound on every profile, and its allow rules are
//      PER-EXECUTABLE, so no rule covered the test binary and the runner (a
//      service) never shows an Allow dialog;
//   2. the TAILNET ACL, asymmetric — a member device may open flows TO a TAGGED
//      resource and the reverse is denied, so the receiver permits inbound from a
//      list the sender is simply absent from.
// Every face that ever appeared to work rode RETURN traffic of flows the receiver
// had opened; a COLD claim in this direction can never cross. The ceremony cannot
// tell that apart from a pairing fault: at B a datagram dropped by a firewall
// looks exactly like one that arrived and was ignored, so the failure was
// triaged as product, then rig, before it was measured as box.
//
// WHY THIS LIVES IN THIS BINARY rather than in a shell script beside it, which
// would have been the smaller change: the Windows rule is PER-EXECUTABLE. An
// allow rule naming pwsh, python, or a bench script says nothing about
// `twohost_web-<hash>.exe`, so a probe written in any of them would today agree
// with reality by coincidence and would go GREEN the moment anyone allowed the
// prober — certifying the exact failure it exists to catch. Only a probe that
// IS the test binary carries the test binary's firewall identity. (It is also
// why the operator rule must be PORT+REMOTE scoped: a program-scoped rule is
// orphaned by the next rebuild's hash, exactly as the dead `_work\spt-core`
// rules were. This probe stays correct under either.)
//
// ── THERE IS NO SHARED CLOCK, SO THERE IS NO FIXED WINDOW ────────────────────
//
// The step boundary orders the probe BEFORE the ceremony on ONE host. It does
// NOT align the two hosts: after the ladder's last rung each half pays its own
// wrapper and its own `cargo test` invocation, and on the Windows side that is a
// fingerprint scan plus Defender's first touch of a fresh exe — seconds to tens
// of seconds, measured on that box all week.
//
// A FIXED SYMMETRIC WINDOW THEREFORE ADDS A THIRD CAUSE THAT LOOKS EXACTLY LIKE
// THE OTHER TWO: "the peer is not up yet" produces the same silence as a
// firewall drop and an ACL denial, and a failure text naming two layers would be
// confidently wrong about a third. MEASURED, not argued: the first hand-run of
// this probe in the A=kitsubito direction (2026-09-09) redded INBOUND_BLOCKED at
// 10.24 s purely because `cargo` startup on the sender put its first datagram
// after the receiver's 10 s window closed. Same silence, innocent box — the
// defect class this probe exists to document, reproduced inside the probe.
//
// So the rendezvous is DATA, not timing:
//   * A listens up to `SPT_TWO_HOST_RENDEZVOUS_SECS` and stops the instant a
//     PROBE datagram arrives. Silence alone cannot prove B ever started.
//   * A BEACONS to B's probe port once a second while it listens. A→B is the
//     direction that is open under both faults (measured 3/3 from this host on
//     2026-09-08 while the reverse read 0/3) — THE DESIGN DEPENDS ON IT, which is
//     why B's timeout for "no beacon at all" is its own fourth outcome below and
//     never INBOUND_BLOCKED.
//   * B binds its probe port BEFORE it sends anything, waits for the first
//     beacon, and only then starts a ten-second clock. B has now excluded "A is
//     not up", so its red can name the two layers honestly.
//   * A ACKs the first PROBE it receives, to the address it came from. B stops on
//     the ACK — the fast green — and the ACK is return traffic of a flow B opened,
//     which crosses under either fault, so it proves nothing on its own and is
//     not asked to.
//
// THREE DISTINCT MAGICS, because a listener that cannot tell a beacon from an ack
// is a clock that never starts and a red that never fires.
//
// THE ASSERTION LIVES ON A, deliberately: B cannot know whether its datagram
// arrived, so a B-side "I sent it" proves nothing about the path. A's own red
// waits the full rendezvous budget ONCE. Without B's scheduler/step receipt,
// A's silence remains ambiguous between a missing sender and a blocked link.
#[test]
fn two_host_inbound_probe_role_a() {
    let Some(rig) = Rig::from_env("a") else {
        return;
    };
    let port = rig.port_a + PROBE_OFFSET;
    let beacon_target = SocketAddr::new(rig.peer_ip, rig.port_b + PROBE_OFFSET);
    let socket = UdpSocket::bind(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), port))
        .unwrap_or_else(|e| panic!("INBOUND_PROBE: role A could not bind udp {port}: {e}"));
    socket
        .set_read_timeout(Some(Duration::from_millis(250)))
        .expect("probe socket takes a read timeout");
    // ⚠ THE LISTENING SOCKET NEVER SENDS. THAT IS THE WHOLE DESIGN.
    //
    // Every outbound datagram from `port` — a beacon, an ack, anything — opens
    // stateful firewall / NAT return state for that port, so B's probe then
    // arrives as SOLICITED return traffic and crosses under exactly the two
    // faults this cell exists to detect. That is this file's own warning —
    // "solicited return traffic works under either fault, so an echo reply
    // proves nothing about this direction" — applied to the guard rather than to
    // the ceremony.
    //
    // MEASURED, not reasoned (2026-09-09): with the beacon sharing the listening
    // socket, the out-of-grant control arm (probe port forced to 7509, outside
    // the operator's udp 7460-7499 rule) went GREEN — the probe certified a path
    // that a listen-only run had measured as BLOCKED eight minutes earlier. The
    // beacon and the ack therefore both leave this ephemeral socket, and B sends
    // from an ephemeral port of its own, so no run repeats a 4-tuple a previous
    // run opened and none can poison the next.
    let beacon_socket = UdpSocket::bind(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 0))
        .expect("probe beacon binds an ephemeral port");
    println!(
        "TWOHOST-WEB probe role A: listening on udp {port} for rendezvous budget {:?}, beaconing to {beacon_target}",
        rig.rendezvous
    );

    let deadline = Instant::now() + rig.rendezvous;
    let mut buf = [0u8; 64];
    let mut seen = 0usize;
    let mut beacons = 0usize;
    let mut last_beacon = Instant::now() - Duration::from_secs(1);
    while Instant::now() < deadline {
        // One beacon a second, so B can start its own clock from evidence that
        // this side is up rather than from a guess about start-up skew.
        if last_beacon.elapsed() >= Duration::from_secs(1) {
            match beacon_socket.send_to(PROBE_BEACON, beacon_target) {
                Ok(_) => beacons += 1,
                Err(e) => println!("TWOHOST-WEB probe role A: beacon send error (continuing): {e}"),
            }
            last_beacon = Instant::now();
        }
        match socket.recv_from(&mut buf) {
            Ok((n, from)) if buf[..n] == *PROBE_MAGIC => {
                seen += 1;
                println!("TWOHOST-WEB probe role A: datagram {seen} from {from} after {beacons} beacon(s)");
                // The ack is B's fast green. It is RETURN traffic of a flow B
                // opened, so it crosses under either fault — which is exactly why
                // nothing is inferred from it beyond "A heard you".
                // OUT OF THE BEACON SOCKET, not this one: an ack sent from the
                // listening port would write return state for the very tuple the
                // next run must find cold, so the guard would poison its own
                // next measurement at the end of every run, green or red.
                //
                // AND ADDRESSED TO B'S BEACON PORT, not to `from`. MEASURED
                // 2026-09-09: acking to `from` (B's ephemeral probe port) leaves
                // A's ack unsolicited at B — it comes from this ephemeral socket,
                // not from the port B dialled, so it matches no return state B
                // opened — and a receiver whose inbound rule is a fixed port
                // RANGE drops it. B then reported INBOUND_BLOCKED on a link its
                // own peer had just certified: the A-side said INBOUND OK and the
                // B-side said blocked, in the same run. The rig's port range is
                // the only inbound address B can be reached on cold, so the ack
                // uses it, and B reads the ack on the socket it already binds.
                if let Err(e) = beacon_socket.send_to(PROBE_ACK, beacon_target) {
                    println!("TWOHOST-WEB probe role A: ack send error (continuing): {e}");
                }
                // ONE ACK IS NOT ENOUGH, and the failure it leaves behind is the
                // self-contradicting run this cell already carries a fix for —
                // arriving by a different road. A single unrepeated UDP datagram
                // that drops leaves A printing INBOUND OK while B reds
                // INBOUND_BLOCKED and blames a firewall for packet loss. So A
                // keeps draining briefly and acks EVERY probe it sees in that
                // window (B sends one per 500 ms, so this covers four), while B
                // stops on the first ack it gets. A's own verdict is unchanged:
                // it was decided by the datagram above.
                let ack_until = Instant::now() + PROBE_ACK_WINDOW;
                while Instant::now() < ack_until {
                    match socket.recv_from(&mut buf) {
                        Ok((n, _)) if buf[..n] == *PROBE_MAGIC => {
                            seen += 1;
                            let _ = beacon_socket.send_to(PROBE_ACK, beacon_target);
                        }
                        _ => {}
                    }
                }
                println!("TWOHOST-WEB probe role A: acked {seen} probe(s) in {PROBE_ACK_WINDOW:?}");
                break;
            }
            // A stray datagram is not this probe's business; keep waiting rather
            // than passing on someone else's traffic.
            Ok((n, from)) => println!("TWOHOST-WEB probe role A: {n} foreign bytes from {from}"),
            Err(_) => {}
        }
    }

    // No datagram is not proof of a blocked link: B may still be queued on its
    // runner. Keep the negative loud, but require independent sender evidence
    // before attributing it to either network-admission layer.
    assert!(
        seen > 0,
        "PROBE_RENDEZVOUS_UNVERIFIED: role A received NO probe datagram on udp {port} from {} within {:?} \
         (and sent {beacons} beacon(s) to {beacon_target} in that time).
This is NOT a pairing failure or proof of an inbound block. This receiver has no evidence that B
started sending. Inspect B's job and probe step in the SAME workflow run and attempt first: queued,
absent, cancelled or failed-before-send is a scheduler/rendezvous failure, not a product verdict.

If B's matching log confirms it received A's beacon and sent probes, investigate the two
network-admission layers below. Neither this silence nor elapsed time identifies which one failed.
The ceremony's SPT_TWO_HOST_WAIT_SECS budget is separate and has not been used by this probe.

  LAYER 1 - the Windows host firewall on the receiver. BlockInbound on every profile, and its allow
  rules are PER-EXECUTABLE, so a rule naming pwsh or python says nothing about this test binary and
  the runner (a service) never shows an Allow dialog. Fix, operator, elevated - allow INBOUND UDP
  {}-{} from {} only: PORT+REMOTE scoped, never program-scoped, or the next rebuild binary hash
  orphans the rule the way the dead _work/spt-core rules were orphaned:
    New-NetFirewallRule -DisplayName 'spt twohost rig inbound' -Direction Inbound -Protocol UDP -LocalPort {}-{} -RemoteAddress {} -Action Allow

  LAYER 2 - the TAILNET ACL, which is asymmetric and denies this direction. The sender is a TAGGED
  resource owned by another tailnet user; a member device may open flows TO it, and the reverse is
  denied, so the receiver permits inbound from a list this sender is simply absent from. Every
  helper-stall face that ever appeared to work rode RETURN traffic of flows the receiver opened; a
  COLD claim in this direction can never cross. Fix, operator - grant it in the tailnet policy
  (asked on releases#272: src tag:eye-tracking-resource -> dst this host, udp 7460-7499). Verify with
  `tailscale debug netmap` on the RECEIVER: the sender IP must appear among the PacketFilter
  permitted inbound sources.

THE DISCRIMINATOR, ten seconds, run it before assuming which layer bit: bind the SAME listener and
send to it once over the LAN and once over Tailscale. LAN receives and Tailscale does not = LAYER 2,
the host firewall is not your problem. Neither receives = LAYER 1 (or both). Solicited return traffic
works under either fault, so an echo reply proves nothing about this direction.

WHAT THIS RED HAS NOT EXCLUDED: the peer being slow to start. B starts its send clock only after a
beacon reaches it, but this receiver cannot observe that event without B's matching run evidence.

Port range above is exactly what THIS rig binds (port_a..probe); releases#272 carries the wider
7460-7499, which covers it. Apply the board's range if they differ - this message and that comment
must never become two different commands.",
        rig.peer_ip,
        rig.rendezvous,
        rig.port_a,
        rig.port_a + PROBE_OFFSET,
        rig.peer_ip,
        rig.port_a,
        rig.port_a + PROBE_OFFSET,
        rig.peer_ip,
    );
    println!("TWOHOST-WEB probe role A: INBOUND OK on udp {port}");
}

/// B's half: wait for proof that A is up, then send for exactly ten seconds.
///
/// The ten seconds are the point — a blocked link is answered in the time it
/// takes to notice, not in the time it takes to give up — but they only start
/// once a beacon has arrived, because a clock started before the peer exists
/// measures start-up skew and calls it a firewall.
#[test]
fn two_host_inbound_probe_role_b() {
    let Some(rig) = Rig::from_env("b") else {
        return;
    };
    let listen = rig.port_b + PROBE_OFFSET;
    let target = SocketAddr::new(rig.peer_ip, rig.port_a + PROBE_OFFSET);
    // BOUND BEFORE ANYTHING IS SENT: A beacons to this port cold, and a beacon
    // that arrives before the bind is a beacon that never happened.
    let socket = UdpSocket::bind(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), listen))
        .unwrap_or_else(|e| panic!("INBOUND_PROBE: role B could not bind udp {listen}: {e}"));
    socket
        .set_read_timeout(Some(Duration::from_millis(250)))
        .expect("probe socket takes a read timeout");
    // ⚠ THE PROBES LEAVE A SEPARATE, EPHEMERAL SOCKET — this fixed one only ever
    // RECEIVES beacons.
    //
    // With a fixed source port here, the tuple (B:port_b+9 → A:probe) REPEATS
    // from run to run, so any earlier outbound from A to that port leaves return
    // state that carries the next run's probe across a link that is actually
    // blocked — at either layer, since WFP and Tailscale's filter both track
    // outbound UDP flows by 4-tuple. An ephemeral source port makes each run's
    // tuple new, so no run can poison the next BY CONSTRUCTION rather than by a
    // cold-start convention nobody will remember. This socket only ever SENDS:
    // the ack comes back to the fixed socket above, because A addresses it to
    // this host's rig port (the only address either host is reachable on cold).
    let probe_socket = UdpSocket::bind(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 0))
        .expect("probe sender binds an ephemeral port");
    println!(
        "TWOHOST-WEB probe role B: waiting for rendezvous up to {:?} on udp {listen} for A's beacon, then sending to {target} for {PROBE_WINDOW:?}",
        rig.rendezvous
    );

    let mut buf = [0u8; 64];
    let beacon_deadline = Instant::now() + rig.rendezvous;
    let mut heard_beacon = false;
    while Instant::now() < beacon_deadline {
        match socket.recv_from(&mut buf) {
            Ok((n, from)) if buf[..n] == *PROBE_BEACON => {
                println!("TWOHOST-WEB probe role B: beacon from {from} — A is up, starting the clock");
                heard_beacon = true;
                break;
            }
            Ok((n, from)) => println!("TWOHOST-WEB probe role B: {n} foreign bytes from {from}"),
            Err(_) => {}
        }
    }

    // THE FOURTH OUTCOME, and it gets its own name. No beacon means this half
    // never established that the other half exists — which is a RENDEZVOUS
    // failure, not an inbound block, and calling it INBOUND_BLOCKED would be the
    // same conflation this whole probe was written to end, one layer up.
    assert!(
        heard_beacon,
        "PROBE_NO_BEACON: role B heard nothing from A on udp {listen} within {:?}.
This is NOT an inbound-block finding and must not be reported as one: it says only that A's beacons
did not reach this host, so this half never learned that the other half is running.

Look, in this order: (1) did A's step run at all on the other host, and did it reach its 'listening on
udp' line; (2) is A->B blocked, which is the direction this rig has always found OPEN (measured 3/3
from A on 2026-09-08 while the reverse read 0/3) — if it has closed, that is a NEW box finding and
the more interesting one; (3) is this host's own udp {listen} bound by something else.
A's own no-probe result is PROBE_RENDEZVOUS_UNVERIFIED: inspect both matching job timelines before
classifying its silence as a blocked link.",
        rig.rendezvous
    );

    // The window that actually measures the link, started from evidence.
    let deadline = Instant::now() + PROBE_WINDOW;
    let mut sent = 0usize;
    let mut acked = false;
    while Instant::now() < deadline {
        match probe_socket.send_to(PROBE_MAGIC, target) {
            Ok(_) => sent += 1,
            Err(e) => println!("TWOHOST-WEB probe role B: send error (continuing): {e}"),
        }
        // The ack arrives on the FIXED socket, not the ephemeral sender: A sends
        // it to this host's rig port so it is addressable cold (see A's cell).
        match socket.recv_from(&mut buf) {
            Ok((n, from)) if buf[..n] == *PROBE_ACK => {
                println!("TWOHOST-WEB probe role B: ack from {from} after {sent} datagram(s)");
                acked = true;
                break;
            }
            // A beacon is not an ack. Reading one as the other would stop the
            // clock on the evidence that started it.
            Ok((n, _)) if buf[..n] == *PROBE_BEACON => {}
            Ok((n, from)) => println!("TWOHOST-WEB probe role B: {n} foreign bytes from {from}"),
            Err(_) => {}
        }
    }
    println!("TWOHOST-WEB probe role B: {sent} datagrams sent to {target}, acked={acked}");
    assert!(sent > 0, "role B could not send a single probe datagram to {target}");

    // B's fast RED, and it is the one IR-89 promises: ten seconds, its own name,
    // with "A is not up yet" already excluded by the beacon above.
    assert!(
        acked,
        "INBOUND_BLOCKED: role B sent {sent} datagram(s) to {target} in {PROBE_WINDOW:?} and A never
acked — while A's beacons were arriving here the whole time, so A IS UP and the reverse direction is
open. B->A is blocked.

THERE ARE TWO LAYERS AND EITHER ONE ALONE PRODUCES THIS EXACT SILENCE; the receiver's own cell
carries the full text and the two operator fixes. In short: LAYER 1 is the receiver's per-executable
host firewall (allow INBOUND UDP for the rig's port range from this host, PORT+REMOTE scoped, never
program-scoped), LAYER 2 is the asymmetric tailnet ACL (this sender must appear among the receiver's
PacketFilter permitted inbound sources — check `tailscale debug netmap` on the RECEIVER). Fixing only
one is the mistake: it is NECESSARY, NOT SUFFICIENT, and the run reds identically afterwards.

Do not spend the ceremony's budget on this: the dial below would fail the same way for the same
reason, and would report it as a pairing failure."
    );
}

// ── Role B: the owner ─────────────────────────────────────────────────────

/// Serves for the rig deadline: two registered files and one endpoint-scoped
/// `WEB` deny for A's node on the locked one. The dispatcher is the
/// production loop; the gate it runs is the production gate.
// [int->REQ-WEB-CROSS-NODE-PROXY]
// [int->REQ-WEB-ACCESS-SURFACE]
#[test]
fn two_host_web_role_b() {
    let Some(rig) = Rig::from_env("b") else {
        return;
    };
    let home = TempDir::new().expect("temp home");
    std::env::set_var("SPT_HOME", home.path());
    let scratch = TempDir::new().expect("scratch");
    println!("TWOHOST-WEB role B: node {} serving on udp {}", rig.b_hex(), rig.port_b);

    let mut subnets = SubnetStore::load();
    subnets
        .add_joined(&rig.subnet, *rig.totp.as_bytes(), 1, None, None)
        .expect("seed subnet");
    subnets.save().expect("save subnets");

    // The served files, registered as two different endpoints would.
    let open = home.path().join("open.md");
    let locked = home.path().join("locked.md");
    std::fs::write(&open, OPEN_BYTES).expect("write open.md");
    std::fs::write(&locked, LOCKED_BYTES).expect("write locked.md");
    let registry_path = perch::serving_registry_file_in(home.path());
    let mut registry = ServingRegistry::load_at(&registry_path).expect("registry");
    registry.add_reference(&open, Some("open.md"), Some(OPEN_ENDPOINT), 1).expect("register open.md");
    registry.add_reference(&locked, Some("locked.md"), Some(LOCKED_ENDPOINT), 1).expect("register locked.md");
    registry.save_at(&registry_path).expect("save registry");

    // The deny: A's node may not read what `w1-locked` serves. Endpoint
    // scope, so `w1-open`'s file stays open — the sibling the deny cell's
    // mutation arm must leave intact.
    let access_path = home.path().join("identity").join("trust").join("access.json");
    let mut store = AccessStore::load_from(&access_path);
    store.endpoint_mut(LOCKED_ENDPOINT).rules.push(AccessRule {
        // The DENY cell's node, named by its offset. With one shared key this
        // rule happened to cover every A cell; now it covers exactly the cell
        // whose refusal the arm is about, which is what it always meant.
        subject: Subject::Node { node: rig.a_hex_for(DENY_OFFSET) },
        surfaces: vec!["WEB".to_owned()],
        origin: OriginQualifier::Any,
        provenance: Provenance::Manual,
        decision: RuleDecision::Deny,
    });
    store.save_to(&access_path).expect("write the deny");

    // [int->REQ-WEB-URL-BOUND-PORT]
    // [int->REQ-SERVING-FIXTURE-CONTROL-PLANE]
    let (serving, docs_port) = ServingFixture::start(
        home.path(),
        B_LABEL,
        scratch.path().join("effects.log"),
        Some(rig_net_host(rig.id_b.clone(), rig.port_b)),
    );
    let broker_name = serving.broker_name.clone();
    println!(
        "TWOHOST-WEB role B: listener {docs_port}, broker udp {} ({broker_name}), pid {}",
        rig.port_b,
        std::process::id()
    );
    let registry_host = Arc::new(RegistryHost::new_at(
        &rig.b_hex(),
        EpochSource::load_from(&perch::epoch_file()),
        spt_daemon::presence::registry_snapshot_dir(),
    ));

    // ── F19a: A must be a ROSTER MEMBER before its row can be merged at all ──
    //
    // `RegistryGatePolicy::admits` is `member subnet ∧ roster.is_member(subnet,
    // origin)`, fail-closed on both, so without this the merge below answers
    // `UntrustedOrigin` and writes nothing — MEASURED on the first run of this
    // fix (`verdicts [UntrustedOrigin]`, and the precondition assert caught it at
    // the seed rather than 120 s later at the helper deadline, which is the whole
    // reason the assert reads through the product's own verb).
    //
    // This is the SAME direction gap as the row itself, one layer down: A's
    // `requester` seeds a roster entry for B (and B's peer address), and nothing
    // ever seeded the reverse. One-way knowledge in a rig that only ever ran one
    // way — it could not surface until a cell needed B to judge something A said.
    // Address stays `None`: B dials A by `rig.a_addr(HELPER_OFFSET)` on the WAN
    // send below, never out of the roster, so seeding one would be an unused
    // second spelling of A's address.
    let mut roster = RosterStore::load();
    roster.merge_entry(RosterEntry {
        pubkey_hex: rig.a_hex_for(HELPER_OFFSET),
        subnet: rig.subnet.clone(),
        label: "rig-web-a".to_owned(),
        machine_id: "rig-web-a-machine".to_owned(),
        address: None,
        last_seen: "1".to_owned(),
        lease_epoch: 1,
    });
    roster.save().expect("seed A as a roster member");

    // ── F19: A's audience row, which `serve_for`'s SECOND conjunct requires ──
    //
    // `webproxy::serve_for` refuses unless `node_hosting_endpoint(home, audience)`
    // names the ORIGIN node — the same refusal text a surface deny produces, on
    // purpose (the two are not distinguishable by probing). This node has no way
    // to place `HELPER_AUDIENCE` on its own: the rig merges B's Gateway row into
    // A's registry (the A helper cell) but nothing ever merged A's audience row
    // into B's, and the rig runs no nethost gossip, so the conjunct was
    // UNSATISFIABLE BY CONSTRUCTION. It went unseen because nothing on either box
    // had ever reached it — every ServeFor fell to `Unknown` at the dispatcher
    // (F17) and was dropped before the gate ran. Measured by doyle 2026-09-07 on
    // the Linux one-box control at f3c8495b + the F17 patch: owner `b.raw` carried
    // `outcome=refused reason=ACCESS_DENIED`, and the ABSENCE of an
    // `ACCESS_REFUSED` breadcrumb for that stream — `gate.rs` prints one on every
    // non-`ImplicitOpen` deny, and stream 1's w1-locked deny did print it — is
    // what places the refusal in the second arm rather than the first.
    //
    // Seeded the same way A seeds B's: through `apply_feed_flips`, the real feed
    // entry point, under the real gate policy. Only the DELIVERY of the record is
    // the rig's; the merge, the gate and the snapshot write are the product's.
    let audience_row = spt_net::net::registry::Instance {
        node: rig.a_hex_for(HELPER_OFFSET),
        status: spt_net::net::registry::Status::Active,
        epoch: 1,
        resources: None,
        last_active_ms: None,
        shell_adapters: Vec::new(),
        node_label: None,
        machine_id: Some("rig-web-a-machine".to_owned()),
        endpoint_type: None,
        bound: true,
        controller_node: None,
        harness_only: false,
        adapter: None,
        recent_projects: Vec::new(),
        controlled: false,
        activation: 0,
    };
    let (audience_verdicts, _) = registry_host.apply_feed_flips(
        &rig.a_hex_for(HELPER_OFFSET),
        &[spt_net::net::replicate::RegistryUpdate {
            subnet: rig.subnet.clone(),
            endpoint_id: HELPER_AUDIENCE.to_owned(),
            instance: audience_row,
        }],
        &spt_daemon::registryhost::RegistryGatePolicy::load(),
    );
    println!("TWOHOST-WEB role B: audience row merged, verdicts {audience_verdicts:?}");

    // The precondition is asserted through the PRODUCT'S OWN READER, not through
    // `instances_of`. They are not the same claim: `instances_of` answers from the
    // in-memory registry, while `serve_for` reads the on-disk SNAPSHOT
    // (`webserve::node_hosting_endpoint` -> `RegistryHost::snapshot_path`), so an
    // in-memory assert would pass green over a snapshot that was never written and
    // the cell would fail later at the arm this seed exists to satisfy. Assert what
    // the gate will read, from where the gate will read it.
    let a_hex = rig.a_hex_for(HELPER_OFFSET);
    let placed = spt_daemon::webserve::node_hosting_endpoint(&perch::spt_home(), HELPER_AUDIENCE);
    assert_eq!(
        placed.as_deref(),
        Some(a_hex.as_str()),
        "PRECONDITION: B must place {HELPER_AUDIENCE} on A in the on-disk registry          snapshot, or `serve_for`'s audience conjunct refuses with the same text a          surface deny produces (got {placed:?}, verdicts {audience_verdicts:?})"
    );

    let stop = Arc::new(AtomicBool::new(false));
    // Completion witness: a MONOTONIC count of what this node actually served.
    // Never the broker's stream table — see the wait loop below.
    let served = Arc::new(AtomicU64::new(0));
    {
        let name = broker_name.clone();
        let flag = Arc::clone(&stop);
        let counter = Arc::clone(&served);
        let paths = DispatchPaths {
            owlery: perch::owlery_dir(),
            releases: home.path().join("releases"),
            scratch: scratch.path().join("dispatch"),
        };
        thread::spawn(move || {
            let _ = run_dispatch_loop_observed(
                &name,
                &paths,
                registry_host,
                &flag,
                Duration::from_millis(50),
                counter,
            );
        });
    }
    // ══ WEBSERVE W2 — the OWNER half of #17's remote helper ════════════════
    // [int->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER]
    //
    // WHAT THIS ROLE COULD NOT DO BEFORE: originate. It served, and never sent.
    // The helper is authorized by a USER'S OWN WORDS, so the owner must have
    // said something AND hold its own record of saying it —
    // `msgid::local_msg_out_in` reads THIS node's `MSG_OUT` row, and every
    // conjunct is judged from that row rather than from anything the requester
    // sends. A claim the asker cannot influence is the only kind worth checking.
    //
    // THE RIG PLAYS THE CLI HERE, AND SAYS SO. In production `spt send` writes
    // the `MSG_OUT` row through the io funnel and ships the same bytes over the
    // WAN leg; there is no CLI in a `spt-daemon` test (the `spt` binary is not
    // built for this package), so the rig writes the row and ships the bytes
    // itself — ONE envelope carrying ONE id, used for both, which is the
    // property the claim actually rests on. Everything the claim tests is still
    // product code: A's WAN ingress, its quoted-path scan, `request_serve_for`
    // over QUIC, and this node's `serve_for` with its access gate and its five
    // conjuncts.
    let quoted = home.path().join("quoted-by-the-user.md");
    std::fs::write(&quoted, HELPER_BYTES).expect("write the quoted file");
    let quoted_text = quoted.to_str().expect("a utf-8 path").to_string();
    perch_record(home.path(), HELPER_GATEWAY, spt_proto::event::GATEWAY_TAG);
    let envelope = helper_envelope(&quoted_text);
    spt_store::iolog::append_at(
        &perch::owlery_dir().join(HELPER_GATEWAY),
        &spt_store::iolog::IoLogRow {
            seq: 0,
            at_ms: 1,
            kind: spt_proto::ioevent::IO_KIND_MSG_OUT.to_string(),
            payload: envelope.clone(),
            truncated: false,
            digest_seq: None,
            peer: Some(HELPER_AUDIENCE.to_string()),
            msg_id: Some(HELPER_MSG_ID.to_string()),
            mid: false,
        },
    )
    .expect("the owner's own record of what its user sent");

    // Ship it to the endpoint on A. Retried until A's ingress confirms, because
    // start order between the two hosts is not ours to fix: the helper cell on
    // A may still be binding when B is ready.
    {
        // BOUNDED carrier (releases#289): `request_wan` refuses an unbounded one
        // at the seam, because a receiver that accepts the stream and never
        // answers has no event that can end an unbounded read. Stderr trace: a
        // rig wants the reader thread's diagnostics.
        let mut brain = Brain::cold_start_pump(
            &broker_name,
            now_ms(),
            Duration::from_secs(30),
            spt_daemon::brain::PumpTrace::Stderr,
        )
        .expect("bounded brain for the WAN send");
        let msg = WanMessage {
            target: HELPER_AUDIENCE.to_string(),
            from: HELPER_GATEWAY.to_string(),
            body: envelope,
            op_id: format!("w2-helper-{}", std::process::id()),
            // The SESSION-PROVEN stamp, which is what the receiver's tier 1
            // keys on. Never `from` (hazard 7.5).
            sender_proven: Some(HELPER_GATEWAY.to_string()),
            sender_origin: None,
        };
        let end = Instant::now() + rig.wait;
        loop {
            let sent = brain
                .net_dial(rig.a_addr(HELPER_OFFSET), None)
                .map_err(|e| e.to_string())
                .and_then(|conn| {
                    spt_daemon::request_wan(&mut brain, &conn, &msg).map_err(|e| e.to_string())
                });
            // `Ok` MEANS THE TRANSPORT WORKED, NOT THAT ANYONE READ IT. A dial
            // that lands and comes back `NoPerch` is a delivered refusal: A has
            // no record of the audience endpoint, so its ingress never runs and
            // the round trip this cell exists to witness never starts. Treating
            // any `Ok` as arrival is what let this loop print "reached A
            // (NoPerch)" and walk on into a 240 s wait for a registration that
            // could not come — a false green one line before the real failure.
            // Only an ADMITTED delivery breaks the loop; every other outcome is
            // "not yet", and the deadline names the last one it saw.
            let last = match sent {
                // `took_custody()` is the PRODUCT's own classifier for "the far
                // node has it" (Delivered / Spooled / Duplicate), reused rather
                // than copied: the audience endpoint here is recorded but not
                // listening, so the honest success is a SPOOL, and a rig with
                // its own hand-written variant list would have to be corrected
                // every time that set moves.
                Ok(outcome) if outcome.took_custody() => {
                    println!("TWOHOST-WEB role B: the user's message was ADMITTED by A ({outcome:?})");
                    break;
                }
                Ok(other) => format!("A answered {other:?} — dialled, but took no custody"),
                Err(why) => why,
            };
            assert!(
                Instant::now() < end,
                "role B never got the user's message ADMITTED by A within {:?}: {last}",
                rig.wait
            );
            println!("TWOHOST-WEB role B: A not ready for the message yet ({last})");
            thread::sleep(Duration::from_secs(2));
        }
    }

    // (vi) — THIS node registered, on its user's authority, and registered it
    // the way the amendment says: a live REFERENCE (the user said look at this,
    // not keep this as it was) with the helper's lifetime, the audience, and the
    // message as its origin. Waited for on the FACT: the round trip is A's to
    // start, and it starts when A's ingress fires.
    eprintln!("F1_ARM=vi: waiting for A's ingress to register on its user's authority");
    let helper_started = Instant::now();
    let helper_end = helper_started + rig.wait;
    let helper_entry = loop {
        let registry = ServingRegistry::load_at(&registry_path).expect("registry");
        let found = registry
            .entries()
            .find(|e| e.origin.as_deref() == Some(HELPER_MSG_ID))
            .cloned();
        if let Some(entry) = found {
            break entry;
        }
        assert!(
            Instant::now() < helper_end,
            "role B was never asked to serve its user's quoted path within {:?} — \
             A's ingress caller did not complete a round trip to this node",
            rig.wait
        );
        thread::sleep(Duration::from_millis(500));
    };
    assert_eq!(
        helper_entry.kind,
        spt_store::serving::ServedKind::File,
        "a live reference, not a snapshot"
    );
    assert_eq!(
        helper_entry.ttl_ms,
        Some(spt_store::serving::HELPER_ENTRY_TTL_MS),
        "the helper's own 24h lifetime"
    );
    assert_eq!(
        helper_entry.audience,
        Some([HELPER_AUDIENCE.to_owned()].into()),
        "addressed to the endpoint the user wrote to, and to nobody else"
    );
    println!(
        "TWOHOST OK: role B registered {} for {HELPER_AUDIENCE} on its user's authority in {} ms \
         (kind={:?} ttl={:?} origin={:?})",
        helper_entry.served_name,
        helper_started.elapsed().as_millis(),
        helper_entry.kind,
        helper_entry.ttl_ms,
        helper_entry.origin
    );

    // Serve until this node has SERVED every A cell — the fetch, the range, and
    // the deny (a refusal the owner produced counts: dispatch maps Sent, Refused
    // and Failed alike to `Served`, reserving `Failed` for a transport error).
    //
    // Counted, NOT sampled. The previous witness polled the broker's stream table
    // for peer-initiated rows every 250 ms, which cannot work on a loopback pair:
    // a served stream is retired out of that table the instant its worker
    // finishes, and each cell here completes in ~0.14-0.23 s, so the row is gone
    // before the next sample. It failed "0 of 3" on the first real pair run while
    // B's own WEB_STREAM breadcrumbs proved it had served all three (200, 206, and
    // the 403). A shorter poll is NOT the fix — a race with a smaller window is
    // the same race. `served` only ever increases, so no interval can miss it.
    // The rig deadline caps the wait; a B that never serves fails loud, not late.
    let expect_cells: usize = env_or("SPT_TWO_HOST_EXPECT_CELLS", A_CELLS);
    let end = Instant::now() + rig.wait;
    println!(
        "TWOHOST OK: role B serving open.md + locked.md (waiting for {expect_cells} served cells{})",
        if expect_cells == A_CELLS { String::new() } else { format!(", narrowed from {A_CELLS}") }
    );
    let mut last = 0u64;
    loop {
        let n = served.load(std::sync::atomic::Ordering::Relaxed);
        if n > last {
            for i in last + 1..=n {
                println!("TWOHOST-WEB role B: served cell {i}/{expect_cells}");
            }
            last = n;
        }
        if n >= expect_cells as u64 {
            break;
        }
        assert!(
            Instant::now() < end,
            "role B served only {n} of {expect_cells} requester cells within {:?} \
             (this counts SERVED outcomes, not stream-table rows — a non-zero \
             count here with a stalled total means A stopped requesting, while \
             zero means no request ever reached this node)",
            rig.wait
        );
        thread::sleep(Duration::from_millis(50));
    }
    thread::sleep(Duration::from_secs(10));
    stop.store(true, std::sync::atomic::Ordering::Relaxed);
    serving.stop();
    println!("TWOHOST OK: role B served all {expect_cells} requester cells");
}

// ── Role A: the requester ─────────────────────────────────────────────────

/// One requester cell: its own home (B known as a subnet member under
/// `B_LABEL`, B's address seeded), its own broker on `port_a + offset`, its
/// own loopback listener wired to that broker. Returns the listener port.
struct RequesterCell {
    _home: TempDir,
    _scratch: TempDir,
    _broker: Arc<Broker>,
    port: u16,
    /// The name this cell's broker actually BOUND.
    ///
    /// Carried rather than re-derived because the two were allowed to disagree
    /// once already: the cell bound `seed_socket_name()` while the dispatch loop
    /// below was started on an invented rig name, so nothing served the WanMsg
    /// family, `run_dispatch_loop_observed` failed its very first connect, and
    /// the error went into a `let _ =` on a spawned thread. The symptom was six
    /// "peer unresponsive" dials at the far end and no hint at this one. One
    /// field means there is no second name to keep in step.
    broker_name: String,
}

/// `seed_broker` is for the ONE cell the product dials INTO.
///
/// `request_quoted_paths` reaches its owner through `endpoint::seed_socket_name()`
/// — `spt-daemon-seed-{home_tag}.sock`, the node's canonical daemon socket — so a
/// cell answering on an invented rig name is unreachable by the product's own
/// helper arm, whatever else is right. Measured 2026-09-07: `HELPER_SERVE_FOR
/// outcome=unanswered reason=no daemon to proxy through … (os error 2)`, with
/// the message delivered, the user-msg gate passed and the path found — the
/// whole arm ran and died at the last hop.
///
/// The name is derived from the home, so this cell claims `SPT_HOME` FIRST and
/// takes the name from the product's own function afterwards. Only the helper
/// cell needs it: nothing dials into the other three.
fn requester(rig: &Rig, offset: u16, seed_broker: bool) -> RequesterCell {
    let home = TempDir::new().expect("temp home");
    let scratch = TempDir::new().expect("scratch");
    if seed_broker {
        std::env::set_var("SPT_HOME", home.path());
    }
    let identity = home.path().join("identity");
    let mut subnets = SubnetStore::load_from(&identity.join("subnet.json"));
    subnets
        .add_joined(&rig.subnet, *rig.totp.as_bytes(), 1, None, None)
        .expect("seed subnet");
    subnets.save_to(&identity.join("subnet.json")).expect("save subnets");
    let mut roster = RosterStore::load_from(&identity.join("roster.json"));
    roster.merge_entry(RosterEntry {
        pubkey_hex: rig.b_hex(),
        subnet: rig.subnet.clone(),
        label: B_LABEL.to_owned(),
        machine_id: "rig-web-b-machine".to_owned(),
        address: Some(rig.b_addr()),
        last_seen: "1".to_owned(),
        lease_epoch: 1,
    });
    roster.save_to(&identity.join("roster.json")).expect("save roster");
    spt_store::peeraddrs::PeerAddrStore::record(&identity.join("peer-addrs.json"), &rig.b_hex(), rig.b_addr())
        .expect("seed B's address");

    let broker_name = if seed_broker {
        spt_daemon::endpoint::seed_socket_name()
    } else {
        format!("spt-twohost-web-a{offset}-{}", std::process::id())
    };
    let broker = rig_broker(&broker_name, rig.id_a_for(offset), rig.port_a + offset, scratch.path());
    let port = spt_daemon::docshost::start_serving_with_broker(
        home.path().to_path_buf(),
        home.path().join("docs"),
        "rig-web-a".to_owned(),
        0,
        Some(broker_name.clone()),
    )
    .expect("bind the requester's listener");
    println!(
        "TWOHOST-WEB role A cell {offset}: listener {port}, broker udp {} ({broker_name}), pid {}",
        rig.port_a + offset,
        std::process::id()
    );
    RequesterCell { _home: home, _scratch: scratch, _broker: broker, port, broker_name }
}

/// Fetch through the requester until B answers with something other than
/// the 502 (B may still be coming up), within the rig deadline.
fn fetch_when_up(cell: &RequesterCell, wait: Duration, path: &str, extra: &[(&str, &str)]) -> (u16, String, Vec<u8>) {
    let end = Instant::now() + wait;
    loop {
        let (status, headers, body) = http(cell.port, "GET", path, extra);
        if status != 502 {
            return (status, headers, body);
        }
        assert!(
            Instant::now() < end,
            "role B never came up within {wait:?}: last answer {status}\n{}",
            String::from_utf8_lossy(&body)
        );
        println!("TWOHOST-WEB role A: B not up yet ({})", String::from_utf8_lossy(&body).trim());
        thread::sleep(Duration::from_secs(2));
    }
}

/// The plain fetch: B's `open.md` through A's own listener, byte-equal.
// [int->REQ-WEB-CROSS-NODE-PROXY]
#[test]
fn two_host_web_fetch_role_a() {
    let Some(rig) = Rig::from_env("a") else {
        return;
    };
    let cell = requester(&rig, FETCH_OFFSET, false);
    let (status, headers, body) = fetch_when_up(&cell, rig.wait, &format!("/{B_LABEL}/f/open.md"), &[]);
    assert_eq!(status, 200, "{headers}\n{}", String::from_utf8_lossy(&body));
    assert_eq!(body, OPEN_BYTES, "byte-equal to the file registered on B");
    assert_eq!(header(&headers, "content-type").as_deref(), Some("text/plain; charset=utf-8"));
    assert_eq!(header(&headers, "content-length").as_deref(), Some(OPEN_BYTES.len().to_string().as_str()));
    println!("TWOHOST OK: fetch of B's open.md through A is byte-equal");
}

/// The range: forwarded verbatim, B's 206 + Content-Range carried back.
// [int->REQ-WEB-CROSS-NODE-PROXY]
#[test]
fn two_host_web_range_role_a() {
    let Some(rig) = Rig::from_env("a") else {
        return;
    };
    let cell = requester(&rig, RANGE_OFFSET, false);
    let (status, headers, body) =
        fetch_when_up(&cell, rig.wait, &format!("/{B_LABEL}/f/open.md"), &[("Range", "bytes=0-3")]);
    assert_eq!(status, 206, "{headers}\n{}", String::from_utf8_lossy(&body));
    assert_eq!(
        header(&headers, "content-range").as_deref(),
        Some(format!("bytes 0-3/{}", OPEN_BYTES.len()).as_str())
    );
    assert_eq!(body, &OPEN_BYTES[..4]);
    println!("TWOHOST OK: Range through A is B's 206 with Content-Range");
}

/// The deny: B's `WEB` rule for A's node turns the fetch into 403 whose body
/// names the SURFACE and no sender.
// [int->REQ-WEB-CROSS-NODE-PROXY]
// [int->REQ-WEB-ACCESS-SURFACE]
#[test]
fn two_host_web_deny_role_a() {
    let Some(rig) = Rig::from_env("a") else {
        return;
    };
    let cell = requester(&rig, DENY_OFFSET, false);
    let (status, headers, body) = fetch_when_up(&cell, rig.wait, &format!("/{B_LABEL}/f/locked.md"), &[]);
    let body = String::from_utf8(body).unwrap();
    assert_eq!(status, 403, "{headers}\n{body}");
    assert!(body.starts_with("ACCESS_DENIED: WEB: "), "the body names the surface: {body}");
    assert!(!body.to_ascii_lowercase().contains("sender"), "no sender is named: {body}");
    println!("TWOHOST OK: B's WEB deny rule answers 403 naming WEB through A");
}

/// WEBSERVE W2 — the REQUESTER half of #17's remote helper: the end the
/// requirement is actually written about.
/// [int->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER]
///
/// WHAT THIS ROLE COULD NOT DO BEFORE: receive. A requester cell was a broker
/// and a listener; it had no dispatch loop, so no `WanMessage` could reach it
/// and the WAN-ingress helper — which only ever runs off `receive_wan` — had no
/// way to fire. It grows one here.
///
/// WHY THIS LIVES ON TWO HOSTS AND NOT ON THE ONE-BOX RIG, measured
/// 2026-09-07: the receiver honors an arriving `user-msg` only if it knows the
/// sender's endpoint is a Gateway (REQ-MSG-6), and only gossip teaches that.
/// Turning the pump on in the one-box rig WORKS — 751 ms to converge, twice —
/// but it destroys that rig's routing, because `known_subnet_node` prefers the
/// registry-ADVERTISED label over the roster's, that label is the OS hostname,
/// and one box gives both daemons the same one, so every peer-prefixed URL is
/// answered locally. Gossip-on and node-label routing are mutually exclusive on
/// ONE box, and the arm below needs both. Two hosts carry two hostnames.
///
/// WHAT IS SEEDED HERE, and what is not: this rig runs no pump either, so the
/// rig supplies the RECORD gossip would have carried — but it supplies it
/// THROUGH `apply_feed_flips`, the same entry point a real feed lands on, under
/// the real gate policy. The merge is the product's; only the delivery of the
/// record is the rig's. What is NOT seeded is anything the claim rests on: the
/// message crosses for real, the quoted-path scan is the product's, and the
/// registration happens on B out of B's own `MSG_OUT` row.
#[test]
fn two_host_web_helper_role_a() {
    let Some(rig) = Rig::from_env("a") else {
        return;
    };
    let cell = requester(&rig, HELPER_OFFSET, true);
    let home = cell._home.path().to_path_buf();

    // The `SPT_HOME` this cell runs under is claimed inside `requester` above
    // (`seed_broker: true`), because the broker's NAME has to be derived from it.
    // It is not a convenience either way: WAN ingress resolves the target perch
    // through the process-global home (`wan.rs`: `perch_exists(&msg.target)` then
    // `perch::resolve_perch_path`), taking no path argument, so a cell recording
    // its audience perch in a TempDir the ingress cannot see is answered
    // `NoPerch` — measured 2026-09-07, and B then sat at its registration wait
    // until the deadline.
    //
    // Safe because nextest runs EACH `#[test]` in its own process — PROVEN in
    // this rig's own log, where the four A cells report four distinct pids, not
    // asserted in a comment.

    // The endpoint the user's words were addressed to. A RECORD is what WAN
    // ingress requires (`perch_exists` reads `info.json`); the delivery below
    // still runs the whole product road — access gate, replay check, the
    // user-msg re-stamp gate, and the spool.
    perch_record(&home, HELPER_AUDIENCE, "live");

    // The registry this node judges the arriving message against, and the
    // dispatch loop that judges it. B's Gateway row goes in through the real
    // merge: without it `origin_user_backed` abstains and the `user-msg` is
    // re-stamped down to a plain `msg`, which is a silent no-op rather than a
    // failure (measured on the one-box rig — two reds before a breadcrumb said
    // so out loud).
    // These are the PRODUCT's own constructors, called unmodified. They resolve
    // through the ambient home — which the line above has just made this cell's
    // own — so the rig keeps no second copy of how a daemon builds its registry
    // host or its gate policy. Before that `set_var`, this same code was
    // answered by the BOX's real fleet home: the policy carried none of the
    // rig's subnets, the merge below returned `NotMember`, and the cell red at
    // its own precondition (measured 2026-09-07).
    let registry_host = Arc::new(RegistryHost::new_at(
        &rig.a_hex_for(HELPER_OFFSET),
        EpochSource::load_from(&perch::epoch_file()),
        spt_daemon::presence::registry_snapshot_dir(),
    ));
    let gateway_row = spt_net::net::registry::Instance {
        node: rig.b_hex(),
        status: spt_net::net::registry::Status::Active,
        epoch: 1,
        resources: None,
        last_active_ms: None,
        shell_adapters: Vec::new(),
        node_label: Some(B_LABEL.to_owned()),
        machine_id: Some("rig-web-b-machine".to_owned()),
        endpoint_type: Some(spt_proto::event::GATEWAY_TAG.to_owned()),
        bound: true,
        controller_node: None,
        harness_only: false,
        adapter: None,
        recent_projects: Vec::new(),
        controlled: false,
        activation: 0,
    };
    let (verdicts, _) = registry_host.apply_feed_flips(
        &rig.b_hex(),
        &[spt_net::net::replicate::RegistryUpdate {
            subnet: rig.subnet.clone(),
            endpoint_id: HELPER_GATEWAY.to_owned(),
            instance: gateway_row,
        }],
        &spt_daemon::registryhost::RegistryGatePolicy::load(),
    );
    println!("TWOHOST-WEB role A helper: gateway row merged, verdicts {verdicts:?}");
    assert!(
        registry_host
            .instances_of(HELPER_GATEWAY)
            .iter()
            .any(|i| i.node == rig.b_hex()),
        "PRECONDITION: A must know B's Gateway before the message lands, or the \
         user-msg is re-stamped and the helper never fires"
    );

    let stop = Arc::new(AtomicBool::new(false));
    let scratch = TempDir::new().expect("dispatch scratch");
    {
        // F9: the name the CELL BOUND, never a second spelling of it. With an
        // invented name here nothing binds it, the dispatch loop's first connect
        // fails, and the WanMsg family is never served — the helper request is
        // then never fired at all.
        let name = cell.broker_name.clone();
        let flag = Arc::clone(&stop);
        let paths = DispatchPaths {
            owlery: home.join("owlery"),
            releases: home.join("releases"),
            scratch: scratch.path().join("dispatch"),
        };
        let host = Arc::clone(&registry_host);
        thread::spawn(move || {
            let _ = run_dispatch_loop_observed(
                &name,
                &paths,
                host,
                &flag,
                Duration::from_millis(50),
                Arc::new(AtomicU64::new(0)),
            );
        });
    }

    // THE OBSERVABLE. B's message arrives, this node's ingress scans the user's
    // words, asks B to serve the path they named, and writes down what came
    // back. The helperline gaining that line is what proves the caller fired
    // AFTER delivery and that `request_serve_for` went daemon to daemon —
    // nothing in this cell calls it, because in production its client lives
    // inside this very ingress path. A deadline widened to make a test's own
    // shape work is the test being wrong about where the client lives.
    let audience_perch = home.join("owlery").join(HELPER_AUDIENCE);
    let started = Instant::now();
    let end = started + rig.wait;
    let line = loop {
        if let Some(line) = spt_store::helperline::read_at(&audience_perch).into_iter().next() {
            break line;
        }
        assert!(
            Instant::now() < end,
            "no helper record on {HELPER_AUDIENCE} within {:?} — either B never sent, \
             the message was re-stamped on arrival, or the round trip to B failed",
            rig.wait
        );
        thread::sleep(Duration::from_secs(1));
    };
    assert_eq!(line.msg_id, HELPER_MSG_ID, "the record names the message it came from");
    assert!(
        line.path.ends_with("quoted-by-the-user.md"),
        "and the path the user quoted, which lives on B: {}",
        line.path
    );
    let served_name = line
        .url
        .rsplit('/')
        .next()
        .expect("the minted URL ends in the served name")
        .to_string();
    println!(
        "TWOHOST OK: A's helperline gained {served_name} for msg {} in {} ms",
        line.msg_id,
        started.elapsed().as_millis()
    );

    // (vii) — and this node can actually PULL it. The file is on B, the user is
    // on B, and the endpoint that was told about it is here.
    let (status, headers, body) = fetch_when_up(
        &cell,
        rig.wait,
        &format!("/{B_LABEL}/f/{served_name}"),
        &[],
    );
    assert_eq!(status, 200, "{headers}\n{}", String::from_utf8_lossy(&body));
    assert_eq!(body, HELPER_BYTES, "byte-equal to the file the user pointed at");
    eprintln!("F1_ARM=vii: EXECUTED, the audience's node pulled the quoted file byte-equal");
    println!("TWOHOST OK: the audience's node fetched the user's quoted file byte-equal");
    stop.store(true, std::sync::atomic::Ordering::Relaxed);
}