feat: unified logging with logging_kit.py

1 files changed, 137 insertions, 3 deletions

diff --git a/app/tests/test_taskview.py b/app/tests/test_taskview.py
index 646fccb..1db708d 100644
--- a/app/tests/test_taskview.py
+++ b/app/tests/test_taskview.py
@@ -9,8 +9,10 @@ transitions through ``handle_event`` + ``_step_mark`` + ``_result_rc``.
 
 import io
 import sys
+import tempfile
 import threading
 import unittest
+from pathlib import Path
 from queue import Empty
 from unittest.mock import patch
 
@@ -30,6 +32,12 @@ class _FakeTui:
         patcher = patch.dict(sys.modules, {"curses": self.curses})
         patcher.start()
         self.addCleanup(patcher.stop)
+        # Console mirroring is disabled by default so tests stay hermetic;
+        # the console-log tests patch day_stream with a real temp file.
+        log_patcher = patch.object(taskview.logging_kit, "day_stream",
+                                   return_value=None)
+        log_patcher.start()
+        self.addCleanup(log_patcher.stop)
         taskview.tui._THEME.clear()
         self.addCleanup(taskview.tui._THEME.clear)
 
@@ -41,6 +49,18 @@ class _FakeTui:
         return view, screen
 
 
+class _TempLog:
+    """A temp target for the tui_ day stream (patched over day_stream)."""
+
+    def _log_target(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        path = Path(tmp.name) / "tui_20260828.log"
+        handle = path.open("w", encoding="utf-8")
+        self.addCleanup(handle.close)
+        return path, handle
+
+
 class RunStepsInlineTests(unittest.TestCase):
     def test_runs_steps_in_order_and_returns_zero(self):
         order = []
@@ -128,10 +148,100 @@ class ProgressParsingTests(_FakeTui, unittest.TestCase):
 
     def test_log_tail_is_capped(self):
         view, _ = self.make_view(steps=[_step("a")])
-        for i in range(taskview._LOG_TAIL + 5):
+        for i in range(taskview._LOG_KEEP + 5):
             view._ingest_line(f"line {i}")
-        self.assertEqual(len(view.log_tail), taskview._LOG_TAIL)
-        self.assertEqual(view.log_tail[-1], f"line {taskview._LOG_TAIL + 4}")
+        self.assertEqual(len(view.log_tail), taskview._LOG_KEEP)
+        self.assertEqual(view.log_tail[-1], f"line {taskview._LOG_KEEP + 4}")
+
+
+class ConsoleLogTests(_TempLog, _FakeTui, unittest.TestCase):
+    """The tui_ day-stream mirror: header, lines, markers, exclusions."""
+
+    def test_run_header_lines_and_step_markers_are_written(self):
+        path, handle = self._log_target()
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=handle):
+            view, _ = self.make_view(steps=[_step("one")])
+            view.handle_event({"kind": "step_start", "index": 0,
+                               "title": "one"})
+            view._ingest_line("cloning into 'audio.cpp'...")
+            view.handle_event({"kind": "step_done", "index": 0, "rc": 0})
+            view.handle_event({"kind": "finish", "phase": "done", "rc": 0})
+            view._on_stop()
+        text = path.read_text(encoding="utf-8")
+        self.assertIn("=== Setup —", text)
+        self.assertIn("--- one ---", text)
+        self.assertIn("cloning into 'audio.cpp'...", text)
+        self.assertIn("[OK] one (exit 0)", text)
+        self.assertIn("=== done (exit 0) ===", text)
+
+    def test_machine_progress_lines_are_not_logged(self):
+        path, handle = self._log_target()
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=handle):
+            view, _ = self.make_view(steps=[_step("a")])
+            view._ingest_line("AUDIOCPP_PROGRESS downloaded=1 total=2")
+            view._ingest_line("[ 45%] building")
+            view._on_stop()
+        text = path.read_text(encoding="utf-8")
+        self.assertNotIn("AUDIOCPP_PROGRESS", text)
+        self.assertIn("[ 45%] building", text)
+
+    def test_failed_step_is_marked_in_the_log(self):
+        path, handle = self._log_target()
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=handle):
+            view, _ = self.make_view(steps=[_step("one")])
+            view.handle_event({"kind": "step_start", "index": 0,
+                               "title": "one"})
+            view.handle_event({"kind": "step_done", "index": 0, "rc": 7})
+            view._on_stop()
+        text = path.read_text(encoding="utf-8")
+        self.assertIn("[FAIL] one (exit 7)", text)
+
+    def test_cancelled_step_is_marked_in_the_log(self):
+        path, handle = self._log_target()
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=handle):
+            view, _ = self.make_view(steps=[_step("one")])
+            view.handle_event({"kind": "step_start", "index": 0,
+                               "title": "one"})
+            view.handle_event({"kind": "step_cancelled", "index": 0})
+            view.handle_event({"kind": "finish", "phase": "cancelled",
+                               "rc": 1})
+            view._on_stop()
+        text = path.read_text(encoding="utf-8")
+        self.assertIn("[x] one (cancelled)", text)
+        self.assertIn("=== cancelled (exit 1) ===", text)
+
+    def test_unwritable_log_disables_the_mirror(self):
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=None):
+            view, _ = self.make_view(steps=[_step("a")])
+            view._ingest_line("hello")   # must not raise
+            view._on_stop()
+        self.assertEqual(view.log_tail, ["hello"])
+
+    def test_lane_output_is_mirrored_with_lane_markers(self):
+        path, handle = self._log_target()
+        with patch.object(taskview.logging_kit, "day_stream",
+                          return_value=handle):
+            screen = FakeScreen(width=80, height=24)
+            view = taskview.LanesView(
+                screen, "Setup",
+                [taskview.TaskLane("Build", [_step("one")])],
+                clock=lambda: 1000.0)
+            lane = view._lanes[0]
+            view._handle_lane_event(lane, {"kind": "step_start",
+                                           "index": 0, "title": "one"})
+            view._ingest_lane_line(lane, "compiling foo.o")
+            view._handle_lane_event(lane, {"kind": "lane_finish", "rc": 0})
+            view._console_log.close()
+        text = path.read_text(encoding="utf-8")
+        self.assertIn("=== Setup —", text)
+        self.assertIn("--- [Build] one ---", text)
+        self.assertIn("compiling foo.o", text)
+        self.assertIn("=== [Build] finished (exit 0) ===", text)
 
 
 class StateTransitionTests(_FakeTui, unittest.TestCase):
@@ -193,6 +303,30 @@ class RenderTests(_FakeTui, unittest.TestCase):
         self.assertIn("two", text)
         self.assertIn("Esc or q: cancel", text)
 
+    def test_log_tail_fills_the_available_height(self):
+        # A tall terminal shows all 20 lines (the old 10-line cap would
+        # have left the bottom blank).
+        view, screen = self.make_view(steps=[_step("one")], height=30)
+        for i in range(20):
+            view._ingest_line(f"line {i}")
+        view.render()
+        shown = [t for _, _, t, _ in screen.strings if t.startswith("line ")]
+        self.assertEqual(len(shown), 20)
+        self.assertEqual(shown[0], "line 0")
+        self.assertEqual(shown[-1], "line 19")
+
+    def test_log_tail_shows_the_most_recent_lines_that_fit(self):
+        # The tail is anchored to the latest output: on a small screen the
+        # earliest lines scroll off while the newest stay visible.
+        view, screen = self.make_view(steps=[_step("one")], height=24)
+        for i in range(20):
+            view._ingest_line(f"line {i}")
+        view.render()
+        shown = [t for _, _, t, _ in screen.strings if t.startswith("line ")]
+        self.assertGreater(len(shown), 10)   # more than the old fixed cap
+        self.assertEqual(shown[-1], "line 19")
+        self.assertNotIn("line 0", shown)
+
     def test_done_screen_shows_the_completion_footer(self):
         view, screen = self.make_view(steps=[_step("one")])
         view.handle_event({"kind": "step_start", "index": 0, "title": "one"})