Extract HTML5 SSE helpers

services/api-server/src/api_server/html5_sse.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+
+def html5_sse_event(event: str, fragment: str) -> str:
+    data = "\n".join(f"data: {line}" for line in fragment.splitlines())
+    return f"event: {event}\nretry: 5000\n{data}\n\n"
+
+
+def html5_sse_if_changed(last_fragments: dict[str, str], event: str, fragment: str) -> Iterator[str]:
+    if last_fragments.get(event) == fragment:
+        return
+    last_fragments[event] = fragment
+    yield html5_sse_event(event, fragment)
+
+
+def html5_sse_comment(message: str) -> str:
+    return f": {message}\n\n"

services/api-server/src/api_server/main.py

@@ -42,6 +42,10 @@ from api_server.html5_responses import (
     html5_response as _html5_response,
     html5_sse_response as _html5_sse_response,
 )
+from api_server.html5_sse import (
+    html5_sse_comment as _html5_sse_comment,
+    html5_sse_if_changed as _html5_sse_if_changed,
+)
 from api_server.html5_inspector import (
     render_html5_flowchart,
     render_html5_object_context,
@@ -8425,22 +8429,6 @@ def _operation_value(value: object) -> str:
     return str(getattr(value, "value", value))
 
 
-def _html5_sse_event(event: str, fragment: str) -> str:
-    data = "\n".join(f"data: {line}" for line in fragment.splitlines())
-    return f"event: {event}\nretry: 5000\n{data}\n\n"
-
-
-def _html5_sse_if_changed(last_fragments: dict[str, str], event: str, fragment: str):
-    if last_fragments.get(event) == fragment:
-        return
-    last_fragments[event] = fragment
-    yield _html5_sse_event(event, fragment)
-
-
-def _html5_sse_comment(message: str) -> str:
-    return f": {message}\n\n"
-
-
 def _project_summaries() -> list[ProjectSummaryResponse]:
     project_ids = set(_project_setup.keys())
     stored_snapshots = _storage.list_snapshot_refs()

services/api-server/tests/test_api.py

@@ -7,6 +7,7 @@ import zipfile
 from fastapi.testclient import TestClient
 
 from api_server import main
+from api_server.html5_sse import html5_sse_comment, html5_sse_event, html5_sse_if_changed
 from api_server.main import app
 from one_c_normalizer import ConfigurationRoot, MetadataGroup, MetadataObject, Module, NormalizedProject
 
@@ -51,6 +52,26 @@ def assert_html5_response_contract(response, *markers: str, full_page: bool = Fa
     assert_html5_contract(response.text, *markers, full_page=full_page)
 
 
+def test_html5_sse_formatters_emit_stable_event_stream_chunks():
+    assert html5_sse_comment("heartbeat") == ": heartbeat\n\n"
+    assert html5_sse_event("status", "<div>\nready</div>") == (
+        "event: status\n"
+        "retry: 5000\n"
+        "data: <div>\n"
+        "data: ready</div>\n\n"
+    )
+
+    last_fragments: dict[str, str] = {}
+    first = list(html5_sse_if_changed(last_fragments, "status", "<div>ready</div>"))
+    second = list(html5_sse_if_changed(last_fragments, "status", "<div>ready</div>"))
+    third = list(html5_sse_if_changed(last_fragments, "status", "<div>done</div>"))
+
+    assert len(first) == 1
+    assert second == []
+    assert len(third) == 1
+    assert "data: <div>done</div>" in third[0]
+
+
 def test_cors_allows_lan_panel_origin():
     client = TestClient(app)
     response = client.options(