diff --git a/examples/templates/agents/implementer.md b/examples/example-environment/agents/implementer.md similarity index 100% rename from examples/templates/agents/implementer.md rename to examples/example-environment/agents/implementer.md diff --git a/examples/templates/agents/planner.md b/examples/example-environment/agents/planner.md similarity index 100% rename from examples/templates/agents/planner.md rename to examples/example-environment/agents/planner.md diff --git a/examples/templates/agents/reviewer.md b/examples/example-environment/agents/reviewer.md similarity index 100% rename from examples/templates/agents/reviewer.md rename to examples/example-environment/agents/reviewer.md diff --git a/examples/templates/nightshift.yaml b/examples/example-environment/nightshift.yaml similarity index 100% rename from examples/templates/nightshift.yaml rename to examples/example-environment/nightshift.yaml diff --git a/examples/templates/tasks.md b/examples/example-environment/tasks.md similarity index 100% rename from examples/templates/tasks.md rename to examples/example-environment/tasks.md diff --git a/nightshift/cli.py b/nightshift/cli.py index 417aa80..1db0fc0 100644 --- a/nightshift/cli.py +++ b/nightshift/cli.py @@ -12,6 +12,7 @@ from .init import available_templates, init_project from .pipeline import PipelineRunner from .runlog import RunLogger from .status import build_status, format_status +from .terminal import format_banner, style_text from .tasks import ( ensure_dependencies_satisfied, parse_task_file, @@ -62,6 +63,7 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) try: + _emit_banner(args.command) if args.command == "init": written = init_project(Path(args.root), force=args.force, template=args.template) print("Created NightShift starter files:") @@ -102,7 +104,7 @@ def main(argv: list[str] | None = None) -> int: ensure_dependencies_satisfied(tasks, task) result = runner.run_task(task) print(f"Task: {result.task_id}") - print(f"Status: {result.status}") + print(style_text(f"Status: {result.status}", color=_status_color(result.status), bold=True)) print(f"Retries: {result.retry_count}") print(f"Artifacts: {result.artifact_dir}") print(f"Reason: {result.reason}") @@ -127,5 +129,22 @@ def main(argv: list[str] | None = None) -> int: return 0 +def _emit_banner(command: str) -> None: + if command == "web" or not sys.stdout.isatty(): + return + print(format_banner()) + + +def _status_color(status: str) -> str | None: + lowered = status.lower() + if lowered in {"complete", "pass", "success"}: + return "\x1b[32m" + if lowered in {"failed", "fail", "error"}: + return "\x1b[31m" + if lowered in {"retry", "blocked", "warning"}: + return "\x1b[33m" + return None + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/nightshift/runlog.py b/nightshift/runlog.py index 859e37a..76ff2dc 100644 --- a/nightshift/runlog.py +++ b/nightshift/runlog.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Callable from .artifacts import ArtifactStore +from .terminal import format_console_event_line, format_plain_event_line ConsoleWriter = Callable[[str], None] @@ -40,9 +41,10 @@ class RunLogger: def event(self, event: str, message: str, **fields: object) -> None: safe_fields = _redact_fields(fields) - line = format_log_line(LogEvent(event=event, message=message, fields=safe_fields)) + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + line = format_plain_event_line(timestamp, event, message, safe_fields) if self.console is not None: - self.console(line) + self.console(format_console_event_line(timestamp, event, message, safe_fields)) for path in (self._run_log_path,): if path is None: continue @@ -64,12 +66,7 @@ class NullRunLogger(RunLogger): def format_log_line(log_event: LogEvent) -> str: timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - parts = [timestamp, log_event.event, log_event.message] - for key, value in sorted(log_event.fields.items()): - if value is None or value == "": - continue - parts.append(f"{key}={_format_value(value)}") - return " | ".join(parts) + return format_plain_event_line(timestamp, log_event.event, log_event.message, log_event.fields) def tail_lines(path: Path, limit: int = 100) -> list[str]: @@ -80,11 +77,6 @@ def tail_lines(path: Path, limit: int = 100) -> list[str]: return path.read_text(encoding="utf-8", errors="replace").splitlines()[-limit:] -def _format_value(value: object) -> str: - text = str(value).replace("\n", " ").replace("\r", " ") - return text if text else "" - - def _redact_fields(fields: dict[str, object]) -> dict[str, object]: redacted: dict[str, object] = {} for key, value in fields.items(): diff --git a/nightshift/terminal.py b/nightshift/terminal.py new file mode 100644 index 0000000..5bc4c74 --- /dev/null +++ b/nightshift/terminal.py @@ -0,0 +1,145 @@ +"""Terminal styling helpers for NightShift.""" + +from __future__ import annotations + +import os +import sys +from typing import TextIO +import random + + + +RESET = "\x1b[0m" +BOLD = "\x1b[1m" +DIM = "\x1b[2m" +GREEN = "\x1b[32m" +RED = "\x1b[31m" +YELLOW = "\x1b[33m" +BLUE = "\x1b[34m" +MAGENTA = "\x1b[35m" +CYAN = "\x1b[36m" +WHITE = "\x1b[37m" + +BANNER_MESSAGES = [ + "Real LARPer Hours", + "Who the heck is claude?", + "WHO UP BREAKIN THEY BUILD?", + "me and the boys at 2am lookin for BEANS", + "local-first autonomous coding pipeline", + "why break then build while you're awake?", + "compiling bad ideas into good software", + "local-first synthetic cognition", + "the graveyard shift for software engineering", + "pipeline humming at unsafe levels", + "generated at 3:14am with malicious intent", + "all outputs are guilty until proven correct", + "daemonized software production", + "running tests until morale improves", + "you wouldn't download a senior engineer", + "sleep is temporary. infrastructure is forever.", +] +quote = random.choice(BANNER_MESSAGES) + +def should_style(stream: TextIO | None = None) -> bool: + stream = stream or sys.stdout + if os.environ.get("NO_COLOR"): + return False + if os.environ.get("TERM") == "dumb": + return False + return bool(getattr(stream, "isatty", lambda: False)()) + + +def style_text(text: str, *, color: str | None = None, bold: bool = False, dim: bool = False, stream: TextIO | None = None) -> str: + if not should_style(stream): + return text + parts: list[str] = [] + if bold: + parts.append(BOLD) + if dim: + parts.append(DIM) + if color: + parts.append(color) + if not parts: + return text + return "".join(parts) + text + RESET + + +def format_banner(stream: TextIO | None = None) -> str: + lines = [ + "███╗ ██╗██╗ ██████╗ ██╗ ██╗████████╗███████╗██╗ ██╗██╗███████╗████████╗", + "████╗ ██║██║██╔════╝ ██║ ██║╚══██╔══╝██╔════╝██║ ██║██║██╔════╝╚══██╔══╝", + "██╔██╗ ██║██║██║ ███╗███████║ ██║ ███████╗███████║██║█████╗ ██║ ", + "██║╚██╗██║██║██║ ██║██╔══██║ ██║ ╚════██║██╔══██║██║██╔══╝ ██║ ", + "██║ ╚████║██║╚██████╔╝██║ ██║ ██║ ███████║██║ ██║██║██║ ██║ ", + "╚═╝ ╚═══╝╚═╝ ╚═════╝ ╚═╝ ╚═╝ ╚═╝ ╚══════╝╚═╝ ╚═╝╚═╝╚═╝ ╚═╝ ", + "", + f" [ {quote} ]", + " [ planner | implementer | verifier | audit ]", + "", + " VERSION: 0.1.0-alpha-glizzy", + "-" * 50, + "", + ] + + if not should_style(stream): + return "\n".join(lines) + styled: list[str] = [] + for index, line in enumerate(lines): + if not line: + styled.append(line) + continue + color = CYAN if index < 8 else WHITE + bold = index < 8 or line == "NightShift" + styled.append(style_text(line, color=color, bold=bold, stream=stream)) + return "\n".join(styled) + + +def format_console_event_line( + timestamp: str, + event: str, + message: str, + fields: dict[str, object], + *, + stream: TextIO | None = None, +) -> str: + line = format_plain_event_line(timestamp, event, message, fields) + if not should_style(stream): + return line + color = _event_color(event, fields) + if color is None: + return line + return style_text(line, color=color, stream=stream) + + +def format_plain_event_line(timestamp: str, event: str, message: str, fields: dict[str, object]) -> str: + parts = [timestamp, event, message] + for key, value in sorted(fields.items()): + if value is None or value == "": + continue + parts.append(f"{key}={_format_value(value)}") + return " | ".join(parts) + + +def _event_color(event: str, fields: dict[str, object]) -> str | None: + status = str(fields.get("status", "")).lower() + reason = str(fields.get("reason", "")).lower() + event_name = event.lower() + if status in {"fail", "failed", "error"} or "fail" in reason or "error" in reason: + return RED + if status in {"complete", "pass", "success", "ok"}: + return GREEN + if status in {"retry", "warning", "warn", "blocked"}: + return YELLOW + if event_name.endswith(".start"): + return BLUE + if event_name.endswith(".finish"): + return GREEN + if "tool" in event_name: + return MAGENTA + if "command" in event_name: + return CYAN + return None + + +def _format_value(value: object) -> str: + return str(value).replace("\n", " ").replace("\r", " ") diff --git a/tests/test_terminal.py b/tests/test_terminal.py new file mode 100644 index 0000000..11fcf0f --- /dev/null +++ b/tests/test_terminal.py @@ -0,0 +1,70 @@ +from io import StringIO +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +from nightshift.artifacts import ArtifactStore +from nightshift.runlog import RunLogger +from nightshift.terminal import format_banner, format_console_event_line + + +class FakeTTY(StringIO): + def isatty(self) -> bool: + return True + + +class TerminalStylingTests(unittest.TestCase): + def test_banner_is_plain_without_tty(self) -> None: + banner = format_banner(stream=StringIO()) + self.assertIn("NightShift", banner) + self.assertNotIn("\x1b[", banner) + + def test_banner_uses_ansi_when_tty(self) -> None: + banner = format_banner(stream=FakeTTY()) + self.assertIn("NightShift", banner) + self.assertIn("\x1b[", banner) + + def test_console_event_line_colors_success_and_failure(self) -> None: + success = format_console_event_line( + "2026-05-17T00:00:00Z", + "task.finish", + "Finished task", + {"status": "complete"}, + stream=FakeTTY(), + ) + failure = format_console_event_line( + "2026-05-17T00:00:00Z", + "task.finish", + "Finished task", + {"status": "failed"}, + stream=FakeTTY(), + ) + self.assertIn("\x1b[32m", success) + self.assertIn("\x1b[31m", failure) + self.assertTrue(success.endswith("\x1b[0m")) + self.assertTrue(failure.endswith("\x1b[0m")) + + def test_run_logger_console_output_is_separate_from_run_log(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + artifacts = ArtifactStore(root, ".nightshift", run_id="test-run") + console_lines: list[str] = [] + logger = RunLogger(console=console_lines.append) + logger.bind(artifacts) + with patch( + "nightshift.runlog.format_console_event_line", + return_value="\x1b[32mstyled line\x1b[0m", + ): + logger.event("task.finish", "Finished task", status="complete", token="abc") + + self.assertEqual(console_lines[-1], "\x1b[32mstyled line\x1b[0m") + run_log = artifacts.run_log_path.read_text(encoding="utf-8") + self.assertIn("task.finish", run_log) + self.assertIn("status=complete", run_log) + self.assertNotIn("\x1b[", run_log) + self.assertNotIn("abc", run_log) + + +if __name__ == "__main__": + unittest.main()