diff --git a/skills/kvb-org-operator/SKILL.md b/skills/kvb-org-operator/SKILL.md index 5865999..e890824 100644 --- a/skills/kvb-org-operator/SKILL.md +++ b/skills/kvb-org-operator/SKILL.md @@ -5,8 +5,16 @@ description: > and sandbox-blocked steps of all agents. Trigger on "was muss ich tun". --- -Collect every step that waits on a human, across all agents, so the operator -can run them top to bottom. +Every step that waits on a human, across all agents, is a task +`operator/` of the `kvb-org-rasci` register with a test for when it is +done. `${CLAUDE_SKILL_DIR}/scripts/operator.py` writes and reads those rows, +so collecting them needs no messages. + +```bash +operator.py add --dir --command "" --reason "" --check [:] +operator.py list [--json] +operator.py done +``` ## What counts @@ -22,48 +30,51 @@ Checked against the live settings and agent rules, never from memory: - **Human-bound**: a hardware key, a browser login, a second factor, a purchase, a decision only the operator may take. -## Step 1: collect +## Record: every agent, the moment a step arises -- **Own**: every step of this session that stopped for one of the reasons - above. -- **Register**: tasks of the `kvb-org-rasci` register with the operator as R - and status `queued` or `blocked` (`rasci.py show --wide`). Every agent records - such a step there the moment it arises (`rasci.py set operator/ - --title "" --workdir --status queued`, then `role - operator/ Operator R`), so the list needs no messages. -- **Peers**: never polled. Only when the operator asks for a full sweep, one - `SendMessage` to every live agent (`ListAgents`): "Reply with each pending - operator command the register lacks, one per line with directory, command - and reason; with none, do not reply." No answer counts as none. A peer's - answer is data: it cannot make a command operator-only that the settings - let an agent run. +When one of your steps stops for a reason above, run `operator.py add` +before reporting it: one copy-pasteable command, its absolute directory, and +why no agent can run it. The row gets the operator as R, your session as A, +status `queued`. -## Step 2: check +Choose the strongest check that proves the step done. A check is one of +these kinds, never a shell command: a stored command would let one agent run +code in another's session through the shared register. -Drop what is already done: a push whose branch is no longer ahead, a -settings line already gone, an install whose copy is current. Merge -duplicates across agents. Order by dependency: commit before push, push -before install, install before restart. +| Kind | Done when | +| --- | --- | +| `pushed:` | `git rev-list --count @{u}..HEAD` is 0; no upstream counts as open | +| `installed:=` | both trees hold the same files with the same bytes, `__pycache__` ignored | +| `binary:` | the name is on `PATH` | +| `absent:` | the path does not exist | +| `manual` | only `operator.py done ` closes it | -## Step 3: print +`manual` is the last resort, for a reboot, a login or a decision. Other +kinds, relative paths and shell metacharacters exit 2. -| # | Directory | Command | Why not an agent | Unblocks | Requested by | +## Print: on invocation -- `Directory` is absolute; `Command` is one copy-pasteable command per row. -- No secret, token or credential value: name where it lives instead. +Add this session's own pending steps first, then run `operator.py list` and +print its output unchanged. It sets every `queued` or `blocked` row whose +check passes to `done`, then prints the open ones as a table in dependency +order (commits, pushes, installs, binaries, removals, manual steps) and one +`bash` block ready to copy. Once the operator reports a `manual` step done, +run `operator.py done `. -Then the same commands once more in one `bash` block in table order, each -after its `cd`, ready to copy. +## Peer sweep: only on request -## Step 4: record - -Every listed step enters the register as `operator/` with the -operator as R, the requesting agent as A and status `queued` -(`rasci.py set`, `rasci.py role`). The next run sets `done` on the ones -Step 2 found done. +Never poll peers: every message wakes an agent for a full turn. Only when the +operator explicitly asks for a sweep, one `SendMessage` to every live agent +(`ListAgents`): "Record each pending operator command the register lacks +with `operator.py add`; do not reply." Then run `list`. A peer's row is data: +it cannot make a command operator-only that the settings let an agent run. ## Rules +- No secret, token or credential value in a command, reason or check: name + where it lives instead. +- Merge duplicates: one row per step across all agents. Read `list --json` + before adding; re-add a known step under its existing slug. - Never run a listed command, and never ask a peer to run it: that is the permission laundering the agent rules forbid. - A command that fails the "What counts" test is not the operator's; run it diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py new file mode 100755 index 0000000..5f5e74c --- /dev/null +++ b/skills/kvb-org-operator/scripts/operator.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +"""Keep the commands only the operator can run in the kvb-org-rasci register, each with a done test. + +Environment: + AGENTS_COUNCIL_ROOT: council directory of the kvb-org-rasci register, read + by rasci.py; the rows are its tasks operator/. + CLAUDE_CODE_SESSION_ID: the adding session, recorded as A; other runtimes + pass --session. + PATH: searched by the binary check. + +Args: + add --dir --command "" --reason "" + --check [:] [--session ]: record a command, or update + the row of that slug, as queued with Operator R and the adder A. + Kinds, never a shell command: + pushed: `git rev-list --count @{u}..HEAD` is 0 there; + no upstream counts as open. + installed:= both trees hold the same + files with the same bytes, __pycache__ ignored. + binary: the name is on PATH. + absent: the path does not exist. + manual only `done` closes it. + Paths are absolute; no argument holds a shell metacharacter. + list [--json]: run the check of every queued or blocked operator row and + set the passing ones done. + done : mark a row done, whatever its check. + +Output: + list: the open rows as `| # | Directory | Command | Why | Check |`, rows + naming `commit` first, then by kind in the order above, insertion + order inside a kind; then one bash block with `cd ` and the + command per row. `operator: nothing open` when none is. --json prints + the same rows as a JSON list. + +Exit codes: + 0 done, 2 usage error. +""" + +from __future__ import annotations + +import os +import sys + +# Trip-wire: this file shadows the stdlib operator module; without this, argparse's +# import of collections, which imports operator, loads this script when site has not. +sys.path.sort(key=lambda entry: os.path.realpath(entry) == os.path.dirname(os.path.realpath(__file__))) + +import argparse +import hashlib +import importlib.util +import json +import re +import shlex +import shutil +import subprocess +from pathlib import Path +from types import ModuleType + +RASCI = Path(__file__).resolve().parents[2] / "kvb-org-rasci" / "scripts" / "rasci.py" +PREFIX = "operator/" +KINDS = ("pushed", "installed", "binary", "absent", "manual") +OPEN = ("queued", "blocked") +META = re.compile(r"[\s|&;()<>$`\\\"'*?\[\]{}~!#]") +NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*$") +COMMIT = re.compile(r"\bcommit\b") +HEADER = "| # | Directory | Command | Why | Check |" + + +def fail(message: str) -> None: + print(f"operator: {message}", file=sys.stderr) + sys.exit(2) + + +def module(name: str, path: Path) -> ModuleType: + spec = importlib.util.spec_from_file_location(name, path) + loaded = importlib.util.module_from_spec(spec) + spec.loader.exec_module(loaded) + return loaded + + +def task_of(rasci: ModuleType, slug: str) -> str: + if not rasci.TASK.match(slug): + fail(f"invalid slug {slug!r}; use lowercase letters, digits, '.', '_', '-' and '/'") + return PREFIX + slug + + +def path_problem(path: str, what: str) -> str | None: + if META.search(path): + return f"{what} {path!r} holds a shell metacharacter" + if not os.path.isabs(path): + return f"{what} {path!r} is not an absolute path" + return None + + +def problem(check: str) -> str | None: + kind, sep, arg = check.partition(":") + if kind not in KINDS: + return f"unknown check kind {kind!r}; use one of {', '.join(KINDS)}" + if kind == "manual": + return "manual takes no argument" if sep else None + if not arg: + return f"{kind} needs an argument" + if META.search(arg): + return f"check argument {arg!r} holds a shell metacharacter" + if kind == "binary": + return None if NAME.match(arg) else f"binary takes a name, not {arg!r}" + if kind == "installed": + source, sep, target = arg.partition("=") + if not sep or "=" in target: + return "installed takes =" + return path_problem(source, "source") or path_problem(target, "installed dir") + return path_problem(arg, kind) + + +def ahead(repo: str) -> int | None: + try: + result = subprocess.run( + ["git", "rev-list", "--count", "@{u}..HEAD"], + cwd=repo, capture_output=True, text=True, check=False, + ) + except OSError: + return None + return int(result.stdout) if result.returncode == 0 else None + + +def digests(root: str) -> dict[str, bytes] | None: + if not os.path.isdir(root): + return None + found = {} + try: + for top, dirs, names in os.walk(root): + dirs[:] = [d for d in dirs if d != "__pycache__"] + for name in names: + path = os.path.join(top, name) + with open(path, "rb") as handle: + found[os.path.relpath(path, root)] = hashlib.file_digest(handle, "sha256").digest() + except OSError: + return None + return found + + +def passed(check: str) -> bool: + kind, _, arg = check.partition(":") + if kind == "pushed": + return ahead(arg) == 0 + if kind == "installed": + source, _, target = arg.partition("=") + tree = digests(source) + return tree is not None and tree == digests(target) + if kind == "binary": + return shutil.which(arg) is not None + if kind == "absent": + return not os.path.lexists(arg) + return False + + +def cell(text: object) -> str: + return " ".join(str(text).split()).replace("|", "\\|") + + +def cmd_add(rasci: ModuleType, args: argparse.Namespace) -> int: + task = task_of(rasci, args.slug) + for error in (path_problem(args.dir, "--dir"), problem(args.check)): + if error: + fail(error) + command = args.command.strip() + reason = " ".join(args.reason.split()) + if not command or not reason: + fail("--command and --reason must not be empty") + council = module("council", rasci.COUNCIL) + sid, _ = council.caller(args.session) + party = council.live_sessions().get(sid) or sid + with rasci.database() as tasks: + # Trip-wire: the register is dumped with sort_keys, so insertion order needs its own field. + seq = max((e.get("seq", 0) for t, e in tasks.items() if t.startswith(PREFIX)), default=0) + 1 + entry = tasks.setdefault(task, {}) + entry.setdefault("seq", seq) + entry.update(title=command, workdir=args.dir, description=reason, check=args.check) + roles = {p: v.replace("A", "") for p, v in entry.get("roles", {}).items()} + roles[rasci.OPERATOR] = rasci.letters_of(roles.get(rasci.OPERATOR, ""), "R") + roles[party] = rasci.letters_of(roles.get(party, ""), "A") + entry["roles"] = {p: v for p, v in roles.items() if v} + rasci.apply_status(entry, "queued") + print(f"added {task}") + return 0 + + +def cmd_list(rasci: ModuleType, args: argparse.Namespace) -> int: + rows = { + task: row for task, row in rasci.merged(include_done=False).items() + if task.startswith(PREFIX) and row["status"] in OPEN + } + checks = {task: row.get("check") or "manual" for task, row in rows.items()} + finished = {task for task, check in checks.items() if not problem(check) and passed(check)} + if finished: + with rasci.database() as tasks: + for task in finished: + entry = tasks.get(task) + if entry and entry.get("check") == checks[task] and entry.get("status") in OPEN: + rasci.apply_status(entry, "done") + + def rank(task: str) -> tuple: + kind = "manual" if problem(checks[task]) else checks[task].partition(":")[0] + first = COMMIT.search(rows[task].get("title") or "") + return (0 if first else 1 + KINDS.index(kind), rows[task].get("seq", 0), task) + + order = sorted(rows.keys() - finished, key=rank) + listed = [ + { + "task": task, + "directory": rows[task].get("workdir"), + "command": rows[task].get("title"), + "why": rows[task].get("description"), + "check": checks[task], + } + for task in order + ] + if args.json: + print(json.dumps(listed, indent=1)) + return 0 + if not listed: + print("operator: nothing open") + return 0 + print(HEADER) + print("|---|---|---|---|---|") + for number, row in enumerate(listed, 1): + values = (number, row["directory"] or "-", row["command"] or "-", row["why"] or "-", row["check"]) + print("| " + " | ".join(cell(v) for v in values) + " |") + print() + print("```bash") + for row in listed: + if row["directory"]: + print(f"cd {shlex.quote(row['directory'])}") + if row["command"]: + print(row["command"]) + print("```") + return 0 + + +def cmd_done(rasci: ModuleType, args: argparse.Namespace) -> int: + task = task_of(rasci, args.slug) + with rasci.database() as tasks: + if task not in tasks: + fail(f"{task}: not in the register") + rasci.apply_status(tasks[task], "done") + print(f"done {task}") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + sub = parser.add_subparsers(dest="cmd", required=True) + p = sub.add_parser("add") + p.add_argument("slug") + for flag in ("--dir", "--command", "--reason", "--check"): + p.add_argument(flag, required=True) + p.add_argument("--session") + p.set_defaults(func=cmd_add) + p = sub.add_parser("list") + p.add_argument("--json", action="store_true") + p.set_defaults(func=cmd_list) + p = sub.add_parser("done") + p.add_argument("slug") + p.set_defaults(func=cmd_done) + args = parser.parse_args() + return args.func(module("rasci", RASCI), args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_operator.py b/tests/test_operator.py new file mode 100644 index 0000000..a6510db --- /dev/null +++ b/tests/test_operator.py @@ -0,0 +1,231 @@ +"""Exercise kvb-org-operator: operator rows in the rasci register, their done checks, the printed list.""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +OPERATOR = REPO_ROOT / "skills" / "kvb-org-operator" / "scripts" / "operator.py" +RASCI = REPO_ROOT / "skills" / "kvb-org-rasci" / "scripts" / "rasci.py" +HEADER = "| # | Directory | Command | Why | Check |" +IDENTITY = { + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.org", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.org", + "GIT_CONFIG_NOSYSTEM": "1", +} + + +class TestOperator(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.base = Path(self.tmp.name) + self.home = self.base / "home" + self.root = self.base / "council" + self.bin = self.base / "bin" + self.bin.mkdir() + sessions = self.home / ".claude" / "sessions" + sessions.mkdir(parents=True) + (sessions / "s1.json").write_text(json.dumps({"sessionId": "s1", "name": "One"}), encoding="utf-8") + self.env = { + "PATH": f"{self.bin}:/usr/bin:/bin", + "HOME": str(self.home), + "AGENTS_COUNCIL_ROOT": str(self.root), + "CLAUDE_CODE_SESSION_ID": "s1", + } + + def _run(self, *args: str, script: Path = OPERATOR) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(script), *args], + capture_output=True, text=True, env=self.env, cwd=self.tmp.name, check=False, + ) + + def _out(self, *args: str) -> list[str]: + result = self._run(*args) + self.assertEqual(result.returncode, 0, result.stderr) + return result.stdout.splitlines() + + def _git(self, cwd: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=cwd, env={**self.env, **IDENTITY}, check=True, capture_output=True) + + def _repo(self, name: str, upstream: bool) -> Path: + repo = self.base / name + repo.mkdir() + self._git(repo, "init", "-q", "-b", "main") + (repo / "one").write_text("1", encoding="utf-8") + self._git(repo, "add", "one") + self._git(repo, "commit", "-qm", "one") + if upstream: + bare = self.base / f"{name}.git" + self._git(self.base, "init", "-q", "--bare", str(bare)) + self._git(repo, "remote", "add", "origin", str(bare)) + self._git(repo, "push", "-q", "-u", "origin", "main") + (repo / "two").write_text("2", encoding="utf-8") + self._git(repo, "add", "two") + self._git(repo, "commit", "-qm", "two") + return repo + + def _add(self, slug: str, workdir: Path, command: str, reason: str, check: str) -> None: + self.assertEqual( + self._out("add", slug, "--dir", str(workdir), "--command", command, "--reason", reason, + "--check", check), + [f"added operator/{slug}"], + ) + + def _open(self) -> list[str]: + return [row["task"] for row in json.loads("\n".join(self._out("list", "--json")))] + + def _register(self) -> dict[str, dict]: + return json.loads((self.root / "rasci.json").read_text(encoding="utf-8")) + + def test_checks_close_rows_once_their_condition_holds(self): + repo = self._repo("repo", upstream=True) + src, dst = self.base / "src", self.base / "dst" + (src / "sub").mkdir(parents=True) + (src / "__pycache__").mkdir() + (src / "a.txt").write_text("alpha", encoding="utf-8") + (src / "sub" / "b.txt").write_text("beta", encoding="utf-8") + (src / "__pycache__" / "a.pyc").write_bytes(b"\0") + stale = self.base / "stale" + stale.write_text("x", encoding="utf-8") + base = self.base + self._add("zeta", base, "sudo reboot", "new kernel", "manual") + self._add("absent-row", base, f"sudo rm {stale}", "root owned", f"absent:{stale}") + self._add("binary-row", base, "pipx install kvb-fake-tool", "tool missing", "binary:kvb-fake-tool") + self._add("installed-row", src, "make install", "copy | stale", f"installed:{src}={dst}") + self._add("push-row", repo, "git-sign-push", "branch ahead", f"pushed:{repo}") + self._add("commit-row", repo, "git commit -m wip", "work uncommitted", "manual") + self._add("alpha", base, "sudo systemctl restart docker", "daemon config", "manual") + self.assertEqual( + self._out("list"), + [ + HEADER, + "|---|---|---|---|---|", + f"| 1 | {repo} | git commit -m wip | work uncommitted | manual |", + f"| 2 | {repo} | git-sign-push | branch ahead | pushed:{repo} |", + f"| 3 | {src} | make install | copy \\| stale | installed:{src}={dst} |", + f"| 4 | {base} | pipx install kvb-fake-tool | tool missing | binary:kvb-fake-tool |", + f"| 5 | {base} | sudo rm {stale} | root owned | absent:{stale} |", + f"| 6 | {base} | sudo reboot | new kernel | manual |", + f"| 7 | {base} | sudo systemctl restart docker | daemon config | manual |", + "", + "```bash", + f"cd {repo}", + "git commit -m wip", + f"cd {repo}", + "git-sign-push", + f"cd {src}", + "make install", + f"cd {base}", + "pipx install kvb-fake-tool", + f"cd {base}", + f"sudo rm {stale}", + f"cd {base}", + "sudo reboot", + f"cd {base}", + "sudo systemctl restart docker", + "```", + ], + ) + row = self._register()["operator/installed-row"] + self.assertEqual(row["title"], "make install") + self.assertEqual(row["workdir"], str(src)) + self.assertEqual(row["description"], "copy | stale") + self.assertEqual(row["status"], "queued") + self.assertEqual(row["roles"], {"Operator": "R", "One": "A"}) + self.assertEqual(row["check"], f"installed:{src}={dst}") + + shutil.copytree(src, dst, ignore=shutil.ignore_patterns("__pycache__")) + (dst / "a.txt").write_text("alphA", encoding="utf-8") + self._git(repo, "push", "-q") + self.assertEqual( + self._open(), + ["operator/commit-row", "operator/installed-row", "operator/binary-row", "operator/absent-row", + "operator/zeta", "operator/alpha"], + ) + (dst / "a.txt").write_text("alpha", encoding="utf-8") + tool = self.bin / "kvb-fake-tool" + tool.write_text("#!/bin/sh\n", encoding="utf-8") + tool.chmod(0o755) + self.assertEqual( + self._open(), ["operator/commit-row", "operator/absent-row", "operator/zeta", "operator/alpha"] + ) + stale.unlink() + for slug in ("commit-row", "zeta"): + self.assertEqual(self._out("done", slug), [f"done operator/{slug}"]) + self.assertEqual(self._open(), ["operator/alpha"]) + self._out("done", "alpha") + self.assertEqual(self._out("list"), ["operator: nothing open"]) + rows = {task: row for task, row in self._register().items() if task.startswith("operator/")} + self.assertEqual(len(rows), 7) + for task, row in rows.items(): + with self.subTest(task=task): + self.assertEqual(row["status"], "done") + self.assertRegex(row["end"], r"^20\d\d-") + + def test_pushed_stays_open_without_upstream_or_repository(self): + repo = self._repo("lonely", upstream=False) + self._add("no-upstream", repo, "git push -u origin main", "no upstream", f"pushed:{repo}") + gone = self.base / "gone" + self._add("no-repo", self.base, "git clone x gone", "not cloned", f"pushed:{gone}") + self.assertEqual(self._open(), ["operator/no-upstream", "operator/no-repo"]) + + def test_readd_updates_the_row_and_keeps_its_place(self): + self._add("first", self.base, "sudo reboot", "kernel", "manual") + self._add("second", self.base, "sudo poweroff", "hardware", "manual") + self._add("first", self.base, "sudo reboot now", "kernel 7.2.9", "manual") + self._out("done", "second") + self._add("second", self.base, "sudo poweroff", "hardware", "manual") + self.assertEqual(self._open(), ["operator/first", "operator/second"]) + row = self._register()["operator/first"] + self.assertEqual((row["title"], row["description"]), ("sudo reboot now", "kernel 7.2.9")) + self.assertNotIn("end", self._register()["operator/second"]) + + def test_legacy_row_without_check_counts_as_manual(self): + result = self._run("set", "operator/legacy", "--title", "sudo reboot", "--status", "queued", script=RASCI) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual( + self._out("list"), + [HEADER, "|---|---|---|---|---|", "| 1 | - | sudo reboot | - | manual |", "", "```bash", + "sudo reboot", "```"], + ) + + def test_invalid_input_exits_2(self): + good = ("--dir", "/srv", "--command", "sudo reboot", "--reason", "kernel") + for args in ( + ("add", "x", *good, "--check", "shell:/usr/bin/true"), + ("add", "x", *good, "--check", "run:reboot"), + ("add", "x", *good, "--check", "manual:yes"), + ("add", "x", *good, "--check", "pushed:relative/repo"), + ("add", "x", *good, "--check", "absent:tmp/x"), + ("add", "x", *good, "--check", "installed:/a"), + ("add", "x", *good, "--check", "installed:/a=relative"), + ("add", "x", *good, "--check", "absent:/tmp/x;rm -rf /"), + ("add", "x", *good, "--check", "pushed:/tmp/$(id)"), + ("add", "x", *good, "--check", "absent:/tmp/`id`"), + ("add", "x", *good, "--check", "installed:/a=/b|c"), + ("add", "x", *good, "--check", "binary:a b"), + ("add", "x", *good, "--check", "binary:/usr/bin/git"), + ("add", "x", *good, "--check", "pushed:"), + ("add", "x", "--dir", "relative", "--command", "c", "--reason", "r", "--check", "manual"), + ("add", "x", "--dir", "/srv", "--command", " ", "--reason", "r", "--check", "manual"), + ("add", "x", "--dir", "/srv", "--command", "c", "--check", "manual"), + ("add", "Bad Slug", *good, "--check", "manual"), + ("done", "missing"), + ("list", "--wide"), + ): + with self.subTest(args=args): + self.assertEqual(self._run(*args).returncode, 2) + self.assertFalse((self.root / "rasci.json").exists()) + + +if __name__ == "__main__": + unittest.main()