# PUMP-ON ONE-BOX RIG — the estate's first datum, preserved 2026-09-07 (todlando, W2 / releases#272) # # WHAT THIS PATCH IS: `crates/spt/tests/webserve_cross_node_e2e.rs` with `peer_pump: true`, # a gossip-convergence gate, and a hermeticism assertion. It is NOT applied on the lane. # # WHAT IT MEASURED, and why it is worth keeping: # * A pump-on `spt daemon run` rig CONVERGES. Node A learned node B's Gateway-typed # endpoint by gossip in 751 ms, IDENTICAL across two invocations. No rig in the repo # had ever run the pump before this (every other one pins `peer_pump: false` and calls # it the hermetic kill switch), so "does it work" was unanswered until now. # * Hermeticism is MEASURABLE, not merely arguable: the two daemons dialed exactly # 1 peer, within the pair, seen 10 ms after the meter started. Read off each daemon's # own `PUMP_DIAL_SUBMIT` breadcrumbs, which is what makes "no live-fleet contact" a # number instead of a claim. # # THE TRAP IT COST ME, which is the real reason this file exists: # My FIRST meter read node A ALONE, at 751 ms, saw an EMPTY set, and would have reported # the rig hermetic. A had learned from B's PUSH and had not yet reached its own cadence # tick — so that zero was the zero of a meter that had not ticked, not the zero of a # quiet network. A guard against a false green produced a false green. The fix is in the # patch: read BOTH daemons, and refuse to judge the set until it is non-empty, so the # meter proves its own liveness before it is allowed to prove anything else. # # WHY IT IS NOT ON THE LANE: gossip-on and node-label routing are mutually exclusive on ONE # box. `webserve::known_subnet_node` prefers the registry-ADVERTISED label over the # roster's; `RegistryHost::new` sets that label from `os_hostname()`; both rig daemons share # one box and therefore one hostname; and the prefix dispatcher takes the LOCAL arm whenever # the first segment equals the local node's label. Pump on, every `//f/…` is answered # locally — arm 1 went `404 NO_DOCS_LANDED`. There is no label override to escape with (no # SPT_NODE_LABEL anywhere, no label field in daemon.json; COMPUTERNAME is Windows-only and # `gethostname` has no hook, so that fix would make the rig behave differently per OS), and # `known_subnet_node` matches labels only, so a node-hex prefix is not an escape either. # # Restore with: git apply docs/spikes/pump-on-one-box-rig.patch (against the W2 tip) diff --git a/crates/spt/tests/webserve_cross_node_e2e.rs b/crates/spt/tests/webserve_cross_node_e2e.rs index e85b65e6..8762b774 100644 --- a/crates/spt/tests/webserve_cross_node_e2e.rs +++ b/crates/spt/tests/webserve_cross_node_e2e.rs @@ -24,9 +24,19 @@ //! router needs to open a stream rather than answer the docs 404. The two //! daemons share this box's hostname, so the owner serves the proxied request //! under its OWN prefix (the production rewrite this rig is the reason for). -//! `relay: disabled` + `peer_pump: false` keep both daemons off every path but -//! the seeded direct one; the rig's subnet seed is its own, so the LIVE fleet -//! daemon on this box is refused by seed-proof membership. +//! `relay: disabled` keeps both daemons off every path but the seeded direct +//! one; the rig's subnet seed is its own, so the LIVE fleet daemon on this box +//! is refused by seed-proof membership. +//! +//! **The peer pump is ON here, and it is the only rig that runs it.** WEBSERVE +//! W2's helper arm needs A to LEARN that B's endpoint is a Gateway, and the only +//! thing that teaches a daemon a peer's endpoint type is gossip: `RegistryHost` +//! starts empty and never reads its own snapshots back, so a seeded registry +//! file cannot reach it (measured — the arm's first two reds were the receiving +//! daemon re-stamping the `user-msg` down to a plain `msg` for exactly that +//! reason). Hermeticism is therefore ASSERTED rather than assumed: the pump +//! dials roster rows only, both rosters hold exactly each other, and the arm +//! reads A's own `PUMP_DIAL_SUBMIT` breadcrumbs and refuses any peer but B. //! //! ## The arms (one `#[test]`, sequential — `SPT_HOME` is process-global) //! @@ -62,6 +72,14 @@ const OWNER_ENDPOINT: &str = "w1-servie"; /// The endpoint an ENTRY may be addressed to (WEBSERVE W2). It lives on the /// requesting node A, which is what makes A's fetch the admitted one. const AUDIENCE_ENDPOINT: &str = "w2-audience"; +/// The endpoint on the OWNER node whose USER sends the message that authorizes +/// the register-on-my-behalf. GATEWAY-typed on purpose and not incidentally: +/// 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), so a Gateway is what +/// "a human's words crossed a node boundary" looks like in production. +const SENDER_GATEWAY: &str = "w2-gateway"; +/// An AGENT-typed endpoint on the SAME node — the one property arm (iii) varies. +const SENDER_AGENT: &str = "w2-agent"; fn now_ms() -> u64 { std::time::SystemTime::now() @@ -219,7 +237,11 @@ fn seed_home(home: &Path) -> String { .expect("node identity") .public_key() .to_hex(); - std::fs::write(home.join("daemon.json"), br#"{"relay":"disabled","peer_pump":false}"#) + // `peer_pump: true` — see the module doc: gossip is the ONLY thing that + // teaches A that B's endpoint is a Gateway, and the user-msg gate at WAN + // ingress fail-closes without it. `relay: disabled` stands, and the + // reachable set is asserted below rather than argued. + std::fs::write(home.join("daemon.json"), br#"{"relay":"disabled","peer_pump":true}"#) .expect("write daemon.json"); let mut subnets = SubnetStore::load_from(&perch::subnet_file()); subnets @@ -300,6 +322,114 @@ fn serve_add(spt: &Path, home: &Path, source: &Path, name: &str) { assert!(done.status.success(), "serve add succeeds on the owner: {done:?}"); } +/// A perch with a TYPE — the record `classify_local_origin` reads to decide +/// whether a sender may author a `user-msg` at all. +fn typed_perch(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("write the perch's info record"); +} + +/// A's registry snapshot, seeded with the LABEL ONLY — deliberately. +/// +/// The label is rendezvous: it is what the router resolves a node-prefixed URL +/// through, and the CLI reads it off this file. B's GATEWAY ROW is NOT seeded +/// here, and that omission is the point: the arm's whole claim is that A LEARNS +/// the endpoint type by gossip, and a pre-seeded row would satisfy the +/// convergence poll below without a single packet crossing. A meter that reads +/// green before the thing it measures has happened is not a meter. +fn seed_a_registry(home: &Path, node_b: &str) { + use spt_net::net::registry::SubnetRegistry; + let reg_dir = home.join("identity").join("registry"); + std::fs::create_dir_all(®_dir).expect("A's registry dir"); + let mut reg = SubnetRegistry::new(); + reg.merge_node_label(node_b, OWNER_LABEL.to_owned(), 1); + std::fs::write( + reg_dir.join(format!("{SUBNET}.json")), + serde_json::to_string(®).expect("registry serializes"), + ) + .expect("seed A's registry before boot"); +} + +/// Every distinct peer this daemon's pump has DIALED, read off its own +/// `PUMP_DIAL_SUBMIT` breadcrumbs. The rig's hermeticism claim is measured +/// through this and not argued: the pump dials roster rows, so the set it +/// produces IS the reachable set. +fn dialed_peers(stderr: &str) -> std::collections::BTreeSet { + stderr + .lines() + .filter_map(|line| line.strip_prefix("PUMP_DIAL_SUBMIT:")) + .filter_map(|rest| rest.split(':').next()) + .map(str::to_owned) + .collect() +} + +/// Whether `home`'s registry snapshot yet shows `endpoint` hosted on `node` +/// with the gateway tag — the exact fact the WAN user-msg gate reads. +fn gossiped_gateway(home: &Path, endpoint: &str, node: &str) -> bool { + let path = home + .join("identity") + .join("registry") + .join(format!("{SUBNET}.json")); + let Ok(text) = std::fs::read_to_string(&path) else { + return false; + }; + let Ok(reg) = serde_json::from_str::(&text) else { + return false; + }; + reg.instances(endpoint).iter().any(|i| { + i.node == node && i.endpoint_type.as_deref() == Some(spt_proto::event::GATEWAY_TAG) + }) +} + +/// One REAL `spt send` out of `home`, as the endpoint `as_id`, body on stdin. +/// +/// `user_msg` REQUESTS the user-authored envelope; the CLI honors the request +/// only from a user-backed origin and degrades it loudly otherwise, which is +/// exactly the property arm (iii) varies. +fn send_as( + spt: &Path, + home: &Path, + as_id: &str, + target: &str, + body: &str, + user_msg: bool, +) -> std::process::Output { + let mut args = vec!["send", target]; + if user_msg { + args.push("--user-msg"); + } + let mut child = Command::new(spt) + .no_window() + .args(&args) + .env("SPT_HOME", home) + .env("SPT_AGENT_ID", as_id) + .env_remove("OWL_SESSION_ID") + .env_remove("SPT_ENDPOINT_ID") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn spt send"); + child + .stdin + .take() + .expect("the send's stdin") + .write_all(body.as_bytes()) + .expect("write the message body"); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(child.wait_with_output()); + }); + rx.recv_timeout(Duration::from_secs(60)) + .expect("spt send completes within its deadline") + .expect("collect spt send output") +} + fn access_file_in(home: &Path) -> PathBuf { home.join("identity").join("trust").join("access.json") } @@ -339,6 +469,14 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { let node_a = seed_home(home_a.path()); let node_b = seed_home(home_b.path()); assert_ne!(node_a, node_b, "two homes, two node identities"); + seed_a_registry(home_a.path(), &node_b); + // The perch RECORDS both ends need, written before either daemon boots so + // B's very first advertise scan carries the Gateway's type. A node + // advertises only its own endpoints' types, and a recordless perch + // advertises none — which the funnel fail-closes into a re-stamp. + typed_perch(home_a.path(), AUDIENCE_ENDPOINT, "live"); + typed_perch(home_b.path(), SENDER_GATEWAY, spt_proto::event::GATEWAY_TAG); + typed_perch(home_b.path(), SENDER_AGENT, "live"); let dae_a = DaemonGuard::start(home_a.path(), &spt, "a"); let mut dae_b = DaemonGuard::start(home_b.path(), &spt, "b"); @@ -357,6 +495,7 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { addr_b.clone(), ) .expect("seed B's address into A"); + let addr_a_row = addr_a.clone(); spt_store::peeraddrs::PeerAddrStore::record( &home_b.path().join("identity").join("peer-addrs.json"), &node_a, @@ -379,16 +518,85 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { lease_epoch: 1, }); roster.save_to(&identity.join("roster.json")).expect("seed A's roster"); - let mut reg = spt_net::net::registry::SubnetRegistry::new(); - reg.merge_node_label(&node_b, OWNER_LABEL.to_owned(), 1); - let reg_dir = identity.join("registry"); - std::fs::create_dir_all(®_dir).expect("registry dir"); - std::fs::write( - reg_dir.join(format!("{SUBNET}.json")), - serde_json::to_string(®).expect("registry serializes"), - ) - .expect("seed A's view of B"); + // And B's roster carries A. The pump dials ROSTER ROWS, so this is both + // what makes the gossip bidirectional and what BOUNDS it: each side's + // roster holds exactly the other, and nothing else is reachable. + let identity_b = home_b.path().join("identity"); + let mut roster_b = RosterStore::load_from(&identity_b.join("roster.json")); + roster_b.merge_entry(RosterEntry { + pubkey_hex: node_a.clone(), + subnet: SUBNET.to_owned(), + label: "rig-a".to_owned(), + machine_id: "rig-a-machine".to_owned(), + address: Some(addr_a_row), + last_seen: "1".to_owned(), + lease_epoch: 1, + }); + roster_b.save_to(&identity_b.join("roster.json")).expect("seed B's roster"); + // A's REGISTRY half of this seeding ran before the daemon booted + // (`seed_a_registry`) — the in-memory copy the WAN gate reads is loaded + // at start, so it cannot be written here. + } + + // ══ GOSSIP CONVERGENCE, and the hermeticism this rig ASSERTS ═══════════ + // A must LEARN that B's endpoint is a Gateway before any `user-msg` from it + // is honored (REQ-MSG-6, read at WAN ingress). Waited for on the FACT rather + // than slept on, and the wait's number goes in the raw so a slow convergence + // is visible as a number instead of arriving as a timeout one day. + let gossip_started = Instant::now(); + let gossip_deadline = gossip_started + Duration::from_secs(180); + while !gossiped_gateway(home_a.path(), SENDER_GATEWAY, &node_b) { + assert!( + Instant::now() < gossip_deadline, + "PRECONDITION: A never learned B's Gateway by gossip in {:?}.\n--- A ---\n{}\n--- B ---\n{}", + gossip_started.elapsed(), + common::daemon_stderr_panel(&home_a.path().join("brain-a.stderr.log")), + common::daemon_stderr_panel(&home_b.path().join("brain-b.stderr.log")) + ); + std::thread::sleep(Duration::from_millis(250)); } + eprintln!( + "=== gossip converged === A learned {SENDER_GATEWAY}@{OWNER_LABEL} as gateway in {} ms", + gossip_started.elapsed().as_millis() + ); + // The kill switch is off, so its PURPOSE — no contact with anything but the + // other rig daemon — is measured instead of assumed. This box hosts the live + // fleet; a pump that reached it would show up here as a second hex. + // Read across BOTH daemons and WAIT for the set to be non-empty before + // judging it. Measured the hard way: the first version read A alone 751 ms + // in, saw {} and called the rig hermetic — but A had learned from B's push + // and had not yet reached its own cadence tick, so that zero was the zero of + // a meter that had not ticked, which is the same false green this rig's + // population floor exists to prevent. A non-empty union is the meter's own + // proof of life; the subset is the claim. + let reach_started = Instant::now(); + let reach_deadline = reach_started + Duration::from_secs(90); + let both_peers = loop { + let mut peers = dialed_peers(&dae_a.stderr()); + peers.extend(dialed_peers(&dae_b.stderr())); + if !peers.is_empty() { + break peers; + } + assert!( + Instant::now() < reach_deadline, + "neither pump dialed anything in {:?} — the reachable-set meter never ticked, so it can prove nothing in either direction", + reach_started.elapsed() + ); + std::thread::sleep(Duration::from_millis(250)); + }; + eprintln!( + "=== the pair's reachable set === {} peer(s) in {} ms: {:?}", + both_peers.len(), + reach_started.elapsed().as_millis(), + both_peers + ); + // This box hosts the LIVE fleet. A pump that reached it would show up here + // as a third hex, which is what turns the hermetic claim into a measurement. + let pair = [node_a.clone(), node_b.clone()]; + assert!( + both_peers.iter().all(|p| pair.contains(p)), + "the two rig daemons dialed NOTHING but each other — anything else is fleet contact: {both_peers:?} against {pair:?}" + ); // ══ THE SERVED FILE, registered on B by the owner endpoint ════════════ let source = home_b.path().join("report.md"); @@ -561,173 +769,220 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { assert_eq!(status, 200, "audience cleared, the fetch is open again"); } - // ══ ARM 6c — REGISTER-ON-MY-BEHALF across the two daemons ══════════════ + // ══ ARM 6c — REGISTER-ON-MY-BEHALF, END TO END through BOTH daemons ════ // [int->REQ-NOW-SIGNAL-FILE-ACCESS-HELPER] + // // The case #17 was minted for: the user who quoted a filepath is on the // OTHER node, so the file is there and the registration has to happen - // there. A asks B to serve one of B's own paths for an endpoint A hosts. + // there. Driven the way production drives it and no other way — ONE real + // `spt send` from a Gateway on B to an endpoint hosted on A, its body + // quoting an absolute path that exists on B — and then WAITED FOR on A's + // helperline store, which is the observable proving the ingress caller + // fired after delivery and that `request_serve_for` went daemon to daemon. + // + // NOTHING HERE CALLS `request_serve_for`. Its client lives INSIDE the + // receiving daemon's WAN ingress, exactly as the fetch path's client does; + // an arm that drove it from the test process was measuring a client that + // does not exist in production, and it ate the brain IPC read deadline + // saying so. A deadline I would have to widen to make my own test shape + // work is the test being wrong about where the client lives. // - // THE AUTHORIZATION IS THE POINT OF THIS ARM. `WEB` is default-on inside a - // subnet, so admitting the request on the surface gate plus "the audience - // lives on the asking node" would let ANY member have B expose ANY absolute - // path on B's disk. What makes the exposure legitimate is that a USER ON B - // said so, and B proves that from its OWN `MSG_OUT` row rather than from - // anything A sends. Each conjunct gets its own refusal here, and every - // refusal is checked to have registered NOTHING — a refusal that still - // exposed the file would pass a check that only read the answer. + // WHAT IS NOT HERE, AND WHERE IT WENT. The five authorization conjuncts are + // `spt_store::msgid::user_authorized_serve_for` — pure, with one unit cell + // per conjunct. Four of them CANNOT be produced end to end at all: the + // receiving daemon only ever asks for a path the user's own words carried, + // addressed to that message's own target, under that message's own id, so + // driving them from here would mean forging a request the product cannot + // make. (iii) is the exception and stays, because it is a DIFFERENT gate — + // the receiver's own "only a user-msg triggers this" — and nothing else + // exercises it. { let quoted = home_b.path().join("quoted-by-the-user.md"); - std::fs::write("ed, b"# the file the user pointed at\n").unwrap(); + let quoted_bytes = b"# the file the user pointed at\n"; + std::fs::write("ed, quoted_bytes).unwrap(); let quoted_text = quoted.to_str().unwrap().to_string(); - let target = spt_daemon::webproxy::ProxyTarget { - node_label: OWNER_LABEL.to_string(), - node_hex: node_b.clone(), - }; - let ttl = spt_store::serving::HELPER_ENTRY_TTL_MS; - - /// B's own record of a message its user sent to `to`, carrying `words`. - fn plant_user_msg_out(home: &Path, short_id: &str, to: &str, words: &str) { - let perch = home.join("owlery").join("the-operator"); - std::fs::create_dir_all(&perch).unwrap(); - let body = spt_proto::event::compose_typed_event( - spt_proto::event::EVENT_TYPE_USER_MSG, - &[("from", "the-operator"), (spt_proto::event::EVENT_ATTR_MSG_ID, short_id)], - words, - ); - spt_store::iolog::append_at( - &perch, - &spt_store::iolog::IoLogRow { - seq: 0, - at_ms: 1, - kind: spt_proto::ioevent::IO_KIND_MSG_OUT.to_string(), - payload: body, - truncated: false, - digest_seq: None, - peer: Some(to.to_string()), - msg_id: Some(short_id.to_string()), - mid: false, - }, - ) - .unwrap(); - } + let words = format!("have a look at {quoted_text} when you get a moment"); - let ask = |origin: Option<&str>, audience: &str, path: &str| { - let origin = origin.map(str::to_owned); - let audience = audience.to_owned(); - let path = path.to_owned(); - in_home(home_a.path(), || { - spt_daemon::webproxy::request_serve_for( - &spt_daemon::endpoint::seed_socket_name(), - home_a.path(), - &target, - &path, - &audience, - ttl, - origin, - ) - }) - .expect("the owner answered") - }; - let registered_names = || { + // The three perch RECORDS this arm needs were written before either + // daemon booted (B has to advertise the Gateway's type on its first + // scan). They are records rather than brought-up endpoints because no + // claim here is about bringup: what is bypassed is the CREATION + // ceremony — WAN ingress asks `perch_exists`, which reads `info.json` + // (measured: a bare directory answers WAN_NO_PERCH). Everything on the + // message's own road still runs: resolution, the QUIC leg, A's access + // gate, the replay check, the user-msg restamp gate, and the spool. + + let audience_perch = home_a.path().join("owlery").join(AUDIENCE_ENDPOINT); + let helper_lines = || spt_store::helperline::read_at(&audience_perch); + let registry_text = || { std::fs::read_to_string(spt_store::perch::serving_registry_file_in(home_b.path())) .unwrap_or_default() }; - let refused = |outcome: spt_daemon::webproxy::ServeForOutcome, why: &str| { - match outcome { - spt_daemon::webproxy::ServeForOutcome::Refused { message, refused } => { - assert!(refused, "{why}: an access decision, not a failure: {message}"); - assert!(message.starts_with("ACCESS_DENIED: WEB: "), "{why}: {message}"); - } - other => panic!("{why}: must be refused, got {other:?}"), - } + assert!(helper_lines().is_empty(), "PRECONDITION: no helper record before this arm"); + assert!( + !registry_text().contains("quoted-by-the-user"), + "PRECONDITION: B has not registered the quoted file" + ); + + // ── (iii) AN AGENT'S message — the receiver's own gate ───────────── + // Same node, same target, same words, same absolute path. The ONE + // varied property is WHO AUTHORED IT: only a `user-msg` — the envelope + // whose authorship the sending daemon proved to be its human — is an + // authority ADR-0058 Amendment 1 will act on. + // + // Judged FIRST, and its zero is not a bare zero: the arm WAITS for A's + // own decline breadcrumb, so what is asserted is "the ingress helper ran + // and declined", not "nothing has happened yet". That ordering is also + // what makes the caller's falsifier readable — with `request_serve_for` + // no-opped this arm still passes, because the decline happens before the + // caller is ever reached, and (vi) below then dies at its wait. + let agent_send = send_as(&spt, home_b.path(), SENDER_AGENT, AUDIENCE_ENDPOINT, &words, false); + eprintln!( + "=== (iii) agent send (B->A) ===\nstatus={}\n{}\n--stderr--\n{}", + agent_send.status, + String::from_utf8_lossy(&agent_send.stdout), + String::from_utf8_lossy(&agent_send.stderr) + ); + assert!( + agent_send.status.success(), + "an agent's message is DELIVERED — it is simply not acted on: {agent_send:?}" + ); + let declined_started = Instant::now(); + let declined_deadline = declined_started + Duration::from_secs(60); + let declined_marker = "outcome=declined"; + while !dae_a.stderr().contains(declined_marker) { assert!( - !registered_names().contains("quoted-by-the-user.md"), - "{why}: a refused request registers NOTHING" + Instant::now() < declined_deadline, + "A never said why it declined the agent's message in {:?} — without that \ + line this arm's zero cannot be told from an ingress helper that never \ + ran.\n--- A ---\n{}", + declined_started.elapsed(), + common::daemon_stderr_panel(&home_a.path().join("brain-a.stderr.log")) ); - }; - - // (i) NO ORIGIN AT ALL — nothing to bind the request to a user's words. - refused(ask(None, AUDIENCE_ENDPOINT, "ed_text), "no origin"); - - // (ii) AN ORIGIN THIS NODE NEVER SENT. B has no such row, so the - // requester's id buys it nothing. - refused(ask(Some("ZZZZZZZZ"), AUDIENCE_ENDPOINT, "ed_text), "unknown origin"); - - // From here B has a real record: its user told the audience about the file. - plant_user_msg_out( - home_b.path(), - "BCDFGH23", - AUDIENCE_ENDPOINT, - &format!("have a look at {quoted_text} when you get a moment"), + std::thread::sleep(Duration::from_millis(200)); + } + let declined_lines: Vec = dae_a + .stderr() + .lines() + .filter(|line| line.contains(declined_marker)) + .map(str::to_owned) + .collect(); + let lines_after_iii = helper_lines(); + let registrations_after_iii = registry_text().matches("quoted-by-the-user").count(); + assert!( + lines_after_iii.is_empty(), + "(iii) an agent-authored message leaves NO helper record: {lines_after_iii:?}" + ); + assert_eq!( + registrations_after_iii, 0, + "(iii) and B registered nothing on its behalf" + ); + eprintln!( + "F1_ARM=iii verdict=refused helper_lines={} registrations={registrations_after_iii} \ + waited_ms={} case=agent-authored message fires no request; A said: {}", + lines_after_iii.len(), + declined_started.elapsed().as_millis(), + declined_lines.last().map(String::as_str).unwrap_or("") ); - // (iii) AN AGENT'S message, not a user's. Same id shape, same node, but - // a plain `msg` envelope carries no proven human behind it. - { - let perch = home_b.path().join("owlery").join("some-agent"); - std::fs::create_dir_all(&perch).unwrap(); - spt_store::iolog::append_at( - &perch, - &spt_store::iolog::IoLogRow { - seq: 0, - at_ms: 1, - kind: spt_proto::ioevent::IO_KIND_MSG_OUT.to_string(), - payload: spt_proto::event::compose_msg_event( - "some-agent", - &format!("I read {quoted_text} earlier"), - ), - truncated: false, - digest_seq: None, - peer: Some(AUDIENCE_ENDPOINT.to_string()), - msg_id: Some("AGENTMSG".to_string()), - mid: false, - }, - ) - .unwrap(); - } - refused(ask(Some("AGENTMSG"), AUDIENCE_ENDPOINT, "ed_text), "agent-sent origin"); - - // (iv) THE WRONG AUDIENCE: the user's message went to the audience - // endpoint, and a request naming a different one is not what they said. - refused(ask(Some("BCDFGH23"), "somebody-else", "ed_text), "audience mismatch"); - - // (v) A PATH THE USER NEVER NAMED — the conjunct that stops a real - // message being used as a key to the rest of the disk. - let elsewhere = home_b.path().join("never-mentioned.md"); - std::fs::write(&elsewhere, b"private\n").unwrap(); - refused( - ask(Some("BCDFGH23"), AUDIENCE_ENDPOINT, elsewhere.to_str().unwrap()), - "path not in the user's words", + // ── (vi) THE POSITIVE ────────────────────────────────────────────── + let user_send = send_as(&spt, home_b.path(), SENDER_GATEWAY, AUDIENCE_ENDPOINT, &words, true); + eprintln!( + "=== (vi) user send (B->A) ===\nstatus={}\n{}\n--stderr--\n{}", + user_send.status, + String::from_utf8_lossy(&user_send.stdout), + String::from_utf8_lossy(&user_send.stderr) + ); + assert!(user_send.status.success(), "the user's message is delivered: {user_send:?}"); + assert!( + !String::from_utf8_lossy(&user_send.stderr).contains("USER_MSG_RESTAMP"), + "PRECONDITION: the Gateway's user-msg was NOT degraded on the way OUT — a \ + restamped send would make every assertion below vacuous:\n{}", + String::from_utf8_lossy(&user_send.stderr) ); - // (vi) THE POSITIVE: every conjunct holds, so B registers — 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. - let (url, served_name) = match ask(Some("BCDFGH23"), AUDIENCE_ENDPOINT, "ed_text) { - spt_daemon::webproxy::ServeForOutcome::Registered { url, served_name } => (url, served_name), - other => panic!("the user's own request must be admitted, got {other:?}"), + // The helper request runs on its OWN THREAD, after delivery, and crosses + // the wire — so the observable is waited for rather than read once. The + // panel carries what both stores held when the wait expired: that is what + // names the mechanism (a decline reason, a refusal, or silence) instead of + // leaving a bare timeout. + let wait_started = Instant::now(); + let deadline = wait_started + Duration::from_secs(90); + let line = loop { + if let Some(line) = helper_lines().into_iter().next() { + break line; + } + assert!( + Instant::now() < deadline, + "A's helperline never gained the fetch line in {:?} — the ingress caller did \ + not complete a round trip to B.\nhelper_lines={} registrations={}\n\ + --- A ---\n{}\n--- B ---\n{}", + wait_started.elapsed(), + helper_lines().len(), + registry_text().matches("quoted-by-the-user").count(), + common::daemon_stderr_panel(&home_a.path().join("brain-a.stderr.log")), + common::daemon_stderr_panel(&home_b.path().join("brain-b.stderr.log")) + ); + std::thread::sleep(Duration::from_millis(250)); }; - assert!(url.contains(&served_name), "the answer names the URL it minted: {url}"); + assert_eq!(line.path, quoted_text, "the record names the path the user quoted"); + assert!(!line.msg_id.is_empty(), "and the message it came from, for the audit trail"); + let served_name = line + .url + .rsplit('/') + .next() + .expect("the minted URL ends in the served name") + .to_string(); + // B REGISTERED it, and registered it as 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. let registry = spt_store::serving::ServingRegistry::load_at( &spt_store::perch::serving_registry_file_in(home_b.path()), ) .unwrap(); - let entry = registry.get(&served_name).expect("B holds the entry it just minted"); + let entry = registry.get(&served_name).expect("B holds the entry it minted"); assert_eq!(entry.kind, spt_store::serving::ServedKind::File, "a live reference, not a snapshot"); - assert_eq!(entry.ttl_ms, Some(ttl), "24h"); + assert_eq!(entry.ttl_ms, Some(spt_store::serving::HELPER_ENTRY_TTL_MS), "24h"); assert_eq!(entry.audience.as_deref(), Some(AUDIENCE_ENDPOINT)); assert_eq!( entry.origin.as_deref(), - Some("BCDFGH23"), + Some(line.msg_id.as_str()), "origin is the message short-ID — the audit trail back to the user's own words" ); + // Both messages quoted the SAME path, so a fired (iii) would have left a + // SECOND record here: the count is the assertion, not the presence. + assert_eq!( + helper_lines().len(), + 1, + "exactly ONE helper record exists — the user's; the agent's added none" + ); + assert_eq!( + registry_text().matches("quoted-by-the-user").count(), + 1, + "and B registered the file exactly once — for the user, not for the agent" + ); + eprintln!( + "F1_ARM=vi verdict=registered kind={:?} ttl_ms={:?} audience={:?} origin={:?} \ + msg_id={} waited_ms={}", + entry.kind, + entry.ttl_ms, + entry.audience, + entry.origin, + line.msg_id, + wait_started.elapsed().as_millis() + ); - // (vii) And the named endpoint's node can actually pull it, which is the - // end the requirement is written about. + // ── (vii) THE AUDIENCE'S NODE PULLS IT ───────────────────────────── + // The end the requirement is written about: the file is on B, the user + // is on B, and the endpoint that was told about it is on A. let (status, _, body) = http(port_a, "GET", &format!("/{OWNER_LABEL}/f/{served_name}"), &[]); assert_eq!(status, 200, "the audience's node fetches what was registered for it"); - assert_eq!(body, b"# the file the user pointed at\n"); + assert_eq!(body, quoted_bytes, "byte-equal to the file the user pointed at"); + eprintln!( + "F1_ARM=vii verdict=fetched status={status} bytes={} served_name={served_name}", + body.len() + ); } // ══ ARM 7 — unknown label = the docs compatibility 404; reserved facet