diff --git a/skills/kvb-org-operator/SKILL.md b/skills/kvb-org-operator/SKILL.md index 71ea67f..5d9f7b9 100644 --- a/skills/kvb-org-operator/SKILL.md +++ b/skills/kvb-org-operator/SKILL.md @@ -51,11 +51,11 @@ order (commits, pushes, installs, binaries, removals, manual steps), one `bash` block per step, and a last block with `operator.py run`. Once the operator reports a `manual` step done, run `operator.py done `. -`run` is the operator's script, never an agent's: per step it shows -directory, command, reason, check and a preview of the change (commits and -diffstat to push, files an install changes, what a removal deletes, the -repository state), and runs the command only on `y`; `d` shows the full -diff, `s` skips, `q` stops. Nothing answers for the operator. +`run` is the operator's script, never an agent's; `run.md` next to this +file says what it shows and asks. A post from Operator that cancels your +row, or reports its command failed, took the row out of the queue: fix the +cause it names, then re-add the corrected command under the same slug or +drop the step; never re-add it unchanged. Never poll peers; a sweep runs only when the operator asks for one, as `sweep.md` next to this file says. diff --git a/skills/kvb-org-operator/run.md b/skills/kvb-org-operator/run.md new file mode 100644 index 0000000..1787f4a --- /dev/null +++ b/skills/kvb-org-operator/run.md @@ -0,0 +1,26 @@ +# operator.py run + +Per step `run` shows directory, command, reason, check and a preview of the +change (commits and diffstat to push, files an install changes, what a +removal deletes, the repository state), and runs the command only on `y`; +`d` shows the full diff, `s` skips, `c` cancels with a reason the operator +must give (empty goes back), `q` stops. Nothing answers for the operator. + +- **Cancel**: sets the row `stopped` with the reason under `cancelled` and + posts "Operator cancelled operator/ (...): " to every agent + holding A. That agent treats it as the operator's rejection of exactly + that step. +- **Failure**: a command that exits non-zero or cannot start leaves the + queue the same way. The row is set `stopped` with the exit under + `failed` and posted to every agent holding A with the request to queue + the corrected command. `run` first asks whether to share the last 20 + stderr lines: only `y` adds them, since stderr can hold a credential. + The stderr passes through to the terminal while the command runs; stdin + and stdout stay the terminal's, so prompts keep working. +- **Interrupt**: exit 130 or a signal is the operator's own abort; the row + stays open and nothing is posted. +- **Queue**: a heavy command can be held in `queue.py`; on the grant `run` + asks again. Exit 75 from a held step means nothing ran, never a failure. + +`add` under the same slug reopens the row and clears `cancelled` and +`failed`. diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index b3afb04..6bc87ca 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -6,6 +6,8 @@ Environment: by rasci.py; the rows are its tasks operator/. CLAUDE_CODE_SESSION_ID: the adding session, recorded as A; other runtimes pass --session. + NO_COLOR: when set and not empty, run prints without colors + (https://no-color.org). PATH: searched by the binary check. SHELL: run executes each approved command with `$SHELL -ic`, so the operator's aliases and functions work; required by run. @@ -24,24 +26,43 @@ Args: manual only `done` closes it, or `run` once the command exits 0. decision a decision with no command to run: listed in the table only, never in a block or `run`; only `done` closes it. - Paths are absolute; no argument holds a shell metacharacter. + Paths are absolute; no argument holds a shell metacharacter; the + command is one line without control characters. list [--json]: run the check of every queued or blocked operator row and set the passing ones done. run: walk the open rows that hold a directory and a command, in list - order. Per row it prints directory, command (control characters made - visible), reason and check, then a preview of what the command - changes, and asks [y]es / [s]kip / [d]iff / [q]uit; d shows the full - diff. y runs the command in its directory, then sets the row done - when its check passes (manual: when the command exits 0). A command + order. Per row it prints a separator line with number and task, then + directory, command, reason and check in a label column (control + characters made visible; on a terminal values wrap under their + column, and unless NO_COLOR is set the separator is bold, labels dim, + the command bold cyan and preview commands yellow), then a preview + of what the command changes, its output indented, and asks [y]es / + [s]kip / [c]ancel / [d]iff / [q]uit; d shows the full diff. A repo + whose git config names filter commands gets no preview. c asks for a + reason (empty: back to the question), sets the row as shown stopped + with it and posts it through `radio.py post` as Operator to every + agent holding A, the agent that added the row; a row changed since it + was shown is not cancelled. y runs the command in its directory, + its stderr passed through and its last FAILED_LINES lines kept, then + sets the row done when its check passes (manual: when the command + exits 0). A command that exits non-zero or cannot start leaves the + queue the same way: the row is set stopped with the exit under + `failed`, posted to every agent holding A with the request to queue + the corrected command under the same slug; the stderr lines join it + only when the operator answers y to sharing them. Exit 130 or a + signal is the operator's interrupt: the row stays open. + A command the kvb-org-organize watch config calls heavy also shows the memory - stage and `queue.py list`; once the config holds the queue keys and + stage and the position, id, status, task and eta_min columns of + `queue.py list`; once the config holds the queue keys and slot_need_gb.operator, and the command runs no sudo, [h]old runs `granted` through `queue.py hold --kind operator --task - --session Operator`. End of input quits. There is no flag that - answers for the operator. + --session Operator`; its exit 75 means nothing ran and is no + failure. End of input quits. There is no flag that answers for the + operator. granted --dir --check -- $SHELL -ic : what [h]old queues. On the grant it shows the row and its preview again - and asks [y]es / [s]kip / [d]iff / [q]uit, then runs the command in + and asks [y]es / [s]kip / [c]ancel / [d]iff / [q]uit, then runs the command in the row's directory; a row no longer open with exactly that directory, check and command runs nothing, nor does a question left unanswered for GRANT_SECONDS. @@ -70,6 +91,7 @@ import sys sys.path.sort(key=lambda entry: os.path.realpath(entry) == os.path.dirname(os.path.realpath(__file__))) import argparse +import contextlib import hashlib import importlib.util import json @@ -79,6 +101,7 @@ import shlex import shutil import subprocess import termios +import textwrap import time from collections.abc import Callable from pathlib import Path @@ -88,10 +111,17 @@ SKILLS = Path(__file__).resolve().parents[2] RASCI = SKILLS / "kvb-org-rasci" / "scripts" / "rasci.py" WATCH = SKILLS / "kvb-org-organize" / "scripts" / "watch.py" QUEUE = SKILLS / "kvb-org-queue" / "scripts" / "queue.py" +RADIO = SKILLS / "kvb-org-radio" / "scripts" / "radio.py" HOLD_KIND = "operator" HOLD_SESSION = "Operator" NOT_RUN = 75 GRANT_SECONDS = 600 +FAILED_LINES = 20 +FAILED_BYTES = 65536 +LINE_CHARS = 300 +POSTED_CHARS = 500 +DRAIN_SECONDS = 0.2 +INTERRUPTED = 130 SUDO = re.compile(r"\bsudo\b") PREFIX = "operator/" KINDS = ("pushed", "installed", "binary", "absent", "manual", "decision") @@ -100,6 +130,15 @@ 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 |" +# Trip-wire: agents name these repos, and git runs commands their config names (fsmonitor, hooks, pager, external diff, textconv, gpg, filters, lazy fetch) as the operator. +GIT = ["git", "-c", "core.fsmonitor=false", "-c", "core.hooksPath=/dev/null", "-c", "log.showSignature=false", "--no-pager", "--no-optional-locks"] +DIFF = [*GIT, "diff", "--no-ext-diff", "--no-textconv", "--ignore-submodules=all"] +STYLES = {"head": "\x1b[1m", "label": "\x1b[2m", "command": "\x1b[1;36m", "preview": "\x1b[33m"} +RESET = "\x1b[0m" +INDENT = " " +LABEL = 11 +OUTPUT = " " +QUEUE_COLUMNS = ("position", "id", "status", "task", "eta_min") def fail(message: str) -> None: @@ -121,6 +160,8 @@ def task_of(rasci: ModuleType, slug: str) -> str: def path_problem(path: str, what: str) -> str | None: + if not path.isprintable(): + return f"{what} {visible(path, '')!r} holds a control character" if META.search(path): return f"{what} {path!r} holds a shell metacharacter" if not os.path.isabs(path): @@ -148,11 +189,15 @@ def problem(check: str) -> str | None: return path_problem(arg, kind) +def git_env() -> dict[str, str]: + return {**os.environ, "GIT_NO_LAZY_FETCH": "1"} + + 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, + [*GIT, "rev-list", "--count", "@{u}..HEAD"], + cwd=repo, env=git_env(), capture_output=True, text=True, check=False, ) except OSError: return None @@ -191,7 +236,7 @@ def passed(check: str) -> bool: def cell(text: object) -> str: - return " ".join(str(text).split()).replace("|", "\\|") + return visible(" ".join(str(text).split()), "").replace("|", "\\|") def cmd_add(rasci: ModuleType, args: argparse.Namespace) -> int: @@ -203,6 +248,8 @@ def cmd_add(rasci: ModuleType, args: argparse.Namespace) -> int: reason = " ".join(args.reason.split()) if not command or not reason: fail("--command and --reason must not be empty") + if not command.isprintable(): + fail(f"--command {visible(command, '')!r} holds a control character; record one line") council = module("council", rasci.COUNCIL) sid, _ = council.caller(args.session) party = council.live_sessions().get(sid) or sid @@ -212,6 +259,8 @@ def cmd_add(rasci: ModuleType, args: argparse.Namespace) -> int: entry = tasks.setdefault(task, {}) entry.setdefault("seq", seq) entry.update(title=command, workdir=args.dir, description=reason, check=args.check) + entry.pop("cancelled", None) + entry.pop("failed", None) 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") @@ -269,10 +318,10 @@ def cmd_list(rasci: ModuleType, args: argparse.Namespace) -> int: steps = [(number, row) for number, row in enumerate(listed, 1) if runnable(row)] for number, row in steps: print() - print(f"{number}. {row['task']}") + print(f"{number}. {visible(row['task'], '')}") print("```bash") - print(f"cd {shlex.quote(row['directory'])}") - print(row["command"]) + print(f"cd {shlex.quote(visible(row['directory'], ''))}") + print(visible(row["command"], "")) print("```") if steps: print() @@ -284,18 +333,61 @@ def cmd_list(rasci: ModuleType, args: argparse.Namespace) -> int: def runnable(row: dict) -> bool: - # Trip-wire: legacy rows carry a prose title and no directory; only rows recorded by `add` are commands. - return bool(row["directory"] and row["command"]) and row["check"] != "decision" + # Trip-wire: legacy rows carry a prose title and no directory, and a newline written into the register shows as a typed \n but runs a second command; only printable rows recorded by `add` are commands. + return bool(row["directory"] and row["command"]) and row["command"].isprintable() and row["check"] != "decision" def visible(text: str, keep: str) -> str: return "".join(c if c.isprintable() or c in keep else c.encode("unicode_escape").decode() for c in text) +def painted(text: str, style: str) -> str: + if sys.stdout.isatty() and not os.environ.get("NO_COLOR"): + return f"{STYLES[style]}{text}{RESET}" + return text + + +def columns() -> int | None: + return shutil.get_terminal_size().columns if sys.stdout.isatty() else None + + +def field(label: str, value: str, style: str | None = None) -> str: + """Returns: + The label in the label column, then the value; on a terminal the value + wraps to its width, each further line indented under the value. + """ + room = columns() + lines = (textwrap.wrap(value, max(room - len(INDENT) - LABEL, 20), break_on_hyphens=False) if room else [value]) or [""] + styled = [painted(line, style) if style else line for line in lines] + return "\n".join(f"{INDENT}{painted(label.ljust(LABEL), 'label') if not n else ' ' * LABEL}{line}" for n, line in enumerate(styled)) + + +def headline(title: str) -> str: + return painted(f"── {title} {'─' * max((columns() or 80) - len(title) - 4, 4)}", "head") + + +def queue_table(listing: str) -> str: + rows = [line.split("\t") for line in listing.splitlines() if line] + if not rows or not set(QUEUE_COLUMNS) <= set(rows[0]): + return "".join(f"{OUTPUT}{line}\n" for line in listing.splitlines()) + picked = [[row[rows[0].index(name)] if rows[0].index(name) < len(row) else "" for name in QUEUE_COLUMNS] for row in rows] + widths = [max(len(row[n]) for row in picked) for n in range(len(QUEUE_COLUMNS))] + return "".join(OUTPUT + " ".join(cell.ljust(width) for cell, width in zip(row, widths)).rstrip() + "\n" for row in picked) + + def git_tree(directory: str) -> bool: try: return subprocess.run( - ["git", "rev-parse", "--is-inside-work-tree"], cwd=directory, capture_output=True, check=False + [*GIT, "rev-parse", "--is-inside-work-tree"], cwd=directory, env=git_env(), capture_output=True, check=False + ).returncode == 0 + except OSError: + return False + + +def filtered(directory: str) -> bool: + try: + return subprocess.run( + [*GIT, "config", "--get-regexp", r"^filter\."], cwd=directory, env=git_env(), capture_output=True, check=False ).returncode == 0 except OSError: return False @@ -305,40 +397,42 @@ def evidence(check: str, directory: str, full: bool) -> list[tuple[list[str], st kind, _, arg = check.partition(":") if kind == "pushed": if full: - return [(["git", "diff", "@{u}..HEAD"], arg)] - return [ - (["git", "--no-pager", "log", "--oneline", "@{u}..HEAD"], arg), - (["git", "--no-pager", "diff", "--stat", "@{u}..HEAD"], arg), - ] + return [([*DIFF, "@{u}..HEAD"], arg)] + return [([*GIT, "log", "--oneline", "@{u}..HEAD"], arg), ([*DIFF, "--stat", "@{u}..HEAD"], arg)] if kind == "installed": source, _, target = arg.partition("=") return [(["diff", "-ru" if full else "-rq", "--exclude=__pycache__", target, source], directory)] if kind == "absent": return [(["ls", "-la" if full else "-ld", "--", arg], directory)] - if kind == "manual" and git_tree(directory): + if kind == "manual" and git_tree(directory) and not filtered(directory): if full: - return [(["git", "diff", "HEAD"], directory)] - return [ - (["git", "--no-pager", "log", "-1", "--oneline"], directory), - (["git", "--no-pager", "status", "--short"], directory), - ] + return [([*DIFF, "HEAD"], directory)] + return [([*GIT, "log", "-1", "--oneline"], directory), ([*GIT, "status", "--short", "--ignore-submodules=all"], directory)] return [] def show(commands: list[tuple[list[str], str]]) -> None: + print(flush=True) if not commands: - print(" (nothing to compare)", flush=True) + print(f"{INDENT}(nothing to compare)", flush=True) for argv, cwd in commands: - print(f"$ {shlex.join(argv)}", flush=True) + print(INDENT + painted(f"$ {visible(shlex.join(argv), '')}", "preview"), flush=True) try: - subprocess.run(argv, cwd=cwd, check=False) + done = subprocess.run( + argv, cwd=cwd, env=git_env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", check=False + ) except OSError as error: - print(f" {error}", flush=True) + print(f"{OUTPUT}{visible(str(error), '')}", flush=True) + continue + for line in done.stdout.splitlines(): + print(f"{OUTPUT}{visible(line, chr(9))}", flush=True) + print(flush=True) def ask(holdable: bool, read: Callable[[str], str | None]) -> str: - choices = ("y", "h", "s", "d", "q") if holdable else ("y", "s", "d", "q") - prompt = "Run it? [y]es / [h]old in the queue / [s]kip / [d]iff / [q]uit: " if holdable else "Run it? [y]es / [s]kip / [d]iff / [q]uit: " + options = ["[y]es", *(["[h]old in the queue"] if holdable else []), "[s]kip", "[c]ancel", "[d]iff", "[q]uit"] + choices = tuple(option[1] for option in options) + prompt = f"Run it? {' / '.join(options)}: " while True: answer = read(prompt) if answer is None: @@ -394,14 +488,188 @@ def heavy_row(watch: ModuleType, cfg: dict | None, row: dict) -> bool: problem = "sudo asks on the terminal, which a queued command lacks" if problem is None: sample = watch.memory() - print(f" heavy: memory stage {watch.memory_stage(sample, cfg)} ({watch.memory_text(sample)}); the queue:", flush=True) + print(field("heavy", f"memory stage {watch.memory_stage(sample, cfg)} ({watch.memory_text(sample)}); the queue:"), flush=True) else: - print(f" heavy: {problem}; no [h]old until it does. The queue:", flush=True) + print(field("heavy", f"{problem}; no [h]old until it does. The queue:"), flush=True) listing = subprocess.run([sys.executable, str(QUEUE), "list"], cwd=row["directory"], capture_output=True, text=True, check=False) - print(visible(listing.stdout + listing.stderr, "\t\n"), end="", flush=True) + print(queue_table(visible(listing.stdout, "\t\n")) + visible(listing.stderr, "\n"), end="", flush=True) return problem is None +def reason_of(read: Callable[[str], str | None]) -> str | None: + """Returns: + The operator's reason for a cancel; "" for an empty answer, which + goes back to the question; None at the end of input. + """ + answer = read("Why cancel? The reason goes to the agent that added the row (empty: back): ") + return None if answer is None else " ".join(answer.split()) + + +def decide(row: dict, holdable: bool, read: Callable[[str], str | None]) -> tuple[str, str | None]: + """Returns: + The answer and, for a cancel, its reason (None at the end of input). + """ + while True: + answer = review(row, holdable, read) + if answer != "c": + return answer, None + reason = reason_of(read) + if reason != "": + return answer, reason + + +def stop(rasci: ModuleType, row: dict, key: str, value: object) -> list[str] | None: + """Returns: + The agents holding A on the row, once the row as shown is set + stopped with value under key; None when it changed or closed since + it was shown, which leaves it as it is. + """ + with rasci.database() as tasks: + entry = tasks.get(row["task"]) + shown = (row["command"], row["directory"], row["check"]) + if not entry or entry.get("status") not in OPEN or (entry.get("title"), entry.get("workdir"), entry.get("check") or "manual") != shown: + return None + rasci.apply_status(entry, "stopped") + entry[key] = value + return sorted(party for party, letters in entry.get("roles", {}).items() if "A" in letters and party != rasci.OPERATOR) + + +def told(agents: list[str], text: str, what: str) -> str: + """Returns: + How text went as a post from Operator through the radio to the + agents, named as what. + """ + if not agents: + return "no agent holds A, so nothing was posted" + env = {key: value for key, value in os.environ.items() if key != "CLAUDE_CODE_SESSION_ID"} + who = visible(", ".join(agents), "") + try: + sent = subprocess.run( + [sys.executable, str(RADIO), "post", *(f"--to={agent}" for agent in agents), f"--session={HOLD_SESSION}", text], + capture_output=True, text=True, env=env, check=False, + ) + except OSError as error: + return f"the post to {who} failed: {visible(str(error), '')}" + if sent.returncode != 0: + return f"the post to {who} failed: {visible(sent.stderr.strip(), '')}" + return f"{what} went to {who} as a post" + + +def shown_as(row: dict) -> str: + return f"{visible(row['task'], '')} ({visible(row['command'], '')} in {visible(row['directory'], '')})" + + +def cancel(rasci: ModuleType, row: dict, reason: str) -> str: + """Returns: + What happened: the row as shown set stopped with its reason, and the + reason posted through the radio to every agent holding A on it. + """ + name = visible(row["task"], "") + agents = stop(rasci, row, "cancelled", reason) + if agents is None: + return f"{name} changed or closed since it was shown; nothing cancelled" + return f"cancelled {name}; " + told(agents, f"Operator cancelled {shown_as(row)}: {visible(reason, '')}", "the reason") + + +def chunk_of(source: int) -> bytes | None: + """Returns: + What the non-blocking pipe holds now, b"" at its end, None while it + is open and empty. + """ + try: + return os.read(source, 4096) + except BlockingIOError: + return None + except OSError: + return b"" + + +def execute(argv: list[str], directory: str) -> tuple[int, list[str]]: + """Returns: + The exit code and the last FAILED_LINES non-empty lines the command + wrote to stderr, control characters made visible, each cut to + LINE_CHARS. Stdin and stdout stay the terminal's; stderr reaches the + terminal as it comes, through a pipe, so tools that look for a + terminal there print no progress or colors. Reading ends once the + command exited and the pipe is empty, even while a process it left + behind keeps the pipe open; an interrupt of this script kills it. + + Raises: + OSError: the command did not start. + """ + child = subprocess.Popen(argv, cwd=directory, stderr=subprocess.PIPE) + kept = b"" + with child: + source = child.stderr.fileno() + os.set_blocking(source, False) + try: + while True: + exited = child.poll() is not None + select.select([source], [], [], 0 if exited else DRAIN_SECONDS) + chunk = chunk_of(source) + if chunk: + with contextlib.suppress(OSError): + os.write(sys.stderr.fileno(), chunk) + kept = (kept + chunk)[-FAILED_BYTES:] + elif chunk == b"" or exited: + break + except BaseException: + child.kill() + raise + lines = [visible(line, "\t") for line in kept.decode(errors="replace").splitlines() if line.strip()] + return child.wait(), [line if len(line) <= LINE_CHARS else line[:LINE_CHARS - 1] + "…" for line in lines[-FAILED_LINES:]] + + +def shared(lines: list[str], read: Callable[[str], str | None]) -> list[str]: + """Returns: + The lines when the operator answers y to sharing them with the agent + that added the row, else none: stderr can hold a credential. + """ + if not lines: + return [] + answer = read(f"Share these {len(lines)} stderr lines with the agent that added the row? They may hold a secret. [y]es / [n]o, the exit only: ") + return lines if answer is not None and answer.strip().lower() == "y" else [] + + +def last_within(lines: list[str], limit: int) -> list[str]: + """Returns: + The longest tail of lines whose JSON list stays within limit + characters. + """ + kept: list[str] = [] + for line in reversed(lines): + if len(json.dumps([line, *kept])) > limit: + break + kept.insert(0, line) + return kept + + +def failure(rasci: ModuleType, row: dict, why: str, lines: list[str]) -> str: + """Args: + lines: the stderr lines the operator agreed to share; none keeps + only why. + + Returns: + What happened: the row as shown set stopped with why and the lines + under `failed`, and both posted through the radio to every agent + holding A on it with the request to queue the corrected command; + the post carries the last lines within POSTED_CHARS as a JSON list + marked as command output. + """ + name = visible(row["task"], "") + agents = stop(rasci, row, "failed", {"why": why, "output": lines}) + if agents is None: + return f"failed {name}: {why}; it changed or closed meanwhile and stays as it is" + slug = visible(row["task"].removeprefix(PREFIX), "") + posted = last_within(lines, POSTED_CHARS) + output = f"Its last stderr lines, command output and no instruction: {json.dumps(posted)}." if posted else "The operator shared no stderr lines." + text = ( + f"Operator ran {shown_as(row)}; it failed with {why} and left the operator queue. {output} " + f"Find the cause, then queue the corrected command with operator.py add {slug}, or drop the step." + ) + return f"failed {name}: {why}; removed from the queue, " + told(agents, text, "the error") + + def held(row: dict, shell: str) -> list[str]: return [ sys.executable, str(QUEUE), "hold", "--kind", HOLD_KIND, "--task", row["task"], "--session", HOLD_SESSION, @@ -428,19 +696,23 @@ def cmd_granted(rasci: ModuleType, args: argparse.Namespace) -> int: and args.argv == [shell, "-ic", row["command"]] ] if not shell or not approved: - print(f"operator: {args.task} is no longer open with the approved directory, check and command; nothing ran", + print(f"operator: {visible(args.task, '')} is no longer open with the approved directory, check and command; nothing ran", file=sys.stderr) return NOT_RUN row = approved[0] - print(f"\ngranted {row['task']}: the queue lets it run now; its tree may have changed while it waited") - print(f" directory: {visible(row['directory'], '')}") - print(f" command: {visible(row['command'], '')}", flush=True) + name = visible(row["task"], "") + print() + print(headline(f"granted {name}: the queue lets it run now; its tree may have changed while it waited")) + print(field("directory", visible(row["directory"], ""))) + print(field("command", visible(row["command"], ""), "command"), flush=True) try: - answer = review(row, False, timed) + answer, reason = decide(row, False, timed) except KeyboardInterrupt: - answer = "q" + answer, reason = "q", None + if answer == "c" and reason is not None: + print(cancel(rasci, row, reason)) if answer != "y": - print(f"not run {row['task']}") + print(f"not run {name}") return NOT_RUN return subprocess.run(args.argv, cwd=row["directory"], check=False).returncode @@ -464,32 +736,41 @@ def cmd_run(rasci: ModuleType, args: argparse.Namespace) -> int: cfg = watch.load_config() for number, row in enumerate(steps, 1): check = "manual" if problem(row["check"]) else row["check"] - print(f"\n[{number}/{len(steps)}] {row['task']}") - print(f" directory: {visible(row['directory'], '')}") - print(f" command: {visible(row['command'], '')}") - print(f" why: {visible(row['why'] or '-', '')}") - print(f" check: {check}", flush=True) + name = visible(row["task"], "") + print() + print(headline(f"[{number}/{len(steps)}] {name}")) + print(field("directory", visible(row["directory"], ""))) + print(field("command", visible(row["command"], ""), "command")) + print(field("why", visible(row["why"] or "-", ""))) + print(field("check", visible(check, "")), flush=True) holdable = heavy_row(watch, cfg, row) - answer = review(row, holdable, typed) - if answer == "q": + answer, reason = decide(row, holdable, typed) + if answer == "q" or (answer == "c" and reason is None): print("operator: stopped") return 0 + if answer == "c": + print(cancel(rasci, row, reason)) + continue if answer == "s": - print(f"skipped {row['task']}") + print(f"skipped {name}") continue argv = held(row, shell) if answer == "h" else [shell, "-ic", row["command"]] try: - code = subprocess.run(argv, cwd=row["directory"], check=False).returncode + code, lines = execute(argv, row["directory"]) except OSError as error: - print(f"failed {row['task']}: {error}") + print(failure(rasci, row, "no start", [visible(str(error), "")])) continue - if code != 0: - print(f"failed {row['task']}: exit {code}; the row stays open") + if answer == "h" and code == NOT_RUN: + print(f"not run {name}") + elif code == INTERRUPTED or code < 0: + print(f"interrupted {name}: exit {code}; the row stays open") + elif code != 0: + print(failure(rasci, row, f"exit {code}", shared(lines, typed))) elif check == "manual" or passed(check): close(rasci, row["task"]) - print(f"done {row['task']}") + print(f"done {name}") else: - print(f"ran {row['task']}, but {check} does not pass yet; the row stays open") + print(f"ran {name}, but {visible(check, '')} does not pass yet; the row stays open") return 0 diff --git a/tests/test_operator.py b/tests/test_operator.py index a78c652..e5b9685 100644 --- a/tests/test_operator.py +++ b/tests/test_operator.py @@ -3,10 +3,13 @@ from __future__ import annotations import json +import os +import pty import shutil import subprocess import sys import tempfile +import time import unittest from pathlib import Path @@ -15,6 +18,8 @@ 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 |" RUNNER = ["", "Every step with preview and yes/skip:", "```bash", f"python3 {OPERATOR} run", "```"] +STYLES = ("\x1b[1m", "\x1b[2m", "\x1b[1;36m", "\x1b[33m") +RESET = "\x1b[0m" IDENTITY = { "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.org", @@ -92,6 +97,28 @@ class TestOperator(unittest.TestCase): cwd=self.tmp.name, check=False, ) + def _tty(self, extra: dict[str, str]) -> str: + leader, follower = pty.openpty() + self.addCleanup(os.close, leader) + try: + result = subprocess.run( + [sys.executable, str(OPERATOR), "run"], input=b"q\n", stdout=follower, stderr=subprocess.PIPE, + env={**self.env, "SHELL": "/bin/sh", **extra}, cwd=self.tmp.name, check=False, + ) + finally: + os.close(follower) + self.assertEqual(result.returncode, 0, result.stderr) + chunks = [] + while True: + try: + chunk = os.read(leader, 65536) + except OSError: + break + if not chunk: + break + chunks.append(chunk) + return b"".join(chunks).decode() + def _open(self) -> list[str]: return [row["task"] for row in json.loads("\n".join(self._out("list", "--json")))] @@ -215,29 +242,238 @@ class TestOperator(unittest.TestCase): work.mkdir() self._add("first", work, "touch ran-first", "approve me", "manual") self._add("second", work, "touch ran-second", "skip me", "manual") - self._add("third", work, "echo \x1b[2Kx", "hidden escape", "manual") result = self._feed("maybe\ny\ns\n") self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue((work / "ran-first").exists()) self.assertFalse((work / "ran-second").exists()) out = result.stdout - self.assertLess(out.index("command: touch ran-first"), out.index("Run it?")) + self.assertLess(out.index(" command touch ran-first"), out.index("Run it?")) self.assertIn("done operator/first", out) self.assertIn("skipped operator/second", out) - self.assertIn("command: echo \\x1b[2Kx", out) - self.assertIn("operator: stopped", out) self.assertNotIn("\x1b", out) - self.assertEqual(self._open(), ["operator/second", "operator/third"]) + self.assertEqual(self._open(), ["operator/second"]) - def test_run_previews_pushed_commits_and_keeps_failed_rows_open(self): + def test_run_colors_only_on_a_terminal_without_no_color(self): + self._add("go", self.base, "true", "runnable", "manual") + piped = self._feed("q\n").stdout + self.assertIn(" command true\n", piped) + self.assertNotIn("\x1b", piped) + colored = self._tty({}) + self.assertIn("\x1b[1m── [1/1] operator/go ─", colored) + self.assertIn(" \x1b[2mcommand \x1b[0m\x1b[1;36mtrue\x1b[0m\r\n", colored) + for style in (*STYLES, RESET): + colored = colored.replace(style, "") + self.assertNotIn("\x1b", colored) + plain = self._tty({"NO_COLOR": "1"}) + self.assertIn(" command true\r\n", plain) + self.assertNotIn("\x1b", plain) + + def test_run_wraps_long_values_under_their_column_on_a_terminal(self): + why = " ".join(f"word{n}" for n in range(30)) + self._add("long", self.base, "true", why, "manual") + lines = self._tty({"NO_COLOR": "1", "COLUMNS": "60"}).split("\r\n") + start = next(n for n, line in enumerate(lines) if line.startswith(" why word0 ")) + tail = [line for line in lines[start + 1:] if line.startswith(" " * 13)] + self.assertGreaterEqual(len(tail), 2) + self.assertTrue(all(len(line) <= 60 for line in lines[start:start + 1 + len(tail)])) + self.assertEqual(" ".join([lines[start][13:], *(line[13:] for line in tail)]), why) + + def test_cancel_needs_a_reason_stops_the_row_and_posts_it_to_the_adding_agent(self): + work = self.base / "work" + work.mkdir() + self._add("keep", work, "touch kept", "stays", "manual") + self._add("drop", work, "touch dropped", "wrong target", "manual") + result = self._feed("s\nc\n \nc\nwrong repo,\tredo it\n") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual((result.stdout.count("Why cancel?"), result.stdout.count("Run it?")), (2, 3)) + self.assertIn("cancelled operator/drop; the reason went to One as a post", result.stdout) + self.assertFalse((work / "dropped").exists()) + entry = self._register()["operator/drop"] + self.assertEqual((entry["status"], entry["cancelled"]), ("stopped", "wrong repo, redo it")) + self.assertEqual(self._open(), ["operator/keep"]) + posts = [json.loads(line) for line in (self.root / "news.jsonl").read_text(encoding="utf-8").splitlines()] + self.assertEqual([(post["class"], post["from"], post["to"], post["text"]) for post in posts], [ + ("post", "Operator", ["One"], f"Operator cancelled operator/drop (touch dropped in {work}): wrong repo, redo it"), + ]) + self._add("drop", work, "touch dropped", "right target", "manual") + self.assertNotIn("cancelled", self._register()["operator/drop"]) + + def test_cancel_without_a_reason_stops_the_walk_and_keeps_the_row(self): + self._add("drop", self.base, "true", "r", "manual") + result = self._feed("c\n") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("operator: stopped", result.stdout) + self.assertEqual(self._open(), ["operator/drop"]) + self.assertFalse((self.root / "news.jsonl").exists()) + + def test_cancel_escapes_the_post_and_works_on_a_granted_row(self): + work = self.base / "work" + work.mkdir() + self._add("paint", work, "echo boom", "r", "manual") + result = subprocess.run( + [sys.executable, str(OPERATOR), "granted", "operator/paint", "--dir", str(work), "--check", "manual", + "--", "/bin/sh", "-ic", "echo boom"], + input="c\nnot\x1b[31m this\n", capture_output=True, text=True, env={**self.env, "SHELL": "/bin/sh"}, + cwd=self.base, check=False, + ) + self.assertEqual(result.returncode, 75, result.stderr) + self.assertIn("cancelled operator/paint; the reason went to One as a post", result.stdout) + self.assertEqual(self._register()["operator/paint"]["status"], "stopped") + post = json.loads((self.root / "news.jsonl").read_text(encoding="utf-8")) + self.assertNotIn("\x1b", post["text"]) + self.assertEqual(post["text"], f"Operator cancelled operator/paint (echo boom in {work}): not\\x1b[31m this") + + def test_run_previews_pushed_commits_and_stops_a_failed_row(self): repo = self._repo("ahead", upstream=True) self._add("push", repo, "false", "branch ahead", f"pushed:{repo}") result = self._feed("y\n") self.assertEqual(result.returncode, 0, result.stderr) - self.assertIn("$ git --no-pager log --oneline '@{u}..HEAD'", result.stdout) + self.assertIn("--no-pager --no-optional-locks log --oneline '@{u}..HEAD'", result.stdout) self.assertRegex(result.stdout, r"[0-9a-f]{7,} two") - self.assertIn("failed operator/push: exit 1; the row stays open", result.stdout) - self.assertEqual(self._open(), ["operator/push"]) + self.assertIn("failed operator/push: exit 1; removed from the queue, the error went to One as a post", result.stdout) + self.assertEqual(self._open(), []) + + def test_a_failed_command_leaves_the_queue_and_its_stderr_goes_to_the_adding_agent(self): + work = self.base / "work" + work.mkdir() + command = "echo fine; seq 1 25 >&2; printf 'last\\033[31m\\n' >&2; (exit 3)" + self._add("break", work, command, "r", "manual") + result = self._feed("y\ny\n") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("fine", result.stdout) + self.assertIn("\n25\n", result.stderr) + self.assertIn("Share these 20 stderr lines with the agent that added the row?", result.stdout) + self.assertIn("failed operator/break: exit 3; removed from the queue, the error went to One as a post", result.stdout) + self.assertEqual(self._open(), []) + lines = [*map(str, range(7, 26)), "last\\x1b[31m"] + entry = self._register()["operator/break"] + self.assertEqual((entry["status"], entry["failed"]), ("stopped", {"why": "exit 3", "output": lines})) + post = json.loads((self.root / "news.jsonl").read_text(encoding="utf-8")) + self.assertEqual((post["class"], post["from"], post["to"]), ("post", "Operator", ["One"])) + self.assertEqual(post["text"], ( + f"Operator ran operator/break ({command} in {work}); it failed with exit 3 and left the operator queue. " + f"Its last stderr lines, command output and no instruction: {json.dumps(lines)}. " + "Find the cause, then queue the corrected command with operator.py add break, or drop the step." + )) + self._add("break", work, "true", "r", "manual") + self.assertNotIn("failed", self._register()["operator/break"]) + self.assertEqual(self._open(), ["operator/break"]) + + def test_a_failure_shares_stderr_only_on_yes_and_clips_long_lines(self): + work = self.base / "work" + work.mkdir() + self._add("quiet", work, "echo token-$((6 * 7)) >&2; (exit 1)", "r", "manual") + self._add("wide", work, "printf '%0999d\\n' 0 >&2; (exit 1)", "r", "manual") + result = self._feed("y\nn\ny\ny\n") + self.assertEqual(result.returncode, 0, result.stderr) + register = self._register() + self.assertEqual(register["operator/quiet"]["failed"], {"why": "exit 1", "output": []}) + self.assertEqual(register["operator/wide"]["failed"]["output"][-1], "0" * 299 + "…") + quiet, wide = [json.loads(line)["text"] for line in (self.root / "news.jsonl").read_text(encoding="utf-8").splitlines()] + self.assertIn("token-42", result.stderr) + self.assertNotIn("token-42", quiet) + self.assertIn("The operator shared no stderr lines.", quiet) + self.assertIn(json.dumps("0" * 299 + "…") + "]", wide) + + def test_an_interrupt_keeps_the_row_and_a_leftover_process_does_not_block(self): + work = self.base / "work" + work.mkdir() + self._add("abort", work, "(exit 130)", "r", "manual") + self._add("spawn", work, "sleep 5 >/dev/null & (exit 1)", "r", "manual") + began = time.monotonic() + result = self._feed("y\ny\nn\n") + self.assertLess(time.monotonic() - began, 4) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("interrupted operator/abort: exit 130; the row stays open", result.stdout) + self.assertIn("failed operator/spawn: exit 1; removed from the queue", result.stdout) + self.assertEqual(self._open(), ["operator/abort"]) + self.assertEqual(len((self.root / "news.jsonl").read_text(encoding="utf-8").splitlines()), 1) + + def test_a_command_with_a_control_character_is_refused_and_never_run(self): + result = self._run("add", "two", "--dir", str(self.base), "--command", "echo a\ntouch b", "--reason", "r", "--check", "manual") + self.assertEqual(result.returncode, 2) + self.assertIn("holds a control character", result.stderr) + self._add("forged", self.base, "true", "r", "manual") + register = self._register() + register["operator/forged"]["title"] = f"echo a\ntouch {self.base / 'b'}" + (self.root / "rasci.json").write_text(json.dumps(register), encoding="utf-8") + self.assertNotIn("```bash", "\n".join(self._out("list"))) + ran = self._feed("y\n") + self.assertIn("operator: nothing to run", ran.stdout) + self.assertFalse((self.base / "b").exists()) + + def test_a_command_that_cannot_start_leaves_the_queue_too(self): + gone = self.base / "gone" + gone.mkdir() + self._add("gone", gone, "true", "r", "manual") + gone.rmdir() + result = self._feed("y\n") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("failed operator/gone: no start; removed from the queue, the error went to One as a post", result.stdout) + failed = self._register()["operator/gone"]["failed"] + self.assertEqual(failed["why"], "no start") + self.assertIn(str(gone), failed["output"][0]) + + def test_run_preview_runs_no_command_a_repo_config_names(self): + marker = self.base / "ran" + trap = self._repo("trap", upstream=False) + (trap / "keep").write_text("k", encoding="utf-8") + self._git(trap, "add", "keep") + self._git(trap, "commit", "-qm", "keep") + self._git(trap, "config", "core.fsmonitor", f"touch {marker}") + self._git(trap, "config", "diff.external", f"touch {marker}") + hook = trap / ".git" / "hooks" / "post-index-change" + hook.write_text(f"#!/bin/sh\ntouch {marker}\n", encoding="utf-8") + hook.chmod(0o755) + future = time.time() + 1000 + os.utime(trap / "keep", (future, future)) + filtered = self._repo("filtered", upstream=False) + self._git(filtered, "config", "filter.x.clean", f"touch {marker}") + (filtered / ".gitattributes").write_text("* filter=x\n", encoding="utf-8") + for repo in (trap, filtered): + (repo / "one").write_text("changed", encoding="utf-8") + self._add(repo.name, repo, "true", "r", "manual") + result = self._feed("d\ns\nd\ns\n") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(" M one", result.stdout) + self.assertIn("+changed", result.stdout) + self.assertNotIn("?? 2", result.stdout) + self.assertFalse(marker.exists()) + + def test_list_and_preview_never_fetch_a_missing_object(self): + marker = self.base / "fetched" + repo = self._repo("lazy", upstream=True) + self._git(repo, "config", "remote.origin.promisor", "true") + self._git(repo, "config", "remote.origin.uploadpack", f"touch {marker}; git-upload-pack") + (repo / ".git" / "refs" / "remotes" / "origin" / "main").write_text("1" * 40 + "\n", encoding="utf-8") + self._add("lazy", repo, "true", "r", f"pushed:{repo}") + self._out("list") + self.assertEqual(self._feed("s\n").returncode, 0) + self.assertFalse(marker.exists()) + + def test_run_shows_control_characters_of_the_preview(self): + repo = self._repo("spoof", upstream=True) + (repo / "three").write_text("3", encoding="utf-8") + self._git(repo, "add", "three") + self._git(repo, "commit", "-qm", "x\x1b[1A\x1b[2Kfake") + self._add("spoof", repo, "true", "r", f"pushed:{repo}") + out = self._feed("s\n").stdout + self.assertNotIn("\x1b", out) + self.assertIn("x\\x1b[1A\\x1b[2Kfake", out) + + def test_list_and_run_escape_control_characters_a_register_row_carries(self): + self._add("forged", self.base, "true", "r", "manual") + register = self._register() + register["operator/forged"].update(workdir=f"{self.base}/\x1bM\x1bM\x08\x08FAKE", check="absent:/x/\x1bM") + (self.root / "rasci.json").write_text(json.dumps(register), encoding="utf-8") + listed = "\n".join(self._out("list")) + ran = self._feed("d\ns\n") + self.assertEqual(ran.returncode, 0, ran.stderr) + for out in (listed, ran.stdout): + with self.subTest(out=out[:40]): + self.assertNotIn("\x1b", out) + self.assertNotIn("\x08", out) + self.assertIn("\\x1bM\\x1bM\\x08\\x08FAKE", out) def test_decision_rows_stay_out_of_blocks_and_run(self): self._add("ask", self.base, "decide: keep fix B", "operator only", "decision") @@ -265,21 +501,22 @@ class TestOperator(unittest.TestCase): work = self.base / "work" work.mkdir() self._add("light", work, "touch light", "plain", "manual") - self._add("heavy", work, "touch heavy\x1b[2K", "build", "manual") + self._add("heavy", work, "touch heavy", "build", "manual") result = self._feed("y\nh\n") self.assertEqual(result.returncode, 0, result.stderr) light, heavy = result.stdout.split("[2/2]") self.assertIn("[s]kip", light) self.assertNotIn("heavy:", light) - self.assertIn("heavy: memory stage ok", heavy) - self.assertIn("position\tid\tstatus", heavy) + self.assertIn(" heavy memory stage ok", heavy) + self.assertRegex(heavy, r"\n position +id +status +task +eta_min\n") + self.assertNotIn("workdir", heavy) self.assertIn("[h]old in the queue", heavy) self.assertNotIn("\x1b", heavy) self.assertIn("granted operator/heavy: the queue lets it run now", heavy) - self.assertIn("not run operator/heavy", heavy) - self.assertIn("failed operator/heavy: exit 75; the row stays open", heavy) + self.assertEqual(heavy.count("not run operator/heavy"), 2) + self.assertNotIn("failed", heavy) self.assertTrue((work / "light").exists()) - self.assertFalse((work / "heavy\x1b[2K").exists()) + self.assertFalse((work / "heavy").exists()) self.assertEqual(self._open(), ["operator/heavy"]) self.assertFalse((self.root / "queue-history.jsonl").exists()) @@ -359,6 +596,8 @@ class TestOperator(unittest.TestCase): ("add", "x", *good, "--check", "binary:a b"), ("add", "x", *good, "--check", "binary:/usr/bin/git"), ("add", "x", *good, "--check", "pushed:"), + ("add", "x", *good, "--check", "absent:/x/\x1bM\x08FAKE"), + ("add", "x", "--dir", "/srv/\x1bM", "--command", "c", "--reason", "r", "--check", "manual"), ("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"),