Something went wrong. Try again.
Monorepo for Aesthetic.Computer aesthetic.computer
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476#!/usr/bin/env python3"""Local macOS pressure sampling and admission. Standard library only."""import datetimeimport fcntlimport jsonimport mathimport osfrom pathlib import Pathimport plistlibimport reimport shleximport shutilimport subprocessimport sysimport time
LABEL = "computer.aesthetic.performance-guard"HERE = Path(__file__).resolve().parentSTATE = Path.home() / ".local/share/slab/performance"FLOOR = 20 * 1024**3INTERVAL = 30MAX_LOG = 5 * 1024**2VERSION = "2026-10-01.2"
def run(args, timeout=2): return subprocess.run(args, capture_output=True, text=True, timeout=timeout).stdout.strip()
def atomic(path, data): """Never replace valid state with an incomplete ENOSPC write.""" temp = path.with_name(path.name + ".next") try: with temp.open("w", encoding="utf-8") as stream: os.chmod(temp, 0o600) stream.write(data) os.replace(temp, path) finally: temp.unlink(missing_ok=True)
def read_json(path): try: value = json.loads(path.read_text()) return value if isinstance(value, dict) else {} except (OSError, ValueError): return {}
def previous_sample(): value = read_json(STATE / "latest.json") for key in ("epoch", "last_alert", "breaches", "swapouts"): number = value.get(key, 0) if not isinstance(number, (int, float)) or not math.isfinite(number) or number < 0: return {} if not isinstance(value.get("reasons", []), list) or not all(isinstance(r, str) for r in value.get("reasons", [])): return {} return value
def existing_parent(path): path = Path(path).absolute() while not path.exists() and path != path.parent: path = path.parent return path
def metrics(destination=None): raw = run(["/usr/sbin/sysctl", "hw.logicalcpu", "kern.memorystatus_level", "vm.loadavg", "vm.swapusage", "kern.boottime"]) values = dict(line.split(": ", 1) for line in raw.splitlines() if ": " in line) memory = values.get("kern.memorystatus_level", "") if not memory.isdigit(): found = re.search(r"free percentage:\s*(\d+)%", run(["/usr/bin/memory_pressure", "-Q"])) memory = found.group(1) if found else "" cores = int(values["hw.logicalcpu"]) free_pct = int(memory) load1 = float(values["vm.loadavg"].strip("{} ").split()[0]) if cores < 1 or not 0 <= free_pct <= 100: raise ValueError("invalid host pressure measurements") disks = [shutil.disk_usage(Path.home()).free] if destination: disks.append(shutil.disk_usage(existing_parent(destination)).free) return {"cores": cores, "free_pct": free_pct, "load1": load1, "disk_free_bytes": min(disks), "disk_floor_bytes": FLOOR, "swapusage": values.get("vm.swapusage", "unknown"), "boot": values.get("kern.boottime", "unknown")}
def reasons_for(m): reasons = [] if m["load1"] > m["cores"] * 1.5: reasons.append("load") if m["free_pct"] < 15: reasons.append("memory") if m["disk_free_bytes"] < FLOOR: reasons.append("disk") return reasons
def admit(operation, destination=None): if os.environ.get("AC_PERFORMANCE_ALLOW_PRESSURE") == "1": print("AC performance guard: explicit pressure override", file=sys.stderr) return 0 try: m = metrics(destination) reasons = reasons_for(m) previous = previous_sample() # Keep the sampler's recent swap/display/session signals; stale files # never deny work indefinitely after a reboot or a disabled sampler. age = time.time() - previous.get("epoch", 0) if 0 <= age <= 90 and previous.get("boot") == m["boot"]: reasons += [r for r in previous.get("reasons", []) if r not in ("disk", "load", "memory") and r not in reasons] if not reasons: return 0 detail = (f"{'+'.join(reasons)} pressure; {m['disk_free_bytes']/1024**3:.1f} GiB free " f"(20 GiB reserve), memory available {m['free_pct']}%, load {m['load1']:.2f}") except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired) as error: detail = f"cannot verify host headroom ({error})" print(f"AC performance guard: deferred {operation}: {detail}. " "Wait for recovery or use a compute host. Exit 75; nothing launched.", file=sys.stderr) return 75
def worktree_destination(args, cwd): """Interpret Git global options, preserving the original argv for execution.""" i = 0 while i < len(args): arg = args[i] if arg in ("-C", "-c", "--git-dir", "--work-tree", "--namespace", "--config-env"): if i + 1 >= len(args): return None if arg == "-C" and args[i + 1]: cwd = os.path.abspath(os.path.join(cwd, args[i + 1])) i += 2 elif arg.startswith("-C") and arg != "-C": cwd = os.path.abspath(os.path.join(cwd, arg[2:])) i += 1 elif arg.startswith("-"): i += 1 else: break if args[i:i + 2] != ["worktree", "add"]: return None if any(arg in ("-h", "--help") for arg in args[i + 2:]): return None i += 2 while i < len(args): arg = args[i] if arg == "--": i += 1 break if arg in ("-b", "-B", "--reason"): i += 2 elif arg.startswith("-"): i += 1 else: break return os.path.abspath(os.path.join(cwd, args[i])) if i < len(args) else None
def git_main(args): real_path = HERE / "real-git" real = real_path.read_text().strip() if real_path.exists() else "/usr/bin/git" destination = worktree_destination(args, os.getcwd()) if destination: status = admit("git worktree add", destination) if status: return status os.execv(real, [real, *args])
def processes(): rows = [] for line in run(["/bin/ps", "-A", "-o", "pid=,ppid=,%cpu=,rss=,comm="]).splitlines(): bits = line.split(None, 4) if len(bits) == 5: try: rows.append({"pid": int(bits[0]), "ppid": int(bits[1]), "cpu": float(bits[2]), "rss_kib": int(bits[3]), "executable": bits[4]}) except ValueError: pass return rows
def safe_git_command(command): """Keep command identity, not config values, messages, URLs, or inline code.""" tokens = shlex.split(command) safe = [] redact_next = False for token in tokens: if redact_next: safe.append("[redacted]") redact_next = False elif token in ("-c", "--config-env", "-m", "--message"): safe.append(token) redact_next = True elif token == "config": safe.extend(["config", "[arguments omitted]"]) break elif "://" in token or "@" in token or re.search(r"(?i)(token|password|secret|authorization|credential|prompt)", token): safe.append("[redacted]") elif token.startswith(("-c", "--config-env=", "--message=", "-m")): safe.append("[redacted]") else: safe.append(token[:256]) return shlex.join(safe)[:2048]
def incident_processes(rows): by_pid = {p["pid"]: p for p in rows} selected = {p["pid"]: dict(p) for p in sorted(rows, key=lambda p: p["rss_kib"], reverse=True)[:8]} for p in sorted(rows, key=lambda p: p["cpu"], reverse=True)[:8]: selected[p["pid"]] = dict(p) git = sorted([p for p in rows if Path(p["executable"]).name == "git"], key=lambda p: p["rss_kib"], reverse=True)[:8] deadline = time.monotonic() + 4 for p in git: item = selected.setdefault(p["pid"], dict(p)) if time.monotonic() >= deadline: break try: # Check start time before and after to avoid attributing a reused PID. start = run(["/bin/ps", "-p", str(p["pid"]), "-o", "lstart="]) command = run(["/bin/ps", "-ww", "-p", str(p["pid"]), "-o", "command="]) cwd = run(["/usr/sbin/lsof", "-a", "-p", str(p["pid"]), "-d", "cwd", "-Fn"], timeout=0.4) if start and start == run(["/bin/ps", "-p", str(p["pid"]), "-o", "lstart="]): item.update(started=start, command=safe_git_command(command), cwd=next((s[1:] for s in cwd.splitlines() if s.startswith("n")), None)) except (OSError, ValueError, subprocess.TimeoutExpired): pass for p in list(selected.values()): chain, parent = [], p["ppid"] while parent in by_pid and len(chain) < 6: ancestor = by_pid[parent] chain.append({k: ancestor[k] for k in ("pid", "ppid", "executable")}) parent = ancestor["ppid"] p["parents"] = chain return list(selected.values())
def notify(message): # argv-based AppleScript avoids interpolating commands into source. script = 'on run argv\ndisplay notification (item 1 of argv) with title "AC performance guard"\nend run' try: run(["/usr/bin/osascript", "-e", script, message], timeout=3) except (OSError, subprocess.TimeoutExpired): pass
def append_log(path, line): if path.exists() and path.stat().st_size > MAX_LOG: with path.open("rb") as stream: stream.seek(-1024**2, os.SEEK_END) tail = stream.read().split(b"\n", 1)[-1].decode(errors="replace") atomic(path, tail) with path.open("a") as stream: os.chmod(path, 0o600) stream.write(line + "\n")
def repair_caddy(rows): repo = os.environ.get("AC_REPO") if not repo: return 0 matches = [] for p in rows: if Path(p["executable"]).name != "caddy": continue try: argv = shlex.split(run(["/bin/ps", "-ww", "-p", str(p["pid"]), "-o", "command="])) cwd = run(["/usr/sbin/lsof", "-a", "-p", str(p["pid"]), "-d", "cwd", "-Fn"]) if argv[1:] == ["run", "--config", "Caddyfile"] and "n" + repo + "/system" in cwd.splitlines(): matches.append(p["pid"]) except (OSError, ValueError, subprocess.TimeoutExpired): pass # SIGTERM only; never escalate against a potentially reused PID. for pid in sorted(matches)[:-1]: try: os.kill(pid, 15) except ProcessLookupError: pass return max(0, len(matches) - 1)
def sample(repair=False): STATE.mkdir(parents=True, exist_ok=True) with (STATE / "sample.lock").open("a") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: return 0 previous = previous_sample() m = metrics() revision_file = HERE / "revision" m.update(version=VERSION, revision=revision_file.read_text().strip() if revision_file.exists() else "source", epoch=time.time(), timestamp=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")) rows = processes() names = [Path(p["executable"]).name for p in rows] m.update(processes=len(rows), codex=names.count("codex"), node=names.count("node"), swift_builds=names.count("swift-build"), caddy=names.count("caddy")) vm = run(["/usr/bin/vm_stat"]) swaps = re.search(r"Swapouts:\s*(\d+)", vm) m["swapouts"] = int(swaps.group(1)) if swaps else 0 elapsed = m["epoch"] - previous.get("epoch", 0) delta = max(0, m["swapouts"] - previous.get("swapouts", m["swapouts"])) if previous.get("boot") != m["boot"] or not 0 < elapsed <= 300: delta = 0 m["swapout_pages_delta"] = delta reasons = reasons_for(m) if delta > 4096 * max(elapsed, 1) / INTERVAL: reasons.append("swap") if m["codex"] > 8: reasons.append("sessions") if m["swift_builds"] > 1: reasons.append("builds") display = sum(p["cpu"] for p in rows if Path(p["executable"]).name in ("Terminal", "WindowServer")) if display > 125: reasons.append("display") m["caddy_repaired"] = repair_caddy(rows) if repair else 0 if m["caddy_repaired"]: reasons.append("caddy") m.update(pressure=int(bool(reasons)), reasons=reasons, reason="+".join(reasons) or "none") recent = 0 < elapsed <= 90 and previous.get("boot") == m["boot"] m["breaches"] = (previous.get("breaches", 0) + 1 if recent else 1) if reasons else 0 m["last_alert"] = previous.get("last_alert", 0) if reasons: # Publish the admission signal before optional incident enrichment. atomic(STATE / "pressure-active", m["reason"] + "\n") critical = m["disk_free_bytes"] < 5 * 1024**3 or m["free_pct"] < 5 if (critical or m["breaches"] >= 3) and m["epoch"] - m["last_alert"] >= 600: notify(f"{m['reason']} pressure: {m['disk_free_bytes']/1024**3:.1f} GiB disk free, memory {m['free_pct']}%.") m["last_alert"] = m["epoch"] else: (STATE / "pressure-active").unlink(missing_ok=True) atomic(STATE / "latest.json", json.dumps(m) + "\n") keys = ("version", "revision", "timestamp", "load1", "cores", "free_pct", "disk_free_bytes", "processes", "codex", "node", "swift_builds", "caddy", "caddy_repaired", "swapout_pages_delta", "pressure", "reason") text = "\n".join(f"{k}={m[k]}" for k in keys) atomic(STATE / "latest.txt", text + "\n") if reasons: append_log(STATE / "performance-guard.log", text.replace("\n", " ")) append_log(STATE / "incidents.jsonl", json.dumps({**m, "process_details": incident_processes(rows)})) return 0
def ensure_shell_path(): """macOS path_helper can reorder zsh PATH after .zshenv has run.""" home = Path.home() shell = os.environ.get("SHELL", "/bin/zsh") expected = [str(home / ".local/bin/git"), str(home / ".local/bin/swift")] if run([shell, "-lc", "command -v git; command -v swift"], timeout=8).splitlines() == expected: return name = Path(shell).name if name == "zsh": profile = home / ".zprofile" block = '\n# AC performance guard PATH\nexport PATH="$HOME/.local/bin:$PATH"\n# End AC performance guard PATH\n' elif name == "bash": profile = next((home / p for p in (".bash_profile", ".bash_login", ".profile") if (home / p).exists()), home / ".profile") block = '\n# AC performance guard PATH\nexport PATH="$HOME/.local/bin:$PATH"\n# End AC performance guard PATH\n' elif name == "fish": profile = home / ".config/fish/conf.d/ac-performance-guard.fish" block = '\n# AC performance guard PATH\nfish_add_path --path --prepend --move "$HOME/.local/bin"\n# End AC performance guard PATH\n' else: raise RuntimeError(f"Add ~/.local/bin before system commands in {shell}") profile = profile.resolve() profile.parent.mkdir(parents=True, exist_ok=True) old = profile.read_text() if profile.exists() else "" atomic(profile, old.replace(block, "").rstrip() + "\n" + block) if run([shell, "-lc", "command -v git; command -v swift"], timeout=8).splitlines() != expected: raise RuntimeError(f"Shell still bypasses guards after updating {profile}")
def install(preserve_git=False): home = Path.home() target = home / ".local/lib/ac-performance-guard" bindir = home / ".local/bin" shims = {"git": "git-guard.sh", "swift": "swift-guard.sh", "ac-performance-guard": "performance-guard.sh"} existing_git = None for name, source in shims.items(): link = bindir / name if link.exists() or link.is_symlink(): resolved = link.resolve() if resolved != target / source and not (link.is_symlink() and resolved.name == source): if name == "git" and preserve_git and link.is_file() and not link.is_symlink(): existing_git = link else: raise RuntimeError(f"Refusing to replace unrelated command: {link}") real_git_file = target / "real-git" real_git = real_git_file.read_text().strip() if real_git_file.exists() else shutil.which("git") if not real_git or Path(real_git).resolve() == (target / "git-guard.sh").resolve(): raise RuntimeError("Cannot resolve underlying Git") for directory in (target, bindir, STATE, home / "Library/LaunchAgents"): directory.mkdir(parents=True, exist_ok=True) if existing_git: backup = bindir / "git.before-ac-guard" if backup.exists(): raise RuntimeError(f"Git wrapper backup already exists: {backup}") shutil.copy2(existing_git, backup) real_git = str(backup) for name in ("performance-guard.sh", "performance_guard.py", "git-guard.sh", "swift-guard.sh", "build-lock.sh"): source = HERE / name if name == "build-lock.sh" and not source.exists(): source = HERE.parent.parent / "slab/bin/build-lock.sh" if source.resolve() != (target / name).resolve(): atomic(target / name, source.read_text()) (target / name).chmod(0o755) atomic(real_git_file, real_git + "\n") if (HERE / "revision").exists() and HERE != target: atomic(target / "revision", (HERE / "revision").read_text()) for name, source in shims.items(): link = bindir / name replacement = bindir / (name + ".guard-next") replacement.unlink(missing_ok=True) replacement.symlink_to(target / source) os.replace(replacement, link) ensure_shell_path() repo = os.environ.get("AC_REPO", str(home / "aesthetic-computer")) atomic(target / "repo-path", repo + "\n") config = {"Label": LABEL, "ProgramArguments": ["/bin/bash", str(target / "performance-guard.sh"), "--once", "--repair"], "RunAtLoad": True, "StartInterval": INTERVAL, "ProcessType": "Background", "LowPriorityIO": True, "EnvironmentVariables": {"AC_REPO": repo, "PATH": f"{bindir}:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"}, "StandardOutPath": str(STATE / "launchd.out"), "StandardErrorPath": str(STATE / "launchd.err")} plist = home / "Library/LaunchAgents" / (LABEL + ".plist") atomic(plist, plistlib.dumps(config).decode()) subprocess.run(["launchctl", "bootout", f"gui/{os.getuid()}/{LABEL}"], capture_output=True) subprocess.run(["launchctl", "bootstrap", f"gui/{os.getuid()}", str(plist)], check=True) print(f"Installed {LABEL}, Git and Swift guards. Put {bindir} first on PATH.")
def main(args): action = args[0] if args else "--once" if action == "--git": return git_main(args[1:]) if action == "--admit": return admit(args[1] if len(args) > 1 else "work", args[2] if len(args) > 2 else None) if action == "--install": install("--preserve-git-wrapper" in args) elif action == "--status": print((STATE / "latest.txt").read_text() if (STATE / "latest.txt").exists() else "no performance sample yet") elif action in ("--once", "--watch"): while True: try: sample("--repair" in args) except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired) as error: print(f"AC performance guard sample failed: {error}", file=sys.stderr) try: atomic(STATE / "pressure-active", "measurement-unavailable\n") except OSError: pass if action == "--once": return 1 if action == "--once": break time.sleep(INTERVAL) elif action == "--uninstall": subprocess.run(["launchctl", "bootout", f"gui/{os.getuid()}/{LABEL}"], capture_output=True) (Path.home() / "Library/LaunchAgents" / (LABEL + ".plist")).unlink(missing_ok=True) for name in ("git", "swift", "ac-performance-guard"): link = Path.home() / ".local/bin" / name if link.is_symlink() and link.resolve().parent == HERE: link.unlink() backup = link.with_name("git.before-ac-guard") if name == "git" and backup.exists(): os.replace(backup, link) (STATE / "pressure-active").unlink(missing_ok=True) else: print("usage: performance-guard.sh [--once [--repair] | --watch | --status | --install | --uninstall | --admit OPERATION [PATH]]") return 0 if action in ("-h", "--help") else 2 return 0
if __name__ == "__main__": sys.exit(main(sys.argv[1:]))