memes

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())

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():

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", " ")

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()