fix(kvb-org-queue): book light jobs by their own footprint

Light single-file runs booked about 17 % RAM and 5 GB learned from heavy runs under a polluted signature. Each entry now stores its aliases, freezes its footprint at grant and learns per signature and kind; heavy kinds book at least the old pooled share, and a job keeps its own learned disk need when the old pool is empty. Hooks take the payload cwd and fall back to the old rule when it is unknown or malformed. Six skill docs now agree on what self-admits. test_queue 70, test_watch 46 and test_operator 30 green with negative controls; two independent reviews.

🎯 Fix: 90 %
💥 Break: 5 %
💸 Cost: 100 %
⏱️ Time: 100 %
🛡️ Security: 100 %
🏆 Score: 86
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

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)