Harden TUI status refresh resilience

Author: hrabanazviking

mythic_vibe_cli/tui/app.py

@@ -773,6 +773,7 @@ def __init__(self, root: Path) -> None:
         # Warm-starts from any existing entries so the panel populates
         # immediately, but only counts truly-new events as "live".
         self._event_reader = EventTailReader(event_log_path_for(self.root))
+        self._refresh_timer: Any | None = None
 
     def compose(self) -> ComposeResult:
         yield Header(show_clock=True)
@@ -789,7 +790,14 @@ def compose(self) -> ComposeResult:
 
     def on_mount(self) -> None:
         self._refresh_panels()
-        self.set_interval(REFRESH_INTERVAL_SECONDS, self._refresh_panels)
+        self._refresh_timer = self.set_interval(REFRESH_INTERVAL_SECONDS, self._refresh_panels)
+
+    def on_unmount(self) -> None:
+        timer = self._refresh_timer
+        self._refresh_timer = None
+        stop = getattr(timer, "stop", None)
+        if callable(stop):
+            stop()
 
     def action_refresh_now(self) -> None:
         self._refresh_panels()
@@ -812,24 +820,58 @@ def action_open_drift(self) -> None:
         self.app.push_screen(DriftScreen(self.root))
 
     def _refresh_panels(self) -> None:
-        data = build_status_data(self.root)
-        loop_nav_data = build_loop_navigator_data(self.root)
-        artifact_data = build_artifact_viewer_data(self.root, loop_nav_data.current_phase)
-        packet_data = build_packet_viewer_data(self.root)
-        diagnostics = self._event_reader.poll()
+        try:
+            data = build_status_data(self.root)
+            status_text = _format_status_bar(data)
+            footer_text = _format_footer_line(data)
+        except Exception as exc:  # noqa: BLE001 - TUI must degrade per panel
+            status_text = _format_panel_error("Status", exc)
+            footer_text = "Last refresh: unavailable"
+
+        try:
+            loop_nav_data = build_loop_navigator_data(self.root)
+            loop_text = _format_loop_navigator(loop_nav_data)
+            current_phase = loop_nav_data.current_phase
+        except Exception as exc:  # noqa: BLE001 - TUI must degrade per panel
+            loop_text = _format_panel_error("Loop", exc)
+            current_phase = ""
+
+        try:
+            artifact_data = build_artifact_viewer_data(self.root, current_phase)
+            artifact_text = _format_artifact_viewer(artifact_data)
+            artifact_phase = artifact_data.phase or "(none)"
+        except Exception as exc:  # noqa: BLE001 - TUI must degrade per panel
+            artifact_text = _format_panel_error("Artefacts", exc)
+            artifact_phase = current_phase or "(none)"
+
+        try:
+            packet_data = build_packet_viewer_data(self.root)
+            packet_text = _format_packet_viewer(packet_data)
+            packet_title = f"Packet ({packet_data.packet_id})" if packet_data.packet_id else "Packet"
+        except Exception as exc:  # noqa: BLE001 - TUI must degrade per panel
+            packet_text = _format_panel_error("Packet", exc)
+            packet_title = "Packet"
+
+        try:
+            diagnostics = self._event_reader.poll()
+            diagnostics_text = _format_diagnostics_panel(diagnostics)
+        except Exception as exc:  # noqa: BLE001 - TUI must degrade per panel
+            diagnostics_text = _format_panel_error("Diagnostics", exc)
+
         self._loop_nav_widget.border_title = "Loop"
         self._events_widget.border_title = "Diagnostics"
-        artifact_phase = artifact_data.phase or "(none)"
         self._artifact_widget.border_title = f"Artefacts ({artifact_phase})"
-        self._packet_widget.border_title = (
-            f"Packet ({packet_data.packet_id})" if packet_data.packet_id else "Packet"
-        )
-        self._loop_nav_widget.update(_format_loop_navigator(loop_nav_data))
-        self._events_widget.update(_format_diagnostics_panel(diagnostics))
-        self._artifact_widget.update(_format_artifact_viewer(artifact_data))
-        self._packet_widget.update(_format_packet_viewer(packet_data))
-        self._status_bar_widget.update(_format_status_bar(data))
-        self._footer_widget.update(_format_footer_line(data))
+        self._packet_widget.border_title = packet_title
+        self._loop_nav_widget.update(loop_text)
+        self._events_widget.update(diagnostics_text)
+        self._artifact_widget.update(artifact_text)
+        self._packet_widget.update(packet_text)
+        self._status_bar_widget.update(status_text)
+        self._footer_widget.update(footer_text)
+
+
+def _format_panel_error(label: str, exc: BaseException) -> str:
+    return f"[yellow]{label} unavailable[/yellow]\n[dim]{type(exc).__name__}: {exc}[/dim]"
 
 
 class MythicTuiApp(App):

tests/test_tui.py

@@ -16,6 +16,7 @@
 import time
 import unittest
 from pathlib import Path
+from unittest import mock
 
 
 textual_unavailable = False
@@ -264,6 +265,65 @@ async def run_test() -> None:
         # Should complete without raising.
         asyncio.run(run_test())
 
+    def test_status_screen_degrades_when_status_data_refresh_fails(self) -> None:
+        from mythic_vibe_cli.tui.app import MythicTuiApp
+
+        async def run_test() -> tuple[str, str]:
+            with tempfile.TemporaryDirectory() as tmp:
+                app = MythicTuiApp(Path(tmp))
+                with mock.patch(
+                    "mythic_vibe_cli.tui.app.build_status_data",
+                    side_effect=RuntimeError("state unavailable"),
+                ):
+                    async with app.run_test() as pilot:
+                        await pilot.pause()
+                        status_bar = app.screen.query_one("#status-bar")
+                        footer_widget = app.screen.query_one("#footer-line")
+                        return str(status_bar.render()), str(footer_widget.render())
+
+        rendered_bar, rendered_footer = asyncio.run(run_test())
+        self.assertIn("Status unavailable", rendered_bar)
+        self.assertIn("state unavailable", rendered_bar)
+        self.assertIn("Last refresh: unavailable", rendered_footer)
+
+    def test_status_screen_degrades_when_diagnostics_poll_fails(self) -> None:
+        from mythic_vibe_cli.tui.app import MythicTuiApp
+
+        async def run_test() -> str:
+            with tempfile.TemporaryDirectory() as tmp:
+                app = MythicTuiApp(Path(tmp))
+                with mock.patch(
+                    "mythic_vibe_cli.tui.app.EventTailReader.poll",
+                    side_effect=OSError("event log unreadable"),
+                ):
+                    async with app.run_test() as pilot:
+                        await pilot.pause()
+                        panel = app.screen.query_one("#events-panel")
+                        return str(panel.render())
+
+        rendered = asyncio.run(run_test())
+        self.assertIn("Diagnostics unavailable", rendered)
+        self.assertIn("event log unreadable", rendered)
+
+    def test_status_screen_stops_refresh_timer_on_unmount(self) -> None:
+        from mythic_vibe_cli.tui.app import StatusScreen
+
+        class FakeTimer:
+            def __init__(self) -> None:
+                self.stopped = False
+
+            def stop(self) -> None:
+                self.stopped = True
+
+        with tempfile.TemporaryDirectory() as tmp:
+            screen = StatusScreen(Path(tmp))
+            timer = FakeTimer()
+            screen._refresh_timer = timer
+            screen.on_unmount()
+
+        self.assertTrue(timer.stopped)
+        self.assertIsNone(screen._refresh_timer)
+
 
 @unittest.skipIf(textual_unavailable, "textual not installed")
 class SlashPickerTests(unittest.TestCase):