tools/guard_check.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""The resource guard: does it warn at the right time, and never block?
The guard is a soft read on whether the box is stretched for the number of
live sessions. These assertions feed ``sysinfo.guard`` synthetic mem / swap /
load so every signal — the RAM floor, swap climbing, load past the cores, and
the session count past a derived soft ceiling — is exercised without needing an
actually-overloaded box. The ceiling is measured from real per-session memory,
so that is checked too.
Then, if the Playwright venv is present, it renders the banner in a real page
and screenshots it — a green DOM check has passed on a visually-broken layout
here before, so the guard's one net-new strip gets a real eyeball.
python3 tools/guard_check.py # logic only
~/.cache/clique-visual/bin/python tools/guard_check.py # + screenshot
Exit status is 0 on pass, 1 on fail.
"""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from clique import sysinfo
def _mem(total_mb: int, used_mb: int) -> dict:
pct = round(100.0 * used_mb / total_mb, 1) if total_mb else 0.0
return {"total_mb": total_mb, "used_mb": used_mb, "percent": pct}
def _load(one: float, cores: int) -> dict:
return {
"one": one,
"five": one,
"fifteen": one,
"cores": cores,
"ratio": round(one / cores, 2) if cores else 0.0,
}
def _swap(used_mb: int) -> dict:
return {"used_mb": used_mb, "total_mb": 4096, "percent": 0.0}
def _calm_swap() -> None:
"""Force the swap signal quiet, so a test isolates the signal it means to.
``_swap_growing`` re-baselines against wall-clock; the tests that are not
about swap monkeypatch it off, and the one that is, forces it on.
"""
sysinfo._swap_growing = lambda used: False # type: ignore[assignment]
def logic() -> dict:
res: dict[str, object] = {}
_calm_swap()
# A roomy box with a few sessions says nothing at all.
big = sysinfo.guard(
3,
[600 * 1024] * 3,
mem=_mem(32768, 8000),
swap_info=_swap(0),
load_info=_load(1.0, 8),
)
res["calm_box_ok"] = big["level"] == "ok" and big["headline"] == ""
res["calm_ceiling_roomy"] = big["ceiling"] > 20
# Ceiling is measured from real session memory, not a fixed guess.
res["avg_measured"] = (
850
<= sysinfo.guard(
2,
[800 * 1024, 1000 * 1024],
mem=_mem(32768, 8000),
swap_info=_swap(0),
load_info=_load(0.5, 8),
)["avg_session_mb"]
<= 950
)
res["avg_default_when_empty"] = (
sysinfo.guard(
0,
[],
mem=_mem(32768, 8000),
swap_info=_swap(0),
load_info=_load(0.1, 8),
)["avg_session_mb"]
== sysinfo._DEFAULT_SESSION_MB
)
# Session count past the soft ceiling → a strong warning.
tight = sysinfo.guard(
8,
[600 * 1024] * 8,
mem=_mem(4096, 1500),
swap_info=_swap(0),
load_info=_load(1.0, 4),
)
res["over_ceiling_high"] = tight["level"] == "high"
res["over_ceiling_reason"] = any("past a comfortable" in r for r in tight["reasons"])
res["over_ceiling_headline"] = bool(tight["headline"])
# Sitting right at the ceiling → a soft heads-up, not the strong one.
at = sysinfo.guard(
tight["ceiling"],
[600 * 1024] * tight["ceiling"],
mem=_mem(4096, 1200),
swap_info=_swap(0),
load_info=_load(1.0, 4),
)
res["at_ceiling_watch"] = at["level"] == "watch"
# Available RAM under the floor → high, whatever the session count.
low = sysinfo.guard(
2,
[600 * 1024] * 2,
mem=_mem(8192, 7800),
swap_info=_swap(0),
load_info=_load(0.5, 8),
)
res["ram_floor_high"] = low["level"] == "high"
res["ram_floor_reason"] = any("RAM free" in r for r in low["reasons"])
# Load sustained over the cores → high at 2x, watch at 1x.
res["load_2x_high"] = (
sysinfo.guard(
1,
[600 * 1024],
mem=_mem(32768, 8000),
swap_info=_swap(0),
load_info=_load(9.0, 4),
)["level"]
== "high"
)
res["load_1x_watch"] = (
sysinfo.guard(
1,
[600 * 1024],
mem=_mem(32768, 8000),
swap_info=_swap(0),
load_info=_load(5.0, 4),
)["level"]
== "watch"
)
# Swap actively climbing → high, even with RAM and load calm.
sysinfo._swap_growing = lambda used: True # type: ignore[assignment]
swp = sysinfo.guard(
1,
[600 * 1024],
mem=_mem(32768, 8000),
swap_info=_swap(300),
load_info=_load(0.5, 8),
)
res["swap_growing_high"] = swp["level"] == "high"
res["swap_growing_reason"] = any("swap is climbing" in r for r in swp["reasons"])
_calm_swap()
# The whole contract: it advises, it never blocks. No allow/deny key, and
# the ceiling is always at least one so the form never reads "runs ~0".
tiny = sysinfo.guard(
1,
[],
mem=_mem(512, 400),
swap_info=_swap(0),
load_info=_load(0.1, 1),
)
res["ceiling_never_zero"] = tiny["ceiling"] >= 1
res["no_block_key"] = not ({"block", "deny", "allowed", "blocked"} & set(big))
res["levels_are_advisory"] = all(
g["level"] in ("ok", "watch", "high") for g in (big, tight, at, low, swp, tiny)
)
# The reap window makes it into the sentence, so the lever is named.
reaped = sysinfo.guard(
8,
[600 * 1024] * 8,
mem=_mem(4096, 1500),
swap_info=_swap(0),
load_info=_load(1.0, 4),
reap_hours=6,
)
res["headline_names_reap"] = "auto-reap in 6h" in reaped["headline"]
return res
def visual() -> dict:
"""Render the banner in a real page and screenshot it. Skipped without the
Playwright venv; the logic above is what always runs."""
try:
from playwright.sync_api import sync_playwright
except ModuleNotFoundError:
return {}
import tools.visual_check as vc
from clique.auth import COOKIE_NAME, Auth
res: dict[str, object] = {}
panel = vc._own_panel()
try:
auth = Auth(vc.PASSWORD, vc.SANDBOX / "secret")
shot = Path("/tmp/clique-visual") # where the other visual checks land
shot.mkdir(parents=True, exist_ok=True)
with sync_playwright() as play:
browser = play.chromium.launch()
ctx = browser.new_context(viewport={"width": 1200, "height": 760})
ctx.add_init_script("try { localStorage.setItem('clique.gpu','0'); } catch (e) {}")
ctx.add_cookies(
[{"name": COOKIE_NAME, "value": auth.issue(), "domain": "127.0.0.1", "path": "/"}]
)
page = ctx.new_page()
errs: list[str] = []
page.on("pageerror", lambda e: errs.append(str(e)))
page.goto(vc.BASE, wait_until="networkidle")
# A calm box draws nothing.
page.evaluate(
"() => { state.stats = state.stats || {}; "
"state.stats.guard = {level:'ok', headline:''}; renderGuard(); }"
)
res["hidden_when_ok"] = page.evaluate("() => document.getElementById('guard').hidden")
# A stretched box draws the strip, on screen, with the sentence.
page.evaluate(
"() => { state.stats.guard = {level:'high', sessions:12, ceiling:8, "
"free_mb:1228, reasons:['12 sessions, past a comfortable ~8 for this box'], "
"headline:'12 sessions, ~1.2 GB free — heavy for this box. "
"Idle ones auto-reap in 6h.'}; renderGuard(); }"
)
g = page.locator("#guard")
res["shown_when_high"] = not page.evaluate(
"() => document.getElementById('guard').hidden"
)
box = g.bounding_box() or {}
res["on_screen"] = bool(box) and box.get("y", -1) >= 0 and box.get("width", 0) > 200
res["carries_headline"] = "heavy for this box" in (g.inner_text() or "")
res["level_styled"] = page.evaluate(
"() => document.getElementById('guard').dataset.level === 'high'"
)
page.screenshot(path=str(shot / "guard.png"))
# Dismiss hides it; the same situation stays gone.
page.evaluate("() => document.querySelector('#guard .guard-x').click()")
res["dismiss_hides"] = page.evaluate(
"() => { renderGuard(); return document.getElementById('guard').hidden; }"
)
res["no_console_errors"] = not errs
if errs:
res["errors"] = errs[:4]
finally:
panel.terminate()
return res
def main() -> int:
res = logic()
res.update(visual())
ok = all(v for k, v in res.items() if k != "errors")
for key, value in res.items():
print(f" {'ok ' if value or key == 'errors' else 'FAIL'} {key}: {value}")
print("guard_check:", "PASS" if ok else "FAIL")
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())