test(cdp): handle split byte counters and EventSource cancellation

Aggregate Fetch dataReceived byte counters after completion instead of
requiring both counters to be positive in the first event. Preserve event
ordering and reject invalid lengths or missing positive totals.

Accept either committed EOF or an explicit ERR_ABORTED cancellation after an
EventSource message handler closes its connection. Keep request correlation,
resource classification, event order, and exactly one terminal checked.

Report failed scenario details when a smoke worker exits without a top-level
traceback. Cover the observed Chromium traces and invalid alternatives with
unit regressions, and document the real Chromium/Moli calibration.

Validation: 64 Python harness tests; release build; all 52 smoke groups on
both pinned Chromium and Moli release; network and browser-semantics repeated
20 times per engine (40/40 worker runs each).

moli-cdp-smoke/README.md

@@ -81,6 +81,22 @@ The `network` and `multi-context` groups require the encoded Fetch body and flag
 then independently verify the original page-visible text after continuation.
 Their Network body and `IO.read` stream assertions retain their own contracts.
 
+The Fetch byte-counter and EventSource-close contracts were probed on 2026-10-20
+with Playwright 1.58.0's pinned Chrome for Testing 145.0.7632.6 and local HTTP
+fixtures. Of 150 ordinary 45-byte JSON fetches, one reported `dataReceived`
+counters `(45, 0)` followed by `(0, 45)`; 146 of 150 gzip responses split their
+decoded and encoded counters similarly. The `network` group therefore checks
+the totals over all data events, their nonnegative lengths, and their order
+between the response and successful terminal. A finite SSE response completed
+with `loadingFinished`, while a response with a withheld trailing chunk ended
+with `loadingFailed`, `net::ERR_ABORTED`, and `canceled: true` when the message
+handler called `close()`. The EventSource scenario accepts those two outcomes
+and still rejects other errors, wrong order, and duplicate terminals. The same
+probe against the Moli release produced both SSE terminals, with its Fetch byte
+counters delivered together. Run
+`--browser chromium --group network --group browser-semantics --repeat 20 --jobs 4`
+to exercise both groups with fresh browser processes.
+
 The default `cdp-ordering` group locks down Renderer native and Inspector wire
 publication order. It pipelines Page trees/layout, DOMSnapshot, DOM queries and
 objects, CSS, AX, and a native backend error before `Runtime.enable`; stylesheet

moli-cdp-smoke/moli_cdp_smoke/groups/browser_semantics.py

@@ -1358,7 +1358,7 @@ async def _event_source_runtime_and_network(state: SmokeState) -> dict[str, Any]
     async with _isolated_page(state) as (_context, page, cdp):
         events = attach_cdp_event_collector(
             cdp,
-            ["Network.requestWillBeSent", "Network.responseReceived", "Network.loadingFinished"],
+            ["Network.requestWillBeSent", "Network.responseReceived", "Network.loadingFinished", "Network.loadingFailed"],
         )
         await cdp.send("Network.enable")
         await page.goto(f"{state.fixture}/plain", wait_until="load", timeout=10_000)
@@ -1402,34 +1402,40 @@ async def _event_source_runtime_and_network(state: SmokeState) -> dict[str, Any]
         request_id = response_event["params"]["requestId"]
         await wait_until(
             lambda: any(
-                event["method"] == "Network.loadingFinished"
+                event["method"] in ("Network.loadingFinished", "Network.loadingFailed")
                 and event["params"].get("requestId") == request_id
                 for event in events
             ),
-            "EventSource Network.loadingFinished",
-        )
-        event_methods = [
-            event["method"]
-            for event in events
-            if event["params"].get("requestId") == request_id
-        ]
-        assert_equal(
-            event_methods,
-            [
-                "Network.requestWillBeSent",
-                "Network.responseReceived",
-                "Network.loadingFinished",
-            ],
-            "EventSource Network event order",
+            "EventSource Network terminal after close()",
         )
         return {
             "runtime": runtime_value,
             "requestId": request_id,
             "resourceType": response_event["params"]["type"],
-            "eventMethods": event_methods,
+            **_event_source_network_summary(events, request_id),
         }
 
 
+def _event_source_network_summary(events: list[dict[str, Any]], request_id: str) -> dict[str, Any]:
+    matching = [event for event in events if event["params"].get("requestId") == request_id]
+    methods = [event["method"] for event in matching]
+    terminal = matching[-1] if matching else {}
+    terminal_method = terminal.get("method")
+    if terminal_method not in ("Network.loadingFinished", "Network.loadingFailed"):
+        raise SmokeError(f"EventSource close() has no Network terminal: {matching}")
+    assert_equal(
+        methods,
+        ["Network.requestWillBeSent", "Network.responseReceived", terminal_method],
+        "EventSource Network event order",
+    )
+    # close() races the finite fixture's EOF. If EOF has not committed yet,
+    # Chromium reports the intentional cancellation instead of successful EOF.
+    if terminal_method == "Network.loadingFailed":
+        assert_equal(terminal["params"].get("errorText"), "net::ERR_ABORTED", "EventSource close() error")
+        assert_equal(terminal["params"].get("canceled"), True, "EventSource close() cancellation")
+    return {"eventMethods": methods, "terminal": terminal}
+
+
 async def _http_cache_event_and_reuse(state: SmokeState) -> dict[str, Any]:
     cache_route = "/semantic-cache.js"
     state.fixture_server.reset_request_count(cache_route)

moli-cdp-smoke/moli_cdp_smoke/groups/network.py

@@ -321,37 +321,42 @@ async def _verify_fetch_network_events(state: SmokeState) -> None:
     observed_fetch = await page.evaluate("async () => await fetch('/api-continue').then(response => response.json())")
     assert_equal(observed_fetch.get("routeHeader"), None, "unrouted fetch should reach fixture server")
 
-    def saw_fetch_events() -> bool:
-        events = subresource_events[fetch_network_start:]
-        request = _find_request(events, "Fetch", f"{fixture}/api-continue")
-        request_id = request and request["params"].get("requestId")
-        if not request_id:
-            return False
-        matching_indices = {
-            method: next(
-                (
-                    index
-                    for index, event in enumerate(events)
-                    if event["method"] == method and event["params"].get("requestId") == request_id
-                ),
-                None,
-            )
-            for method in ("Network.responseReceived", "Network.dataReceived", "Network.loadingFinished")
-        }
-        response_index = matching_indices["Network.responseReceived"]
-        data_index = matching_indices["Network.dataReceived"]
-        finished_index = matching_indices["Network.loadingFinished"]
-        if response_index is None or data_index is None or finished_index is None:
-            return False
-        data_params = events[data_index]["params"]
-        return bool(
-            response_index < data_index < finished_index
-            and data_params.get("dataLength", 0) > 0
-            and data_params.get("encodedDataLength", 0) > 0
-        )
+    await wait_until(
+        lambda: _completed_request_id(
+            subresource_events[fetch_network_start:], "Fetch", f"{fixture}/api-continue"
+        ) is not None,
+        "Fetch Network events",
+    )
+    events = subresource_events[fetch_network_start:]
+    request_id = _completed_request_id(events, "Fetch", f"{fixture}/api-continue")
+    if request_id is None:
+        raise SmokeError("completed Fetch request disappeared from the event trace")
+    state.record("fetch_network_events", {
+        "requestId": request_id,
+        **_fetch_network_event_summary(events, request_id),
+    })
 
-    await wait_until(saw_fetch_events, "Fetch Network events")
-    state.record("fetch_network_events")
+
+def _fetch_network_event_summary(events: list[dict[str, Any]], request_id: str) -> dict[str, Any]:
+    matching = [event for event in events if event["params"].get("requestId") == request_id]
+    methods = [event["method"] for event in matching]
+    assert_equal(methods[:2], ["Network.requestWillBeSent", "Network.responseReceived"], "Fetch Network start order")
+    assert_equal(methods[-1:], ["Network.loadingFinished"], "Fetch Network terminal")
+    data_events = matching[2:-1]
+    assert_equal({event["method"] for event in data_events}, {"Network.dataReceived"}, "Fetch Network data order")
+    totals = {"dataLength": 0, "encodedDataLength": 0}
+    for event in data_events:
+        for key in totals:
+            value = event["params"].get(key)
+            if type(value) is not int or value < 0:
+                raise SmokeError(f"invalid Fetch Network {key}: {event}")
+            totals[key] += value
+    # Chromium can deliver decoded bytes first, then flush encoded byte counts
+    # in a later dataReceived event whose dataLength is zero. Neither counter
+    # has to be positive in the first chunk; the completed nonempty body does.
+    if any(value <= 0 for value in totals.values()):
+        raise SmokeError(f"nonempty Fetch Network body must report positive byte totals: {matching}")
+    return {"eventMethods": methods, **totals}
 
 
 async def _verify_xhr_network_events(state: SmokeState) -> None:

moli-cdp-smoke/moli_cdp_smoke/supervisor.py

@@ -307,7 +307,9 @@ async def _run_worker_job(
     else:
         status = "failed"
         worker_error = payload.get("error")
-        error = str(worker_error) if worker_error else f"worker exited with {exit_code}"
+        error = str(worker_error) if worker_error else (
+            _coverage_error(job, payload) or f"worker exited with {exit_code}"
+        )
 
     outcome = WorkerOutcome(
         job=job,

moli-cdp-smoke/tests/test_network_events.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import unittest
+from typing import Any
+
+from moli_cdp_smoke.assertions import SmokeError
+from moli_cdp_smoke.groups.browser_semantics import _event_source_network_summary
+from moli_cdp_smoke.groups.network import _fetch_network_event_summary
+
+
+REQUEST = "request"
+
+
+def event(method: str, **params: Any) -> dict[str, Any]:
+    return {"method": f"Network.{method}", "params": {"requestId": REQUEST, **params}}
+
+
+def fetch_trace(chunks: list[tuple[int, int]]) -> list[dict[str, Any]]:
+    return [
+        event("requestWillBeSent"),
+        event("responseReceived"),
+        *(event("dataReceived", dataLength=decoded, encodedDataLength=encoded)
+          for decoded, encoded in chunks),
+        event("loadingFinished"),
+    ]
+
+
+class FetchNetworkEventsTests(unittest.TestCase):
+    def test_counters_can_arrive_together_or_in_separate_chunks(self) -> None:
+        # Observed on pinned Chromium 145.0.7632.6: a normal 45-byte JSON
+        # response can split into (45, 0), (0, 45). Gzip does this frequently.
+        for chunks, decoded, encoded in [
+            ([(45, 45)], 45, 45),
+            ([(45, 0), (0, 45)], 45, 45),
+            ([(45, 0), (0, 35)], 45, 35),
+            ([(1044, 0), (31737, 82)], 32781, 82),
+        ]:
+            with self.subTest(chunks=chunks):
+                summary = _fetch_network_event_summary(fetch_trace(chunks), REQUEST)
+                self.assertEqual(summary["dataLength"], decoded)
+                self.assertEqual(summary["encodedDataLength"], encoded)
+                self.assertEqual(summary["eventMethods"][-1], "Network.loadingFinished")
+
+    def test_other_requests_cannot_supply_missing_byte_counts(self) -> None:
+        events = fetch_trace([(45, 0)])
+        events.insert(-1, event("dataReceived", requestId="other", dataLength=0, encodedDataLength=45))
+        with self.assertRaisesRegex(SmokeError, "positive byte totals"):
+            _fetch_network_event_summary(events, REQUEST)
+
+    def test_nonempty_body_requires_both_positive_totals(self) -> None:
+        for chunks in [[(45, 0)], [(0, 45)], [(0, 0)], []]:
+            with self.subTest(chunks=chunks), self.assertRaises(SmokeError):
+                _fetch_network_event_summary(fetch_trace(chunks), REQUEST)
+
+    def test_invalid_chunk_lengths_are_rejected_before_summing(self) -> None:
+        for key in ["dataLength", "encodedDataLength"]:
+            for value in [-1, None, "45", True]:
+                events = fetch_trace([(45, 45), (1, 1)])
+                events[2]["params"][key] = value
+                with self.subTest(key=key, value=value), self.assertRaisesRegex(SmokeError, "invalid Fetch"):
+                    _fetch_network_event_summary(events, REQUEST)
+
+    def test_data_must_follow_response_and_precede_one_successful_terminal(self) -> None:
+        valid = fetch_trace([(45, 45)])
+        for events in [
+            [valid[0], valid[2], valid[1], valid[3]],
+            [valid[0], valid[1], valid[3], valid[2]],
+            [*valid, valid[-1]],
+            [*valid[:-1], event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True)],
+        ]:
+            with self.subTest(events=events), self.assertRaises(SmokeError):
+                _fetch_network_event_summary(events, REQUEST)
+
+
+class EventSourceNetworkEventsTests(unittest.TestCase):
+    def test_close_accepts_committed_eof_or_an_explicit_abort(self) -> None:
+        for terminal in [
+            event("loadingFinished"),
+            event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True),
+        ]:
+            events = [event("requestWillBeSent"), event("responseReceived"), terminal]
+            with self.subTest(terminal=terminal):
+                summary = _event_source_network_summary(events, REQUEST)
+                self.assertEqual(summary["terminal"], terminal)
+                self.assertEqual(summary["eventMethods"], [entry["method"] for entry in events])
+
+    def test_close_does_not_accept_unrelated_network_errors(self) -> None:
+        for params in [
+            {"errorText": "net::ERR_CONNECTION_RESET", "canceled": True},
+            {"errorText": "net::ERR_ABORTED", "canceled": False},
+            {"errorText": "net::ERR_ABORTED"},
+        ]:
+            events = [event("requestWillBeSent"), event("responseReceived"), event("loadingFailed", **params)]
+            with self.subTest(params=params), self.assertRaises(SmokeError):
+                _event_source_network_summary(events, REQUEST)
+
+    def test_close_requires_ordered_response_and_exactly_one_terminal(self) -> None:
+        request, response, finished = event("requestWillBeSent"), event("responseReceived"), event("loadingFinished")
+        aborted = event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True)
+        for events in [
+            [request, response],
+            [request, finished, response],
+            [request, response, finished, aborted],
+            [request, response, finished, finished],
+        ]:
+            with self.subTest(events=events), self.assertRaises(SmokeError):
+                _event_source_network_summary(events, REQUEST)
+
+    def test_other_request_terminals_are_ignored(self) -> None:
+        terminal = event("loadingFinished")
+        summary = _event_source_network_summary([
+            event("requestWillBeSent"), event("responseReceived"),
+            event("loadingFailed", requestId="other", errorText="net::ERR_FAILED"), terminal,
+        ], REQUEST)
+        self.assertEqual(summary["terminal"], terminal)
+
+
+if __name__ == "__main__":
+    unittest.main()

moli-cdp-smoke/tests/test_supervisor.py

@@ -22,6 +22,40 @@ from moli_cdp_smoke.supervisor import (
 
 
 class SupervisorWorkerTests(unittest.IsolatedAsyncioTestCase):
+    async def test_failed_scenario_is_reported_when_worker_has_no_top_level_error(self) -> None:
+        job = WorkerJob(group=GROUPS_BY_NAME["browser-semantics"], attempt=1, repeat=1)
+        for worker_error in (None, "worker traceback takes precedence"):
+            with self.subTest(worker_error=worker_error), tempfile.TemporaryDirectory() as temporary:
+                payload = {
+                    "ok": False,
+                    "group": "browser-semantics",
+                    "results": [{
+                        "name": "cdp_contract_event_source_runtime_and_network",
+                        "ok": False,
+                        "errorType": "SmokeError",
+                        "error": "timed out waiting for EventSource Network terminal",
+                    }],
+                }
+                if worker_error is not None:
+                    payload["error"] = worker_error
+
+                def fake_argv(_: WorkerJob, result_path: Path) -> list[str]:
+                    return [sys.executable, "-c", (
+                        "from pathlib import Path; import sys; "
+                        f"Path(sys.argv[1]).write_text({json.dumps(payload)!r}); "
+                        "print('later scenario passed'); sys.exit(1)"
+                    ), str(result_path)]
+
+                with patch("moli_cdp_smoke.supervisor._worker_argv", fake_argv):
+                    outcome = await _run_worker_job(job, Path(temporary), 5)
+                self.assertEqual(outcome.status, "failed")
+                if worker_error is not None:
+                    self.assertEqual(outcome.error, worker_error)
+                else:
+                    self.assertIn("cdp_contract_event_source_runtime_and_network", outcome.error)
+                    self.assertIn("SmokeError", outcome.error)
+                    self.assertIn("timed out waiting for EventSource", outcome.error)
+
     async def test_worker_result_is_read_from_its_isolated_file(self) -> None:
         group = GROUPS_BY_NAME["protocol"]
         job = WorkerJob(group=group, attempt=1, repeat=1)