feat(kvb-org-operator): color the command in run on a terminal
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
85462df5ab
commit
99323abc35
2 files changed
+49
-3
No files matched your search
@@ -6,6 +6,8 @@ Environment:
|
|||||||
by rasci.py; the rows are its tasks operator/<slug>.
|
by rasci.py; the rows are its tasks operator/<slug>.
|
||||||
CLAUDE_CODE_SESSION_ID: the adding session, recorded as A; other runtimes
|
CLAUDE_CODE_SESSION_ID: the adding session, recorded as A; other runtimes
|
||||||
pass --session.
|
pass --session.
|
||||||
|
NO_COLOR: when set and not empty, run prints the command uncolored
|
||||||
|
(https://no-color.org).
|
||||||
PATH: searched by the binary check.
|
PATH: searched by the binary check.
|
||||||
SHELL: run executes each approved command with `$SHELL -ic`, so the
|
SHELL: run executes each approved command with `$SHELL -ic`, so the
|
||||||
operator's aliases and functions work; required by run.
|
operator's aliases and functions work; required by run.
|
||||||
@@ -29,7 +31,8 @@ Args:
|
|||||||
set the passing ones done.
|
set the passing ones done.
|
||||||
run: walk the open rows that hold a directory and a command, in list
|
run: walk the open rows that hold a directory and a command, in list
|
||||||
order. Per row it prints directory, command (control characters made
|
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
|
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
|
diff. A repo whose git config names filter commands gets no
|
||||||
preview. y runs the command in its directory, then sets the row done
|
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.
|
# 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"]
|
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"]
|
DIFF = [*GIT, "diff", "--no-ext-diff", "--no-textconv", "--ignore-submodules=all"]
|
||||||
|
HIGHLIGHT = "\x1b[1;36m"
|
||||||
|
RESET = "\x1b[0m"
|
||||||
|
|
||||||
|
|
||||||
def fail(message: str) -> None:
|
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)
|
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:
|
def git_tree(directory: str) -> bool:
|
||||||
try:
|
try:
|
||||||
return subprocess.run(
|
return subprocess.run(
|
||||||
@@ -450,7 +461,7 @@ def cmd_granted(rasci: ModuleType, args: argparse.Namespace) -> int:
|
|||||||
row = approved[0]
|
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"\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" directory: {visible(row['directory'], '')}")
|
||||||
print(f" command: {visible(row['command'], '')}", flush=True)
|
print(f" command: {highlighted(visible(row['command'], ''))}", flush=True)
|
||||||
try:
|
try:
|
||||||
answer = review(row, False, timed)
|
answer = review(row, False, timed)
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
@@ -482,7 +493,7 @@ def cmd_run(rasci: ModuleType, args: argparse.Namespace) -> int:
|
|||||||
check = "manual" if problem(row["check"]) else row["check"]
|
check = "manual" if problem(row["check"]) else row["check"]
|
||||||
print(f"\n[{number}/{len(steps)}] {visible(row['task'], '')}")
|
print(f"\n[{number}/{len(steps)}] {visible(row['task'], '')}")
|
||||||
print(f" directory: {visible(row['directory'], '')}")
|
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" why: {visible(row['why'] or '-', '')}")
|
||||||
print(f" check: {visible(check, '')}", flush=True)
|
print(f" check: {visible(check, '')}", flush=True)
|
||||||
holdable = heavy_row(watch, cfg, row)
|
holdable = heavy_row(watch, cfg, row)
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
|
import pty
|
||||||
import shutil
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
@@ -94,6 +95,28 @@ class TestOperator(unittest.TestCase):
|
|||||||
cwd=self.tmp.name, check=False,
|
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]:
|
def _open(self) -> list[str]:
|
||||||
return [row["task"] for row in json.loads("\n".join(self._out("list", "--json")))]
|
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.assertNotIn("\x1b", out)
|
||||||
self.assertEqual(self._open(), ["operator/second", "operator/third"])
|
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):
|
def test_run_previews_pushed_commits_and_keeps_failed_rows_open(self):
|
||||||
repo = self._repo("ahead", upstream=True)
|
repo = self._repo("ahead", upstream=True)
|
||||||
self._add("push", repo, "false", "branch ahead", f"pushed:{repo}")
|
self._add("push", repo, "false", "branch ahead", f"pushed:{repo}")
|
||||||
|
|||||||
Reference in new issue
Block a user