feat(kvb-check-ansible): rank where an Ansible run spends its time

New skill with ansible_tasks.py: gaps, durations, kinds, kind and task rank a run's log by wall-clock gaps or profile_tasks lines; profile-report merges cProfile dumps of the controller and its forked workers (sitecustomize hook); bench-carrier and the bench playbook measure per-task overhead against variable size. kvb-code-tune hands its Ansible timing to it.

🎯 Fix: 92 %
💥 Break: 2 %
💸 Cost: new
⏱️ Time: new
🛡️ Security: 100 %
🏆 Score: 90
🚦 Gate: cost new, time new: cleared by the operator's commit word for kvb-check-ansible (2026-11-07)

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

README.md

@@ -68,6 +68,7 @@ flowchart LR
   check --> check_gate[kvb-check-gate]
   check --> check_efficiency[kvb-check-efficiency]
   check --> check_speed[kvb-check-speed]
+  check --> check_ansible[kvb-check-ansible]
   check --> check_score[kvb-check-score]
   org --> org_organize[kvb-org-organize]
   org --> org_split[kvb-org-split]
@@ -173,6 +174,7 @@ flowchart LR
     gate[kvb-check-gate]
     efficiency[kvb-check-efficiency]
     speed[kvb-check-speed]
+    check_ansible[kvb-check-ansible]
     score[kvb-check-score]
   end
   subgraph cat_org [org]
@@ -306,6 +308,8 @@ flowchart LR
   tune --> funnel
   tune --> merge_fix
   tune --> dialectic
+  tune --> check_ansible
+  check_ansible -. heavy commands .-> queue
 
   dialectic --> active_listening
   criticize --> dialectic

skills/kvb-check-ansible/SKILL.md

@@ -0,0 +1,86 @@
+---
+name: kvb-check-ansible
+description: >
+  Rank where an Ansible run spends its time from its log, profile the
+  controller, bench per-task overhead. Trigger on a slow playbook or deploy.
+---
+
+Where an Ansible run spends its time: its log, a controller profile, a bench.
+Report the script's lines with their log and range, never a guess. The
+script and the bench are light runs (`watch.py admit --kind test-nostack` of
+`kvb-org-organize`); a deploy to profile is heavy and goes through the
+`kvb-org-queue` skill.
+
+## 1. Pick the log
+
+- The newest log of the current code; an older one may time a task since
+  replaced.
+- A log whose lines start with an ISO timestamp (GitHub Actions) times every
+  task by wall clock. A local run needs
+  `ANSIBLE_CALLBACKS_ENABLED=ansible.posix.profile_tasks`, streamed to a file.
+- One deploy among several: bound it with `--from` and `--to`, line numbers
+  from `grep -n 'PLAY \['`.
+
+## 2. Rank
+
+```bash
+python3 ${CLAUDE_SKILL_DIR}/scripts/ansible_tasks.py gaps --log <log> --top 20
+```
+
+| Subcommand | Prints |
+| --- | --- |
+| `gaps` | wall seconds and count per task name, then `TOTAL` |
+| `durations` | the same from `profile_tasks` `(h:mm:ss.sss)` lines |
+| `kinds` | per kind (ran, skipped, included, other): count, sum, mean, buckets `<0.25` to `>=2` s |
+| `kind --kind <kind>` | per task name, that kind only |
+| `task --needle <text>` | every occurrence: start line, seconds, kind |
+
+`kinds`, `kind` and `task` read wall gaps when a header carries a timestamp,
+else `profile_tasks` lines; `--source wall|profile` forces one. A task runs
+from its header to the next header or `PLAY RECAP`. `--json` prints JSON;
+exit 2 means nothing measurable in the range.
+
+Rank by sum, a name's cost times its count: 400 skips of 0.3 s outweigh one
+60 s download.
+
+## 3. Profile the controller
+
+A skipped task does no work on any host; its time is controller overhead
+(templating, variable merging, one worker fork per task). Profile only when
+`kinds` shows it: a `skipped` mean of 0.1 s or more, or `skipped` and
+`included` together holding 10 % of the sum or more.
+
+```bash
+ANSIBLE_PROFILE_DIR=<new dir> PYTHONPATH=${CLAUDE_SKILL_DIR}/scripts/profile ansible-playbook <the run's arguments>
+python3 ${CLAUDE_SKILL_DIR}/scripts/ansible_tasks.py profile-report --dir <new dir> --top 30
+```
+
+- `sitecustomize.py` profiles only a process named `ansible-playbook`: the
+  controller into `main-<pid>.prof`, each forked worker into
+  `worker-<pid>.prof`; without `ANSIBLE_PROFILE_DIR` into
+  `$TMPDIR/ansible-profile`.
+- The report merges every file of the directory: a new one per run.
+- `PYTHONPATH` must reach the controller's interpreter (virtualenv,
+  container) and shadows a distribution's own `sitecustomize`.
+- A profiled run is slower: compare profiles with profiles, never its wall
+  clock with an unprofiled run.
+
+## 4. Bench the per-task overhead
+
+```bash
+python3 ${CLAUDE_SKILL_DIR}/scripts/ansible_tasks.py bench-carrier --kb <size> --out <dir>/carrier.json
+ANSIBLE_CALLBACKS_ENABLED=ansible.posix.profile_tasks ansible-playbook -i ${CLAUDE_SKILL_DIR}/bench/hosts.yml ${CLAUDE_SKILL_DIR}/bench/bench.yml -e bench_count=200 -e bench_carrier_file=<dir>/carrier.json > <dir>/bench-<size>.log 2>&1
+python3 ${CLAUDE_SKILL_DIR}/scripts/ansible_tasks.py kinds --log <dir>/bench-<size>.log
+```
+
+Each item includes one skipped and one run `debug` task on localhost. Run
+once without `bench_carrier_file` and once per `<size>` in KiB, up to the
+run's biggest variable; the change of the `skipped` mean is that fact's cost
+per task. Without `ansible.posix`, divide the wall-time difference of
+`bench_count=0` and `bench_count=200` by 400.
+
+## Report
+
+Per finding: subcommand, log and range, its line as printed, and the saving
+ceiling, count times the measured cost per task. Hand the ranking to the
+`kvb-code-tune` loop.

skills/kvb-check-ansible/bench/bench.yml

@@ -0,0 +1,12 @@
+- name: Bench the controller cost per task
+  hosts: bench
+  gather_facts: false
+  tasks:
+    - name: Park the synthetic fact
+      ansible.builtin.set_fact:
+        bench_carrier: "{{ lookup('ansible.builtin.file', bench_carrier_file) | from_json }}"
+      when: bench_carrier_file | default('') | length > 0
+
+    - name: Include one skipped and one run task per item
+      ansible.builtin.include_tasks: one.yml
+      loop: "{{ range(0, bench_count | int) | list }}"

skills/kvb-check-ansible/bench/hosts.yml

@@ -0,0 +1,5 @@
+all:
+  hosts:
+    bench:
+      ansible_connection: local
+      ansible_python_interpreter: "{{ ansible_playbook_python }}"

skills/kvb-check-ansible/bench/one.yml

@@ -0,0 +1,8 @@
+- name: Skipped
+  ansible.builtin.debug:
+    msg: never
+  when: false
+
+- name: Ran
+  ansible.builtin.debug:
+    msg: once

skills/kvb-check-ansible/scripts/ansible_bench.py

@@ -0,0 +1,52 @@
+"""Write the synthetic JSON fact the bench playbook parks on its host.
+
+Args:
+    size: exact size of the JSON text in bytes.
+    kb: exact size of the written file in KiB.
+    out: file the fact goes to.
+
+Output:
+    A dict of nested service entries, many small objects as real inventories carry them,
+    filled up to the exact size by one pad string. write returns a (report, lines) pair.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+ENTRY = {
+    "services": {
+        f"s{service}": {
+            "image": "x" * 20,
+            "version": "1.2.3",
+            "env": {f"K{key}": "v" * 10 for key in range(10)},
+        }
+        for service in range(5)
+    }
+}
+PAD = "pad"
+
+
+def carrier(size: int) -> dict:
+    value = len(json.dumps(ENTRY))
+    room = len(f', "{PAD}": ""')
+    fact: dict[str, object] = {}
+    used = len("{}")
+    while True:
+        key = f"entry-{len(fact)}"
+        grow = len(", ") * bool(fact) + len(json.dumps(key)) + len(": ") + value
+        if used + grow + room > size:
+            break
+        fact[key] = ENTRY
+        used += grow
+    fact[PAD] = "p" * (size - used - len(", ") * bool(fact) - len(f'"{PAD}": ""'))
+    return fact
+
+
+def write(kb: int, out: Path) -> tuple[dict, list[str]]:
+    fact = carrier(kb * 1024)
+    text = json.dumps(fact)
+    out.write_text(text, encoding="ascii")
+    report = {"out": str(out), "bytes": len(text), "entries": len(fact) - 1}
+    return report, [f"{report['out']}\t{report['bytes']} bytes\t{report['entries']} entries"]

skills/kvb-check-ansible/scripts/ansible_log.py

@@ -0,0 +1,170 @@
+"""Parse Ansible output into timed task occurrences.
+
+Args:
+    path: the Ansible output to read.
+    first, last: first and last line to read, 1-based and inclusive; None reads to that end.
+    source: wall (ISO timestamps) or profile (profile_tasks); None picks wall when a header
+        in range carries a timestamp, else profile.
+
+Output:
+    A task lasts from its TASK or RUNNING HANDLER header to the next header or PLAY RECAP;
+    profile_tasks prints that duration under the next header. Its kind is ran on an ok,
+    changed, failed or fatal result line, else skipped on a skipping line, else included
+    on an included line, else other. A stamp without zone counts as UTC. Unmeasurable
+    names what the range lacks.
+"""
+
+from __future__ import annotations
+
+import calendar
+import re
+from collections.abc import Iterator
+from pathlib import Path
+
+HEADER = re.compile(r"(?:TASK|RUNNING HANDLER) \[(?P<name>.*)\][\s*]*$")
+RECAP = re.compile(r"PLAY RECAP \*")
+STAMP = re.compile(
+    r"^(?P<day>\d{4}-\d{2}-\d{2})[T ](?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})"
+    r"(?:[.,](?P<fraction>\d+))?(?P<zone>Z|[+-]\d{2}:?\d{2})? ?"
+)
+ELAPSED = re.compile(r"\((?P<hours>\d+):(?P<minutes>\d+):(?P<seconds>\d+(?:\.\d+)?)\)")
+ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+RESULTS = (
+    ("ran", re.compile(r"^\s*(?:ok|changed|failed|fatal): \[")),
+    ("skipped", re.compile(r"^\s*skipping: \[")),
+    ("included", re.compile(r"^\s*included: ")),
+)
+OTHER = "other"
+KINDS = (*(kind for kind, _ in RESULTS), OTHER)
+SOURCES = ("wall", "profile")
+
+Rows = Iterator[tuple[int, float | None, str]]
+
+
+class Unmeasurable(Exception):
+    pass
+
+
+class NoStamps(Unmeasurable):
+    pass
+
+
+class Task:
+    def __init__(self, name: str, line: int, start: float | None) -> None:
+        self.name = name
+        self.line = line
+        self.start = start
+        self.seconds = 0.0
+        self.seen: set[str] = set()
+
+    def note(self, content: str) -> None:
+        self.seen.update(kind for kind, pattern in RESULTS if pattern.search(content))
+
+    @property
+    def kind(self) -> str:
+        return next((kind for kind, _ in RESULTS if kind in self.seen), OTHER)
+
+
+def offset(zone: str | None) -> int:
+    if zone is None or zone == "Z":
+        return 0
+    digits = zone[1:].replace(":", "")
+    seconds = int(digits[:2]) * 3600 + int(digits[2:]) * 60
+    return -seconds if zone.startswith("-") else seconds
+
+
+def epoch(match: re.Match[str]) -> float:
+    year, month, day = (int(part) for part in match["day"].split("-"))
+    whole = calendar.timegm((year, month, day, int(match["hour"]), int(match["minute"]), int(match["second"])))
+    fraction = float(f"0.{match['fraction']}") if match["fraction"] else 0.0
+    return whole + fraction - offset(match["zone"])
+
+
+def split_stamp(text: str) -> tuple[float | None, str]:
+    match = STAMP.match(text)
+    if match is None:
+        return None, text
+    return epoch(match), text[match.end():]
+
+
+def read(path: Path, first: int | None, last: int | None) -> Rows:
+    with path.open(encoding="utf-8-sig", errors="replace") as handle:
+        for number, raw in enumerate(handle, 1):
+            if first is not None and number < first:
+                continue
+            if last is not None and number > last:
+                return
+            stamp, content = split_stamp(ANSI.sub("", raw.rstrip("\r\n")))
+            yield number, stamp, content
+
+
+def boundary(content: str) -> tuple[bool, str | None]:
+    header = HEADER.search(content)
+    if header is not None:
+        return True, header["name"]
+    return RECAP.search(content) is not None, None
+
+
+def wall(rows: Rows) -> list[Task]:
+    done: list[Task] = []
+    current: Task | None = None
+    stamped = False
+    for number, stamp, content in rows:
+        found, name = boundary(content)
+        if not found:
+            if current is not None:
+                current.note(content)
+            continue
+        if stamp is None:
+            continue
+        stamped = True
+        if current is not None:
+            current.seconds = stamp - current.start
+            done.append(current)
+        current = None if name is None else Task(name, number, stamp)
+    if not stamped:
+        raise NoStamps("no TASK, RUNNING HANDLER or PLAY RECAP header with an ISO timestamp in range")
+    if not done:
+        raise Unmeasurable("no task closed by a later timestamped header in range")
+    return done
+
+
+def elapsed(match: re.Match[str]) -> float:
+    return int(match["hours"]) * 3600 + int(match["minutes"]) * 60 + float(match["seconds"])
+
+
+def profile(rows: Rows) -> list[Task]:
+    done: list[Task] = []
+    current: Task | None = None
+    waiting: Task | None = None
+    for number, _, content in rows:
+        found, name = boundary(content)
+        if found:
+            if current is not None:
+                waiting = current
+            current = None if name is None else Task(name, number, None)
+            continue
+        duration = ELAPSED.search(content)
+        if duration is not None and waiting is not None:
+            waiting.seconds = elapsed(duration)
+            done.append(waiting)
+            waiting = None
+        elif current is not None:
+            current.note(content)
+    if not done:
+        raise Unmeasurable(
+            "no profile_tasks (h:mm:ss.sss) duration under a header in range; "
+            "enable the ansible.posix.profile_tasks callback, or use gaps on a timestamped log"
+        )
+    return done
+
+
+def tasks(path: Path, first: int | None, last: int | None, source: str | None) -> tuple[str, list[Task]]:
+    if source == "profile":
+        return source, profile(read(path, first, last))
+    if source == "wall":
+        return source, wall(read(path, first, last))
+    try:
+        return "wall", wall(read(path, first, last))
+    except NoStamps:
+        return "profile", profile(read(path, first, last))

skills/kvb-check-ansible/scripts/ansible_profile.py

@@ -0,0 +1,74 @@
+"""Merge the main and worker cProfile files of an ansible-playbook run.
+
+Args:
+    directory: profile directory; the default of profile/sitecustomize.py when None.
+    top: functions per ranking.
+
+Output:
+    A (report, lines) pair: per role (main, worker) the profile count, own seconds and the
+    top functions by cumulative and by own time. Unmeasurable when the directory holds no
+    profile or an unreadable one.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import pstats
+from pathlib import Path
+
+import ansible_log
+
+PROFILER = Path(__file__).resolve().parent / "profile" / "sitecustomize.py"
+ROLES = ("main", "worker")
+TOTAL_TIME = 2
+CUMULATIVE_TIME = 3
+
+
+def default_dir() -> Path:
+    spec = importlib.util.spec_from_file_location("ansible_profile_site", PROFILER)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return Path(module.profile_dir())
+
+
+def hottest(stats: pstats.Stats, field: int, top: int) -> list[dict]:
+    rows = sorted(stats.stats.items(), key=lambda item: -item[1][field])[:top]
+    return [
+        {"function": pstats.func_std_string(func), "calls": calls, "tottime": round(own, 6), "cumtime": round(cumulative, 6)}
+        for func, (_, calls, own, cumulative, _) in rows
+    ]
+
+
+def merged(directory: Path, top: int) -> dict[str, dict]:
+    roles: dict[str, dict] = {}
+    for role in ROLES:
+        files = sorted(directory.glob(f"{role}-*.prof"))
+        if not files:
+            continue
+        try:
+            stats = pstats.Stats(*(str(file) for file in files))
+        except (EOFError, TypeError, ValueError) as error:
+            raise ansible_log.Unmeasurable(f"unreadable profile in {directory}: {error}") from None
+        roles[role] = {
+            "profiles": len(files),
+            "seconds": round(stats.total_tt, 3),
+            "cumulative": hottest(stats, CUMULATIVE_TIME, top),
+            "tottime": hottest(stats, TOTAL_TIME, top),
+        }
+    if not roles:
+        raise ansible_log.Unmeasurable(f"no main-*.prof or worker-*.prof in {directory}")
+    return roles
+
+
+def report(directory: Path | None, top: int) -> tuple[dict, list[str]]:
+    chosen = default_dir() if directory is None else directory
+    roles = merged(chosen, top)
+    lines: list[str] = []
+    for role, found in roles.items():
+        lines.append(f"== {role}: {found['profiles']} profiles, {found['seconds']:.3f} s ==")
+        for order in ("cumulative", "tottime"):
+            lines.append(f"-- {order}: cumtime, tottime, calls, function --")
+            lines += [
+                f"{row['cumtime']:.3f}\t{row['tottime']:.3f}\t{row['calls']}\t{row['function']}" for row in found[order]
+            ]
+    return {"dir": str(chosen), **roles}, lines

skills/kvb-check-ansible/scripts/ansible_tables.py

@@ -0,0 +1,103 @@
+"""Rank timed task occurrences into the reports of durations, gaps, kinds, kind and task.
+
+Args:
+    source: the source the occurrences came from, echoed in the JSON report.
+    found: the task occurrences of ansible_log.
+    top: rows per table; every row when None.
+    kind: ran, skipped, included or other.
+    needle: substring of the task name.
+
+Output:
+    A (report, lines) pair: the JSON-ready report and its tab-separated text, biggest
+    first. Name tables end in a TOTAL row over all their tasks; kind tables bucket each
+    occurrence into <0.25, <0.5, <1, <2 or >=2 s.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import ansible_log
+
+BUCKETS = ((0.25, "<0.25"), (0.5, "<0.5"), (1.0, "<1"), (2.0, "<2"))
+LAST_BUCKET = ">=2"
+LABELS = (*(label for _, label in BUCKETS), LAST_BUCKET)
+
+Found = list[ansible_log.Task]
+
+
+def bucket(seconds: float) -> str:
+    return next((label for limit, label in BUCKETS if seconds < limit), LAST_BUCKET)
+
+
+def ranked(rows: Iterable[dict]) -> list[dict]:
+    return sorted(rows, key=lambda row: -row["sum"])
+
+
+def of_kind(found: Found, kind: str) -> Found:
+    return [item for item in found if item.kind == kind]
+
+
+def names(source: str, found: Found, top: int | None) -> tuple[dict, list[str]]:
+    table: dict[str, dict] = {}
+    for item in found:
+        row = table.setdefault(item.name, {"name": item.name, "sum": 0.0, "count": 0})
+        row["sum"] += item.seconds
+        row["count"] += 1
+    rows = ranked(table.values())[:top]
+    total = sum(item.seconds for item in found)
+    report = {
+        "source": source,
+        "rows": [{**row, "sum": round(row["sum"], 3)} for row in rows],
+        "total": {"sum": round(total, 3), "count": len(found)},
+    }
+    lines = [f"{row['sum']:.1f}\t{row['count']}\t{row['name']}" for row in rows]
+    return report, [*lines, f"{total:.1f}\t{len(found)}\tTOTAL"]
+
+
+def kind_table(source: str, found: Found, top: int | None) -> tuple[dict, list[str]]:
+    table: dict[str, dict] = {}
+    for item in found:
+        row = table.setdefault(
+            item.kind, {"kind": item.kind, "count": 0, "sum": 0.0, "buckets": dict.fromkeys(LABELS, 0)}
+        )
+        row["count"] += 1
+        row["sum"] += item.seconds
+        row["buckets"][bucket(item.seconds)] += 1
+    rows = ranked(table.values())[:top]
+    report = {
+        "source": source,
+        "kinds": [
+            {**row, "sum": round(row["sum"], 3), "mean": round(row["sum"] / row["count"], 4)} for row in rows
+        ],
+    }
+    lines = ["\t".join(("kind", "count", "sum", "mean", *LABELS))]
+    lines += [
+        "\t".join(
+            (
+                row["kind"],
+                str(row["count"]),
+                f"{row['sum']:.1f}",
+                f"{row['sum'] / row['count']:.4f}",
+                *(str(row["buckets"][label]) for label in LABELS),
+            )
+        )
+        for row in rows
+    ]
+    return report, lines
+
+
+def occurrences(source: str, found: Found, needle: str, top: int | None) -> tuple[dict, list[str]]:
+    hits = sorted((item for item in found if needle in item.name), key=lambda item: -item.seconds)
+    shown = hits[:top]
+    report = {
+        "source": source,
+        "needle": needle,
+        "count": len(hits),
+        "sum": round(sum(item.seconds for item in hits), 3),
+        "occurrences": [
+            {"line": item.line, "seconds": round(item.seconds, 3), "kind": item.kind, "name": item.name}
+            for item in shown
+        ],
+    }
+    return report, [f"{item.line}\t{item.seconds:.3f}\t{item.kind}\t{item.name}" for item in shown]

skills/kvb-check-ansible/scripts/ansible_tasks.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""Measure where an Ansible run spends its time: from its log, its controller profiles or a bench.
+
+Subcommands:
+    durations: per task name sum and count of the profile_tasks `(h:mm:ss.sss)` durations.
+    gaps: per task name sum and count of the wall time between the headers of a log
+        whose lines start with an ISO timestamp.
+    kinds: per kind (ran, skipped, included, other) count, sum, mean and the buckets
+        <0.25, <0.5, <1, <2 and >=2 s.
+    kind: per task name sum and count of one --kind.
+    task: every occurrence of the task names holding --needle: start line, seconds, kind.
+    profile-report: the main and worker profiles of --dir merged, by cumulative and own time.
+    bench-carrier: a synthetic JSON fact of exactly --kb KiB written to --out.
+
+Args:
+    --log: the Ansible output to read.
+    --from, --to: first and last line to read, 1-based and inclusive; the whole log when omitted.
+    --top: rows per table; every row when omitted for a log, required for profile-report.
+    --json: JSON instead of tab-separated text.
+    --source: wall (ISO timestamps) or profile (profile_tasks) for kinds, kind and task;
+        wall when a header in range carries a timestamp, else profile, when omitted.
+    --kind: ran, skipped, included or other.
+    --needle: substring of the task name.
+    --dir: profile directory; $ANSIBLE_PROFILE_DIR, else <TMPDIR>/ansible-profile, when omitted.
+    --kb: size of the synthetic fact in KiB.
+    --out: file the synthetic fact goes to.
+
+Exit codes:
+    0 success, 2 usage error or nothing measurable in the range.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+# Trip-wire: python3 -I keeps the script directory off sys.path, and the subcommands live there.
+sys.path.append(str(Path(__file__).resolve().parent))
+
+import ansible_bench
+import ansible_log
+import ansible_profile
+import ansible_tables
+
+Report = tuple[dict, list[str]]
+
+
+def parsed(args: argparse.Namespace, source: str | None) -> tuple[str, ansible_tables.Found]:
+    return ansible_log.tasks(args.log, args.first, args.last, source)
+
+
+def durations(args: argparse.Namespace) -> Report:
+    return ansible_tables.names(*parsed(args, "profile"), args.top)
+
+
+def gaps(args: argparse.Namespace) -> Report:
+    return ansible_tables.names(*parsed(args, "wall"), args.top)
+
+
+def kinds(args: argparse.Namespace) -> Report:
+    return ansible_tables.kind_table(*parsed(args, args.source), args.top)
+
+
+def kind(args: argparse.Namespace) -> Report:
+    source, found = parsed(args, args.source)
+    return ansible_tables.names(source, ansible_tables.of_kind(found, args.kind), args.top)
+
+
+def task(args: argparse.Namespace) -> Report:
+    return ansible_tables.occurrences(*parsed(args, args.source), args.needle, args.top)
+
+
+def profile_report(args: argparse.Namespace) -> Report:
+    return ansible_profile.report(args.dir, args.top)
+
+
+def bench_carrier(args: argparse.Namespace) -> Report:
+    return ansible_bench.write(args.kb, args.out)
+
+
+COMMANDS: dict[str, Callable[[argparse.Namespace], Report]] = {
+    "durations": durations,
+    "gaps": gaps,
+    "kinds": kinds,
+    "kind": kind,
+    "task": task,
+    "profile-report": profile_report,
+    "bench-carrier": bench_carrier,
+}
+
+
+def positive(text: str) -> int:
+    try:
+        number = int(text)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"{text!r}: expected a whole number") from None
+    if number < 1:
+        raise argparse.ArgumentTypeError(f"{text!r} is below 1")
+    return number
+
+
+def parse(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    commands = parser.add_subparsers(dest="command", required=True)
+    output = argparse.ArgumentParser(add_help=False)
+    output.add_argument("--json", action="store_true")
+    log = argparse.ArgumentParser(add_help=False, parents=[output])
+    log.add_argument("--log", required=True, type=Path)
+    log.add_argument("--from", dest="first", type=positive)
+    log.add_argument("--to", dest="last", type=positive)
+    log.add_argument("--top", type=positive)
+    detect = argparse.ArgumentParser(add_help=False, parents=[log])
+    detect.add_argument("--source", choices=ansible_log.SOURCES)
+    commands.add_parser("durations", parents=[log])
+    commands.add_parser("gaps", parents=[log])
+    commands.add_parser("kinds", parents=[detect])
+    commands.add_parser("kind", parents=[detect]).add_argument("--kind", required=True, choices=ansible_log.KINDS)
+    commands.add_parser("task", parents=[detect]).add_argument("--needle", required=True)
+    report = commands.add_parser("profile-report", parents=[output])
+    report.add_argument("--dir", type=Path)
+    report.add_argument("--top", required=True, type=positive)
+    bench = commands.add_parser("bench-carrier", parents=[output])
+    bench.add_argument("--kb", required=True, type=positive)
+    bench.add_argument("--out", required=True, type=Path)
+    args = parser.parse_args(argv)
+    first, last = getattr(args, "first", None), getattr(args, "last", None)
+    if first is not None and last is not None and first > last:
+        parser.error(f"--from {first} is after --to {last}")
+    return args
+
+
+def main(argv: list[str]) -> int:
+    args = parse(argv)
+    try:
+        report, lines = COMMANDS[args.command](args)
+    except (ansible_log.Unmeasurable, OSError) as error:
+        print(f"ansible_tasks: {error}", file=sys.stderr)
+        return 2
+    print(json.dumps(report, indent=2) if args.json else "\n".join(lines))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

skills/kvb-check-ansible/scripts/profile/sitecustomize.py

@@ -0,0 +1,55 @@
+"""Profile ansible-playbook and every worker it forks with cProfile, once on PYTHONPATH.
+
+Args:
+    ANSIBLE_PROFILE_DIR: directory for main-<pid>.prof and worker-<pid>.prof;
+        <TMPDIR>/ansible-profile when unset.
+
+Output:
+    Only a process whose argv[0] is named ansible-playbook is profiled. The controller
+    writes its file at exit, each forked worker when it leaves through os._exit.
+"""
+
+import atexit
+import cProfile
+import os
+import sys
+import tempfile
+
+PROGRAM = "ansible-playbook"
+
+
+def profile_dir() -> str:
+    # Trip-wire: TMPDIR is the one scratch directory every sandbox lets the run write to.
+    return os.environ.get("ANSIBLE_PROFILE_DIR") or os.path.join(tempfile.gettempdir(), "ansible-profile")
+
+
+def start(out: str) -> None:
+    os.makedirs(out, exist_ok=True)
+    controller = cProfile.Profile()
+
+    def dump_controller() -> None:
+        controller.disable()
+        controller.dump_stats(os.path.join(out, f"main-{os.getpid()}.prof"))
+
+    def after_fork() -> None:
+        # Trip-wire: the fork inherits the enabled controller profiler; no second one starts beside it.
+        controller.disable()
+        worker = cProfile.Profile()
+        real_exit = os._exit
+
+        def exit_dumping(code: int) -> None:
+            worker.disable()
+            worker.dump_stats(os.path.join(out, f"worker-{os.getpid()}.prof"))
+            real_exit(code)
+
+        # Trip-wire: multiprocessing workers leave through os._exit, which skips atexit.
+        os._exit = exit_dumping
+        worker.enable()
+
+    atexit.register(dump_controller)
+    os.register_at_fork(after_in_child=after_fork)
+    controller.enable()
+
+
+if os.path.basename(sys.argv[0] if sys.argv else "") == PROGRAM:
+    start(profile_dir())

skills/kvb-code-tune/SKILL.md

@@ -41,6 +41,7 @@ operator to decide, never applied as a tuning.
   beats a 60 s download paid once.
 - **The baseline is the newest log of the current code**; an older one may
   carry a step since replaced, and a saving against it is fiction.
+- **Ansible**: measure with the `kvb-check-ansible` skill first.
 
 ## Coordinate every cost cluster
 

tests/ansible_support.py

@@ -0,0 +1,125 @@
+"""Shared paths, script modules, synthetic logs and the CLI test case of the kvb-check-ansible tests."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SKILL = Path(__file__).resolve().parents[1] / "skills" / "kvb-check-ansible"
+sys.path.insert(0, str(SKILL / "scripts"))
+
+import ansible_bench
+import ansible_log
+import ansible_tables
+
+__all__ = [
+    "BENCH",
+    "MIDNIGHT_LOG",
+    "PROFILE_DIR",
+    "PROFILE_LOG",
+    "SCRIPT",
+    "WALL_KINDS",
+    "WALL_LOG",
+    "LogCase",
+    "ansible_bench",
+    "ansible_log",
+    "ansible_tables",
+]
+
+SCRIPT = SKILL / "scripts" / "ansible_tasks.py"
+PROFILE_DIR = SKILL / "scripts" / "profile"
+BENCH = SKILL / "bench"
+STARS = " " + "*" * 40
+
+WALL_LOG = "\n".join(
+    [
+        "2026-11-07T06:45:40.0000000Z PLAY [deploy]" + STARS,
+        "2026-11-07T06:45:40.5000000Z TASK [setup : install]" + STARS,
+        "2026-11-07T06:45:43.0000000Z \x1b[0;33mchanged: [host]\x1b[0m",
+        "2026-11-07T06:45:44.0000000Z TASK [setup : check]" + STARS,
+        "2026-11-07T06:45:44.1000000Z skipping: [host]",
+        "2026-11-07T06:45:44.2Z TASK [setup : install]" + STARS,
+        "2026-11-07T06:45:44.5000000Z skipping: [host] => (item=a)",
+        "2026-11-07T06:45:45.0000000Z ok: [host] => (item=b)",
+        "2026-11-07T06:45:45.700000000Z TASK [setup : include]" + STARS,
+        "2026-11-07T06:45:45.8000000Z included: /x/y.yml for host",
+        "2026-11-07T06:45:46.0 TASK [setup : flush]" + STARS,
+        "2026-11-07T06:45:46.5000000Z [WARNING]: nothing to flush",
+        "2026-11-07T06:45:46.8000000Z RUNNING HANDLER [setup : restart]" + STARS,
+        "2026-11-07T06:45:48.0000000Z fatal: [host]: FAILED! => {}",
+        "2026-11-07T06:45:48.0000000Z PLAY RECAP" + STARS,
+        "2026-11-07T06:45:48.0000000Z host : ok=2 changed=1 skipped=1 failed=1",
+    ]
+) + "\n"
+
+PROFILE_LINE = "Saturday 07 November 2026  06:45:40 +0000 ({})       0:00:00.020 " + "*" * 20
+PROFILE_LOG = "\n".join(
+    [
+        "PLAY [deploy]" + STARS,
+        "",
+        "TASK [Gathering Facts]" + STARS,
+        PROFILE_LINE.format("0:00:00.020"),
+        "ok: [host]",
+        "TASK [setup : install]" + STARS,
+        PROFILE_LINE.format("0:00:02.000"),
+        "changed: [host]",
+        "TASK [setup : check]" + STARS,
+        PROFILE_LINE.format("0:00:03.500"),
+        "skipping: [host]",
+        "TASK [setup : install]" + STARS,
+        PROFILE_LINE.format("0:00:00.100"),
+        "included: /x/y.yml for host",
+        "RUNNING HANDLER [setup : restart]" + STARS,
+        PROFILE_LINE.format("0:01:01.500"),
+        "ok: [host]",
+        "PLAY RECAP" + STARS,
+        "host : ok=3 changed=1 skipped=1",
+        "TASKS RECAP" + STARS,
+        PROFILE_LINE.format("0:00:00.400"),
+        "=" * 40,
+        "setup : install ----------------------- 63.50s",
+    ]
+) + "\n"
+
+MIDNIGHT_LOG = chr(0xFEFF) + "2026-11-07T23:59:59.5Z TASK [late]" + STARS + "\n2026-11-08T00:00:00.25Z PLAY RECAP" + STARS + "\n"
+
+WALL_KINDS = [
+    "kind\tcount\tsum\tmean\t<0.25\t<0.5\t<1\t<2\t>=2",
+    "ran\t3\t6.2\t2.0667\t0\t0\t0\t2\t1",
+    "other\t1\t0.8\t0.8000\t0\t0\t1\t0\t0",
+    "included\t1\t0.3\t0.3000\t0\t1\t0\t0\t0",
+    "skipped\t1\t0.2\t0.2000\t1\t0\t0\t0\t0",
+]
+
+
+class LogCase(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.base = Path(tmp.name)
+        self.wall = self.write("wall.log", WALL_LOG)
+        self.profile = self.write("profile.log", PROFILE_LOG)
+
+    def write(self, name: str, text: str) -> Path:
+        path = self.base / name
+        path.write_text(text, encoding="utf-8")
+        return path
+
+    def run_script(self, *args: str, env: dict | None = None) -> subprocess.CompletedProcess:
+        return subprocess.run(
+            [sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False, env=env
+        )
+
+    def lines(self, *args: str) -> list[str]:
+        result = self.run_script(*args)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        return result.stdout.splitlines()
+
+    def report(self, *args: str) -> dict:
+        result = self.run_script(*args, "--json")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        return json.loads(result.stdout)

tests/test_ansible_bench.py

@@ -0,0 +1,79 @@
+"""Exercise kvb-check-ansible's bench: the synthetic carrier fact and a real run of the bench playbook."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from ansible_support import BENCH, SCRIPT, LogCase, ansible_bench, ansible_log
+
+
+class TestBenchCarrier(LogCase):
+    def test_writes_exactly_the_asked_size(self):
+        out = self.base / "carrier.json"
+        result = self.run_script("bench-carrier", "--kb", "3", "--out", str(out))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(out.stat().st_size, 3 * 1024)
+        fact = json.loads(out.read_text(encoding="ascii"))
+        self.assertEqual(result.stdout, f"{out}\t3072 bytes\t{len(fact) - 1} entries\n")
+        self.assertIn("pad", fact)
+        for size in (1024, 1500, 400 * 1024):
+            with self.subTest(size=size):
+                self.assertEqual(len(json.dumps(ansible_bench.carrier(size))), size)
+        self.assertGreater(len(ansible_bench.carrier(400 * 1024)), 100)
+
+    def test_bench_inventory_pins_no_interpreter_path(self):
+        self.assertNotRegex((BENCH / "hosts.yml").read_text(encoding="utf-8"), r"interpreter:\s*/")
+
+
+@unittest.skipUnless(shutil.which("ansible-playbook"), "ansible-playbook not installed")
+class TestBench(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        tmp = tempfile.TemporaryDirectory()
+        cls.addClassCleanup(tmp.cleanup)
+        base = Path(tmp.name)
+        carrier = base / "carrier.json"
+        carrier.write_text(json.dumps(ansible_bench.carrier(8 * 1024)), encoding="ascii")
+        cls.log = base / "bench.log"
+        env = {
+            "PATH": os.environ["PATH"],
+            "HOME": str(base),
+            "ANSIBLE_LOCAL_TEMP": str(base / "tmp"),
+            "ANSIBLE_NOCOLOR": "1",
+            "ANSIBLE_CALLBACKS_ENABLED": "ansible.posix.profile_tasks",
+        }
+        cls.result = subprocess.run(
+            [
+                "ansible-playbook", "-i", str(BENCH / "hosts.yml"), str(BENCH / "bench.yml"),
+                "-e", "bench_count=3", "-e", f"bench_carrier_file={carrier}",
+            ],
+            cwd=base, env=env, capture_output=True, text=True, check=False,
+        )
+        cls.log.write_text(cls.result.stdout, encoding="utf-8")
+
+    def test_runs_one_skipped_and_one_ran_task_per_item(self):
+        self.assertEqual(self.result.returncode, 0, self.result.stdout + self.result.stderr)
+        self.assertEqual(self.result.stdout.count("skipping: [bench]"), 3)
+        self.assertEqual(self.result.stdout.count("included: "), 3)
+
+    def test_profile_tasks_output_ranks_by_kind(self):
+        if ansible_log.ELAPSED.search(self.result.stdout) is None:
+            self.skipTest("ansible.posix.profile_tasks not installed")
+        result = subprocess.run(
+            [sys.executable, str(SCRIPT), "kinds", "--log", str(self.log), "--json"],
+            capture_output=True, text=True, check=False,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        counts = {row["kind"]: row["count"] for row in json.loads(result.stdout)["kinds"]}
+        self.assertEqual(counts, {"ran": 4, "skipped": 3, "included": 1})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ansible_log.py

@@ -0,0 +1,96 @@
+"""Exercise kvb-check-ansible's log parsing: timestamps, sources, line ranges and the task boundaries."""
+
+from __future__ import annotations
+
+import calendar
+import unittest
+
+from ansible_support import MIDNIGHT_LOG, PROFILE_LOG, LogCase, ansible_log
+
+
+class TestStamps(unittest.TestCase):
+    def test_fraction_digits_zone_and_missing_z(self):
+        base = calendar.timegm((2026, 11, 7, 6, 45, 47))
+        for text, fraction in (
+            ("2026-11-07T06:45:47.1234567Z x", 0.1234567),
+            ("2026-11-07T06:45:47.1Z x", 0.1),
+            ("2026-11-07T06:45:47Z x", 0.0),
+            ("2026-11-07T06:45:47.123456789 x", 0.123456789),
+            ("2026-11-07T07:45:47.5+01:00 x", 0.5),
+            ("2026-11-07T01:45:47.5-0500 x", 0.5),
+            ("2026-11-07 06:45:47,25 x", 0.25),
+        ):
+            with self.subTest(text=text):
+                stamp, content = ansible_log.split_stamp(text)
+                self.assertAlmostEqual(stamp, base + fraction, places=5)
+                self.assertEqual(content, "x")
+        self.assertEqual(ansible_log.split_stamp("TASK [x] ***"), (None, "TASK [x] ***"))
+
+
+class TestWallParsing(LogCase):
+    def test_line_range_drops_the_task_its_end_cuts_off(self):
+        self.assertEqual(
+            self.lines("gaps", "--log", str(self.wall), "--from", "4", "--to", "13"),
+            [
+                "1.5\t1\tsetup : install",
+                "0.8\t1\tsetup : flush",
+                "0.3\t1\tsetup : include",
+                "0.2\t1\tsetup : check",
+                "2.8\t4\tTOTAL",
+            ],
+        )
+
+    def test_midnight_bom_and_recap_close_the_last_task(self):
+        midnight = self.write("midnight.log", MIDNIGHT_LOG)
+        self.assertEqual(self.report("gaps", "--log", str(midnight))["rows"], [{"name": "late", "sum": 0.75, "count": 1}])
+
+
+class TestProfileParsing(LogCase):
+    def test_durations_pair_each_header_with_the_next_elapsed_line(self):
+        self.assertEqual(
+            self.lines("durations", "--log", str(self.profile)),
+            [
+                "65.0\t2\tsetup : install",
+                "2.0\t1\tGathering Facts",
+                "0.4\t1\tsetup : restart",
+                "0.1\t1\tsetup : check",
+                "67.5\t5\tTOTAL",
+            ],
+        )
+
+    def test_kinds_fall_back_to_the_profile_source(self):
+        self.assertEqual(
+            self.lines("kinds", "--log", str(self.profile)),
+            [
+                "kind\tcount\tsum\tmean\t<0.25\t<0.5\t<1\t<2\t>=2",
+                "included\t1\t61.5\t61.5000\t0\t0\t0\t0\t1",
+                "ran\t3\t5.9\t1.9667\t0\t1\t0\t0\t2",
+                "skipped\t1\t0.1\t0.1000\t1\t0\t0\t0\t0",
+            ],
+        )
+        self.assertEqual(self.report("kinds", "--log", str(self.profile))["source"], "profile")
+
+    def test_profile_source_reads_a_timestamped_log_too(self):
+        stamped = self.write("stamped.log", "".join(f"2026-11-07T06:45:40.1234Z {line}\n" for line in PROFILE_LOG.splitlines()))
+        self.assertEqual(
+            self.lines("kind", "--log", str(stamped), "--source", "profile", "--kind", "skipped"),
+            ["0.1\t1\tsetup : check", "0.1\t1\tTOTAL"],
+        )
+
+
+class TestUnmeasurable(LogCase):
+    def assertFails(self, needle: str, *args: str):
+        result = self.run_script(*args)
+        self.assertEqual((result.returncode, result.stdout), (2, ""))
+        self.assertIn(needle, result.stderr)
+
+    def test_wrong_source_or_range_exits_2(self):
+        self.assertFails("ISO timestamp", "gaps", "--log", str(self.profile))
+        self.assertFails("ISO timestamp", "kinds", "--log", str(self.profile), "--source", "wall")
+        self.assertFails("profile_tasks", "durations", "--log", str(self.wall))
+        self.assertFails("no task closed", "gaps", "--log", str(self.wall), "--from", "2", "--to", "2")
+        self.assertFails("No such file", "gaps", "--log", str(self.base / "missing.log"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ansible_profile.py

@@ -0,0 +1,98 @@
+"""Exercise kvb-check-ansible's controller profiler and its profile-report subcommand."""
+
+from __future__ import annotations
+
+import cProfile
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from ansible_support import PROFILE_DIR, LogCase
+
+FORKING = (
+    "import multiprocessing\n"
+    "def work():\n"
+    "    return sum(range(1000))\n"
+    "if __name__ == '__main__':\n"
+    "    child = multiprocessing.get_context('fork').Process(target=work)\n"
+    "    child.start()\n"
+    "    child.join()\n"
+)
+
+
+def busy() -> int:
+    return sum(number * number for number in range(20000))
+
+
+class TestProfileReport(LogCase):
+    def setUp(self):
+        super().setUp()
+        self.profiles = self.base / "profiles"
+        self.profiles.mkdir()
+        for name in ("main-1.prof", "worker-2.prof", "worker-3.prof"):
+            profiler = cProfile.Profile()
+            profiler.enable()
+            busy()
+            profiler.disable()
+            profiler.dump_stats(self.profiles / name)
+
+    def test_merges_main_and_worker_profiles(self):
+        found = self.report("profile-report", "--dir", str(self.profiles), "--top", "3")
+        self.assertEqual((found["main"]["profiles"], found["worker"]["profiles"]), (1, 2))
+        for role, calls in (("main", 1), ("worker", 2)):
+            with self.subTest(role=role):
+                self.assertLessEqual(len(found[role]["cumulative"]), 3)
+                self.assertLessEqual(len(found[role]["tottime"]), 3)
+                rows = [row for row in found[role]["cumulative"] if row["function"].endswith("(busy)")]
+                self.assertEqual([row["calls"] for row in rows], [calls])
+        lines = self.lines("profile-report", "--dir", str(self.profiles), "--top", "2")
+        self.assertRegex(lines[0], r"^== main: 1 profiles, \d+\.\d{3} s ==$")
+        self.assertIn("-- tottime: cumtime, tottime, calls, function --", lines)
+        self.assertTrue(any(line.startswith("== worker: 2 profiles") for line in lines))
+
+    def test_dir_comes_from_the_environment_and_empty_exits_2(self):
+        env = {"PATH": os.environ["PATH"], "ANSIBLE_PROFILE_DIR": str(self.profiles)}
+        result = self.run_script("profile-report", "--top", "1", env=env)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertTrue(result.stdout.startswith("== main: 1 profiles"))
+        empty = self.base / "empty"
+        empty.mkdir()
+        result = self.run_script("profile-report", "--dir", str(empty), "--top", "1")
+        self.assertEqual(result.returncode, 2)
+        self.assertIn("no main-*.prof or worker-*.prof", result.stderr)
+
+
+class TestProfiler(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.base = Path(tmp.name)
+
+    def fork(self, program: str, **env: str) -> None:
+        script = self.base / program
+        script.write_text(FORKING, encoding="utf-8")
+        result = subprocess.run(
+            [sys.executable, str(script)],
+            env={"PATH": os.environ["PATH"], "PYTHONPATH": str(PROFILE_DIR), "TMPDIR": str(self.base), **env},
+            capture_output=True, text=True, check=False,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_ansible_playbook_and_its_workers_land_under_tmpdir(self):
+        self.fork("ansible-playbook")
+        out = self.base / "ansible-profile"
+        self.assertEqual(len(list(out.glob("main-*.prof"))), 1)
+        self.assertEqual(len(list(out.glob("worker-*.prof"))), 1)
+
+    def test_profile_dir_wins_and_other_programs_stay_unprofiled(self):
+        self.fork("ansible-playbook", ANSIBLE_PROFILE_DIR=str(self.base / "chosen"))
+        self.assertEqual(len(list((self.base / "chosen").glob("*.prof"))), 2)
+        self.fork("ansible", ANSIBLE_PROFILE_DIR=str(self.base / "other"))
+        self.assertFalse((self.base / "other").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ansible_tables.py

@@ -0,0 +1,97 @@
+"""Exercise kvb-check-ansible's reporting subcommands: name, kind and occurrence tables, JSON and usage errors."""
+
+from __future__ import annotations
+
+import unittest
+
+from ansible_support import WALL_KINDS, LogCase, ansible_tables
+
+
+class TestNameTables(LogCase):
+    def test_gaps_sum_per_name_biggest_first_with_total(self):
+        self.assertEqual(
+            self.lines("gaps", "--log", str(self.wall)),
+            [
+                "5.0\t2\tsetup : install",
+                "1.2\t1\tsetup : restart",
+                "0.8\t1\tsetup : flush",
+                "0.3\t1\tsetup : include",
+                "0.2\t1\tsetup : check",
+                "7.5\t6\tTOTAL",
+            ],
+        )
+
+    def test_gaps_json_and_top_keep_the_full_total(self):
+        self.assertEqual(
+            self.report("gaps", "--log", str(self.wall), "--top", "2"),
+            {
+                "source": "wall",
+                "rows": [
+                    {"name": "setup : install", "sum": 5.0, "count": 2},
+                    {"name": "setup : restart", "sum": 1.2, "count": 1},
+                ],
+                "total": {"sum": 7.5, "count": 6},
+            },
+        )
+
+    def test_kind_sums_one_kind_per_name(self):
+        self.assertEqual(
+            self.lines("kind", "--log", str(self.wall), "--kind", "ran"),
+            ["5.0\t2\tsetup : install", "1.2\t1\tsetup : restart", "6.2\t3\tTOTAL"],
+        )
+        self.assertEqual(self.lines("kind", "--log", str(self.wall), "--kind", "other"), ["0.8\t1\tsetup : flush", "0.8\t1\tTOTAL"])
+
+
+class TestKindTable(LogCase):
+    def test_kinds_detect_ran_before_skipped_and_fill_the_buckets(self):
+        self.assertEqual(self.lines("kinds", "--log", str(self.wall)), WALL_KINDS)
+        found = self.report("kinds", "--log", str(self.wall))
+        self.assertEqual(found["source"], "wall")
+        self.assertEqual(
+            found["kinds"][0],
+            {"kind": "ran", "count": 3, "sum": 6.2, "mean": 2.0667,
+             "buckets": {"<0.25": 0, "<0.5": 0, "<1": 0, "<2": 2, ">=2": 1}},
+        )
+
+    def test_bucket_edges(self):
+        for seconds, label in ((0.0, "<0.25"), (0.2499, "<0.25"), (0.25, "<0.5"), (0.5, "<1"), (1.0, "<2"), (1.999, "<2"), (2.0, ">=2")):
+            with self.subTest(seconds=seconds):
+                self.assertEqual(ansible_tables.bucket(seconds), label)
+
+
+class TestOccurrences(LogCase):
+    def test_task_lists_every_occurrence_of_the_needle(self):
+        self.assertEqual(
+            self.lines("task", "--log", str(self.wall), "--needle", "install"),
+            ["2\t3.500\tran\tsetup : install", "6\t1.500\tran\tsetup : install"],
+        )
+        found = self.report("task", "--log", str(self.wall), "--needle", "install", "--top", "1")
+        self.assertEqual(
+            found,
+            {
+                "source": "wall",
+                "needle": "install",
+                "count": 2,
+                "sum": 5.0,
+                "occurrences": [{"line": 2, "seconds": 3.5, "kind": "ran", "name": "setup : install"}],
+            },
+        )
+
+
+class TestUsage(LogCase):
+    def test_usage_errors_exit_2(self):
+        for args in (
+            ("gaps",),
+            ("gaps", "--log", str(self.wall), "--from", "5", "--to", "4"),
+            ("gaps", "--log", str(self.wall), "--top", "0"),
+            ("kind", "--log", str(self.wall), "--kind", "slow"),
+            ("task", "--log", str(self.wall)),
+            ("profile-report", "--dir", str(self.base)),
+            ("bench-carrier", "--kb", "0", "--out", str(self.base / "c.json")),
+        ):
+            with self.subTest(args=args):
+                self.assertEqual(self.run_script(*args).returncode, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()