From cd543d20cf4e1d4f0dca059e0552267f84ff472f Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Tue, 10 Nov 2026 21:15:48 +0000 Subject: [PATCH] feat(kvb-org-operator): failed and cancelled rows leave the queue and reach the adding agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A command that exits non-zero or cannot start sets its row stopped with the exit under failed and posts it to every agent holding A, asking for the corrected command; its last stderr lines go along only when the operator answers y, since stderr can hold a credential. Exit 130 or a signal keeps the row open. A cancel needs a reason and posts it the same way. add refuses commands with control characters and run never offers such a row, so a newline can no longer run a hidden second command. A leftover background process holding stderr no longer blocks run. run.md describes what run shows and asks. 🎯 Fix: 90 % 💥 Break: 4 % 💸 Cost: 140 % ⏱️ Time: 100 % 🛡️ Security: 100 % 🏆 Score: 62 🚦 Gate: ⚠️ cost 140 % > 100: operator approved the commit Co-Authored-By: Claude Opus 5.5 --- skills/kvb-org-operator/SKILL.md | 10 +- skills/kvb-org-operator/run.md | 26 ++ skills/kvb-org-operator/scripts/operator.py | 352 +++++++++++++++++--- tests/test_operator.py | 193 +++++++++-- 4 files changed, 508 insertions(+), 73 deletions(-) create mode 100644 skills/kvb-org-operator/run.md 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 027ac4d..6bc87ca 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -6,7 +6,7 @@ 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 the command uncolored + 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 @@ -26,26 +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; bold cyan when stdout is a terminal and NO_COLOR is not - set), 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. A repo whose git config names filter commands gets no - preview. 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. @@ -74,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 @@ -83,6 +101,7 @@ import shlex import shutil import subprocess import termios +import textwrap import time from collections.abc import Callable from pathlib import Path @@ -92,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") @@ -107,8 +133,12 @@ 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"] -HIGHLIGHT = "\x1b[1;36m" +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: @@ -130,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): @@ -216,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 @@ -225,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") @@ -284,7 +320,7 @@ def cmd_list(rasci: ModuleType, args: argparse.Namespace) -> int: print() print(f"{number}. {visible(row['task'], '')}") print("```bash") - print(f"cd {shlex.quote(row['directory'])}") + print(f"cd {shlex.quote(visible(row['directory'], ''))}") print(visible(row["command"], "")) print("```") if steps: @@ -297,20 +333,48 @@ 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 highlighted(text: str) -> str: +def painted(text: str, style: str) -> str: if sys.stdout.isatty() and not os.environ.get("NO_COLOR"): - return f"{HIGHLIGHT}{text}{RESET}" + 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( @@ -348,24 +412,27 @@ def evidence(check: str, directory: str, full: bool) -> list[tuple[list[str], st 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: 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(visible(line, "\t"), flush=True) + 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: @@ -421,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, @@ -455,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 {visible(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: {highlighted(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 @@ -491,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)}] {visible(row['task'], '')}") - print(f" directory: {visible(row['directory'], '')}") - print(f" command: {highlighted(visible(row['command'], ''))}") - print(f" why: {visible(row['why'] or '-', '')}") - print(f" check: {visible(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 804a0e9..e5b9685 100644 --- a/tests/test_operator.py +++ b/tests/test_operator.py @@ -18,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", @@ -240,41 +242,177 @@ 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_colors_the_command_only_on_a_terminal_without_no_color(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.assertIn(" command true\n", piped) self.assertNotIn("\x1b", piped) colored = self._tty({}) - self.assertIn(" command: \x1b[1;36mtrue\x1b[0m\r\n", colored) - self.assertEqual(colored.count("\x1b"), 2) + 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.assertIn(" command true\r\n", plain) self.assertNotIn("\x1b", plain) - def test_run_previews_pushed_commits_and_keeps_failed_rows_open(self): + 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("--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" @@ -323,6 +461,20 @@ class TestOperator(unittest.TestCase): 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") self._add("go", self.base, "true", "runnable", "manual") @@ -349,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()) @@ -443,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"),