Merge fix/kvb-org-queue-light-footprint into unify/skills-20261112

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

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.
 

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

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"]):

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.

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 <target>`, 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 <targets>`.
         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:

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 <gate.py> 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
 

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 `<repo>: <command
+signature>`, 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.

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():

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 `<repository>: <command signature>`
-(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:
-        `<repository>: <history.signature of command_words>`: 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:
+        `<repository>: <runner_signature or else history.signature of
+        command_words>`, then `<repository>: <history.signature>`, 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:
+        `<runner> 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 `<runner>
+        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 <kind> --task <register task> -- <command>` in the background; it runs the command on its grant.",

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

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)