supermachine 0.7.108

//! The two-test prefix-reuse vertical slice (supermachine-vm#70, second half).
//!
//! Proves the whole tree loop on real parts: a plain-Playwright "prefix" test
//! runs against Chromium INSIDE the guest (reached over the vsock TSI forward),
//! the world is captured as a ~20 ms dirty diff node, and two INDEPENDENT
//! forks of that node each run a different test with the shared prefix
//! no-oped (`SUPERMACHINE_TREE_SKIP_STEPS`) and the prefix-hash guard armed
//! (`SUPERMACHINE_TREE_EXPECT_PREFIX`). The two suffix tests assert MUTUALLY
//! EXCLUSIVE world states — each sees the prefix's effects but not the other
//! fork's suffix — which is the tree property itself, not just skip
//! accounting.
//!
//! Env (no machine-local defaults):
//!   SUPERMACHINE_SLICE_SNAPSHOT   full snapshot dir with node+chromium baked
//!                                 (restore.snap inside is the dirty-capture base)
//!   SUPERMACHINE_SLICE_WORKDIR    host scratch dir (node dir, playwright out)
//!   SUPERMACHINE_SLICE_FLEET      supermachine-fleet checkout (harness/preload.cjs)
//!   SUPERMACHINE_SLICE_CHROME     chrome binary path inside the guest
//!   SUPERMACHINE_SLICE_MEM_MIB    guest RAM (default 8192)

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn main() {
    slice::run();
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
mod slice {
    use std::io::{Read, Write};
    use std::path::Path;
    use std::time::{Duration, Instant};
    use supermachine::{Image, Vm, VmConfig};

    const APP_PORT: u16 = 3000;
    const CDP_PORT: u16 = 9222;

    pub fn run() {
        let snap = env("SUPERMACHINE_SLICE_SNAPSHOT");
        let workdir = env("SUPERMACHINE_SLICE_WORKDIR");
        let fleet = env("SUPERMACHINE_SLICE_FLEET");
        let chrome = env("SUPERMACHINE_SLICE_CHROME");
        let mem: u32 = std::env::var("SUPERMACHINE_SLICE_MEM_MIB")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(8192);

        std::fs::create_dir_all(&workdir).expect("workdir");
        write_host_suite(&workdir);
        npm_install(&workdir);

        let cfg = VmConfig::new().with_memory_mib(mem);
        let node_dir = format!("{workdir}/tree-node-prefix");
        let _ = std::fs::remove_dir_all(&node_dir);

        // --- Parent world: boot, start app + in-guest chromium ---
        eprintln!("SLICE step=boot_parent");
        let vm = Image::from_snapshot(&snap)
            .expect("from_snapshot")
            .start(&cfg)
            .expect("start");
        wait_exec(&vm);
        start_world(&vm, &chrome);
        // Host port == guest port so devtools' Host-header-echoed ws URLs
        // stay dialable from the host.
        let fwd_p = vm
            .expose_tcp(CDP_PORT, CDP_PORT)
            .expect("expose parent cdp");
        // Differential probes: does ANY new guest listener get a TSI route?
        {
            let fwd_app = vm.expose_tcp(13000, APP_PORT).expect("expose app");
            let app_ok = try_http("127.0.0.1:13000", "/", 8);
            eprintln!("SLICE diag app_forward_ok={app_ok}");
            drop(fwd_app);
            match vm.connect_port(CDP_PORT) {
                Ok(mut s) => {
                    let _ = s.set_read_timeout(Some(Duration::from_secs(3)));
                    let _ = s.write_all(b"GET /json/version HTTP/1.1\r\nHost: 127.0.0.1:9222\r\nConnection: close\r\n\r\n");
                    let mut buf = String::new();
                    let _ = s.read_to_string(&mut buf);
                    eprintln!("SLICE diag connect_port_9222 head={:?}", buf.lines().next());
                }
                Err(e) => eprintln!("SLICE diag connect_port_9222 err={e}"),
            }
        }
        if !try_http(&format!("127.0.0.1:{CDP_PORT}"), "/json/version", 30) {
            // Dump guest state before dying — which process died, what the
            // kernel said, who is still listening.
            let diag = vm
                .exec_builder()
                .argv(["/bin/sh", "-c", "echo ---dmesg---; dmesg | tail -8; echo ---chrome---; tail -5 /root/chrome.log 2>/dev/null; echo ---relay---; tail -5 /root/relay.log 2>/dev/null; echo ---procs---; pgrep -c chrome 2>/dev/null; pgrep -c -f cdp-relay 2>/dev/null; echo ---inguest-cdp---; node -e \"require('http').get('http://127.0.0.1:9222/json/version',r=>{console.log('relay ok',r.statusCode);process.exit(0)}).on('error',e=>{console.log('relay ERR',e.message);process.exit(0)})\""])
                .output();
            if let Ok(o) = diag {
                eprintln!("GUEST DIAG:\n{}", String::from_utf8_lossy(&o.stdout));
            }
            panic!("host->guest cdp readiness failed; diagnostics above");
        }

        // --- Prefix test P against the in-guest browser ---
        eprintln!("SLICE step=run_prefix_test");
        if !run_spec(&workdir, &fleet, "p.spec.js", "out-p", CDP_PORT, None, None) {
            let diag = vm.exec_builder().argv(["/bin/sh","-c","echo ---applog---; tail -5 /root/app.log; echo ---dmesg---; dmesg | tail -6; echo ---chrome-tabs---; node -e \"require('http').get('http://127.0.0.1:9224/json/list',r=>{let b='';r.on('data',d=>b+=d);r.on('end',()=>{console.log(b.slice(0,600));process.exit(0)})}).on('error',e=>{console.log('ERR',e.message);process.exit(0)})\""]).output();
            if let Ok(o) = diag {
                eprintln!("GUEST DIAG P:\n{}", String::from_utf8_lossy(&o.stdout));
            }
            panic!("p.spec.js failed; diagnostics above");
        }
        let events_p = read_events(&format!("{workdir}/out-p"));
        assert!(
            events_p.len() >= 2,
            "prefix test emitted {} events",
            events_p.len()
        );
        let prefix_hash = events_p[1]["prefixHash"]
            .as_str()
            .expect("prefixHash")
            .to_string();
        eprintln!("SLICE prefix_hash={prefix_hash}");

        // --- Capture the node: instant dirty diff vs the restored base ---
        let base_snap = Path::new(&snap).join("restore.snap");
        eprintln!("SLICE step=capture_node");
        let t = Instant::now();
        let node = vm
            .snapshot_diff_live_dirty(&node_dir, &base_snap)
            .expect("dirty capture");
        let capture_ms = t.elapsed().as_secs_f64() * 1e3;
        eprintln!(
            "SLICE capture_ms={capture_ms:.1} node_bytes={}",
            dir_bytes(&node_dir)
        );

        // Free host port CDP_PORT for the forks (they run sequentially and
        // must bind the same number — see the relay comment).
        drop(fwd_p);

        // --- Two independent forks, two different suffix tests ---
        // B clicks #suffix-b and must see suffixA == 0; C clicks #suffix-a
        // and must see suffixB == 0. Both must see shared == 1 (the prefix's
        // click) WITHOUT re-running it.
        let b = fork_and_run(
            &node,
            &cfg,
            &workdir,
            &fleet,
            "b.spec.js",
            "out-b",
            CDP_PORT,
            &prefix_hash,
            &chrome,
        );
        let c = fork_and_run(
            &node,
            &cfg,
            &workdir,
            &fleet,
            "c.spec.js",
            "out-c",
            CDP_PORT,
            &prefix_hash,
            &chrome,
        );

        // Skip accounting: exactly the two prefix steps are marked skipped.
        for out in ["out-b", "out-c"] {
            let ev = read_events(&format!("{workdir}/{out}"));
            assert!(
                ev.len() >= 3,
                "{out}: expected >=3 events, got {}",
                ev.len()
            );
            assert!(
                ev[0]["skipped"].as_bool() == Some(true),
                "{out}: event 1 not skipped"
            );
            assert!(
                ev[1]["skipped"].as_bool() == Some(true),
                "{out}: event 2 not skipped"
            );
            assert!(
                ev[2]["skipped"].as_bool() != Some(true),
                "{out}: suffix event marked skipped"
            );
        }
        eprintln!(
            "SLICE ledger: capture_ms={capture_ms:.1} fork_b_restore_ms={:.1} fork_b_test_s={:.1} fork_c_restore_ms={:.1} fork_c_test_s={:.1}",
            b.0, b.1, c.0, c.1
        );
        vm.stop().ok();
        eprintln!("SLICE result=PASS (two forks of one prefix node, mutually exclusive suffix states, prefix never re-executed)");
    }

    /// Restore the node, re-expose CDP, run one suffix spec with skip-2 +
    /// prefix guard. Returns (restore_ms, test_seconds).
    fn fork_and_run(
        node: &Image,
        cfg: &VmConfig,
        workdir: &str,
        fleet: &str,
        spec: &str,
        out: &str,
        host_port: u16,
        prefix_hash: &str,
        _chrome: &str,
    ) -> (f64, f64) {
        eprintln!("SLICE step=fork spec={spec}");
        let t = Instant::now();
        let child = node.start(cfg).expect("fork restore");
        let restore_ms = t.elapsed().as_secs_f64() * 1e3;
        wait_exec(&child);
        let _fwd = child
            .expose_tcp(host_port, CDP_PORT)
            .expect("expose fork cdp");
        wait_http(&format!("127.0.0.1:{host_port}"), "/json/version", 30);
        let t = Instant::now();
        assert!(
            run_spec(
                workdir,
                fleet,
                spec,
                out,
                host_port,
                Some(2),
                Some(prefix_hash)
            ),
            "{spec} failed on fork"
        );
        let test_s = t.elapsed().as_secs_f64();
        child.stop().ok();
        (restore_ms, test_s)
    }

    /// Start the demo app and headless chromium (CDP on 9222) inside the guest.
    fn start_world(vm: &Vm, chrome: &str) {
        let app = r#"const http=require('node:http');
http.createServer((q,res)=>{console.log('REQ',q.method,q.url);res.setHeader('content-type','text/html');res.end(`<!doctype html><button id="shared">shared</button><button id="suffix-a">sa</button><button id="suffix-b">sb</button><script>
window.__hits={shared:0,suffixA:0,suffixB:0};
document.querySelector('#shared').addEventListener('click',()=>{window.__hits.shared+=1});
document.querySelector('#suffix-a').addEventListener('click',()=>{window.__hits.suffixA+=1});
document.querySelector('#suffix-b').addEventListener('click',()=>{window.__hits.suffixB+=1});
</script>`)}).listen(3000,'127.0.0.1',()=>console.log('app ready'));"#;
        let b64 = base64(app.as_bytes());
        exec_ok(vm, &format!("echo {b64} | base64 -d > /root/app.cjs"));
        // Chromium must NOT accept host-forwarded connections directly: it
        // calls getsockopt(SOL_IPV6, IPV6_V6ONLY) on accepted sockets, and
        // the TSI hijack's ops table has a NULL there — guest kernel oops,
        // Chrome IO thread dies (observed twice, 2026-07-12; tracking issue
        // filed). Node's accept path never queries SOL_IPV6 (fleet forwards
        // to node apps daily), so a tiny in-guest node relay fronts the
        // devtools port: host → TSI forward → relay (node, safe) → guest
        // loopback (real socket, safe) → chromium on CDP_PORT+2. The relay
        // listens on 0.0.0.0:CDP_PORT (the TSI hijack only exposes non-loopback
        // listeners to the host) and the host forward uses the SAME port
        // number, so the ws URLs devtools echoes from the Host header stay
        // dialable end-to-end.
        let relay = format!(
            "const net=require('node:net');net.createServer(c=>{{const u=net.connect({backend},'127.0.0.1');c.pipe(u);u.pipe(c);c.on('error',()=>u.destroy());u.on('error',()=>c.destroy())}}).listen({front},'0.0.0.0',()=>console.log('relay ready'));",
            backend = CDP_PORT + 2,
            front = CDP_PORT
        );
        let relay_b64 = base64(relay.as_bytes());
        exec_ok(
            vm,
            &format!("echo {relay_b64} | base64 -d > /root/cdp-relay.cjs"),
        );
        exec_ok(
            vm,
            "nohup node /root/app.cjs > /root/app.log 2>&1 & sleep 0.2; true",
        );
        exec_ok(
            vm,
            &format!(
                "nohup {chrome} --headless --no-sandbox --disable-gpu --disable-dev-shm-usage \
                 --remote-debugging-port={} --remote-debugging-address=127.0.0.1 \
                 --user-data-dir=/root/cdp-profile about:blank \
                 > /root/chrome.log 2>&1 & sleep 0.2; true",
                CDP_PORT + 2
            ),
        );
        exec_ok(
            vm,
            "nohup node /root/cdp-relay.cjs > /root/relay.log 2>&1 & sleep 0.2; true",
        );
        // Readiness from INSIDE the guest (loopback): app, chromium direct,
        // then through the relay.
        exec_retry(vm, &format!("node -e \"require('http').get('http://127.0.0.1:{APP_PORT}',r=>process.exit(0)).on('error',()=>process.exit(1))\""), 30);
        exec_retry(vm, &format!("node -e \"require('http').get('http://127.0.0.1:{}/json/version',r=>process.exit(0)).on('error',()=>process.exit(1))\"", CDP_PORT + 2), 30);
        exec_retry(vm, &format!("node -e \"require('http').get('http://127.0.0.1:{CDP_PORT}/json/version',r=>process.exit(0)).on('error',()=>process.exit(1))\""), 30);
    }

    fn write_host_suite(workdir: &str) {
        let w = Path::new(workdir);
        std::fs::write(
            w.join("package.json"),
            r#"{"type":"commonjs","devDependencies":{"@playwright/test":"^1.48.0"}}"#,
        )
        .unwrap();
        std::fs::write(w.join("playwright.config.js"), "module.exports = { testDir: '.', timeout: 45000, workers: 1, use: { browserName: 'chromium' } }\n").unwrap();
        // The world page is a data: URL: the demo state (window.__hits) lives
        // in the in-guest chromium's JS heap, which is exactly what the VM
        // snapshot captures and the forks must NOT share. (Guest-TCP page
        // loads are kept out of this slice: chromium's response streams
        // stall through the TSI loopback path — tracked separately with the
        // getsockopt oops.) The URL string is byte-identical across specs so
        // the prefix hash chain matches.
        let page_js = r#"window.__hits={shared:0,suffixA:0,suffixB:0};document.getElementById("shared").addEventListener("click",()=>{window.__hits.shared+=1});document.getElementById("suffix-a").addEventListener("click",()=>{window.__hits.suffixA+=1});document.getElementById("suffix-b").addEventListener("click",()=>{window.__hits.suffixB+=1});"#;
        let data_url = format!(
            "data:text/html,<button id=\"shared\">s</button><button id=\"suffix-a\">sa</button><button id=\"suffix-b\">sb</button><script>{}</script>",
            page_js.replace('#', "%23")
        );
        let prefix = format!("  await page.goto('{data_url}')\n  await page.click('#shared')\n");
        std::fs::write(w.join("p.spec.js"), format!(
            "const {{ test, expect }} = require('@playwright/test')\ntest('prefix builder', async ({{ page }}) => {{\n{prefix}  await expect(page.locator('#shared')).toBeVisible()\n}})\n"
        )).unwrap();
        std::fs::write(w.join("b.spec.js"), format!(
            "const {{ test, expect }} = require('@playwright/test')\ntest('fork b: suffix-b only', async ({{ page }}) => {{\n{prefix}  await page.click('#suffix-b')\n  const hits = await page.evaluate(() => window.__hits)\n  expect(hits).toEqual({{ shared: 1, suffixA: 0, suffixB: 1 }})\n}})\n"
        )).unwrap();
        std::fs::write(w.join("c.spec.js"), format!(
            "const {{ test, expect }} = require('@playwright/test')\ntest('fork c: suffix-a only', async ({{ page }}) => {{\n{prefix}  await page.click('#suffix-a')\n  const hits = await page.evaluate(() => window.__hits)\n  expect(hits).toEqual({{ shared: 1, suffixA: 1, suffixB: 0 }})\n}})\n"
        )).unwrap();
    }

    fn npm_install(workdir: &str) {
        if Path::new(workdir)
            .join("node_modules/@playwright/test")
            .exists()
        {
            return;
        }
        let st = std::process::Command::new("npm")
            .args(["install", "--no-audit", "--no-fund"])
            .current_dir(workdir)
            .status()
            .expect("npm install");
        assert!(st.success(), "npm install failed");
    }

    fn run_spec(
        workdir: &str,
        fleet: &str,
        spec: &str,
        out: &str,
        cdp_host_port: u16,
        skip: Option<u32>,
        expect_prefix: Option<&str>,
    ) -> bool {
        let mut cmd = std::process::Command::new("npx");
        cmd.args(["playwright", "test", spec])
            .current_dir(workdir)
            .env("SUPERMACHINE_OUT", format!("{workdir}/{out}"))
            .env(
                "SUPERMACHINE_TREE_CDP_ENDPOINT",
                format!("http://127.0.0.1:{cdp_host_port}"),
            )
            .env(
                "NODE_OPTIONS",
                format!("--require {fleet}/harness/preload.cjs"),
            );
        if let Some(k) = skip {
            cmd.env("SUPERMACHINE_TREE_SKIP_STEPS", k.to_string());
        }
        if let Some(p) = expect_prefix {
            cmd.env("SUPERMACHINE_TREE_EXPECT_PREFIX", p);
        }
        let outp = cmd.output().expect("npx playwright test");
        if !outp.status.success() {
            eprintln!(
                "--- {spec} stdout ---\n{}",
                String::from_utf8_lossy(&outp.stdout)
            );
            eprintln!(
                "--- {spec} stderr ---\n{}",
                String::from_utf8_lossy(&outp.stderr)
            );
            return false;
        }
        true
    }

    fn read_events(out_dir: &str) -> Vec<serde_json::Value> {
        let p = format!("{out_dir}/step-boundaries/events.ndjson");
        std::fs::read_to_string(&p)
            .unwrap_or_else(|e| panic!("read {p}: {e}"))
            .lines()
            .filter(|l| !l.trim().is_empty())
            .map(|l| serde_json::from_str(l).expect("event json"))
            .collect()
    }

    fn wait_exec(vm: &Vm) {
        for _ in 0..120 {
            if let Ok(o) = vm
                .exec_builder()
                .argv(["/bin/sh", "-c", "echo up"])
                .output()
            {
                if o.success() {
                    return;
                }
            }
            std::thread::sleep(Duration::from_millis(1000));
        }
        panic!("guest exec never came up");
    }

    fn exec_ok(vm: &Vm, cmd: &str) {
        let o = vm
            .exec_builder()
            .argv(["/bin/sh", "-c", cmd])
            .output()
            .expect("exec");
        assert!(
            o.success(),
            "guest cmd failed: {cmd}\n{}",
            String::from_utf8_lossy(&o.stderr)
        );
    }

    fn exec_retry(vm: &Vm, cmd: &str, secs: u32) {
        for _ in 0..secs {
            if let Ok(o) = vm.exec_builder().argv(["/bin/sh", "-c", cmd]).output() {
                if o.success() {
                    return;
                }
            }
            std::thread::sleep(Duration::from_millis(1000));
        }
        panic!("guest readiness timed out: {cmd}");
    }

    /// HTTP GET readiness from the HOST through the TSI forward.
    fn wait_http(addr: &str, path: &str, secs: u32) {
        if !try_http(addr, path, secs) {
            panic!("host->guest http readiness timed out: {addr}{path}");
        }
    }

    fn try_http(addr: &str, path: &str, secs: u32) -> bool {
        for _ in 0..secs {
            if let Ok(mut s) = std::net::TcpStream::connect(addr) {
                let _ = s.set_read_timeout(Some(Duration::from_secs(2)));
                let req =
                    format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
                if s.write_all(req.as_bytes()).is_ok() {
                    let mut buf = String::new();
                    let r = s.read_to_string(&mut buf);
                    if buf.starts_with("HTTP/1.1 200") {
                        return true;
                    }
                    eprintln!("try_http {addr}: read={r:?} head={:?}", buf.lines().next());
                }
            }
            std::thread::sleep(Duration::from_millis(1000));
        }
        false
    }

    fn base64(data: &[u8]) -> String {
        const T: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
        let mut out = String::new();
        for chunk in data.chunks(3) {
            let b = [
                chunk[0],
                *chunk.get(1).unwrap_or(&0),
                *chunk.get(2).unwrap_or(&0),
            ];
            let n = ((b[0] as u32) << 16) | ((b[1] as u32) << 8) | b[2] as u32;
            out.push(T[(n >> 18) as usize & 63] as char);
            out.push(T[(n >> 12) as usize & 63] as char);
            out.push(if chunk.len() > 1 {
                T[(n >> 6) as usize & 63] as char
            } else {
                '='
            });
            out.push(if chunk.len() > 2 {
                T[n as usize & 63] as char
            } else {
                '='
            });
        }
        out
    }

    fn dir_bytes(d: &str) -> String {
        fn walk(p: &Path) -> u64 {
            std::fs::read_dir(p)
                .map(|rd| {
                    rd.flatten()
                        .map(|e| {
                            let m = e.metadata().unwrap();
                            if m.is_dir() {
                                walk(&e.path())
                            } else {
                                m.len()
                            }
                        })
                        .sum()
                })
                .unwrap_or(0)
        }
        format!("{:.1} MiB", walk(Path::new(d)) as f64 / 1048576.0)
    }

    fn env(k: &str) -> String {
        std::env::var(k).unwrap_or_else(|_| panic!("set {k}"))
    }
}

#[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
fn main() {
    eprintln!("linux/x86_64 only");
}