path: root/app/tests/test_taskview.py

"""Tests for the in-TUI task view (ui/taskview.py).

The view is driven the same way as the other TUI widgets: the fake curses
module and recording screen from test_tui stand in for a terminal. The
step-sequencing logic is exercised through ``run_steps_inline`` (no thread),
the progress-line parsing through ``TaskView._ingest_line``, and state
transitions through ``handle_event`` + ``_step_mark`` + ``_result_rc``.
"""

import io
import sys
import threading
import unittest
from queue import Empty
from unittest.mock import patch

from tests.test_tui import FakeCurses, FakeScreen
from ui import taskview


def _step(title, rc=0):
    def work(emit, cancel):
        return rc
    return taskview.TaskStep(title, work)


class _FakeTui:
    def setUp(self):
        self.curses = FakeCurses()
        patcher = patch.dict(sys.modules, {"curses": self.curses})
        patcher.start()
        self.addCleanup(patcher.stop)
        taskview.tui._THEME.clear()
        self.addCleanup(taskview.tui._THEME.clear)

    def make_view(self, steps=(), width=80, height=24):
        screen = FakeScreen(width=width, height=height)
        with patch.object(taskview.TaskView, "_worker_main", lambda self: None):
            view = taskview.TaskView(screen, "Setup", list(steps),
                                     clock=lambda: 1000.0)
        return view, screen


class RunStepsInlineTests(unittest.TestCase):
    def test_runs_steps_in_order_and_returns_zero(self):
        order = []
        steps = [
            taskview.TaskStep("a", lambda emit, cancel: order.append("a") or 0),
            taskview.TaskStep("b", lambda emit, cancel: order.append("b") or 0),
        ]
        self.assertEqual(taskview.run_steps_inline(steps), 0)
        self.assertEqual(order, ["a", "b"])

    def test_returns_first_bad_rc_and_continues(self):
        order = []
        steps = [
            taskview.TaskStep("a", lambda emit, cancel: order.append("a") or 1),
            taskview.TaskStep("b", lambda emit, cancel: order.append("b") or 2),
        ]
        self.assertEqual(taskview.run_steps_inline(steps), 1)
        # The second step still ran (warn-and-continue semantics).
        self.assertEqual(order, ["a", "b"])

    def test_passes_emit_and_cancel_to_each_step(self):
        seen = []
        emit = object()
        cancel = object()
        steps = [taskview.TaskStep(
            "a", lambda e, c: seen.append((e, c)) or 0)]
        taskview.run_steps_inline(steps, emit=emit, cancel=cancel)
        self.assertEqual(seen, [(emit, cancel)])


class LineWriterTests(unittest.TestCase):
    def _split(self, text):
        lines = []
        writer = taskview._LineWriter(lines.append)
        writer.write(text)
        writer.flush()
        return lines

    def test_splits_on_newline(self):
        self.assertEqual(self._split("one\ntwo\n"), ["one", "two"])

    def test_splits_on_carriage_return(self):
        # git/tqdm progress updates use \r; each update is its own line.
        self.assertEqual(self._split("a\rb\rc"), ["a", "b", "c"])

    def test_handles_mixed_terminators_and_no_final_newline(self):
        self.assertEqual(self._split("x\ny\r\nz"), ["x", "y", "z"])


class ProgressParsingTests(_FakeTui, unittest.TestCase):
    def _ingest(self, line):
        view, _ = self.make_view(steps=[_step("a")])
        view._ingest_line(line)
        return view

    def test_bytes_progress_is_hidden_from_the_log(self):
        view = self._ingest("AUDIOCPP_PROGRESS downloaded=512 total=2048")
        self.assertEqual(view._progress, (512, 2048))
        self.assertEqual(view._progress_kind, "bytes")
        self.assertEqual(view.log_tail, [])

    def test_percent_progress_kept_in_log(self):
        view = self._ingest("[ 45%] Building CXX object foo.o")
        self.assertEqual(view._progress, (45, 100))
        self.assertEqual(view._progress_kind, "percent")
        self.assertEqual(view.log_tail, ["[ 45%] Building CXX object foo.o"])

    def test_count_progress_from_ninja(self):
        view = self._ingest("[123/456] Compiling bar.cpp")
        self.assertEqual(view._progress, (123, 456))
        self.assertEqual(view._progress_kind, "count")

    def test_git_percent_progress(self):
        view = self._ingest("Receiving objects:  33% (99/300), 1.2 MiB")
        self.assertEqual(view._progress, (33, 100))

    def test_percent_above_one_hundred_ignored(self):
        view = self._ingest("CPU 150% usage")
        self.assertIsNone(view._progress)

    def test_plain_line_only_logs(self):
        view = self._ingest("[INFO] doing work")
        self.assertIsNone(view._progress)
        self.assertEqual(view.log_tail, ["[INFO] doing work"])

    def test_log_tail_is_capped(self):
        view, _ = self.make_view(steps=[_step("a")])
        for i in range(taskview._LOG_TAIL + 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}")


class StateTransitionTests(_FakeTui, unittest.TestCase):
    def _steps(self):
        return [_step("one"), _step("two")]

    def test_success_flow_marks_steps_ok(self):
        view, _ = self.make_view(steps=self._steps())
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        self.assertEqual(view.current, 0)
        view.handle_event({"kind": "step_done", "index": 0, "rc": 0})
        view.handle_event({"kind": "step_start", "index": 1, "title": "two"})
        view.handle_event({"kind": "step_done", "index": 1, "rc": 0})
        view.handle_event({"kind": "finish", "phase": "done", "rc": 0})
        self.assertEqual(view.phase, "done")
        self.assertEqual(view._result_rc(), 0)
        self.assertEqual(view._step_mark(0), ("[OK]", "ok"))
        self.assertEqual(view._step_mark(1), ("[OK]", "ok"))

    def test_failure_marks_step_failed_and_returns_bad_rc(self):
        view, _ = self.make_view(steps=self._steps())
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view.handle_event({"kind": "step_done", "index": 0, "rc": 7})
        view.handle_event({"kind": "finish", "phase": "error", "rc": 7})
        self.assertEqual(view.phase, "error")
        self.assertEqual(view._result_rc(), 7)
        self.assertEqual(view._step_mark(0), ("[FAIL]", "err"))
        self.assertEqual(view._step_mark(1), ("[  ]", "dim"))

    def test_cancelled_run_returns_nonzero(self):
        view, _ = self.make_view(steps=self._steps())
        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})
        self.assertEqual(view.phase, "cancelled")
        self.assertEqual(view._result_rc(), 1)
        # The step interrupted by cancel is marked cancelled, not failed.
        self.assertEqual(view._step_mark(0), ("[x]", "warn"))
        self.assertEqual(view._step_mark(1), ("[  ]", "dim"))

    def test_running_step_shows_a_spinner_mark(self):
        view, _ = self.make_view(steps=self._steps())
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        mark, kind = view._step_mark(0)
        self.assertEqual(kind, "warn")
        self.assertIn("[", mark)


class RenderTests(_FakeTui, unittest.TestCase):
    def _strings(self, screen):
        return " ".join(text for _, _, text, _ in screen.strings)

    def test_running_screen_lists_steps_and_cancel_footer(self):
        view, screen = self.make_view(steps=[_step("one"), _step("two")])
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view.render()
        text = self._strings(screen)
        self.assertIn("one", text)
        self.assertIn("two", text)
        self.assertIn("Esc or q: cancel", text)

    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"})
        view.handle_event({"kind": "step_done", "index": 0, "rc": 0})
        view.handle_event({"kind": "finish", "phase": "done", "rc": 0})
        view.render()
        text = self._strings(screen)
        self.assertIn("completed", text)
        self.assertIn("press any key", text)

    def test_progress_bar_drawn_when_known(self):
        view, screen = self.make_view(steps=[_step("one")])
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("AUDIOCPP_PROGRESS downloaded=512 total=2048")
        view.render()
        text = self._strings(screen)
        self.assertIn("Progress", text)

    def test_progress_label_never_paints_over_the_border(self):
        # Worst-case byte label ("1023.7GB / 1024.0GB") on an 80-col
        # terminal must end before the outer box's border column.
        view, screen = self.make_view(steps=[_step("one")], width=80)
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("AUDIOCPP_PROGRESS downloaded=1099240726528 "
                          "total=1099511627776")
        view.render()
        labels = [t for _, _, t, _ in screen.strings]
        self.assertIn("1023.7GB / 1024.0GB", labels)
        for _, x, text, _ in screen.strings:
            self.assertLessEqual(x + len(text) - 1, 78,
                                 f"{text!r} painted over the border")


class NoWaitTests(_FakeTui, unittest.TestCase):
    """wait_on_finish=False: the view returns to the caller on finish."""

    def make_view(self, steps=(), width=80, height=24, wait_on_finish=False):
        screen = FakeScreen(width=width, height=height)
        with patch.object(taskview.TaskView, "_worker_main", lambda self: None):
            view = taskview.TaskView(screen, "Setup", list(steps),
                                     clock=lambda: 1000.0,
                                     wait_on_finish=wait_on_finish)
        return view, screen

    def _strings(self, screen):
        return " ".join(text for _, _, text, _ in screen.strings)

    def _run_view(self, wait_on_finish):
        """Run() with a synchronous worker that posts a successful finish."""
        screen = FakeScreen(width=80, height=24)
        step = _step("one")

        def fake_worker_main(self):
            self._queue.put({"kind": "step_start", "index": 0,
                             "title": "one"})
            self._queue.put({"kind": "step_done", "index": 0, "rc": 0})
            self._queue.put({"kind": "finish", "phase": "done", "rc": 0})

        with patch.object(taskview.TaskView, "_worker_main",
                          fake_worker_main):
            view = taskview.TaskView(screen, "Setup", [step],
                                     clock=lambda: 1000.0,
                                     wait_on_finish=wait_on_finish)
            # Replace the daemon thread with a synchronous runner so the
            # events are queued before run() enters its loop (deterministic).
            view._worker = _SyncWorker(view._worker_main)
        return view.run()

    def test_returns_immediately_when_not_waiting(self):
        # No scripted keys: run() must return on finish without asking for one.
        self.assertEqual(self._run_view(wait_on_finish=False), 0)

    def test_done_footer_omits_key_hint_when_not_waiting(self):
        view, screen = 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": 0})
        view.handle_event({"kind": "finish", "phase": "done", "rc": 0})
        view.render()
        text = self._strings(screen)
        self.assertIn("completed", text)
        self.assertNotIn("press any key", text)


class InputModeRestoreTests(_FakeTui, unittest.TestCase):
    """run() must leave the screen blocking again when the view exits.

    The views drive their redraw loop with a timed getch; if that cadence
    leaked into the hub, single-getch dialogs like tui.flash would dismiss
    themselves after one timeout instead of waiting for a key.
    """

    def test_task_view_run_restores_blocking_getch(self):
        def fake_worker_main(view):
            view._queue.put({"kind": "step_start", "index": 0,
                             "title": "one"})
            view._queue.put({"kind": "step_done", "index": 0, "rc": 0})
            view._queue.put({"kind": "finish", "phase": "done", "rc": 0})

        screen = FakeScreen(width=80, height=24)
        with patch.object(taskview.TaskView, "_worker_main",
                          fake_worker_main):
            view = taskview.TaskView(screen, "Setup", [_step("one")],
                                     clock=lambda: 1000.0,
                                     wait_on_finish=False)
            view._worker = _SyncWorker(view._worker_main)
            view.run()
        self.assertEqual(screen.timeouts[-1], -1)

    def test_lanes_view_run_restores_blocking_getch(self):
        screen = FakeScreen(keys=[27])   # any key leaves the finished view
        lanes = [taskview.TaskLane("A", [_step("a")]),
                 taskview.TaskLane("B", [_step("b")])]
        view = taskview.LanesView(screen, "Setup", lanes,
                                  clock=lambda: 1000.0)
        with patch.object(taskview.threading, "Thread", _SyncThread):
            view.run()
        self.assertEqual(screen.timeouts[-1], -1)


class _SyncWorker:
    """A stand-in for threading.Thread that runs the target synchronously."""

    def __init__(self, fn):
        self._fn = fn

    def start(self):
        self._fn()


class _SyncThread:
    """A threading.Thread stand-in that runs its target on start()."""

    def __init__(self, target=None, args=(), kwargs=None, daemon=None):
        self._target = target
        self._args = args

    def start(self):
        self._target(*self._args)

    def join(self, timeout=None):
        pass


class LabelTests(unittest.TestCase):
    def test_fmt_bytes(self):
        self.assertEqual(taskview._fmt_bytes(512), "512B")
        self.assertEqual(taskview._fmt_bytes(2048), "2.0KB")
        self.assertEqual(taskview._fmt_bytes(5 * 1024 * 1024), "5.0MB")

    def test_progress_label_bytes(self):
        self.assertEqual(taskview._progress_label((512, 2048), "bytes"),
                         "512B / 2.0KB")

    def test_progress_label_count(self):
        self.assertEqual(taskview._progress_label((3, 10), "count"), "3/10")

    def test_progress_label_percent(self):
        self.assertEqual(taskview._progress_label((45, 100), "percent"),
                         "45%")


class RunLanesTests(_FakeTui, unittest.TestCase):
    """run_lanes: one lane falls back to run_steps, two use the split view."""

    def test_single_lane_delegates_to_run_steps(self):
        lane = taskview.TaskLane("Build", [_step("one")])
        with patch.object(taskview, "run_steps", return_value=0) as mk_run:
            rc = taskview.run_lanes(None, "Setup", [lane])
        self.assertEqual(rc, 0)
        mk_run.assert_called_once()
        self.assertEqual(mk_run.call_args[0][2], lane.steps)

    def test_empty_lanes_are_dropped(self):
        lane = taskview.TaskLane("Build", [_step("one")])
        with patch.object(taskview, "run_steps", return_value=0) as mk_run:
            rc = taskview.run_lanes(
                None, "Setup",
                [taskview.TaskLane("Empty", []), lane])
        self.assertEqual(rc, 0)
        self.assertEqual(mk_run.call_args[0][2], lane.steps)

    def test_no_lanes_returns_zero_without_running(self):
        with patch.object(taskview, "run_steps") as mk_run, \
                patch.object(taskview, "LanesView") as mk_view:
            rc = taskview.run_lanes(None, "Setup", [])
        self.assertEqual(rc, 0)
        mk_run.assert_not_called()
        mk_view.assert_not_called()

    def test_two_lanes_uses_the_split_view(self):
        lanes = [taskview.TaskLane("Build", [_step("one")]),
                 taskview.TaskLane("Download", [_step("two")])]
        with patch.object(taskview, "LanesView") as mk_view:
            mk_view.return_value.run.return_value = 0
            rc = taskview.run_lanes(None, "Setup", lanes)
        self.assertEqual(rc, 0)
        mk_view.assert_called_once_with(None, "Setup", lanes)
        mk_view.return_value.run.assert_called_once_with()


class ThreadRouterTests(unittest.TestCase):
    def test_routes_to_registered_thread_and_falls_back(self):
        fallback = io.StringIO()
        router = taskview._ThreadRouter(fallback)
        captured = []
        writer = taskview._LineWriter(captured.append)
        with router.for_thread(writer):
            router.write("hello\n")
        self.assertEqual(captured, ["hello"])
        # Unregistered thread falls through to the original stream.
        router.write("fallback\n")
        self.assertEqual(fallback.getvalue(), "fallback\n")

    def test_concurrent_prints_land_in_their_own_writer(self):
        registry = {}
        router = taskview._ThreadRouter(io.StringIO(), registry)
        lines_a, lines_b = [], []
        wa = taskview._LineWriter(lines_a.append)
        wb = taskview._LineWriter(lines_b.append)
        ready_a, ready_b = threading.Event(), threading.Event()
        go = threading.Event()

        def body(writer, tag, ready):
            with router.for_thread(writer):
                ready.set()
                go.wait()
                for i in range(50):
                    print(f"{tag}-{i}")

        with patch.object(sys, "stdout", router):
            threads = [threading.Thread(target=body,
                                        args=(wa, "A", ready_a)),
                       threading.Thread(target=body,
                                        args=(wb, "B", ready_b))]
            for t in threads:
                t.start()
            ready_a.wait()
            ready_b.wait()
            go.set()
            for t in threads:
                t.join()
        self.assertTrue(lines_a)
        self.assertTrue(lines_b)
        self.assertTrue(all(line.startswith("A-") for line in lines_a))
        self.assertTrue(all(line.startswith("B-") for line in lines_b))


class LanesViewTests(_FakeTui, unittest.TestCase):
    """LanesView: parallel lanes, isolated logs/progress, split rendering."""

    def make_view(self, lanes, width=80, height=24):
        screen = FakeScreen(width=width, height=height)
        view = taskview.LanesView(screen, "Setup", lanes,
                                  clock=lambda: 1000.0)
        return view, screen

    def _two_lanes(self):
        return [taskview.TaskLane("Build", [_step("one"), _step("two")]),
                taskview.TaskLane("Download models", [_step("dl")])]

    def test_two_lane_workers_run_both_steps(self):
        ran = []

        def work(name):
            def _w(emit, cancel):
                ran.append(name)
                return 0
            return _w

        lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work("A"))]),
                 taskview.TaskLane("B", [taskview.TaskStep("b", work("B"))])]
        view, _ = self.make_view(lanes)
        registry = {}
        router = taskview._ThreadRouter(sys.stdout, registry)
        threads = [threading.Thread(target=view._lane_worker,
                                    args=(lane_state, router, view._cancel))
                   for lane_state in view._lanes]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        view._drain()
        self.assertCountEqual(ran, ["A", "B"])
        self.assertTrue(all(lane.finished for lane in view._lanes))

    def test_lane_worker_routes_prints_to_its_lane(self):
        def work(emit, cancel):
            print("lane-A-log")
            return 0

        lanes = [taskview.TaskLane("A", [taskview.TaskStep("a", work)])]
        view, _ = self.make_view(lanes)
        registry = {}
        router = taskview._ThreadRouter(io.StringIO(), registry)
        lane_state = view._lanes[0]
        with patch.object(sys, "stdout", router):
            view._lane_worker(lane_state, router, view._cancel)
        events = []
        while True:
            try:
                events.append(lane_state.queue.get_nowait())
            except Empty:
                break
        lines = [e["text"] for e in events if e.get("kind") == "line"]
        self.assertIn("lane-A-log", lines)
        self.assertTrue(any(e.get("kind") == "lane_finish" for e in events))

    def test_progress_is_isolated_per_lane(self):
        view, _ = self.make_view(self._two_lanes())
        view._ingest_lane_line(view._lanes[0],
                               "AUDIOCPP_PROGRESS downloaded=512 total=2048")
        self.assertEqual(view._lanes[0].progress, (512, 2048))
        self.assertIsNone(view._lanes[1].progress)
        view._ingest_lane_line(view._lanes[1], "[ 45%] building")
        self.assertEqual(view._lanes[1].progress, (45, 100))
        self.assertEqual(view._lanes[0].progress_kind, "bytes")

    def test_result_rc_returns_first_nonzero_across_lanes(self):
        view, _ = self.make_view(self._two_lanes())
        # Lane 0 step 0 succeeds, lane 0 step 1 fails, lane 1 succeeds.
        view._lanes[0].results = [0, 7]
        view._lanes[1].results = [0]
        self.assertEqual(view._result_rc(), 7)

    def test_all_lanes_finished_flips_phase_to_done(self):
        view, _ = self.make_view(self._two_lanes())
        for lane in view._lanes:
            lane.finished = True
            lane.rc = 0
        view._drain()
        self.assertEqual(view.phase, "done")
        self.assertEqual(view._result_rc(), 0)

    def test_error_phase_when_a_lane_reports_failure(self):
        view, _ = self.make_view(self._two_lanes())
        view._lanes[0].finished = True
        view._lanes[0].rc = 3
        view._lanes[1].finished = True
        view._lanes[1].rc = 0
        view._drain()
        self.assertEqual(view.phase, "error")

    def test_cancelled_run_returns_nonzero(self):
        view, _ = self.make_view(self._two_lanes())
        view._cancel.set()
        for lane in view._lanes:
            lane.finished = True
        view._drain()
        self.assertEqual(view.phase, "cancelled")
        self.assertTrue(view.cancelled)
        self.assertEqual(view._result_rc(), 1)

    def test_split_render_draws_both_lane_titles(self):
        view, screen = self.make_view(self._two_lanes())
        view.render()
        text = " ".join(t for _, _, t, _ in screen.strings)
        self.assertIn("Build", text)
        self.assertIn("Download models", text)
        self.assertIn("Esc or q: cancel", text)

    def test_split_render_stacks_on_narrow_terminal(self):
        view, screen = self.make_view(self._two_lanes(), width=60)
        view.render()
        text = " ".join(t for _, _, t, _ in screen.strings)
        self.assertIn("Build", text)
        self.assertIn("Download models", text)

    def test_split_render_progress_labels_stay_inside_their_panes(self):
        # Regression: the label used to start on the pane's right border
        # column, so the download pane drew over it ("52/794" -> "52").
        view, screen = self.make_view(self._two_lanes(), width=92)
        view._ingest_lane_line(view._lanes[0],
                               "[52/794] Building CUDA object ggml")
        view._ingest_lane_line(view._lanes[1], "[ 45%] building foo.o")
        view.render()
        pane_w = (screen.width - 3) // 2
        rects = [(1, pane_w), (1 + pane_w + 1, (screen.width - 3) - pane_w)]
        for expected, (px, pw) in zip(("52/794", "45%"), rects,
                                      strict=False):
            found = [(x, t) for _, x, t, _ in screen.strings if t == expected]
            self.assertEqual(len(found), 1, expected)
            x, label = found[0]
            # The whole label sits inside its own pane's interior.
            self.assertGreaterEqual(x, px + 1)
            self.assertLessEqual(x + len(label) - 1, px + pw - 2)

    def test_split_render_bytes_label_fits_in_its_pane(self):
        view, screen = self.make_view(self._two_lanes(), width=76)
        view._ingest_lane_line(view._lanes[0],
                               "AUDIOCPP_PROGRESS downloaded=512 total=2048")
        view.render()
        found = [(x, t) for _, x, t, _ in screen.strings
                 if t == "512B / 2.0KB"]
        self.assertEqual(len(found), 1)
        x, label = found[0]
        pane_w = (screen.width - 3) // 2
        self.assertLessEqual(x + len(label) - 1, 1 + pane_w - 2)


class SilenceCueTests(_FakeTui, unittest.TestCase):
    """The "(no output 6m)" cue for a running step that stopped emitting."""

    def test_silence_cue_text_format(self):
        self.assertEqual(taskview._silence_cue_text(90), "(no output 1m)")
        self.assertEqual(taskview._silence_cue_text(3700), "(no output 61m)")

    def test_no_cue_while_output_is_recent(self):
        view, _ = self.make_view(steps=[_step("one")])
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("working")
        self.assertIsNone(taskview._silent_secs(view.last_line_at,
                                                view.last_line_at + 59))

    def test_cue_after_the_threshold(self):
        view, _ = self.make_view(steps=[_step("one")])
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("working")
        self.assertEqual(taskview._silent_secs(view.last_line_at,
                                               view.last_line_at + 60), 60)

    def test_step_start_resets_and_done_clears_the_tracking(self):
        view, _ = self.make_view(steps=[_step("one"), _step("two")])
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        start = view.last_line_at
        view.handle_event({"kind": "step_done", "index": 0, "rc": 0})
        self.assertIsNone(view.last_line_at)
        view.handle_event({"kind": "step_start", "index": 1, "title": "two"})
        self.assertGreaterEqual(view.last_line_at, start)

    def test_render_shows_the_cue_for_a_silent_step(self):
        now = [1000.0]

        def clock():
            return now[0]

        screen = FakeScreen(width=80, height=24)
        with patch.object(taskview.TaskView, "_worker_main",
                          lambda self: None):
            view = taskview.TaskView(screen, "Setup", [_step("one")],
                                     clock=clock)
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("working")
        now[0] = view.last_line_at + 300
        view.render()
        text = " ".join(t for _, _, t, _ in screen.strings)
        self.assertIn("(no output 5m)", text)

    def test_render_hides_the_cue_when_output_is_fresh(self):
        now = [1000.0]

        def clock():
            return now[0]

        screen = FakeScreen(width=80, height=24)
        with patch.object(taskview.TaskView, "_worker_main",
                          lambda self: None):
            view = taskview.TaskView(screen, "Setup", [_step("one")],
                                     clock=clock)
        view.handle_event({"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_line("working")
        now[0] = view.last_line_at + 5
        view.render()
        text = " ".join(t for _, _, t, _ in screen.strings)
        self.assertNotIn("(no output", text)

    def test_lane_pane_shows_the_cue_for_a_silent_lane(self):
        now = [1000.0]

        def clock():
            return now[0]

        screen = FakeScreen(width=80, height=24)
        view = taskview.LanesView(
            screen, "Setup",
            [taskview.TaskLane("Build", [_step("one")])], clock=clock)
        lane = view._lanes[0]
        view._handle_lane_event(
            lane, {"kind": "step_start", "index": 0, "title": "one"})
        view._ingest_lane_line(lane, "working")
        now[0] = lane.last_line_at + 120
        view.render()
        text = " ".join(t for _, _, t, _ in screen.strings)
        self.assertIn("(no output 2m)", text)


if __name__ == "__main__":
    unittest.main()