diff --git a/skills/kvb-ci-triage/funnel.md b/skills/kvb-ci-triage/funnel.md index 9ba07c0..2f985ac 100644 --- a/skills/kvb-ci-triage/funnel.md +++ b/skills/kvb-ci-triage/funnel.md @@ -15,8 +15,8 @@ skill's Rules apply here too. first error. 3. **Fix at the root**: apply the real fix for each proven root cause in the sidetree, with a test and a negative control as the `kvb-org-swarm` skill - demands. Light runs (no Docker, about a minute: targeted pytest or - unittest, ruff, single lint files) self-admit with + demands. Light runs (no Docker, about a minute: one test file or + module, ruff) self-admit with `${CLAUDE_SKILL_DIR}/../kvb-org-organize/scripts/watch.py admit --kind test-nostack`; `make test`, gates, deploys and anything with Docker run in the background through diff --git a/skills/kvb-code-refactor/SKILL.md b/skills/kvb-code-refactor/SKILL.md index 26bd024..a2b688b 100644 --- a/skills/kvb-code-refactor/SKILL.md +++ b/skills/kvb-code-refactor/SKILL.md @@ -26,7 +26,7 @@ cover the change set run now and give the green baseline; a suite already red is reported, not refactored around. No coverage: read `net.md` next to this file before editing. -Light runs (targeted tests, ruff, single lint files) start once `watch.py +Light runs (one test file or module, ruff) start once `watch.py admit --kind test-nostack` of the `kvb-org-organize` skill prints `grant`; no chair request. diff --git a/skills/kvb-code-test/SKILL.md b/skills/kvb-code-test/SKILL.md index 85dd4ba..3c1f89f 100644 --- a/skills/kvb-code-test/SKILL.md +++ b/skills/kvb-code-test/SKILL.md @@ -38,8 +38,8 @@ any listed name. Never silently truncate. ## Step 3: run Before each run, admit it under the `kvb-org-organize` Capacity rules: a -light run (no Docker, about a minute: targeted pytest or unittest, ruff, a -single lint file) self-admits with +light run (no Docker, about a minute: one test file or module, ruff) +self-admits with `${CLAUDE_SKILL_DIR}/../kvb-org-organize/scripts/watch.py admit --kind test-nostack`; `make test`, quality gates and anything with Docker run in the background through diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index 79f73d1..d322a88 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -505,7 +505,7 @@ def heavy_row(watch: ModuleType, cfg: dict | None, row: dict) -> bool: 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"])): + if cfg is None or not watch.heavy(row["command"], cfg, tuple(cfg["never_block"]), row["directory"]): 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"]): diff --git a/skills/kvb-org-organize/SKILL.md b/skills/kvb-org-organize/SKILL.md index 0cdf91d..f4cf759 100644 --- a/skills/kvb-org-organize/SKILL.md +++ b/skills/kvb-org-organize/SKILL.md @@ -37,14 +37,14 @@ ends or goes stale, or after a restart. ## Capacity -- Light runs (no Docker, about a minute: targeted pytest or unittest, ruff, - single lint files) self-admit: `watch.py admit --kind test-nostack`, start - on `grant`. -- `make test`, `make test-lint`, quality gates, deploys, iteration loops, - swarm, image builds, cloud, shared-data cleanup, anything with Docker or - over a few minutes: run through `queue.py hold` (`kvb-org-queue`), which - grants by rule and releases on exit; the chair only handles exceptions. In - a worktree wrap `gate.py run` or `gate.py hold` (`kvb-check-gate`). +- Light runs (no Docker, about a minute: one test file or module, ruff) + self-admit: `watch.py admit --kind test-nostack`, start on `grant`. +- Other test runs, `make test`, `make test-lint`, quality gates, deploys, + iteration loops, swarm, image builds, cloud, shared-data cleanup, anything + with Docker or over a few minutes: run through `queue.py hold` + (`kvb-org-queue`), which grants by rule and releases on exit; the chair only + handles exceptions. In a worktree wrap `gate.py run` or `gate.py hold` + (`kvb-check-gate`). - No running job is cut short for capacity; only the operator orders one down. Exception: void by a missing merged hotfix (`kvb-git-hotfix`). `RESOURCE` hook lines address the chair; others adjust plans. diff --git a/skills/kvb-org-organize/scripts/watch.py b/skills/kvb-org-organize/scripts/watch.py index caee087..4ff1b50 100755 --- a/skills/kvb-org-organize/scripts/watch.py +++ b/skills/kvb-org-organize/scripts/watch.py @@ -64,7 +64,9 @@ from __future__ import annotations import argparse import contextlib import fcntl +import fnmatch import importlib.util +import itertools import json import math import os @@ -186,6 +188,9 @@ PYTEST_VALUE_OPTIONS = frozenset({ UNITTEST_VALUE_OPTIONS = frozenset({"-k"}) DISCOVER_VALUE_OPTIONS = frozenset({"-s", "-p", "-t", "-k", "--start-directory", "--pattern", "--top-level-directory"}) PATTERN_OPTIONS = frozenset({"-p", "--pattern"}) +START_OPTIONS = frozenset({"-s", "--start-directory"}) +CHDIR_OPTIONS = frozenset({"-C", "--chdir"}) +CHDIR_LONG = "--chdir=" MAKE_VALUE_OPTIONS = frozenset({ "-C", "-f", "-I", "-o", "-W", "--directory", "--file", "--makefile", "--include-dir", "--old-file", "--assume-old", "--what-if", "--new-file", "--assume-new", @@ -619,13 +624,71 @@ def python_module(args: list[str]) -> tuple[str, list[str]]: return "", [] -def light_unittest(args: list[str]) -> bool: +def tool(run: list[str]) -> tuple[str, list[str]]: + """Returns: + The program's name and arguments, the module and its arguments for + `python -m` or a Python script. + """ + name, args = os.path.basename(run[0]), run[1:] + return python_module(args) if PYTHON.fullmatch(name) else (name, args) + + +def discover_patterns(rest: list[str]) -> list[str]: + """Returns: + The file patterns `unittest discover` gets after `discover`: -p or + --pattern values, else the second operand; empty for the default. + """ + patterns = [rest[at + 1] for at, word in enumerate(rest[:-1]) if word in PATTERN_OPTIONS] + return patterns or operands(rest, DISCOVER_VALUE_OPTIONS)[1:2] + + +def started_in(prefix: list[str], cwd: str | None) -> str | None: + """Args: + prefix: the words in front of a command's program. + cwd: the directory the shell runs the command line in, None when + unknown. + + Returns: + The directory the command starts in: cwd joined with every `env + -C|--chdir` value of prefix in order, each relative to the one + before; None when cwd is None. + """ + if cwd is None: + return None + dirs = [word[len(CHDIR_LONG):] if word.startswith(CHDIR_LONG) else prefix[at + 1] + for at, word in enumerate(prefix) + if word.startswith(CHDIR_LONG) or (word in CHDIR_OPTIONS and at + 1 < len(prefix))] + return os.path.join(cwd, *dirs) + + +def one_discovered(rest: list[str], base: str | None) -> bool: + """Args: + rest: the arguments after `unittest discover`. + base: the directory the command starts in (started_in()), None when + unknown. + + Returns: + Whether exactly one file below the start directory matches the + pattern: a plain name can match a file in every package of a tree. + Without base or start directory, whether the pattern is one file + name (TEST_FILE), so an unknown tree never blocks a single file. + """ + patterns = discover_patterns(rest) + if not patterns: + return False + starts = [rest[at + 1] for at, word in enumerate(rest[:-1]) if word in START_OPTIONS] + start = (starts or operands(rest, DISCOVER_VALUE_OPTIONS)[:1] or ["."])[-1] + if base is None or not os.path.isdir(os.path.join(base, start)): + return TEST_FILE.fullmatch(patterns[-1]) is not None + found = (name for _, _, files in os.walk(os.path.join(base, start)) for name in files + if fnmatch.fnmatchcase(name, patterns[-1])) + return len([*itertools.islice(found, 2)]) == 1 + + +def light_unittest(args: list[str], base: str | None) -> 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 + return one_discovered(args[1:], base) def heavy_pytest(args: list[str]) -> bool: @@ -633,8 +696,8 @@ def heavy_pytest(args: list[str]) -> bool: 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 heavy_unittest(args: list[str], base: str | None) -> bool: + return not (HELP.intersection(args) or light_unittest(args, base)) def make_targets(args: list[str]) -> list[str]: @@ -660,22 +723,26 @@ def heavy_docker(args: list[str]) -> bool: HEAVY_TOOLS = { - "pytest": heavy_pytest, "py.test": heavy_pytest, "unittest": heavy_unittest, "make": heavy_make, + "pytest": heavy_pytest, "py.test": heavy_pytest, "make": heavy_make, "docker": heavy_docker, "docker-compose": heavy_compose, } -def heavy_words(run: list[str], cfg: dict, exempt: tuple[str, ...]) -> bool: +def heavy_words(run: list[str], cfg: dict, exempt: tuple[str, ...], base: str | None) -> bool: + """Args: + base: the directory the command starts in (started_in()), None when + unknown. + """ 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 heavy(code, cfg, exempt, base) + name, args = tool(run) + if PYTHON.fullmatch(os.path.basename(run[0])) and name in WRAPPERS: + return heavy_words(program([name, *args]), cfg, exempt, base) + if name == "unittest": + return heavy_unittest(args, base) return name in HEAVY_TOOLS and HEAVY_TOOLS[name](args) @@ -683,23 +750,24 @@ 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: +def judged(texts: list[str], run: list[str], cfg: dict, exempt: tuple[str, ...], base: str | None) -> 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) + return heavy_words(run, cfg, exempt, base) -def heavy_part(words: list[str], cfg: dict, exempt: tuple[str, ...]) -> bool: +def heavy_part(words: list[str], cfg: dict, exempt: tuple[str, ...], cwd: str | None) -> 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) + base = started_in(words[: len(words) - len(run)], cwd) 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) + return any(judged([f"make {target}"], ["make", target], cfg, exempt, base) for target in make_targets(run[1:])) + return judged(segment_texts(words, run), run, cfg, exempt, base) def marked_heredocs(command: str) -> tuple[str, list[re.Match[str]]]: @@ -795,6 +863,15 @@ def substituted(text: str) -> tuple[str, list[str]]: return "".join(kept) + text[at:], codes +def payload_cwd(payload: dict) -> str | None: + """Returns: + The hook payload's cwd when it is an absolute path, else None + (unknown, heavy()). + """ + cwd = payload.get("cwd") + return cwd if isinstance(cwd, str) and os.path.isabs(cwd) else None + + def cmd_block(cfg: dict, problem: str | None) -> int: """Args: cfg: the config. @@ -803,7 +880,7 @@ def cmd_block(cfg: dict, problem: str | None) -> int: """ payload = json.load(sys.stdin) raw = (payload.get("tool_input") or {}).get("command", "") - if not heavy(raw, cfg, tuple(cfg["never_block"])): + if not heavy(raw, cfg, tuple(cfg["never_block"]), payload_cwd(payload)): return 0 reason = block_reason(cfg, problem) if reason is None: @@ -990,20 +1067,23 @@ def swap_holders(base: Path, names: dict[str, str]) -> list[str]: 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: +def heavy(command: str, cfg: dict, exempt: tuple[str, ...], cwd: str | None) -> 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. + cwd: the directory the shell runs it in (the hook payload's cwd), + None when unknown; a cd, pushd or popd in it makes it unknown. 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`, + pytest or unittest on anything but exactly one test file or module + (a discover pattern only when one_discovered()), 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 @@ -1021,7 +1101,10 @@ def heavy(command: str, cfg: dict, exempt: tuple[str, ...]) -> bool: 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) + if any(words[:1] in (["cd"], ["pushd"], ["popd"]) for words in parts): + cwd = None + return (any(heavy_part(words, cfg, exempt, cwd) for words in parts) + or any(heavy(code, cfg, exempt, cwd) for code in nested)) def cmd_admit(cfg: dict, args: argparse.Namespace) -> int: diff --git a/skills/kvb-org-queue/SKILL.md b/skills/kvb-org-queue/SKILL.md index 296f851..6a23e53 100644 --- a/skills/kvb-org-queue/SKILL.md +++ b/skills/kvb-org-queue/SKILL.md @@ -27,8 +27,9 @@ turn: hold runs the command on its grant and exits with its code. - In a worktree the command stays `gate.py run|hold` (`kvb-check-gate`): `queue.py hold ... -- python3 run ...`. -Light runs (no Docker, about a minute) self-admit with -`watch.py admit --kind test-nostack` (`kvb-org-organize`). +One test file or module, or ruff, without Docker self-admits with +`watch.py admit --kind test-nostack` (`kvb-org-organize`); any other test +run goes through hold with kind `test`. ## Order and grant diff --git a/skills/kvb-org-queue/learning.md b/skills/kvb-org-queue/learning.md index 32bb723..fd7bfeb 100644 --- a/skills/kvb-org-queue/learning.md +++ b/skills/kvb-org-queue/learning.md @@ -6,10 +6,25 @@ kind, minutes, disk, RAM and CPU) to `queue-history.jsonl`; the From the last 30 days, once 5 runs exist: -- per signature, the disk need is p90 disk taken plus 1 GB; +- per job, the disk need is p90 disk taken plus 1 GB over the runs of its + signature and kind; until 5 exist there, the largest such need its + signature and the command's older signature learned over every kind; - per kind, the slot limit is how many p90 RAM and median CPU shares fit between the median load at the grant and `load_limits`, at least 1; -- per job, the footprint is its signature's share, else its kind's. +- per job, the footprint is its signature's share over the runs of its + kind; until 5 exist there, the largest of its kind's share and what its + signature and the command's older signature learned over every kind. + +A job of a kind outside `light_kinds` never books less than the queue before +the per-kind learning: at least the share its older signature learned over +every kind, else its kind's, and at least that signature's disk need once +learned. Only light kinds learn lower. + +A pytest or unittest signature counts what the run names and its xdist +workers (`pytest files=1`, `pytest paths=2 workers=auto`), so one test file +never learns from a suite; the older signature is `: `, which older rows carry. A granted job keeps the footprint it +was granted on. Until then the seeds hold: `slot_need_gb[kind]` for disk, 1 slot per kind, the whole budget as footprint. diff --git a/skills/kvb-org-queue/scripts/history.py b/skills/kvb-org-queue/scripts/history.py index e179839..3008fb4 100644 --- a/skills/kvb-org-queue/scripts/history.py +++ b/skills/kvb-org-queue/scripts/history.py @@ -13,10 +13,11 @@ from __future__ import annotations import json import math +import operator import os import re import statistics -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Hashable, Iterable, Iterator from pathlib import Path MIN_RUNS = 5 @@ -27,6 +28,10 @@ SEED_LIMIT = 1 INTERPRETERS = re.compile(r"python[\d.]*|bash|sh|zsh|node|perl|ruby") MAKE_VALUE_OPTIONS = frozenset({"-C", "-f", "-I", "-o", "-W"}) RESOURCES = ("ram", "cpu") +SIGNATURE_KEY = operator.itemgetter("kind", "signature") +NAME_KEY = operator.itemgetter("signature") +KIND_KEY = operator.itemgetter("kind") +ANY_KIND = "*" SIGNATURE_HEADER = ("Signature", "Kind", "Runs", "Median min", "p90 disk GB", "p90 RAM %", "Need GB") KIND_HEADER = ("Kind", "Runs", "p90 RAM share %", "Median CPU share %", "Slot limit", "Source") @@ -142,18 +147,15 @@ def shares(rows: list[dict], resource: str) -> list[float]: return [r[f"{resource}_max"] - r[f"{resource}_before"] for r in rows] -def grouped(rows: list[dict], key: str) -> dict[str, list[dict]]: - groups: dict[str, list[dict]] = {} +def grouped(rows: list[dict], key: Callable[[dict], Hashable]) -> dict[Hashable, list[dict]]: + groups: dict[Hashable, list[dict]] = {} for entry in rows: - groups.setdefault(entry[key], []).append(entry) + groups.setdefault(key(entry), []).append(entry) return groups def command_stats(rows: list[dict]) -> dict: - stats = { - "kinds": sorted({r["kind"] for r in rows}), "runs": len(rows), - "minutes": None, "disk_gb": None, "ram_share": None, "cpu_share": None, "need_gb": None, - } + stats = {"runs": len(rows), "minutes": None, "disk_gb": None, "ram_share": None, "cpu_share": None, "need_gb": None} if len(rows) >= MIN_RUNS: disk = p90([r["disk_gb"] for r in rows]) stats.update( @@ -179,49 +181,113 @@ def kind_stats(rows: list[dict], limits: dict) -> dict: return stats -def learn(rows: list[dict], now: float, cfg: dict) -> tuple[dict[str, dict], dict[str, dict]]: +def learn(rows: list[dict], now: float, cfg: dict) -> tuple[dict[tuple[str, str], dict], dict[str, dict]]: """Derive disk need per command and slot limit per kind from recent rows. Args: rows: history rows, as read() returns them. now: epoch seconds; rows that ended more than WINDOW_SECONDS earlier are ignored. - cfg: the config; load_limits.mem_percent and cpu_percent bound a limit. + cfg: the config; load_limits.mem_percent and cpu_percent bound a limit, + light_kinds marks a kind light. Returns: - (signatures, kinds). signatures: {signature: {kinds, runs, minutes, - disk_gb, ram_share, cpu_share, need_gb}}; kinds: {kind: {runs, ram_share, - cpu_share, limit}}. With fewer than MIN_RUNS rows every value but kinds - and runs is None. minutes is the median, disk_gb and ram_share the p90 - of the disk taken and of max minus before in percentage points, - cpu_share the median of max minus before; need_gb is ceil(disk_gb + DISK_MARGIN_GB); limit is + (signatures, kinds). signatures: {(kind, signature): {runs, minutes, + disk_gb, ram_share, cpu_share, need_gb}}, and {(ANY_KIND, signature): + ...} over the rows of every kind with that signature, which is what a + queue before the per-kind learning booked; kinds: {kind: {runs, + ram_share, cpu_share, limit, light}}. With fewer than MIN_RUNS rows + every value but runs and light is None. + minutes is the median, disk_gb and ram_share the p90 of the disk taken + and of max minus before in percentage points, cpu_share the median of + max minus before; need_gb is ceil(disk_gb + DISK_MARGIN_GB); limit is max(1, min over RAM and CPU of floor((load limit - median before) / max(share, MIN_SHARE_PERCENT))). """ recent = [r for r in rows if now - r["end"] <= WINDOW_SECONDS] - signatures = {name: command_stats(group) for name, group in grouped(recent, "signature").items()} - kinds = {name: kind_stats(group, cfg["load_limits"]) for name, group in grouped(recent, "kind").items()} + light = set(cfg.get("light_kinds", [])) + signatures = {(ANY_KIND, name): command_stats(group) for name, group in grouped(recent, NAME_KEY).items()} + signatures.update((key, command_stats(group)) for key, group in grouped(recent, SIGNATURE_KEY).items()) + kinds = { + name: {**kind_stats(group, cfg["load_limits"]), "light": name in light} + for name, group in grouped(recent, KIND_KEY).items() + } return signatures, kinds -def footprint(kind: str, name: str, signatures: dict[str, dict], kinds: dict[str, dict]) -> dict[str, float] | None: +def own(signatures: dict[tuple[str, str], dict], kind: str, name: str) -> dict | None: + stats = signatures.get((kind, name)) + return stats if stats is not None and stats["runs"] >= MIN_RUNS else None + + +def wider(signatures: dict[tuple[str, str], dict], aliases: tuple[str, ...]) -> list[dict]: + found = [signatures.get((ANY_KIND, name)) for name in aliases] + return [stats for stats in found if stats is not None and stats["runs"] >= MIN_RUNS] + + +def footprint(kind: str, aliases: tuple[str, ...], signatures: dict[tuple[str, str], dict], + kinds: dict[str, dict]) -> dict[str, float] | None: """Predict the RAM and CPU share a job takes, in percentage points. Args: kind: the job's kind. - name: its command signature. + aliases: its signature, then the signature rows before the runner + naming carry for the same command. signatures: the signature map of learn(). kinds: the kind map of learn(). Returns: - {ram, cpu}: the signature's learned share once learned, else the kind's, at - least MIN_SHARE_PERCENT. None while the kind is not learned yet. + {ram, cpu}: the signature's share under kind once MIN_RUNS rows exist + there, for a kind not light at least booked_before(); else the + largest of the kind's share and each alias's share over every kind, + so a shape without runs of its own never books less than its kind or + the runs it was learned from; at least MIN_SHARE_PERCENT. None while + the kind is not learned yet. """ if kind not in kinds or kinds[kind]["ram_share"] is None: return None - own = signatures.get(name) - source = own if own and own["ram_share"] is not None else kinds[kind] - return {resource: max(MIN_SHARE_PERCENT, source[f"{resource}_share"]) for resource in RESOURCES} + mine = own(signatures, kind, aliases[0]) + if mine is None: + return largest([kinds[kind], *wider(signatures, aliases)]) + return largest([mine] if kinds[kind]["light"] else [mine, booked_before(kind, aliases, signatures, kinds)]) + + +def booked_before(kind: str, aliases: tuple[str, ...], signatures: dict[tuple[str, str], dict], + kinds: dict[str, dict]) -> dict: + """Returns: + The stats a queue before the per-kind learning booked the job on: + its last alias over every kind once learned, else its kind's. + """ + pooled = wider(signatures, aliases[-1:]) + return pooled[0] if pooled else kinds[kind] + + +def ceiling(kind: str, aliases: tuple[str, ...], signatures: dict[tuple[str, str], dict], + kinds: dict[str, dict]) -> dict[str, float] | None: + """Returns: + The largest of every share footprint() can pick for the job: its + signature under kind once learned, its kind's and each alias's over + every kind, which holds booked_before(); None while the kind is not + learned yet. + """ + if kind not in kinds or kinds[kind]["ram_share"] is None: + return None + mine = own(signatures, kind, aliases[0]) + return largest([kinds[kind], *wider(signatures, aliases), *([mine] if mine is not None else [])]) + + +def largest(sources: list[dict]) -> dict[str, float]: + return {resource: max(MIN_SHARE_PERCENT, *(s[f"{resource}_share"] for s in sources)) for resource in RESOURCES} + + +def learned_value(signatures: dict[tuple[str, str], dict], kind: str, aliases: tuple[str, ...], + field: str) -> float | None: + """Returns: + field of the signature under kind once MIN_RUNS rows exist there, + else the largest field of the aliases over every kind, else None. + """ + mine = own(signatures, kind, aliases[0]) + return mine[field] if mine is not None else max((s[field] for s in wider(signatures, aliases)), default=None) def seed_ram(kinds: dict[str, dict]) -> float: @@ -232,8 +298,16 @@ def seed_ram(kinds: dict[str, dict]) -> float: return max((stats["ram_share"] for stats in kinds.values() if stats["ram_share"] is not None), default=0.0) -def need_gb(signatures: dict[str, dict], name: str) -> int | None: - return signatures[name]["need_gb"] if name in signatures else None +def need_gb(signatures: dict[tuple[str, str], dict], kinds: dict[str, dict], kind: str, + aliases: tuple[str, ...]) -> int | None: + """Returns: + The disk need learned_value() finds, None while nothing is learned + (slot_need_gb); for a kind not light at least that of the last alias + over every kind, as a queue before the per-kind learning booked. + """ + learned = learned_value(signatures, kind, aliases, "need_gb") + pooled = [] if kind in kinds and kinds[kind]["light"] else wider(signatures, aliases[-1:]) + return max(learned, pooled[0]["need_gb"]) if pooled else learned def slot_limits(kinds: dict[str, dict], caps: dict[str, int], names: Iterable[str]) -> dict[str, tuple[int, str]]: @@ -260,22 +334,24 @@ def slot_limits(kinds: dict[str, dict], caps: dict[str, int], names: Iterable[st return limits -def eta(signatures: dict[str, dict], running: list[tuple[str, float]], ahead: list[str], now: float) -> float | None: +def eta(signatures: dict[tuple[str, str], dict], running: list[tuple[tuple[str, tuple[str, ...]], float]], + ahead: list[tuple[str, tuple[str, ...]]], now: float) -> float | None: """Estimate the minutes until a waiting job starts, or until a running one ends. Args: - signatures: the signature map of learn(); its median minutes drive the - estimate. - running: (signature, grant epoch seconds) of the running jobs counted. - ahead: signatures of the waiting jobs before it. + signatures: the signature map of learn(); its median minutes + (learned_value()) drive the estimate. + running: ((kind, aliases), grant epoch seconds) of the running jobs + counted. + ahead: (kind, aliases) of the waiting jobs before it. now: epoch seconds. Returns: The remaining median minutes of running plus the median minutes of ahead, as if run one after another; None when one has no median yet. """ - def median(name: str) -> float | None: - return signatures[name]["minutes"] if name in signatures else None + def median(key: tuple[str, tuple[str, ...]]) -> float | None: + return learned_value(signatures, key[0], key[1], "minutes") parts = [(median(name), (now - granted) / 60) for name, granted in running] parts += [(median(name), 0.0) for name in ahead] @@ -293,7 +369,7 @@ def table(header: tuple[str, ...], rows: list[list[str]]) -> str: return "".join(f"| {' | '.join(line)} |\n" for line in lines) -def report(signatures: dict[str, dict], kinds: dict[str, dict], limits: dict[str, tuple[int, str]]) -> str: +def report(signatures: dict[tuple[str, str], dict], kinds: dict[str, dict], limits: dict[str, tuple[int, str]]) -> str: """Render the learned values as two markdown tables. Args: @@ -302,12 +378,13 @@ def report(signatures: dict[str, dict], kinds: dict[str, dict], limits: dict[str limits: slot_limits() for every kind to list. Returns: - A table per signature, a blank line, a table per kind. + A table per signature and kind, sorted by signature, a blank line, a + table per kind. """ commands = [ - [name, ", ".join(s["kinds"]), str(s["runs"]), cell(s["minutes"]), cell(s["disk_gb"]), + [name, kind, str(s["runs"]), cell(s["minutes"]), cell(s["disk_gb"]), cell(s["ram_share"]), "-" if s["need_gb"] is None else str(s["need_gb"])] - for name, s in sorted(signatures.items()) + for (kind, name), s in sorted(signatures.items(), key=lambda item: item[0][::-1]) if kind != ANY_KIND ] per_kind = [] for kind, (limit, source) in limits.items(): diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index b0ecb2c..0f7b9eb 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/skills/kvb-org-queue/scripts/queue.py @@ -80,7 +80,7 @@ also needs CPU busy below cpu_percent and RAM used plus the job's learned RAM share below mem_percent; both bound the learned slot limits), memory_limits (no grant at memory stage stop), heavy_commands and never_block are used as watch.py documents them. A signature is `: ` -(qualified()). +(aliases()). Exit codes: hold: the command's exit code unchanged, 128+N when signal N ended it; @@ -148,6 +148,12 @@ RULE_PART = re.compile(r"( \*|\*)") MANAGED = Path("/etc/claude-code") BUDGET_KEYS = ("budget_percent.ram", "budget_percent.cpu") HOTFIX = "hotfix/" +PYTEST = frozenset({"pytest", "py.test"}) +XDIST = frozenset({"-n", "--numprocesses"}) +XDIST_LONG = "--numprocesses=" +ADDOPTS = "PYTEST_ADDOPTS=" +FROZEN = "footprint" +ALIASES = "aliases" @functools.cache @@ -265,18 +271,99 @@ def script_tail(run: list[str]) -> list[str] | None: return None -def qualified(workdir: str, argv: list[str]) -> str: - """Returns: - `: `: the directory - holding the main checkout's .git, so every worktree of one repository - shares it, else the workdir's name; one make test of two repositories - stays apart. +def aliases(workdir: str, argv: list[str], cwd: str | None) -> tuple[str, ...]: + """Args: + workdir: the hold's workdir, which names the repository. + argv: the command as held. + cwd: the directory the hold runs it in, None when unknown + (runner_signature()). + + Returns: + `: `, then `: `, the + signature history rows before the runner naming carry, when it + differs. The repository is the directory holding the main checkout's + .git, so every worktree of one repository shares it, else the + workdir's name; one make test of two repositories stays apart. """ - return f"{repository(workdir)}: {history.signature(command_words(argv))}" + words = command_words(argv) + repo = repository(workdir) + plain = f"{repo}: {history.signature(words)}" + runner = runner_signature(words, argv, cwd) + return (plain,) if runner is None else tuple(dict.fromkeys((f"{repo}: {runner}", plain))) -def signature_of(entry: dict) -> str: - return qualified(entry["workdir"], shlex.split(entry["cmd"])) +def runner_signature(words: list[str], argv: list[str], cwd: str | None) -> str | None: + """Args: + words: the command, command_words() of argv. + argv: the command as held; a PYTEST_ADDOPTS= word in it adds its + options before the command's own. + cwd: the directory the hold runs it in, None when unknown; a + discover run then counts by its pattern alone + (watch.one_discovered()). + + Returns: + ` files=N` when each of the N tests a pytest or unittest run + names is one file or module, a discover pattern only when exactly + one file below its start directory matches it, else ` + paths=N` (0 for the whole suite), then ` workers=W` for the last + pytest-xdist count; None for other commands. One test file and a + suite never share what they learn. + """ + name, args = watch.tool(words) + named = tests_named(name, args) + if named is None: + return None + if name == "unittest" and args[:1] == ["discover"]: + prefix = argv[: argv.index(words[0])] if words[0] in argv else [] + files = watch.one_discovered(args[1:], watch.started_in(prefix, cwd)) + else: + files = bool(named) and all(name == "unittest" or watch.TEST_FILE.fullmatch(test) for test in named) + shape = f"{'unittest' if name == 'unittest' else 'pytest'} {'files' if files else 'paths'}={len(named)}" + count = workers([*addopts(argv), *args]) if name in PYTEST else None + return shape if count is None else f"{shape} workers={count}" + + +def addopts(argv: list[str]) -> list[str]: + found = [word[len(ADDOPTS):] for word in argv if word.startswith(ADDOPTS)] + return found[-1].split() if found else [] + + +def tests_named(name: str, args: list[str]) -> list[str] | None: + if name in PYTEST: + return watch.operands(args, watch.PYTEST_VALUE_OPTIONS) + if name != "unittest": + return None + if args[:1] == ["discover"]: + return watch.discover_patterns(args[1:]) + return watch.operands(args, watch.UNITTEST_VALUE_OPTIONS) + + +def workers(args: list[str]) -> str | None: + found = None + for at, word in enumerate(args): + if word in XDIST and at + 1 < len(args): + found = args[at + 1] + elif word.startswith(XDIST_LONG): + found = word[len(XDIST_LONG):] + elif word.startswith("-n") and not word.startswith("--") and len(word) > 2: + found = word[2:].removeprefix("=") + return found + + +def aliases_of(entry: dict) -> tuple[str, ...]: + """Returns: + The aliases the holder stored at enqueue, in the directory the + command starts in, else (an entry of older code, whose directory is + unknown) those derived here. + """ + if ALIASES in entry: + return tuple(entry[ALIASES]) + return aliases(entry["workdir"], shlex.split(entry["cmd"]), None) + + +def key_of(entry: dict) -> tuple[str, tuple[str, ...]]: + return entry["kind"], aliases_of(entry) def identity(pid: int) -> str | None: @@ -401,7 +488,7 @@ def evaluate(entries: list[dict], free_gb: float, floor: float, stage: str, memo silenced: whether a radio-silence mayday holds. slot_limits: {kind: max concurrent jobs}, an upper cap beside the budget. footprints: {entry id: {ram, cpu}}, the learned share of every entry - (history.footprint), in percentage points. + (footprint(), an active one's as granted), in percentage points. budget: {ram, cpu}: the sum of the footprints of active and granted entries may not pass it; a job alone always fits. ranks: {task: index in the register's ranking}; a task missing there @@ -560,8 +647,19 @@ def silenced(now: float) -> bool: return "until" in state and state["until"] > now -def footprint(entry: dict, signatures: dict[str, dict], kinds: dict[str, dict]) -> dict[str, float] | None: - return history.footprint(entry["kind"], signature_of(entry), signatures, kinds) +def footprint(entry: dict, signatures: dict[tuple[str, str], dict], kinds: dict[str, dict]) -> dict[str, float] | None: + """Returns: + The footprint an active entry was granted on (its footprint field, + so what history learns later never moves a running job's + reservation); history.ceiling of an active entry an older queue + granted without one, so it never books less than that queue did; + else history.footprint of its aliases. + """ + if entry["status"] == "active": + if FROZEN in entry: + return entry[FROZEN] + return history.ceiling(entry["kind"], aliases_of(entry), signatures, kinds) + return history.footprint(entry["kind"], aliases_of(entry), signatures, kinds) def held_back(cfg: dict, entries: list[dict], now: float, signatures: dict[str, dict], kinds: dict[str, dict], @@ -702,7 +800,7 @@ def wait(cfg: dict, entry_id: str, caught: list[int]) -> tuple[int | None, int]: mine[0]["reason"] = reason if reason is None: concurrent = sum(e["status"] == "active" for e in data["entries"]) - mine[0].update(status="active", granted=now) + mine[0].update(status="active", granted=now, **{FROZEN: footprint(mine[0], signatures, kinds)}) return None, concurrent position = list(verdicts).index(entry_id) + 1 if (position, reason) != shown: @@ -766,11 +864,11 @@ def cmd_hold(args: argparse.Namespace) -> int: if args.hotfix is not None and HOTFIX + args.hotfix not in rasci.load(): fail(f"the register has no row {HOTFIX}{args.hotfix}") here = workdir() - signature = qualified(here, args.command) - signatures, _ = learned(cfg, time.time()) + names = aliases(here, args.command, os.getcwd()) + signatures, kinds = learned(cfg, time.time()) need = args.need_gb if need is None: - need = history.need_gb(signatures, signature) + need = history.need_gb(signatures, kinds, args.kind, names) if need is None: need = cfg["slot_need_gb"][args.kind] if need < 0: @@ -794,7 +892,7 @@ def cmd_hold(args: argparse.Namespace) -> int: "workdir": here, "session": session, "name": council.live_sessions().get(session) or session, "lock": tag, "code": CODE, "pid": os.getpid(), "pid_start": identity(os.getpid()), "child": None, "child_start": None, "status": "waiting", "enqueued": time.time(), "granted": None, - "cmd": shlex.join(args.command), + "cmd": shlex.join(args.command), ALIASES: list(names), }) try: code, concurrent = wait(cfg, entry["id"], caught) @@ -814,7 +912,7 @@ def cmd_hold(args: argparse.Namespace) -> int: update(entry["id"], child=child.pid, child_start=identity(child.pid)) code = supervise(cfg, child, samples) code = code if code >= 0 else 128 - code - job = {"id": f"{entry['id']}-{time.time_ns()}", "kind": args.kind, "signature": signature, + job = {"id": f"{entry['id']}-{time.time_ns()}", "kind": args.kind, "signature": names[0], "workdir": entry["workdir"]} record(job, start, code, samples, concurrent) return code @@ -832,17 +930,17 @@ def cmd_list(_: argparse.Namespace) -> int: entries = list(data["entries"]) verdicts = decide(cfg, entries, free, now, signatures, kinds, {}) by_id = {e["id"]: e for e in entries} - running = [(signature_of(e), e["granted"]) for e in entries if e["status"] == "active"] + running = [(key_of(e), e["granted"]) for e in entries if e["status"] == "active"] rows = [ - ("run", e, history.eta(signatures, [(signature_of(e), e["granted"])], [], now), "-") + ("run", e, history.eta(signatures, [(key_of(e), e["granted"])], [], now), "-") for e in entries if e["status"] == "active" ] - ahead: list[str] = [] + ahead: list[tuple[str, str]] = [] for n, (entry_id, reason) in enumerate(verdicts.items(), 1): e = by_id[entry_id] polled = reason if silent(e, now) else e.get("reason") or reason rows.append((str(n), e, history.eta(signatures, running, ahead, now), polled or "granted on its next poll")) - ahead.append(signature_of(e)) + ahead.append(key_of(e)) host = code_of(HOST_SKILLS) or CODE print("position\tid\tstatus\tkind\ttask\tneed_gb\twaited_min\teta_min\tworkdir\treason") for position, e, eta, reason in rows: @@ -964,7 +1062,7 @@ def cmd_block() -> int: if cfg is None or cfg.get("queue_enforce") is not True: return 0 gated = {**cfg, "heavy_commands": [*cfg["heavy_commands"], GATE.pattern]} - if watch.heavy(raw, gated, (THROUGH_QUEUE.pattern, *cfg["never_block"])): + if watch.heavy(raw, gated, (THROUGH_QUEUE.pattern, *cfg["never_block"]), watch.payload_cwd(payload)): print( f"kvb-org-queue: heavy jobs run through the queue: start `python3 {Path(__file__).resolve()} hold " "--kind --task -- ` in the background; it runs the command on its grant.", diff --git a/tests/test_queue.py b/tests/test_queue.py index 8d6c6cf..622f814 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -155,12 +155,15 @@ class TestEvaluate(unittest.TestCase): "GIT_COMMITTER_EMAIL": "t@x", "GIT_CONFIG_NOSYSTEM": "1"} for args in (["init", "-q"], ["commit", "-q", "--allow-empty", "-m", "x"], ["worktree", "add", "-q", str(base / "wt")]): subprocess.run(["git", *args], cwd=main, env=identity, check=True, capture_output=True) - self.assertEqual(self.queue.qualified(str(main), ["make", "test"]), "skills: make test") - self.assertEqual(self.queue.qualified(str(base / "wt"), ["make", "test"]), "skills: make test") + def qualified(workdir: Path, argv: list[str]) -> str: + return self.queue.aliases(str(workdir), argv, None)[0] + + self.assertEqual(qualified(main, ["make", "test"]), "skills: make test") + self.assertEqual(qualified(base / "wt", ["make", "test"]), "skills: make test") plain = base / "plain" plain.mkdir() - self.assertEqual(self.queue.qualified(str(plain), ["true"]), "plain: true") - self.assertEqual(self.queue.qualified(str(base / "gone"), ["true"]), "gone: true") + self.assertEqual(qualified(plain, ["true"]), "plain: true") + self.assertEqual(qualified(base / "gone", ["true"]), "gone: true") for argv, expected in { ("env", "TMPDIR=/t", "-i", "make", "test"): "make test", ("nice", "-n", "10", "make", "compose-deploy"): "make compose-deploy", @@ -174,9 +177,94 @@ class TestEvaluate(unittest.TestCase): ("bash", "-c", "source .venv/bin/activate && make test"): "make test", }.items(): with self.subTest(argv=argv): - self.assertEqual(self.queue.qualified(str(plain), list(argv)), f"plain: {expected}") + self.assertEqual(qualified(plain, list(argv)), f"plain: {expected}") module = ["python3", "-m", "pytest", "--", "tests/"] self.assertEqual(self.queue.command_words(module), module) + single = ["env", "PYTHONDONTWRITEBYTECODE=1", "/v/bin/python", "-m", "pytest", "-q", "-p", "no:cacheprovider", + "tests/unit/test_x.py"] + self.assertEqual(qualified(plain, single), "plain: pytest files=1") + self.assertEqual(self.queue.aliases(str(plain), single, None), + ("plain: pytest files=1", "plain: pytest no:cacheprovider")) + self.assertEqual(self.queue.aliases(str(plain), ["make", "test"], None), ("plain: make test",)) + + def test_a_test_run_is_named_by_the_files_or_paths_it_names_and_its_xdist_workers(self): + nine = [f"tests/unit/test_{n}.py" for n in range(9)] + with tempfile.TemporaryDirectory() as tmp: + for name in ("tests/test_x.py", "tree/a/test_main.py", "tree/b/test_main.py", "a/x/test_main.py", + "a/y/test_main.py"): + (Path(tmp) / name).parent.mkdir(parents=True, exist_ok=True) + (Path(tmp) / name).write_text("", encoding="utf-8") + + def runner(argv: list[str], cwd: str | None) -> str | None: + return self.queue.runner_signature(self.queue.command_words(argv), argv, cwd) + + discover = ["python3", "-m", "unittest", "discover"] + self.assertEqual(runner([*discover, "-s", "tests", "-p", "test_x.py"], tmp), "unittest files=1") + self.assertEqual(runner(["env", "-C", "tree", "env", "--chdir=a", *discover, "-p", "test_main.py"], tmp), + "unittest files=1") + self.assertEqual(runner([*discover, "-s", "tree", "-p", "test_main.py"], tmp), "unittest paths=1") + self.assertEqual(runner([*discover, "-s", "tree", "-p", "test_main.py"], None), "unittest files=1") + self.assertEqual(runner([*discover, "-s", "nowhere", "-p", "test_main.py"], tmp), "unittest files=1") + self.assertEqual(runner([*discover, "-s", "nowhere", "-p", "test_*.py"], tmp), "unittest paths=1") + there = ("env", "-C", tmp, "python3", "-m", "unittest", "discover") + for argv, expected in { + ("python3", "-m", "pytest", "-q", "-p", "no:cacheprovider", "tests/unit/test_x.py"): "pytest files=1", + ("pytest", "tests/unit/test_x.py::TestA::test_b", "-k", "a or b"): "pytest files=1", + ("python3", "-m", "pytest", "-q", "-p", "no:cacheprovider", *nine): "pytest files=9", + ("python3", "-m", "pytest", "tests/unit", "tests/integration", "-p", "no:cacheprovider"): "pytest paths=2", + ("pytest", "tests/unit/test_x.py", "tests/lint"): "pytest paths=2", + ("python3", "-m", "pytest", "-q"): "pytest paths=0", + ("pytest", "-n", "auto", "tests/x.py"): "pytest files=1 workers=auto", + ("pytest", "-n4", "tests/"): "pytest paths=1 workers=4", + ("pytest", "-n=4", "tests/x.py"): "pytest files=1 workers=4", + ("pytest", "--numprocesses", "6", "tests/x.py"): "pytest files=1 workers=6", + ("py.test", "--numprocesses=8"): "pytest paths=0 workers=8", + ("env", "PYTEST_ADDOPTS=-q -n 4", "pytest", "tests/x.py"): "pytest files=1 workers=4", + ("env", "PYTEST_ADDOPTS=-n 4", "pytest", "-n", "2", "tests/x.py"): "pytest files=1 workers=2", + (*there, "-s", "tests", "-p", "test_x.py"): "unittest files=1", + (*there, "--start-directory", "tests", "--pattern", "test_x.py"): "unittest files=1", + (*there, "-s", "tree", "-p", "test_main.py"): "unittest paths=1", + (*there, "-s", "tests", "-p", "test_gone.py"): "unittest paths=1", + (*there, "-s", "tests", "-p", "test_*.py"): "unittest files=1", + (*there, "-s", "tree", "-p", "test_*.py"): "unittest paths=1", + (*there, "-s", "tests"): "unittest paths=0", + ("python3", "-m", "unittest", "-v", "tests.test_queue", "tests.test_watch"): "unittest files=2", + ("python3", "-m", "unittest"): "unittest paths=0", + }.items(): + with self.subTest(argv=argv): + self.assertEqual(runner(list(argv), "/"), expected) + for argv in (["make", "test"], ["python3", "/s/gate.py", "run"], ["python3", "-c", "print(1)"], ["true"]): + with self.subTest(argv=argv): + self.assertIsNone(self.queue.runner_signature(argv, argv, "/")) + + def test_an_active_entry_keeps_the_footprint_it_was_granted_on_while_history_moves(self): + history = _load(HISTORY) + rows = [history_row(n, signature="w: make test", mem_max=70, cpu_max=90) for n in range(5)] + signatures, kinds = history.learn(rows, NOW, {"load_limits": {"mem_percent": 90, "cpu_percent": 90}}) + granted = {"ram": 12.0, "cpu": 30.0} + running = {**job(1, status="active", granted=NOW - 60), "workdir": "/w", "cmd": "make test", "footprint": granted} + self.assertEqual(self.queue.footprint(running, signatures, kinds), granted) + unlearned = {**running, "footprint": None} + self.assertIsNone(self.queue.footprint(unlearned, signatures, kinds)) + waiting = {**running, "status": "waiting"} + self.assertEqual(self.queue.footprint(waiting, signatures, kinds), {"ram": 50, "cpu": 80}) + older = {k: v for k, v in running.items() if k != "footprint"} + self.assertEqual(self.queue.footprint(older, signatures, kinds), {"ram": 50, "cpu": 80}) + self.assertEqual(self.queue.pending_ram([running], "j1", NOW, signatures, kinds), 12 + 12) + + def test_an_active_entry_an_older_queue_granted_never_books_less_than_that_queue_did(self): + history = _load(HISTORY) + light = [history_row(n, kind="test-nostack", signature="w: gate.py hold", mem_max=23, cpu_max=22) for n in range(5)] + heavy = [history_row(n + 5, kind="quality-high", signature="w: gate.py hold", mem_max=60, cpu_max=80) for n in range(5)] + signatures, kinds = history.learn([*light, *heavy], NOW, {"load_limits": {"mem_percent": 90, "cpu_percent": 90}, + "light_kinds": ["test-nostack"]}) + landed = {**job(1, kind="test-nostack", status="active", granted=NOW - 60), "workdir": "/w", + "cmd": "python3 /x/gate.py hold --repo ."} + self.assertEqual(self.queue.footprint(landed, signatures, kinds), {"ram": 40, "cpu": 41}) + self.assertEqual(self.queue.footprint({**landed, "status": "waiting"}, signatures, kinds), {"ram": 3, "cpu": 12}) + self.assertEqual(self.queue.footprint({**landed, "footprint": {"ram": 3, "cpu": 12}}, signatures, kinds), + {"ram": 3, "cpu": 12}) + self.assertIsNone(self.queue.footprint({**landed, "kind": "gui-dind"}, signatures, kinds)) def test_memory_stage_stop_blocks_every_job_and_warn_none(self): entries = [job(1, need_gb=0), job(2, task="work/b")] @@ -362,7 +450,8 @@ class TestHistory(unittest.TestCase): cls.history = _load(HISTORY) def learn(self, rows: list[dict]) -> tuple[dict, dict]: - return self.history.learn(rows, NOW, {"load_limits": {"mem_percent": 80, "cpu_percent": 90}}) + return self.history.learn(rows, NOW, {"load_limits": {"mem_percent": 80, "cpu_percent": 90}, + "light_kinds": ["test-nostack"]}) def test_signature_names_a_command_apart_from_paths_options_and_variables(self): cases = { @@ -385,24 +474,96 @@ class TestHistory(unittest.TestCase): def test_learn_waits_for_min_runs_inside_the_30_day_window(self): four = [history_row(n) for n in range(4)] signatures, kinds = self.learn(four) - self.assertEqual((signatures["make test"]["runs"], signatures["make test"]["need_gb"]), (4, None)) + self.assertEqual((signatures["test", "make test"]["runs"], signatures["test", "make test"]["need_gb"]), (4, None)) self.assertEqual((kinds["test"]["runs"], kinds["test"]["limit"]), (4, None)) stale = history_row(4, end=NOW - self.history.WINDOW_SECONDS - 1) - self.assertIsNone(self.learn([*four, stale])[0]["make test"]["need_gb"]) + self.assertIsNone(self.learn([*four, stale])[0]["test", "make test"]["need_gb"]) signatures, kinds = self.learn([*four, history_row(4, end=NOW - self.history.WINDOW_SECONDS)]) - self.assertEqual((signatures["make test"]["need_gb"], signatures["make test"]["minutes"]), (2, 10)) + self.assertEqual((signatures["test", "make test"]["need_gb"], signatures["test", "make test"]["minutes"]), (2, 10)) self.assertEqual(kinds["test"]["limit"], 2) def test_learn_needs_the_p90_disk_plus_one_gb_and_takes_the_median_minutes(self): runs = zip((4, 1, 10, 3, 2), (5, 9, 7, 1, 30)) - stats = self.learn([history_row(n, disk_gb=gb, minutes=m) for n, (gb, m) in enumerate(runs)])[0]["make test"] + rows = [history_row(n, disk_gb=gb, minutes=m) for n, (gb, m) in enumerate(runs)] + stats = self.learn(rows)[0]["test", "make test"] self.assertAlmostEqual(stats["disk_gb"], 7.6) self.assertEqual((stats["need_gb"], stats["minutes"]), (9, 7)) + def test_a_signature_learns_per_kind_once_the_kind_has_its_own_runs_and_over_every_kind_until_then(self): + heavy = [history_row(n, mem_max=60, cpu_max=95, disk_gb=9) for n in range(5)] + light = [history_row(n + 5, kind="test-nostack", mem_max=21, cpu_max=12, disk_gb=0) for n in range(5)] + signatures, kinds = self.learn([*heavy, *light]) + footprint, need, tests = self.history.footprint, self.history.need_gb, ("make test",) + self.assertEqual(footprint("test-nostack", tests, signatures, kinds), {"ram": 1, "cpu": 2}) + self.assertEqual(footprint("test", tests, signatures, kinds), {"ram": 40, "cpu": 85}) + self.assertEqual((need(signatures, kinds, "test-nostack", tests), need(signatures, kinds, "test", tests)), (1, 10)) + self.assertEqual(need(signatures, kinds, "deploy", tests), 10) + self.assertIsNone(need(signatures, kinds, "test", ("make lint",))) + other = [history_row(n + 10, kind="test-nostack", signature="true", mem_max=20, cpu_max=10) for n in range(5)] + signatures, kinds = self.learn([*heavy, *light[:4], *other]) + self.assertEqual(footprint("test-nostack", tests, signatures, kinds), {"ram": 40, "cpu": 85}) + + def test_a_shape_below_min_runs_books_the_largest_of_its_kind_and_every_alias_so_it_never_under_reserves(self): + heavy = [history_row(n, mem_max=60, cpu_max=95) for n in range(2)] + light = [history_row(n + 2, kind="test-nostack", mem_max=21, cpu_max=12) for n in range(5)] + lint = [history_row(n + 7, signature="make lint", mem_max=50, cpu_max=60) for n in range(5)] + signatures, kinds = self.learn([*heavy, *light, *lint]) + self.assertEqual(self.history.footprint("test", ("make test",), signatures, kinds), {"ram": 40, "cpu": 50}) + heavy = [history_row(n, mem_max=60, cpu_max=95) for n in range(5)] + tiny = [history_row(n + 5, kind="test-nostack", signature="true", mem_max=21, cpu_max=11) for n in range(5)] + signatures, kinds = self.learn([*heavy, *light[:2], *tiny]) + self.assertEqual(self.history.footprint("test-nostack", ("make test",), signatures, kinds), {"ram": 40, "cpu": 85}) + + def test_a_new_signature_without_runs_learns_from_the_rows_older_queues_wrote_under_its_old_signature(self): + old, new = "core: pytest no:cacheprovider", "core: pytest files=1" + legacy = [history_row(n, kind="test-nostack", signature=old, mem_max=35, cpu_max=70, disk_gb=4) for n in range(5)] + tiny = [history_row(n + 5, kind="test-nostack", signature="true", mem_max=21, cpu_max=11, disk_gb=0) for n in range(5)] + signatures, kinds = self.learn([*legacy, *tiny]) + footprint, need = self.history.footprint, self.history.need_gb + self.assertEqual(footprint("test-nostack", (new, old), signatures, kinds), {"ram": 15, "cpu": 60}) + self.assertEqual(need(signatures, kinds, "test-nostack", (new, old)), 5) + self.assertEqual(footprint("test-nostack", (new,), signatures, kinds)["ram"], kinds["test-nostack"]["ram_share"]) + fresh = [history_row(n + 10, kind="test-nostack", signature=new, legacy_signature=old, mem_max=21, cpu_max=12, + disk_gb=0) for n in range(5)] + signatures, kinds = self.learn([*legacy, *tiny, *fresh]) + self.assertEqual(footprint("test-nostack", (new, old), signatures, kinds), {"ram": 1, "cpu": 2}) + self.assertEqual(need(signatures, kinds, "test-nostack", (new, old)), 1) + self.assertEqual(signatures[self.history.ANY_KIND, old]["runs"], 5) + + def test_a_heavy_kind_never_books_less_than_the_queue_before_the_per_kind_learning_and_a_light_one_may(self): + gate, plain, suite = "w: gate.py hold", "core: pytest no:cacheprovider", "core: pytest paths=1" + shared = [history_row(n, signature=gate, mem_max=40, cpu_max=30, disk_gb=2) for n in range(5)] + shared += [history_row(n + 5, kind="quality-high", signature=gate, mem_max=80, cpu_max=70, disk_gb=9) + for n in range(5)] + diluted = [history_row(n, signature=plain, mem_max=50, cpu_max=40, disk_gb=6) for n in range(5)] + diluted += [history_row(n + 5, signature="core: make lint", mem_max=25, cpu_max=15, disk_gb=0) for n in range(50)] + diluted += [history_row(n + 55, kind="test-nostack", signature="core: pytest files=1", legacy_signature=plain, + mem_max=21, cpu_max=11, disk_gb=0) for n in range(60)] + booked_at_7c0872e = [ + (shared, "test", (gate,), {"ram": 60, "cpu": 40}, 10), + (shared, "quality-high", (gate,), {"ram": 60, "cpu": 40}, 10), + (diluted, "test", (suite, plain), {"ram": 30, "cpu": 30}, 7), + ] + for rows, kind, aliases, old, old_need in booked_at_7c0872e: + with self.subTest(kind=kind, aliases=aliases): + signatures, kinds = self.learn(rows) + for booked in (self.history.footprint, self.history.ceiling): + new = booked(kind, aliases, signatures, kinds) + self.assertTrue(all(new[resource] >= old[resource] for resource in old), (booked.__name__, new)) + self.assertGreaterEqual(self.history.need_gb(signatures, kinds, kind, aliases), old_need) + signatures, kinds = self.learn(diluted) + light = ("core: pytest files=1", plain) + self.assertEqual(self.history.footprint("test-nostack", light, signatures, kinds), {"ram": 1, "cpu": 1}) + self.assertEqual(self.history.need_gb(signatures, kinds, "test-nostack", light), 1) + xdist = "core: pytest paths=1 workers=auto" + signatures, kinds = self.learn([history_row(n, signature=xdist, disk_gb=25) for n in range(5)]) + self.assertEqual(self.history.need_gb(signatures, kinds, "test", (xdist, "core: pytest auto")), 26) + self.assertIsNone(self.history.need_gb(signatures, kinds, "test", (suite, plain))) + def test_a_neighbours_cpu_spike_in_one_row_does_not_lift_the_learned_cpu_share(self): rows = [*(history_row(n, cpu_before=10, cpu_max=12) for n in range(4)), history_row(4, cpu_before=10, cpu_max=90)] signatures, kinds = self.learn(rows) - self.assertEqual((signatures["make test"]["cpu_share"], kinds["test"]["cpu_share"]), (2, 2)) + self.assertEqual((signatures["test", "make test"]["cpu_share"], kinds["test"]["cpu_share"]), (2, 2)) self.assertEqual(kinds["test"]["ram_share"], 10) def test_learn_fits_the_slot_limit_between_the_median_load_and_the_load_limits(self): @@ -418,14 +579,16 @@ class TestHistory(unittest.TestCase): self.assertEqual(self.learn(crowded)[1]["test"]["limit"], 1) def test_a_light_command_of_a_heavy_kind_and_an_unlearned_one_are_not_booked_at_the_heaviest_share(self): + light, heavy = {"runs": 5, "ram_share": 3.3, "cpu_share": 4}, {"runs": 5, "ram_share": 42.6, "cpu_share": 93.5} signatures = { - "skills: make test": {"ram_share": 3.3, "cpu_share": 4, "kinds": {"test"}}, - "gate.py hold": {"ram_share": 42.6, "cpu_share": 93.5, "kinds": {"test", "test-nostack"}}, + ("test", "skills: make test"): light, (self.history.ANY_KIND, "skills: make test"): light, + ("test", "gate.py hold"): heavy, (self.history.ANY_KIND, "gate.py hold"): heavy, } - kinds = {"test": {"ram_share": 35.2, "cpu_share": 83.6}, "test-nostack": {"ram_share": 5.6, "cpu_share": 73.8}} + kinds = {"test": {"ram_share": 35.2, "cpu_share": 83.6, "light": False}, + "test-nostack": {"ram_share": 5.6, "cpu_share": 73.8, "light": True}} footprint = self.history.footprint - self.assertEqual(footprint("test", "skills: make test", signatures, kinds), {"ram": 3.3, "cpu": 4}) - self.assertEqual(footprint("test-nostack", "core: docker build", signatures, kinds), {"ram": 5.6, "cpu": 73.8}) + self.assertEqual(footprint("test", ("skills: make test",), signatures, kinds), {"ram": 3.3, "cpu": 4}) + self.assertEqual(footprint("test-nostack", ("core: docker build",), signatures, kinds), {"ram": 5.6, "cpu": 73.8}) def test_footprint_takes_the_learned_signature_else_the_kind_at_least_the_minimum_and_none_for_a_new_kind(self): small = [history_row(n, signature="make test", mem_max=25, cpu_max=20) for n in range(5)] @@ -436,13 +599,14 @@ class TestHistory(unittest.TestCase): idle = [history_row(n + 10, kind="idle", signature="true", mem_max=20, cpu_max=5) for n in range(5)] signatures, kinds = self.learn([*small, *big, *idle]) footprint = self.history.footprint - self.assertEqual(footprint("test", "make test", signatures, kinds), {"ram": 5, "cpu": 10}) - self.assertEqual(footprint("test", "make verify-test", signatures, kinds), {"ram": 60, "cpu": 80}) - self.assertEqual(footprint("test", "make unseen", signatures, kinds), {"ram": 5, "cpu": 10}) - self.assertEqual(footprint("idle", "true", signatures, kinds), {"ram": 1, "cpu": 1}) - self.assertIsNone(footprint("gui-dind", "make test-up", signatures, kinds)) + self.assertEqual(footprint("test", ("make test",), signatures, kinds), {"ram": 5, "cpu": 10}) + self.assertEqual(footprint("quality-high", ("make verify-test",), signatures, kinds), {"ram": 60, "cpu": 80}) + self.assertEqual(footprint("test", ("make verify-test",), signatures, kinds), {"ram": 60, "cpu": 80}) + self.assertEqual(footprint("test", ("make unseen",), signatures, kinds), {"ram": 5, "cpu": 10}) + self.assertEqual(footprint("idle", ("true",), signatures, kinds), {"ram": 1, "cpu": 1}) + self.assertIsNone(footprint("gui-dind", ("make test-up",), signatures, kinds)) signatures, kinds = self.learn(small[:4]) - self.assertIsNone(footprint("test", "make test", signatures, kinds)) + self.assertIsNone(footprint("test", ("make test",), signatures, kinds)) def test_slot_limits_take_the_learned_limit_else_the_seed_and_a_smaller_cap(self): kinds = {"deploy": {"limit": 3}, "build": {"limit": None}} @@ -477,10 +641,12 @@ class TestHistory(unittest.TestCase): def test_eta_adds_the_remaining_and_the_waiting_median_minutes(self): signatures = self.learn([history_row(n) for n in range(5)])[0] eta = self.history.eta - self.assertEqual(eta(signatures, [("make test", NOW - 240)], ["make test"], NOW), 16) - self.assertEqual(eta(signatures, [("make test", NOW - 900)], [], NOW), 0) + tests = ("test", ("make test",)) + self.assertEqual(eta(signatures, [(tests, NOW - 240)], [tests], NOW), 16) + self.assertEqual(eta(signatures, [(tests, NOW - 900)], [], NOW), 0) self.assertEqual(eta(signatures, [], [], NOW), 0) - self.assertIsNone(eta(signatures, [], ["make test", "true"], NOW)) + self.assertIsNone(eta(signatures, [], [tests, ("test", ("true",))], NOW)) + self.assertEqual(eta(signatures, [], [("deploy", ("make test",))], NOW), 10) class TestQueueCommands(unittest.TestCase): @@ -561,7 +727,11 @@ class TestQueueCommands(unittest.TestCase): ) def bash(self, command: str) -> int: - return self.hook(json.dumps({"tool_input": {"command": command}})).returncode + return self.bash_in(command, self.base) + + def bash_in(self, command: str, cwd: Path | None) -> int: + payload = {"tool_input": {"command": command}, **({} if cwd is None else {"cwd": str(cwd)})} + return self.hook(json.dumps(payload)).returncode def wait_for(self, condition: Callable[[], bool]) -> None: deadline = time.monotonic() + 30 @@ -581,11 +751,13 @@ class TestQueueCommands(unittest.TestCase): [entry[k] for k in ("id", "status", "kind", "task", "need_gb", "hotfix", "workdir", "session")], ["test-1", "active", "test", "work/a", 0.001, "fix-1", str(self.base), "s1"], ) + signature = f"{self.base.name}: {Path(sys.executable).name} -c" + self.assertEqual((entry["aliases"], entry["footprint"]), ([signature], None)) self.assertEqual(self.entries(), []) [row] = [json.loads(line) for line in self.history_file.read_text(encoding="utf-8").splitlines()] self.assertEqual( [row[k] for k in ("kind", "signature", "workdir", "exit", "concurrent")], - ["test", f"{self.base.name}: {Path(sys.executable).name} -c", str(self.base), 7, 0], + ["test", signature, str(self.base), 7, 0], ) self.assertTrue(row["id"].startswith("test-1-")) self.assertLessEqual(row["start"], row["end"]) @@ -885,7 +1057,15 @@ class TestQueueCommands(unittest.TestCase): self.cfg["queue_enforce"] = True self.cfg["heavy_commands"] = [] self.write_config() + for name in ("tests/test_queue.py", "tests/test_watch.py", "sub/tests/test_queue.py", "tree/a/test_main.py", + "tree/b/test_main.py"): + (self.base / name).parent.mkdir(parents=True, exist_ok=True) + (self.base / name).write_text("", encoding="utf-8") for command in ( + "python3 -m unittest discover -s tree -p test_main.py", + "python3 -m unittest discover -s tests -p test_gone.py", + "python3 -m unittest discover -s . -p test_queue.py", + "python3 -m unittest discover tree test_main.py", "docker compose up -d --build", "docker compose -f x.yml build web", "docker-compose up --build", @@ -979,6 +1159,8 @@ class TestQueueCommands(unittest.TestCase): "python3 /x/queue.py hold --kind test --task work/a -- pytest tests/", 'pytest "tests/unit/test_x.py"', "python3 -m unittest discover -s tests -p 'test_queue.py'", + "env -C sub python3 -m unittest discover -s tests -p test_queue.py", + f"env --chdir={self.base / 'sub'} python3 -m unittest discover -s tests -p 'test_q*.py'", "grep -rn pytest tests/", "which pytest", "pytest --version", @@ -1012,9 +1194,12 @@ class TestQueueCommands(unittest.TestCase): "bash script.sh <<'EOF'\nmake install\nEOF", "out=$(cat <<'EOF'\nmake install\nEOF\n)", "echo kvbheredoc1 kvbheredoczz", + "cd sub && python3 -m unittest discover -s . -p test_queue.py", ): with self.subTest(command=command): self.assertEqual(self.bash(command), 0) + self.assertEqual(self.bash_in("python3 -m unittest discover -s tests -p 'test_*.py'", self.base / "sub"), 0) + self.assertEqual(self.bash_in("python3 -m unittest discover -s tree -p test_main.py", None), 0) def test_block_matches_a_config_regex_on_the_wrapped_program_too(self): self.cfg["queue_enforce"] = True diff --git a/tests/test_watch.py b/tests/test_watch.py index 136a58d..ce0b57c 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -174,6 +174,18 @@ class TestWatchCommands(unittest.TestCase): self._stage("warn") self.assertEqual(self._run("block", stdin=self._bash("make compose-deploy")).returncode, 0) + def test_block_counts_discovered_files_in_the_payload_cwd_and_an_unusable_cwd_as_unknown(self): + self._stage("stop") + for name in ("tree/a/test_main.py", "tree/b/test_main.py"): + (self.base / name).parent.mkdir(parents=True, exist_ok=True) + (self.base / name).write_text("", encoding="utf-8") + command = {"command": "python3 -m unittest discover -s tree -p test_main.py"} + self.assertEqual(self._run("block", stdin=json.dumps({"tool_input": command, "cwd": str(self.base)})).returncode, 2) + for cwd in (str(self.base / "tree"), 5, ["/"], {"a": 1}, "tree", None): + with self.subTest(cwd=cwd): + blocked = self._run("block", stdin=json.dumps({"tool_input": command, "cwd": cwd})) + self.assertEqual((blocked.returncode, blocked.stderr), (0, "")) + def test_block_heavy_jobs_while_a_mayday_holds_radio_silence(self): self._stage("ok") self.root.mkdir(exist_ok=True)