diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index d1d6ddc..b3afb04 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -32,8 +32,19 @@ Args: visible), reason and check, then a preview of what the command changes, and asks [y]es / [s]kip / [d]iff / [q]uit; d shows the full diff. y runs the command in its directory, then sets the row done - when its check passes (manual: when the command exits 0). End of - input quits. There is no flag that answers for the operator. + when its check passes (manual: when the command exits 0). A command + the kvb-org-organize watch config calls heavy also shows the memory + stage and `queue.py list`; once the config holds the queue keys and + slot_need_gb.operator, and the command runs no sudo, [h]old runs + `granted` through `queue.py hold --kind operator --task + --session Operator`. End of input quits. There is no flag that + answers for the operator. + granted --dir --check -- $SHELL -ic : what + [h]old queues. On the grant it shows the row and its preview again + and asks [y]es / [s]kip / [d]iff / [q]uit, then runs the command in + the row's directory; a row no longer open with exactly that + directory, check and command runs nothing, nor does a question left + unanswered for GRANT_SECONDS. done : mark a row done, whatever its check. Output: @@ -45,7 +56,8 @@ Output: prints the same rows as a JSON list. Exit codes: - 0 done, 2 usage error. + 0 done, 2 usage error. granted: the command's exit code, or 75 + (EX_TEMPFAIL, which queue.py hold does not learn from) when nothing ran. """ from __future__ import annotations @@ -62,13 +74,25 @@ import hashlib import importlib.util import json import re +import select import shlex import shutil import subprocess +import termios +import time +from collections.abc import Callable from pathlib import Path from types import ModuleType -RASCI = Path(__file__).resolve().parents[2] / "kvb-org-rasci" / "scripts" / "rasci.py" +SKILLS = Path(__file__).resolve().parents[2] +RASCI = SKILLS / "kvb-org-rasci" / "scripts" / "rasci.py" +WATCH = SKILLS / "kvb-org-organize" / "scripts" / "watch.py" +QUEUE = SKILLS / "kvb-org-queue" / "scripts" / "queue.py" +HOLD_KIND = "operator" +HOLD_SESSION = "Operator" +NOT_RUN = 75 +GRANT_SECONDS = 600 +SUDO = re.compile(r"\bsudo\b") PREFIX = "operator/" KINDS = ("pushed", "installed", "binary", "absent", "manual", "decision") OPEN = ("queued", "blocked") @@ -264,8 +288,8 @@ def runnable(row: dict) -> bool: return bool(row["directory"] and row["command"]) and row["check"] != "decision" -def visible(text: str) -> str: - return "".join(c if c.isprintable() else c.encode("unicode_escape").decode() for c in text) +def visible(text: str, keep: str) -> str: + return "".join(c if c.isprintable() or c in keep else c.encode("unicode_escape").decode() for c in text) def git_tree(directory: str) -> bool: @@ -312,14 +336,113 @@ def show(commands: list[tuple[list[str], str]]) -> None: print(f" {error}", flush=True) -def ask() -> str: +def ask(holdable: bool, read: Callable[[str], str | None]) -> str: + choices = ("y", "h", "s", "d", "q") if holdable else ("y", "s", "d", "q") + prompt = "Run it? [y]es / [h]old in the queue / [s]kip / [d]iff / [q]uit: " if holdable else "Run it? [y]es / [s]kip / [d]iff / [q]uit: " while True: - try: - answer = input("Run it? [y]es / [s]kip / [d]iff / [q]uit: ").strip().lower() - except EOFError: + answer = read(prompt) + if answer is None: return "q" - if answer in ("y", "s", "d", "q"): - return answer + if answer.strip().lower() in choices: + return answer.strip().lower() + + +def typed(prompt: str) -> str | None: + try: + return input(prompt) + except EOFError: + return None + + +def timed(prompt: str) -> str | None: + """Returns: + One line read unbuffered from stdin, None at its end or when no line + came within GRANT_SECONDS, so an unanswered grant frees its slot; + what was typed while the job waited is dropped first. + """ + if sys.stdin.isatty(): + termios.tcflush(sys.stdin.fileno(), termios.TCIFLUSH) + print(prompt, end="", flush=True) + deadline = time.monotonic() + GRANT_SECONDS + line = b"" + while not line.endswith(b"\n"): + ready, _, _ = select.select([sys.stdin.fileno()], [], [], max(0.0, deadline - time.monotonic())) + byte = os.read(sys.stdin.fileno(), 1) if ready else b"" + if not byte: + print(flush=True) + return None + line += byte + return line.decode(errors="replace") + + +def needs_terminal(command: str) -> bool: + return SUDO.search(command) is not None + + +def heavy_row(watch: ModuleType, cfg: dict | None, row: dict) -> bool: + """Returns: + Whether [h]old can queue the row's command: a heavy job by the + kvb-org-organize watch config, which holds every queue key and the + kind operator, and no sudo, whose prompt needs the terminal a queued + command lacks; prints the memory stage and `queue.py list` for a + heavy one, its output control characters made visible. + """ + if cfg is None or not watch.heavy(row["command"], cfg, tuple(cfg["never_block"])): + return False + problem = watch.missing(cfg, (*watch.MEMORY_KEYS, "budget_percent.ram", "budget_percent.cpu", f"slot_need_gb.{HOLD_KIND}")) + if problem is None and needs_terminal(row["command"]): + problem = "sudo asks on the terminal, which a queued command lacks" + if problem is None: + sample = watch.memory() + print(f" heavy: memory stage {watch.memory_stage(sample, cfg)} ({watch.memory_text(sample)}); the queue:", flush=True) + else: + print(f" heavy: {problem}; no [h]old until it does. The queue:", flush=True) + listing = subprocess.run([sys.executable, str(QUEUE), "list"], cwd=row["directory"], capture_output=True, text=True, check=False) + print(visible(listing.stdout + listing.stderr, "\t\n"), end="", flush=True) + return problem is None + + +def held(row: dict, shell: str) -> list[str]: + return [ + sys.executable, str(QUEUE), "hold", "--kind", HOLD_KIND, "--task", row["task"], "--session", HOLD_SESSION, + "--", sys.executable, str(Path(__file__).resolve()), "granted", row["task"], "--dir", row["directory"], + "--check", row["check"], "--", shell, "-ic", row["command"], + ] + + +def review(row: dict, holdable: bool, read: Callable[[str], str | None]) -> str: + check = "manual" if problem(row["check"]) else row["check"] + show(evidence(check, row["directory"], full=False)) + answer = ask(holdable, read) + while answer == "d": + show(evidence(check, row["directory"], full=True)) + answer = ask(holdable, read) + return answer + + +def cmd_granted(rasci: ModuleType, args: argparse.Namespace) -> int: + shell = os.environ.get("SHELL") + approved = [ + row for row in open_rows(rasci) + if runnable(row) and (row["task"], row["directory"], row["check"]) == (args.task, args.dir, args.check) + and args.argv == [shell, "-ic", row["command"]] + ] + if not shell or not approved: + print(f"operator: {args.task} is no longer open with the approved directory, check and command; nothing ran", + file=sys.stderr) + return NOT_RUN + row = approved[0] + print(f"\ngranted {row['task']}: the queue lets it run now; its tree may have changed while it waited") + print(f" directory: {visible(row['directory'], '')}") + print(f" command: {visible(row['command'], '')}", flush=True) + try: + answer = review(row, False, timed) + except KeyboardInterrupt: + answer = "q" + if answer != "y": + print(f"not run {row['task']}") + return NOT_RUN + return subprocess.run(args.argv, cwd=row["directory"], check=False).returncode def close(rasci: ModuleType, task: str) -> None: @@ -337,26 +460,26 @@ def cmd_run(rasci: ModuleType, args: argparse.Namespace) -> int: if not steps: print("operator: nothing to run") return 0 + watch = module("kvb_operator_watch", WATCH) + cfg = watch.load_config() for number, row in enumerate(steps, 1): check = "manual" if problem(row["check"]) else row["check"] print(f"\n[{number}/{len(steps)}] {row['task']}") - print(f" directory: {visible(row['directory'])}") - print(f" command: {visible(row['command'])}") - print(f" why: {visible(row['why'] or '-')}") + print(f" directory: {visible(row['directory'], '')}") + print(f" command: {visible(row['command'], '')}") + print(f" why: {visible(row['why'] or '-', '')}") print(f" check: {check}", flush=True) - show(evidence(check, row["directory"], full=False)) - answer = ask() - while answer == "d": - show(evidence(check, row["directory"], full=True)) - answer = ask() + holdable = heavy_row(watch, cfg, row) + answer = review(row, holdable, typed) if answer == "q": print("operator: stopped") return 0 if answer == "s": print(f"skipped {row['task']}") continue + argv = held(row, shell) if answer == "h" else [shell, "-ic", row["command"]] try: - code = subprocess.run([shell, "-ic", row["command"]], cwd=row["directory"], check=False).returncode + code = subprocess.run(argv, cwd=row["directory"], check=False).returncode except OSError as error: print(f"failed {row['task']}: {error}") continue @@ -394,6 +517,12 @@ def main() -> int: p.set_defaults(func=cmd_list) p = sub.add_parser("run") p.set_defaults(func=cmd_run) + p = sub.add_parser("granted") + p.add_argument("task") + p.add_argument("--dir", required=True) + p.add_argument("--check", required=True) + p.add_argument("argv", nargs="+") + p.set_defaults(func=cmd_granted) p = sub.add_parser("done") p.add_argument("slug") p.set_defaults(func=cmd_done) diff --git a/skills/kvb-org-organize/scripts/watch.py b/skills/kvb-org-organize/scripts/watch.py index fc21c53..b308673 100755 --- a/skills/kvb-org-organize/scripts/watch.py +++ b/skills/kvb-org-organize/scripts/watch.py @@ -9,11 +9,14 @@ Usage: (kvb-check-security) whose id it lacks to the same name under ~/.local/state/agents-council/ watch.py block hook (PreToolUse on Bash): exit 2 for a heavy job while the - stage is stop or critical, or while a kvb-org-radio mayday - holds radio silence (/radio-silence.json, until > now) + disk stage is stop or critical, the memory stage is stop + or cannot be judged without memory_limits, or a + kvb-org-radio mayday holds radio silence + (/radio-silence.json, until > now) watch.py admit (--need-gb N | --kind KIND) exit 0 when the grant leaves at least the stop threshold - free, 3 otherwise + free, CPU busy and RAM used stay below load_limits and + the memory stage is not stop, 3 otherwise watch.py cleanup (--caches | --full) trim caches to their budget; --full also runs the configured low-risk cleanup commands @@ -45,8 +48,15 @@ compose project name is skipped), worktree_marker (path relative to a worktree that must exist), busy_processes (process names; a worktree where one of them runs is skipped), cache_dirs, cache_budget_percent, slot_need_gb {kind: GB}, notify (argv the alert text is appended to), -load_limits {cpu_percent, mem_percent, sample_seconds, kinds}: admit for a -kind listed in kinds also needs CPU busy and RAM used below the limits. +load_limits {cpu_percent, mem_percent, sample_seconds}: every admit and every +queue grant needs CPU busy and RAM used below them, memory_limits {warn, stop}, +each {available, swap_free, psi_full, container_swap_gb}: the memory stage is +the first of stop and warn whose MemAvailable or SwapFree percentage lies below +its bound, whose /proc/pressure/memory full avg10 reaches its bound, or where +one container (a docker-*.scope cgroup under system.slice, nested groups +counted) holds at least container_swap_gb GB of swap, else ok; a host without +swap or PSI skips that bound. heavy_commands extend the built-in rules of +heavy(). """ from __future__ import annotations @@ -56,6 +66,7 @@ import contextlib import fcntl import importlib.util import json +import math import os import re import shlex @@ -99,6 +110,86 @@ CLIP = 1000 CLEANUP_LOCK = "cleanup.lock" LOCK_TTL = 1800 GB = 1e9 +MEMINFO = Path("/proc/meminfo") +PRESSURE = Path("/proc/pressure/memory") +CGROUPS = Path("/sys/fs/cgroup/system.slice") +SWAP_HOLDERS = 3 +NEST_DEPTH = 8 +MEMORY_STAGES = ("ok", "warn", "stop") +MEMORY_BOUNDS = ("available", "swap_free", "psi_full", "container_swap_gb") +MEMORY_KEYS = tuple(f"memory_limits.{stage}.{bound}" for stage in MEMORY_STAGES[1:] for bound in MEMORY_BOUNDS) +CONTAINER_ID = re.compile(r"[0-9a-f]{64}") +COMPOSE_VALUES = frozenset({ + "-f", "--file", "-p", "--project-name", "--project-directory", "--env-file", "--profile", "--ansi", "--parallel", + "--progress", +}) +BUILDING = frozenset({"up", "run", "create"}) +SEPARATOR = re.compile(r"[();<>|&\n]+") +PUNCTUATION = "();<>|&\n" +WORD_BREAK = " \t\n;&|()<>" +ASSIGNMENT = re.compile(r"[A-Za-z_]\w*=.*", re.DOTALL) +WRAPPERS = frozenset({ + "env", "sudo", "nice", "nohup", "timeout", "exec", "xargs", "stdbuf", "ionice", "uv", "uvx", "poetry", "pdm", + "pipenv", "coverage", "time", "command", +}) +UV_VALUES = frozenset({ + "--with", "--extra", "--group", "--python", "-p", "--project", "--directory", "--package", "--env-file", "--from", +}) +WRAPPER_VALUES = { + "sudo": frozenset({ + "-u", "-g", "-h", "-p", "-C", "-D", "-r", "-t", "-U", "-T", "--user", "--group", "--host", "--chdir", "--role", + "--type", "--prompt", "--other-user", "--command-timeout", "--close-from", + }), + "timeout": frozenset({"-s", "-k", "--signal", "--kill-after"}), + "env": frozenset({"-u", "-C", "--unset", "--chdir"}), + "nice": frozenset({"-n", "--adjustment"}), + "ionice": frozenset({"-c", "-n", "-p", "--class", "--classdata"}), + "xargs": frozenset({"-a", "-d", "-E", "-I", "-L", "-n", "-P", "-s"}), + "stdbuf": frozenset({"-i", "-o", "-e"}), + "time": frozenset({"-f", "-o", "--format", "--output"}), + "exec": frozenset({"-a"}), + "uv": UV_VALUES, + "uvx": UV_VALUES, + "poetry": frozenset({"-C", "-P", "--directory", "--project"}), + "pdm": frozenset({"-p", "--project"}), + "coverage": frozenset({"--source", "--include", "--omit", "--rcfile", "--context", "--data-file"}), +} +WRAPPER_WORDS = frozenset({"run", "tool"}) +RESERVED = frozenset({"!", "{", "if", "then", "else", "elif", "do", "while", "until"}) +WRAPPER_ARGUMENT = re.compile(r"-.*|\d+(\.\d+)?[smhd]?", re.DOTALL) +SHELLS = frozenset({"bash", "sh", "zsh", "dash"}) +INLINE = re.compile(r"-[A-Za-z]*c[A-Za-z]*") +REDIRECT = re.compile(r"&?[<>]+&?") +TRAILING_REDIRECT = re.compile(r"[<>]+&?$") +PAREN = {"(": 1, ")": -1} +PYTHON = re.compile(r"python[\d.]*") +PYTHON_VALUE_OPTIONS = frozenset({"-X", "-W", "-Q"}) +DRY_RUN = frozenset({"-n", "--dry-run", "--just-print", "--recon", "-q", "--question"}) +SHORT_DRY = re.compile(r"-(?:(?![CfIoOWjl])[A-Za-z])*[nq][A-Za-z]*") +SHELL_VALUE = re.compile(r"[-+][A-Za-z]*[oO]|--rcfile|--init-file") +STDIN_FLAG = re.compile(r"-[A-Za-z]*s[A-Za-z]*") +SPACE = re.compile(r"\s") +ESCAPED = re.compile(r"\\[`$]") +OPENER = re.compile(r"`|\$\(") +PLACEHOLDER = "-_" +HEREDOC_MARK = "kvbheredoc" +MARK_WORD = re.compile(rf"{HEREDOC_MARK}((?:[0-9a-f]{{2}})*)") +HELP = frozenset({"-h", "--help", "--version"}) +PYTEST_LIGHT = HELP | {"--collect-only", "--co"} +TEST_FILE = re.compile(r"[^*?\[\]]+\.py(::.*)?", re.DOTALL) +PYTEST_VALUE_OPTIONS = frozenset({ + "-p", "-k", "-m", "-c", "-o", "-W", "-n", "--numprocesses", "-r", "--maxfail", "--tb", "--rootdir", "--basetemp", "--deselect", + "--ignore", "--ignore-glob", "--confcutdir", "--durations", "--junitxml", "--log-level", "--dist", "--cov", + "--cov-report", "--cov-config", "--timeout", "--import-mode", "--capture", "--color", "--log-cli-level", + "--log-file", "--reruns", "--tx", "--max-worker-restart", +}) +UNITTEST_VALUE_OPTIONS = frozenset({"-k"}) +DISCOVER_VALUE_OPTIONS = frozenset({"-s", "-p", "-t", "-k", "--start-directory", "--pattern", "--top-level-directory"}) +PATTERN_OPTIONS = frozenset({"-p", "--pattern"}) +MAKE_VALUE_OPTIONS = frozenset({ + "-C", "-f", "-I", "-o", "-W", "--directory", "--file", "--makefile", "--include-dir", "--old-file", "--assume-old", + "--what-if", "--new-file", "--assume-new", +}) def guard_root(hook: bool) -> int | None: @@ -123,6 +214,35 @@ def load_config() -> dict | None: return json.loads(CONFIG.read_text(encoding="utf-8")) +def absent_part(cfg: dict, key: str) -> str | None: + node: object = cfg + parts = key.split(".") + for depth, part in enumerate(parts, 1): + if not isinstance(node, dict) or part not in node: + return ".".join(parts[:depth]) + node = node[part] + return None if isinstance(node, (int, float)) and not isinstance(node, bool) and math.isfinite(node) else key + + +def missing(cfg: dict, keys: tuple[str, ...]) -> str | None: + """Args: + cfg: the config. + keys: required number keys, nested ones dotted + (`memory_limits.stop.psi_full`). + + Returns: + None when every key holds a number, else the error naming the + shortest missing part of each, or the key whose value is no number. + """ + absent = list(dict.fromkeys(part for key in keys if (part := absent_part(cfg, key)) is not None)) + if not absent: + return None + return ( + f"agents council: {CONFIG} lacks {', '.join(absent)}; add them as numbers as the watch.py and queue.py " + "docstrings document" + ) + + def disk(cfg: dict) -> tuple[float, float]: st = os.statvfs(cfg["disk"]) return st.f_bavail * st.f_frsize / GB, st.f_blocks * st.f_frsize / GB @@ -237,6 +357,19 @@ def act(cfg: dict, stage: str, previous: str, free: float) -> None: notify(cfg, f"Agents council: {head}. Heavy jobs blocked; a running stack needs the operator's OK to go.") +def act_memory(cfg: dict, stage: str, previous: str, sample: dict[str, float | None]) -> None: + head = memory_text(sample) + if MEMORY_STAGES.index(stage) < MEMORY_STAGES.index(previous): + alert(stage, f"RESOURCE MEMORY {stage.upper()}: recovered, {head}.") + return + holders = "; ".join(swap_holders(CGROUPS, container_names())) or "no container holds swap" + if stage == "warn": + alert(stage, f"RESOURCE MEMORY WARN: {head}. Top swap holders: {holders}. Chair: find the holder; peers: start nothing heavy.") + else: + alert(stage, f"RESOURCE MEMORY STOP: {head}. New heavy jobs are blocked. Top swap holders: {holders}.") + notify(cfg, f"Agents council: memory nearly exhausted ({head}). Heavy jobs blocked; top swap holders: {holders}.") + + def history_rows(text: str) -> list[tuple[str, str]]: rows = [] for line in text.splitlines(): @@ -280,27 +413,42 @@ def stale(state: dict, cfg: dict, now: float) -> bool: return now - state.get("checked", 0) >= cfg["interval_seconds"] -def cmd_watch(cfg: dict) -> int: +def measure_and_act(cfg: dict, state: dict, now: float, problem: str | None) -> None: + free, _ = disk(cfg) + samples = [s for s in state.get("samples", []) if now - s[0] <= cfg["projection_minutes"] * 60] + stage = stage_for(free, samples, cfg, now) + previous = state.get("stage", "ok") + samples.append([now, free]) + state.update(checked=now, free=free, stage=stage, samples=samples) + sample = None if problem else memory() + was_short = state.get("memory_stage", "ok") + if sample is not None: + state["memory_stage"] = memory_stage(sample, cfg) + due = now - state.get("caches_checked", 0) >= cfg["cache_interval_seconds"] + if due: + state["caches_checked"] = now + write_json(STATE, state) + if stage != previous: + act(cfg, stage, previous, free) + elif due: + spawn_cleanup("--caches") + if sample is not None and state["memory_stage"] != was_short: + act_memory(cfg, state["memory_stage"], was_short, sample) + + +def cmd_watch(cfg: dict, problem: str | None) -> int: + """Args: + cfg: the config. + problem: why the memory stage cannot be judged, None when it can; the + disk part still runs and every session sees the problem. + """ payload = json.load(sys.stdin) now = time.time() state = read_json(STATE) if stale(state, cfg, now): - free, _ = disk(cfg) - samples = [s for s in state.get("samples", []) if now - s[0] <= cfg["projection_minutes"] * 60] - stage = stage_for(free, samples, cfg, now) - previous = state.get("stage", "ok") - samples.append([now, free]) - state.update(checked=now, free=free, stage=stage, samples=samples) - due = now - state.get("caches_checked", 0) >= cfg["cache_interval_seconds"] - if due: - state["caches_checked"] = now - write_json(STATE, state) - if stage != previous: - act(cfg, stage, previous, free) - elif due: - spawn_cleanup("--caches") + measure_and_act(cfg, state, now, problem) keep_history() - texts = unseen_alerts(payload["session_id"]) + texts = unseen_alerts(payload["session_id"]) + ([] if problem is None else [problem]) if texts: print(json.dumps({"hookSpecificOutput": { "hookEventName": payload["hook_event_name"], @@ -309,7 +457,10 @@ def cmd_watch(cfg: dict) -> int: return 0 -HEREDOC = re.compile(r"<<-?\s*(['\"]?)(\w+)\1[^\n]*\n.*?\n\s*\2(?=\s|\)|$)", re.DOTALL) +HEREDOC = re.compile( + r"<<-?\s*(?P['\"]?)(?P\\?)(?P[\w.-]+)(?P=quote)(?P[^\n]*)\n(?P.*?)\n\s*(?P=tag)(?=\s|\)|$)", + re.DOTALL, +) QUOTED = re.compile(r"'[^']*'|\"(?:\\.|[^\"\\])*\"") @@ -322,32 +473,369 @@ def silence() -> dict | None: return state if state.get("until", 0) > time.time() else None -def cmd_block(cfg: dict) -> int: +def uncommented(command: str) -> str: + kept: list[str] = [] + quote = "" + escaped = skipping = False + word_start = True + for char in command: + if skipping and char != "\n": + continue + skipping = False + if escaped: + escaped = False + elif quote == "'": + quote = "" if char == "'" else quote + elif char == "\\": + escaped = True + elif quote == '"': + quote = "" if char == '"' else quote + elif char in "'\"": + quote = char + elif char == "#" and word_start: + skipping = True + continue + kept.append(char) + word_start = not quote and not escaped and char in WORD_BREAK + return "".join(kept) + + +def segments(command: str) -> list[list[str]]: + """Args: + command: a shell command line. + + Returns: + The words of each simple command, split at newlines, `;`, `&`, `|` + and parentheses; a redirection, its target and a file descriptor + number in front of it are dropped and the command goes on. + + Raises: + ValueError: on an unbalanced quote. + """ + lexer = shlex.shlex(uncommented(command.replace("\\\n", "")), posix=True, punctuation_chars=PUNCTUATION) + lexer.whitespace = " \t\r" + lexer.whitespace_split = True + lexer.commenters = "" + parts: list[list[str]] = [[]] + target = False + for token in lexer: + if REDIRECT.fullmatch(token): + if parts[-1] and parts[-1][-1].isdigit(): + parts[-1].pop() + target = True + elif SEPARATOR.fullmatch(token): + parts.append([]) + target = TRAILING_REDIRECT.search(token) is not None + elif target: + target = False + else: + parts[-1].append(token) + return parts + + +def program(words: list[str]) -> list[str]: + """Returns: + The words from the program on: shell reserved words, VAR=value, + wrappers such as env, sudo, time, timeout or `uv run`, their options, + the values of their options and their numbers dropped; the split + string of `env -S`, nothing for the lookup `command -v`. + """ + at, wrapper = 0, "" + while at < len(words): + word = words[at] + if os.path.basename(word) in WRAPPERS: + wrapper = os.path.basename(word) + elif word in WRAPPER_VALUES.get(wrapper, ()): + at += 1 + elif word in WRAPPER_TURNS.get(wrapper, {}): + return WRAPPER_TURNS[wrapper][word](words[at + 1 :]) + elif not (word in RESERVED or ASSIGNMENT.fullmatch(word) or (wrapper and wrapped_word(word))): + break + at += 1 + return words[at:] + + +def lookup(rest: list[str]) -> list[str]: + return [] + + +def split_string(rest: list[str]) -> list[str]: + return program(" ".join(rest[:1]).split() + rest[1:]) + + +WRAPPER_TURNS = { + "command": {"-v": lookup, "-V": lookup}, + "env": {"-S": split_string, "--split-string": split_string}, +} + + +def wrapped_word(word: str) -> bool: + return word in WRAPPER_WORDS or WRAPPER_ARGUMENT.fullmatch(word) is not None + + +def inline_code(words: list[str]) -> str | None: + """Args: + words: a simple command from its program on. + + Returns: + The code a shell runs with -c (also clustered, such as -lc or -ic, + and after a `--`), else None. + """ + if not words or os.path.basename(words[0]) not in SHELLS: + return None + rest = words[1:] + for at, word in enumerate(rest): + if INLINE.fullmatch(word): + code = [later for later in rest[at + 1 :] if later != "--"][:1] + return code[0] if code else None + return None + + +def operands(args: list[str], value_options: frozenset[str]) -> list[str]: + found, skip = [], False + for word in args: + if skip: + skip = False + elif word in value_options: + skip = True + elif not word.startswith("-"): + found.append(word) + return found + + +def python_module(args: list[str]) -> tuple[str, list[str]]: + skip = False + for at, word in enumerate(args): + if skip: + skip = False + elif word == "-m": + return (args[at + 1], args[at + 2 :]) if at + 1 < len(args) else ("", []) + elif word.startswith("-m"): + return word[2:], args[at + 1 :] + elif word in PYTHON_VALUE_OPTIONS: + skip = True + elif not word.startswith("-"): + return os.path.basename(word), args[at + 1 :] + return "", [] + + +def light_unittest(args: list[str]) -> bool: + if args[:1] != ["discover"]: + return len(operands(args, UNITTEST_VALUE_OPTIONS)) == 1 + rest = args[1:] + patterns = [rest[at + 1] for at, word in enumerate(rest[:-1]) if word in PATTERN_OPTIONS] + patterns = patterns or operands(rest, DISCOVER_VALUE_OPTIONS)[1:2] + return bool(patterns) and TEST_FILE.fullmatch(patterns[-1]) is not None + + +def heavy_pytest(args: list[str]) -> bool: + files = operands(args, PYTEST_VALUE_OPTIONS) + return not (PYTEST_LIGHT.intersection(args) or (len(files) == 1 and TEST_FILE.fullmatch(files[0]))) + + +def heavy_unittest(args: list[str]) -> bool: + return not (HELP.intersection(args) or light_unittest(args)) + + +def make_targets(args: list[str]) -> list[str]: + return [word for word in operands(args, MAKE_VALUE_OPTIONS) if "=" not in word and not word.isdigit()] + + +def dry_run(args: list[str]) -> bool: + return bool(DRY_RUN.intersection(args)) or any(SHORT_DRY.fullmatch(word) for word in args) + + +def heavy_make(args: list[str]) -> bool: + return not dry_run(args) and any(target == "install" or target.startswith("verify-") for target in make_targets(args)) + + +def heavy_compose(args: list[str]) -> bool: + words = operands(args, COMPOSE_VALUES) + command = words[0] if words else "" + return command == "build" or (command in BUILDING and "--build" in args) + + +def heavy_docker(args: list[str]) -> bool: + return args[:1] == ["compose"] and heavy_compose(args[1:]) + + +HEAVY_TOOLS = { + "pytest": heavy_pytest, "py.test": heavy_pytest, "unittest": heavy_unittest, "make": heavy_make, + "docker": heavy_docker, "docker-compose": heavy_compose, +} + + +def heavy_words(run: list[str], cfg: dict, exempt: tuple[str, ...]) -> bool: + if not run: + return False + code = inline_code(run) + if code is not None: + return heavy(code, cfg, exempt) + name, args = os.path.basename(run[0]), run[1:] + if PYTHON.fullmatch(name): + name, args = python_module(args) + if name in WRAPPERS: + return heavy_words(program([name, *args]), cfg, exempt) + return name in HEAVY_TOOLS and HEAVY_TOOLS[name](args) + + +def segment_texts(words: list[str], run: list[str]) -> list[str]: + return list(dict.fromkeys(" ".join("''" if SPACE.search(word) else word for word in part) for part in (words, run))) + + +def judged(texts: list[str], run: list[str], cfg: dict, exempt: tuple[str, ...]) -> bool: + if any(re.search(pattern, text) for pattern in exempt for text in texts): + return False + if any(re.search(pattern, text) for pattern in cfg["heavy_commands"] for text in texts): + return True + return heavy_words(run, cfg, exempt) + + +def heavy_part(words: list[str], cfg: dict, exempt: tuple[str, ...]) -> bool: + """Returns: + Whether one simple command is heavy; a make call is judged per target + as `make `, so an exempt target never covers the next one. + """ + run = program(words) + if run and os.path.basename(run[0]) == "make" and not dry_run(run[1:]): + return any(judged([f"make {target}"], ["make", target], cfg, exempt) for target in make_targets(run[1:])) + return judged(segment_texts(words, run), run, cfg, exempt) + + +def marked_heredocs(command: str) -> tuple[str, list[re.Match[str]]]: + """Returns: + The command with each heredoc replaced by the word HEREDOC_MARK + followed by its body in hex, so a nested call still reads it, the + rest of its first line kept, and the heredoc matches. + """ + found: list[re.Match[str]] = [] + + def mark(match: re.Match[str]) -> str: + found.append(match) + return f" {HEREDOC_MARK}{match['body'].encode(errors='surrogatepass').hex()} {match['rest']}" + + return HEREDOC.sub(mark, command), found + + +def shell_operands(args: list[str]) -> list[str]: + found, skip = [], False + for word in args: + if skip: + skip = False + elif word.startswith(("-", "+")): + skip = SHELL_VALUE.fullmatch(word) is not None + else: + found.append(word) + return found + + +def reads_stdin(run: list[str]) -> bool: + """Returns: + Whether a simple command from its program on is a shell that runs the + code on its stdin: no -c code, and -s or no operand but heredoc marks. + """ + if not run or os.path.basename(run[0]) not in SHELLS or inline_code(run) is not None: + return False + args = run[1:] + return any(STDIN_FLAG.fullmatch(word) for word in args) or all(MARK_WORD.fullmatch(word) for word in shell_operands(args)) + + +def expanded_codes(heredocs: list[re.Match[str]]) -> list[str]: + """Returns: + The command substitutions in the bodies of heredocs whose delimiter + is unquoted, which the shell expands. + """ + return [code for match in heredocs if not (match["quote"] or match["escape"]) for code in substituted(match["body"])[1]] + + +def fed_bodies(parts: list[list[str]]) -> list[str]: + marks = [MARK_WORD.fullmatch(word) for words in parts if reads_stdin(program(words)) for word in words] + return [bytes.fromhex(mark[1]).decode(errors="ignore") for mark in marks if mark] + + +def closing(text: str, start: int) -> int: + depth = 1 + for at in range(start, len(text)): + depth += PAREN.get(text[at], 0) + if depth == 0: + return at + return len(text) + + +def blanked_single(quoted: re.Match[str]) -> str: + text = quoted[0] + return text if text.startswith('"') else "'" + " " * (len(text) - 2) + "'" + + +def substitution_end(shown: str, found: re.Match[str]) -> int: + if found[0] == "`": + end = shown.find("`", found.end()) + return len(shown) if end < 0 else end + return closing(shown, found.end()) + + +def substituted(text: str) -> tuple[str, list[str]]: + """Returns: + The text with every `...` and $(...) outside single quotes and + escapes replaced by the option-like word PLACEHOLDER, so a wrapper + around it still parses, and the code of each, which the shell runs + first; nested ones stay inside their code. + """ + shown = QUOTED.sub(blanked_single, ESCAPED.sub(" ", text)) + kept: list[str] = [] + codes: list[str] = [] + at = 0 + found = OPENER.search(shown) + while found: + end = substitution_end(shown, found) + kept.append(text[at : found.start()] + PLACEHOLDER) + codes.append(text[found.end() : end]) + at = end + 1 + found = OPENER.search(shown, at) + return "".join(kept) + text[at:], codes + + +def cmd_block(cfg: dict, problem: str | None) -> int: + """Args: + cfg: the config. + problem: why the memory stage cannot be judged, None when it can; a + heavy job is then refused with it. + """ payload = json.load(sys.stdin) - command = executable_text((payload.get("tool_input") or {}).get("command", "")) + raw = (payload.get("tool_input") or {}).get("command", "") + if not heavy(raw, cfg, tuple(cfg["never_block"])): + return 0 + reason = block_reason(cfg, problem) + if reason is None: + return 0 + print(f"kvb-org-organize: {reason}. Teardown commands still run. Ask the resource chair.", file=sys.stderr) + return 2 + + +def disk_stage(cfg: dict) -> str: state = read_json(STATE) now = time.time() - stage = state.get("stage", "ok") if stale(state, cfg, now): - stage = stage_for(disk(cfg)[0], state.get("samples", []), cfg, now) + return stage_for(disk(cfg)[0], state.get("samples", []), cfg, now) + return state.get("stage", "ok") + + +def block_reason(cfg: dict, problem: str | None) -> str | None: mayday = silence() - if stage not in ("stop", "critical") and not mayday: - return 0 - if any(re.search(p, command) for p in cfg["never_block"]): - return 0 - if any(re.search(p, command) for p in cfg["heavy_commands"]): - reason = ( + if mayday: + return ( f"a mayday by {clip(mayday.get('by', '?'))} holds radio silence: {clip(mayday.get('text', ''))}; " "it ends with radio.py clear" - if mayday - else f"resource stage is {stage}; new heavy jobs are blocked until the disk recovers" ) - print( - f"kvb-org-organize: {reason}. Teardown commands still run. Ask the resource chair.", - file=sys.stderr, - ) - return 2 - return 0 + stage = disk_stage(cfg) + if stage in ("stop", "critical"): + return f"resource stage is {stage}; new heavy jobs are blocked until the disk recovers" + if problem is not None: + return f"{problem}; new heavy jobs are blocked until it does" + sample = memory() + if memory_stage(sample, cfg) == "stop": + return f"memory stage is stop ({memory_text(sample)}); new heavy jobs are blocked until memory recovers" + return None def cpu_times() -> tuple[int, int]: @@ -361,12 +849,17 @@ def cpu_busy_percent(before: tuple[int, int], after: tuple[int, int]) -> float: return 100.0 * (total - idle) / total -def mem_used_percent(meminfo: str) -> float: +def meminfo_kb(meminfo: str) -> dict[str, int]: values = {} for line in meminfo.splitlines(): key, _, rest = line.partition(":") if rest: values[key] = int(rest.split()[0]) + return values + + +def mem_used_percent(meminfo: str) -> float: + values = meminfo_kb(meminfo) return 100.0 * (1 - values["MemAvailable"] / values["MemTotal"]) @@ -374,7 +867,161 @@ def load(cfg: dict) -> tuple[float, float]: before = cpu_times() time.sleep(cfg["load_limits"]["sample_seconds"]) cpu = cpu_busy_percent(before, cpu_times()) - return cpu, mem_used_percent(Path("/proc/meminfo").read_text(encoding="utf-8")) + return cpu, mem_used_percent(MEMINFO.read_text(encoding="utf-8")) + + +def psi_full(pressure: str) -> float | None: + for line in pressure.splitlines(): + if line.startswith("full "): + return float(dict(field.split("=", 1) for field in line.split()[1:])["avg10"]) + return None + + +def memory_sample(meminfo: str, pressure: str | None, container_swap_gb: float) -> dict[str, float | None]: + """Args: + meminfo: the text of /proc/meminfo. + pressure: the text of /proc/pressure/memory, None without PSI. + container_swap_gb: the swap of the container holding the most, in GB. + + Returns: + {available, swap_free, psi_full, container_swap_gb}: MemAvailable and + SwapFree in percent of their totals, the full avg10 stall percentage + and container_swap_gb unchanged; swap_free is None without swap, + psi_full None without PSI. + """ + values = meminfo_kb(meminfo) + swap = values["SwapTotal"] + return { + "available": 100.0 * values["MemAvailable"] / values["MemTotal"], + "swap_free": 100.0 * values["SwapFree"] / swap if swap else None, + "psi_full": None if pressure is None else psi_full(pressure), + "container_swap_gb": container_swap_gb, + } + + +def memory() -> dict[str, float | None]: + try: + pressure = PRESSURE.read_text(encoding="utf-8") + except OSError: + pressure = None + largest = max((size for size, _ in container_swaps(CGROUPS)), default=0) + return memory_sample(MEMINFO.read_text(encoding="utf-8"), pressure, largest / GB) + + +def memory_stage(sample: dict[str, float | None], cfg: dict) -> str: + for name in ("stop", "warn"): + bound = cfg["memory_limits"][name] + if ( + sample["available"] < bound["available"] + or (sample["swap_free"] is not None and sample["swap_free"] < bound["swap_free"]) + or (sample["psi_full"] is not None and sample["psi_full"] >= bound["psi_full"]) + or sample["container_swap_gb"] >= bound["container_swap_gb"] + ): + return name + return "ok" + + +def memory_text(sample: dict[str, float | None]) -> str: + swap = "no swap" if sample["swap_free"] is None else f"swap free {sample['swap_free']:.0f} %" + psi = "no PSI" if sample["psi_full"] is None else f"PSI full avg10 {sample['psi_full']:.1f}" + return f"MemAvailable {sample['available']:.0f} %, {swap}, {psi}, largest container swap {sample['container_swap_gb']:.1f} GB" + + +def swap_bytes(cgroup: Path) -> int: + try: + return int((cgroup / "memory.swap.current").read_text(encoding="utf-8")) + except (OSError, ValueError): + return 0 + + +def container_swaps(base: Path) -> list[tuple[int, Path]]: + try: + return [(swap_bytes(scope), scope) for scope in base.glob("docker-*.scope")] + except OSError: + return [] + + +def cgroup_label(name: str, names: dict[str, str]) -> str: + bare = name.removeprefix("docker-").removesuffix(".scope") + if CONTAINER_ID.fullmatch(bare): + return names[bare] if bare in names else bare[:12] + return bare + + +def container_names() -> dict[str, str]: + try: + out = subprocess.run( + ["docker", "ps", "--no-trunc", "--format", "{{.ID}}\t{{.Names}}"], + capture_output=True, text=True, timeout=30, check=False, + ).stdout + except (OSError, subprocess.TimeoutExpired): + return {} + return dict(line.split("\t", 1) for line in out.splitlines() if "\t" in line) + + +def swap_chain(cgroup: Path, held: int, names: dict[str, str]) -> list[str]: + chain = [cgroup_label(cgroup.name, names)] + for _ in range(NEST_DEPTH): + try: + children = [(swap_bytes(child), child) for child in cgroup.iterdir() if child.is_dir()] + except OSError: + break + if not children: + break + size, child = max(children) + if size * 2 < held: + break + chain.append(cgroup_label(child.name, names)) + cgroup, held = child, size + return chain + + +def swap_holders(base: Path, names: dict[str, str]) -> list[str]: + """Args: + base: the cgroup directory holding the docker-.scope groups. + names: {full container id: name} for the labels. + + Returns: + Up to SWAP_HOLDERS lines, the largest swap first: the container, then + each nested group holding at least half of its parent's swap, and the + container's memory.swap.current in GB. + """ + ranked = sorted((pair for pair in container_swaps(base) if pair[0] > 0), reverse=True)[:SWAP_HOLDERS] + return [f"{' > '.join(swap_chain(scope, size, names))} {size / GB:.1f} GB" for size, scope in ranked] + + +def heavy(command: str, cfg: dict, exempt: tuple[str, ...]) -> bool: + """Args: + command: the shell command line as typed, quotes included. + cfg: the config; its heavy_commands search each simple command with + quoted text blanked, and a make call also as `make `. + exempt: regexes; a simple command one of them finds is never heavy + (never_block, the queue's own hold), the others still count. + + Returns: + Whether one simple command is a heavy job: an image build through + docker compose (build, or up, run or create with --build), a + heavy_commands match, make install or verify-* (not as a dry run), or + pytest or unittest on anything but exactly one test file or module, + help and version aside. Wrappers (program()), the code of `sh -c`, + command substitutions (substituted(), also in heredocs the shell + expands) and heredocs a shell reads as its code count the same. A + command with an unbalanced quote, which no shell runs, counts by the + regexes on its whole text alone. A guard against accidental heavy + runs, not a sandbox: a shell reading code from a pipe or a + here-string, a script of its own, a program named by a substitution, + quotes inside an expanding heredoc and an escaped backslash in front + of a substitution are not looked into. + """ + marked, heredocs = marked_heredocs(command) + plain, codes = substituted(uncommented(marked)) + try: + parts = segments(plain) + except ValueError: + text = executable_text(command) + return not any(re.search(p, text) for p in exempt) and any(re.search(p, text) for p in cfg["heavy_commands"]) + nested = codes + expanded_codes(heredocs) + fed_bodies(parts) + return any(heavy_part(words, cfg, exempt) for words in parts) or any(heavy(code, cfg, exempt) for code in nested) def cmd_admit(cfg: dict, args: argparse.Namespace) -> int: @@ -385,12 +1032,12 @@ def cmd_admit(cfg: dict, args: argparse.Namespace) -> int: passed = [left >= floor] report = [f"free {free:.1f} GB, need {need} GB, leaves {left:.1f} GB (floor {floor} GB)"] limits = cfg["load_limits"] - if args.kind in limits["kinds"]: - cpu, mem = load(cfg) - passed += [cpu < limits["cpu_percent"], mem < limits["mem_percent"]] - report.append( - f"cpu {cpu:.0f} % (limit {limits['cpu_percent']}), ram {mem:.0f} % (limit {limits['mem_percent']})" - ) + cpu, mem = load(cfg) + sample = memory() + stage = memory_stage(sample, cfg) + passed += [cpu < limits["cpu_percent"], mem < limits["mem_percent"], stage != "stop"] + report.append(f"cpu {cpu:.0f} % (limit {limits['cpu_percent']}), ram {mem:.0f} % (limit {limits['mem_percent']})") + report.append(f"memory stage {stage} ({memory_text(sample)})") verdict = "grant" if all(passed) else "deny" print(f"{verdict}: {'; '.join(report)}") return 0 if verdict == "grant" else 3 @@ -511,7 +1158,11 @@ def cmd_status(cfg: dict) -> int: now = time.time() samples = state.get("samples", []) age = f"{now - state['checked']:.0f}s" if "checked" in state else "unmeasured" - print(f"stage={stage_for(free, samples, cfg, now)} free={free:.1f}GB total={total:.1f}GB samples={len(samples)} age={age}") + sample = memory() + print( + f"stage={stage_for(free, samples, cfg, now)} free={free:.1f}GB total={total:.1f}GB samples={len(samples)} age={age}" + f" memory={memory_stage(sample, cfg)} ({memory_text(sample)})" + ) return 0 @@ -537,10 +1188,14 @@ def main() -> int: cfg = load_config() if cfg is None: return 0 + problem = missing(cfg, MEMORY_KEYS) if args.cmd == "watch": - return cmd_watch(cfg) + return cmd_watch(cfg, problem) if args.cmd == "block": - return cmd_block(cfg) + return cmd_block(cfg, problem) + if problem is not None: + print(problem, file=sys.stderr) + return 2 if args.cmd == "admit": return cmd_admit(cfg, args) if args.cmd == "cleanup": diff --git a/skills/kvb-org-queue/SKILL.md b/skills/kvb-org-queue/SKILL.md index dc0d920..12370f1 100644 --- a/skills/kvb-org-queue/SKILL.md +++ b/skills/kvb-org-queue/SKILL.md @@ -15,7 +15,7 @@ queue.py hold --kind --task -- ``` Start it in the background (`run_in_background` or equivalent) and end the -turn: hold waits, runs the command on its grant and exits with its code. +turn: hold runs the command on its grant and exits with its code. - ``: a key of `slot_need_gb` in `~/.claude/agents-council/config.json`. - ``: your open `kvb-org-rasci` task; its rank places you. @@ -37,31 +37,32 @@ Every 20 s each waiting hold evaluates the whole queue under one lock: 1. Hotfix jobs first, then the register rank (`rasci.py show`), one place earlier per whole hour waited, then arrival. 2. In that order, a job is blocked at disk stage stop or critical, at warn - when it needs disk, during a radio-silence mayday, while its kind's slots - are taken, or while its workdir runs another job. -3. A job whose need would cut below the stop threshold reserves that disk - against smaller jobs behind it. + when it needs disk, at memory stage stop, during a radio-silence mayday, + while its kind's slots are taken, or while its workdir runs another job. +3. A job whose disk need would cut below the stop threshold, or whose + footprint would push the running ones past `budget_percent`, reserves it + against the jobs behind it; a job alone always fits. 4. An unblocked job counts as granted for the jobs behind it: none takes its slot. -5. Kinds in `load_limits.kinds` also need CPU and RAM below the limits. - -A waiting hold prints a stderr line when its position or reason changes. +5. Every job, alone too, needs CPU, and RAM used plus what it and the + running jobs may still take, below `load_limits`. ## History and learning -Every finished job appends a row to the council's `queue-history.jsonl`: -signature (`make `, `