diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index 4ae6ae2029..4cfd70c64d 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -470,6 +470,54 @@ jobs: fi echo "gate-contract test coverage allowlist is current" + # Report-only (#5046). Measures steady-state RSS of an embedded openhuman_core + # agent roster and uploads the raw samples + a human summary. Deliberately NOT + # in `pr-ci-gate.needs`, so it can never fail a PR — it exists to accrue a + # baseline and its runner variance before the 30 MiB gate is flipped to + # blocking in a follow-up (add this job to pr-ci-gate + a threshold step then). + rust-rss-bench: + name: Rust RSS Benchmark (report-only) + needs: [changes] + if: needs.changes.outputs['rust-core'] == 'true' + runs-on: ubuntu-22.04 + timeout-minutes: 40 + container: + image: ghcr.io/tinyhumansai/openhuman_ci:rust-1.93.0 + env: + CARGO_INCREMENTAL: "0" + steps: + - name: Checkout code + uses: actions/checkout@v7 + with: + fetch-depth: 1 + persist-credentials: false + submodules: recursive + + - name: Cache Rust build artifacts + uses: Swatinem/rust-cache@v2 + with: + workspaces: | + . -> target + cache-on-failure: true + shared-key: pr-rust-rss-bench + + # Fixture-contract signal: the gated bin's unit tests (build_roster, + # warm-up turn) never enter the default coverage lane, so run them here. + - name: Run rss-bench fixture tests + run: bash scripts/ci-cancel-aware.sh cargo test --features rss-bench --bin rss-bench + + - name: Build stripped-release rss-bench + run: bash scripts/ci-cancel-aware.sh cargo build --release --features rss-bench --bin rss-bench + + - name: Measure embedded RSS (5 fresh procs x {1,8} agents) + run: ./target/release/rss-bench --out bench-rss.json | tee -a "$GITHUB_STEP_SUMMARY" + + - name: Upload raw RSS samples + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: rss-bench-report + path: bench-rss.json + rust-core-coverage: name: Rust Core Coverage (cargo-llvm-cov) needs: [changes, rust-quality] diff --git a/Cargo.toml b/Cargo.toml index aacb8d574b..c540354def 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,14 @@ path = "src/bin/test_mcp_stub.rs" name = "openhuman-fleet" path = "src/bin/fleet.rs" +# Embedded-RSS benchmark harness (#5046). Gated behind the default-OFF +# `rss-bench` feature so no benchmark code enters the shipped build. Build with +# `cargo build --release --features rss-bench --bin rss-bench`. +[[bin]] +name = "rss-bench" +path = "src/bin/rss_bench.rs" +required-features = ["rss-bench"] + [lib] name = "openhuman_core" crate-type = ["rlib"] @@ -491,6 +499,10 @@ whatsapp-web = ["tinychannels/whatsapp-web"] # build (app/scripts/e2e-build.sh) flips it on. Shipped binaries never have # this feature so the wipe RPC isn't even registered, let alone reachable. e2e-test-support = [] +# Builds the `rss-bench` benchmark binary (#5046). Default-OFF, so it is never +# part of the shipped desktop/library build and the feature-forwarding gate +# (which only inspects the `default` list) never requires forwarding it. +rss-bench = [] [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage)'] } diff --git a/scripts/__tests__/feature-forwarding.test.mjs b/scripts/__tests__/feature-forwarding.test.mjs index 745c3a6c82..3bdd62379d 100644 --- a/scripts/__tests__/feature-forwarding.test.mjs +++ b/scripts/__tests__/feature-forwarding.test.mjs @@ -7,6 +7,7 @@ import { fileURLToPath } from 'node:url'; import { diffForwarding, + INTENTIONALLY_NOT_FORWARDED, parseCoreDefaultFeatures, parseShellForwardedFeatures, stripComments, @@ -179,6 +180,9 @@ test('the real shell manifest forwards every real core default', () => { assert.ok(coreDefaults.length > 0, 'expected to parse at least one core default gate'); assert.equal(shell.defaultFeatures, false, 'shell is expected to set default-features = false'); for (const gate of coreDefaults) { + // Gates the shell intentionally does not forward (e.g. `tui` — a terminal + // subcommand the desktop app never runs) are exempt, matching the checker. + if (INTENTIONALLY_NOT_FORWARDED[gate]) continue; assert.ok( shell.features.includes(gate), `core default gate not forwarded to the shell: ${gate}` diff --git a/scripts/ci/check-feature-forwarding.mjs b/scripts/ci/check-feature-forwarding.mjs index a346f4c7dd..4a8e410dad 100644 --- a/scripts/ci/check-feature-forwarding.mjs +++ b/scripts/ci/check-feature-forwarding.mjs @@ -15,26 +15,13 @@ import { fileURLToPath } from 'node:url'; import { diffForwarding, formatReport, + INTENTIONALLY_NOT_FORWARDED, parseCoreDefaultFeatures, parseShellForwardedFeatures, } from '../lib/feature-forwarding.mjs'; const REPO_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); -/** - * Gates the desktop shell intentionally does NOT forward, mapped to why. - * - * Empty by design: every current default-ON gate belongs in the shipped app. - * Adding an entry is a deliberate product decision, not a way to silence this - * check — the reason string is what a future reader (and reviewer) relies on to - * tell "excluded on purpose" from "forgotten". That ambiguity is exactly what - * let #4918 sit unnoticed since #4123. - */ -const INTENTIONALLY_NOT_FORWARDED = { - // 'some-gate': 'Reason it must not ship in the desktop build.', - tui: 'Terminal UI subcommand (openhuman tui/chat); the desktop app ships its own Tauri UI and never runs the ratatui terminal front-end.', -}; - function usage() { return 'Usage: check-feature-forwarding.mjs [core-manifest] [shell-manifest]'; } diff --git a/scripts/lib/feature-forwarding.mjs b/scripts/lib/feature-forwarding.mjs index a105e0646b..91195ef39f 100644 --- a/scripts/lib/feature-forwarding.mjs +++ b/scripts/lib/feature-forwarding.mjs @@ -20,6 +20,23 @@ // only needs two well-known shapes, and the repo has no TOML dependency for // Node. It is regex/scanner-based in the same spirit as `checklist-parser.mjs`. +/** + * Gates the desktop shell intentionally does NOT forward, mapped to why. + * + * Adding an entry is a deliberate product decision, not a way to silence the + * forwarding guard — the reason string is what a future reader (and reviewer) + * relies on to tell "excluded on purpose" from "forgotten". That ambiguity is + * exactly what let #4918 sit unnoticed since #4123. + * + * Lives here (not in the checker) so both the CI checker and the self-test read + * the same source of truth — otherwise the self-test can demand a forward the + * checker legitimately exempts, which is exactly the drift #5084's `tui` gate hit. + */ +export const INTENTIONALLY_NOT_FORWARDED = { + // 'some-gate': 'Reason it must not ship in the desktop build.', + tui: 'Terminal UI subcommand (openhuman tui/chat); the desktop app ships its own Tauri UI and never runs the ratatui terminal front-end.', +}; + /** * Strip TOML `#` comments while respecting quoted strings, so a `#` inside a * value (or an issue number in a comment) can't truncate a real line. diff --git a/src/bin/rss_bench.rs b/src/bin/rss_bench.rs new file mode 100644 index 0000000000..493cc63df0 --- /dev/null +++ b/src/bin/rss_bench.rs @@ -0,0 +1,436 @@ +//! `rss-bench` — steady-state RSS benchmark for an embedded `openhuman_core` +//! agent roster (#5046). +//! +//! Mirrors the OpenCompany embedding contract: a bare [`Agent`] built directly +//! via [`Agent::builder`] (no `CoreBuilder`, no RPC, no background services) +//! with an injected mock provider, an in-process `"none"` memory backend, and a +//! per-agent temp workspace. Builds a 1-agent and an 8-agent roster, runs one +//! deterministic warm-up turn per agent to fault in lazy allocations, settles, +//! then samples `/proc/self/{status,smaps_rollup}`. +//! +//! Two modes: +//! * `--child --roster N` builds one roster in a **fresh process**, warms up, +//! settles, and prints one [`ProcSample`] JSON line. This is the isolated +//! measured workload. +//! * default (parent) re-execs itself `--repeat` times per roster size to get +//! independent cold samples, aggregates, writes the raw JSON report +//! (`--out`), and prints a human summary. +//! +//! Gated behind the default-OFF `rss-bench` feature so no benchmark code enters +//! the shipped desktop/library build. Build & run: +//! `cargo build --release --features rss-bench --bin rss-bench`. +//! +//! The pure sampling/aggregation logic lives in +//! [`openhuman_core::openhuman::proc_metrics`]; this binary is the fixture + +//! process driver. + +use anyhow::{Context, Result}; +use async_trait::async_trait; +use openhuman_core::openhuman::agent::dispatcher::NativeToolDispatcher; +use openhuman_core::openhuman::agent::Agent; +use openhuman_core::openhuman::inference::provider::{ + ChatRequest, ChatResponse, Provider, UsageInfo, +}; +use openhuman_core::openhuman::memory::{ + Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts, +}; +use openhuman_core::openhuman::proc_metrics::{ + self, BenchReport, ProcSample, RosterResult, REPORT_SCHEMA_VERSION, RSS_BUDGET_KIB, + RSS_HARD_CAP_KIB, +}; +use openhuman_core::openhuman::tools::{Tool, ToolResult}; +use std::path::PathBuf; +use std::process::Stdio; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; + +/// Roster sizes measured by default: the 1-agent baseline and the +/// representative 8-agent company roster from #5046. +const DEFAULT_ROSTER_SIZES: &[usize] = &[1, 8]; +/// Fresh processes sampled per roster size (≥ 5 per the issue). +const DEFAULT_REPEAT: usize = 5; +/// Per-child wall-clock budget. A child does bounded work (build a roster, one +/// warm-up turn, a ≤2 s settle), so anything beyond this is a stall — kill it and +/// fail the run rather than letting one bad child block the whole benchmark until +/// the outer CI job timeout. +const CHILD_TIMEOUT: Duration = Duration::from_secs(120); + +/// Provider that never touches the network: returns a fixed assistant message +/// with a `stop` shape (no tool calls) so a turn completes in one round-trip. +struct MockProvider; + +#[async_trait] +impl Provider for MockProvider { + async fn chat_with_system( + &self, + _system_prompt: Option<&str>, + _message: &str, + _model: &str, + _temperature: f64, + ) -> Result { + Ok("ok".into()) + } + + async fn chat( + &self, + _request: ChatRequest<'_>, + _model: &str, + _temperature: f64, + ) -> Result { + Ok(ChatResponse { + text: Some("ok".into()), + tool_calls: vec![], + usage: Some(UsageInfo { + input_tokens: 8, + output_tokens: 2, + context_window: 8000, + charged_amount_usd: 0.0, + ..Default::default() + }), + reasoning_content: None, + }) + } +} + +/// Trivial host-supplied tool so the roster mirrors a real embedding (the host +/// injects its own tools). Never invoked — the provider returns no tool calls. +struct EchoTool; + +#[async_trait] +impl Tool for EchoTool { + fn name(&self) -> &str { + "echo" + } + + fn description(&self) -> &str { + "echo" + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ "type": "object" }) + } + + async fn execute(&self, _args: serde_json::Value) -> Result { + Ok(ToolResult::success("echo")) + } +} + +/// Zero-allocation no-op `Memory` for the fixture. +/// +/// The benchmark measures a bare agent under the OpenCompany embedding contract +/// (host supplies its own `Memory` over its context store), *not* a memory +/// store. `create_memory(MemoryConfig{ backend: "none", .. })` does **not** +/// select a no-op backend — it always builds a `UnifiedMemory` (SQLite + the +/// default cloud embedder), which would inflate the measured RSS with +/// memory-store setup. Injecting a real no-op keeps the reading on the agent +/// harness itself. +struct NoopMemory; + +#[async_trait] +impl Memory for NoopMemory { + fn name(&self) -> &str { + "noop" + } + async fn store( + &self, + _namespace: &str, + _key: &str, + _content: &str, + _category: MemoryCategory, + _session_id: Option<&str>, + ) -> Result<()> { + Ok(()) + } + async fn recall( + &self, + _query: &str, + _limit: usize, + _opts: RecallOpts<'_>, + ) -> Result> { + Ok(Vec::new()) + } + async fn get(&self, _namespace: &str, _key: &str) -> Result> { + Ok(None) + } + async fn list( + &self, + _namespace: Option<&str>, + _category: Option<&MemoryCategory>, + _session_id: Option<&str>, + ) -> Result> { + Ok(Vec::new()) + } + async fn forget(&self, _namespace: &str, _key: &str) -> Result { + Ok(false) + } + async fn namespace_summaries(&self) -> Result> { + Ok(Vec::new()) + } + async fn count(&self) -> Result { + Ok(0) + } + async fn health_check(&self) -> bool { + true + } +} + +/// A built roster plus the temp workspaces that must outlive it — dropping the +/// `TempDir`s would delete the agents' workspaces mid-measurement. +struct Roster { + agents: Vec, + _workspaces: Vec, +} + +/// Build `n` bare agents, each with its own temp workspace, mock provider, +/// `"none"` memory backend, and a single host-supplied tool. +fn build_roster(n: usize) -> Result { + let mut agents = Vec::with_capacity(n); + let mut workspaces = Vec::with_capacity(n); + for i in 0..n { + let workspace = TempDir::new().context("create temp workspace")?; + let path = workspace.path().to_path_buf(); + + let memory: Arc = Arc::new(NoopMemory); + + let agent = Agent::builder() + .provider(Box::new(MockProvider)) + .tools(vec![Box::new(EchoTool)]) + .memory(memory) + .tool_dispatcher(Box::new(NativeToolDispatcher)) + .model_name("bench-mock".into()) + .agent_definition_name(format!("bench-{i}")) + .workspace_dir(path.clone()) + .action_dir(path) + .auto_save(false) + .build() + .context("build bench agent")?; + agents.push(agent); + workspaces.push(workspace); + } + Ok(Roster { + agents, + _workspaces: workspaces, + }) +} + +/// One deterministic warm-up turn per agent, forcing first-touch allocations +/// (prompt build, tokenizer, provider adapter) to fault in before measuring. +async fn warm_up(roster: &mut Roster) -> Result<()> { + for agent in &mut roster.agents { + let _ = agent.turn("warmup").await.context("warm-up turn")?; + } + Ok(()) +} + +/// Poll RSS until it stops climbing (Δ < 256 KiB between reads ~200 ms apart) +/// or a 2 s cap, draining async task-allocation jitter. No-op off Linux. +async fn settle() { + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let mut last = proc_metrics::sample_self().map(|s| s.rss_kib).unwrap_or(0); + loop { + tokio::time::sleep(Duration::from_millis(200)).await; + let now = proc_metrics::sample_self() + .map(|s| s.rss_kib) + .unwrap_or(last); + if now.abs_diff(last) < 256 || tokio::time::Instant::now() >= deadline { + break; + } + last = now; + } +} + +/// Child mode: build one roster in this fresh process, warm up, settle, sample, +/// and print the sample as a single JSON line on stdout. +async fn run_child(roster_size: usize) -> Result<()> { + // Diagnostics go to stderr so they never corrupt the single JSON line the + // parent parses from stdout. + eprintln!("[rss-bench] child: building {roster_size}-agent roster"); + let mut roster = build_roster(roster_size)?; + eprintln!("[rss-bench] child: warming up {roster_size} agent(s)"); + warm_up(&mut roster).await?; + eprintln!("[rss-bench] child: settling"); + settle().await; + let sample = proc_metrics::sample_self()?; + eprintln!( + "[rss-bench] child: sampled rss={}KiB threads={}", + sample.rss_kib, sample.threads + ); + println!("{}", serde_json::to_string(&sample)?); + drop(roster); // keep the roster alive until after the sample is taken + Ok(()) +} + +/// Parent mode: re-exec the child `repeat` times per roster size, aggregate, and +/// write the report. +async fn run_parent(out: Option, repeat: usize, roster_sizes: &[usize]) -> Result<()> { + let exe = std::env::current_exe().context("resolve current exe")?; + let mut rosters = Vec::with_capacity(roster_sizes.len()); + for &size in roster_sizes { + let mut samples = Vec::with_capacity(repeat); + for run in 0..repeat { + eprintln!("[rss-bench] spawn child roster={size} run={run}"); + // `kill_on_drop` + a `timeout` around `wait_with_output` gives a + // robust kill-and-reap: on timeout the cancelled future drops the + // child, `kill_on_drop` sends SIGKILL, and the tokio runtime reaps it. + let child = tokio::process::Command::new(&exe) + .arg("--child") + .arg("--roster") + .arg(size.to_string()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .with_context(|| format!("spawn child roster={size} run={run}"))?; + let output = match tokio::time::timeout(CHILD_TIMEOUT, child.wait_with_output()).await { + Ok(res) => res.with_context(|| format!("await child roster={size} run={run}"))?, + Err(_) => { + eprintln!( + "[rss-bench] child roster={size} run={run} timed out after {}s; killed", + CHILD_TIMEOUT.as_secs() + ); + anyhow::bail!( + "child roster={size} run={run} timed out after {}s", + CHILD_TIMEOUT.as_secs() + ); + } + }; + if !output.status.success() { + eprintln!("[rss-bench] child roster={size} run={run} exited non-zero"); + anyhow::bail!( + "child roster={size} run={run} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + let stdout = String::from_utf8_lossy(&output.stdout); + let line = stdout + .lines() + .rev() + .find(|l| !l.trim().is_empty()) + .unwrap_or("") + .trim(); + let sample: ProcSample = serde_json::from_str(line) + .with_context(|| format!("parse child sample (roster={size}): {line:?}"))?; + eprintln!( + "[rss-bench] child roster={size} run={run} ok rss={}KiB", + sample.rss_kib + ); + samples.push(sample); + } + rosters.push(RosterResult::from_samples(size, samples)); + } + + let report = BenchReport { + schema_version: REPORT_SCHEMA_VERSION, + git_sha: git_sha(), + kernel: kernel(), + rss_budget_kib: RSS_BUDGET_KIB, + rss_hard_cap_kib: RSS_HARD_CAP_KIB, + rosters, + }; + + if let Some(path) = out { + std::fs::write(&path, serde_json::to_string_pretty(&report)?) + .with_context(|| format!("write report to {}", path.display()))?; + } + println!("{}", proc_metrics::human_summary(&report)); + Ok(()) +} + +/// Best-effort commit id for the report header (`GITHUB_SHA` in CI). +fn git_sha() -> String { + std::env::var("GITHUB_SHA").unwrap_or_else(|_| "unknown".into()) +} + +/// Best-effort kernel version for the report header. +fn kernel() -> String { + std::fs::read_to_string("/proc/sys/kernel/osrelease") + .map(|s| s.trim().to_string()) + .unwrap_or_else(|_| std::env::consts::OS.to_string()) +} + +/// Minimal flag parse — avoids a clap dependency for four flags. +struct Args { + child: bool, + roster: usize, + repeat: usize, + out: Option, +} + +fn parse_args() -> Result { + let mut child = false; + let mut roster = 1usize; + let mut repeat = DEFAULT_REPEAT; + let mut out = None; + let mut it = std::env::args().skip(1); + while let Some(arg) = it.next() { + match arg.as_str() { + "--child" => child = true, + "--roster" => { + roster = it + .next() + .context("--roster needs a value")? + .parse() + .context("--roster value")?; + } + "--repeat" => { + repeat = it + .next() + .context("--repeat needs a value")? + .parse() + .context("--repeat value")?; + } + "--out" => out = Some(PathBuf::from(it.next().context("--out needs a path")?)), + other => anyhow::bail!("unknown argument: {other}"), + } + } + Ok(Args { + child, + roster, + repeat, + out, + }) +} + +#[tokio::main] +async fn main() -> Result<()> { + let args = parse_args()?; + if args.child { + run_child(args.roster).await + } else { + run_parent(args.out, args.repeat, DEFAULT_ROSTER_SIZES).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_roster_constructs_bare_agents_with_isolated_workspaces() { + let roster = build_roster(8).expect("8-agent roster builds"); + assert_eq!(roster.agents.len(), 8); + assert_eq!(roster._workspaces.len(), 8); + // Each agent got a distinct workspace directory. + let mut dirs: Vec<_> = roster + ._workspaces + .iter() + .map(|w| w.path().to_path_buf()) + .collect(); + dirs.sort(); + dirs.dedup(); + assert_eq!(dirs.len(), 8, "workspaces must be isolated per agent"); + } + + #[tokio::test] + async fn warm_up_turn_completes_without_network() { + let mut roster = build_roster(1).expect("1-agent roster builds"); + warm_up(&mut roster).await.expect("warm-up turn completes"); + // The mock provider reports usage, so last_turn_usage is populated — + // proving the embedding cost-metering contract works on the bare Agent. + assert!( + roster.agents[0].last_turn_usage().is_some(), + "usage should be readable after a turn" + ); + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index dc26574017..e0a89af1c0 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -96,6 +96,7 @@ pub mod orchestration; pub mod overlay; pub mod people; pub mod plan_review; +pub mod proc_metrics; pub mod profiles; pub mod prompt_injection; pub mod provider_surfaces; diff --git a/src/openhuman/proc_metrics/mod.rs b/src/openhuman/proc_metrics/mod.rs new file mode 100644 index 0000000000..44b0365310 --- /dev/null +++ b/src/openhuman/proc_metrics/mod.rs @@ -0,0 +1,446 @@ +//! Process memory sampling from Linux `/proc`. +//! +//! Reads the current process's resident-memory breakdown from +//! `/proc/self/smaps_rollup` + `/proc/self/status` and aggregates repeated +//! samples into a [`RosterResult`] / [`BenchReport`]. Written for the +//! `rss-bench` benchmark harness (#5046), which measures the steady-state RSS +//! of an embedded `openhuman_core` agent roster against the 20–30 MiB budget, +//! but [`sample_self`] is a general capability: any caller wanting this +//! process's RSS / PSS / private-page / peak-RSS figures on Linux can use it. +//! +//! The parsers ([`parse_status`], [`parse_smaps_rollup`]) are OS-agnostic and +//! take `&str`, so they are unit-tested without a live `/proc`. [`sample_self`] +//! is Linux-only and returns a structured error elsewhere — it never fabricates +//! a reading (a macOS local run fails loudly rather than emitting garbage). + +use serde::{Deserialize, Serialize}; + +/// Product budget for the embedded roster, in KiB (#5046). Target the agent +/// roster should land under. +pub const RSS_BUDGET_KIB: u64 = 20 * 1024; +/// Hard cap for the embedded roster, in KiB (#5046). Steady-state RSS above +/// this fails the (eventually blocking) CI gate. +pub const RSS_HARD_CAP_KIB: u64 = 30 * 1024; + +/// One resident-memory sample of a single process. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProcSample { + /// Resident set size (`/proc/self/status` `VmRSS`). + pub rss_kib: u64, + /// Proportional set size (`/proc/self/smaps_rollup` `Pss`). + pub pss_kib: u64, + /// Private clean pages (`smaps_rollup` `Private_Clean`). + pub private_clean_kib: u64, + /// Private dirty pages (`smaps_rollup` `Private_Dirty`). + pub private_dirty_kib: u64, + /// Peak resident set size (`status` `VmHWM`). + pub vm_hwm_kib: u64, + /// Live thread count (`status` `Threads`). + pub threads: u64, + /// On-disk size of the running executable, in bytes. + pub binary_size_bytes: u64, +} + +/// Fields extracted from `/proc//status`. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct StatusFields { + pub vm_rss_kib: u64, + pub vm_hwm_kib: u64, + pub threads: u64, +} + +/// Fields extracted from `/proc//smaps_rollup`. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct SmapsRollupFields { + pub pss_kib: u64, + pub private_clean_kib: u64, + pub private_dirty_kib: u64, +} + +/// First whitespace-separated integer in a `/proc` value tail +/// (e.g. `"\t 1234 kB"` → `1234`). Zero when absent or unparsable. +fn first_u64(rest: &str) -> u64 { + rest.split_whitespace() + .next() + .and_then(|token| token.parse().ok()) + .unwrap_or(0) +} + +/// Parse the `VmRSS` / `VmHWM` / `Threads` lines out of `/proc//status`. +/// Missing keys stay zero. OS-agnostic — feed it the file contents. +pub fn parse_status(contents: &str) -> StatusFields { + let mut fields = StatusFields::default(); + for line in contents.lines() { + if let Some(rest) = line.strip_prefix("VmRSS:") { + fields.vm_rss_kib = first_u64(rest); + } else if let Some(rest) = line.strip_prefix("VmHWM:") { + fields.vm_hwm_kib = first_u64(rest); + } else if let Some(rest) = line.strip_prefix("Threads:") { + fields.threads = first_u64(rest); + } + } + fields +} + +/// Parse the `Pss` / `Private_Clean` / `Private_Dirty` lines out of +/// `/proc//smaps_rollup` (the pre-summed variant of `smaps`). Missing keys +/// stay zero. `strip_prefix` with the trailing colon avoids matching the +/// `Pss_Anon:` / `Pss_Dirty:` breakdown lines. +pub fn parse_smaps_rollup(contents: &str) -> SmapsRollupFields { + let mut fields = SmapsRollupFields::default(); + for line in contents.lines() { + if let Some(rest) = line.strip_prefix("Pss:") { + fields.pss_kib = first_u64(rest); + } else if let Some(rest) = line.strip_prefix("Private_Clean:") { + fields.private_clean_kib = first_u64(rest); + } else if let Some(rest) = line.strip_prefix("Private_Dirty:") { + fields.private_dirty_kib = first_u64(rest); + } + } + fields +} + +/// Sample this process's resident memory. Linux-only. +#[cfg(target_os = "linux")] +pub fn sample_self() -> anyhow::Result { + use anyhow::Context; + let status = std::fs::read_to_string("/proc/self/status").context("read /proc/self/status")?; + let smaps = std::fs::read_to_string("/proc/self/smaps_rollup") + .context("read /proc/self/smaps_rollup")?; + let status = parse_status(&status); + let smaps = parse_smaps_rollup(&smaps); + let binary_size_bytes = std::env::current_exe() + .and_then(std::fs::metadata) + .map(|meta| meta.len()) + .unwrap_or(0); + Ok(ProcSample { + rss_kib: status.vm_rss_kib, + pss_kib: smaps.pss_kib, + private_clean_kib: smaps.private_clean_kib, + private_dirty_kib: smaps.private_dirty_kib, + vm_hwm_kib: status.vm_hwm_kib, + threads: status.threads, + binary_size_bytes, + }) +} + +/// Sample this process's resident memory. Non-Linux stub — fails loudly rather +/// than fabricating a reading. +#[cfg(not(target_os = "linux"))] +pub fn sample_self() -> anyhow::Result { + anyhow::bail!( + "proc_metrics::sample_self requires Linux /proc/self/smaps_rollup + status (this is a {} build)", + std::env::consts::OS + ) +} + +/// Median of a slice of `u64`, averaging the two middle values for even counts. +/// Empty input yields zero. +fn median_u64(values: &[u64]) -> u64 { + if values.is_empty() { + return 0; + } + let mut sorted = values.to_vec(); + sorted.sort_unstable(); + let mid = sorted.len() / 2; + if sorted.len() % 2 == 1 { + sorted[mid] + } else { + // Average without overflow. + sorted[mid - 1] + (sorted[mid] - sorted[mid - 1]) / 2 + } +} + +/// Aggregated result for one roster size across several fresh-process samples. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RosterResult { + pub roster_size: usize, + pub sample_count: usize, + pub median_rss_kib: u64, + pub min_rss_kib: u64, + pub max_rss_kib: u64, + pub mean_rss_kib: u64, + pub median_pss_kib: u64, + pub max_vm_hwm_kib: u64, + pub median_threads: u64, + pub binary_size_bytes: u64, + /// The raw per-process samples, retained as a CI artifact. + pub samples: Vec, +} + +impl RosterResult { + /// Aggregate raw samples into the reported statistics. RSS is summarised as + /// median (steady-state), min/max/mean for spread; PSS/threads as median; + /// `VmHWM` as the max (peak) across processes. + pub fn from_samples(roster_size: usize, samples: Vec) -> Self { + let rss: Vec = samples.iter().map(|s| s.rss_kib).collect(); + let pss: Vec = samples.iter().map(|s| s.pss_kib).collect(); + let threads: Vec = samples.iter().map(|s| s.threads).collect(); + let count = samples.len(); + let mean_rss_kib = if count == 0 { + 0 + } else { + rss.iter().sum::() / count as u64 + }; + Self { + roster_size, + sample_count: count, + median_rss_kib: median_u64(&rss), + min_rss_kib: rss.iter().copied().min().unwrap_or(0), + max_rss_kib: rss.iter().copied().max().unwrap_or(0), + mean_rss_kib, + median_pss_kib: median_u64(&pss), + max_vm_hwm_kib: samples.iter().map(|s| s.vm_hwm_kib).max().unwrap_or(0), + median_threads: median_u64(&threads), + binary_size_bytes: samples.first().map(|s| s.binary_size_bytes).unwrap_or(0), + samples, + } + } +} + +/// The full benchmark report, serialized to the raw JSON CI artifact. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BenchReport { + pub schema_version: u32, + pub git_sha: String, + pub kernel: String, + pub rss_budget_kib: u64, + pub rss_hard_cap_kib: u64, + pub rosters: Vec, +} + +/// Current schema version for [`BenchReport`]; bump on any field change so +/// downstream trend tooling can detect format shifts. +pub const REPORT_SCHEMA_VERSION: u32 = 1; + +impl BenchReport { + /// Marginal steady-state RSS cost of each additional agent (#5046), derived + /// from the smallest and largest rosters measured: + /// `(median_rss(max) - median_rss(min)) / (max_size - min_size)`. + /// + /// Returns `(min_roster_size, max_roster_size, kib_per_agent)`, or `None` when + /// fewer than two distinct roster sizes were measured (no incremental cost is + /// derivable). For the default `{1, 8}` rosters this is the per-agent cost of + /// agents 2–8. + pub fn per_agent_increment_kib(&self) -> Option<(usize, usize, u64)> { + let min = self.rosters.iter().min_by_key(|r| r.roster_size)?; + let max = self.rosters.iter().max_by_key(|r| r.roster_size)?; + let span = max.roster_size.checked_sub(min.roster_size)?; + if span == 0 { + return None; + } + let per_agent = max.median_rss_kib.saturating_sub(min.median_rss_kib) / span as u64; + Some((min.roster_size, max.roster_size, per_agent)) + } +} + +fn kib_to_mib(kib: u64) -> f64 { + kib as f64 / 1024.0 +} + +/// Human-readable Markdown summary for stdout + `$GITHUB_STEP_SUMMARY`. +pub fn human_summary(report: &BenchReport) -> String { + use std::fmt::Write as _; + let mut out = String::new(); + let _ = writeln!(out, "### Embedded `openhuman_core` RSS benchmark (#5046)"); + let _ = writeln!(out); + let _ = writeln!( + out, + "kernel `{}` · git `{}` · target ≤ {:.0} MiB · hard cap ≤ {:.0} MiB", + report.kernel, + report.git_sha, + kib_to_mib(report.rss_budget_kib), + kib_to_mib(report.rss_hard_cap_kib), + ); + let _ = writeln!(out); + let _ = writeln!( + out, + "| roster | n | median RSS | min–max RSS | median PSS | peak VmHWM | threads | binary |" + ); + let _ = writeln!( + out, + "| ------ | - | ---------- | ----------- | ---------- | ---------- | ------- | ------ |" + ); + for r in &report.rosters { + let over = if r.median_rss_kib > report.rss_hard_cap_kib { + " ⚠️" + } else { + "" + }; + let _ = writeln!( + out, + "| {} agent{} | {} | {:.1} MiB{} | {:.1}–{:.1} MiB | {:.1} MiB | {:.1} MiB | {} | {:.1} MiB |", + r.roster_size, + if r.roster_size == 1 { "" } else { "s" }, + r.sample_count, + kib_to_mib(r.median_rss_kib), + over, + kib_to_mib(r.min_rss_kib), + kib_to_mib(r.max_rss_kib), + kib_to_mib(r.median_pss_kib), + kib_to_mib(r.max_vm_hwm_kib), + r.median_threads, + r.binary_size_bytes as f64 / (1024.0 * 1024.0), + ); + } + if let Some((min_size, max_size, per_agent_kib)) = report.per_agent_increment_kib() { + let _ = writeln!(out); + let _ = writeln!( + out, + "Per-agent increment (roster {min_size}→{max_size}): {:.2} MiB/agent", + kib_to_mib(per_agent_kib), + ); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + const SAMPLE_STATUS: &str = "Name:\trss-bench\nVmPeak:\t 123456 kB\nVmRSS:\t 20480 kB\nVmHWM:\t 24576 kB\nThreads:\t8\n"; + + const SAMPLE_SMAPS_ROLLUP: &str = "00400000-7fff00000000 ---p 00000000 00:00 0 [rollup]\nRss:\t 20480 kB\nPss:\t 18000 kB\nPss_Anon:\t 1000 kB\nPss_Dirty:\t 500 kB\nPrivate_Clean:\t 4096 kB\nPrivate_Dirty:\t 12000 kB\n"; + + #[test] + fn parse_status_extracts_rss_hwm_threads() { + let f = parse_status(SAMPLE_STATUS); + assert_eq!(f.vm_rss_kib, 20480); + assert_eq!(f.vm_hwm_kib, 24576); + assert_eq!(f.threads, 8); + } + + #[test] + fn parse_status_missing_keys_stay_zero() { + let f = parse_status("Name:\tx\nState:\tR\n"); + assert_eq!(f, StatusFields::default()); + } + + #[test] + fn parse_smaps_rollup_extracts_pss_and_private_pages() { + let f = parse_smaps_rollup(SAMPLE_SMAPS_ROLLUP); + assert_eq!(f.pss_kib, 18000); + assert_eq!(f.private_clean_kib, 4096); + assert_eq!(f.private_dirty_kib, 12000); + } + + #[test] + fn parse_smaps_rollup_does_not_match_pss_breakdown_lines() { + // `Pss_Anon:` / `Pss_Dirty:` must not be read as `Pss:`. + let f = parse_smaps_rollup("Pss_Anon:\t 9999 kB\nPss_Dirty:\t 8888 kB\n"); + assert_eq!(f.pss_kib, 0); + } + + fn sample(rss: u64, pss: u64, hwm: u64, threads: u64) -> ProcSample { + ProcSample { + rss_kib: rss, + pss_kib: pss, + private_clean_kib: 0, + private_dirty_kib: 0, + vm_hwm_kib: hwm, + threads, + binary_size_bytes: 1024, + } + } + + #[test] + fn median_handles_odd_and_even() { + assert_eq!(median_u64(&[]), 0); + assert_eq!(median_u64(&[5]), 5); + assert_eq!(median_u64(&[3, 1, 2]), 2); + assert_eq!(median_u64(&[1, 2, 3, 4]), 2); // (2+3)/2 floored -> 2 + } + + #[test] + fn from_samples_aggregates_rss_pss_and_peak() { + let samples = vec![ + sample(20000, 18000, 21000, 8), + sample(22000, 19000, 26000, 8), + sample(21000, 18500, 24000, 8), + ]; + let r = RosterResult::from_samples(8, samples); + assert_eq!(r.roster_size, 8); + assert_eq!(r.sample_count, 3); + assert_eq!(r.median_rss_kib, 21000); + assert_eq!(r.min_rss_kib, 20000); + assert_eq!(r.max_rss_kib, 22000); + assert_eq!(r.mean_rss_kib, 21000); + assert_eq!(r.max_vm_hwm_kib, 26000); // peak across processes + assert_eq!(r.median_threads, 8); + } + + #[test] + fn report_serde_round_trips() { + let report = BenchReport { + schema_version: REPORT_SCHEMA_VERSION, + git_sha: "abc123".into(), + kernel: "6.1.0".into(), + rss_budget_kib: RSS_BUDGET_KIB, + rss_hard_cap_kib: RSS_HARD_CAP_KIB, + rosters: vec![RosterResult::from_samples( + 1, + vec![sample(15000, 14000, 16000, 6)], + )], + }; + let json = serde_json::to_string(&report).unwrap(); + let back: BenchReport = serde_json::from_str(&json).unwrap(); + assert_eq!(back.rosters.len(), 1); + assert_eq!(back.rosters[0].samples[0], report.rosters[0].samples[0]); + assert_eq!(back.rss_hard_cap_kib, RSS_HARD_CAP_KIB); + } + + #[test] + fn human_summary_flags_over_cap_roster() { + let report = BenchReport { + schema_version: REPORT_SCHEMA_VERSION, + git_sha: "deadbeef".into(), + kernel: "6.1.0".into(), + rss_budget_kib: RSS_BUDGET_KIB, + rss_hard_cap_kib: RSS_HARD_CAP_KIB, + rosters: vec![RosterResult::from_samples( + 8, + vec![sample(40000, 30000, 42000, 12)], + )], + }; + let summary = human_summary(&report); + assert!(summary.contains("8 agents")); + assert!(summary.contains("⚠️"), "over-cap roster must be flagged"); + } + + #[test] + fn per_agent_increment_from_min_and_max_rosters() { + let report = BenchReport { + schema_version: REPORT_SCHEMA_VERSION, + git_sha: "x".into(), + kernel: "6.1.0".into(), + rss_budget_kib: RSS_BUDGET_KIB, + rss_hard_cap_kib: RSS_HARD_CAP_KIB, + rosters: vec![ + RosterResult::from_samples(1, vec![sample(20_000, 0, 0, 6)]), + RosterResult::from_samples(8, vec![sample(27_000, 0, 0, 6)]), + ], + }; + // (27000 - 20000) / (8 - 1) = 1000 KiB per agent. + assert_eq!(report.per_agent_increment_kib(), Some((1, 8, 1000))); + assert!(human_summary(&report).contains("Per-agent increment (roster 1→8)")); + } + + #[test] + fn per_agent_increment_none_for_single_roster() { + let report = BenchReport { + schema_version: REPORT_SCHEMA_VERSION, + git_sha: "x".into(), + kernel: "6.1.0".into(), + rss_budget_kib: RSS_BUDGET_KIB, + rss_hard_cap_kib: RSS_HARD_CAP_KIB, + rosters: vec![RosterResult::from_samples(1, vec![sample(20_000, 0, 0, 6)])], + }; + assert_eq!(report.per_agent_increment_kib(), None); + } + + #[cfg(not(target_os = "linux"))] + #[test] + fn sample_self_is_linux_only() { + assert!(sample_self().is_err()); + } +}