Merge feature/workbench-w4-resources into unify/skills-20261112

Conflicts combined W1 (Inbox leaves the Stream for Tasks, Waits on Operator) with W4 (Resources tab): five tabs with Resources, Stream keeps only the Firewall, the inbox hash keeps its redirect; index.html.j2 includes both diffs.js and resources.js.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-12 16:26:29 +00:00
commit 515fe61009
21 files changed
+1937 -18

No files matched your search

@@ -739,6 +739,8 @@ flowchart TD
### Phase 6: Resources (S7) ### 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 - [ ] The tab shows tokens per agent and accumulated, and CPU, RAM, disk and
load, on the shared time axis, several values per chart. load, on the shared time axis, several values per chart.
- [ ] Token counts equal the sum of the transcript usage fields of a fixture - [ ] Token counts equal the sum of the transcript usage fields of a fixture
+11 -7
View File
@@ -301,22 +301,22 @@ flowchart TD
### W4 Resources ### 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 `make orchestrator-install` enables, appends one row: used and total per
mount, CPU, RAM and swap percent, load, GPU utilization and VRAM per mount, CPU, RAM and swap percent, load, GPU utilization and VRAM per
GPU, waiting and active queue jobs. 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`. 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 tokens per agent and model from transcript usage fields since the last
byte offset; no sampler or parser calls an LLM. 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. days are dropped.
- [ ] The Resources entry shows on one time axis disks, CPU, RAM, swap, GPU, - [ ] The Resources entry shows on one time axis disks, CPU, RAM, swap, GPU,
VRAM and stacked tokens per agent; bands mark idle (< 30 %) and VRAM and stacked tokens per agent; bands mark idle (< 30 %) and
saturated (> 90 %); a marker flags spans where jobs waited while CPU and saturated (> 90 %); a marker flags spans where jobs waited while CPU and
RAM stayed under 50 %; the view reads only the sampler files. 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 ### Contracts
@@ -344,7 +344,7 @@ The work packages meet only through these interfaces:
used, total}], cpu, ram, swap, load, gpus: [{name, util, vram_used, used, total}], cpu, ram, swap, load, gpus: [{name, util, vram_used,
vram_total} | {name, asleep: true}], queue: {waiting, active}}`), vram_total} | {name, asleep: true}], queue: {waiting, active}}`),
`tokens.jsonl` (`{t, agent, model, input, output, cache_read, `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/`. `~/.local/state/agents-council/resources/`.
- **Mounts**: the hub mounts `~/.local/state/agents-council` read-only and - **Mounts**: the hub mounts `~/.local/state/agents-council` read-only and
opens `gates.db` read-only. opens `gates.db` read-only.
@@ -394,6 +394,10 @@ The work packages meet only through these interfaces:
- [ ] Resources sample parsing on fixtures (df, meminfo, nvidia-smi, asleep - [ ] Resources sample parsing on fixtures (df, meminfo, nvidia-smi, asleep
GPU); token offsets; retention fold; diff cap and masking; feedback GPU); token offsets; retention fold; diff cap and masking; feedback
payload. 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 - [ ] Bubble schedule: scale per queue position, word reveal times from a
duration, reading-pace duration, one bubble per agent. duration, reading-pace duration, one bubble per agent.
- Integration: - Integration:
@@ -403,7 +407,7 @@ The work packages meet only through these interfaces:
- [ ] The host executor applies a signed board batch to a scratch register - [ ] The host executor applies a signed board batch to a scratch register
and refuses a forged one; deletes a local merged branch without a and refuses a forged one; deletes a local merged branch without a
token and refuses a remote one. 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: - E2E, headless Chromium and Firefox at 1920, 1366 and 390 px:
- [ ] Tasks badge, no Inbox entry, operator cards with bulk actions, review - [ ] Tasks badge, no Inbox entry, operator cards with bulk actions, review
diff expanded, feedback by right-click and `f` reaching the radio diff expanded, feedback by right-click and `f` reaching the radio
+590
View File
@@ -0,0 +1,590 @@
#!/usr/bin/env python3
"""Args:
sample: append one row to <state>/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 <state>/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 <state>/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. <state>/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 <state>/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 <state>/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())
+4
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import re import re
import secrets import secrets
import time
from collections.abc import Callable from collections.abc import Callable
from http import HTTPStatus from http import HTTPStatus
from urllib.parse import urlsplit from urllib.parse import urlsplit
@@ -9,6 +10,7 @@ from urllib.parse import urlsplit
from .diffs import DIFF_BODY_LIMIT from .diffs import DIFF_BODY_LIMIT
from .engines import STT_LIMIT from .engines import STT_LIMIT
from .exchange import PLAIN_CSP, Exchange from .exchange import PLAIN_CSP, Exchange
from .resources import RESOURCES, sampled
BODY_LIMIT = 1 << 20 BODY_LIMIT = 1 << 20
UPLOAD_LIMIT = 50 << 20 UPLOAD_LIMIT = 50 << 20
@@ -114,6 +116,8 @@ class Handler(Exchange):
self.answer(lambda: hub.events(self.query()), HTTPStatus.OK) self.answer(lambda: hub.events(self.query()), HTTPStatus.OK)
elif path == "/api/tree": elif path == "/api/tree":
self.answer(hub.repositories, HTTPStatus.OK) 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": elif path == "/api/subagents":
self.answer(lambda: hub.subagents(self.query()), HTTPStatus.OK) self.answer(lambda: hub.subagents(self.query()), HTTPStatus.OK)
elif path == "/api/runner/jobs": elif path == "/api/runner/jobs":
@@ -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)}
@@ -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
@@ -0,0 +1,9 @@
[Unit]
Description=Agent Orchestrator resources every minute
[Timer]
OnCalendar=*-*-* *:*:00
AccuracySec=1s
[Install]
WantedBy=timers.target
+6
View File
@@ -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); } .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-hierarchy.css" %}
{% include "style-resources.css" %}
{% include "style-inbox.css" %} {% include "style-inbox.css" %}
{% include "style-ui.css" %} {% include "style-ui.css" %}
@@ -463,6 +465,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi
{% include "diffs.js" %} {% include "diffs.js" %}
{% include "resources.js" %}
{% include "nav.js" %} {% include "nav.js" %}
{% include "ancestry.js" %} {% include "ancestry.js" %}
@@ -578,6 +582,8 @@ pre.notes { margin: 0; padding: 8px 10px; background: var(--sunken); border-radi
{% include "page-alerts.js" %} {% include "page-alerts.js" %}
{% include "page-resources.js" %}
{% include "page-rules.js" %} {% include "page-rules.js" %}
{% include "page-keyboard.js" %} {% include "page-keyboard.js" %}
+2 -1
View File
@@ -1,6 +1,6 @@
const Nav = (() => { const Nav = (() => {
"use strict"; "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: [["rules", "Firewall"]]}; const SUBS = {stream: [["rules", "Firewall"]]};
const PANES = {inbox: ["tasks", "operator"]}; const PANES = {inbox: ["tasks", "operator"]};
const VIEWS = [["orbit", "Orbit"], ["chord", "Chord"], ["swimlane", "Swimlane"], ["tree", "Tree"], ["3d", "3D"]]; const VIEWS = [["orbit", "Orbit"], ["chord", "Chord"], ["swimlane", "Swimlane"], ["tree", "Tree"], ["3d", "3D"]];
@@ -10,6 +10,7 @@ const Nav = (() => {
stream: "bi-activity", inbox: "bi-inbox", rules: "bi-card-checklist", agents: "bi-people", artifacts: "bi-git", tasks: "bi-list-check", 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", 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", 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 OLD = {graph: ["agents", "orbit"], tree: ["artifacts", "tree"]};
const viewsOf = (screen) => (screen === "tasks" ? TASK_VIEWS : [...VIEWS, ...(OWN[screen] || [])]); const viewsOf = (screen) => (screen === "tasks" ? TASK_VIEWS : [...VIEWS, ...(OWN[screen] || [])]);
+174
View File
@@ -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);
}
}
+2 -2
View File
@@ -55,7 +55,7 @@
} }
function screenOf() { 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) { function render(force) {
@@ -73,7 +73,7 @@
const main = document.getElementById("main"); const main = document.getElementById("main");
main.setAttribute("aria-labelledby", `tab-${ui.screen}`); main.setAttribute("aria-labelledby", `tab-${ui.screen}`);
main.replaceChildren(screenOf()); 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.screen = ui.screen;
drawn.sig = sig; drawn.sig = sig;
if (g3.host && !g3.host.isConnected) { if (g3.host && !g3.host.isConnected) {
+144
View File
@@ -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;
}
@@ -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); }
+22 -2
View File
@@ -4,7 +4,8 @@
# write its .env, build and start the container, register the MCP # write its .env, build and start the container, register the MCP
# server for Claude Code, build pkgmgr from the newest signed tag of # server for Claude Code, build pkgmgr from the newest signed tag of
# PKGMGR_REPO and enable the tree timer and the cleanup path unit, # 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 hook when the migration took them over; an enabled runner gets
# the fresh unit and a restart # the fresh unit and a restart
# migrate move every trace of the earlier name to the kvb-orchestrator names # 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" CLEANUP_UNIT="${NAME}-cleanup"
OFFICES_UNIT="${NAME}-offices" OFFICES_UNIT="${NAME}-offices"
SESSIONS_UNIT="${NAME}-sessions" SESSIONS_UNIT="${NAME}-sessions"
RESOURCES_UNIT="${NAME}-resources"
INSTALLED_SKILLS="${HOME}/.claude/skills" 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) 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" TREE_DIR="${HOME}/.local/state/pkgmgr/tree"
@@ -155,7 +157,7 @@ install_hub() {
cp -a "${REPO_ROOT}/orchestrator/." "${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 # 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 {} + 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: 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=<blob>'; until then every executor refuses" log "skills: approvals need a pinned passkey: settings, 'Passkey registrieren', then 'make orchestrator-pin CREDENTIAL=<blob>'; 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' \ 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" 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() { skills_current() {
local script local script
for script in "${HOST_SCRIPTS[@]}"; do for script in "${HOST_SCRIPTS[@]}"; do
@@ -271,6 +285,7 @@ install_all() {
install_hub install_hub
tree_timer tree_timer
sessions_timer sessions_timer
resources_timer
offices_unit offices_unit
if ((TOOK_RUNNER)) || runner_enabled; then runner || true; fi if ((TOOK_RUNNER)) || runner_enabled; then runner || true; fi
if ((TOOK_HOOK)); then if ((TOOK_HOOK)); then
@@ -354,6 +369,11 @@ check() {
else else
todo "${SESSIONS_UNIT}.timer off: agent states wait for make orchestrator-install" todo "${SESSIONS_UNIT}.timer off: agent states wait for make orchestrator-install"
fi 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 if [[ "$(systemctl --user is-active "${OFFICES_UNIT}.path" 2>/dev/null || true)" == "active" ]]; then
ok "${OFFICES_UNIT}.path active" ok "${OFFICES_UNIT}.path active"
else else
+3 -3
View File
@@ -35,9 +35,9 @@ OUT = Path(os.environ.get("E2E_OUT", "/tmp/orchestrator-phase4-shots"))
WIDTHS = ((1920, 1080), (1366, 768), (390, 844)) WIDTHS = ((1920, 1080), (1366, 768), (390, 844))
VIEWS = ("orbit", "chord", "swimlane", "tree", "3d") VIEWS = ("orbit", "chord", "swimlane", "tree", "3d")
TASK_VIEWS = ("list", "board", "rasci", "gantt", "burn", "3d") TASK_VIEWS = ("list", "board", "rasci", "gantt", "burn", "3d")
SCREENS = ["stream", "tasks/operator", "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", "tasks/operator", "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 = ["Firewall"] SUBS = ["Firewall"]
TABS = ["Stream", "Agents", "Artifacts", "Tasks"] TABS = ["Stream", "Agents", "Artifacts", "Tasks", "Resources"]
HELPERS = """ HELPERS = """
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const until = async (check, ms) => { const until = async (check, ms) => {
@@ -1785,7 +1785,7 @@ class TestOrchestratorE2E(HubCase):
self.assertEqual((probe["title"], probe["h1"]), ("(2) Agent Orchestrator", "Agent Orchestrator")) 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["header"], ["radio-open", "bubbles-toggle", "palette-open", "settings-open"])
self.assertEqual((probe["tabs"], probe["subs"]), (TABS, SUBS)) self.assertEqual((probe["tabs"], probe["subs"]), (TABS, SUBS))
self.assertEqual(probe["badges"], {"tab-stream": None, "tab-rules": None, "tab-agents": "1", "tab-artifacts": None, "tab-tasks": "2"}) self.assertEqual(probe["badges"], {"tab-stream": None, "tab-rules": None, "tab-agents": "1", "tab-artifacts": None, "tab-tasks": "2", "tab-resources": None})
self.assertEqual(probe["banner"], "MAYDAY · Docs: disk full") self.assertEqual(probe["banner"], "MAYDAY · Docs: disk full")
self.assertFalse(probe["rail"]) self.assertFalse(probe["rail"])
self.assertEqual((probe["time"], probe["replay"]), ("24 h ▾", True)) self.assertEqual((probe["time"], probe["replay"]), ("24 h ▾", True))
+152
View File
@@ -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()
+2 -2
View File
@@ -65,9 +65,9 @@ class TestNavModule(unittest.TestCase):
done = subprocess.run(["node", "-e", NAV_PROBE, str(NAV)], capture_output=True, text=True, timeout=60, check=True) done = subprocess.run(["node", "-e", NAV_PROBE, str(NAV)], capture_output=True, text=True, timeout=60, check=True)
cls.data = json.loads(done.stdout) cls.data = json.loads(done.stdout)
def test_the_navigation_has_four_entries_and_no_inbox_and_stream_keeps_only_the_firewall(self): def test_the_navigation_has_five_entries_and_no_inbox_and_stream_keeps_only_the_firewall(self):
stream = ["stream", "Stream", [["rules", "Firewall"]]] stream = ["stream", "Stream", [["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", []]])
self.assertNotIn("inbox", json.dumps(self.data["tabs"])) self.assertNotIn("inbox", json.dumps(self.data["tabs"]))
def test_agents_and_artifacts_offer_five_views_in_order_and_tasks_six(self): def test_agents_and_artifacts_offer_five_views_in_order_and_tasks_six(self):
+229
View File
@@ -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'<symbol id="bi-{name}"', html)
done = subprocess.run(["node", "-e", "const N = require(process.argv[1]); console.log(JSON.stringify([N.TABS.at(-1), N.ICONS.resources, N.route('#resources')]));", str(NAV)], capture_output=True, text=True, timeout=60, check=True)
self.assertEqual(json.loads(done.stdout), [["resources", "Resources"], "bi-speedometer2", {"screen": "resources", "view": None, "redirect": None}])
if __name__ == "__main__":
unittest.main()
+2 -1
View File
@@ -123,6 +123,7 @@ class TestOrchestratorMigration(unittest.TestCase):
self.assertEqual(((config / NEW).stat().st_mode & 0o777, (config / NEW / "approvers.json").stat().st_mode & 0o777), (0o700, 0o600)) self.assertEqual(((config / NEW).stat().st_mode & 0o777, (config / NEW / "approvers.json").stat().st_mode & 0o777), (0o700, 0o600))
self.assertTrue((units / f"{NEW}-runner.service").is_file()) self.assertTrue((units / f"{NEW}-runner.service").is_file())
self.assertTrue((units / f"{NEW}-sessions.timer").is_file()) self.assertTrue((units / f"{NEW}-sessions.timer").is_file())
self.assertTrue((units / f"{NEW}-resources.timer").is_file())
self.assertIn(f"KVB_ORCHESTRATOR_STATE={local / NEW}", (share / NEW / ".env").read_text(encoding="utf-8")) self.assertIn(f"KVB_ORCHESTRATOR_STATE={local / NEW}", (share / NEW / ".env").read_text(encoding="utf-8"))
settings = self.home / ".claude" / "settings.json" settings = self.home / ".claude" / "settings.json"
data = json.loads(settings.read_text(encoding="utf-8")) data = json.loads(settings.read_text(encoding="utf-8"))
@@ -131,7 +132,7 @@ class TestOrchestratorMigration(unittest.TestCase):
self.assertIn(f"{share / NEW}/client/permission_hook.py", hooks[1]["hooks"][0]["command"]) self.assertIn(f"{share / NEW}/client/permission_hook.py", hooks[1]["hooks"][0]["command"])
self.assertEqual(settings.stat().st_mode & 0o777, 0o640) self.assertEqual(settings.stat().st_mode & 0o777, 0o640)
self.assertIn(f"{OLD}/client", (self.home / ".claude" / f"settings.json.before-{NEW}-migration").read_text(encoding="utf-8")) self.assertIn(f"{OLD}/client", (self.home / ".claude" / f"settings.json.before-{NEW}-migration").read_text(encoding="utf-8"))
self.assertEqual((first["images"], first["mcp"], first["enabled"]), ([f"{NEW}:local"], {NEW: "http://127.0.0.1:8737/mcp"}, [f"{NEW}-offices.path", f"{NEW}-runner.service", f"{NEW}-sessions.timer"])) self.assertEqual((first["images"], first["mcp"], first["enabled"]), ([f"{NEW}:local"], {NEW: "http://127.0.0.1:8737/mcp"}, [f"{NEW}-offices.path", f"{NEW}-resources.timer", f"{NEW}-runner.service", f"{NEW}-sessions.timer"]))
self.assertIn(["docker", "compose", "--file", str(share / OLD / "compose.yml"), "down", "--remove-orphans"], first["calls"]) self.assertIn(["docker", "compose", "--file", str(share / OLD / "compose.yml"), "down", "--remove-orphans"], first["calls"])
before = snapshot(self.home) before = snapshot(self.home)
+294
View File
@@ -0,0 +1,294 @@
from __future__ import annotations
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"
UNITS = hub_support.HUB / "systemd"
NOW = 1_800_000_000
DAY = 86400
MIB = 1 << 20
DF = """Filesystem 1024-blocks Used Available Capacity Mounted on
/dev/nvme0n1p2 976762584 512000000 460000000 53% /
devtmpfs 8000000 0 8000000 0% /dev
tmpfs 8000000 1000 7999000 1% /dev/shm
/dev/nvme0n1p2 976762584 512000000 460000000 53% /home
/dev/nvme0n1p1 523248 12000 511248 3% /boot/efi
/dev/loop3 65536 65536 0 100% /snap/core/1
/dev/sda1 1953514584 100000000 1853514584 6% /mnt/Backup Disk
overlay 976762584 512000000 460000000 53% /var/lib/docker/overlay2/x/merged
"""
DISKS = [
{"mount": "/", "used": 512000000 * 1024, "total": 976762584 * 1024},
{"mount": "/boot/efi", "used": 12000 * 1024, "total": 523248 * 1024},
{"mount": "/mnt/Backup Disk", "used": 100000000 * 1024, "total": 1953514584 * 1024},
]
MEMINFO = """MemTotal: 32000000 kB
MemFree: 2000000 kB
MemAvailable: 24000000 kB
Buffers: 500000 kB
SwapTotal: 8000000 kB
SwapFree: 6000000 kB
HugePages_Total: 0
"""
STAT_BEFORE = "cpu 1000 0 500 8000 500 0 0 0 0 0\ncpu0 500 0 250 4000 250 0 0 0 0 0\n"
STAT_AFTER = "cpu 1600 0 700 8400 600 0 0 0 0 0\ncpu0 800 0 350 4200 300 0 0 0 0 0\n"
STAT_LATER = "cpu 2000 0 900 8600 700 0 0 0 0 0\n"
LOADAVG = "1.25 0.80 0.50 2/1234 56789\n"
NVIDIA_SMI = "00000000:01:00.0, 37, 2048, 8188\n00000000:03:00.0, [N/A], [N/A], 4096\n"
AWAKE, ASLEEP = "0000:01:00.0", "0000:02:00.0"
STUB = """#!/usr/bin/env python3
import json, os, sys
from pathlib import Path
with Path(os.environ["STUB_LOG"]).open("a", encoding="utf-8") as handle:
handle.write(json.dumps([Path(sys.argv[0]).name, *sys.argv[1:]]) + "\\n")
sys.stdout.write(os.environ["STUB_" + Path(sys.argv[0]).name.replace("-", "_").upper()])
"""
def device(root: Path, address: str, vendor: str, kind: str, status: str | None) -> 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()
+212
View File
@@ -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, "<synthetic>", "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()