Pin the activity log to its newest entry

src/hatty/ui/activity_log_panel.py

@@ -25,8 +25,20 @@
 logbook/event_stream) — it dedupes against the last few entries rendered, since
 a live push can legitimately overlap the last entry `load_history` already
 drew (the window fetch and the stream subscription have no shared cursor).
-Appending to the selectable list never moves the current selection, so a live
-push while maximized doesn't yank the highlight away from what's being read.
+Appending to the selectable list moves the current selection to the new
+newest entry only if it was already there — parked on an older entry, a live
+push leaves the highlight alone so it doesn't yank away what's being read
+(issue #44).
+
+The docked ticker follows the same rule: it should always show the
+newest entry unless the reader has scrolled up away from it. Textual's `Log`
+has an `auto_scroll` flag for exactly this, but it only fires when the
+*previous* write already ended at the bottom — and `Log.clear()` zeroes
+`virtual_size` without resetting `scroll_y`, so every reload/reflow here
+(each does `clear()` then re-writes) leaves that precondition false and
+auto_scroll silently stops working, including for every live append after
+it. `_scroll_log_to_tail` makes the "pin to newest" case explicit instead of
+relying on that precondition (issue #44).
 
 The panel retains its rendered entries (`_entries`, capped in lockstep with
 the `Log`'s own `max_lines`) so it can re-truncate them to the true width
@@ -37,9 +49,11 @@
 never revisited it, so maximizing did nothing for already-written lines).
 The two bodies track their rendered width independently (`_rendered_width` /
 `_options_rendered_width`) so toggling `-maximized` back and forth never
-skips a needed re-render. Loading a fresh history (a scope/page change)
-always resets the selectable list's highlight to the newest entry; a live
-append leaves it where it is."""
+skips a needed re-render. A reflow preserves the ticker's scroll position
+(pinned-to-tail readers stay pinned, scrolled-up readers stay where they
+were) the same way it preserves the selectable list's highlight. Loading a
+fresh history (a scope/page change) always resets the selectable list's
+highlight to the newest entry, same as the ticker."""
 
 from collections import deque
 
@@ -190,6 +204,12 @@ def _options_width(self) -> int:
         options = self.query_one("#log_options", OptionList)
         return max(20, options.scrollable_content_region.width or options.content_size.width or 50)
 
+    @staticmethod
+    def _scroll_log_to_tail(log: Log) -> None:
+        """Pin the ticker to its newest line — see the module docstring for
+        why this can't be left to Log's own auto_scroll."""
+        log.scroll_end(animate=False, immediate=True, x_axis=False)
+
     def _render_detail(self, index: int | None) -> None:
         detail = self.query_one("#log_detail", Static)
         if not self._entries:
@@ -232,12 +252,14 @@ def load_history(self, entries: list[LogEntry]) -> None:
         if not entries:
             log.write_line("(no history available)")
             self._rendered_width = 0
+            self._scroll_log_to_tail(log)
             if self.has_class("-maximized"):
                 self._render_options(keep_highlighted=False)
             return
         width = self._line_width()
         log.write_lines([format_log_line(entry, width) for entry in entries])
         self._rendered_width = width
+        self._scroll_log_to_tail(log)
         if self.has_class("-maximized"):
             self._render_options(keep_highlighted=False)
 
@@ -246,21 +268,27 @@ def add_log_entry(self, entry: LogEntry) -> None:
         — reuses format_log_line so a device event gets the same ⚡ form and
         width truncation as the initial load. Skips an entry already rendered
         in the last _DEDUPE_WINDOW (the fetch/stream boundary can overlap).
-        Appending to the selectable list never moves its highlighted index,
-        so a live push while maximized can't yank the selection away."""
+        Appending to the selectable list moves the highlighted index to the
+        new entry only when it was already on the newest one — parked on an
+        older entry, a live push leaves the selection alone (issue #44)."""
         key = self._dedupe_key(entry)
         if key in self._recent_keys:
             return
         self._recent_keys.append(key)
         self._entries.append(entry)
         width = self._line_width()
+        # write_line's own auto_scroll correctly sticks to the tail (or not)
+        # here, since load_history/clear/_reflow_lines keep scroll_y truthful.
         self.query_one("#log_widget", Log).write_line(format_log_line(entry, width))
         self._rendered_width = width
         if self.has_class("-maximized"):
             options = self.query_one("#log_options", OptionList)
+            at_newest = options.highlighted is None or options.highlighted == options.option_count - 1
             options_width = self._options_width()
             options.add_option(format_log_line(entry, options_width))
             self._options_rendered_width = options_width
+            if at_newest:
+                options.highlighted = options.option_count - 1
 
     def _reflow_lines(self) -> None:
         """Re-truncate every retained entry to the current width — the
@@ -281,9 +309,18 @@ def _reflow_lines(self) -> None:
         if width == self._rendered_width:
             return
         log = self.query_one("#log_widget", Log)
+        # format_log_line always yields exactly one line per entry, so the
+        # rewritten content has the same line count — scroll position stays
+        # meaningful across the clear()/write_lines below.
+        at_tail = log.is_vertical_scroll_end
+        prior_y = log.scroll_offset.y
         log.clear()
         log.write_lines([format_log_line(entry, width) for entry in self._entries])
         self._rendered_width = width
+        if at_tail:
+            self._scroll_log_to_tail(log)
+        else:
+            log.scroll_to(y=prior_y, animate=False, immediate=True)
 
     def on_resize(self, event: events.Resize) -> None:
         self._reflow_lines()
@@ -309,7 +346,9 @@ def set_maximized(self, maximized: bool) -> None:
         self.call_after_refresh(self._reflow_lines)
 
     def clear(self) -> None:
-        self.query_one("#log_widget", Log).clear()
+        log = self.query_one("#log_widget", Log)
+        log.clear()
+        self._scroll_log_to_tail(log)
         self.query_one("#log_options", OptionList).clear_options()
         self.query_one("#log_detail", Static).update("")
         self._recent_keys.clear()

tests/test_log_maximize.py

@@ -213,6 +213,33 @@ async def test_live_append_while_maximized_preserves_the_selection(make_app, sam
         assert "Living Room Lamp" in str(detail.content)
 
 
+async def test_live_append_while_maximized_follows_the_newest_selection(make_app, sample_entities):
+    app = make_app(entities=sample_entities, config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = [
+            {"when": "2024-01-15T10:30:00+00:00", "name": "Living Room Lamp", "state": "on"},
+            {"when": "2024-01-15T10:31:00+00:00", "name": "Kitchen Light", "state": "off"},
+        ]
+        await pilot.press("a")
+        await pilot.press("f")  # lands the highlight on the newest entry
+        await pilot.pause()
+
+        panel = app.query_one("#activity_log_panel", ActivityLogPanel)
+        options = panel.query_one("#log_options", OptionList)
+        assert options.highlighted == 1
+
+        app.client.inject_logbook_event(
+            [{"when": "2024-01-15T10:32:00+00:00", "name": "Fan Switch", "state": "on"}]
+        )
+        await pilot.pause()
+
+        assert options.option_count == 3
+        assert options.highlighted == 2  # followed to the new newest entry
+        detail = panel.query_one("#log_detail", Static)
+        assert "Fan Switch" in str(detail.content)
+
+
 async def test_reopening_log_is_not_maximized(make_app):
     app = make_app(config_data=NO_LIST_CONFIG)
     async with app.run_test() as pilot:

tests/test_log_scroll.py

@@ -0,0 +1,190 @@
+# hatty — MIT License. See LICENSE file for details.
+"""The docked activity log ticker stays pinned to its newest entry unless
+the reader has scrolled away from it (issue #44) — reloads, live appends and
+reflows all respect that rule."""
+
+from textual.widgets import Log
+
+from hatty.ui.activity_log_panel import ActivityLogPanel
+from tests.conftest import NO_LIST_CONFIG
+
+# 60 entries comfortably overflows the ~22-row docked log body at the
+# default 80x24 test terminal size, so max_scroll_y > 0.
+_MANY = 60
+
+
+def _entries(n: int, *, start_minute: int = 0) -> list[dict]:
+    return [
+        {
+            "when": f"2024-01-15T10:{(start_minute + i) % 60:02d}:00+00:00",
+            "name": f"Entity {i}",
+            "state": "on",
+        }
+        for i in range(n)
+    ]
+
+
+def _log_widget(app) -> Log:
+    panel = app.query_one("#activity_log_panel", ActivityLogPanel)
+    return panel.query_one("#log_widget", Log)
+
+
+async def test_long_history_opens_pinned_to_the_newest_entry(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        log = _log_widget(app)
+        assert log.max_scroll_y > 0
+        assert log.scroll_offset.y == log.max_scroll_y
+
+
+async def test_live_append_follows_when_pinned_to_the_newest(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        log = _log_widget(app)
+        count_before = log.line_count
+        app.client.inject_logbook_event(
+            [{"when": "2024-01-15T11:30:00+00:00", "name": "New Entity", "state": "on"}]
+        )
+        await pilot.pause()
+
+        assert log.line_count == count_before + 1
+        assert log.scroll_offset.y == log.max_scroll_y
+
+
+async def test_live_append_leaves_a_scrolled_up_reader_alone(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        log = _log_widget(app)
+        log.scroll_to(y=0, animate=False, immediate=True)
+        await pilot.pause()
+        assert log.scroll_offset.y == 0
+        count_before = log.line_count
+
+        app.client.inject_logbook_event(
+            [{"when": "2024-01-15T11:30:00+00:00", "name": "New Entity", "state": "on"}]
+        )
+        await pilot.pause()
+
+        assert log.line_count == count_before + 1
+        assert log.scroll_offset.y == 0
+
+
+async def test_a_longer_reload_repins_to_the_newest(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(40)
+        await pilot.press("a")
+        await pilot.pause()
+
+        app.client._logbook_data = _entries(120)
+        await pilot.press("left")  # pages back, forcing a refetch/reload
+        await pilot.pause()
+
+        log = _log_widget(app)
+        assert log.max_scroll_y > 0
+        assert log.scroll_offset.y == log.max_scroll_y
+
+
+async def test_scrolled_up_reader_is_repinned_by_a_reload_and_then_follows_again(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        log = _log_widget(app)
+        log.scroll_to(y=0, animate=False, immediate=True)
+        await pilot.pause()
+        assert log.scroll_offset.y == 0
+
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("left")  # any reload starts a reader fresh, pinned to newest
+        await pilot.pause()
+        assert log.scroll_offset.y == log.max_scroll_y
+
+        app.client.inject_logbook_event(
+            [{"when": "2024-01-15T11:30:00+00:00", "name": "New Entity", "state": "on"}]
+        )
+        await pilot.pause()
+        assert log.scroll_offset.y == log.max_scroll_y
+
+
+async def test_empty_history_is_visible_after_a_scrolled_up_reload(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        log = _log_widget(app)
+        log.scroll_to(y=0, animate=False, immediate=True)
+        await pilot.pause()
+
+        app.client._logbook_data = []
+        await pilot.press("left")
+        await pilot.pause()
+
+        assert log.scroll_offset.y == 0
+        assert "(no history available)" in log.lines[0]
+
+
+async def test_reflow_keeps_a_pinned_reader_pinned(make_app):
+    """_reflow_lines only ever runs when the log's rendered width has
+    actually changed (unreachable via a docked, fixed-width panel in the
+    headless test driver), so this drives it directly rather than through a
+    terminal resize — see the plan's investigation notes."""
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        panel = app.query_one("#activity_log_panel", ActivityLogPanel)
+        log = _log_widget(app)
+        assert log.scroll_offset.y == log.max_scroll_y
+
+        panel._rendered_width = 0  # force _reflow_lines to treat width as changed
+        panel._reflow_lines()
+        await pilot.pause()
+
+        assert log.scroll_offset.y == log.max_scroll_y
+
+
+async def test_reflow_preserves_a_scrolled_up_position(make_app):
+    app = make_app(config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.client._logbook_data = _entries(_MANY)
+        await pilot.press("a")
+        await pilot.pause()
+
+        panel = app.query_one("#activity_log_panel", ActivityLogPanel)
+        log = _log_widget(app)
+        log.scroll_to(y=3, animate=False, immediate=True)
+        await pilot.pause()
+        assert log.scroll_offset.y == 3
+
+        panel._rendered_width = 0
+        panel._reflow_lines()
+        await pilot.pause()
+
+        assert log.scroll_offset.y == 3