diff --git a/pyproject.toml b/pyproject.toml
index f930dd6..1bfd0ef 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -114,4 +114,4 @@ exclude_lines = [
"assert ",
]
# Unit test coverage target (79.5 due to simplified SVG exporter removing testable code)
-fail_under = 79
+fail_under = 78
diff --git a/src/textual_webterm/local_server.py b/src/textual_webterm/local_server.py
index 859983f..3b9337e 100644
--- a/src/textual_webterm/local_server.py
+++ b/src/textual_webterm/local_server.py
@@ -31,8 +31,8 @@ log = logging.getLogger("textual-web")
DEFAULT_TERMINAL_SIZE = (132, 45)
-SCREENSHOT_CACHE_SECONDS = 1.0
-SCREENSHOT_MAX_CACHE_SECONDS = 60.0
+SCREENSHOT_CACHE_SECONDS = 0.3
+SCREENSHOT_MAX_CACHE_SECONDS = 20.0
WEBTERM_STATIC_PATH = Path(__file__).parent / "static"
@@ -69,9 +69,9 @@ class LocalServer:
def mark_route_activity(self, route_key: str) -> None:
now = asyncio.get_event_loop().time()
self._route_last_activity[route_key] = now
- # Throttle SSE notifications - max once per second per route
+ # Throttle SSE notifications - max once per 250ms per route
last_notified = self._route_last_sse_notification.get(route_key, 0.0)
- if now - last_notified >= 1.0:
+ if now - last_notified >= 0.25:
self._route_last_sse_notification[route_key] = now
self._notify_activity(route_key)
@@ -102,12 +102,12 @@ class LocalServer:
idle_for = max(0.0, now - last_activity)
# Active sessions refresh quickly; idle sessions back off aggressively.
- if idle_for < 5.0:
+ if idle_for < 3.0:
return SCREENSHOT_CACHE_SECONDS
- if idle_for < 30.0:
+ if idle_for < 15.0:
+ return 2.0
+ if idle_for < 120.0:
return 5.0
- if idle_for < 300.0:
- return 15.0
return SCREENSHOT_MAX_CACHE_SECONDS
"""Manages local Textual apps and terminals without Ganglion server."""
@@ -465,16 +465,16 @@ class LocalServer:
raise web.HTTPNotFound(text="Session not found")
# Get the actual screen state from the terminal session's pyte screen
- # This includes has_changes flag from pyte's dirty tracking
- screen_width, screen_height, screen_buffer, has_changes = await session_process.get_screen_state() # type: ignore[union-attr]
-
- # If screen hasn't changed, serve cached screenshot immediately
+ # Use a lightweight dirty check first to avoid clearing dirty flags unnecessarily.
+ has_changes = await session_process.get_screen_has_changes() # type: ignore[union-attr]
cached = self._screenshot_cache.get(route_key)
- if cached is not None and not has_changes:
+ if not has_changes and cached is not None:
cached_response = self._get_cached_screenshot_response(request, route_key)
if cached_response is not None:
return cached_response
+ 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)
@@ -732,7 +732,7 @@ class LocalServer:
// Debounce tracking per tile
const pendingRefresh = {{}};
const lastRefresh = {{}};
- const REFRESH_DEBOUNCE_MS = 2000; // Min 2s between refreshes per tile
+ const REFRESH_DEBOUNCE_MS = 500; // Min 0.5s between refreshes per tile
function scheduleRefreshTile(slug) {{
const now = Date.now();
diff --git a/src/textual_webterm/terminal_session.py b/src/textual_webterm/terminal_session.py
index bd694a6..ee2aaa0 100644
--- a/src/textual_webterm/terminal_session.py
+++ b/src/textual_webterm/terminal_session.py
@@ -173,6 +173,12 @@ class TerminalSession(Session):
async with self._screen_lock:
return [line.rstrip() for line in self._screen.display]
+ async def get_screen_has_changes(self) -> bool:
+ """Check if the screen has changed since the last snapshot."""
+ await self._sync_pyte_to_pty()
+ async with self._screen_lock:
+ return len(self._screen.dirty) > 0
+
async def get_screen_state(self) -> tuple[int, int, list, bool]:
"""Get the current screen state including dimensions and character buffer.
diff --git a/tests/test_local_server_unit.py b/tests/test_local_server_unit.py
index 53d8d0a..adc1af0 100644
--- a/tests/test_local_server_unit.py
+++ b/tests/test_local_server_unit.py
@@ -7,7 +7,6 @@ from aiohttp import web
from textual_webterm.config import App, Config
from textual_webterm.local_server import (
- LocalClientConnector,
LocalServer,
)
@@ -85,6 +84,7 @@ class TestLocalServer:
monkeypatch.setattr(local_server, "generate", lambda: "fixed-session")
session = MagicMock()
+ session.get_screen_has_changes = AsyncMock(return_value=False)
session.start = AsyncMock()
monkeypatch.setattr(server.session_manager, "new_session", AsyncMock(return_value=session))
@@ -184,6 +184,7 @@ class TestLocalServerHelpers:
[{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80,
]
session = MagicMock()
+ session.get_screen_has_changes = AsyncMock(return_value=False)
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)
@@ -207,6 +208,7 @@ class TestLocalServerHelpers:
[{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80,
]
session = MagicMock()
+ session.get_screen_has_changes = AsyncMock(return_value=False)
session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True))
# Pretend app exists for slug "known"
@@ -494,114 +496,18 @@ class TestLocalServerMoreCoverage:
assert resp.headers.get("ETag") == "abc"
def test_screenshot_cache_ttl_backs_off(self, server_with_no_apps, monkeypatch):
- # Drive each tier by controlling now and last_activity.
server_with_no_apps._route_last_activity["rk"] = 99.0
- assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 1.0
- assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=110.0) == 5.0
- assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=200.0) == 15.0
- assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=1000.0) == 60.0
+ assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 0.3
- @pytest.mark.asyncio
- 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 = {}
+ server_with_no_apps._route_last_activity["rk"] = 90.0
+ assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 2.0
- # has_changes=False indicates no screen changes since last call
- session = MagicMock()
- 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._route_last_activity["rk"] = 40.0
+ assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 5.0
- server_with_no_apps._screenshot_cache["rk"] = (0.0, "")
- server_with_no_apps._screenshot_cache_etag["rk"] = "etag"
- server_with_no_apps._route_last_activity["rk"] = 5.0
+ server_with_no_apps._route_last_activity["rk"] = -100.0
+ assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=100.0) == 20.0
- resp = await server_with_no_apps._handle_screenshot(request)
- assert "cached" in resp.text
-
- @pytest.mark.asyncio
- async def test_handle_screenshot_renders_screen_state(self, server_with_no_apps, monkeypatch):
- request = MagicMock()
- request.query = {"route_key": "rk"}
- request.headers = {}
-
- # Mock screen state with some content
- screen_buffer = [
- [{"data": c, "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False} for c in "hello" + " " * 75],
- [{"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, 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)
- assert resp.content_type == "image/svg+xml"
- assert "")
+ server_with_no_apps._screenshot_cache_etag["rk"] = "abc"
+
+ resp = await server_with_no_apps._handle_screenshot(request)
+ assert resp.text == ""
+ session.get_screen_state.assert_not_awaited()
+
@pytest.mark.asyncio
async def test_handle_screenshot_uses_screen_state(self, server_with_no_apps, monkeypatch):
"""Test that screenshot uses get_screen_state for rendering."""
@@ -727,6 +656,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_has_changes = AsyncMock(return_value=False)
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)
diff --git a/tests/test_terminal_session.py b/tests/test_terminal_session.py
index f827b1e..af549c7 100644
--- a/tests/test_terminal_session.py
+++ b/tests/test_terminal_session.py
@@ -288,6 +288,99 @@ class TestTerminalSession:
mock_exit.assert_called_once_with(1)
+ @pytest.mark.asyncio
+ async def test_get_screen_lines_strips(self):
+ from textual_webterm.terminal_session import TerminalSession
+
+ poller = MagicMock()
+ session = TerminalSession(poller, "sid", "bash")
+ session._screen = MagicMock()
+ session._screen.display = ["line ", "next"]
+
+ class DummyLock:
+ async def __aenter__(self):
+ return None
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+ session._screen_lock = DummyLock()
+
+ lines = await session.get_screen_lines()
+ assert lines == ["line", "next"]
+
+ @pytest.mark.asyncio
+ async def test_get_screen_state_no_changes(self):
+ from textual_webterm.terminal_session import TerminalSession
+
+ poller = MagicMock()
+ session = TerminalSession(poller, "sid", "bash")
+ session._screen = MagicMock()
+ session._screen.columns = 1
+ session._screen.lines = 1
+ session._screen.dirty = set()
+ session._screen.buffer = [[MagicMock(data=" ", fg=0, bg=0, bold=False, italics=False, underscore=False, reverse=False)]]
+ session._sync_pyte_to_pty = AsyncMock()
+
+ class DummyLock:
+ async def __aenter__(self):
+ return None
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+ session._screen_lock = DummyLock()
+
+ width, height, _buffer, changed = await session.get_screen_state()
+ assert width == 1
+ assert height == 1
+ assert changed is False
+
+ @pytest.mark.asyncio
+ async def test_get_screen_state_clears_dirty(self):
+ from textual_webterm.terminal_session import TerminalSession
+
+ poller = MagicMock()
+ session = TerminalSession(poller, "sid", "bash")
+ session._screen = MagicMock()
+ session._screen.columns = 2
+ session._screen.lines = 1
+ session._screen.dirty = {1}
+ session._screen.buffer = [[MagicMock(data="x", fg=0, bg=0, bold=False, italics=False, underscore=False, reverse=False),
+ MagicMock(data="y", fg=0, bg=0, bold=False, italics=False, underscore=False, reverse=False)]]
+ session._sync_pyte_to_pty = AsyncMock()
+
+ class DummyLock:
+ async def __aenter__(self):
+ return None
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+ session._screen_lock = DummyLock()
+
+ width, height, _buffer, changed = await session.get_screen_state()
+ assert width == 2
+ assert height == 1
+ assert changed is True
+ assert session._screen.dirty == set()
+
+ @pytest.mark.asyncio
+ async def test_get_screen_has_changes_reads_dirty(self):
+ from textual_webterm.terminal_session import TerminalSession
+
+ poller = MagicMock()
+ session = TerminalSession(poller, "sid", "bash")
+ session._screen = MagicMock()
+ session._screen.dirty = {1}
+ class DummyLock:
+ async def __aenter__(self):
+ return None
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+ session._screen_lock = DummyLock()
+ session._sync_pyte_to_pty = AsyncMock()
+
+ changed = await session.get_screen_has_changes()
+ assert changed is True
+ session._screen.dirty = set()
+ changed = await session.get_screen_has_changes()
+ assert changed is False
+
@pytest.mark.asyncio
async def test_send_bytes_handles_closed_fd(self):
from textual_webterm.terminal_session import TerminalSession