From 85462df5abd880b18d38a322a682486a5a01ae32 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Fri, 6 Nov 2026 09:46:14 +0000 Subject: [PATCH 1/3] fix(kvb-org-operator): run no repo-configured command before the operator's yes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agents name the directories of operator rows and can write the repos behind them. Before the operator answered, list and run started git there unsandboxed, and git ran commands the repo config names: core.fsmonitor, the post-index-change hook on an index refresh, filter clean drivers, the pager, external diff and textconv drivers, gpg through log.showSignature, and a lazy fetch of a missing object that runs remote uploadpack through sh. Preview output and task names also reached the terminal raw, so escape sequences could overwrite the displayed command. Every git call before the yes now runs as git -c core.fsmonitor=false -c core.hooksPath=/dev/null -c log.showSignature=false --no-pager --no-optional-locks with GIT_NO_LAZY_FETCH=1, diffs add --no-ext-diff --no-textconv --ignore-submodules=all, a repo whose config names filter commands gets no working-tree preview, and preview output, task, check, table cells and listed commands go through visible(). The approved command itself still runs through $SHELL -ic unchanged. Found by a skeptic while scanning the color change; each hole was reproduced red before the fix. Accepted MEDUSA false positives after kvb-think-criticize (rows in security-accepted.jsonl): MEDUSA-MM-SCAN-058 x3, MEDUSA-GENAI-SCAN-1270 x4, MEDUSA-WM-SCAN-354, MEDUSA-CGS-SCAN-040. 🎯 Fix: 90 % 💥 Break: 3 % 💸 Cost: – ⏱️ Time: – 🛡️ Security: 100 % 🏆 Score: 87 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % ``` /home/kevinveenbirkenbach/Repositories/github.com/kevinveenbirkenbach/skills/.worktrees/feature/kvb-org-operator-run-color ``` Co-Authored-By: Claude Opus 5.5 --- skills/kvb-org-operator/scripts/operator.py | 60 +++++++++++++-------- tests/test_operator.py | 51 +++++++++++++++++- 2 files changed, 88 insertions(+), 23 deletions(-) diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index b3afb04..c3bc231 100755 --- a/skills/kvb-org-operator/scripts/operator.py +++ b/skills/kvb-org-operator/scripts/operator.py @@ -31,7 +31,8 @@ Args: 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 + 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 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 @@ -100,6 +101,9 @@ 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"] def fail(message: str) -> None: @@ -148,11 +152,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 +199,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: @@ -269,10 +277,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(visible(row["command"], "")) print("```") if steps: print() @@ -295,7 +303,16 @@ def visible(text: str, keep: str) -> str: 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,23 +322,17 @@ 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 [] @@ -331,9 +342,14 @@ def show(commands: list[tuple[list[str], str]]) -> None: for argv, cwd in commands: print(f"$ {shlex.join(argv)}", 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) + continue + for line in done.stdout.splitlines(): + print(visible(line, "\t"), flush=True) def ask(holdable: bool, read: Callable[[str], str | None]) -> str: @@ -432,7 +448,7 @@ def cmd_granted(rasci: ModuleType, args: argparse.Namespace) -> int: 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"\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: {visible(row['command'], '')}", flush=True) try: @@ -464,11 +480,11 @@ 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"\n[{number}/{len(steps)}] {visible(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) + print(f" check: {visible(check, '')}", flush=True) holdable = heavy_row(watch, cfg, row) answer = review(row, holdable, typed) if answer == "q": diff --git a/tests/test_operator.py b/tests/test_operator.py index a78c652..6c098ae 100644 --- a/tests/test_operator.py +++ b/tests/test_operator.py @@ -3,10 +3,12 @@ from __future__ import annotations import json +import os import shutil import subprocess import sys import tempfile +import time import unittest from pathlib import Path @@ -234,11 +236,58 @@ class TestOperator(unittest.TestCase): 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"]) + 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_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") From 99323abc3544bbf1e9d1a8a54a28888afaaa1856 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Fri, 6 Nov 2026 09:59:29 +0000 Subject: [PATCH 2/3] feat(kvb-org-operator): color the command in run on a terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The operator found the command hard to spot in the run output. run now prints each step's command bold cyan when stdout is a terminal and NO_COLOR is unset or empty (no-color.org); piped or captured output stays byte-identical. Only the command line is colored. The new test runs run on a pseudo-terminal with and without NO_COLOR and piped. Operator order relayed by the Maintainer session on 2026-11-06; the change was written by a subagent and reviewed here. 🎯 Fix: 95 % 💥 Break: 1 % 💸 Cost: – ⏱️ Time: – 🛡️ Security: 100 % 🏆 Score: 94 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % ``` /home/kevinveenbirkenbach/Repositories/github.com/kevinveenbirkenbach/skills/.worktrees/feature/kvb-org-operator-run-color ``` Co-Authored-By: Claude Opus 5.5 --- skills/kvb-org-operator/scripts/operator.py | 17 ++++++++-- tests/test_operator.py | 35 +++++++++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/skills/kvb-org-operator/scripts/operator.py b/skills/kvb-org-operator/scripts/operator.py index c3bc231..027ac4d 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 the command uncolored + (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. @@ -29,7 +31,8 @@ Args: 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 + 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 @@ -104,6 +107,8 @@ 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" +RESET = "\x1b[0m" def fail(message: str) -> None: @@ -300,6 +305,12 @@ 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: + if sys.stdout.isatty() and not os.environ.get("NO_COLOR"): + return f"{HIGHLIGHT}{text}{RESET}" + return text + + def git_tree(directory: str) -> bool: try: return subprocess.run( @@ -450,7 +461,7 @@ def cmd_granted(rasci: ModuleType, args: argparse.Namespace) -> int: 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: {visible(row['command'], '')}", flush=True) + print(f" command: {highlighted(visible(row['command'], ''))}", flush=True) try: answer = review(row, False, timed) except KeyboardInterrupt: @@ -482,7 +493,7 @@ def cmd_run(rasci: ModuleType, args: argparse.Namespace) -> int: 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: {visible(row['command'], '')}") + print(f" command: {highlighted(visible(row['command'], ''))}") print(f" why: {visible(row['why'] or '-', '')}") print(f" check: {visible(check, '')}", flush=True) holdable = heavy_row(watch, cfg, row) diff --git a/tests/test_operator.py b/tests/test_operator.py index 6c098ae..804a0e9 100644 --- a/tests/test_operator.py +++ b/tests/test_operator.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import pty import shutil import subprocess import sys @@ -94,6 +95,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")))] @@ -231,6 +254,18 @@ class TestOperator(unittest.TestCase): self.assertNotIn("\x1b", out) self.assertEqual(self._open(), ["operator/second", "operator/third"]) + def test_run_colors_the_command_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(" command: \x1b[1;36mtrue\x1b[0m\r\n", colored) + self.assertEqual(colored.count("\x1b"), 2) + plain = self._tty({"NO_COLOR": "1"}) + self.assertIn(" command: true\r\n", plain) + self.assertNotIn("\x1b", plain) + def test_run_previews_pushed_commits_and_keeps_failed_rows_open(self): repo = self._repo("ahead", upstream=True) self._add("push", repo, "false", "branch ahead", f"pushed:{repo}") From cd543d20cf4e1d4f0dca059e0552267f84ff472f Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Tue, 10 Nov 2026 21:15:48 +0000 Subject: [PATCH 3/3] 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"),