diff --git a/docs/requirements/001-agent-orchestrator.md b/docs/requirements/001-agent-orchestrator.md index 09bdb9f..17c44fc 100644 --- a/docs/requirements/001-agent-orchestrator.md +++ b/docs/requirements/001-agent-orchestrator.md @@ -739,6 +739,8 @@ flowchart TD ### Phase 6: Resources (S7) +Replaced by [003 W4 Resources](003-operator-workbench.md#w4-resources). + - [ ] The tab shows tokens per agent and accumulated, and CPU, RAM, disk and load, on the shared time axis, several values per chart. - [ ] Token counts equal the sum of the transcript usage fields of a fixture diff --git a/docs/requirements/003-operator-workbench.md b/docs/requirements/003-operator-workbench.md index d1b34c9..352c7a3 100644 --- a/docs/requirements/003-operator-workbench.md +++ b/docs/requirements/003-operator-workbench.md @@ -248,22 +248,22 @@ flowchart TD ### W4 Resources -- [ ] `resources.py sample`, run every 60 s by a user timer that +- [x] `resources.py sample`, run every 60 s by a user timer that `make orchestrator-install` enables, appends one row: used and total per mount, CPU, RAM and swap percent, load, GPU utilization and VRAM per GPU, waiting and active queue jobs. -- [ ] An NVIDIA GPU whose PCI `runtime_status` is not `active` is recorded as +- [x] An NVIDIA GPU whose PCI `runtime_status` is not `active` is recorded as asleep without running `nvidia-smi`. -- [ ] `resources.py tokens` adds input, output, cache-read and cache-write +- [x] `resources.py tokens` adds input, output, cache-read and cache-write tokens per agent and model from transcript usage fields since the last byte offset; no sampler or parser calls an LLM. -- [ ] Rows older than 7 days fold into 15-minute means; means older than 90 +- [x] Rows older than 7 days fold into 15-minute means; means older than 90 days are dropped. - [ ] The Resources entry shows on one time axis disks, CPU, RAM, swap, GPU, VRAM and stacked tokens per agent; bands mark idle (< 30 %) and saturated (> 90 %); a marker flags spans where jobs waited while CPU and RAM stayed under 50 %; the view reads only the sampler files. -- [ ] 001 S7 points to this section. +- [x] 001 S7 points to this section. ### Contracts @@ -289,7 +289,7 @@ The work packages meet only through these interfaces: used, total}], cpu, ram, swap, load, gpus: [{name, util, vram_used, vram_total} | {name, asleep: true}], queue: {waiting, active}}`), `tokens.jsonl` (`{t, agent, model, input, output, cache_read, - cache_write}`, t the hour) and `offsets.json` under + cache_write}`, t the hour), `offsets.json` and `cpu.json` under `~/.local/state/agents-council/resources/`. - **Mounts**: the hub mounts `~/.local/state/agents-council` read-only and opens `gates.db` read-only. @@ -339,6 +339,10 @@ The work packages meet only through these interfaces: - [ ] Resources sample parsing on fixtures (df, meminfo, nvidia-smi, asleep GPU); token offsets; retention fold; diff cap and masking; feedback payload. + - [x] W4 part of the item above: sample parsing on fixtures (df, meminfo, + CPU ticks between runs, nvidia-smi, asleep and unread GPU); token + offsets, a message counted once across fork transcripts; retention + fold. - [ ] Bubble schedule: scale per queue position, word reveal times from a duration, reading-pace duration, one bubble per agent. - Integration: @@ -348,7 +352,7 @@ The work packages meet only through these interfaces: - [ ] The host executor applies a signed board batch to a scratch register and refuses a forged one; deletes a local merged branch without a token and refuses a remote one. - - [ ] The sampler writes rows in a scratch HOME. + - [x] The sampler writes rows in a scratch HOME. - E2E, headless Chromium and Firefox at 1920, 1366 and 390 px: - [ ] Tasks badge, no Inbox entry, operator cards with bulk actions, review diff expanded, feedback by right-click and `f` reaching the radio diff --git a/orchestrator/client/resources.py b/orchestrator/client/resources.py new file mode 100755 index 0000000..25106a5 --- /dev/null +++ b/orchestrator/client/resources.py @@ -0,0 +1,590 @@ +#!/usr/bin/env python3 +"""Args: + sample: append one row to /samples.jsonl: {t, disks: [{mount, + used, total}], cpu, ram, swap, load, gpus: [{name, util, vram_used, + vram_total} | {name, asleep: true}], queue: {waiting, active}}; t in + epoch seconds, used, total and vram in bytes, cpu, ram, swap and util + in percent, load the 1-minute load average, a GPU named by its PCI + address. cpu is the busy share of the /proc/stat ticks since those + the run before kept in /cpu.json as [t, idle, total], None on + the first run and after a gap of more than 15 minutes. An NVIDIA GPU + whose PCI runtime_status is not active is recorded asleep, and + nvidia-smi is not run for it; an awake one nvidia-smi gives no + reading for (no driver, a driver mismatch) keeps its name with util + and vram None. In the first minute of a 15-minute bucket, rows older + than 7 days fold into 15-minute means and means older than 90 days + are dropped. + tokens: add the input, output, cache-read and cache-write tokens of the + usage fields in the Claude Code transcripts under --transcripts since + each file's byte offset to /tokens.jsonl as {t, agent, model, + input, output, cache_read, cache_write}, t the hour; agent is the + name the council claim of the session carries, else the session id's + first 8 characters. A message counts once by its message id and + request id, also when written over several lines or copied into a + fork's transcript. /offsets.json holds {files: {path: offset}, + seen: {hour: [key digest]}}, seen keeping the hours this run read and + the last two, and is replaced only when it changes. In the first + minute of an hour, rows of the same hour, agent and model merge and + rows older than 90 days are dropped. + --state: the directory of these files, the unit passes + ~/.local/state/agents-council/resources. + --council: the council directory with queue.json and the claims. + --transcripts (tokens): Claude Code's projects directory. + --proc, --sys (sample): where /proc and /sys are mounted. + +Returns: + Runs on the host every minute through kvb-orchestrator-resources.timer, + because the hub's container sees neither the host's disks, memory and + GPUs nor its transcripts' offsets; nothing here calls a language model. + The exit code is 0. +""" + +from __future__ import annotations + +import argparse +import base64 +import datetime +import hashlib +import json +import os +import shutil +import stat +import subprocess +import tempfile +import time +from collections.abc import Callable +from pathlib import Path + +INTERVAL = 60 +BUCKET = 900 +HOUR = 3600 +FINE = 7 * 86400 +COARSE = 90 * 86400 +COMMAND_SECONDS = 10 +KEY_BYTES = 9 +KIB = 1 << 10 +MIB = 1 << 20 +JSON_BYTES = 1 << 24 +BUDGET = 64 << 20 +NVIDIA = "0x10de" +DISPLAY = "0x03" +NVIDIA_QUERY = "pci.bus_id,utilization.gpu,memory.used,memory.total" +UNREAD = {"util": None, "vram_used": None, "vram_total": None} +COUNTS = (("input", "input_tokens"), ("output", "output_tokens"), ("cache_read", "cache_read_input_tokens"), ("cache_write", "cache_creation_input_tokens")) +NAMES = tuple(name for name, _ in COUNTS) +HOST_DIGITS = (("cpu", 1), ("ram", 1), ("swap", 1), ("load", 2)) +CLAIMS = ("*/claim.json", "*/*/claim.json") +SAMPLES = "samples.jsonl" +TOKENS = "tokens.jsonl" +OFFSETS = "offsets.json" +TICKS = "cpu.json" + + +def read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return "" + + +def read_json(path: Path) -> object: + """Returns: + The parsed JSON of path, read without following a symlink and only + from a regular file of at most JSON_BYTES; None when that fails. + """ + try: + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as handle: + if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode): + return None + data = handle.read(JSON_BYTES + 1) + return None if len(data) > JSON_BYTES else json.loads(data.decode("utf-8")) + except (OSError, ValueError, RecursionError): + return None + + +def number(text: str) -> float | None: + try: + return float(text) + except ValueError: + return None + + +def percent(part: float, whole: float | None) -> float | None: + return round(100 * part / whole, 1) if whole else None + + +def command(argv: list[str]) -> str | None: + if shutil.which(argv[0]) is None: + return None + try: + done = subprocess.run(argv, capture_output=True, text=True, timeout=COMMAND_SECONDS, check=False) + except (OSError, subprocess.SubprocessError): + return None + return done.stdout if done.returncode == 0 else None + + +def parse_df(text: str) -> list[dict]: + """Returns: + One {mount, used, total} in bytes per block device in the output of + df -P -k, at the first mount df lists for it; loop devices and + filesystems without a device are left out. + """ + found, seen = [], set() + for line in text.splitlines()[1:]: + parts = line.split(None, 5) + if len(parts) < 6 or not parts[0].startswith("/dev/") or parts[0].startswith("/dev/loop") or parts[0] in seen: + continue + total, used = number(parts[1]), number(parts[2]) + if total is None or used is None: + continue + seen.add(parts[0]) + found.append({"mount": parts[5], "used": int(used) * KIB, "total": int(total) * KIB}) + return found + + +def parse_meminfo(text: str) -> dict[str, int]: + values = {} + for line in text.splitlines(): + key, _, rest = line.partition(":") + fields = rest.split() + if fields and fields[0].isdigit(): + values[key.strip()] = int(fields[0]) + return values + + +def memory(values: dict[str, int]) -> tuple[float | None, float | None]: + """Returns: + RAM in use ((MemTotal - MemAvailable) / MemTotal) and swap in use in + percent; swap is 0 without a swap device, None when unknown. + """ + total, available = values.get("MemTotal"), values.get("MemAvailable") + ram = percent(total - available, total) if total and available is not None else None + swap_total, swap_free = values.get("SwapTotal"), values.get("SwapFree") + if swap_total == 0: + return ram, 0.0 + return ram, percent(swap_total - swap_free, swap_total) if swap_total and swap_free is not None else None + + +def cpu_ticks(text: str) -> tuple[int, int] | None: + """Returns: + (idle, total) ticks of the summary line of /proc/stat: idle counts + idle and iowait, total the first eight fields, which already hold + guest time. + """ + fields = text.split("\n", 1)[0].split() + if len(fields) < 6 or fields[0] != "cpu" or not all(field.isdigit() for field in fields[1:9]): + return None + values = [int(field) for field in fields[1:9]] + return values[3] + values[4], sum(values) + + +def cpu_percent(before: tuple[int, int] | None, after: tuple[int, int] | None) -> float | None: + if before is None or after is None: + return None + idle, total = after[0] - before[0], after[1] - before[1] + return percent(total - idle, total) if total > 0 else None + + +def parse_loadavg(text: str) -> float | None: + fields = text.split() + return number(fields[0]) if fields else None + + +def parse_nvidia(text: str) -> dict[str, dict]: + """Returns: + Per PCI address (domain of 4 hex digits) {util, vram_used, + vram_total} from the CSV nvidia-smi prints for NVIDIA_QUERY with + noheader,nounits; vram in bytes, an unknown value None. + """ + found = {} + for line in text.splitlines(): + parts = [part.strip() for part in line.split(",")] + if len(parts) != 4: + continue + util, used, total = (number(part) for part in parts[1:]) + found[parts[0].lower()[-12:]] = {"util": util, "vram_used": None if used is None else int(used * MIB), "vram_total": None if total is None else int(total * MIB)} + return found + + +def nvidia_gpus(sys_root: Path) -> list[tuple[str, bool]]: + """Returns: + The PCI address of every NVIDIA display controller under sys_root and + whether its runtime_status reads active. + """ + found = [] + for device in sorted((sys_root / "bus" / "pci" / "devices").glob("*")): + if read_text(device / "vendor").strip() == NVIDIA and read_text(device / "class").strip().startswith(DISPLAY): + found.append((device.name, read_text(device / "power" / "runtime_status").strip() == "active")) + return found + + +def gpus(sys_root: Path, run: Callable[[list[str]], str | None]) -> list[dict]: + devices = nvidia_gpus(sys_root) + awake = [address for address, active in devices if active] + readings = parse_nvidia(run(["nvidia-smi", f"--id={','.join(awake)}", f"--query-gpu={NVIDIA_QUERY}", "--format=csv,noheader,nounits"]) or "") if awake else {} + return [{"name": address, **readings.get(address, UNREAD)} if active else {"name": address, "asleep": True} for address, active in devices] + + +def queue(council: Path) -> dict: + data = read_json(council / "queue.json") + entries = data.get("entries") if isinstance(data, dict) else None + statuses = [entry.get("status") for entry in entries if isinstance(entry, dict)] if isinstance(entries, list) else [] + return {"waiting": statuses.count("waiting"), "active": statuses.count("active")} + + +def integers(value: object, size: int) -> bool: + return isinstance(value, list) and len(value) == size and all(isinstance(item, int) and not isinstance(item, bool) for item in value) + + +def cpu_since(proc: Path, state: Path, t: int) -> float | None: + """Returns: + The busy share of the ticks in proc's stat since those the run + before kept in /cpu.json at most BUCKET seconds before t, + else None; this run's ticks replace them. + """ + ticks = cpu_ticks(read_text(proc / "stat")) + stored = read_json(state / TICKS) + before = (stored[1], stored[2]) if integers(stored, 3) and t - BUCKET <= stored[0] <= t else None + if ticks is not None: + replace(state / TICKS, json.dumps([t, *ticks])) + return cpu_percent(before, ticks) + + +def sample(proc: Path, sys_root: Path, council: Path, run: Callable[[list[str]], str | None], state: Path) -> dict: + t = int(time.time()) + ram, swap = memory(parse_meminfo(read_text(proc / "meminfo"))) + return { + "t": t, + "disks": parse_df(run(["df", "-P", "-k", "-l"]) or ""), + "cpu": cpu_since(proc, state, t), + "ram": ram, + "swap": swap, + "load": parse_loadavg(read_text(proc / "loadavg")), + "gpus": gpus(sys_root, run), + "queue": queue(council), + } + + +def average(values: list[object], digits: int) -> float | None: + found = [value for value in values if isinstance(value, int | float) and not isinstance(value, bool)] + return round(sum(found) / len(found), digits) if found else None + + +def whole(values: list[object]) -> int | None: + found = average(values, 0) + return None if found is None else int(found) + + +def listed(rows: list[dict], key: str) -> list[dict]: + return [item for row in rows if isinstance(row.get(key), list) for item in row[key] if isinstance(item, dict)] + + +def grouped(items: list[dict], key: str) -> dict[object, list[dict]]: + found: dict[object, list[dict]] = {} + for item in items: + found.setdefault(item.get(key), []).append(item) + return found + + +def gpu_mean(name: object, readings: list[dict]) -> dict: + awake = [reading for reading in readings if reading.get("asleep") is not True] + if not awake: + return {"name": name, "asleep": True} + return {"name": name, "util": average([reading.get("util") for reading in awake], 1), "vram_used": whole([reading.get("vram_used") for reading in awake]), "vram_total": whole([reading.get("vram_total") for reading in awake])} + + +def disk_means(rows: list[dict]) -> list[dict]: + return [{"mount": mount, "used": whole([disk.get("used") for disk in disks]), "total": whole([disk.get("total") for disk in disks])} for mount, disks in grouped(listed(rows, "disks"), "mount").items()] + + +def queue_mean(rows: list[dict]) -> dict: + queues = [row["queue"] for row in rows if isinstance(row.get("queue"), dict)] + return {key: average([entry.get(key) for entry in queues], 2) for key in ("waiting", "active")} + + +def mean_row(start: int, rows: list[dict]) -> dict: + """Returns: + The 15-minute mean of rows at start: per mount and per GPU the mean + of the rows that list it, a GPU asleep in every row stays asleep; + a single row already at start comes back unchanged. + """ + if len(rows) == 1 and rows[0]["t"] == start: + return rows[0] + host = {key: average([row.get(key) for row in rows], digits) for key, digits in HOST_DIGITS} + gpu_rows = [gpu_mean(name, readings) for name, readings in grouped(listed(rows, "gpus"), "name").items()] + return {"t": start, "disks": disk_means(rows), **host, "gpus": gpu_rows, "queue": queue_mean(rows)} + + +def timed(row: object) -> bool: + return isinstance(row, dict) and isinstance(row.get("t"), int | float) and not isinstance(row.get("t"), bool) + + +def retain(rows: list[dict], now: float) -> list[dict]: + """Returns: + rows without those older than 90 days, every 15-minute bucket that + ended more than 7 days ago folded into one mean at its start, the + newer rows unchanged and in order. + """ + buckets: dict[int, list[dict]] = {} + recent = [] + for row in rows: + if not timed(row) or row["t"] < now - COARSE: + continue + start = int(row["t"] // BUCKET * BUCKET) + if start + BUCKET <= now - FINE: + buckets.setdefault(start, []).append(row) + else: + recent.append(row) + return [mean_row(start, buckets[start]) for start in sorted(buckets)] + recent + + +def read_rows(path: Path) -> list[dict]: + rows = [] + for line in read_text(path).splitlines(): + try: + row = json.loads(line) + except (ValueError, RecursionError): + continue + if isinstance(row, dict): + rows.append(row) + return rows + + +def encoded(rows: list[dict]) -> str: + return "".join(json.dumps(row, separators=(",", ":")) + "\n" for row in rows) + + +def append_rows(path: Path, rows: list[dict]) -> None: + if rows: + with path.open("a", encoding="utf-8") as handle: + handle.write(encoded(rows)) + + +def replace(path: Path, text: str) -> None: + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False) as handle: + try: + handle.write(text) + except BaseException: + os.unlink(handle.name) + raise + os.replace(handle.name, path) + + +def rewrite(path: Path, keep: Callable[[list[dict], float], list[dict]], now: float) -> None: + rows = read_rows(path) + kept = keep(rows, now) + if kept != rows: + replace(path, encoded(kept)) + + +def record(state: Path, row: dict, now: float) -> None: + path = state / SAMPLES + append_rows(path, [row]) + if now % BUCKET < INTERVAL: + rewrite(path, retain, now) + + +def claim_names(council: Path) -> dict[str, str]: + names = {} + for pattern in CLAIMS: + for path in sorted(council.glob(pattern)): + data = read_json(path) + if isinstance(data, dict) and isinstance(data.get("session"), str) and isinstance(data.get("name"), str) and data["name"].strip(): + names[data["session"]] = data["name"].strip() + return names + + +def epoch(value: object) -> float | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.datetime.fromisoformat(value) + except ValueError: + return None + return (parsed if parsed.tzinfo else parsed.replace(tzinfo=datetime.UTC)).timestamp() + + +def count(value: object) -> int: + return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0 + + +def assistant(raw: bytes) -> tuple[dict, dict, dict] | None: + """Returns: + (row, message, usage) of an assistant transcript line whose message + carries a usage map and whose row a session id, else None. + """ + try: + row = json.loads(raw) + except (ValueError, RecursionError): + return None + message = row.get("message") if isinstance(row, dict) and row.get("type") == "assistant" else None + fields = message.get("usage") if isinstance(message, dict) else None + return (row, message, fields) if isinstance(fields, dict) and isinstance(row.get("sessionId"), str) else None + + +def message_key(ident: object, request: object) -> str: + """Returns: + A short digest of (message id, request id), "" without a message id. + """ + if not isinstance(ident, str) or not ident: + return "" + text = f"{ident}\0{request if isinstance(request, str) else ''}" + return base64.urlsafe_b64encode(hashlib.blake2b(text.encode("utf-8"), digest_size=KEY_BYTES).digest()).decode("ascii") + + +def usage(raw: bytes) -> tuple[str, float, str, str, tuple[int, ...]] | None: + """Returns: + (message key, time, session, model, counts in COUNTS order) of an + assistant transcript line with a usage field, a time and a model, + else None. + """ + found = assistant(raw) + if found is None: + return None + row, message, fields = found + ts, model = epoch(row.get("timestamp")), message.get("model") + if ts is None or not isinstance(model, str): + return None + return message_key(message.get("id"), row.get("requestId")), ts, row["sessionId"], model, tuple(count(fields.get(field)) for _, field in COUNTS) + + +def first_sight(key: str, ts: float, seen: dict[int, set[str]], keys: dict[int, set[str]]) -> bool: + """Returns: + Whether the message key at ts is in neither seen nor keys at its hour + or a neighbouring one, then files it in keys at its hour; a line + without a key is always a first sight. The neighbours catch a copy a + fork wrote seconds after the original across the hour. + """ + if not key: + return True + hour = int(ts // HOUR * HOUR) + known = any(key in found.get(near, ()) for found in (seen, keys) for near in (hour - HOUR, hour, hour + HOUR)) + keys.setdefault(hour, set()).add(key) + return not known + + +def scan(path: Path, start: int, budget: int, seen: dict[int, set[str]]) -> tuple[int, list[tuple[float, str, str, tuple[int, ...]]], dict[int, set[str]]]: + """Returns: + The offset after the last complete line read from start, at most + about budget bytes on, the usage of every message seen holds no key + of, and per hour the keys of every message read. + """ + found = [] + keys: dict[int, set[str]] = {} + position = start + with path.open("rb") as handle: + handle.seek(start) + for raw in handle: + if not raw.endswith(b"\n") or position - start >= budget: + break + position += len(raw) + read = usage(raw) if b'"usage"' in raw else None + if read is not None and first_sight(read[0], read[1], seen, keys) and any(read[4]): + found.append(read[1:]) + return position, found, keys + + +def token_rows(sums: dict[tuple, list[int]]) -> list[dict]: + return [{"t": t, "agent": agent, "model": model, **dict(zip(NAMES, counts, strict=True))} for (t, agent, model), counts in sorted(sums.items())] + + +def add(sums: dict[tuple, list[int]], key: tuple, counts: tuple[int, ...]) -> None: + total = sums.setdefault(key, [0] * len(COUNTS)) + for index, value in enumerate(counts): + total[index] += value + + +def compact(rows: list[dict], now: float) -> list[dict]: + sums: dict[tuple, list[int]] = {} + for row in rows: + if timed(row) and row["t"] >= now - COARSE and isinstance(row.get("agent"), str) and isinstance(row.get("model"), str): + add(sums, (row["t"], row["agent"], row["model"]), tuple(count(row.get(name)) for name in NAMES)) + return token_rows(sums) + + +def start_of(entry: object, info: os.stat_result, now: float) -> int: + if isinstance(entry, int) and not isinstance(entry, bool) and entry >= 0: + return entry if entry <= info.st_size else 0 + return info.st_size if info.st_mtime < now - COARSE else 0 + + +def read_transcript(path: Path, entry: object, budget: int, now: float, seen: dict[int, set[str]]) -> tuple[int, int, list, dict[int, set[str]]]: + """Returns: + (start, end, usages, keys per hour) of path read from the offset + entry records, without reading once budget is spent. + """ + info = path.stat() + start = start_of(entry, info, now) + if budget <= 0 or start >= info.st_size: + return start, start, [], {} + return start, *scan(path, start, budget, seen) + + +def seen_of(value: object) -> dict[int, set[str]]: + if not isinstance(value, dict): + return {} + return {int(hour): {key for key in keys if isinstance(key, str)} for hour, keys in value.items() if hour.isdigit() and isinstance(keys, list)} + + +def kept_seen(seen: dict[int, set[str]], touched: set[int], now: float) -> dict[str, list[str]]: + floor = now // HOUR * HOUR - HOUR + return {str(hour): sorted(keys) for hour, keys in sorted(seen.items()) if hour in touched or hour >= floor} + + +def count_tokens(transcripts: Path, council: Path, state: Path, now: float, budget: int) -> list[dict]: + """Returns: + The rows appended to /tokens.jsonl by this run. + """ + stored = read_json(state / OFFSETS) + offsets = stored.get("files") if isinstance(stored, dict) and isinstance(stored.get("files"), dict) else {} + seen = seen_of(stored.get("seen") if isinstance(stored, dict) else None) + names = claim_names(council) + sums: dict[tuple, list[int]] = {} + files, touched = {}, set() + for path in sorted(transcripts.rglob("*.jsonl")): + key = path.relative_to(transcripts).as_posix() + try: + start, end, found, keys = read_transcript(path, offsets.get(key), budget, now, seen) + except OSError: + if key in offsets: + files[key] = offsets[key] + continue + files[key] = end + budget -= end - start + for hour, read in keys.items(): + seen.setdefault(hour, set()).update(read) + touched.add(hour) + for ts, session, model, counts in found: + add(sums, (int(ts // HOUR * HOUR), names.get(session) or session[:8], model), counts) + rows = token_rows(sums) + append_rows(state / TOKENS, rows) + kept = {"files": files, "seen": kept_seen(seen, touched, now)} + if kept != stored: + replace(state / OFFSETS, json.dumps(kept, sort_keys=True, separators=(",", ":"))) + if now % HOUR < INTERVAL: + rewrite(state / TOKENS, compact, now) + return rows + + +def main() -> int: + parser = argparse.ArgumentParser(description="Sample the host's resources and count the transcripts' tokens for the Agent Orchestrator.") + commands = parser.add_subparsers(dest="command", required=True) + sampling, counting = commands.add_parser("sample"), commands.add_parser("tokens") + for sub in (sampling, counting): + sub.add_argument("--state", type=Path, required=True) + sub.add_argument("--council", type=Path, required=True) + sampling.add_argument("--proc", type=Path, required=True) + sampling.add_argument("--sys", type=Path, required=True) + counting.add_argument("--transcripts", type=Path, required=True) + args = parser.parse_args() + args.state.mkdir(parents=True, exist_ok=True) + if args.command == "sample": + row = sample(args.proc, args.sys, args.council, command, args.state) + record(args.state, row, row["t"]) + else: + count_tokens(args.transcripts, args.council, args.state, time.time(), BUDGET) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/orchestrator/kvb_orchestrator/handler.py b/orchestrator/kvb_orchestrator/handler.py index 5b35909..2a05cf5 100644 --- a/orchestrator/kvb_orchestrator/handler.py +++ b/orchestrator/kvb_orchestrator/handler.py @@ -2,12 +2,14 @@ from __future__ import annotations import re import secrets +import time from collections.abc import Callable from http import HTTPStatus from urllib.parse import urlsplit from .engines import STT_LIMIT from .exchange import PLAIN_CSP, Exchange +from .resources import RESOURCES, sampled BODY_LIMIT = 1 << 20 UPLOAD_LIMIT = 50 << 20 @@ -107,6 +109,8 @@ class Handler(Exchange): self.answer(lambda: hub.events(self.query()), HTTPStatus.OK) elif path == "/api/tree": self.answer(hub.repositories, HTTPStatus.OK) + elif path == "/api/resources": + self.answer(lambda: sampled(hub.kept / RESOURCES, hub.span(self.query(), time.time())), HTTPStatus.OK) elif path == "/api/subagents": self.answer(lambda: hub.subagents(self.query()), HTTPStatus.OK) elif path == "/api/runner/jobs": diff --git a/orchestrator/kvb_orchestrator/resources.py b/orchestrator/kvb_orchestrator/resources.py new file mode 100644 index 0000000..948d006 --- /dev/null +++ b/orchestrator/kvb_orchestrator/resources.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from pathlib import Path + +from .records import epoch, parse_line, read_regular + +RESOURCES = "resources" +SAMPLES = "samples.jsonl" +TOKENS = "tokens.jsonl" +HOUR = 3600 +FILE_LIMIT = 64 << 20 + + +def rows_between(path: Path, begin: float, end: float, width: float) -> list[dict]: + try: + lines = read_regular(path, FILE_LIMIT).splitlines() + except (OSError, ValueError): + return [] + found = [] + for row in map(parse_line, lines): + t = None if row is None else epoch(row.get("t")) + if t is not None and begin <= t + width and t <= end: + found.append(row) + return found + + +def sampled(folder: Path, span: tuple[float, float]) -> dict: + """Args: + folder: the host sampler's directory, mounted read-only. + span: (begin, end) of the page's time window in epoch seconds. + + Returns: + {"since", "until", "samples", "tokens"}: the rows of samples.jsonl + within the span and those of tokens.jsonl whose hour overlaps it, as + the host unit wrote them; the hub reads nothing of the host itself. + """ + begin, end = span + return {"since": begin, "until": end, "samples": rows_between(folder / SAMPLES, begin, end, 0), "tokens": rows_between(folder / TOKENS, begin, end, HOUR)} diff --git a/orchestrator/systemd/kvb-orchestrator-resources.service b/orchestrator/systemd/kvb-orchestrator-resources.service new file mode 100644 index 0000000..81d3e28 --- /dev/null +++ b/orchestrator/systemd/kvb-orchestrator-resources.service @@ -0,0 +1,12 @@ +[Unit] +Description=Agent Orchestrator resources: disks, CPU, RAM, swap, load, GPU, the queue and the transcripts' tokens, into the files the orchestrator reads + +[Service] +Type=oneshot +Environment=PATH=/usr/bin:/bin +EnvironmentFile=%h/.local/share/kvb-orchestrator/.env +ExecStart=/usr/bin/env python3 %h/.local/share/kvb-orchestrator/client/resources.py sample --state ${KVB_ORCHESTRATOR_KEPT}/resources --council ${KVB_ORCHESTRATOR_COUNCIL} --proc /proc --sys /sys +ExecStart=/usr/bin/env python3 %h/.local/share/kvb-orchestrator/client/resources.py tokens --state ${KVB_ORCHESTRATOR_KEPT}/resources --council ${KVB_ORCHESTRATOR_COUNCIL} --transcripts ${KVB_ORCHESTRATOR_PROJECTS} +Nice=10 +IOSchedulingClass=idle +TimeoutStartSec=300 diff --git a/orchestrator/systemd/kvb-orchestrator-resources.timer b/orchestrator/systemd/kvb-orchestrator-resources.timer new file mode 100644 index 0000000..094c8ba --- /dev/null +++ b/orchestrator/systemd/kvb-orchestrator-resources.timer @@ -0,0 +1,9 @@ +[Unit] +Description=Agent Orchestrator resources every minute + +[Timer] +OnCalendar=*-*-* *:*:00 +AccuracySec=1s + +[Install] +WantedBy=timers.target diff --git a/orchestrator/templates/index.html.j2 b/orchestrator/templates/index.html.j2 index f3640e2..c426963 100644 --- a/orchestrator/templates/index.html.j2 +++ b/orchestrator/templates/index.html.j2 @@ -406,6 +406,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi .float-tooltip-kap { position: absolute; z-index: 9; padding: 4px 8px; border-radius: 6px; background: var(--glass-card); color: var(--text); font-size: var(--fs-sm); white-space: nowrap; pointer-events: none; box-shadow: var(--shadow); } {% include "style-hierarchy.css" %} +{% include "style-resources.css" %} + {% include "style-inbox.css" %} {% include "style-ui.css" %} @@ -455,6 +457,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi {% include "bubbles.js" %} +{% include "resources.js" %} + {% include "nav.js" %} {% include "popup.js" %} @@ -560,6 +564,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi {% include "page-alerts.js" %} +{% include "page-resources.js" %} + {% include "page-rules.js" %} {% include "page-keyboard.js" %} diff --git a/orchestrator/templates/nav.js b/orchestrator/templates/nav.js index 5550599..3d3020e 100644 --- a/orchestrator/templates/nav.js +++ b/orchestrator/templates/nav.js @@ -1,6 +1,6 @@ const Nav = (() => { "use strict"; - const TABS = [["stream", "Stream"], ["agents", "Agents"], ["artifacts", "Artifacts"], ["tasks", "Tasks"]]; + const TABS = [["stream", "Stream"], ["agents", "Agents"], ["artifacts", "Artifacts"], ["tasks", "Tasks"], ["resources", "Resources"]]; const SUBS = {stream: [["inbox", "Inbox"], ["rules", "Firewall"]]}; const VIEWS = [["orbit", "Orbit"], ["chord", "Chord"], ["swimlane", "Swimlane"], ["tree", "Tree"], ["3d", "3D"]]; const TASK_VIEWS = [["list", "List"], ["board", "Board"], ["rasci", "RASCI"], ["gantt", "Gantt"], ["burn", "Burndown and burnup"], ["3d", "3D"]]; @@ -9,6 +9,7 @@ const Nav = (() => { stream: "bi-activity", inbox: "bi-inbox", rules: "bi-card-checklist", agents: "bi-people", artifacts: "bi-git", tasks: "bi-list-check", orbit: "bi-globe2", chord: "bi-bezier2", swimlane: "bi-bar-chart-steps", tree: "bi-diagram-3", "3d": "bi-badge-3d", hierarchy: "bi-diagram-2", list: "bi-list-task", board: "bi-kanban", rasci: "bi-grid-3x3", gantt: "bi-calendar-range", burn: "bi-graph-down-arrow", + resources: "bi-speedometer2", }; const OLD = {graph: ["agents", "orbit"], tree: ["artifacts", "tree"]}; const viewsOf = (screen) => (screen === "tasks" ? TASK_VIEWS : [...VIEWS, ...(OWN[screen] || [])]); diff --git a/orchestrator/templates/page-resources.js b/orchestrator/templates/page-resources.js new file mode 100644 index 0000000..1e93721 --- /dev/null +++ b/orchestrator/templates/page-resources.js @@ -0,0 +1,174 @@ + const RESOURCE = {left: 46, right: 10, top: 6, bottom: 6, axis: 18, poll: 5000, stale: 60, tones: 6, hour: 3600, narrow: 600}; + const RESOURCE_PANELS = [["disks", "bi-hdd-stack", "Disks"], ["host", "bi-cpu", "CPU, RAM and swap"], ["gpu", "bi-gpu-card", "GPU and VRAM"], ["tokens", "bi-coin", "Tokens per agent"]]; + const RESOURCE_ZONES = {idle: "Idle: below 30 %", saturated: "Saturated: above 90 %"}; + const RESOURCE_MARKS = { + wait: ["#bi-hourglass-split", (mark) => `${mark.peak} jobs waited while CPU and RAM stayed under 50 %`], + asleep: ["#bi-moon", (mark) => `${mark.name} asleep`], + unknown: ["#bi-question-circle", (mark) => `${mark.name} unknown: no reading`], + }; + const resourceView = {data: null, dom: null, busy: false, key: "", at: 0, timer: 0}; + const resourceKey = () => JSON.stringify([ui.preset, ui.from, ui.to, ui.asOf]); + const compactCount = (value) => (value >= 1e6 ? `${fixed(value / 1e6)}M` : value >= 1e3 ? `${fixed(value / 1e3)}k` : String(Math.round(value))); + const resourceLabel = (line) => (line.kind === "vram" ? `${line.name} VRAM` : line.name); + + function resourcesScreen() { + const count = h("span", {class: "count", title: "Samples", "aria-label": "Samples"}, "0"); + const panels = RESOURCE_PANELS.map(([kind, name, label]) => { + const legend = h("div", {class: "res-legend"}); + const plot = h("div", {class: "res-plot"}); + const panel = h("section", {class: `card res-panel res-${kind}`, "data-panel": kind, "aria-label": label}, + h("div", {class: "res-head"}, h("span", {class: "res-title", title: label, "aria-label": label}, icon(name)), legend), plot); + return {kind, label, legend, plot, panel}; + }); + resourceView.dom = {count, panels}; + if (!resourceView.timer) { + resourceView.timer = window.setInterval(pollResources, RESOURCE.poll); + window.addEventListener("resize", drawResources); + } + window.setTimeout(() => { + drawResources(); + fetchResources(); + }, 0); + const reload = iconBtn("bi-arrow-clockwise", "Reload", "btn small soft", fetchResources, "Reload: read the sampler files again"); + return h("div", {class: "graph resources"}, + h("div", {class: "card controls"}, h("div", {class: "toolbar", role: "toolbar", "aria-label": "Resources"}, reload, count)), + h("div", {class: "res-panels"}, panels.map((entry) => entry.panel))); + } + + function pollResources() { + const dom = resourceView.dom; + if (!dom || !dom.count.isConnected || document.hidden || resourceView.busy) { + return; + } + if (resourceKey() !== resourceView.key || nowSeconds() - resourceView.at >= RESOURCE.stale) { + fetchResources(); + } + } + + async function fetchResources() { + if (resourceView.busy) { + return; + } + resourceView.busy = true; + resourceView.key = resourceKey(); + try { + const response = await fetch(`/api/resources?${windowQuery()}`, {cache: "no-store"}); + if (response.ok) { + resourceView.data = await response.json(); + } + } catch { + ui.offline = true; + } + resourceView.at = nowSeconds(); + resourceView.busy = false; + drawResources(); + } + + function resourceFrame(entry, span, size) { + const axis = entry.kind === "tokens"; + const bottom = size.height - (axis ? RESOURCE.axis : RESOURCE.bottom); + const x = (t) => RESOURCE.left + ((t - span.from) / (span.to - span.from)) * (size.width - RESOURCE.left - RESOURCE.right); + const svg = s("svg", {class: `chart res-chart res-${entry.kind}`, "data-panel": entry.kind, "data-from": span.from, "data-to": span.to, viewBox: `0 0 ${size.width} ${size.height}`, role: "img", "aria-label": entry.label}); + const ticks = size.width < RESOURCE.narrow ? 2 : 4; + for (let index = 0; index <= ticks; index += 1) { + const t = span.from + ((span.to - span.from) * index) / ticks; + svg.append(s("line", {class: "grid", x1: fixed(x(t)), x2: fixed(x(t)), y1: RESOURCE.top, y2: fixed(bottom)})); + if (axis) { + svg.append(s("text", {class: "axis", x: fixed(x(t)), y: size.height - 4, "text-anchor": index === 0 ? "start" : index === ticks ? "end" : "middle"}, (span.to - span.from > DAY ? absolute : clock).format(stampOf(t)))); + } + } + return {svg, x, top: RESOURCE.top, bottom, right: size.width - RESOURCE.right}; + } + + function resourceScale(frame, values, label) { + const y = (value) => frame.top + (1 - value / values[values.length - 1]) * (frame.bottom - frame.top); + for (const value of values) { + frame.svg.append(s("line", {class: "grid", x1: RESOURCE.left, x2: fixed(frame.right), y1: fixed(y(value)), y2: fixed(y(value))})); + frame.svg.append(s("text", {class: "axis", x: RESOURCE.left - 6, y: fixed(y(value)), dy: "0.35em", "text-anchor": "end"}, label(value))); + } + return y; + } + + function resourceMarks(frame, marks, span) { + for (const mark of marks) { + const left = frame.x(Math.max(mark.from, span.from)); + const right = frame.x(Math.min(mark.to, span.to)); + if (right <= left) { + continue; + } + const [href, words] = RESOURCE_MARKS[mark.kind]; + frame.svg.append(s("rect", {class: `res-mark ${mark.kind}`, "data-from": mark.from, "data-to": mark.to, "data-peak": mark.peak, x: fixed(left), width: fixed(Math.max(right - left, 2)), y: frame.top, height: fixed(frame.bottom - frame.top)}, s("title", {}, words(mark)))); + frame.svg.append(s("use", {class: `res-icon ${mark.kind}`, href, x: fixed(left + 2), y: frame.top + 2, width: 11, height: 11})); + } + } + + function resourcePercent(entry, list, marks, span, size) { + const frame = resourceFrame(entry, span, size); + const y = resourceScale(frame, [0, 50, 100], (value) => `${value} %`); + for (const zone of Resources.ZONES) { + frame.svg.append(s("rect", {class: `band ${zone.kind}`, "data-zone": zone.kind, x: RESOURCE.left, width: fixed(frame.right - RESOURCE.left), y: fixed(y(zone.to)), height: fixed(y(zone.from) - y(zone.to))}, s("title", {}, RESOURCE_ZONES[zone.kind]))); + } + resourceMarks(frame, marks, span); + const tones = new Map(); + const legend = []; + for (const line of list) { + if (!tones.has(line.name)) { + tones.set(line.name, tones.size % RESOURCE.tones); + } + const tone = `res-c${tones.get(line.name)} k-${line.kind}`; + const last = [...line.points].reverse().find(([, value]) => value !== null); + const words = `${resourceLabel(line)} ${dash(last === undefined ? null : last[1])} %`; + for (const part of Resources.segments(line.points)) { + const d = part.map(([t, value], index) => `${index ? "L" : "M"}${fixed(frame.x(t))},${fixed(y(Math.min(Math.max(value, 0), 100)))}`).join(" "); + frame.svg.append(s("path", {class: `res-line ${tone}`, "data-series": line.name, "data-kind": line.kind, "data-points": part.length, d: part.length === 1 ? `${d} h1` : d}, s("title", {}, words))); + } + legend.push(h("span", {class: `res-key ${tone}`, "data-series": line.name, "data-kind": line.kind}, words)); + } + return {svg: frame.svg, legend}; + } + + function resourceTokens(entry, stack, span, size) { + const frame = resourceFrame(entry, span, size); + const y = resourceScale(frame, [0, Math.max(stack.top, 1)], compactCount); + const tones = new Map(stack.agents.map((agent, index) => [agent, index % RESOURCE.tones])); + for (const bar of stack.bars) { + const left = frame.x(Math.max(bar.t, span.from)); + const right = frame.x(Math.min(bar.t + RESOURCE.hour, span.to)); + if (right <= left) { + continue; + } + for (const part of bar.parts) { + const words = `${part.agent} · ${absolute.format(stampOf(bar.t))}: ${part.total} tokens (input ${part.input}, output ${part.output}, cache read ${part.cache_read}, cache write ${part.cache_write}) · ${part.models.join(", ")}`; + frame.svg.append(s("rect", {class: `res-bar res-c${tones.get(part.agent)}`, "data-agent": part.agent, "data-t": bar.t, "data-total": part.total, x: fixed(left), width: fixed(Math.max(right - left - 1, 1)), y: fixed(y(part.to)), height: fixed(Math.max(y(part.from) - y(part.to), 0.5))}, s("title", {}, words))); + } + } + const legend = stack.agents.map((agent) => h("span", {class: `res-key res-c${tones.get(agent)}`, "data-series": agent}, `${agent} ${compactCount(stack.totals[agent])}`)); + return {svg: frame.svg, legend}; + } + + function drawResources() { + const dom = resourceView.dom; + const data = resourceView.data; + if (!dom || !dom.count.isConnected || data === null) { + return; + } + const span = {from: data.since, to: Math.max(data.until, data.since + 60)}; + const lines = Resources.lines(data.samples); + const marks = { + disks: [], + host: Resources.waits(data.samples).map((mark) => ({...mark, kind: "wait"})), + gpu: ["asleep", "unknown"].flatMap((kind) => Resources[kind](data.samples).flatMap((gpu) => gpu.spans.map((mark) => ({...mark, kind, name: gpu.name})))), + }; + dom.count.textContent = String(data.samples.length); + for (const entry of dom.panels) { + const box = entry.plot.getBoundingClientRect(); + const size = {width: Math.round(box.width), height: Math.round(box.height)}; + if (size.width < RESOURCE.left + RESOURCE.right + 10 || size.height < RESOURCE.axis + 10) { + entry.plot.replaceChildren(); + continue; + } + const drawn = entry.kind === "tokens" ? resourceTokens(entry, Resources.stack(data.tokens), span, size) : resourcePercent(entry, lines[entry.kind], marks[entry.kind], span, size); + entry.plot.replaceChildren(drawn.svg); + entry.legend.replaceChildren(...drawn.legend); + } + } diff --git a/orchestrator/templates/page-screens.js b/orchestrator/templates/page-screens.js index 4de18a5..5177309 100644 --- a/orchestrator/templates/page-screens.js +++ b/orchestrator/templates/page-screens.js @@ -55,7 +55,7 @@ } function screenOf() { - return {stream: streamScreen, inbox: tasksScreen, rules: rulesScreen, agents: agentsScreen, artifacts: artifactsScreen, tasks: tasksScreen}[ui.screen](); + return {stream: streamScreen, inbox: tasksScreen, rules: rulesScreen, agents: agentsScreen, artifacts: artifactsScreen, tasks: tasksScreen, resources: resourcesScreen}[ui.screen](); } function render(force) { @@ -73,7 +73,7 @@ const main = document.getElementById("main"); main.setAttribute("aria-labelledby", `tab-${ui.screen}`); main.replaceChildren(screenOf()); - document.body.classList.toggle("fit", ["tasks", "inbox"].includes(ui.screen) || (["agents", "artifacts"].includes(ui.screen) && !["tree", "hierarchy"].includes(ui.views[ui.screen]))); + document.body.classList.toggle("fit", ["tasks", "inbox", "resources"].includes(ui.screen) || (["agents", "artifacts"].includes(ui.screen) && !["tree", "hierarchy"].includes(ui.views[ui.screen]))); drawn.screen = ui.screen; drawn.sig = sig; if (g3.host && !g3.host.isConnected) { diff --git a/orchestrator/templates/resources.js b/orchestrator/templates/resources.js new file mode 100644 index 0000000..9822af1 --- /dev/null +++ b/orchestrator/templates/resources.js @@ -0,0 +1,144 @@ +const Resources = (() => { + "use strict"; + const IDLE = 30; + const SATURATED = 90; + const CALM = 50; + const GAP = 1800; + const STEP = 60; + const COUNTS = ["input", "output", "cache_read", "cache_write"]; + const ZONES = [{kind: "idle", from: 0, to: IDLE}, {kind: "saturated", from: SATURATED, to: 100}]; + const finite = (value) => typeof value === "number" && Number.isFinite(value); + const share = (part, whole) => (finite(part) && finite(whole) && whole > 0 ? Math.round((1000 * part) / whole) / 10 : null); + const listOf = (row, key) => (Array.isArray(row[key]) ? row[key].filter((item) => item && typeof item === "object") : []); + const timed = (samples) => samples.filter((row) => row && finite(row.t)); + + function series(samples, pick) { + const found = new Map(); + for (const row of timed(samples)) { + for (const [name, kind, value] of pick(row)) { + const key = `${kind}\u0000${name}`; + if (!found.has(key)) { + found.set(key, {name: String(name), kind, points: []}); + } + found.get(key).points.push([row.t, finite(value) ? value : null]); + } + } + return [...found.values()]; + } + + function lines(samples) { + return { + disks: series(samples, (row) => listOf(row, "disks").map((disk) => [disk.mount, "disk", share(disk.used, disk.total)])), + host: series(samples, (row) => [["CPU", "host", row.cpu], ["RAM", "host", row.ram], ["Swap", "host", row.swap]]), + gpu: series(samples, (row) => listOf(row, "gpus").flatMap((gpu) => [[gpu.name, "util", gpu.asleep ? null : gpu.util], [gpu.name, "vram", gpu.asleep ? null : share(gpu.vram_used, gpu.vram_total)]])), + }; + } + + function segments(points) { + const found = []; + let current = []; + let last = null; + for (const [t, value] of points) { + if (current.length && (value === null || t - last > GAP)) { + found.push(current); + current = []; + } + if (value !== null) { + current.push([t, value]); + } + last = t; + } + if (current.length) { + found.push(current); + } + return found; + } + + function spans(samples, test, value) { + const rows = timed(samples); + const found = []; + let open = null; + rows.forEach((row, index) => { + if (open && index && row.t - rows[index - 1].t > GAP) { + found.push(open); + open = null; + } + if (!test(row)) { + if (open) { + found.push(open); + } + open = null; + return; + } + const next = rows[index + 1]; + const end = row.t + (next && next.t - row.t <= GAP ? next.t - row.t : STEP); + const peak = value(row); + open = open ? {from: open.from, to: end, peak: Math.max(open.peak, peak)} : {from: row.t, to: end, peak}; + }); + if (open) { + found.push(open); + } + return found; + } + + function waits(samples) { + const waiting = (row) => (row.queue && finite(row.queue.waiting) ? row.queue.waiting : 0); + return spans(samples, (row) => waiting(row) > 0 && finite(row.cpu) && finite(row.ram) && row.cpu < CALM && row.ram < CALM, waiting); + } + + function gpuSpans(samples, test) { + const names = [...new Set(timed(samples).flatMap((row) => listOf(row, "gpus").filter(test).map((gpu) => String(gpu.name))))]; + return names.map((name) => ({name, spans: spans(samples, (row) => listOf(row, "gpus").some((gpu) => String(gpu.name) === name && test(gpu)), () => 1)})); + } + + const asleep = (samples) => gpuSpans(samples, (gpu) => gpu.asleep === true); + const unknown = (samples) => gpuSpans(samples, (gpu) => gpu.asleep !== true && !finite(gpu.util) && !finite(gpu.vram_used)); + + function stack(tokens) { + const totals = new Map(); + const hours = new Map(); + for (const row of tokens) { + if (!row || !finite(row.t) || typeof row.agent !== "string") { + continue; + } + const counts = COUNTS.map((name) => (finite(row[name]) && row[name] > 0 ? row[name] : 0)); + const sum = counts.reduce((total, value) => total + value, 0); + if (!totals.has(row.agent)) { + totals.set(row.agent, 0); + } + totals.set(row.agent, totals.get(row.agent) + sum); + if (!hours.has(row.t)) { + hours.set(row.t, new Map()); + } + const hour = hours.get(row.t); + if (!hour.has(row.agent)) { + hour.set(row.agent, {agent: row.agent, total: 0, models: [], ...Object.fromEntries(COUNTS.map((name) => [name, 0]))}); + } + const entry = hour.get(row.agent); + COUNTS.forEach((name, index) => { + entry[name] += counts[index]; + }); + entry.total += sum; + if (typeof row.model === "string" && !entry.models.includes(row.model)) { + entry.models.push(row.model); + } + } + const agents = [...totals].sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0])).map(([agent]) => agent); + const bars = [...hours.keys()].sort((a, b) => a - b).map((t) => { + let base = 0; + const parts = agents.filter((agent) => hours.get(t).has(agent)).map((agent) => { + const entry = hours.get(t).get(agent); + const part = {...entry, from: base, to: base + entry.total}; + base += entry.total; + return part; + }); + return {t, total: base, parts}; + }); + return {agents, totals: Object.fromEntries(totals), bars, top: Math.max(0, ...bars.map((bar) => bar.total))}; + } + + return {IDLE, SATURATED, CALM, GAP, COUNTS, ZONES, lines, segments, spans, waits, asleep, unknown, stack}; +})(); +if (typeof module === "object" && module.exports) { + module.exports = Resources; +} diff --git a/orchestrator/templates/style-resources.css b/orchestrator/templates/style-resources.css new file mode 100644 index 0000000..6d8a99a --- /dev/null +++ b/orchestrator/templates/style-resources.css @@ -0,0 +1,27 @@ +body.fit .graph.resources > .res-panels { flex: 1; } +.res-panels { display: flex; flex-direction: column; gap: 8px; min-height: 0; } +.res-panel { display: flex; flex: 1 1 0; flex-direction: column; gap: 2px; min-height: 72px; padding: 6px 10px; overflow: hidden; } +.res-head { display: flex; align-items: center; gap: 4px 12px; min-width: 0; overflow: hidden; color: var(--muted); font-size: var(--fs-xs); white-space: nowrap; } +.res-title { display: inline-flex; color: var(--text); font-size: var(--fs-md); } +.res-legend { display: flex; gap: 4px 12px; min-width: 0; overflow: hidden; font-variant-numeric: tabular-nums; } +.res-key { display: inline-flex; align-items: center; gap: 5px; } +.res-key::before { content: ""; width: 12px; height: 3px; border-radius: 2px; background: var(--c); } +.res-key.k-vram::before { background: repeating-linear-gradient(90deg, var(--c) 0 4px, transparent 4px 6px); } +.res-plot { position: relative; flex: 1; min-height: 0; } +.res-plot > svg { position: absolute; inset: 0; width: 100%; height: 100%; } +svg .band.idle { fill: var(--ok-soft); } +svg .band.saturated { fill: var(--bad-soft); } +svg .res-line { fill: none; stroke: var(--c); stroke-width: 1.8; stroke-linecap: round; stroke-linejoin: round; } +svg .res-line.k-vram { stroke-dasharray: 5 3; } +svg .res-mark.wait { fill: var(--warn-soft); stroke: var(--warn); stroke-dasharray: 3 3; } +svg .res-mark.asleep { fill: var(--neutral-soft); } +svg .res-mark.unknown { fill: var(--neutral-soft); stroke: var(--muted); stroke-dasharray: 2 3; } +svg .res-icon.wait { fill: var(--warn); } +svg .res-icon.asleep, svg .res-icon.unknown { fill: var(--muted); } +svg .res-bar { fill: var(--c); } +.res-c0 { --c: var(--info); } +.res-c1 { --c: var(--ok); } +.res-c2 { --c: var(--accent); } +.res-c3 { --c: var(--warn); } +.res-c4 { --c: var(--violet); } +.res-c5 { --c: var(--bad); } diff --git a/scripts/orchestrator.sh b/scripts/orchestrator.sh index 67f4552..78ef794 100644 --- a/scripts/orchestrator.sh +++ b/scripts/orchestrator.sh @@ -4,7 +4,8 @@ # write its .env, build and start the container, register the MCP # server for Claude Code, build pkgmgr from the newest signed tag of # PKGMGR_REPO and enable the tree timer and the cleanup path unit, -# enable the sessions timer and the offices path unit; re-enable the runner and re-register +# enable the sessions timer and the offices path unit, enable the +# resources timer; re-enable the runner and re-register # the hook when the migration took them over; an enabled runner gets # the fresh unit and a restart # migrate move every trace of the earlier name to the kvb-orchestrator names @@ -40,6 +41,7 @@ TREE_UNIT="${NAME}-tree" CLEANUP_UNIT="${NAME}-cleanup" OFFICES_UNIT="${NAME}-offices" SESSIONS_UNIT="${NAME}-sessions" +RESOURCES_UNIT="${NAME}-resources" INSTALLED_SKILLS="${HOME}/.claude/skills" HOST_SCRIPTS=(kvb-org-organize/scripts/council.py kvb-org-organize/scripts/nofollow.py kvb-org-rasci/scripts/rasci.py) TREE_DIR="${HOME}/.local/state/pkgmgr/tree" @@ -155,7 +157,7 @@ install_hub() { cp -a "${REPO_ROOT}/orchestrator/." "${HUB}/" # Python loads a cached .pyc whose header matches the source mtime and size, and cp -a keeps the mtime: one planted in the sandbox-writable tree would run on the host find "${HUB}" -name __pycache__ -type d -prune -exec rm -rf {} + - chmod +x "${HUB}/client/permission_hook.py" "${HUB}/client/runner.py" "${HUB}/client/approval.py" "${HUB}/client/cleanup.py" "${HUB}/client/offices.py" "${HUB}/client/sessions.py" + chmod +x "${HUB}/client/permission_hook.py" "${HUB}/client/runner.py" "${HUB}/client/approval.py" "${HUB}/client/cleanup.py" "${HUB}/client/offices.py" "${HUB}/client/sessions.py" "${HUB}/client/resources.py" log "skills: the PermissionRequest hook client is ${HUB}/client/permission_hook.py" log "skills: approvals need a pinned passkey: settings, 'Passkey registrieren', then 'make orchestrator-pin CREDENTIAL='; until then every executor refuses" printf 'KVB_ORCHESTRATOR_USER=%s:%s\nKVB_ORCHESTRATOR_STATE=%s\nKVB_ORCHESTRATOR_COUNCIL=%s\nKVB_ORCHESTRATOR_SKILLS=%s\nKVB_ORCHESTRATOR_PROJECTS=%s\nKVB_ORCHESTRATOR_TREE=%s\nKVB_ORCHESTRATOR_TIMERS=%s\nKVB_ORCHESTRATOR_APPROVERS=%s\nKVB_ORCHESTRATOR_KEPT=%s\n' \ @@ -243,6 +245,18 @@ sessions_timer() { log "skills: ${SESSIONS_UNIT}.timer writes ${STATE}/agent-sessions.json every 15 seconds while the Agent Orchestrator page polls" } +resources_timer() { + if ! has_systemd; then + warn "skills: no systemd user session; the Resources view stays empty." + return 0 + fi + cp -f "${HUB}/systemd/${RESOURCES_UNIT}.service" "${HUB}/systemd/${RESOURCES_UNIT}.timer" "${UNITS}/" + systemctl --user daemon-reload + systemctl --user enable --now "${RESOURCES_UNIT}.timer" + systemctl --user restart "${RESOURCES_UNIT}.timer" + log "skills: ${RESOURCES_UNIT}.timer samples the host and counts the transcripts' tokens into ${HOME}/.local/state/agents-council/resources every minute" +} + skills_current() { local script for script in "${HOST_SCRIPTS[@]}"; do @@ -271,6 +285,7 @@ install_all() { install_hub tree_timer sessions_timer + resources_timer offices_unit if ((TOOK_RUNNER)) || runner_enabled; then runner || true; fi if ((TOOK_HOOK)); then @@ -354,6 +369,11 @@ check() { else todo "${SESSIONS_UNIT}.timer off: agent states wait for make orchestrator-install" fi + if [[ "$(systemctl --user is-active "${RESOURCES_UNIT}.timer" 2>/dev/null || true)" == "active" ]]; then + ok "${RESOURCES_UNIT}.timer active" + else + todo "${RESOURCES_UNIT}.timer off: the Resources view waits for make orchestrator-install" + fi if [[ "$(systemctl --user is-active "${OFFICES_UNIT}.path" 2>/dev/null || true)" == "active" ]]; then ok "${OFFICES_UNIT}.path active" else diff --git a/tests/e2e/test_orchestrator_e2e.py b/tests/e2e/test_orchestrator_e2e.py index e790560..9f9b0a0 100644 --- a/tests/e2e/test_orchestrator_e2e.py +++ b/tests/e2e/test_orchestrator_e2e.py @@ -35,9 +35,9 @@ OUT = Path(os.environ.get("E2E_OUT", "/tmp/orchestrator-phase4-shots")) WIDTHS = ((1920, 1080), (1366, 768), (390, 844)) VIEWS = ("orbit", "chord", "swimlane", "tree", "3d") TASK_VIEWS = ("list", "board", "rasci", "gantt", "burn", "3d") -SCREENS = ["stream", "stream/inbox", "stream/rules", *(f"{tab}/{view}" for tab in ("agents", "artifacts") for view in VIEWS), "agents/hierarchy", *(f"tasks/{view}" for view in TASK_VIEWS)] +SCREENS = ["stream", "stream/inbox", "stream/rules", *(f"{tab}/{view}" for tab in ("agents", "artifacts") for view in VIEWS), "agents/hierarchy", *(f"tasks/{view}" for view in TASK_VIEWS), "resources"] SUBS = ["Inbox", "Firewall"] -TABS = ["Stream", "Agents", "Artifacts", "Tasks"] +TABS = ["Stream", "Agents", "Artifacts", "Tasks", "Resources"] HELPERS = """ const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); const until = async (check, ms) => { @@ -1768,7 +1768,7 @@ class TestOrchestratorE2E(HubCase): self.assertEqual((probe["title"], probe["h1"]), ("(2) Agent Orchestrator", "Agent Orchestrator")) self.assertEqual(probe["header"], ["radio-open", "bubbles-toggle", "palette-open", "settings-open"]) self.assertEqual((probe["tabs"], probe["subs"]), (TABS, SUBS)) - self.assertEqual(probe["badges"], {"tab-stream": None, "tab-inbox": "2", "tab-rules": None, "tab-agents": "1", "tab-artifacts": None, "tab-tasks": None}) + self.assertEqual(probe["badges"], {"tab-stream": None, "tab-inbox": "2", "tab-rules": None, "tab-agents": "1", "tab-artifacts": None, "tab-tasks": None, "tab-resources": None}) self.assertEqual(probe["banner"], "MAYDAY · Docs: disk full") self.assertFalse(probe["rail"]) self.assertEqual((probe["time"], probe["replay"]), ("24 h ▾", True)) diff --git a/tests/e2e/test_workbench_w4_e2e.py b/tests/e2e/test_workbench_w4_e2e.py new file mode 100644 index 0000000..17f9254 --- /dev/null +++ b/tests/e2e/test_workbench_w4_e2e.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import json +import os +import shutil +import sys +import time +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path[:0] = [str(HERE.parent), str(HERE)] + +import test_orchestrator_e2e as base +from hub_case import HubCase + +OUT = Path(os.environ.get("E2E_OUT", "/tmp/workbench-w4-shots")) +GPU = "0000:01:00.0" +ROWS = 360 +WAITED = (120, 180) +ASLEEP = 180 +UNREAD = (300, 320) +PANELS = {"disks": "Disks", "host": "CPU, RAM and swap", "gpu": "GPU and VRAM", "tokens": "Tokens per agent"} +GUARD = """() => { + window.e2eViolations = []; + document.addEventListener("securitypolicyviolation", (event) => window.e2eViolations.push(`${event.violatedDirective} ${event.blockedURI}`)); +}""" +CHARTS = """(async () => {@HELPERS@ + const counted = () => text(document.querySelector('#main [aria-label="Samples"]')); + if (!(await until(() => all("#main svg.res-chart").length === 4 && counted() === "@ROWS@", 15000))) { + return {error: text(document.getElementById("main")), count: counted()}; + } + await sleep(300); + const charts = all("#main svg.res-chart"); + const number = (node, key) => Number(node.dataset[key]); + const attr = (node, key) => Number(node.getAttribute(key)); + const stacks = {}; + for (const rect of all("#main rect.res-bar")) { + (stacks[rect.dataset.t] = stacks[rect.dataset.t] || []).push({agent: rect.dataset.agent, y: attr(rect, "y"), height: attr(rect, "height")}); + } + const piles = Object.values(stacks).map((parts) => parts.sort((a, b) => b.y - a.y)); + const labelled = (node) => (node ? [node.getAttribute("title"), node.getAttribute("aria-label")] : null); + return { + panels: all("#main .res-panel").map((panel) => panel.dataset.panel), + axes: [...new Set(charts.map((chart) => `${chart.dataset.from}|${chart.dataset.to}`))], + since: number(charts[0], "from"), + bands: charts.filter((chart) => chart.dataset.panel !== "tokens").map((chart) => { + const [idle, saturated] = [chart.querySelector("rect.band.idle"), chart.querySelector("rect.band.saturated")]; + return [chart.dataset.panel, attr(idle, "height") / attr(saturated, "height"), attr(idle, "y") > attr(saturated, "y")]; + }), + series: charts.map((chart) => [chart.dataset.panel, [...new Set(all("path.res-line", chart).map((path) => `${path.dataset.series}|${path.dataset.kind}`))]]), + waits: all("#main rect.res-mark.wait").map((rect) => [number(rect, "from"), number(rect, "to"), number(rect, "peak")]), + asleep: all("#main rect.res-mark.asleep").map((rect) => [number(rect, "from"), number(rect, "to")]), + unknown: all("#main rect.res-mark.unknown").map((rect) => [number(rect, "from"), number(rect, "to")]), + agents: [...new Set(all("#main rect.res-bar").map((rect) => rect.dataset.agent))].sort(), + bottoms: [...new Set(piles.map((parts) => parts[0].agent))], + stacked: piles.every((parts) => parts.every((part, index) => !index || Math.abs(part.y + part.height - parts[index - 1].y) < 0.2)), + legend: all("#main .res-host .res-key").map(text), + labels: {tab: labelled(document.getElementById("tab-resources")), reload: labelled(document.querySelector('#main [aria-label="Reload"]')), titles: all("#main .res-title").map(labelled)}, + unlabelled: all("#main svg.bi").map((svg) => svg.parentElement).filter((owner) => !text(owner) && !(owner.getAttribute("title") && owner.getAttribute("aria-label"))).length, + plots: all("#main .res-plot").map((plot) => plot.getBoundingClientRect().height > 20), + tall: document.documentElement.scrollHeight > innerHeight, + wide: document.documentElement.scrollWidth > innerWidth + 1, + violations: window.e2eViolations || null, + }; +})()""" +KEY = """(async () => {@HELPERS@ + document.body.dispatchEvent(new KeyboardEvent("keydown", {key: "5", bubbles: true})); + const opened = Boolean(await until(() => location.hash === "#resources" && document.querySelector("#main svg.res-chart"), 8000)); + return {opened, current: document.getElementById("tab-resources").getAttribute("aria-current")}; +})()""" + + +def probe(template: str) -> str: + return template.replace("@HELPERS@", base.HELPERS).replace("@ROWS@", str(ROWS)) + + +def quiet(found: list[str] | None) -> list[str] | None: + return None if found is None else [seen for seen in found if not seen.endswith("/favicon.ico")] + + +@unittest.skipUnless(shutil.which(os.environ.get("CHROMIUM", "chromium")), "no Chromium") +class TestWorkbenchResources(HubCase): + drive = base.TestOrchestratorE2E.drive + + def setUp(self): + super().setUp() + OUT.mkdir(parents=True, exist_ok=True) + now = int(time.time()) + self.start = now - now % 60 - ROWS * 60 + samples = [self.sample(index) for index in range(ROWS)] + hour = self.start - self.start % 3600 + tokens = [{"t": hour + step * 3600, "agent": "Skills", "model": "claude-opus-5-5", "input": 1000 * (step + 1), "output": 500, "cache_read": 20000, "cache_write": 1000} for step in range(7)] + tokens += [{"t": hour + step * 3600, "agent": "Stage", "model": "claude-haiku-4-5", "input": 200, "output": 100, "cache_read": 3000, "cache_write": 0} for step in range(0, 7, 2)] + folder = self.kept / "resources" + folder.mkdir() + (folder / "samples.jsonl").write_text(self.jsonl(samples), encoding="utf-8") + (folder / "tokens.jsonl").write_text(self.jsonl([row for row in tokens if row["t"] <= now]), encoding="utf-8") + self.base = f"http://{self.host}/" + + def sample(self, index: int) -> dict: + waited, saturated = WAITED[0] <= index < WAITED[1], WAITED[1] <= index < WAITED[1] + 60 + cpu, ram, waiting = (25, 35, 2) if waited else (95, 60, 1) if saturated else (10, 20, 0) if index < WAITED[0] else (60, 55, 0) + gpu = {"name": GPU, "asleep": True} if index < ASLEEP else {"name": GPU, "util": None, "vram_used": None, "vram_total": None} if UNREAD[0] <= index < UNREAD[1] else {"name": GPU, "util": 50, "vram_used": 2048 << 20, "vram_total": 8192 << 20} + root = 950 if index >= ROWS - 60 else 400 + disks = [{"mount": "/", "used": root << 30, "total": 1000 << 30}, {"mount": "/boot/efi", "used": 3 << 20, "total": 100 << 20}] + return {"t": self.start + index * 60, "disks": disks, "cpu": cpu, "ram": ram, "swap": 5, "load": cpu / 25, "gpus": [gpu], "queue": {"waiting": waiting, "active": 1}} + + def steps(self, browser: str) -> list[dict]: + steps = [{"width": width, "height": height, "url": f"{self.base}?{width}#resources", "wait": 1200, "probe": probe(CHARTS), "shot": str(OUT / f"{browser}-{width}-resources.png")} for width, height in base.WIDTHS] + steps.append({"width": 1366, "height": 768, "url": f"{self.base}?key#stream", "wait": 1200, "probe": probe(KEY), "shot": None}) + return steps + + def check(self, data: dict, browser: str) -> None: + results = data["results"] + for (width, _), found in zip(base.WIDTHS, results, strict=False): + with self.subTest(browser=browser, width=width): + self.assertNotIn("error", found, found) + self.assertEqual(found["panels"], list(PANELS)) + self.assertEqual(len(found["axes"]), 1, found["axes"]) + self.assertLess(abs(found["since"] - (time.time() - 86400)), 600) + self.assertEqual([(panel, lower) for panel, _, lower in found["bands"]], [("disks", True), ("host", True), ("gpu", True)]) + for panel, ratio, _ in found["bands"]: + self.assertAlmostEqual(ratio, 3, delta=0.2, msg=panel) + self.assertEqual(found["series"], [["disks", ["/|disk", "/boot/efi|disk"]], ["host", ["CPU|host", "RAM|host", "Swap|host"]], ["gpu", [f"{GPU}|util", f"{GPU}|vram"]], ["tokens", []]]) + self.assertEqual(found["waits"], [[self.start + WAITED[0] * 60, self.start + WAITED[1] * 60, 2]]) + self.assertEqual(found["asleep"], [[self.start, self.start + ASLEEP * 60]]) + self.assertEqual(found["unknown"], [[self.start + UNREAD[0] * 60, self.start + UNREAD[1] * 60]]) + self.assertEqual((found["agents"], found["bottoms"], found["stacked"]), (["Skills", "Stage"], ["Skills"], True)) + self.assertEqual(found["legend"], ["CPU 60 %", "RAM 55 %", "Swap 5 %"]) + self.assertEqual(found["labels"], {"tab": ["Resources (key 5)", "Resources"], "reload": ["Reload: read the sampler files again", "Reload"], "titles": [[name, name] for name in PANELS.values()]}) + self.assertEqual((found["unlabelled"], found["plots"]), (0, [True] * 4)) + self.assertEqual((found["tall"], found["wide"]), (False, False), "the page scrolls") + self.assertEqual(quiet(found["violations"]), []) + self.assertEqual(results[len(base.WIDTHS)], {"opened": True, "current": "page"}) + + def test_chromium_draws_the_resources_charts_from_the_sampler_files_at_three_widths(self): + data = self.drive(self.steps("chromium"), None, init=GUARD) + (OUT / "chromium-resources.json").write_text(json.dumps(data, indent=1), encoding="utf-8") + self.assertEqual(data["errors"], []) + self.check(data, "chromium") + + @unittest.skipUnless(shutil.which(os.environ.get("FIREFOX", "firefox")), "no Firefox") + def test_firefox_draws_the_resources_charts_from_the_sampler_files_at_three_widths(self): + data = self.drive(self.steps("firefox"), None, "bidi.mjs", os.environ.get("FIREFOX", "firefox"), init=GUARD) + (OUT / "firefox-resources.json").write_text(json.dumps(data, indent=1), encoding="utf-8") + self.assertEqual([error for error in data["errors"] if "/favicon.ico" not in error], []) + self.check(data, "firefox") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hub_nav.py b/tests/test_hub_nav.py index 74a4a4a..b019eba 100644 --- a/tests/test_hub_nav.py +++ b/tests/test_hub_nav.py @@ -64,9 +64,9 @@ class TestNavModule(unittest.TestCase): done = subprocess.run(["node", "-e", NAV_PROBE, str(NAV)], capture_output=True, text=True, timeout=60, check=True) cls.data = json.loads(done.stdout) - def test_the_navigation_has_four_entries_and_stream_has_inbox_and_rules(self): + def test_the_navigation_has_five_entries_and_stream_has_inbox_and_rules(self): stream = ["stream", "Stream", [["inbox", "Inbox"], ["rules", "Firewall"]]] - self.assertEqual(self.data["tabs"], [stream, ["agents", "Agents", []], ["artifacts", "Artifacts", []], ["tasks", "Tasks", []]]) + self.assertEqual(self.data["tabs"], [stream, ["agents", "Agents", []], ["artifacts", "Artifacts", []], ["tasks", "Tasks", []], ["resources", "Resources", []]]) def test_agents_and_artifacts_offer_five_views_in_order_and_tasks_six(self): shared = [["orbit", "Orbit"], ["chord", "Chord"], ["swimlane", "Swimlane"], ["tree", "Tree"], ["3d", "3D"]] diff --git a/tests/test_hub_resources.py b/tests/test_hub_resources.py new file mode 100644 index 0000000..e83937a --- /dev/null +++ b/tests/test_hub_resources.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import json +import subprocess +import time +import unittest + +from hub_case import HubCase +from hub_support import HUB, NAV + +MODULE = HUB / "templates" / "resources.js" +PAGE = HUB / "templates" / "page-resources.js" +STYLE = HUB / "templates" / "style-resources.css" +T = 1_760_000_000 +FIXTURE = """ +const asleep = {name: "0000:01:00.0", asleep: true}; +const awake = (util, used) => ({name: "0000:01:00.0", util, vram_used: used, vram_total: 1000}); +const unread = {name: "0000:01:00.0", util: null, vram_used: null, vram_total: null}; +const row = (t, cpu, ram, waiting, gpu, disks) => ({t, cpu, ram, swap: 0, load: 1, disks, gpus: [gpu], queue: {waiting, active: 1}}); +const root = [{mount: "/", used: 250, total: 1000}]; +const samples = [ + row(T, 20, 30, 0, asleep, root), + row(T + 60, 40, 45, 2, asleep, root), + row(T + 120, 45, 40, 3, awake(80, 500), [...root, {mount: "/mnt", used: 1, total: 4}]), + row(T + 180, 70, 40, 3, awake(90, 600), root), + row(T + 240, 30, 30, 1, awake(10, 100), root), + row(T + 300, 35, 30, 0, unread, root), + row(T + 5000, 10, 10, 1, asleep, root), + row(T + 5060, 10, 10, 0, asleep, root), +]; +const tokens = [ + {t: T, agent: "Skills", model: "opus", input: 10, output: 20, cache_read: 300, cache_write: 40}, + {t: T, agent: "Stage", model: "opus", input: 1, output: 2, cache_read: 3, cache_write: 4}, + {t: T, agent: "Skills", model: "haiku", input: 5, output: 0, cache_read: 0, cache_write: 0}, + {t: T + 3600, agent: "Stage", model: "opus", input: 500, output: 0, cache_read: 0, cache_write: 0}, + {t: "x", agent: "Bad", model: "opus", input: 1, output: 0, cache_read: 0, cache_write: 0}, +]; +""" +PROBE = """ +const R = require(process.argv[1]); +const T = Number(process.argv[2]); +@FIXTURE@ +const lines = R.lines(samples); +const pick = (list) => list.map((entry) => [entry.name, entry.kind, entry.points]); +console.log(JSON.stringify({ + zones: R.ZONES, + disks: pick(lines.disks), + host: pick(lines.host), + gpu: pick(lines.gpu), + segments: R.segments([[T, 1], [T + 60, null], [T + 120, 3], [T + 180, 4], [T + 4000, 5]]), + waits: R.waits(samples), + asleep: R.asleep(samples), + unknown: R.unknown(samples), + stack: R.stack(tokens), +})); +""".replace("@FIXTURE@", FIXTURE) +PAGE_PROBE = """ +const fs = require("fs"); +const vm = require("vm"); +const Resources = require(process.argv[1]); +const code = fs.readFileSync(process.argv[2], "utf8"); +const T = Number(process.argv[3]); +@FIXTURE@ +const make = (tag, attrs, children) => { + const node = {tag, attrs: {}, children: [], textContent: "", isConnected: true, + append(...items) { this.children.push(...items); }, + replaceChildren(...items) { this.children = items; }, + getBoundingClientRect() { return {width: 800, height: 120}; }}; + for (const [key, value] of Object.entries(attrs)) { + if (key !== "on" && value !== null && value !== undefined && value !== false) { + node.attrs[key] = String(value); + } + } + node.children = children.flat(Infinity).filter((child) => child !== null && child !== undefined && child !== false); + return node; +}; +const h = (tag, attrs, ...children) => make(tag, attrs, children); +const icon = (name) => make("svg", {class: "bi", "data-icon": name}, []); +const fetched = []; +const context = { + Resources, h, s: h, icon, DAY: 86400, ui: {preset: 86400, from: null, to: null, asOf: null, offline: false}, + iconBtn: (name, label, cls, run, hint) => make("button", {title: hint, "aria-label": label}, [icon(name)]), + fixed: (value) => value.toFixed(1), dash: (value) => (value === null ? "–" : value), absolute: {format: () => "d"}, clock: {format: () => "c"}, stampOf: (t) => new Date(t * 1000), + window: {setTimeout: (run) => run(), setInterval: () => 1, addEventListener: () => null}, document: {hidden: false}, + windowQuery: () => "from=1", nowSeconds: () => 0, + fetch: async (url) => { + fetched.push(url); + return {ok: true, json: async () => ({since: T - 600, until: T + 6000, samples, tokens})}; + }, +}; +vm.runInNewContext(`${code}\\nscreen = resourcesScreen();`, context); +const walk = (node, found) => { + if (node && typeof node === "object") { + found.push(node); + node.children.forEach((child) => walk(child, found)); + } + return found; +}; +const has = (node, name) => (node.attrs.class || "").split(" ").includes(name); +(async () => { + for (let round = 0; round < 5; round += 1) { + await new Promise((resolve) => setImmediate(resolve)); + } + const all = walk(context.screen, []); + const charts = all.filter((node) => has(node, "res-chart")); + const marks = (kind) => all.filter((node) => has(node, "res-mark") && has(node, kind)).map((node) => [Number(node.attrs["data-from"]), Number(node.attrs["data-to"])]); + console.log(JSON.stringify({ + fetched, + count: all.find((node) => node.attrs["aria-label"] === "Samples").textContent, + charts: charts.map((chart) => [chart.attrs["data-panel"], chart.attrs["data-from"], chart.attrs["data-to"]]), + bands: charts.map((chart) => walk(chart, []).filter((node) => has(node, "band")).map((node) => node.attrs["data-zone"])), + lines: [...new Set(all.filter((node) => has(node, "res-line")).map((node) => `${node.attrs["data-series"]}|${node.attrs["data-kind"]}`))], + waits: marks("wait"), + asleep: marks("asleep"), + unknown: marks("unknown"), + words: all.filter((node) => has(node, "res-mark") && has(node, "unknown")).map((node) => node.children[0].children.join("")), + bars: all.filter((node) => has(node, "res-bar")).map((node) => [node.attrs["data-agent"], Number(node.attrs["data-t"])]), + legends: all.filter((node) => has(node, "res-key")).map((node) => node.children.join("")), + titles: all.filter((node) => has(node, "res-title")).map((node) => [node.attrs.title, node.attrs["aria-label"], node.children[0].attrs["data-icon"]]), + })); +})(); +""".replace("@FIXTURE@", FIXTURE) + + +class TestResourcesModule(unittest.TestCase): + @classmethod + def setUpClass(cls): + done = subprocess.run(["node", "-e", PROBE, str(MODULE), str(T)], capture_output=True, text=True, timeout=60, check=True) + cls.data = json.loads(done.stdout) + + def test_bands_mark_idle_below_thirty_and_saturated_above_ninety_percent(self): + self.assertEqual(self.data["zones"], [{"kind": "idle", "from": 0, "to": 30}, {"kind": "saturated", "from": 90, "to": 100}]) + + def test_disks_cpu_ram_swap_gpu_and_vram_become_percent_lines_on_one_time_axis(self): + times = [T, T + 60, T + 120, T + 180, T + 240, T + 300, T + 5000, T + 5060] + self.assertEqual(self.data["disks"], [["/", "disk", [[t, 25] for t in times]], ["/mnt", "disk", [[T + 120, 25]]]]) + self.assertEqual([name for name, _, _ in self.data["host"]], ["CPU", "RAM", "Swap"]) + self.assertEqual(self.data["host"][0][2], [[t, cpu] for t, cpu in zip(times, (20, 40, 45, 70, 30, 35, 10, 10), strict=True)]) + gpu, vram = self.data["gpu"] + self.assertEqual((gpu[0], gpu[1], vram[0], vram[1]), ("0000:01:00.0", "util", "0000:01:00.0", "vram")) + self.assertEqual(gpu[2], [[T, None], [T + 60, None], [T + 120, 80], [T + 180, 90], [T + 240, 10], [T + 300, None], [T + 5000, None], [T + 5060, None]]) + self.assertEqual(vram[2][2:5], [[T + 120, 50], [T + 180, 60], [T + 240, 10]]) + + def test_a_line_breaks_at_a_missing_value_and_at_a_gap_of_more_than_half_an_hour(self): + self.assertEqual(self.data["segments"], [[[T, 1]], [[T + 120, 3], [T + 180, 4]], [[T + 4000, 5]]]) + + def test_a_marker_spans_rows_where_jobs_waited_while_cpu_and_ram_stayed_under_half(self): + self.assertEqual(self.data["waits"], [{"from": T + 60, "to": T + 180, "peak": 3}, {"from": T + 240, "to": T + 300, "peak": 1}, {"from": T + 5000, "to": T + 5060, "peak": 1}]) + + def test_an_asleep_gpu_gets_its_own_spans(self): + self.assertEqual(self.data["asleep"], [{"name": "0000:01:00.0", "spans": [{"from": T, "to": T + 120, "peak": 1}, {"from": T + 5000, "to": T + 5120, "peak": 1}]}]) + + def test_an_awake_gpu_without_a_reading_gets_unknown_spans_apart_from_asleep_ones(self): + self.assertEqual(self.data["unknown"], [{"name": "0000:01:00.0", "spans": [{"from": T + 300, "to": T + 360, "peak": 1}]}]) + + def test_tokens_stack_per_hour_with_the_biggest_agent_at_the_bottom(self): + stack = self.data["stack"] + self.assertEqual((stack["agents"], stack["top"]), (["Stage", "Skills"], 500)) + first, second = stack["bars"] + self.assertEqual((first["t"], first["total"]), (T, 385)) + self.assertEqual([(part["agent"], part["from"], part["to"], part["models"]) for part in first["parts"]], [("Stage", 0, 10, ["opus"]), ("Skills", 10, 385, ["opus", "haiku"])]) + self.assertEqual({key: first["parts"][1][key] for key in ("input", "output", "cache_read", "cache_write")}, {"input": 15, "output": 20, "cache_read": 300, "cache_write": 40}) + self.assertEqual((second["t"], [(part["agent"], part["to"]) for part in second["parts"]]), (T + 3600, [("Stage", 500)])) + + +class TestResourcesPage(unittest.TestCase): + def test_the_view_draws_four_panels_on_one_time_axis_from_the_api_rows_alone(self): + done = subprocess.run(["node", "-e", PAGE_PROBE, str(MODULE), str(PAGE), str(T)], capture_output=True, text=True, timeout=60, check=True) + found = json.loads(done.stdout) + self.assertEqual((found["fetched"], found["count"]), (["/api/resources?from=1"], "8")) + axis = [str(T - 600), str(T + 6000)] + self.assertEqual(found["charts"], [[panel, *axis] for panel in ("disks", "host", "gpu", "tokens")]) + self.assertEqual(found["bands"], [["idle", "saturated"]] * 3 + [[]]) + self.assertEqual(found["lines"], ["/|disk", "/mnt|disk", "CPU|host", "RAM|host", "Swap|host", "0000:01:00.0|util", "0000:01:00.0|vram"]) + self.assertEqual(found["waits"], [[T + 60, T + 180], [T + 240, T + 300], [T + 5000, T + 5060]]) + self.assertEqual(found["asleep"], [[T, T + 120], [T + 5000, T + 5120]]) + self.assertEqual((found["unknown"], found["words"]), ([[T + 300, T + 360]], ["0000:01:00.0 unknown: no reading"])) + self.assertEqual(found["bars"], [["Stage", T], ["Skills", T], ["Stage", T + 3600]]) + self.assertEqual(found["legends"], ["/ 25 %", "/mnt 25 %", "CPU 10 %", "RAM 10 %", "Swap 0 %", "0000:01:00.0 10 %", "0000:01:00.0 VRAM 10 %", "Stage 510", "Skills 375"]) + names = [("Disks", "bi-hdd-stack"), ("CPU, RAM and swap", "bi-cpu"), ("GPU and VRAM", "bi-gpu-card"), ("Tokens per agent", "bi-coin")] + self.assertEqual(found["titles"], [[name, name, icon] for name, icon in names]) + + +class TestHubResources(HubCase): + def write(self, name: str, rows: list[dict], tail: str = "") -> None: + folder = self.kept / "resources" + folder.mkdir(exist_ok=True) + (folder / name).write_text(self.jsonl(rows) + tail, encoding="utf-8") + + def fetch(self, query: str) -> tuple[int, bytes]: + status, _, body = self.get(f"/api/resources?{query}") + return status, body + + def test_the_view_reads_only_the_sampler_files_within_the_window(self): + samples = [{"t": T + step, "cpu": step / 60, "ram": 1, "swap": 0, "load": 0, "disks": [], "gpus": [], "queue": {"waiting": 0, "active": 0}} for step in (0, 60, 120, 180)] + tokens = [{"t": T - 3600, "agent": "Skills", "model": "opus", "input": 1, "output": 0, "cache_read": 0, "cache_write": 0}, {"t": T + 3600, "agent": "Skills", "model": "opus", "input": 2, "output": 0, "cache_read": 0, "cache_write": 0}] + self.write("samples.jsonl", samples, "{broken\n[1]\n") + self.write("tokens.jsonl", tokens) + status, body = self.fetch(f"from={T + 60}&to={T + 120}") + self.assertEqual(status, 200, body) + self.assertEqual(json.loads(body), {"since": T + 60, "until": T + 120, "samples": samples[1:3], "tokens": []}) + status, body = self.fetch(f"from={T - 1800}&to={T + 3600}") + self.assertEqual(json.loads(body)["tokens"], tokens) + self.assertEqual(json.loads(body)["samples"], samples) + + def test_missing_files_answer_empty_rows_and_a_reversed_window_is_refused(self): + status, body = self.fetch("") + self.assertEqual(status, 200, body) + found = json.loads(body) + self.assertEqual((found["samples"], found["tokens"]), ([], [])) + self.assertAlmostEqual(found["until"] - found["since"], 86400) + self.assertLess(abs(found["until"] - time.time()), 30) + self.assertEqual(self.fetch(f"from={T + 10}&to={T}")[0], 400) + self.assertEqual(self.fetch("from=x")[0], 400) + + def test_the_page_carries_the_resources_tab_module_view_and_icons(self): + html = self.index() + self.assertIn(MODULE.read_text(encoding="utf-8").strip(), html) + self.assertIn(PAGE.read_text(encoding="utf-8").strip(), html) + self.assertIn(STYLE.read_text(encoding="utf-8").strip(), html) + for name in ("speedometer2", "hdd-stack", "cpu", "gpu-card", "coin", "hourglass-split", "moon", "question-circle"): + with self.subTest(icon=name): + self.assertIn(f' None: + folder = root / "bus" / "pci" / "devices" / address + (folder / "power").mkdir(parents=True) + (folder / "vendor").write_text(f"{vendor}\n", encoding="utf-8") + (folder / "class").write_text(f"{kind}\n", encoding="utf-8") + if status is not None: + (folder / "power" / "runtime_status").write_text(f"{status}\n", encoding="utf-8") + + +def host_sys(root: Path) -> Path: + device(root, AWAKE, "0x10de", "0x030000", "active") + device(root, "0000:01:00.1", "0x10de", "0x040300", "active") + device(root, ASLEEP, "0x10de", "0x030200", "suspended") + device(root, "0000:00:02.0", "0x8086", "0x030000", "active") + return root + + +def row(t: int, cpu: float, used: int, waiting: int, gpu: dict) -> dict: + return {"t": t, "disks": [{"mount": "/", "used": used, "total": 1000}], "cpu": cpu, "ram": cpu / 2, "swap": 0.0, "load": cpu / 10, "gpus": [gpu], "queue": {"waiting": waiting, "active": 1}} + + +class TestParsing(unittest.TestCase): + def test_df_keeps_each_local_block_device_once_with_its_first_mount_in_bytes(self): + self.assertEqual(sampler.parse_df(DF), DISKS) + self.assertEqual(sampler.parse_df(""), []) + + def test_meminfo_gives_ram_and_swap_in_percent_and_no_swap_as_zero(self): + self.assertEqual(sampler.memory(sampler.parse_meminfo(MEMINFO)), (25.0, 25.0)) + without = MEMINFO.replace("SwapTotal: 8000000", "SwapTotal: 0").replace("SwapFree: 6000000", "SwapFree: 0") + self.assertEqual(sampler.memory(sampler.parse_meminfo(without)), (25.0, 0.0)) + self.assertEqual(sampler.memory({}), (None, None)) + + def test_cpu_percent_is_the_busy_share_of_the_ticks_between_two_reads(self): + before, after = sampler.cpu_ticks(STAT_BEFORE), sampler.cpu_ticks(STAT_AFTER) + self.assertEqual((before, after), ((8500, 10000), (9000, 11300))) + self.assertEqual(sampler.cpu_percent(before, after), 61.5) + self.assertIsNone(sampler.cpu_percent(after, after)) + self.assertIsNone(sampler.cpu_ticks("intr 1 2 3\n")) + + def test_loadavg_is_the_one_minute_load(self): + self.assertEqual(sampler.parse_loadavg(LOADAVG), 1.25) + self.assertIsNone(sampler.parse_loadavg("")) + + def test_nvidia_smi_csv_gives_util_and_vram_in_bytes_per_pci_address(self): + self.assertEqual(sampler.parse_nvidia(NVIDIA_SMI), { + AWAKE: {"util": 37.0, "vram_used": 2048 * MIB, "vram_total": 8188 * MIB}, + "0000:03:00.0": {"util": None, "vram_used": None, "vram_total": 4096 * MIB}, + }) + + def test_the_queue_counts_waiting_and_active_entries(self): + with tempfile.TemporaryDirectory() as tmp: + council = Path(tmp) + self.assertEqual(sampler.queue(council), {"waiting": 0, "active": 0}) + entries = [{"status": "waiting"}, {"status": "waiting"}, {"status": "active"}, {"status": "done"}, "junk"] + (council / "queue.json").write_text(json.dumps({"entries": entries}), encoding="utf-8") + self.assertEqual(sampler.queue(council), {"waiting": 2, "active": 1}) + + +class TestCpu(unittest.TestCase): + def test_cpu_is_the_busy_share_since_the_ticks_the_run_before_kept_and_the_sampler_never_sleeps(self): + with tempfile.TemporaryDirectory() as tmp: + proc, state = Path(tmp) / "proc", Path(tmp) / "state" + proc.mkdir() + state.mkdir() + (proc / "stat").write_text(STAT_BEFORE, encoding="utf-8") + with mock.patch.object(sampler.time, "sleep", side_effect=AssertionError("slept")), mock.patch.object(sampler.time, "time", return_value=NOW - 60): + self.assertIsNone(sampler.sample(proc, Path(tmp) / "sys", Path(tmp), lambda argv: None, state)["cpu"]) + (proc / "stat").write_text(STAT_AFTER, encoding="utf-8") + self.assertEqual(sampler.cpu_since(proc, state, NOW), 61.5) + (proc / "stat").write_text(STAT_LATER, encoding="utf-8") + self.assertIsNone(sampler.cpu_since(proc, state, NOW + sampler.BUCKET + 1)) + self.assertEqual(json.loads((state / "cpu.json").read_text(encoding="utf-8")), [NOW + sampler.BUCKET + 1, 9300, 12200]) + + +class TestGpus(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.root = Path(tmp.name) + self.calls = [] + + def run_smi(self, argv: list[str]) -> str: + self.calls.append(argv) + return NVIDIA_SMI + + def test_an_asleep_gpu_is_recorded_asleep_and_nvidia_smi_queries_only_the_awake_one(self): + found = sampler.gpus(host_sys(self.root), self.run_smi) + self.assertEqual(found, [{"name": AWAKE, "util": 37.0, "vram_used": 2048 * MIB, "vram_total": 8188 * MIB}, {"name": ASLEEP, "asleep": True}]) + self.assertEqual(len(self.calls), 1) + self.assertEqual(self.calls[0][0], "nvidia-smi") + self.assertIn(f"--id={AWAKE}", self.calls[0]) + self.assertNotIn(ASLEEP, " ".join(self.calls[0])) + + def test_nvidia_smi_never_runs_while_every_nvidia_gpu_sleeps_or_lacks_a_runtime_status(self): + device(self.root, ASLEEP, "0x10de", "0x030000", "suspended") + device(self.root, "0000:04:00.0", "0x10de", "0x030000", None) + self.assertEqual(sampler.gpus(self.root, self.run_smi), [{"name": ASLEEP, "asleep": True}, {"name": "0000:04:00.0", "asleep": True}]) + self.assertEqual(self.calls, []) + + def test_an_awake_gpu_nvidia_smi_gives_no_reading_for_is_recorded_unknown(self): + unknown = {"name": AWAKE, "util": None, "vram_used": None, "vram_total": None} + self.assertEqual(sampler.gpus(host_sys(self.root), lambda argv: None), [unknown, {"name": ASLEEP, "asleep": True}]) + self.assertEqual(sampler.gpus(self.root, lambda argv: NVIDIA_SMI.splitlines()[1]), [unknown, {"name": ASLEEP, "asleep": True}]) + self.assertEqual(sampler.mean_row(NOW, [row(NOW + 60, 1.0, 1, 0, unknown), row(NOW + 120, 1.0, 1, 0, unknown)])["gpus"], [unknown]) + + def test_no_nvidia_gpu_means_no_gpu_row_and_no_call(self): + device(self.root, "0000:00:02.0", "0x8086", "0x030000", "active") + self.assertEqual(sampler.gpus(self.root, self.run_smi), []) + self.assertEqual(sampler.gpus(self.root / "missing", self.run_smi), []) + self.assertEqual(self.calls, []) + + +class TestRetention(unittest.TestCase): + def test_rows_older_than_seven_days_fold_into_fifteen_minute_means_and_means_past_ninety_days_drop(self): + bucket = NOW - 8 * DAY + asleep, awake = {"name": AWAKE, "asleep": True}, {"name": AWAKE, "util": 50.0, "vram_used": 300, "vram_total": 1000} + mean = {**row(NOW - 30 * DAY, 12.0, 100, 0, asleep)} + rows = [ + row(NOW - 91 * DAY, 5.0, 1, 0, asleep), + mean, + row(bucket, 10.0, 100, 0, asleep), + row(bucket + 60, 20.0, 200, 1, asleep), + {**row(bucket + 120, 60.0, 300, 2, awake), "disks": [{"mount": "/", "used": 300, "total": 1000}, {"mount": "/mnt", "used": 10, "total": 20}]}, + row(bucket + 1000, 40.0, 400, 0, awake), + row(NOW - 60, 70.0, 700, 3, awake), + ] + kept = sampler.retain(rows, NOW) + self.assertEqual([entry["t"] for entry in kept], [NOW - 30 * DAY, bucket, bucket + 900, NOW - 60]) + self.assertIs(kept[0], mean) + self.assertEqual(kept[1], { + "t": bucket, "disks": [{"mount": "/", "used": 200, "total": 1000}, {"mount": "/mnt", "used": 10, "total": 20}], "cpu": 30.0, "ram": 15.0, "swap": 0.0, "load": 3.0, + "gpus": [{"name": AWAKE, "util": 50.0, "vram_used": 300, "vram_total": 1000}], "queue": {"waiting": 1.0, "active": 1.0}, + }) + self.assertEqual((kept[2]["cpu"], kept[2]["disks"]), (40.0, [{"mount": "/", "used": 400, "total": 1000}])) + self.assertEqual(kept[3], rows[-1]) + self.assertEqual(sampler.retain(kept, NOW), kept) + + def test_a_bucket_still_reaching_into_the_last_seven_days_stays_raw_and_asleep_buckets_stay_asleep(self): + now = NOW + 300 + asleep = {"name": ASLEEP, "asleep": True} + rows = [row(NOW - 7 * DAY - 890, 10.0, 1, 0, asleep), row(NOW - 7 * DAY - 830, 30.0, 3, 0, asleep), row(NOW - 7 * DAY + 10, 50.0, 5, 0, asleep)] + kept = sampler.retain(rows, now) + self.assertEqual([entry["t"] for entry in kept], [NOW - 7 * DAY - 900, NOW - 7 * DAY + 10]) + self.assertEqual((kept[0]["cpu"], kept[0]["gpus"]), (20.0, [asleep])) + self.assertEqual(kept[1], rows[2]) + + def test_record_appends_and_folds_only_on_the_first_minute_of_a_bucket(self): + with tempfile.TemporaryDirectory() as tmp: + state = Path(tmp) + asleep = {"name": AWAKE, "asleep": True} + old = [row(NOW - 8 * DAY, 10.0, 1, 0, asleep), row(NOW - 8 * DAY + 60, 30.0, 1, 0, asleep)] + (state / "samples.jsonl").write_text("".join(json.dumps(entry) + "\n" for entry in old), encoding="utf-8") + later = row(NOW + 400, 1.0, 1, 0, asleep) + sampler.record(state, later, NOW + 400) + lines = [json.loads(line) for line in (state / "samples.jsonl").read_text(encoding="utf-8").splitlines()] + self.assertEqual(lines, [*old, later]) + first = row(NOW + 930, 2.0, 1, 0, asleep) + sampler.record(state, first, NOW + 930) + lines = [json.loads(line) for line in (state / "samples.jsonl").read_text(encoding="utf-8").splitlines()] + self.assertEqual([entry["t"] for entry in lines], [NOW - 8 * DAY, NOW + 400, NOW + 930]) + self.assertEqual(lines[0]["cpu"], 20.0) + + +class TestSamplerInScratchHome(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + base = Path(tmp.name) + self.home, self.council, stubs, self.proc = base / "home", base / "council", base / "bin", base / "proc" + for folder in (self.home, self.council, stubs, self.proc): + folder.mkdir() + for name, text in (("stat", STAT_BEFORE), ("meminfo", MEMINFO), ("loadavg", LOADAVG)): + (self.proc / name).write_text(text, encoding="utf-8") + for tool in ("df", "nvidia-smi"): + (stubs / tool).write_text(STUB, encoding="utf-8") + (stubs / tool).chmod(0o755) + self.sys = host_sys(base / "sys") + self.log = base / "calls.jsonl" + (self.council / "queue.json").write_text(json.dumps({"entries": [{"status": "waiting"}, {"status": "active"}, {"status": "active"}]}), encoding="utf-8") + self.env = { + **os.environ, "HOME": str(self.home), "PATH": f"{stubs}:/usr/bin:/bin", + "STUB_LOG": str(self.log), "STUB_DF": DF, "STUB_NVIDIA_SMI": NVIDIA_SMI.splitlines()[0] + "\n", + } + self.resources = self.home / ".local" / "state" / "agents-council" / "resources" + + def sample(self) -> None: + argv = [sys.executable, str(SCRIPT), "sample", "--state", str(self.resources), "--council", str(self.council), "--proc", str(self.proc), "--sys", str(self.sys)] + done = subprocess.run(argv, env=self.env, capture_output=True, text=True, timeout=60, check=False) + self.assertEqual(done.returncode, 0, done.stderr) + + def test_the_sampler_appends_one_row_per_run_under_the_scratch_home(self): + self.sample() + (self.proc / "stat").write_text(STAT_AFTER, encoding="utf-8") + self.sample() + samples = self.resources / "samples.jsonl" + rows = [json.loads(line) for line in samples.read_text(encoding="utf-8").splitlines()] + self.assertEqual(len(rows), 2) + for found in rows: + with self.subTest(t=found["t"]): + self.assertEqual(sorted(found), ["cpu", "disks", "gpus", "load", "queue", "ram", "swap", "t"]) + self.assertEqual(found["disks"], DISKS) + self.assertEqual(found["gpus"], [{"name": AWAKE, "util": 37.0, "vram_used": 2048 * MIB, "vram_total": 8188 * MIB}, {"name": ASLEEP, "asleep": True}]) + self.assertEqual(found["queue"], {"waiting": 1, "active": 2}) + self.assertEqual((found["ram"], found["swap"], found["load"]), (25.0, 25.0, 1.25)) + self.assertEqual([found["cpu"] for found in rows], [None, 61.5]) + self.assertLessEqual(rows[0]["t"], rows[1]["t"]) + calls = [json.loads(line) for line in self.log.read_text(encoding="utf-8").splitlines()] + self.assertEqual([call[0] for call in calls], ["df", "nvidia-smi"] * 2) + self.assertTrue(all(ASLEEP not in " ".join(call) for call in calls), calls) + self.assertEqual(sorted(path.name for path in samples.parent.iterdir()), ["cpu.json", "samples.jsonl"]) + + +class TestTimer(unittest.TestCase): + def test_a_user_timer_runs_sample_and_tokens_every_minute_without_a_page_condition(self): + timer = (UNITS / "kvb-orchestrator-resources.timer").read_text(encoding="utf-8").splitlines() + service = (UNITS / "kvb-orchestrator-resources.service").read_text(encoding="utf-8").splitlines() + self.assertIn("OnCalendar=*-*-* *:*:00", timer) + self.assertIn("WantedBy=timers.target", timer) + script = "/usr/bin/env python3 %h/.local/share/kvb-orchestrator/client/resources.py" + shared = "--state ${KVB_ORCHESTRATOR_KEPT}/resources --council ${KVB_ORCHESTRATOR_COUNCIL}" + self.assertEqual([line for line in service if line.startswith("ExecStart=")], [ + f"ExecStart={script} sample {shared} --proc /proc --sys /sys", + f"ExecStart={script} tokens {shared} --transcripts ${{KVB_ORCHESTRATOR_PROJECTS}}", + ]) + self.assertIn("EnvironmentFile=%h/.local/share/kvb-orchestrator/.env", service) + self.assertIn("Type=oneshot", service) + self.assertFalse(any(line.startswith("ExecCondition=") for line in service)) + + def test_orchestrator_install_enables_the_resources_timer(self): + script = (hub_support.REPO_ROOT / "scripts" / "orchestrator.sh").read_text(encoding="utf-8") + self.assertIn('RESOURCES_UNIT="${NAME}-resources"', script) + self.assertIn('systemctl --user enable --now "${RESOURCES_UNIT}.timer"', script) + self.assertIn('"${HUB}/client/resources.py"', script) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resources_tokens.py b/tests/test_resources_tokens.py new file mode 100644 index 0000000..e66cb31 --- /dev/null +++ b/tests/test_resources_tokens.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import ast +import datetime +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import hub_support +import resources as sampler + +SCRIPT = hub_support.HUB / "client" / "resources.py" +NOW = 1_800_000_000 +HOUR = 3600 +SESSION, OTHER = "5d88e9aa-0000-4000-8000-000000000001", "f45bf5bb-0000-4000-8000-000000000002" +OPUS, HAIKU = "claude-opus-5-5", "claude-haiku-4-5" + + +def iso(ts: float) -> str: + return datetime.datetime.fromtimestamp(ts, datetime.UTC).isoformat().replace("+00:00", "Z") + + +def assistant(ts: float, session: str, model: str, ident: str, counts: tuple[int, int, int, int], request: str = "") -> dict: + usage = dict(zip(("input_tokens", "output_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"), counts, strict=True)) + message = {"id": ident, "type": "message", "role": "assistant", "model": model, "content": [{"type": "text", "text": "ok"}], "usage": {**usage, "service_tier": "standard"}} + return {"type": "assistant", "sessionId": session, "timestamp": iso(ts), "message": message, **({"requestId": request} if request else {})} + + +def lines(rows: list[dict]) -> str: + return "".join(json.dumps(entry) + "\n" for entry in rows) + + +class TokensCase(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + base = Path(tmp.name) + self.projects, self.council, self.state = base / "projects", base / "council", base / "state" + for folder in (self.projects / "-repo", self.council / "work" / "s1", self.state): + folder.mkdir(parents=True) + (self.council / "work" / "s1" / "claim.json").write_text(json.dumps({"office": "work/s1", "session": SESSION, "name": " Skills ", "beat": NOW}), encoding="utf-8") + self.transcript = self.projects / "-repo" / f"{SESSION}.jsonl" + + def count(self, now: float = NOW + 120, budget: int = sampler.BUDGET) -> list[dict]: + return sampler.count_tokens(self.projects, self.council, self.state, now, budget) + + def offsets(self) -> dict: + return json.loads((self.state / "offsets.json").read_text(encoding="utf-8")) + + def written(self) -> list[dict]: + path = self.state / "tokens.jsonl" + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] if path.exists() else [] + + def totals(self) -> dict[tuple[int, str, str], tuple[int, ...]]: + found: dict[tuple[int, str, str], tuple[int, ...]] = {} + for entry in self.written(): + key = (entry["t"], entry["agent"], entry["model"]) + before = found.get(key, (0, 0, 0, 0)) + found[key] = tuple(a + b for a, b in zip(before, (entry["input"], entry["output"], entry["cache_read"], entry["cache_write"]), strict=True)) + return found + + +class TestTokens(TokensCase): + def test_usage_fields_add_up_per_hour_agent_and_model_and_a_message_split_over_lines_counts_once(self): + hour = NOW + rows = [ + {"type": "user", "sessionId": SESSION, "timestamp": iso(NOW + 5), "message": {"role": "user", "content": "x", "usage": {"input_tokens": 999}}}, + assistant(NOW + 10, SESSION, OPUS, "msg_1", (10, 20, 300, 40)), + assistant(NOW + 10, SESSION, OPUS, "msg_1", (10, 20, 300, 40)), + assistant(NOW + 20, SESSION, HAIKU, "msg_2", (1, 2, 3, 4)), + assistant(NOW + 30, SESSION, OPUS, "msg_3", (5, 6, 7, 8)), + assistant(NOW + 40, SESSION, "", "msg_4", (0, 0, 0, 0)), + assistant(NOW - HOUR + 10, OTHER, OPUS, "msg_5", (100, 200, 0, 0)), + ] + self.transcript.write_text(lines(rows) + "{not json\n", encoding="utf-8") + added = self.count() + self.assertEqual(sorted((entry["t"], entry["agent"], entry["model"]) for entry in added), sorted(self.totals())) + self.assertEqual(self.totals(), { + (hour, "Skills", OPUS): (15, 26, 307, 48), + (hour, "Skills", HAIKU): (1, 2, 3, 4), + (hour - HOUR, OTHER[:8], OPUS): (100, 200, 0, 0), + }) + self.assertEqual(sorted(self.written()[0]), ["agent", "cache_read", "cache_write", "input", "model", "output", "t"]) + + def test_a_second_run_adds_only_the_bytes_after_the_last_offset(self): + hour = NOW + self.transcript.write_text(lines([assistant(NOW + 10, SESSION, OPUS, "msg_1", (10, 20, 30, 40))]), encoding="utf-8") + self.count() + self.assertEqual(self.offsets()["files"], {f"-repo/{SESSION}.jsonl": self.transcript.stat().st_size}) + self.assertEqual(self.count(), []) + last = json.dumps(assistant(NOW + 50, SESSION, OPUS, "msg_3", (7, 7, 7, 7))) + more = [assistant(NOW + 10, SESSION, OPUS, "msg_1", (10, 20, 30, 40)), assistant(NOW + 40, SESSION, OPUS, "msg_2", (1, 1, 1, 1))] + with self.transcript.open("a", encoding="utf-8") as handle: + handle.write(lines(more) + last[:40]) + self.count() + self.assertEqual(self.totals(), {(hour, "Skills", OPUS): (11, 21, 31, 41)}) + with self.transcript.open("a", encoding="utf-8") as handle: + handle.write(last[40:] + "\n") + self.count() + self.assertEqual(self.totals(), {(hour, "Skills", OPUS): (18, 28, 38, 48)}) + self.assertEqual(self.offsets()["files"][f"-repo/{SESSION}.jsonl"], self.transcript.stat().st_size) + + def test_a_byte_budget_spreads_a_long_backlog_over_runs_without_losing_tokens(self): + rows = [assistant(NOW + index, SESSION, OPUS, f"msg_{index}", (1, 0, 0, 0)) for index in range(40)] + self.transcript.write_text(lines(rows), encoding="utf-8") + line = len(lines(rows[:1])) + self.count(budget=10 * line) + self.assertEqual(sum(entry["input"] for entry in self.written()), 10) + for _ in range(3): + self.count(budget=10 * line) + self.assertEqual(sum(entry["input"] for entry in self.written()), 40) + + def test_a_shrunk_file_restarts_at_zero_a_vanished_one_leaves_the_offsets_and_an_old_one_starts_at_its_end(self): + self.transcript.write_text(lines([assistant(NOW - 40, SESSION, OPUS, "msg_1", (5, 0, 0, 0)), assistant(NOW - 30, SESSION, OPUS, "msg_2", (5, 0, 0, 0))]), encoding="utf-8") + old = self.projects / "-repo" / f"{OTHER}.jsonl" + old.write_text(lines([assistant(NOW - 100 * 86400, OTHER, OPUS, "msg_9", (9, 0, 0, 0))]), encoding="utf-8") + os.utime(old, (NOW - 100 * 86400, NOW - 100 * 86400)) + self.count() + self.assertEqual(sum(entry["input"] for entry in self.written()), 10) + self.transcript.write_text(lines([assistant(NOW - 20, SESSION, OPUS, "msg_7", (1, 0, 0, 0))]), encoding="utf-8") + old.unlink() + self.count() + self.assertEqual(sum(entry["input"] for entry in self.written()), 11) + self.assertEqual(sorted(self.offsets()["files"]), [f"-repo/{SESSION}.jsonl"]) + + def test_a_message_that_fork_transcripts_copy_counts_once_across_files_and_runs(self): + shared = assistant(NOW + 10, SESSION, OPUS, "msg_1", (10, 20, 30, 40), "req_1") + subagents = self.projects / "-repo" / SESSION / "subagents" + subagents.mkdir(parents=True) + self.transcript.write_text(lines([shared]), encoding="utf-8") + (subagents / "agent-a.jsonl").write_text(lines([{**shared, "timestamp": iso(NOW + 40)}, assistant(NOW + 50, SESSION, OPUS, "msg_2", (1, 1, 1, 1), "req_2")]), encoding="utf-8") + self.count() + self.assertEqual(self.totals(), {(NOW, "Skills", OPUS): (11, 21, 31, 41)}) + (subagents / "agent-b.jsonl").write_text(lines([shared, assistant(NOW + 60, SESSION, OPUS, "msg_1", (5, 5, 5, 5), "req_9")]), encoding="utf-8") + self.count(now=NOW + 180) + self.assertEqual(self.totals(), {(NOW, "Skills", OPUS): (16, 26, 36, 46)}) + + def test_the_seen_messages_keep_the_hours_of_this_run_and_the_last_two_hours(self): + self.transcript.write_text(lines([assistant(NOW - 5 * HOUR, SESSION, OPUS, "msg_1", (1, 0, 0, 0), "req_1"), assistant(NOW + 10, SESSION, OPUS, "msg_2", (1, 0, 0, 0), "req_2")]), encoding="utf-8") + self.count() + self.assertEqual({hour: len(keys) for hour, keys in self.offsets()["seen"].items()}, {str(NOW - 5 * HOUR): 1, str(NOW): 1}) + self.count(now=NOW + HOUR + 120) + self.assertEqual(sorted(self.offsets()["seen"]), [str(NOW)]) + self.count(now=NOW + 2 * HOUR + 120) + self.assertEqual(self.offsets()["seen"], {}) + + def test_a_transcript_that_fails_to_read_keeps_its_offset_and_counts_nothing_twice(self): + old = NOW - 5 * HOUR + self.transcript.write_text(lines([assistant(old + 10, SESSION, OPUS, "msg_1", (5, 0, 0, 0), "req_1")]), encoding="utf-8") + self.count() + end = self.transcript.stat().st_size + with self.transcript.open("a", encoding="utf-8") as handle: + handle.write(lines([assistant(old + 20, SESSION, OPUS, "msg_2", (1, 0, 0, 0), "req_2")])) + with mock.patch.object(sampler, "scan", side_effect=PermissionError("denied")): + self.assertEqual(self.count(now=NOW + HOUR + 120), []) + self.assertEqual(self.offsets()["files"], {f"-repo/{SESSION}.jsonl": end}) + self.count(now=NOW + 2 * HOUR + 120) + self.assertEqual(self.totals(), {(old, "Skills", OPUS): (6, 0, 0, 0)}) + + def test_a_run_without_new_bytes_leaves_offsets_json_unwritten(self): + self.transcript.write_text(lines([assistant(NOW + 10, SESSION, OPUS, "msg_1", (1, 0, 0, 0), "req_1")]), encoding="utf-8") + self.count() + stored = self.offsets() + with mock.patch.object(sampler, "replace", side_effect=AssertionError("offsets.json rewritten")): + self.assertEqual(self.count(now=NOW + 180), []) + self.assertEqual(self.offsets(), stored) + + def test_the_first_minute_of_an_hour_merges_rows_and_drops_those_older_than_ninety_days(self): + hour = NOW + stale = {"t": hour - 91 * 86400, "agent": "Old", "model": OPUS, "input": 1, "output": 1, "cache_read": 1, "cache_write": 1} + parts = [{"t": hour - HOUR, "agent": "Skills", "model": OPUS, "input": value, "output": 0, "cache_read": 0, "cache_write": 0} for value in (1, 2, 3)] + (self.state / "tokens.jsonl").write_text(lines([stale, *parts]), encoding="utf-8") + self.transcript.write_text(lines([assistant(NOW + 10, SESSION, OPUS, "msg_1", (4, 0, 0, 0))]), encoding="utf-8") + self.count(now=hour + 1800) + self.assertEqual(len(self.written()), 5) + self.count(now=hour + 30) + self.assertEqual(self.written(), [{**parts[0], "input": 6}, {"t": hour, "agent": "Skills", "model": OPUS, "input": 4, "output": 0, "cache_read": 0, "cache_write": 0}]) + + +class TestNoLanguageModel(unittest.TestCase): + def test_the_sampler_and_the_parser_import_no_network_client(self): + tree = ast.parse(SCRIPT.read_text(encoding="utf-8")) + imported = {alias.name.split(".")[0] for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names} + imported |= {node.module.split(".")[0] for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module} + self.assertEqual(imported & {"urllib", "http", "socket", "ssl", "requests", "hub_api", "anthropic", "openai"}, set()) + + +class TestTokensInScratchHome(unittest.TestCase): + def test_the_tokens_command_reads_the_scratch_home_transcripts_and_writes_beside_the_samples(self): + with tempfile.TemporaryDirectory() as tmp: + home, council = Path(tmp) / "home", Path(tmp) / "council" + folder = home / ".claude" / "projects" / "-repo" + folder.mkdir(parents=True) + council.mkdir() + ts = datetime.datetime.now(datetime.UTC).timestamp() - 60 + (folder / f"{SESSION}.jsonl").write_text(lines([assistant(ts, SESSION, OPUS, "msg_1", (3, 4, 5, 6))]), encoding="utf-8") + state = home / ".local" / "state" / "agents-council" / "resources" + argv = [sys.executable, str(SCRIPT), "tokens", "--state", str(state), "--council", str(council), "--transcripts", str(home / ".claude" / "projects")] + done = subprocess.run(argv, env={**os.environ, "HOME": str(home), "PATH": "/usr/bin:/bin"}, capture_output=True, text=True, timeout=60, check=False) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertEqual(sorted(path.name for path in state.iterdir()), ["offsets.json", "tokens.jsonl"]) + written = [json.loads(line) for line in (state / "tokens.jsonl").read_text(encoding="utf-8").splitlines()] + self.assertEqual(written, [{"t": int(ts - ts % HOUR), "agent": SESSION[:8], "model": OPUS, "input": 3, "output": 4, "cache_read": 5, "cache_write": 6}]) + + +if __name__ == "__main__": + unittest.main()