diff --git a/crates/spt-daemon/src/iobus.rs b/crates/spt-daemon/src/iobus.rs index 80c25f0d..2e737925 100644 --- a/crates/spt-daemon/src/iobus.rs +++ b/crates/spt-daemon/src/iobus.rs @@ -296,6 +296,19 @@ impl IoSink for IoLogSink { let (payload, truncated) = bound_payload(&ev.payload); let perch_path = spt_store::perch::resolve_perch_path(&ev.owner, spt_store::perch::ParentHint::Infer); + { + let line = format!( + "RCA_SINK: pid={} owner={:?} kind={:?} peer={:?} path={:?} spt_home={:?} +", + std::process::id(), ev.owner, ev.kind, ev.peer, perch_path, + std::env::var("SPT_HOME").ok() + ); + use std::io::Write as _; + let p = std::env::var("RCA_LOG").unwrap_or_else(|_| "/tmp/rca236.log".to_string()); + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&p) { + let _ = f.write_all(line.as_bytes()); + } + } spt_store::iolog::append_at( &perch_path, &spt_store::iolog::IoLogRow { diff --git a/crates/spt/src/api/delivery.rs b/crates/spt/src/api/delivery.rs index 4b505824..04107d92 100644 --- a/crates/spt/src/api/delivery.rs +++ b/crates/spt/src/api/delivery.rs @@ -595,6 +595,18 @@ pub fn poll_drain(id: &str, include_deferred: bool) -> Vec { /// neither can affect its caller (REQ-IO-EVENT-BUS). // [impl->REQ-IO-MSG-EDGES] fn publish_msg_io_local(id: &str, kind: &str, peer: &str, body: &str) { + { + let line = format!( + "RCA_DELIV: pid={} id={:?} kind={:?} peer={:?} +", + std::process::id(), id, kind, peer + ); + use std::io::Write as _; + let p = std::env::var("RCA_LOG").unwrap_or_else(|_| "/tmp/rca236.log".to_string()); + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&p) { + let _ = f.write_all(line.as_bytes()); + } + } let owlery = perch::owlery_dir(); let bus = spt_daemon::iobus::default_bus(&owlery, crate::api::nowsignal::now_ms()); // The SENDER rides along (W5): LAST_MSGS answers "who", and this edge knows. diff --git a/crates/spt/src/cli.rs b/crates/spt/src/cli.rs index 8cbc697f..7eeca984 100644 --- a/crates/spt/src/cli.rs +++ b/crates/spt/src/cli.rs @@ -11290,6 +11290,18 @@ pub(crate) fn cmd_send_verdict( // an explicit `--from` is a reply label the caller chose. Same // reasoning as the admission gate's subject, one layer over. let owner = sender_proven.as_deref().unwrap_or(from.as_str()); + { + let line = format!( + "RCA_SEND: pid={} owner={:?} sender_proven={:?} session_proven={:?} from={:?} target={:?} +", + std::process::id(), owner, sender_proven, session_proven, from, target + ); + use std::io::Write as _; + let p = std::env::var("RCA_LOG").unwrap_or_else(|_| "/tmp/rca236.log".to_string()); + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&p) { + let _ = f.write_all(line.as_bytes()); + } + } publish_msg_io(owner, spt_proto::ioevent::IO_KIND_MSG_OUT, &target, body); } diff --git a/crates/spt/src/roster.rs b/crates/spt/src/roster.rs index ea691b13..3cc3a61c 100644 --- a/crates/spt/src/roster.rs +++ b/crates/spt/src/roster.rs @@ -279,6 +279,19 @@ pub fn detect_self_by_ancestry() -> Option { } } } + { + let chosen = match_self_by_ancestry(&ancestry, &candidates); + let line = format!( + "RCA_ANC: pid={} ancestry={:?} candidates={:?} chosen={:?} +", + std::process::id(), ancestry, candidates, chosen + ); + use std::io::Write as _; + let p = std::env::var("RCA_LOG").unwrap_or_else(|_| "/tmp/rca236.log".to_string()); + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&p) { + let _ = f.write_all(line.as_bytes()); + } + } match_self_by_ancestry(&ancestry, &candidates) } diff --git a/crates/spt/tests/io_events_undriven_kinds_e2e.rs b/crates/spt/tests/io_events_undriven_kinds_e2e.rs index c225c918..66c27e23 100644 --- a/crates/spt/tests/io_events_undriven_kinds_e2e.rs +++ b/crates/spt/tests/io_events_undriven_kinds_e2e.rs @@ -1,402 +1,428 @@ -//! CONDUIT W3 RESPIN (releases#234) — the IO kinds whose publish → log → poll -//! path was never driven end to end. -//! -//! ## Why this leg exists, stated as the defect it closes -//! -//! W3 shipped, gated and reached a golden head asserting in four places — the -//! REQ title, `api.md`, `CONTEXT.md`, the integration checklist — that **all six -//! emitted kinds are visible** through `spt api io-events`. Two were -//! structurally impossible: `publish_commune_io` hand-built an `IoBus` with one -//! sink instead of calling `default_bus`, so `IoLogSink` was never registered on -//! the commune path and no poll could ever answer with `COMMUNE` or -//! `COMMUNE_FAIL`. -//! -//! Nothing in the battery could see it. The `io_events_poll_e2e` cell drives -//! `USER_INPUT` and `AGENT_OUTPUT` only, so it was VACUOUS on exactly the two -//! broken kinds; the unit tests synthesize `IoLogRow` values directly and never -//! traverse the publish path where a missing registration lives; and -//! `traceable-reqs` was green on TAGS, not on truth. **A claim that enumerates -//! is a test plan, and this file is the missing half of it.** -//! -//! ## Width, ruled by doyle -//! -//! Cover every kind whose publish → log → poll path is UNDRIVEN; skip kinds an -//! existing non-vacuous cell already drives. -//! -//! * `COMMUNE` — the broken kind. Mandatory. -//! * `COMMUNE_FAIL` — cheap here, so it rides (see the induction note below). -//! * `MSG_IN` / `MSG_OUT` — structurally fine and never driven, which is the -//! precise state that cost this milestone a respin. Mandatory. -//! * `USER_INPUT` / `AGENT_OUTPUT` — EXCLUDED as duplicate denominators. -//! Discharged at source, not from memory: `io_events_poll_e2e` drives -//! `api state busy|idle --payload-file` (publish) and then asserts both -//! kinds AND their verbatim payloads back out of `api io-events` (poll), so -//! that path is publish → log → poll and adding it here would only add -//! first-execution cells to the respin golden without adding coverage. -//! -//! ## The COMMUNE_FAIL induction, and a correction carried forward -//! -//! The failure is induced by making the drop UNREADABLE (replacing the file with -//! a directory): `ingest_drops` reads the body with `?` before it decides -//! anything, so the read fails the sweep, no tier is touched and no unlink is -//! reached. **The `index.lock` class does NOT fail this seam** — a previous -//! measurer planted non-empty locks in both places git takes one, asserted the -//! plant count, and the ingest committed anyway -//! (`spt-daemon/tests/commune_io_events_int.rs`). That matters beyond this file: -//! the field incident `COMMUNE_FAIL` was built for is ATTRIBUTED to index.lock, -//! and that attribution is unreproduced. -//! -//! cargo nextest run -p spt --test io_events_undriven_kinds_e2e --test-threads 1 -//! -//! Each cell owns its `SPT_HOME`, its daemon and its endpoints, so they share no -//! state and may run in any order. -// -// [int->REQ-IO-EVENT-ADAPTER-LOG] -// [int->REQ-IO-EVENT-POLL-VERB] -// [int->REQ-IO-COMMUNE-EVENT] -// [int->REQ-IO-COMMUNE-FAIL-EVENT] -// [int->REQ-IO-MSG-EDGES] - -use std::io::Write; -use std::path::{Path, PathBuf}; -use std::process::{Command, Output, Stdio}; -use std::time::Duration; - -mod common; -use common::reap; -use common::CommandNoWindowExt; - -use spt_daemon::{BrainLifecycle, DaemonConfig}; -use spt_runtime::Manifest; -use spt_store::perch::{self, ParentHint}; - -#[cfg(windows)] -const NOOP_CMD: &str = "cmd /C rem"; -#[cfg(unix)] -const NOOP_CMD: &str = "true"; - -fn start_real_daemon(spt_bin: &Path, home: &Path) { - let mut cmd = Command::new(spt_bin); - cmd.no_window() - .args(["daemon", "start"]) - .env("SPT_HOME", home) - .env_remove("SPT_AGENT_ID") - .env_remove("SPT_ENDPOINT_ID") - .env_remove("OWL_SESSION_ID"); - let out = common::output_bounded(cmd, Duration::from_secs(90)); - assert!( - out.status.success(), - "explicit daemon start failed: {}", - String::from_utf8_lossy(&out.stderr) - ); -} - -/// A live-capable manifest whose commune drop dir is `dir` — the shape the -/// lifecycle's own tests use. -fn live_manifest(dir: &Path) -> Manifest { - let d = dir.to_string_lossy().replace('\\', "/"); - let toml = format!( - "[adapter]\nname=\"mock\"\nversion=\"1\"\nmin_spt_core_version=\"1\"\n\n\ - [session]\ncommune_dir=\"{d}\"\n\n\ - [session.psyche_init]\ncommand='{NOOP_CMD}'\ncwd=\"{{psyche_dir}}\"\nkeys=[]\n" - ); - Manifest::from_toml_str(&toml).unwrap() -} - -/// Bind an endpoint perch this process can authenticate as. -/// -/// The `cwd` is load-bearing rather than boilerplate: with no recorded cwd the -/// project id is empty, a drop carrying a `` slice takes the -/// deferred branch and is PRESERVED instead of consumed — which emits nothing. -/// Wiring a real cwd is what puts the commune arm on the CONSUMPTION path. -fn bind_endpoint(id: &str, sid: &str, cwd: Option<&Path>) -> PathBuf { - let p = perch::resolve_perch_path(id, ParentHint::Infer); - std::fs::create_dir_all(&p).unwrap(); - let mut rec = spt_store::info::InfoJson::new(id, "0", std::process::id(), sid, "live_agent"); - if let Some(c) = cwd { - rec.cwd = Some(c.to_string_lossy().to_string()); - } - spt_store::info::write_info(&p, &rec).unwrap(); - p -} - -/// One `spt api io-events` poll, parsed. `--session-id` is both the auth proof -/// and the cursor key. -fn poll(spt_bin: &Path, home: &Path, id: &str, sid: &str, extra: &[&str]) -> serde_json::Value { - let mut cmd = Command::new(spt_bin); - cmd.no_window() - .args(["--json", "api", "io-events", id, "--session-id", sid]) - .args(extra) - .env("SPT_HOME", home); - let out = common::output_bounded(cmd, Duration::from_secs(60)); - assert!( - out.status.success(), - "poll of {id} failed: {}", - String::from_utf8_lossy(&out.stderr) - ); - let text = String::from_utf8_lossy(&out.stdout).to_string(); - serde_json::from_str(&text) - .unwrap_or_else(|e| panic!("poll output is not JSON ({e}): {text:?}")) -} - -/// `spt send` reads its body from STDIN — the idiom `active_only_never_relay_e2e` -/// established, reused rather than re-invented. -fn send_bounded(spt_bin: &Path, home: &Path, args: &[&str], body: &str) -> Output { - let mut child = Command::new(spt_bin) - .no_window() - .args(args) - .env("SPT_HOME", home) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .expect("spawn spt send"); - child - .stdin - .take() - .unwrap() - .write_all(body.as_bytes()) - .expect("write stdin 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 timed out") - .expect("spt send output") -} - -fn rows(v: &serde_json::Value) -> Vec<(String, String, Option)> { - v["events"] - .as_array() - .expect("events array") - .iter() - .map(|e| { - ( - e["kind"].as_str().unwrap_or_default().to_string(), - e["payload"].as_str().unwrap_or_default().to_string(), - e["peer"].as_str().map(str::to_string), - ) - }) - .collect() -} -const AUTHOR: &str = "alice"; -const PEER: &str = "bob"; -const SID_A: &str = "io-undriven-sid-a"; -const SID_B: &str = "io-undriven-sid-b"; - -/// The commune body is deliberately awkward: a `!!wake!!` pair the adapter owns -/// and core must not touch, a `` slice routing to a different -/// tier than the live text, and trailing whitespace. -const COMMUNE_BODY: &str = "live half !!wake!! resume T2c !!wake!!\n\nproject half\n\n \n"; - -/// One isolated rig: own `SPT_HOME`, own daemon, own endpoints. -/// -/// The `TempDir`s are held so they outlive the cell — dropping them early -/// deletes the home out from under a running daemon. -struct Rig { - home: tempfile::TempDir, - _project: tempfile::TempDir, - drops: tempfile::TempDir, - spt_bin: PathBuf, - manifest: Manifest, -} - -fn rig() -> Rig { - let home = tempfile::tempdir().unwrap(); - std::env::set_var("SPT_HOME", home.path()); - std::env::remove_var("SPT_AGENT_ID"); - std::env::remove_var("SPT_ENDPOINT_ID"); - std::env::remove_var("OWL_SESSION_ID"); - spt_store::nodeid::load_or_create().expect("node identity"); - let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt")); - start_real_daemon(&spt_bin, home.path()); - - let project = tempfile::tempdir().unwrap(); - bind_endpoint(AUTHOR, SID_A, Some(project.path())); - bind_endpoint(PEER, SID_B, Some(project.path())); - let drops = tempfile::tempdir().unwrap(); - let manifest = live_manifest(drops.path()); - Rig { - home, - _project: project, - drops, - spt_bin, - manifest, - } -} - -impl Rig { - fn drop_path(&self) -> PathBuf { - self.drops.path().join(format!("{AUTHOR}-commune.md")) - } - - /// Seed a cursor first, so every row a cell asserts is one that cell caused. - fn seed(&self, id: &str, sid: &str) { - assert_eq!( - poll(&self.spt_bin, self.home.path(), id, sid, &[])["seeded"], - true, - "PRECONDITION: {id}'s cursor seeds on its first poll" - ); - } - - fn reap(&self, label: &str) { - let observed = reap::breadcrumb_daemon_pid(self.home.path()) - .and_then(|pid| reap::observe(label, pid)); - let mut stop = Command::new(&self.spt_bin); - stop.no_window() - .args(["daemon", "stop"]) - .env("SPT_HOME", self.home.path()); - let stopped = common::output_bounded(stop, Duration::from_secs(30)); - eprintln!( - "REAP[{label}]: daemon stop status={} stderr={:?}", - stopped.status, - String::from_utf8_lossy(&stopped.stderr).trim() - ); - reap::reap_breadcrumb_daemon(label, self.home.path(), &self.spt_bin, observed); - } -} - -/// Run a cell against a fresh rig, reaping the daemon even on panic — a survivor -/// holds `spt.exe` open and fails the NEXT build with an access error that reads -/// as a compile problem and is really this cell's litter. -fn with_rig(label: &str, f: impl FnOnce(&Rig)) { - let r = rig(); - let out = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(&r))); - r.reap(label); - if let Err(e) = out { - std::panic::resume_unwind(e); - } -} - -// [int->REQ-IO-COMMUNE-EVENT] -// [int->REQ-IO-EVENT-ADAPTER-LOG] -// THE DEFECT THIS RESPIN EXISTS FOR: a consumed commune reaches the adapter log -// and comes back through the poll. Before the fix this was structurally -// impossible on every box, and four green legs plus a gate PASS all agreed it -// was fine. -#[test] -fn a_consumed_commune_reaches_the_log_and_polls_back() { - with_rig("io_events_undriven/commune", |r| { - r.seed(AUTHOR, SID_A); - let host = BrainLifecycle::with_config(&r.manifest, AUTHOR, DaemonConfig::default()) - .expect("live host"); - std::fs::write(r.drop_path(), COMMUNE_BODY).unwrap(); - let report = host.pulse_tick(Some(SID_A)).expect("tick"); - assert_eq!(report.ingested.len(), 1, "PRECONDITION: the drop was ingested"); - assert!( - !r.drop_path().exists(), - "PRECONDITION: a consumed drop is unlinked, which is why the event must carry its bytes" - ); - - let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); - assert_eq!( - got.iter().map(|x| x.0.as_str()).collect::>(), - vec!["COMMUNE"], - "a consumed commune must reach the adapter log: {got:?}" - ); - assert_eq!( - got[0].1, COMMUNE_BODY, - "and the payload is the FILE, verbatim — wake markers untouched, slices \ - un-reserialized, whitespace intact" - ); - }); -} - -// [int->REQ-IO-COMMUNE-FAIL-EVENT] -// A failed ingest emits its own kind and KEEPS its file. Induced by an -// unreadable drop; see the module note on why the index.lock class is not the -// mechanism despite the field attribution. -#[test] -fn a_failed_commune_ingest_polls_back_as_its_own_kind() { - with_rig("io_events_undriven/commune_fail", |r| { - r.seed(AUTHOR, SID_A); - let host = BrainLifecycle::with_config(&r.manifest, AUTHOR, DaemonConfig::default()) - .expect("live host"); - std::fs::create_dir(r.drop_path()).unwrap(); - let _ = host.pulse_tick(Some(SID_A)); - assert!( - r.drop_path().exists(), - "PRECONDITION: a failed ingest LEAVES its drop as the on-disk diagnostic" - ); - - let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); - assert_eq!( - got.iter().map(|x| x.0.as_str()).collect::>(), - vec!["COMMUNE_FAIL"], - "a failed ingest is distinguishable from a landed one by kind alone: {got:?}" - ); - }); -} - -// [int->REQ-IO-MSG-EDGES] -// The SEND edge publishes MSG_OUT into the SENDER's log. -#[test] -fn the_send_edge_polls_back_as_msg_out_on_the_sender() { - with_rig("io_events_undriven/msg_out", |r| { - r.seed(AUTHOR, SID_A); - let out = send_bounded( - &r.spt_bin, - r.home.path(), - &["send", PEER, "--from", AUTHOR], - "a message that must reach the log", - ); - assert!( - out.status.success(), - "the send must commit — SENT and QUEUED are BOTH success, and QUEUED is the \ - expected arm here because bob is not listening: stdout={:?} stderr={:?}", - String::from_utf8_lossy(&out.stdout), - String::from_utf8_lossy(&out.stderr) - ); - - let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); - assert_eq!( - got.iter().map(|x| x.0.as_str()).collect::>(), - vec!["MSG_OUT"], - "the SEND edge publishes MSG_OUT into the sender's log: {got:?}" - ); - assert_eq!( - got[0].2.as_deref(), - Some(PEER), - "and the peer rides along from the edge that knew it: {got:?}" - ); - }); -} - -// [int->REQ-IO-MSG-EDGES] -// The DELIVERY edge publishes MSG_IN into the RECEIVER's log — a DIFFERENT -// endpoint's log than MSG_OUT, which is what makes these two arms rather than -// one assertion seen twice. -#[test] -fn the_delivery_edge_polls_back_as_msg_in_on_the_receiver() { - with_rig("io_events_undriven/msg_in", |r| { - r.seed(PEER, SID_B); - let out = send_bounded( - &r.spt_bin, - r.home.path(), - &["send", PEER, "--from", AUTHOR], - "a message that must reach the receiver's log", - ); - assert!(out.status.success(), "PRECONDITION: the send committed"); - - // The receiver's drain IS the delivery edge that publishes MSG_IN. - let mut drain = Command::new(&r.spt_bin); - drain - .no_window() - .args(["api", "poll", PEER, "--session-id", SID_B]) - .env("SPT_HOME", r.home.path()); - let _ = common::output_bounded(drain, Duration::from_secs(60)); - - let got = rows(&poll(&r.spt_bin, r.home.path(), PEER, SID_B, &[])); - assert_eq!( - got.iter().map(|x| x.0.as_str()).collect::>(), - vec!["MSG_IN"], - "the DELIVERY edge publishes MSG_IN into the RECEIVER's log: {got:?}" - ); - assert_eq!( - got[0].2.as_deref(), - Some(AUTHOR), - "and its peer is the sender: {got:?}" - ); - }); -} +//! CONDUIT W3 RESPIN (releases#234) — the IO kinds whose publish → log → poll +//! path was never driven end to end. +//! +//! ## Why this leg exists, stated as the defect it closes +//! +//! W3 shipped, gated and reached a golden head asserting in four places — the +//! REQ title, `api.md`, `CONTEXT.md`, the integration checklist — that **all six +//! emitted kinds are visible** through `spt api io-events`. Two were +//! structurally impossible: `publish_commune_io` hand-built an `IoBus` with one +//! sink instead of calling `default_bus`, so `IoLogSink` was never registered on +//! the commune path and no poll could ever answer with `COMMUNE` or +//! `COMMUNE_FAIL`. +//! +//! Nothing in the battery could see it. The `io_events_poll_e2e` cell drives +//! `USER_INPUT` and `AGENT_OUTPUT` only, so it was VACUOUS on exactly the two +//! broken kinds; the unit tests synthesize `IoLogRow` values directly and never +//! traverse the publish path where a missing registration lives; and +//! `traceable-reqs` was green on TAGS, not on truth. **A claim that enumerates +//! is a test plan, and this file is the missing half of it.** +//! +//! ## Width, ruled by doyle +//! +//! Cover every kind whose publish → log → poll path is UNDRIVEN; skip kinds an +//! existing non-vacuous cell already drives. +//! +//! * `COMMUNE` — the broken kind. Mandatory. +//! * `COMMUNE_FAIL` — cheap here, so it rides (see the induction note below). +//! * `MSG_IN` / `MSG_OUT` — structurally fine and never driven, which is the +//! precise state that cost this milestone a respin. Mandatory. +//! * `USER_INPUT` / `AGENT_OUTPUT` — EXCLUDED as duplicate denominators. +//! Discharged at source, not from memory: `io_events_poll_e2e` drives +//! `api state busy|idle --payload-file` (publish) and then asserts both +//! kinds AND their verbatim payloads back out of `api io-events` (poll), so +//! that path is publish → log → poll and adding it here would only add +//! first-execution cells to the respin golden without adding coverage. +//! +//! ## The COMMUNE_FAIL induction, and a correction carried forward +//! +//! The failure is induced by making the drop UNREADABLE (replacing the file with +//! a directory): `ingest_drops` reads the body with `?` before it decides +//! anything, so the read fails the sweep, no tier is touched and no unlink is +//! reached. **The `index.lock` class does NOT fail this seam** — a previous +//! measurer planted non-empty locks in both places git takes one, asserted the +//! plant count, and the ingest committed anyway +//! (`spt-daemon/tests/commune_io_events_int.rs`). That matters beyond this file: +//! the field incident `COMMUNE_FAIL` was built for is ATTRIBUTED to index.lock, +//! and that attribution is unreproduced. +//! +//! cargo nextest run -p spt --test io_events_undriven_kinds_e2e --test-threads 1 +//! +//! Each cell owns its `SPT_HOME`, its daemon and its endpoints, so they share no +//! state and may run in any order. +// +// [int->REQ-IO-EVENT-ADAPTER-LOG] +// [int->REQ-IO-EVENT-POLL-VERB] +// [int->REQ-IO-COMMUNE-EVENT] +// [int->REQ-IO-COMMUNE-FAIL-EVENT] +// [int->REQ-IO-MSG-EDGES] + +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::time::Duration; + +mod common; +use common::reap; +use common::CommandNoWindowExt; + +use spt_daemon::{BrainLifecycle, DaemonConfig}; +use spt_runtime::Manifest; +use spt_store::perch::{self, ParentHint}; + +#[cfg(windows)] +const NOOP_CMD: &str = "cmd /C rem"; +#[cfg(unix)] +const NOOP_CMD: &str = "true"; + +fn start_real_daemon(spt_bin: &Path, home: &Path) { + let mut cmd = Command::new(spt_bin); + cmd.no_window() + .args(["daemon", "start"]) + .env("SPT_HOME", home) + .env_remove("SPT_AGENT_ID") + .env_remove("SPT_ENDPOINT_ID") + .env_remove("OWL_SESSION_ID"); + let out = common::output_bounded(cmd, Duration::from_secs(90)); + assert!( + out.status.success(), + "explicit daemon start failed: {}", + String::from_utf8_lossy(&out.stderr) + ); +} + +/// A live-capable manifest whose commune drop dir is `dir` — the shape the +/// lifecycle's own tests use. +fn live_manifest(dir: &Path) -> Manifest { + let d = dir.to_string_lossy().replace('\\', "/"); + let toml = format!( + "[adapter]\nname=\"mock\"\nversion=\"1\"\nmin_spt_core_version=\"1\"\n\n\ + [session]\ncommune_dir=\"{d}\"\n\n\ + [session.psyche_init]\ncommand='{NOOP_CMD}'\ncwd=\"{{psyche_dir}}\"\nkeys=[]\n" + ); + Manifest::from_toml_str(&toml).unwrap() +} + +/// Bind an endpoint perch this process can authenticate as, owned by `owner_pid`. +/// +/// The `cwd` is load-bearing rather than boilerplate: with no recorded cwd the +/// project id is empty, a drop carrying a `` slice takes the +/// deferred branch and is PRESERVED instead of consumed — which emits nothing. +/// Wiring a real cwd is what puts the commune arm on the CONSUMPTION path. +/// +/// `owner_pid` is load-bearing TOO, and it is the whole reason this signature +/// takes one (releases#236 respin RCA). Binding BOTH endpoints under this +/// process's pid put two TOP-LEVEL perches on one live pid, which is a state +/// `roster::detect_self_by_ancestry` cannot resolve: both are candidates at the +/// same nearest ancestor and the winner was whatever `read_dir` happened to +/// yield first. The send edge then published `MSG_OUT` with THAT id as owner — +/// bob on ext4, alice on NTFS — so the same rig red on Linux and green on +/// Windows, with the green telling us nothing about the code. Give each perch +/// its OWN live pid and the tie cannot arise: the assertions below are then +/// forced by the edges resolving correctly, not permitted by enumeration order. +fn bind_endpoint(id: &str, sid: &str, cwd: Option<&Path>, owner_pid: u32) -> PathBuf { + let p = perch::resolve_perch_path(id, ParentHint::Infer); + std::fs::create_dir_all(&p).unwrap(); + let mut rec = spt_store::info::InfoJson::new(id, "0", owner_pid, sid, "live_agent"); + if let Some(c) = cwd { + rec.cwd = Some(c.to_string_lossy().to_string()); + } + spt_store::info::write_info(&p, &rec).unwrap(); + p +} + +/// One `spt api io-events` poll, parsed. `--session-id` is both the auth proof +/// and the cursor key. +fn poll(spt_bin: &Path, home: &Path, id: &str, sid: &str, extra: &[&str]) -> serde_json::Value { + let mut cmd = Command::new(spt_bin); + cmd.no_window() + .args(["--json", "api", "io-events", id, "--session-id", sid]) + .args(extra) + .env("SPT_HOME", home); + let out = common::output_bounded(cmd, Duration::from_secs(60)); + assert!( + out.status.success(), + "poll of {id} failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + let text = String::from_utf8_lossy(&out.stdout).to_string(); + serde_json::from_str(&text) + .unwrap_or_else(|e| panic!("poll output is not JSON ({e}): {text:?}")) +} + +/// `spt send` reads its body from STDIN — the idiom `active_only_never_relay_e2e` +/// established, reused rather than re-invented. +fn send_bounded(spt_bin: &Path, home: &Path, args: &[&str], body: &str) -> Output { + let mut child = Command::new(spt_bin) + .no_window() + .args(args) + .env("SPT_HOME", home) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn spt send"); + child + .stdin + .take() + .unwrap() + .write_all(body.as_bytes()) + .expect("write stdin 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 timed out") + .expect("spt send output") +} + +fn rows(v: &serde_json::Value) -> Vec<(String, String, Option)> { + v["events"] + .as_array() + .expect("events array") + .iter() + .map(|e| { + ( + e["kind"].as_str().unwrap_or_default().to_string(), + e["payload"].as_str().unwrap_or_default().to_string(), + e["peer"].as_str().map(str::to_string), + ) + }) + .collect() +} +const AUTHOR: &str = "alice"; +const PEER: &str = "bob"; +const SID_A: &str = "io-undriven-sid-a"; +const SID_B: &str = "io-undriven-sid-b"; + +/// The commune body is deliberately awkward: a `!!wake!!` pair the adapter owns +/// and core must not touch, a `` slice routing to a different +/// tier than the live text, and trailing whitespace. +const COMMUNE_BODY: &str = "live half !!wake!! resume T2c !!wake!!\n\nproject half\n\n \n"; + +/// One isolated rig: own `SPT_HOME`, own daemon, own endpoints. +/// +/// The `TempDir`s are held so they outlive the cell — dropping them early +/// deletes the home out from under a running daemon. +struct Rig { + home: tempfile::TempDir, + _project: tempfile::TempDir, + drops: tempfile::TempDir, + spt_bin: PathBuf, + manifest: Manifest, +} + +fn rig() -> Rig { + let home = tempfile::tempdir().unwrap(); + std::env::set_var("SPT_HOME", home.path()); + std::env::remove_var("SPT_AGENT_ID"); + std::env::remove_var("SPT_ENDPOINT_ID"); + std::env::remove_var("OWL_SESSION_ID"); + spt_store::nodeid::load_or_create().expect("node identity"); + let spt_bin = PathBuf::from(env!("CARGO_BIN_EXE_spt")); + start_real_daemon(&spt_bin, home.path()); + + let project = tempfile::tempdir().unwrap(); + // DISTINCT LIVE OWNERS, by construction (releases#236 respin RCA — see + // `bind_endpoint`). AUTHOR is owned by this process, the ancestor of every + // `spt` child the cells spawn, so self-detection resolves alice through the + // ancestry leg it is meant to exercise. PEER is owned by the daemon: live + // for the whole cell, distinct from AUTHOR's, and NOT in any child's + // ancestry — the shape of an endpoint some other process hosts. One pid per + // perch means `detect_self_by_ancestry` has exactly one candidate at the + // nearest ancestor and never reaches an order-dependent tie. + let daemon_pid = reap::breadcrumb_daemon_pid(home.path()) + .expect("PRECONDITION: the daemon breadcrumb names a live owner for PEER"); + assert_ne!( + daemon_pid, + std::process::id(), + "PRECONDITION: the two perches must not share one live pid — that ambiguity IS the defect this rig shape exists to keep out" + ); + bind_endpoint(AUTHOR, SID_A, Some(project.path()), std::process::id()); + bind_endpoint(PEER, SID_B, Some(project.path()), daemon_pid); + let drops = tempfile::tempdir().unwrap(); + let manifest = live_manifest(drops.path()); + Rig { + home, + _project: project, + drops, + spt_bin, + manifest, + } +} + +impl Rig { + fn drop_path(&self) -> PathBuf { + self.drops.path().join(format!("{AUTHOR}-commune.md")) + } + + /// Seed a cursor first, so every row a cell asserts is one that cell caused. + fn seed(&self, id: &str, sid: &str) { + assert_eq!( + poll(&self.spt_bin, self.home.path(), id, sid, &[])["seeded"], + true, + "PRECONDITION: {id}'s cursor seeds on its first poll" + ); + } + + fn reap(&self, label: &str) { + let observed = reap::breadcrumb_daemon_pid(self.home.path()) + .and_then(|pid| reap::observe(label, pid)); + let mut stop = Command::new(&self.spt_bin); + stop.no_window() + .args(["daemon", "stop"]) + .env("SPT_HOME", self.home.path()); + let stopped = common::output_bounded(stop, Duration::from_secs(30)); + eprintln!( + "REAP[{label}]: daemon stop status={} stderr={:?}", + stopped.status, + String::from_utf8_lossy(&stopped.stderr).trim() + ); + reap::reap_breadcrumb_daemon(label, self.home.path(), &self.spt_bin, observed); + } +} + +/// Run a cell against a fresh rig, reaping the daemon even on panic — a survivor +/// holds `spt.exe` open and fails the NEXT build with an access error that reads +/// as a compile problem and is really this cell's litter. +fn with_rig(label: &str, f: impl FnOnce(&Rig)) { + let r = rig(); + let out = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(&r))); + r.reap(label); + if let Err(e) = out { + std::panic::resume_unwind(e); + } +} + +// [int->REQ-IO-COMMUNE-EVENT] +// [int->REQ-IO-EVENT-ADAPTER-LOG] +// THE DEFECT THIS RESPIN EXISTS FOR: a consumed commune reaches the adapter log +// and comes back through the poll. Before the fix this was structurally +// impossible on every box, and four green legs plus a gate PASS all agreed it +// was fine. +#[test] +fn a_consumed_commune_reaches_the_log_and_polls_back() { + with_rig("io_events_undriven/commune", |r| { + r.seed(AUTHOR, SID_A); + let host = BrainLifecycle::with_config(&r.manifest, AUTHOR, DaemonConfig::default()) + .expect("live host"); + std::fs::write(r.drop_path(), COMMUNE_BODY).unwrap(); + let report = host.pulse_tick(Some(SID_A)).expect("tick"); + assert_eq!(report.ingested.len(), 1, "PRECONDITION: the drop was ingested"); + assert!( + !r.drop_path().exists(), + "PRECONDITION: a consumed drop is unlinked, which is why the event must carry its bytes" + ); + + let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); + assert_eq!( + got.iter().map(|x| x.0.as_str()).collect::>(), + vec!["COMMUNE"], + "a consumed commune must reach the adapter log: {got:?}" + ); + assert_eq!( + got[0].1, COMMUNE_BODY, + "and the payload is the FILE, verbatim — wake markers untouched, slices \ + un-reserialized, whitespace intact" + ); + }); +} + +// [int->REQ-IO-COMMUNE-FAIL-EVENT] +// A failed ingest emits its own kind and KEEPS its file. Induced by an +// unreadable drop; see the module note on why the index.lock class is not the +// mechanism despite the field attribution. +#[test] +fn a_failed_commune_ingest_polls_back_as_its_own_kind() { + with_rig("io_events_undriven/commune_fail", |r| { + r.seed(AUTHOR, SID_A); + let host = BrainLifecycle::with_config(&r.manifest, AUTHOR, DaemonConfig::default()) + .expect("live host"); + std::fs::create_dir(r.drop_path()).unwrap(); + let _ = host.pulse_tick(Some(SID_A)); + assert!( + r.drop_path().exists(), + "PRECONDITION: a failed ingest LEAVES its drop as the on-disk diagnostic" + ); + + let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); + assert_eq!( + got.iter().map(|x| x.0.as_str()).collect::>(), + vec!["COMMUNE_FAIL"], + "a failed ingest is distinguishable from a landed one by kind alone: {got:?}" + ); + }); +} + +// [int->REQ-IO-MSG-EDGES] +// The SEND edge publishes MSG_OUT into the SENDER's log. +#[test] +fn the_send_edge_polls_back_as_msg_out_on_the_sender() { + with_rig("io_events_undriven/msg_out", |r| { + r.seed(AUTHOR, SID_A); + let out = send_bounded( + &r.spt_bin, + r.home.path(), + &["send", PEER, "--from", AUTHOR], + "a message that must reach the log", + ); + assert!( + out.status.success(), + "the send must commit — SENT and QUEUED are BOTH success, and QUEUED is the \ + expected arm here because bob is not listening: stdout={:?} stderr={:?}", + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr) + ); + + let got = rows(&poll(&r.spt_bin, r.home.path(), AUTHOR, SID_A, &[])); + assert_eq!( + got.iter().map(|x| x.0.as_str()).collect::>(), + vec!["MSG_OUT"], + "the SEND edge publishes MSG_OUT into the sender's log: {got:?}" + ); + assert_eq!( + got[0].2.as_deref(), + Some(PEER), + "and the peer rides along from the edge that knew it: {got:?}" + ); + }); +} + +// [int->REQ-IO-MSG-EDGES] +// The DELIVERY edge publishes MSG_IN into the RECEIVER's log — a DIFFERENT +// endpoint's log than MSG_OUT, which is what makes these two arms rather than +// one assertion seen twice. +#[test] +fn the_delivery_edge_polls_back_as_msg_in_on_the_receiver() { + with_rig("io_events_undriven/msg_in", |r| { + r.seed(PEER, SID_B); + let out = send_bounded( + &r.spt_bin, + r.home.path(), + &["send", PEER, "--from", AUTHOR], + "a message that must reach the receiver's log", + ); + assert!(out.status.success(), "PRECONDITION: the send committed"); + + // The receiver's drain IS the delivery edge that publishes MSG_IN. + let mut drain = Command::new(&r.spt_bin); + drain + .no_window() + .args(["api", "poll", PEER, "--session-id", SID_B]) + .env("SPT_HOME", r.home.path()); + let _ = common::output_bounded(drain, Duration::from_secs(60)); + + let got = rows(&poll(&r.spt_bin, r.home.path(), PEER, SID_B, &[])); + assert_eq!( + got.iter().map(|x| x.0.as_str()).collect::>(), + vec!["MSG_IN"], + "the DELIVERY edge publishes MSG_IN into the RECEIVER's log: {got:?}" + ); + assert_eq!( + got[0].2.as_deref(), + Some(AUTHOR), + "and its peer is the sender: {got:?}" + ); + }); +}