from __future__ import annotations
import argparse
import json
import os
import platform
import signal
import statistics
import subprocess
import sys
import tempfile
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_BINARY = ROOT / "target" / "debug" / ("bot-forge.exe" if os.name == "nt" else "bot-forge")
def summary(samples: list[float]) -> dict[str, float | int]:
ordered = sorted(samples)
index = min(len(ordered) - 1, max(0, int((len(ordered) - 1) * 0.95)))
return {
"samples": len(samples),
"min_ms": round(min(samples), 3),
"median_ms": round(statistics.median(samples), 3),
"p95_ms": round(ordered[index], 3),
"max_ms": round(max(samples), 3),
}
def run_timed(command: list[str], env: dict[str, str], cwd: Path) -> tuple[float, bytes]:
started = time.perf_counter_ns()
result = subprocess.run(command, cwd=cwd, env=env, stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
elapsed = (time.perf_counter_ns() - started) / 1_000_000
if result.returncode:
raise RuntimeError(f"command failed: {command}\n{result.stderr.decode(errors='replace')}")
return elapsed, result.stdout
def write_config(path: Path, count: int) -> None:
names = ", ".join(json.dumps(f"tool-{index}") for index in range(count))
lines = ["catalog = \"rust-dev\"", "[profiles.smoke]", f"components = [{names}]",
"[profiles.standard]"]
for index in range(count):
lines.extend(["[[components]]", f'id = "tool-{index}"', 'platforms = ["*"]',
"[components.detect]", 'kind = "command"', 'program = "true"',
"[components.install]", 'backend = "apt"', f'packages = ["tool-{index}"]'])
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def benchmark_plan(binary: Path, env: dict[str, str], cwd: Path, count: int, runs: int) -> dict[str, object]:
config = cwd / f"plan-{count}.toml"
write_config(config, count)
command = [
str(binary),
"plan",
"smoke",
"--format",
"json",
"--config",
str(config),
]
samples = []
outcome = {}
for _ in range(runs):
elapsed, output = run_timed(command, env, cwd)
document = json.loads(output)
samples.append(elapsed)
outcome = {"components": len(document["components"]), "nodes": len(document["nodes"])}
return {**summary(samples), "last_outcome": outcome}
def process_tree_rss_kib(root_pid: int) -> int:
result = subprocess.run(["ps", "-axo", "pid=,ppid=,rss="], capture_output=True, text=True)
rows = [tuple(map(int, line.split())) for line in result.stdout.splitlines() if len(line.split()) == 3]
descendants = {root_pid}
changed = True
while changed:
changed = False
for pid, parent, _ in rows:
if parent in descendants and pid not in descendants:
descendants.add(pid)
changed = True
return sum(rss for pid, _, rss in rows if pid in descendants)
def shell_probe_config(path: Path, component: str, detect: str, command: str) -> None:
path.write_text(
"catalog = \"rust-dev\"\n[policy]\nallow_shell = true\n[profiles.smoke]\n"
f'components = ["{component}"]\n[profiles.standard]\n'
f'[[components]]\nid = "{component}"\nplatforms = ["*"]\n'
f'[components.detect]\nkind = "shell"\ncommand = {json.dumps(detect)}\n'
f'[components.install]\nbackend = "shell"\ncommand = {json.dumps(command)}\n'
'resources = ["benchmark"]\ntimeout_secs = 60\n', encoding="utf-8")
def high_output(binary: Path, env: dict[str, str], cwd: Path) -> dict[str, object]:
if os.name == "nt":
return {"supported": False, "reason": "RSS sampling is POSIX-only"}
config = cwd / "output.toml"
command = f'{sys.executable} -c "import sys;sys.stdout.buffer.write(b\'x\'*(16*1024*1024))" && touch output.done'
shell_probe_config(config, "output", "test -f output.done", command)
process = subprocess.Popen([str(binary), "install", "smoke", "--yes", "--config", str(config)],
cwd=cwd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
start_new_session=True)
peak = 0
while process.poll() is None:
peak = max(peak, process_tree_rss_kib(process.pid))
time.sleep(0.01)
if process.returncode:
raise RuntimeError("high-output probe failed")
return {"supported": True, "bytes_written": 16 * 1024 * 1024,
"process_tree_peak_rss_kib": peak}
def cancellation(binary: Path, env: dict[str, str], cwd: Path) -> dict[str, object]:
if os.name == "nt":
return {"supported": False, "reason": "signal probe is POSIX-only"}
config = cwd / "cancel.toml"
shell_probe_config(config, "slow", "false", "sleep 30")
process = subprocess.Popen([str(binary), "install", "smoke", "--yes", "--config", str(config)],
cwd=cwd, env=env, stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(0.25)
started = time.perf_counter_ns()
process.send_signal(signal.SIGINT)
process.wait(timeout=5)
return {"supported": True, "latency_ms": round((time.perf_counter_ns() - started) / 1_000_000, 3),
"return_code": process.returncode}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--binary", type=Path, default=DEFAULT_BINARY)
parser.add_argument("--output", type=Path)
parser.add_argument("--runs", type=int, default=5)
args = parser.parse_args()
if args.runs < 1:
parser.error("--runs must be positive")
binary = args.binary.resolve()
with tempfile.TemporaryDirectory(prefix="bot-forge-release-benchmark-") as directory:
cwd = Path(directory)
user_home = cwd / "user-home"
user_home.mkdir()
env = os.environ.copy()
env.update({"BOT_FORGE_HOME": str(cwd / "forge-home"), "HOME": str(user_home),
"USERPROFILE": str(user_home), "NO_COLOR": "1"})
report = {
"kind": "release-gate",
"binary": str(binary),
"platform": {"system": platform.system(), "machine": platform.machine()},
"runs": args.runs,
"startup": summary([run_timed([str(binary), "help"], env, cwd)[0] for _ in range(args.runs)]),
"plan_500_components": benchmark_plan(binary, env, cwd, 500, args.runs),
"standard_plan_100": benchmark_plan(binary, env, cwd, 100, min(args.runs, 3)),
"high_output_command": high_output(binary, env, cwd),
"cancellation": cancellation(binary, env, cwd),
}
encoded = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
if args.output:
args.output.write_text(encoded, encoding="utf-8")
print(encoded, end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())