From bf735e235bed01c79f6ea58c1eb12a35108697f7 Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Thu, 30 Jul 2026 04:23:28 +1000 Subject: [PATCH] test(mesh): prove agent replies reach a live relay (P5 fixes + P6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P5 could never pass: install_fake_shell returned the bin directory, but resolve_bash requires BUZZ_SHELL to name a file, so it silently fell back to real bash and the marker was never written. The fake shell also matched `-lc` while shell_flag() emits `-c` on Unix. Return the executable, accept both flags, and append rather than overwrite so a probing model does not erase the recorded send. P5 stops at the shell boundary, so it cannot show a reply ever lands. Add P6: the agent runs the genuine `buzz messages send` (dev-mcp's multicall shim, real buzz-cli) against a running relay, and the harness then queries the relay itself — the agent's prose is not evidence. Infra-gated on BUZZ_E2E_RELAY_URL / _PRIVATE_KEY / _CHANNEL; skipped when unset so P1-P5 still run without Postgres and Redis. Also drive the agent with the real base_prompt.md instead of a terse hand-written string. A short prompt is not a proxy for production: tool adherence in small models depends on it, so testing with 56 characters where Buzz injects ~3.2k tokens measures the harness, not Buzz. Signed-off-by: Michael Neale --- crates/buzz-relay/examples/mesh_agent_e2e.rs | 295 +++++++++++++++++-- 1 file changed, 276 insertions(+), 19 deletions(-) diff --git a/crates/buzz-relay/examples/mesh_agent_e2e.rs b/crates/buzz-relay/examples/mesh_agent_e2e.rs index b6f723f35a..53c9042082 100644 --- a/crates/buzz-relay/examples/mesh_agent_e2e.rs +++ b/crates/buzz-relay/examples/mesh_agent_e2e.rs @@ -8,17 +8,31 @@ //! P2 auto-model chat — agent sends `model: "auto"`; mesh router picks. //! P3 context-fit regression — an oversized output budget (150k tokens) //! must FAIL with the router's context error (proves the router's fit -//! gate — the failure mode the 1024 preset cap protects against). +//! gate — the failure mode the 4096 preset cap protects against). //! P4 agentic tool use — agent + buzz-dev-mcp writes a file on disk. +//! P5 Buzz reply command — agent asks the real developer MCP to run the +//! exact `buzz messages send` command used to publish a reply. A fake +//! shell records the command at the execution boundary, so this proves +//! the model emits the right command but NOT that a reply lands. +//! P6 live relay send — the genuine `buzz messages send` (dev-mcp's +//! multicall shim, real buzz-cli) against a running relay. The harness +//! then queries the relay itself: the agent's prose is not evidence. +//! Skipped unless the live-relay env vars below are set. //! //! The serve node is the same `mesh_llm_sdk::serve` path Share-compute uses //! (publish off, mdns, loopback). The agent legs spawn the real //! `buzz-agent` binary with the exact env vars the relay-mesh preset ships. //! //! Hardware-gated, not CI. Run: -//! cargo build --release -p buzz-agent -p buzz-dev-mcp +//! cargo build --release -p buzz-agent -p buzz-dev-mcp -p buzz-cli //! cargo run -p buzz-relay --example mesh_agent_e2e //! Env: MESH_E2E_MODEL overrides the served model ref. +//! +//! P6 additionally needs a relay (`just relay` — Postgres + Redis) plus: +//! BUZZ_E2E_RELAY_URL=ws://localhost:3000 +//! BUZZ_E2E_PRIVATE_KEY= +//! BUZZ_E2E_CHANNEL= +//! All three must be set or P6 is skipped (P1-P5 still run). use std::process::Stdio; use std::time::Duration; @@ -26,14 +40,43 @@ use mesh_llm_sdk::{serve, MeshDiscoveryMode}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::process::{Child, Command}; -// Qwen3-8B: cached GGUF *and* complete layer package on this class of -// machine, so the serve node starts in seconds. Qwen3-30B-A3B works too but -// mesh-llm serves it from layer packages and will download them on first -// serve (~7GB) — fine in the app (progress UI), too slow for a smoke. -const DEFAULT_MODEL: &str = "unsloth/Qwen3-8B-GGUF:Q4_K_M"; +// Non-reasoning Gemma 4 is the small Buzz-curated default. It reaches tool +// calls promptly instead of spending the output budget in a hidden reasoning +// loop. MESH_E2E_MODEL can still exercise any other served model explicitly. +const DEFAULT_MODEL: &str = "unsloth/gemma-4-E4B-it-GGUF:Q4_K_M"; const API_PORT: u16 = 19437; const CONSOLE_PORT: u16 = 13231; +#[derive(Clone)] +struct McpServerEnv { + name: String, + value: String, +} + +#[derive(Clone)] +struct McpServerSpec { + name: String, + command: String, + env: Vec, +} + +impl McpServerSpec { + fn new(name: impl Into, command: impl Into) -> Self { + Self { + name: name.into(), + command: command.into(), + env: Vec::new(), + } + } + + fn set_env(&mut self, name: impl Into, value: impl Into) { + self.env.push(McpServerEnv { + name: name.into(), + value: value.into(), + }); + } +} + fn main() -> anyhow::Result<()> { tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -167,7 +210,7 @@ async fn run() -> anyhow::Result<()> { let prompt = format!( "Use your developer tools to create {marker_name} in the current working directory containing exactly the text BUZZ_OK (no quotes, no newline commentary). Then confirm." ); - let mcp = vec![("dev".to_string(), repo_bin("buzz-dev-mcp")?)]; + let mcp = vec![McpServerSpec::new("dev", repo_bin("buzz-dev-mcp")?)]; let (r, marker) = agent_chat_with_marker(&base, &served_id, None, &prompt, &mcp, &marker_name).await; let file_ok = std::fs::read_to_string(&marker) @@ -187,6 +230,83 @@ async fn run() -> anyhow::Result<()> { } let _ = std::fs::remove_file(&marker); + // P5: prove the model follows Buzz's real reply path. buzz-dev-mcp always + // prepends its own multicall `buzz` shim to PATH, so a fake `buzz` later on + // PATH cannot isolate this test. Instead, inject a fake shell into the real + // MCP server. It records the exact command at the execution boundary and + // returns the same accepted-write shape as buzz-cli, without contacting a + // community. + let send_marker = format!("mesh-e2e-buzz-send-{}.txt", std::process::id()); + let prompt = "Use the developer shell tool to run exactly this command: buzz messages send --channel test-channel --content BUZZ_REPLY_OK. Do not simulate or substitute another command. After it succeeds, confirm briefly."; + let (r, marker) = + agent_chat_with_fake_shell(&base, &served_id, prompt, &mcp, &send_marker).await; + let recorded = std::fs::read_to_string(&marker).unwrap_or_default(); + // The fake shell appends every command it is handed. Small models often + // probe with an unrelated command first, so the pass condition is that the + // exact Buzz send reached the execution boundary on some line — not that it + // was the only command attempted. + const EXPECTED_SEND: &str = "buzz messages send --channel test-channel --content BUZZ_REPLY_OK"; + let attempts: Vec<&str> = recorded + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .collect(); + let send_ok = attempts.contains(&EXPECTED_SEND); + match r { + Ok(text) => record( + "P5 Buzz reply command", + send_ok, + format!("commands attempted {attempts:?}; agent said: {text}"), + ), + Err(e) => record( + "P5 Buzz reply command", + send_ok, + format!("agent error: {e}; commands attempted: {attempts:?}"), + ), + } + let _ = std::fs::remove_file(&marker); + + // P6: live relay end-to-end. P5 stops at the shell boundary with a fake + // shell, so it cannot prove a reply ever reaches a channel. Here the agent + // runs the genuine `buzz messages send` (dev-mcp's multicall shim, real + // buzz-cli code) against a running relay, and the harness independently + // queries the relay afterwards rather than trusting the agent's prose. + // + // Infra-gated: needs `just relay` (Postgres + Redis) and an identity: + // BUZZ_E2E_RELAY_URL, BUZZ_E2E_PRIVATE_KEY, BUZZ_E2E_CHANNEL + match ( + std::env::var("BUZZ_E2E_RELAY_URL").ok(), + std::env::var("BUZZ_E2E_PRIVATE_KEY").ok(), + std::env::var("BUZZ_E2E_CHANNEL").ok(), + ) { + (Some(relay_url), Some(private_key), Some(channel)) => { + let live_marker = format!("BUZZ_LIVE_OK_{}", std::process::id()); + let mut dev = McpServerSpec::new("dev", repo_bin("buzz-dev-mcp")?); + dev.set_env("BUZZ_RELAY_URL", relay_url.as_str()); + dev.set_env("BUZZ_PRIVATE_KEY", private_key.as_str()); + let prompt = format!( + "You are a Buzz agent with shell access. The `buzz` CLI is on your PATH; `buzz messages send` publishes to a channel. Post a message containing exactly the token {live_marker} to channel {channel}." + ); + let r = agent_chat(&base, &served_id, None, &prompt, &[dev]).await; + let landed = relay_has_message(&relay_url, &private_key, &channel, &live_marker).await; + let landed_ok = matches!(landed, Ok(true)); + let detail = match (&r, &landed) { + (Ok(text), Ok(true)) => { + format!("relay returned {live_marker}; agent said: {text}") + } + (Ok(text), Ok(false)) => { + format!("{live_marker} never reached the relay; agent said: {text}") + } + (Ok(text), Err(e)) => format!("relay verification failed: {e}; agent said: {text}"), + (Err(e), _) => format!("agent error: {e}"), + }; + record("P6 live relay send", landed_ok, detail); + } + _ => eprintln!( + "[e2e] SKIP P6 live relay send (set BUZZ_E2E_RELAY_URL, BUZZ_E2E_PRIVATE_KEY, BUZZ_E2E_CHANNEL)" + ), + } + eprintln!("[e2e] {pass} passed, {fail} failed"); if fail > 0 { eprintln!("[e2e] FAIL: {fail} permutation(s) failed"); @@ -220,6 +340,45 @@ fn repo_bin(name: &str) -> anyhow::Result { Ok(path.to_string_lossy().into_owned()) } +/// Ask the relay directly whether `marker` is present in `channel`. +/// +/// This is the independent check that makes P6 meaningful: the agent's prose is +/// not evidence, and neither is the shell exit code. The harness performs its +/// own authenticated read against the relay in a separate process and only +/// trusts what comes back. Polls because the relay persists and fans out +/// asynchronously, so an immediate read can race the write. +async fn relay_has_message( + relay_url: &str, + private_key: &str, + channel: &str, + marker: &str, +) -> anyhow::Result { + let buzz = repo_bin("buzz")?; + for _ in 0..15 { + let output = Command::new(&buzz) + .args([ + "--format", + "compact", + "messages", + "get", + "--channel", + channel, + "--limit", + "50", + ]) + .env("PATH", std::env::var("PATH").unwrap_or_default()) + .env("BUZZ_RELAY_URL", relay_url) + .env("BUZZ_PRIVATE_KEY", private_key) + .output() + .await?; + if output.status.success() && String::from_utf8_lossy(&output.stdout).contains(marker) { + return Ok(true); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + Ok(false) +} + /// Spawn the real buzz-agent with relay-mesh preset env and drive one ACP /// session/prompt over stdio. Returns the concatenated agent message text, /// or Err carrying the agent's error message. @@ -228,10 +387,11 @@ async fn agent_chat( model: &str, max_output_tokens: Option<&str>, prompt: &str, - mcp_servers: &[(String, String)], + mcp_servers: &[McpServerSpec], ) -> anyhow::Result { let (result, _) = - agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers).await; + agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers, None) + .await; result } @@ -240,11 +400,25 @@ async fn agent_chat_with_marker( model: &str, max_output_tokens: Option<&str>, prompt: &str, - mcp_servers: &[(String, String)], + mcp_servers: &[McpServerSpec], marker_name: &str, ) -> (anyhow::Result, std::path::PathBuf) { let (result, home) = - agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers).await; + agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers, None) + .await; + (result, home.join(marker_name)) +} + +async fn agent_chat_with_fake_shell( + base: &str, + model: &str, + prompt: &str, + mcp_servers: &[McpServerSpec], + marker_name: &str, +) -> (anyhow::Result, std::path::PathBuf) { + let (result, home) = + agent_chat_in_isolated_home(base, model, None, prompt, mcp_servers, Some(marker_name)) + .await; (result, home.join(marker_name)) } @@ -253,7 +427,8 @@ async fn agent_chat_in_isolated_home( model: &str, max_output_tokens: Option<&str>, prompt: &str, - mcp_servers: &[(String, String)], + mcp_servers: &[McpServerSpec], + fake_shell_marker_name: Option<&str>, ) -> (anyhow::Result, std::path::PathBuf) { let agent = match repo_bin("buzz-agent") { Ok(agent) => agent, @@ -265,10 +440,28 @@ async fn agent_chat_in_isolated_home( return (Err(error.into()), home); } + let child_path = std::env::var("PATH").unwrap_or_default(); + let mut effective_mcp_servers = mcp_servers.to_vec(); + if let Some(marker_name) = fake_shell_marker_name { + let fake_shell = match install_fake_shell(&home) { + Ok(fake_shell) => fake_shell, + Err(error) => return (Err(error), home), + }; + let marker = home.join(marker_name); + let Some(dev_server) = effective_mcp_servers + .iter_mut() + .find(|server| server.name == "dev") + else { + return (Err(anyhow::anyhow!("P5 requires the dev MCP server")), home); + }; + dev_server.set_env("BUZZ_SHELL", fake_shell.to_string_lossy()); + dev_server.set_env("BUZZ_E2E_SEND_MARKER", marker.to_string_lossy()); + } + let mut command = Command::new(&agent); command .env_clear() - .env("PATH", std::env::var("PATH").unwrap_or_default()) + .env("PATH", child_path) .env("HOME", &home) // Exactly the environment apply_relay_mesh_env() supplies. .env("BUZZ_AGENT_PROVIDER", "openai") @@ -293,15 +486,69 @@ async fn agent_chat_in_isolated_home( Err(error) => return (Err(error.into()), home), }; - let result = drive_acp(&mut child, prompt, mcp_servers, &home).await; + let result = drive_acp(&mut child, prompt, &effective_mcp_servers, &home).await; let _ = child.kill().await; (result, home) } +#[cfg(unix)] +fn install_fake_shell(home: &std::path::Path) -> anyhow::Result { + use std::os::unix::fs::PermissionsExt; + + let bin_dir = home.join("bin"); + std::fs::create_dir_all(&bin_dir)?; + let executable = bin_dir.join("mesh-e2e-shell"); + // buzz-dev-mcp spawns ` -c ""` on Unix (`shell_flag()` + // returns `-c` for every non-cmd/pwsh shell). Accept `-lc` too so a future + // login-shell dialect change does not silently stop recording. + // Every command is appended, one per line: a small model often probes with + // an unrelated command first, and overwriting would discard the real send. + std::fs::write( + &executable, + r#"#!/bin/sh +case "$1" in + -c|-lc) command_string="$2" ;; + *) + printf 'unsupported fake shell invocation: %s\n' "$*" >&2 + exit 2 + ;; +esac +printf '%s\n' "$command_string" >> "$BUZZ_E2E_SEND_MARKER" +if [ "$command_string" = "buzz messages send --channel test-channel --content BUZZ_REPLY_OK" ]; then + printf '%s\n' '{"event_id":"mesh-e2e-event","accepted":true,"message":"ok"}' + exit 0 +fi +printf 'unsupported fake shell command: %s\n' "$command_string" >&2 +exit 2 +"#, + )?; + let mut permissions = std::fs::metadata(&executable)?.permissions(); + permissions.set_mode(0o755); + std::fs::set_permissions(&executable, permissions)?; + // BUZZ_SHELL must name the executable itself. Returning the directory made + // `resolve_bash()` reject it as "not a file" and fall back to real bash, so + // the fake shell never ran and P5 could never record a command. + Ok(executable) +} + +#[cfg(not(unix))] +fn install_fake_shell(_home: &std::path::Path) -> anyhow::Result { + anyhow::bail!("the mesh agent hardware harness currently requires Unix") +} + +/// The real Buzz system prompt every managed agent receives, straight from the +/// harness crate rather than a paraphrase. +/// +/// A terse hand-written test prompt is not a proxy for this: it is ~3.2k tokens +/// of instructions, and the reply-delivery rules the model must follow to get a +/// message into a channel live in here. Testing with a short prompt measures a +/// toy, not Buzz. +const BUZZ_BASE_PROMPT: &str = include_str!("../../buzz-acp/src/base_prompt.md"); + async fn drive_acp( child: &mut Child, prompt: &str, - mcp_servers: &[(String, String)], + mcp_servers: &[McpServerSpec], cwd: &std::path::Path, ) -> anyhow::Result { let mut stdin = child @@ -316,8 +563,18 @@ async fn drive_acp( let mcp_json: Vec = mcp_servers .iter() - .map(|(name, command)| { - serde_json::json!({ "name": name, "command": command, "args": [], "env": [] }) + .map(|server| { + let env: Vec = server + .env + .iter() + .map(|entry| serde_json::json!({ "name": &entry.name, "value": &entry.value })) + .collect(); + serde_json::json!({ + "name": &server.name, + "command": &server.command, + "args": [], + "env": env, + }) }) .collect(); @@ -338,7 +595,7 @@ async fn drive_acp( "params": { "cwd": cwd.to_string_lossy(), "mcpServers": mcp_json, - "systemPrompt": "You are a terse test agent. Follow instructions exactly." + "systemPrompt": BUZZ_BASE_PROMPT } })) .as_bytes(),