From b0b786fde9d24f68ab1d58aad5a3e2cb43be22bf Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 15:28:01 +0530 Subject: [PATCH 1/8] feat(bench): add embedded-RSS benchmark harness for openhuman_core (#5046) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the reproducible steady-state RSS benchmark that the 20-30 MiB embedded budget is measured against, plus a report-only CI lane. - `openhuman::proc_metrics` (ungated): Linux /proc/self/{status,smaps_rollup} sampling + roster aggregation + serde report + human summary. OS-agnostic parsers are unit-tested; `sample_self()` is Linux-only and errors elsewhere rather than fabricating a reading. - `rss-bench` binary (gated behind the default-OFF `rss-bench` feature): mirrors the OpenCompany embedding contract — a bare `Agent` via `Agent::builder()` (no CoreBuilder / RPC / background services), injected mock provider + `"none"` memory + per-agent temp workspace, 1- and 8-agent rosters, one warm-up turn + settle, sampled across 5 fresh child processes per roster. - Report-only `rust-rss-bench` CI job: uploads raw samples + a human summary; deliberately not in `pr-ci-gate.needs`, so it can never fail a PR until the 30 MiB gate is flipped to blocking in a follow-up. The default/shipped build is byte-identical (feature is default-OFF, no new deps). --- .github/workflows/ci-lite.yml | 48 ++++ Cargo.toml | 12 + src/bin/rss_bench.rs | 343 ++++++++++++++++++++++++++ src/openhuman/mod.rs | 1 + src/openhuman/proc_metrics/mod.rs | 386 ++++++++++++++++++++++++++++++ 5 files changed, 790 insertions(+) create mode 100644 src/bin/rss_bench.rs create mode 100644 src/openhuman/proc_metrics/mod.rs diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index a5162b892c..1bc7fb47a8 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -469,6 +469,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 14b8e35b92..c33e76a67f 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"] @@ -470,6 +478,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/src/bin/rss_bench.rs b/src/bin/rss_bench.rs new file mode 100644 index 0000000000..aedca19d9a --- /dev/null +++ b/src/bin/rss_bench.rs @@ -0,0 +1,343 @@ +//! `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::config::MemoryConfig; +use openhuman_core::openhuman::inference::provider::{ + ChatRequest, ChatResponse, Provider, UsageInfo, +}; +use openhuman_core::openhuman::memory::Memory; +use openhuman_core::openhuman::memory_store::create_memory; +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::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; + +/// 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")) + } +} + +/// 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_cfg = MemoryConfig { + backend: "none".into(), + ..MemoryConfig::default() + }; + let memory: Arc = + Arc::from(create_memory(&memory_cfg, &path).context("create 'none' memory backend")?); + + 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<()> { + let mut roster = build_roster(roster_size)?; + warm_up(&mut roster).await?; + settle().await; + let sample = proc_metrics::sample_self()?; + 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 { + let output = tokio::process::Command::new(&exe) + .arg("--child") + .arg("--roster") + .arg(size.to_string()) + .output() + .await + .with_context(|| format!("spawn child roster={size}"))?; + if !output.status.success() { + 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:?}"))?; + 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 b890b5bcaa..3f360d73f4 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -97,6 +97,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..272602a64a --- /dev/null +++ b/src/openhuman/proc_metrics/mod.rs @@ -0,0 +1,386 @@ +//! 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; + +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), + ); + } + 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"); + } + + #[cfg(not(target_os = "linux"))] + #[test] + fn sample_self_is_linux_only() { + assert!(sample_self().is_err()); + } +} From bdf02f7b70918899f183295e037282f0ec1b2066 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 16:21:48 +0530 Subject: [PATCH 2/8] feat(bench): report per-agent RSS increment from the roster spread (#5046) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add BenchReport::per_agent_increment_kib(), which derives the marginal steady-state RSS cost of each additional agent from the smallest and largest rosters — (median_rss(max) - median_rss(min)) / (max_size - min_size). For the default {1, 8} rosters this is the per-agent cost of agents 2-8, which the human summary now prints. Returns None when fewer than two distinct roster sizes were measured. Two unit tests cover the computed value and the single-roster case. Addresses the issue's "bounded per-agent growth" criterion (the report now reports the incremental cost, not just the two roster totals). --- src/openhuman/proc_metrics/mod.rs | 60 +++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/src/openhuman/proc_metrics/mod.rs b/src/openhuman/proc_metrics/mod.rs index 272602a64a..44b0365310 100644 --- a/src/openhuman/proc_metrics/mod.rs +++ b/src/openhuman/proc_metrics/mod.rs @@ -213,6 +213,27 @@ pub struct BenchReport { /// 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 } @@ -262,6 +283,14 @@ pub fn human_summary(report: &BenchReport) -> String { 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 } @@ -378,6 +407,37 @@ mod tests { 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() { From 1ab0459cdff6935c7b111a86b2c20794a26696d3 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 17:05:13 +0530 Subject: [PATCH 3/8] feat(bench): per-child timeout + reap + flow diagnostics for rss-bench (#5046) From CodeRabbit review: - Give each benchmark child a 120s wall-clock budget: spawn with kill_on_drop and wrap wait_with_output in a timeout, so a stalled child is killed and reaped rather than blocking the whole run until the outer CI job timeout. - Add stderr-only diagnostics for the benchmark flow (child spawn/timeout/exit, and roster build/warm-up/settle/sample in the child) so a run is observable; kept on stderr so they never corrupt the single JSON line the parent parses from stdout. --- src/bin/rss_bench.rs | 47 ++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 43 insertions(+), 4 deletions(-) diff --git a/src/bin/rss_bench.rs b/src/bin/rss_bench.rs index aedca19d9a..1f2bf77da2 100644 --- a/src/bin/rss_bench.rs +++ b/src/bin/rss_bench.rs @@ -40,6 +40,7 @@ use openhuman_core::openhuman::proc_metrics::{ }; 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; @@ -49,6 +50,11 @@ use tempfile::TempDir; 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. @@ -183,10 +189,19 @@ async fn settle() { /// 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(()) @@ -200,14 +215,34 @@ async fn run_parent(out: Option, repeat: usize, roster_sizes: &[usize]) for &size in roster_sizes { let mut samples = Vec::with_capacity(repeat); for run in 0..repeat { - let output = tokio::process::Command::new(&exe) + 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()) - .output() - .await - .with_context(|| format!("spawn child roster={size}"))?; + .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) @@ -222,6 +257,10 @@ async fn run_parent(out: Option, repeat: usize, roster_sizes: &[usize]) .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)); From 03f9546d99d1cd7fbca13d8abcf72a035822c9a7 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 17:14:48 +0530 Subject: [PATCH 4/8] fix(event-bus): raise test-only capacity floors to fix a full-suite flake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under cfg(test) the crate's ~12.6k unit tests share two process-global bounded resources sized for production: the event bus's 256-slot broadcast channel (src/core/event_bus/bus.rs) and the learning candidate ring's 1024 entries (src/openhuman/learning/candidate.rs, whose drainer does not run in the unit suite). Under the concurrent full-suite publish load a slow subscriber lags and drops events (RecvError::Lagged), and the candidate ring evicts a test's own candidates — intermittently failing bus-delivery tests such as learning::startup::tests::learning_subscriber_fires_with_no_channel_configured. Raise both floors to 65536 for the TEST BINARY ONLY (cfg(test)); production keeps 256 / 1024 and ships byte-identical. No new dependency, no serialization, no weakened assertion — the shared globals simply get enough headroom that neither drops nor evicts under the artificial suite-wide load. --- src/core/event_bus/bus.rs | 9 +++++++++ src/openhuman/learning/candidate.rs | 10 +++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/src/core/event_bus/bus.rs b/src/core/event_bus/bus.rs index 3faa9d63d0..b608e0de41 100644 --- a/src/core/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -41,6 +41,15 @@ pub fn init_global(capacity: usize) -> &'static EventBus { // is always safe from anywhere in the process once the bus is up. init_native_registry(); GLOBAL_BUS.get_or_init(|| { + // Under `cfg(test)` the crate's ~12.6k unit tests share this one + // process-global bus and publish concurrently, so the default 256-slot + // broadcast buffer lets a slow subscriber fall behind and drop events + // (`RecvError::Lagged`) — intermittently flaking bus-delivery tests such + // as `learning::startup::tests::learning_subscriber_fires_with_no_channel_configured`. + // Raise the buffer floor for the test binary only; production is + // unaffected (it publishes at human speed and keeps the caller's capacity). + #[cfg(test)] + let capacity = capacity.max(1 << 16); tracing::debug!(capacity, "[event_bus] initializing global singleton"); EventBus::create(capacity) }) diff --git a/src/openhuman/learning/candidate.rs b/src/openhuman/learning/candidate.rs index 95285df258..d063401fae 100644 --- a/src/openhuman/learning/candidate.rs +++ b/src/openhuman/learning/candidate.rs @@ -233,7 +233,15 @@ static GLOBAL_BUFFER: OnceLock = OnceLock::new(); /// Initialised on first call with a default capacity of 1024. All producers /// push into this buffer; the stability detector drains it. pub fn global() -> &'static Buffer { - GLOBAL_BUFFER.get_or_init(|| Buffer::new(1024)) + // Production default: a 1024-entry ring drained by the stability detector. + // Under `cfg(test)` the crate's full parallel suite shares this one global + // ring with no drainer running, so concurrent producers can evict a test's + // own candidates before it asserts on them. Enlarge for the test binary only. + #[cfg(not(test))] + let capacity = 1024; + #[cfg(test)] + let capacity = 1 << 16; + GLOBAL_BUFFER.get_or_init(|| Buffer::new(capacity)) } // ── Tests ──────────────────────────────────────────────────────────────────── From 56d30941fea5430acb9dc8a28bd0213de3b5cc06 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 17:38:52 +0530 Subject: [PATCH 5/8] fix(bench): inject a real no-op Memory so the RSS baseline isn't inflated (#5046) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review: `create_memory(MemoryConfig{ backend: "none", .. })` does not select a no-op backend — `create_memory_full` always builds a `UnifiedMemory` (SQLite + the default cloud embedder) regardless of `backend`. So every measured agent was allocating a memory store + embedder, inflating the 1/8-agent RSS baseline with memory-store setup rather than measuring the bare agent under the OpenCompany embedding contract (host supplies its own Memory). Replace the fixture's `create_memory` call with a zero-allocation `NoopMemory` that implements the `Memory` trait as inert (store discards, recall/list/get return empty, count 0). This mirrors the host-supplied-memory contract and keeps the reading on the agent harness itself. --- src/bin/rss_bench.rs | 72 ++++++++++++++++++++++++++++++++++++++------ 1 file changed, 63 insertions(+), 9 deletions(-) diff --git a/src/bin/rss_bench.rs b/src/bin/rss_bench.rs index 1f2bf77da2..493cc63df0 100644 --- a/src/bin/rss_bench.rs +++ b/src/bin/rss_bench.rs @@ -28,12 +28,12 @@ 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::config::MemoryConfig; use openhuman_core::openhuman::inference::provider::{ ChatRequest, ChatResponse, Provider, UsageInfo, }; -use openhuman_core::openhuman::memory::Memory; -use openhuman_core::openhuman::memory_store::create_memory; +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, @@ -116,6 +116,65 @@ impl Tool for EchoTool { } } +/// 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 { @@ -132,12 +191,7 @@ fn build_roster(n: usize) -> Result { let workspace = TempDir::new().context("create temp workspace")?; let path = workspace.path().to_path_buf(); - let memory_cfg = MemoryConfig { - backend: "none".into(), - ..MemoryConfig::default() - }; - let memory: Arc = - Arc::from(create_memory(&memory_cfg, &path).context("create 'none' memory backend")?); + let memory: Arc = Arc::new(NoopMemory); let agent = Agent::builder() .provider(Box::new(MockProvider)) From 1fb6cd1756e252c0f95e1d53903f67bd9c1ec04f Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 18:20:48 +0530 Subject: [PATCH 6/8] fix(test): isolate event-bus flakes properly; revert ineffective capacity floor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The earlier test-only capacity floor (5c49a969) did not fix the flake: the two failing tests share the process-global broadcast bus, so under the parallel suite they receive events published by *other* tests — a capacity bump cannot fix cross-test interference. Replace it with real isolation. - learning::startup: rewrite `learning_subscriber_fires_*` to publish a DocumentCanonicalized event on a private `EventBus::create(64)` with a locally-subscribed `EmailSignatureSubscriber`, instead of asserting on the shared global bus. Add a synchronous `register_email_signature_subscriber` guard test for the channel-independent registration path (#5003). - web_chat::event_bus: the automation-halt test reads from the shared web-channel bus; drain foreign events and match on `event == "automation_halt"` rather than taking whatever is first in the buffer. - Revert the `#[cfg(test)]` capacity floors in event_bus/bus.rs and learning/candidate.rs back to origin/main — no longer needed. --- src/core/event_bus/bus.rs | 9 ------ src/openhuman/learning/candidate.rs | 10 +----- src/openhuman/learning/startup.rs | 49 +++++++++++++++++++++++------ src/openhuman/web_chat/event_bus.rs | 29 +++++++++++++---- 4 files changed, 64 insertions(+), 33 deletions(-) diff --git a/src/core/event_bus/bus.rs b/src/core/event_bus/bus.rs index b608e0de41..3faa9d63d0 100644 --- a/src/core/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -41,15 +41,6 @@ pub fn init_global(capacity: usize) -> &'static EventBus { // is always safe from anywhere in the process once the bus is up. init_native_registry(); GLOBAL_BUS.get_or_init(|| { - // Under `cfg(test)` the crate's ~12.6k unit tests share this one - // process-global bus and publish concurrently, so the default 256-slot - // broadcast buffer lets a slow subscriber fall behind and drop events - // (`RecvError::Lagged`) — intermittently flaking bus-delivery tests such - // as `learning::startup::tests::learning_subscriber_fires_with_no_channel_configured`. - // Raise the buffer floor for the test binary only; production is - // unaffected (it publishes at human speed and keeps the caller's capacity). - #[cfg(test)] - let capacity = capacity.max(1 << 16); tracing::debug!(capacity, "[event_bus] initializing global singleton"); EventBus::create(capacity) }) diff --git a/src/openhuman/learning/candidate.rs b/src/openhuman/learning/candidate.rs index d063401fae..95285df258 100644 --- a/src/openhuman/learning/candidate.rs +++ b/src/openhuman/learning/candidate.rs @@ -233,15 +233,7 @@ static GLOBAL_BUFFER: OnceLock = OnceLock::new(); /// Initialised on first call with a default capacity of 1024. All producers /// push into this buffer; the stability detector drains it. pub fn global() -> &'static Buffer { - // Production default: a 1024-entry ring drained by the stability detector. - // Under `cfg(test)` the crate's full parallel suite shares this one global - // ring with no drainer running, so concurrent producers can evict a test's - // own candidates before it asserts on them. Enlarge for the test binary only. - #[cfg(not(test))] - let capacity = 1024; - #[cfg(test)] - let capacity = 1 << 16; - GLOBAL_BUFFER.get_or_init(|| Buffer::new(capacity)) + GLOBAL_BUFFER.get_or_init(|| Buffer::new(1024)) } // ── Tests ──────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/learning/startup.rs b/src/openhuman/learning/startup.rs index 2da5a5de62..2336bca79b 100644 --- a/src/openhuman/learning/startup.rs +++ b/src/openhuman/learning/startup.rs @@ -270,13 +270,20 @@ mod tests { } #[tokio::test] - async fn learning_subscriber_fires_with_no_channel_configured() { - init_global(DEFAULT_CAPACITY); - let (tmp, _client) = test_client(); - // Make the memory client ready so the full Platform wiring runs — no - // channel runtime is ever constructed in this test. - let _ = crate::openhuman::memory::global::init(tmp.path().join("workspace")); - register_learning_subscribers(tmp.path().to_path_buf()); + async fn learning_subscriber_fires_on_document_event() { + // The email-signature producer reacts to `DocumentCanonicalized` and pushes + // Identity candidates into the global candidate buffer. Driving this through + // the process-global event bus (as it originally did) is flaky under the full + // parallel suite: that singleton is shared by every test, so its receiver + // starves under load and drops this test's event. Exercise the subscriber on + // an ISOLATED bus instead — the repo's canonical test-isolation pattern (see + // `core::event_bus::bus`) — so delivery is deterministic. The #5003 + // regression (the subscriber registers on the channel-independent boot path) + // is guarded separately by + // `email_signature_subscriber_registers_on_channel_independent_path` below. + use crate::core::event_bus::EventBus; + use crate::openhuman::learning::extract::signature::EmailSignatureSubscriber; + use std::sync::Arc; let source_id = unique_source_id("e2e"); let body = signature_body(); @@ -286,12 +293,36 @@ mod tests { "signature body must yield at least one identity candidate" ); - publish_email_doc(&source_id, &body); + let bus = EventBus::create(64); + let _handle = bus.subscribe(Arc::new(EmailSignatureSubscriber)); + bus.publish(DomainEvent::DocumentCanonicalized { + source_id: source_id.clone(), + source_kind: "email".to_string(), + chunks_written: 1, + chunk_ids: vec![format!("{source_id}-c1")], + canonicalized_at: 0.0, + body_preview: Some(body), + }); + let got = wait_for_candidates(&source_id, expected).await; assert_eq!( got, expected, "email-signature subscriber must push the parsed identity candidates \ - with no channel configured anywhere (#5003)" + from a DocumentCanonicalized event" + ); + } + + /// #5003 regression guard (channel-independent boot path): the email-signature + /// subscriber must register even when no memory client / channel is configured. + /// Synchronous (no bus delivery), so it is not subject to the shared-bus flake. + #[tokio::test] + async fn email_signature_subscriber_registers_on_channel_independent_path() { + init_global(DEFAULT_CAPACITY); + assert!( + crate::openhuman::learning::extract::signature::register_email_signature_subscriber() + .is_some(), + "email-signature subscriber must register on the global bus once it is \ + initialised, independent of any channel configuration (#5003)" ); } diff --git a/src/openhuman/web_chat/event_bus.rs b/src/openhuman/web_chat/event_bus.rs index 0afdbe7646..ff525518ef 100644 --- a/src/openhuman/web_chat/event_bus.rs +++ b/src/openhuman/web_chat/event_bus.rs @@ -505,6 +505,27 @@ mod tests { /// - `args.reason` and `args.source` echoed from the domain event #[tokio::test] async fn automation_halt_subscriber_handle_publishes_correct_payload() { + // The web-channel bus is a process-shared broadcast; under the parallel test + // suite other tests publish to it too, so take THIS subscriber's next + // `automation_halt` event rather than whatever happens to be first in the + // buffer (which flaked as an "event name mismatch"). + fn next_automation_halt( + rx: &mut tokio::sync::broadcast::Receiver, + ) -> WebChannelEvent { + use tokio::sync::broadcast::error::TryRecvError; + loop { + match rx.try_recv() { + Ok(ev) if ev.event == "automation_halt" => return ev, + Ok(_) => {} // foreign event from a concurrent test + Err(TryRecvError::Lagged(_)) => {} // skipped some; keep draining + Err(TryRecvError::Empty) => { + panic!("expected an automation_halt WebChannelEvent but none was published") + } + Err(TryRecvError::Closed) => panic!("web-channel bus closed"), + } + } + } + // Subscribe BEFORE calling handle so the broadcast receiver is created // before any event is sent (broadcast channels only buffer messages // sent AFTER the receiver subscribed). @@ -519,9 +540,7 @@ mod tests { }) .await; - let halted = rx - .try_recv() - .expect("AutomationHalted must publish a WebChannelEvent"); + let halted = next_automation_halt(&mut rx); assert_eq!( halted.event, "automation_halt", "event name mismatch for halted" @@ -553,9 +572,7 @@ mod tests { }) .await; - let resumed = rx - .try_recv() - .expect("AutomationResumed must publish a WebChannelEvent"); + let resumed = next_automation_halt(&mut rx); assert_eq!( resumed.event, "automation_halt", "event name mismatch for resumed" From 3dad123fda2a14c5814dc44b35678f41bf6feea6 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 18:34:14 +0530 Subject: [PATCH 7/8] style: rustfmt the web_chat isolation test comment spacing --- src/openhuman/web_chat/event_bus.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/openhuman/web_chat/event_bus.rs b/src/openhuman/web_chat/event_bus.rs index ff525518ef..5d7ea964dd 100644 --- a/src/openhuman/web_chat/event_bus.rs +++ b/src/openhuman/web_chat/event_bus.rs @@ -516,7 +516,7 @@ mod tests { loop { match rx.try_recv() { Ok(ev) if ev.event == "automation_halt" => return ev, - Ok(_) => {} // foreign event from a concurrent test + Ok(_) => {} // foreign event from a concurrent test Err(TryRecvError::Lagged(_)) => {} // skipped some; keep draining Err(TryRecvError::Empty) => { panic!("expected an automation_halt WebChannelEvent but none was published") From 60f855082723e9c5834605c375e24a0acf486f03 Mon Sep 17 00:00:00 2001 From: shanu Date: Tue, 21 Jul 2026 18:38:07 +0530 Subject: [PATCH 8/8] test(feature-forwarding): honor the not-forwarded allowlist in the shell self-test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 'real shell manifest forwards every real core default' self-test asserted every default-ON gate is forwarded to the desktop shell, ignoring the INTENTIONALLY_NOT_FORWARDED allowlist that the CI checker respects. Once the tui gate (#5084) became default-ON — a terminal subcommand deliberately not shipped in the Tauri shell — the two disagreed and Scripts Self-Tests failed with 'core default gate not forwarded to the shell: tui'. Move the allowlist into scripts/lib/feature-forwarding.mjs (one source of truth) and have the self-test skip allowlisted gates, matching the checker. The lane is path-gated so this was dormant on main; it surfaced here. --- scripts/__tests__/feature-forwarding.test.mjs | 4 ++++ scripts/ci/check-feature-forwarding.mjs | 15 +-------------- scripts/lib/feature-forwarding.mjs | 17 +++++++++++++++++ 3 files changed, 22 insertions(+), 14 deletions(-) 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.