open-standards/jcp/jcp-rs/crates/jcp-gateway/tests/upstream_tree.rs

//! The gateway metering real calls on the upstream server's own process
//! tree. Each test spawns an MCP-ish server of its own (a process group of
//! its own, killed at the end) that answers one JSON-RPC `tools/call` per
//! line, burning the loop count its arguments name before it replies, and
//! relays `tools/call`s to it through [`Gateway::relay`].
//!
//! The oracle shares no code with the meter or the gateway: the server's
//! `proc_pid_rusage(pid, RUSAGE_INFO_V6)` read into a raw buffer and taken
//! at the SDK's byte offsets (energy at 336, reaped-descendant CPU time at
//! 96 and 104, billed CPU time at 216, billed and serviced energy at 264
//! and 272), read before and after each relay once the server is quiet
//! on the oracle's own terms: no thread on a core or queued
//! (`proc_pidinfo(PROC_PIDTASKINFO)` into a raw buffer, `pti_numrunning` at
//! byte 88) and none in a wait no signal ends (each thread's `pth_run_state`,
//! read raw as well) at two reads 20 ms apart whose energy agrees. A server blocked
//! waiting for its next request does not move its count, so between the
//! meter's reads and the oracle's nothing is spent: the call's figure must
//! be the oracle's advance across the relay, in nanojoules rounded UP to
//! the receipt's microjoules (an enforcement point never debits less than
//! was read). That is the tolerance — exact to the µJ: the figure is at
//! least the oracle's advance and less than 1 µJ over it.
//!
//! What a call is sealed as is read back from its receipt, verified, and
//! compared with the plaintext beside the response. The tests log what they
//! saw (`upstream tree:`), and run one at a time (a lock) so one test's
//! server is not loading the host under another's.

#![cfg(all(target_os = "macos", target_arch = "aarch64"))]

use ed25519_dalek::SigningKey;
use jcp_gateway::{CallMeter, Gateway, Relayed, Upstream, JCP_META_NS};
use serde_json::{json, Value};
use std::io::{BufRead, BufReader, Write};
use std::os::unix::process::CommandExt;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::{Mutex, MutexGuard};
use std::time::{Duration, Instant};

static ONE_AT_A_TIME: Mutex<()> = Mutex::new(());

fn alone() -> MutexGuard<'static, ()> {
    ONE_AT_A_TIME.lock().unwrap_or_else(|e| e.into_inner())
}

/// How long the gateway's tree meter may wait for the server to go quiet in
/// these tests: far past the default bound, because the host runs at a load
/// of 14 to 260 and a reply's tail can wait that long for a core. A server
/// still running after it would be a gap and the call `Partial`, failing a
/// test that expects it whole, which is the point: a whole reading is never
/// a stale one.
const TEST_SETTLE: Duration = Duration::from_secs(2);

/// The JSON-RPC result for the request line in `l`: its id, and its
/// `arguments.n` as a loop count burned before replying.
const REPLY: &str = r#"id = "null"; n = 0
  if (match(l, /"id":[0-9]+/)) id = substr(l, RSTART + 5, RLENGTH - 5)
  if (match(l, /"n":[0-9]+/)) n = substr(l, RSTART + 4, RLENGTH - 4) + 0
  for (i = 0; i < n; i++) x += i
  printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %d\"}]}}\n", id, n"#;

const STEWARD: &str = "did:jcp:gateway:burner";

fn steward_sk() -> SigningKey {
    SigningKey::from_bytes(&[0x2c; 32])
}

/// `pid`'s `rusage_info_v6`, raw: `proc_pid_rusage(pid, 6)` into a buffer
/// larger than the struct's 464 bytes, as 64-bit words.
fn oracle_rusage(pid: i32) -> Option<[u64; 64]> {
    let mut buf = [0u64; 64];
    // SAFETY: `buf` is writable and larger than `rusage_info_v6`.
    let rc = unsafe { libc::proc_pid_rusage(pid, 6, buf.as_mut_ptr() as *mut libc::rusage_info_t) };
    (rc == 0).then_some(buf)
}

fn energy_nj(ru: &[u64; 64]) -> u64 {
    ru[336 / 8]
}

fn reaped_cpu_ticks(ru: &[u64; 64]) -> u64 {
    ru[96 / 8] + ru[104 / 8]
}

/// `ri_billed_system_time`, mach ticks: the CPU time of work other
/// processes did on the server's behalf, billed to it.
fn billed_ticks(ru: &[u64; 64]) -> u64 {
    ru[216 / 8]
}

/// `ri_billed_energy`, nJ: that work's energy.
fn billed_nj(ru: &[u64; 64]) -> u64 {
    ru[264 / 8]
}

/// `ri_serviced_energy`, nJ: the energy of work the server did for others.
fn serviced_nj(ru: &[u64; 64]) -> u64 {
    ru[272 / 8]
}

/// `pti_numrunning` of `pid`: `proc_pidinfo(pid, PROC_PIDTASKINFO = 4)` into
/// a raw 96-byte buffer, the int32 at byte 88.
fn oracle_running(pid: i32) -> Option<i32> {
    let mut buf = [0u8; 96];
    // SAFETY: `buf` is writable for the 96 bytes passed.
    let n = unsafe { libc::proc_pidinfo(pid, 4, 0, buf.as_mut_ptr().cast(), 96) };
    (n == 96).then(|| i32::from_ne_bytes(buf[88..92].try_into().unwrap()))
}

/// Whether every thread of `pid` is at rest by its own run state: none on a
/// core or queued (`TH_STATE_RUNNING`, 1) and none in a wait that no signal
/// ends (`TH_STATE_UNINTERRUPTIBLE`, 4: waiting on the disk, the pager or a
/// kernel lock, to run again without anything arriving), the rule the meter
/// settles by. `proc_pidinfo(pid, PROC_PIDLISTTHREADS = 6)` into a raw buffer
/// of 64-bit thread handles, then `proc_pidinfo(pid, PROC_PIDTHREADINFO = 5,
/// handle)` into a raw 112-byte buffer, `pth_run_state` the int32 at byte 24.
/// `None` if the listing fails (an exited process has no task).
fn oracle_at_rest(pid: i32) -> Option<bool> {
    let mut handles = [0u64; 512];
    // SAFETY: `handles` is writable for the bytes passed.
    let n = unsafe { libc::proc_pidinfo(pid, 6, 0, handles.as_mut_ptr().cast(), (handles.len() * 8) as libc::c_int) };
    if n <= 0 {
        return None;
    }
    let n = n as usize / 8;
    assert!(n < handles.len(), "{pid} has more threads than the oracle's buffer holds");
    Some(handles[..n].iter().all(|&handle| {
        let mut buf = [0u8; 112];
        // SAFETY: `buf` is writable for the 112 bytes passed.
        let got = unsafe { libc::proc_pidinfo(pid, 5, handle, buf.as_mut_ptr().cast(), 112) };
        // A thread gone since the listing waits on nothing.
        got != 112 || !matches!(i32::from_ne_bytes(buf[24..28].try_into().unwrap()), 1 | 4)
    }))
}

/// `pid`'s rusage once it is quiet by the oracle's rule: no thread running,
/// queued or in a wait no signal ends at two reads 20 ms apart, with the
/// same energy at both. Panics
/// after 20 s.
fn quiet(pid: i32) -> [u64; 64] {
    let deadline = Instant::now() + Duration::from_secs(20);
    loop {
        let first = oracle_rusage(pid).expect("oracle read");
        let before = oracle_running(pid) == Some(0) && oracle_at_rest(pid) == Some(true);
        std::thread::sleep(Duration::from_millis(20));
        let second = oracle_rusage(pid).expect("oracle read");
        let after = oracle_running(pid) == Some(0) && oracle_at_rest(pid) == Some(true);
        if before && after && energy_nj(&first) == energy_nj(&second) {
            return second;
        }
        assert!(Instant::now() < deadline, "the oracle never saw {pid} quiet");
    }
}

/// An MCP-ish server: a child in a process group of its own, one JSON-RPC
/// request per line on its stdin and one response per line on its stdout,
/// killed with its whole group on drop. It names its pid to the gateway
/// (`named`, normally its own), and is never waited on while it serves, so
/// its pid stays its own.
struct Server {
    child: Child,
    stdin: ChildStdin,
    stdout: BufReader<ChildStdout>,
    named: Option<u32>,
}

impl Server {
    fn spawn(program: &str, args: &[&str]) -> Server {
        let mut child = Command::new(program)
            .args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null())
            .process_group(0)
            .spawn()
            .expect("spawn the server");
        let stdin = child.stdin.take().unwrap();
        let stdout = BufReader::new(child.stdout.take().unwrap());
        let named = Some(child.id());
        Server { child, stdin, stdout, named }
    }

    /// awk answering every request itself: a single-process server.
    fn single_process() -> Server {
        Server::spawn("/usr/bin/awk", &[&format!("{{ l = $0; {REPLY}; fflush() }}")])
    }

    /// A shell answering each request with an awk worker it forks, waits on
    /// and reaps before it replies: every call reaps a descendant.
    fn reaping_a_worker_per_call() -> Server {
        Server::spawn(
            "/bin/sh",
            &[
                "-c",
                r#"while IFS= read -r l; do r=$(L="$l" /usr/bin/awk "$1"); printf '%s\n' "$r"; done"#,
                "sh",
                &format!(r#"BEGIN {{ l = ENVIRON["L"]; {REPLY} }}"#),
            ],
        )
    }

    fn pid(&self) -> i32 {
        self.child.id() as i32
    }
}

impl Upstream for Server {
    fn call(&mut self, line: &str) -> std::io::Result<String> {
        writeln!(self.stdin, "{line}")?;
        self.stdin.flush()?;
        let mut reply = String::new();
        self.stdout.read_line(&mut reply)?;
        Ok(reply.trim_end().to_string())
    }

    fn pid(&self) -> Option<u32> {
        self.named
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        // SAFETY: signals this server's own process group.
        unsafe { libc::kill(-self.pid(), libc::SIGKILL) };
        let _ = self.child.wait();
    }
}

/// A gateway granting `burn` a budget no test reaches.
fn gateway(declared: bool) -> Gateway {
    let mut g = Gateway::new("burner", &[("burn", 50_000_000_000)], steward_sk());
    g.set_upstream_settle_bound(TEST_SETTLE);
    if declared {
        g.declare_upstream_self_contained();
    }
    g
}

fn burn(id: i64, n: u64) -> String {
    json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"burn","arguments":{"n":n}}}).to_string()
}

/// A `burn` call that also asks the server to hand `k` lookups to a system
/// service (the delegating server below; others ignore `k`).
fn burn_delegating(id: i64, n: u64, k: u64) -> String {
    json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"burn","arguments":{"n":n,"k":k}}}).to_string()
}

/// What one allowed call came back as.
struct Call {
    cost_uj: u64,
    /// The call's sealed receipt, verified and decoded.
    sealed: Value,
    /// The plaintext beside the response, without the receipt's bytes.
    meta: Value,
}

fn relay(g: &mut Gateway, up: &mut Server, id: i64, n: u64) -> Call {
    relay_line(g, up, id, n, &burn(id, n))
}

/// [`relay`] for a request line of the caller's: its id is `id`, and the
/// server answers it `burned <n>`.
fn relay_line(g: &mut Gateway, up: &mut Server, id: i64, n: u64, line: &str) -> Call {
    let Relayed::Allowed { response, cost_uj, .. } = g.relay(line, up).expect("relay") else {
        panic!("the call was not allowed")
    };
    let pubkey = steward_sk().verifying_key().to_bytes();
    let core = jcp_receipt::jcr1::verify_cose_sign1(g.receipts().last().unwrap(), STEWARD, &pubkey).expect("the receipt verifies");
    let sealed: Value = jcp_schema::from_canonical_cbor(&core).expect("the receipt decodes");
    let response: Value = serde_json::from_str(&response).expect("a JSON response");
    let mut meta = response.pointer(&format!("/result/_meta/{}", JCP_META_NS.replace('/', "~1"))).expect("the jcp _meta").clone();
    meta.as_object_mut().unwrap().remove("receipt_b64");
    assert_eq!(response["id"].as_i64(), Some(id), "the server's own response, relayed: {response}");
    assert_eq!(response["result"]["content"][0]["text"].as_str(), Some(format!("burned {n}").as_str()), "{response}");
    assert_eq!(sealed["cost_uj"].as_u64(), Some(cost_uj), "{sealed}");
    assert_eq!(meta["cost_uj"].as_u64(), Some(cost_uj), "{meta}");
    // The invariant: the plaintext grade is the sealed grade.
    assert_eq!(meta.get("energy_provenance"), sealed.get("energy_provenance"), "{meta} vs {sealed}");
    assert_eq!(meta.get("energy_coverage"), sealed.get("energy_coverage"), "{meta} vs {sealed}");
    Call { cost_uj, sealed, meta }
}

fn grade(c: &Call) -> String {
    match c.sealed.get("energy_coverage").and_then(Value::as_str) {
        Some(coverage) => format!("{}, {coverage}", c.sealed["energy_provenance"].as_str().unwrap_or("?")),
        None => c.sealed["energy_provenance"].as_str().unwrap_or("?").to_string(),
    }
}

fn the_log_verifies(g: &Gateway) {
    let pubkey = steward_sk().verifying_key().to_bytes();
    assert_eq!(jcp_receipt::verify_log(g.receipts(), STEWARD, &pubkey), Ok(g.receipts().len()));
}

/// A single-process server whose operator declared it self-contained: each
/// call, from a bare round trip to a few milliseconds of CPU, is sealed
/// `HwShunt` and whole, read on the server's own process tree, and its
/// figure is the oracle's advance across the relay, exact to the µJ.
#[test]
fn a_declared_self_contained_server_is_read_whole_on_its_own_process() {
    let _one = alone();
    let mut s = Server::single_process();
    let pid = s.pid();
    let mut g = gateway(true);
    assert_eq!(
        g.call_meter(&s),
        CallMeter::UpstreamTree { pid: pid as u32 },
        "a child of this process, on a host whose per-process counter is live (every host this runs on)"
    );
    for (rep, n) in [0u64, 20_000, 150_000, 300_000, 0].into_iter().enumerate() {
        let before = energy_nj(&quiet(pid));
        let c = relay(&mut g, &mut s, rep as i64, n);
        let oracle = energy_nj(&quiet(pid)) - before;
        eprintln!(
            "upstream tree: declared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}",
            c.cost_uj,
            grade(&c),
            c.meta
        );
        assert!(oracle > 0, "the server's count advanced over the call");
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert!(c.sealed.get("energy_coverage").is_none(), "whole: {}", c.sealed);
        assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
        assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta);
        assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ");
    }
    the_log_verifies(&g);
}

/// The same server with no declaration: the same reading, to the µJ, but
/// sealed `Partial`, because nothing in the tree can show that the server
/// hands no work to a process outside it; the plaintext says that is why.
#[test]
fn an_undeclared_server_is_read_partial_for_want_of_the_declaration() {
    let _one = alone();
    let mut s = Server::single_process();
    let pid = s.pid();
    let mut g = gateway(false);
    assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
    for (rep, n) in [0u64, 150_000, 300_000].into_iter().enumerate() {
        let before = energy_nj(&quiet(pid));
        let c = relay(&mut g, &mut s, rep as i64, n);
        let oracle = energy_nj(&quiet(pid)) - before;
        eprintln!(
            "upstream tree: undeclared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}",
            c.cost_uj,
            grade(&c),
            c.meta
        );
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed);
        assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
        assert_eq!(c.meta["energy_partial_because"], json!(["not-declared-self-contained"]), "{}", c.meta);
        assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ");
    }
    the_log_verifies(&g);
}

/// A server that answers each call with a worker it forks and reaps (a
/// shell loop running awk) is `Partial` even when declared: the worker's
/// energy is in no counter once it is reaped. The figure is the shell's own
/// advance, exactly, and the plaintext names the reaped descendant as why.
#[test]
fn a_server_that_reaps_a_worker_per_call_is_partial_even_when_declared() {
    let _one = alone();
    let mut s = Server::reaping_a_worker_per_call();
    let pid = s.pid();
    let mut g = gateway(true);
    assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
    for (rep, n) in [20_000u64, 200_000, 200_000].into_iter().enumerate() {
        let before = quiet(pid);
        let c = relay(&mut g, &mut s, rep as i64, n);
        let after = quiet(pid);
        let oracle = energy_nj(&after) - energy_nj(&before);
        let reaped = reaped_cpu_ticks(&after) - reaped_cpu_ticks(&before);
        eprintln!(
            "upstream tree: declared shell reaping a worker, loop {n} rep {rep}: {} µJ [{}], oracle (the shell's own) {oracle} nJ, reaped {reaped} ticks, _meta {}",
            c.cost_uj,
            grade(&c),
            c.meta
        );
        assert!(reaped > 0, "setup: the worker ran and was reaped in the call");
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed);
        assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
        assert_eq!(c.meta["energy_partial_because"], json!(["reaped-descendant"]), "{}", c.meta);
        assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the shell's own advance, to the µJ");
    }
    the_log_verifies(&g);
}

/// A declared server whose worker writes the reply ITSELF, keeps computing
/// for a few milliseconds and exits, and is reaped by the server after the
/// reply has gone (perl, with 60 idle children beside it, so a walk of its
/// tree takes long enough for the reap to land inside one). The worker was
/// born in the call and spent energy in it, and is in no counter once
/// reaped, so no call is sealed whole — not even one whose stop walk read
/// the server before the reap and looked for the worker after it.
#[test]
fn a_worker_reaped_after_writing_the_reply_never_leaves_a_call_whole() {
    let _one = alone();
    let mut s = Server::spawn(
        "/usr/bin/perl",
        &[
            "-e",
            r#"$| = 1;
               for (1 .. 60) { if (fork() == 0) { sleep 1000 while 1 } }
               while (my $l = <STDIN>) {
                   my ($id) = $l =~ /"id":(\d+)/; my ($n) = $l =~ /"n":(\d+)/; $n //= 0;
                   my $p = fork();
                   if ($p == 0) {
                       print qq({"jsonrpc":"2.0","id":$id,"result":{"content":[{"type":"text","text":"burned $n"}]}}\n);
                       my $x = 0; $x += $_ for 1 .. $n; exit 0
                   }
                   waitpid($p, 0);
               }"#,
        ],
    );
    let pid = s.pid();
    let mut g = gateway(true);
    assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
    let _ = relay(&mut g, &mut s, 0, 0);
    let mut reasons: Vec<String> = Vec::new();
    for rep in 1..31 {
        let c = relay(&mut g, &mut s, rep, 150_000);
        eprintln!("upstream tree: declared server, worker reaped after its reply, rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a call whose worker is in no reading sealed whole: {}", c.meta);
        let because = c.meta["energy_partial_because"].as_array().expect("why it is partial");
        assert!(
            because.iter().any(|k| matches!(k.as_str(), Some("reaped-descendant" | "exited" | "unsettled-at-start" | "unsettled-at-stop" | "tree-changing"))),
            "{}",
            c.meta
        );
        reasons.extend(because.iter().filter_map(Value::as_str).map(str::to_string));
    }
    reasons.sort();
    reasons.dedup();
    eprintln!("upstream tree: declared server, worker reaped after its reply: reasons seen {reasons:?}");
    the_log_verifies(&g);
}

/// A Metal compute server speaking the tests' JSON-RPC, built from this
/// source when the test first runs: each request's `arguments.n` picks what
/// it does before it replies — 0 nothing on the GPU, 1 a compute kernel on
/// the command queue it keeps, 2 the same kernel on a queue made for the
/// request and released after it — and it replies `burned <n>`.
const METAL_SERVER: &str = r##"#import <Metal/Metal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
static NSString *src = @"#include <metal_stdlib>\nusing namespace metal;\n"
  "kernel void burn(device float *o [[buffer(0)]], uint i [[thread_position_in_grid]]) {\n"
  "  float x = float(i); for (int k = 0; k < 20000; k++) { x = x * 1.0000001f + 0.5f; } o[i] = x; }\n";
static long field(const char *line, const char *key) {
  const char *p = strstr(line, key);
  return p ? strtol(p + strlen(key), NULL, 10) : 0;
}
int main(void) {
  @autoreleasepool {
    id<MTLDevice> d = MTLCreateSystemDefaultDevice();
    if (!d) return 1;
    NSError *err = nil;
    id<MTLLibrary> lib = [d newLibraryWithSource:src options:nil error:&err];
    if (!lib) return 1;
    id<MTLComputePipelineState> ps = [d newComputePipelineStateWithFunction:[lib newFunctionWithName:@"burn"] error:&err];
    id<MTLCommandQueue> kept = [d newCommandQueue];
    id<MTLBuffer> buf = [d newBufferWithLength:sizeof(float) * 65536 options:MTLResourceStorageModeShared];
    char line[4096];
    while (fgets(line, sizeof line, stdin)) {
      long rid = field(line, "\"id\":"), n = field(line, "\"n\":");
      if (n == 1 || n == 2) {
        @autoreleasepool {
          id<MTLCommandQueue> q = n == 2 ? [d newCommandQueue] : kept;
          id<MTLCommandBuffer> b = [q commandBuffer];
          id<MTLComputeCommandEncoder> e = [b computeCommandEncoder];
          [e setComputePipelineState:ps];
          [e setBuffer:buf offset:0 atIndex:0];
          [e dispatchThreads:MTLSizeMake(65536, 1, 1) threadsPerThreadgroup:MTLSizeMake(256, 1, 1)];
          [e endEncoding];
          [b commit];
          [b waitUntilCompleted];
        }
      }
      printf("{\"jsonrpc\":\"2.0\",\"id\":%ld,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %ld\"}]}}\n", rid, n);
      fflush(stdout);
    }
  }
  return 0;
}
"##;

/// A declared server that holds a GPU client is never sealed whole, whatever
/// a call did: its kept queue's work is recorded GPU time ("gpu"), and a
/// call whose work ran on a queue released inside it, or that did no GPU
/// work, leaves the server's recorded time unmoved, which does not show it
/// used none ("gpu-unknown"). The meter waits only 200 ms for the server to
/// go quiet: a Metal process keeps threads that wake on their own, and a
/// server still running at an end is a reason of its own.
#[test]
fn a_declared_server_holding_a_gpu_client_is_never_sealed_whole() {
    let _one = alone();
    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"));
    let source = dir.join("upstream_tree_metal_server.m");
    let binary = dir.join(format!("upstream_tree_metal_server-{}", std::process::id()));
    std::fs::write(&source, METAL_SERVER).expect("write the Metal server's source");
    let out = Command::new("/usr/bin/cc")
        .args(["-fobjc-arc", "-O2", "-framework", "Foundation", "-framework", "Metal", "-o"])
        .arg(&binary)
        .arg(&source)
        .output()
        .expect("run the C compiler");
    assert!(out.status.success(), "the Metal server did not build:\n{}", String::from_utf8_lossy(&out.stderr));
    let mut s = Server::spawn(binary.to_str().unwrap(), &[]);
    let pid = s.pid();
    let mut g = gateway(true);
    g.set_upstream_settle_bound(Duration::from_millis(200));
    assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
    for (rep, n) in [1u64, 2, 0, 2, 1, 0].into_iter().enumerate() {
        let c = relay(&mut g, &mut s, rep as i64, n);
        eprintln!("upstream tree: declared Metal server, mode {n} rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a GPU-holding server sealed whole: {}", c.meta);
        let because: Vec<&str> = c.meta["energy_partial_because"].as_array().expect("why").iter().filter_map(Value::as_str).collect();
        let want = if n == 1 { "gpu" } else { "gpu-unknown" };
        assert!(because.contains(&want) || because.contains(&"gpu"), "{want} not among {because:?}");
    }
    the_log_verifies(&g);
}

/// An upstream naming a process this one may not read (launchd, another
/// user's) gets no tree meter, and its calls are metered machine-wide,
/// sealed as that meter grades them.
#[test]
fn an_upstream_naming_a_process_this_one_cannot_read_is_metered_machine_wide() {
    // SAFETY: no preconditions.
    if unsafe { libc::geteuid() } == 0 {
        eprintln!("upstream fallback: running as root, launchd is readable; nothing to check");
        return;
    }
    let _one = alone();
    let mut s = Server::single_process();
    s.named = Some(1);
    let mut g = gateway(true);
    let machine = if g.metering_is_hardware() { CallMeter::MachineHardware } else { CallMeter::Estimator };
    assert_eq!(g.call_meter(&s), machine);
    let c = relay(&mut g, &mut s, 1, 20_000);
    eprintln!("upstream fallback: pid 1 named: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta);
    assert_eq!(c.meta["energy_domain"].as_str(), Some("machine"), "{}", c.meta);
    assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta);
    if !g.metering_is_hardware() {
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("Estimator"), "{}", c.sealed);
    }
    the_log_verifies(&g);
}

/// A server for the delegated-work test, speaking the tests' JSON-RPC, built
/// from this source when the test first runs: each request burns its
/// `arguments.n` loop, then looks up `arguments.k` account names no account
/// has (`getpwnam`). A system directory service does each lookup, on a
/// message that carries this process's voucher, so the kernel bills the
/// service's CPU work for it to this process. When it looked any up, it
/// then waits — up to 5 s, in 1 ms sleeps — for its own billed counts to
/// move before it replies `burned <n>`: a servicing thread is credited when
/// it next changes voucher, which can be after the reply, and the test is
/// of how a bill that landed in the call is sealed, not of when the kernel
/// lands it. One process, no children, no GPU client.
const DELEGATING_SERVER: &str = r##"#include <libproc.h>
#include <pwd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <unistd.h>
static long field(const char *line, const char *key) {
  const char *p = strstr(line, key);
  return p ? strtol(p + strlen(key), NULL, 10) : 0;
}
static void billed(unsigned long long *nj, unsigned long long *ticks) {
  struct rusage_info_v6 r;
  memset(&r, 0, sizeof r);
  proc_pid_rusage(getpid(), RUSAGE_INFO_V6, (rusage_info_t *)&r);
  *nj = r.ri_billed_energy;
  *ticks = r.ri_billed_system_time;
}
int main(void) {
  char line[4096];
  unsigned long seq = 0;
  while (fgets(line, sizeof line, stdin)) {
    long rid = field(line, "\"id\":"), n = field(line, "\"n\":"), k = field(line, "\"k\":");
    volatile unsigned long long x = 1;
    for (long i = 0; i < n; i++) x = x * 6364136223846793005ULL + 1;
    if (k > 0) {
      unsigned long long nj0, t0, nj1, t1;
      billed(&nj0, &t0);
      for (long i = 0; i < k; i++) {
        char name[96];
        snprintf(name, sizeof name, "no-such-account-%d-%lu", (int)getpid(), seq++);
        (void)getpwnam(name);
      }
      for (int waited = 0; waited < 5000; waited++) {
        billed(&nj1, &t1);
        if (nj1 > nj0 || t1 > t0) break;
        usleep(1000);
      }
    }
    printf("{\"jsonrpc\":\"2.0\",\"id\":%ld,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %ld\"}]}}\n", rid, n);
    fflush(stdout);
  }
  return 0;
}
"##;

/// The server's rusage once its billed counts have held for 300 ms:
/// whatever was billed for its start, which can land late, has landed.
/// Panics after 20 s.
fn billed_quiet(pid: i32) {
    let deadline = Instant::now() + Duration::from_secs(20);
    loop {
        let a = oracle_rusage(pid).expect("oracle read");
        std::thread::sleep(Duration::from_millis(300));
        let b = oracle_rusage(pid).expect("oracle read");
        if (billed_nj(&a), billed_ticks(&a)) == (billed_nj(&b), billed_ticks(&b)) {
            return;
        }
        assert!(Instant::now() < deadline, "the oracle never saw {pid}'s billed counts hold");
    }
}

/// A server whose operator declared it self-contained, and whose calls hand
/// work to a system service, is sealed `Partial` with `delegated` among the
/// reasons: the kernel billed the service's CPU work to it, which is
/// evidence the declaration does not hold for that call, and the evidence
/// wins. Its figure holds the bill: at least the oracle's own advance less
/// what it serviced plus what it saw billed (the oracle reads inside the
/// call's span, so a bill landing late is in the gateway's figure and maybe
/// not the oracle's), rounded up to the µJ, and more than the server's own
/// advance alone. The same server's calls that delegate nothing come first
/// (once anything billed for its start has landed), as the control: whole,
/// and exact to the µJ against the oracle.
#[test]
fn a_declared_server_whose_calls_delegate_is_sealed_partial_for_the_delegation() {
    let _one = alone();
    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"));
    let source = dir.join("upstream_tree_delegating_server.c");
    let binary = dir.join(format!("upstream_tree_delegating_server-{}", std::process::id()));
    std::fs::write(&source, DELEGATING_SERVER).expect("write the delegating server's source");
    let out = Command::new("/usr/bin/cc").args(["-O2", "-o"]).arg(&binary).arg(&source).output().expect("run the C compiler");
    assert!(out.status.success(), "the delegating server did not build:\n{}", String::from_utf8_lossy(&out.stderr));
    let mut s = Server::spawn(binary.to_str().unwrap(), &[]);
    let pid = s.pid();
    let mut g = gateway(true);
    assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 });
    // The first call starts the server's work; its start's bills land after.
    let _ = relay(&mut g, &mut s, 0, 0);
    billed_quiet(pid);
    for rep in 1..4 {
        let before = quiet(pid);
        let c = relay(&mut g, &mut s, rep, 150_000);
        let after = quiet(pid);
        let own = energy_nj(&after) - energy_nj(&before);
        eprintln!(
            "upstream tree: declared delegating server, control rep {rep}: {} µJ [{}], oracle own {own} nJ, billed {} nJ, _meta {}",
            c.cost_uj,
            grade(&c),
            billed_nj(&after) - billed_nj(&before),
            c.meta
        );
        assert_eq!((billed_nj(&after), billed_ticks(&after)), (billed_nj(&before), billed_ticks(&before)), "setup: the control billed nothing");
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert!(c.sealed.get("energy_coverage").is_none(), "the control is whole: {}", c.sealed);
        assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta);
        assert_eq!(c.cost_uj, own.div_ceil(1_000), "the control against the oracle's advance, to the µJ");
    }
    for rep in 4..8 {
        let before = quiet(pid);
        let c = relay_line(&mut g, &mut s, rep, 20_000, &burn_delegating(rep, 20_000, 3));
        let after = quiet(pid);
        let own = i128::from(energy_nj(&after) - energy_nj(&before));
        let serviced = i128::from(serviced_nj(&after) - serviced_nj(&before));
        let billed = billed_nj(&after) - billed_nj(&before);
        let billed_cpu = billed_ticks(&after) - billed_ticks(&before);
        eprintln!(
            "upstream tree: declared delegating server, delegating rep {rep}: {} µJ [{}], oracle own {own} nJ, serviced {serviced} nJ, billed {billed} nJ / {billed_cpu} ticks, _meta {}",
            c.cost_uj,
            grade(&c),
            c.meta
        );
        assert!(billed > 0 || billed_cpu > 0, "setup: the oracle saw no work billed to the server");
        assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed);
        assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a delegating call sealed whole under the declaration: {}", c.sealed);
        assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta);
        assert_eq!(c.meta["energy_partial_because"], json!(["delegated"]), "{}", c.meta);
        let floor = u64::try_from(own - serviced + i128::from(billed)).expect("a positive oracle figure");
        assert!(c.cost_uj >= floor.div_ceil(1_000), "the figure ({} µJ) is under the oracle's own − serviced + billed ({floor} nJ)", c.cost_uj);
        assert!(billed == 0 || i128::from(c.cost_uj) * 1_000 > own - serviced, "the bill was not added to the server's own figure");
    }
    the_log_verifies(&g);
}