Use SSE for real-time screenshot updates

- New /events SSE endpoint pushes activity notifications to browsers
- Dashboard subscribes to SSE stream instead of polling
- Screenshots refresh instantly when terminal activity occurs
- Sparklines still poll every 30s (appropriate for 30min history)
- SSE includes keepalive every 30s and auto-reconnect on error
- Removes inefficient 5s polling; updates only on actual changes
This commit is contained in:
GitHub Copilot
2026-01-24 11:44:29 +00:00
parent 1ba4ce2a34
commit bd477c1b3c
2 changed files with 148 additions and 19 deletions
+39
View File
@@ -769,3 +769,42 @@ class TestLocalServerMoreCoverage:
assert resp.content_type == "image/svg+xml"
assert "<svg" in resp.text
session.get_screen_state.assert_awaited_once()
def test_notify_activity_pushes_to_subscribers(self, server_with_no_apps):
"""Test that activity notifications are pushed to SSE subscribers."""
import asyncio
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=10)
server_with_no_apps._sse_subscribers.append(queue)
server_with_no_apps._notify_activity("test-route")
assert not queue.empty()
assert queue.get_nowait() == "test-route"
def test_notify_activity_handles_full_queue(self, server_with_no_apps):
"""Test that full queues don't cause errors."""
import asyncio
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=1)
queue.put_nowait("existing")
server_with_no_apps._sse_subscribers.append(queue)
# Should not raise even though queue is full
server_with_no_apps._notify_activity("test-route")
# Only the original item should be there
assert queue.get_nowait() == "existing"
@pytest.mark.asyncio
async def test_mark_route_activity_triggers_notification(self, server_with_no_apps):
"""Test that mark_route_activity triggers SSE notification."""
import asyncio
queue: asyncio.Queue[str] = asyncio.Queue(maxsize=10)
server_with_no_apps._sse_subscribers.append(queue)
server_with_no_apps.mark_route_activity("my-route")
assert not queue.empty()
assert queue.get_nowait() == "my-route"