Optimize screenshot updates using pyte dirty tracking

- get_screen_state() now returns has_changes flag indicating if screen changed
- pyte's dirty set tracks which rows have been modified since last read
- Screenshot handler returns cached SVG immediately when no changes detected
- Removed _screenshot_last_rendered_activity tracking (replaced by dirty flag)
- Added test for dirty flag behavior

Bump version to 0.1.16

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "textual-webterm"
-version = "0.1.15"
+version = "0.1.16"
 description = "Serve terminal sessions over the web"
 authors = ["Will McGugan <will@textualize.io>"]
 license = "MIT"

src/textual_webterm/local_server.py

@@ -211,7 +211,6 @@ class LocalServer:
         self._screenshot_cache_etag: dict[str, str] = {}
         self._screenshot_locks: dict[str, asyncio.Lock] = {}
         self._route_last_activity: dict[str, float] = {}
-        self._screenshot_last_rendered_activity: dict[str, float] = {}
 
     @property
     def app_count(self) -> int:
@@ -516,30 +515,21 @@ class LocalServer:
         if session_process is None or not hasattr(session_process, "get_screen_state"):
             raise web.HTTPNotFound(text="Session not found")
 
-        # If nothing has changed since the last render, serve cached screenshot without
+        # Get the actual screen state from the terminal session's pyte screen
-        # touching the session replay buffer.
+        # This includes has_changes flag from pyte's dirty tracking
-        last_activity = self._route_last_activity.get(route_key, 0.0)
+        screen_width, screen_height, screen_buffer, has_changes = await session_process.get_screen_state()  # type: ignore[union-attr]
-        last_rendered_activity = self._screenshot_last_rendered_activity.get(route_key, -1.0)
+
-        if last_activity <= last_rendered_activity:
+        # If screen hasn't changed, serve cached screenshot immediately
+        cached = self._screenshot_cache.get(route_key)
+        if cached is not None and not has_changes:
             cached_response = self._get_cached_screenshot_response(request, route_key)
             if cached_response is not None:
                 return cached_response
 
-        # Get the actual screen state from the terminal session's pyte screen
-        # This uses the correct dimensions that match what the terminal is rendering
-        screen_width, screen_height, screen_buffer = await session_process.get_screen_state()  # type: ignore[union-attr]
-
         now = asyncio.get_event_loop().time()
         ttl = self._get_screenshot_cache_ttl(route_key, now)
-        cached = self._screenshot_cache.get(route_key)
-
-        # If we have a cached screenshot and the session is idle, keep serving it until
-        # new activity occurs (no periodic re-render).
-        if cached is not None and self._route_last_activity.get(route_key, 0.0) == 0.0:
-            cached_response = self._get_cached_screenshot_response(request, route_key)
-            if cached_response is not None:
-                return cached_response
 
+        # Also check time-based cache for recently rendered screenshots
         if cached is not None and (now - cached[0]) < ttl:
             cached_response = self._get_cached_screenshot_response(request, route_key)
             if cached_response is not None:
@@ -620,9 +610,6 @@ class LocalServer:
             etag = hashlib.sha1(svg.encode("utf-8"), usedforsecurity=False).hexdigest()
             self._screenshot_cache[route_key] = (asyncio.get_event_loop().time(), svg)
             self._screenshot_cache_etag[route_key] = etag
-            self._screenshot_last_rendered_activity[route_key] = self._route_last_activity.get(
-                route_key, 0.0
-            )
             headers = {"Cache-Control": "no-cache", "ETag": etag}
             return web.Response(text=svg, content_type="image/svg+xml", headers=headers)
 

src/textual_webterm/terminal_session.py

@@ -140,16 +140,23 @@ class TerminalSession(Session):
         async with self._screen_lock:
             return [line.rstrip() for line in self._screen.display]
 
-    async def get_screen_state(self) -> tuple[int, int, list]:
+    async def get_screen_state(self) -> tuple[int, int, list, bool]:
         """Get the current screen state including dimensions and character buffer.
 
         Returns:
-            Tuple of (width, height, buffer) where buffer is a list of rows,
+            Tuple of (width, height, buffer, has_changes) where:
-            each row containing character data with styling attributes.
+            - width: screen width in columns
+            - height: screen height in rows
+            - buffer: list of rows, each containing character data with styling
+            - has_changes: True if screen has changed since last call
         """
         async with self._screen_lock:
             width = self._screen.columns
             height = self._screen.lines
+            # Check if any rows are dirty (changed since last clear)
+            has_changes = len(self._screen.dirty) > 0
+            # Clear dirty set after checking
+            self._screen.dirty.clear()
             # Copy the buffer data to avoid holding the lock
             buffer = []
             for row in range(height):
@@ -166,7 +173,7 @@ class TerminalSession(Session):
                         "reverse": char.reverse,
                     })
                 buffer.append(row_data)
-            return (width, height, buffer)
+            return (width, height, buffer, has_changes)
 
     def update_connector(self, connector: SessionConnector) -> None:
         """Update the connector for reconnection without restarting the session."""

tests/test_local_server_unit.py

@@ -199,7 +199,7 @@ class TestLocalServerHelpers:
             [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80,
         ]
         session = MagicMock()
-        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer))
+        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
 
         monkeypatch.setattr(server.session_manager, "get_session_by_route_key", lambda _rk: session)
 
@@ -222,7 +222,7 @@ class TestLocalServerHelpers:
             [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80,
         ]
         session = MagicMock()
-        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer))
+        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
 
         # Pretend app exists for slug "known"
         server.session_manager.apps_by_slug["known"] = App(
@@ -548,23 +548,23 @@ class TestLocalServerMoreCoverage:
         assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=1000.0) == 60.0
 
     @pytest.mark.asyncio
-    async def test_handle_screenshot_uses_cache_when_no_new_activity(self, server_with_no_apps, monkeypatch):
+    async def test_handle_screenshot_uses_cache_when_no_changes(self, server_with_no_apps, monkeypatch):
+        """Test that cached screenshot is returned when pyte reports no changes."""
         request = MagicMock()
         request.query = {"route_key": "rk"}
         request.headers = {}
 
+        # has_changes=False indicates no screen changes since last call
         session = MagicMock()
-        session.get_screen_state = AsyncMock(return_value=(80, 2, []))
+        session.get_screen_state = AsyncMock(return_value=(80, 2, [], False))
         monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
 
         server_with_no_apps._screenshot_cache["rk"] = (0.0, "<svg>cached</svg>")
         server_with_no_apps._screenshot_cache_etag["rk"] = "etag"
         server_with_no_apps._route_last_activity["rk"] = 5.0
-        server_with_no_apps._screenshot_last_rendered_activity["rk"] = 5.0
 
         resp = await server_with_no_apps._handle_screenshot(request)
         assert "cached" in resp.text
-        session.get_screen_state.assert_not_awaited()
 
     @pytest.mark.asyncio
     async def test_handle_screenshot_renders_screen_state(self, server_with_no_apps, monkeypatch):
@@ -578,7 +578,7 @@ class TestLocalServerMoreCoverage:
             [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80,
         ]
         session = MagicMock()
-        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer))
+        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
         monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
 
         resp = await server_with_no_apps._handle_screenshot(request)
@@ -760,7 +760,7 @@ class TestLocalServerMoreCoverage:
             [{"data": c, "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False} for c in "line2" + " " * 75],
         ]
         session = MagicMock()
-        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer))
+        session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
         monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
 
         server_with_no_apps._route_last_activity["rk"] = 1.0

tests/test_terminal_session.py

@@ -130,6 +130,32 @@ class TestTerminalSession:
         assert lines[1] == "Updated Line 2"
         assert lines[2] == "Line 3"
 
+    @pytest.mark.asyncio
+    async def test_get_screen_state_returns_dirty_flag(self):
+        """Test that get_screen_state returns has_changes flag based on pyte dirty tracking."""
+        from textual_webterm.terminal_session import TerminalSession
+
+        mock_poller = MagicMock()
+        session = TerminalSession(mock_poller, "test-session", "bash")
+
+        # After creation, all rows are dirty (initialized)
+        _w, _h, _buf, has_changes = await session.get_screen_state()
+        assert has_changes is True  # Initial state marks all rows dirty
+
+        # After getting state, dirty set is cleared
+        # Without new data, has_changes should be False
+        _, _, _, has_changes = await session.get_screen_state()
+        assert has_changes is False  # No changes since last call
+
+        # Feed new data
+        await session._update_screen(b"New content\r\n")
+        _, _, _, has_changes = await session.get_screen_state()
+        assert has_changes is True  # Screen was updated
+
+        # Check again without new data
+        _, _, _, has_changes = await session.get_screen_state()
+        assert has_changes is False  # No changes
+
     def test_update_connector(self):
         """Test updating connector."""
         from textual_webterm.terminal_session import TerminalSession