diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py
index 78c57a4fb7e..171aa4108bd 100644
--- a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/enginex/conftest.py
@@ -10,16 +10,21 @@
 import io
 import json
 import logging
+from pathlib import Path
 from typing import TYPE_CHECKING, Generator, cast
 
 import pytest
 from hive.client import Client, ClientType
-from hive.testing import HiveTest
+from hive.testing import HiveTest, HiveTestSuite
 
 from execution_testing.fixtures import BlockchainEngineXFixture
 from execution_testing.fixtures.blockchain import FixtureHeader
 from execution_testing.fixtures.pre_alloc_groups import PreAllocGroup
 
+from ....pytest_hive.reporting import (
+    retry_on_connection_error,
+    write_reported_test_count,
+)
 from ..helpers.test_tracker import (
     PreAllocGroupTestTracker,
     enginex_group_counts_key,
@@ -47,6 +52,10 @@
 def pytest_configure(config: pytest.Config) -> None:
     """Set the supported fixture formats for the enginex simulator."""
     config.supported_fixture_formats = [BlockchainEngineXFixture]  # type: ignore[attr-defined]
+    # Detect silent test loss: after the run, assert that every collected
+    # test reported its result to hive (see `_verify_reported_test_count`
+    # in the pytest_hive plugin).
+    config.assert_reported_test_count = True  # type: ignore[attr-defined]
 
 
 @pytest.hookimpl(trylast=True)
@@ -139,7 +148,29 @@ def _per_test_reporting(
     the hive node still exists, before `client` teardown calls
     `mark_test_completed` / `client.stop()`.
     """
-    hive_test.register_multi_test_client(client)
+    retry_on_connection_error(
+        "register multi-test client",
+        lambda: hive_test.register_multi_test_client(client),
+    )
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _reported_test_count_flush(
+    test_suite: HiveTestSuite,
+    session_temp_folder: Path,
+    request: pytest.FixtureRequest,
+) -> Generator[None, None, None]:
+    """
+    Persist this worker's reported-test counts before suite teardown.
+
+    Depending on `test_suite` guarantees this fixture's teardown runs
+    before the suite teardown, in which the last-finishing xdist worker
+    aggregates all workers' counts and asserts that no test results
+    were silently lost (never started or never reported to hive).
+    """
+    del test_suite
+    yield
+    write_reported_test_count(request.config, session_temp_folder)
 
 
 @pytest.fixture(scope="function")
diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py
new file mode 100644
index 00000000000..2b1293ccbf9
--- /dev/null
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/__init__.py
@@ -0,0 +1 @@
+"""Hive pytest plugin providing common functionality for simulators."""
diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py
index 89e24f1f5e9..c774cd0c853 100644
--- a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/pytest_hive.py
@@ -50,6 +50,13 @@
 from execution_testing.logging import get_logger
 
 from .hive_info import ClientFile, HiveInfo
+from .reporting import (
+    HiveReportedTestCountError,
+    count_skipped_test,
+    end_test_and_count,
+    retry_on_connection_error,
+    verify_reported_test_count,
+)
 
 logger = get_logger(__name__)
 
@@ -162,6 +169,8 @@ def pytest_runtest_makereport(
     outcome = yield
     report = outcome.get_result()
     setattr(item, f"result_{report.when}", report)
+    if report.when == "setup" and report.skipped:
+        count_skipped_test(item.config)
 
 
 @pytest.fixture(scope="session")
@@ -199,6 +208,7 @@ def get_test_suite_scope(fixture_name: str, config: pytest.Config) -> str:
 
 @pytest.fixture(scope=get_test_suite_scope)  # type: ignore[arg-type]
 def test_suite(
+    request: pytest.FixtureRequest,
     simulator: Simulation,
     session_temp_folder: Path,
     test_suite_name: str,
@@ -235,6 +245,7 @@ def test_suite(
 
     yield suite
 
+    count_check_error: str | None = None
     with FileLock(users_lock_file):
         with open(users_file, "r") as f:
             users = json.load(f)
@@ -242,9 +253,15 @@ def test_suite(
         with open(users_file, "w") as f:
             json.dump(users, f)
         if users == 0:
+            if getattr(request.config, "assert_reported_test_count", False):
+                count_check_error = verify_reported_test_count(
+                    request, suite, session_temp_folder
+                )
             suite.end()
             suite_file.unlink()
             users_file.unlink()
+    if count_check_error is not None:
+        raise HiveReportedTestCountError(count_check_error)
 
 
 @pytest.fixture(scope="module")
@@ -325,9 +342,12 @@ def hive_test(
         )
 
     test_parameter_string = request.node.name
-    test: HiveTest = test_suite.start_test(
-        name=test_parameter_string,
-        description=test_case_description,
+    test: HiveTest = retry_on_connection_error(
+        f"start test {test_parameter_string}",
+        lambda: test_suite.start_test(
+            name=test_parameter_string,
+            description=test_case_description,
+        ),
     )
     yield test
 
@@ -403,10 +423,10 @@ def hive_test(
                 "unknown).\n\n" + captured_output
             )
 
-        test.end(
-            result=HiveTestResult(
-                test_pass=test_passed, details=test_result_details
-            )
+        end_test_and_count(
+            request.config,
+            test,
+            HiveTestResult(test_pass=test_passed, details=test_result_details),
         )
         logger.verbose(
             f"Finished processing logs for test: {request.node.nodeid}"
@@ -420,8 +440,8 @@ def hive_test(
         test_result_details = (
             f"Exception whilst processing test result: {str(e)}"
         )
-        test.end(
-            result=HiveTestResult(
-                test_pass=test_passed, details=test_result_details
-            )
+        end_test_and_count(
+            request.config,
+            test,
+            HiveTestResult(test_pass=test_passed, details=test_result_details),
         )
diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py
new file mode 100644
index 00000000000..f5361ed184f
--- /dev/null
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/reporting.py
@@ -0,0 +1,197 @@
+"""
+Helpers guarding hive test-result reporting against connection errors
+and silent test loss.
+
+At very high test throughput (e.g. `consume-enginex`), the simulator
+can transiently exhaust its ephemeral port range toward the hive API
+endpoint: `connect()` then fails with `EADDRNOTAVAIL` for a few seconds
+on all xdist workers at once. Tests in flight during such a burst
+either error at setup (`start_test` failed: the test is then silently
+missing from the hive results), or run and pass but never report their
+verdict (`end_test` failed: the dangling hive test case is later
+force-closed as "Test was terminated by host").
+
+The root fix is connection pooling in the `ethereum-hive` package;
+these helpers are the complementary hardening on the simulator side:
+
+- `retry_on_connection_error` retries hive API calls with backoff so a
+  seconds-long burst does not error tests or lose results.
+- The reported-test counting (`end_test_and_count`,
+  `write_reported_test_count`, `verify_reported_test_count`) detects
+  any remaining silent loss by comparing the number of test results
+  successfully reported to hive against the number of collected tests
+  once the run has finished.
+"""
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Callable, Dict, TypeVar
+
+import pytest
+import requests
+from hive.testing import HiveTest, HiveTestResult, HiveTestSuite
+
+from execution_testing.logging import get_logger
+
+logger = get_logger(__name__)
+
+reported_test_count_key = pytest.StashKey[int]()
+"""Tests whose results this process successfully reported to hive."""
+
+skipped_test_count_key = pytest.StashKey[int]()
+"""Tests this process skipped during setup (never reported to hive)."""
+
+T = TypeVar("T")
+
+
+class HiveReportedTestCountError(Exception):
+    """Fewer test results than collected tests were reported to hive."""
+
+
+def retry_on_connection_error(
+    description: str,
+    call: Callable[[], T],
+    *,
+    attempts: int = 5,
+    initial_backoff: float = 0.5,
+) -> T:
+    """
+    Call `call`, retrying connection errors with exponential backoff.
+
+    A safety net for hive API calls during transient client-side
+    ephemeral-port exhaustion (`EADDRNOTAVAIL`) bursts; see the module
+    docstring. Connection errors are raised before the request has been
+    sent, so retrying cannot duplicate a non-idempotent call.
+    """
+    backoff = initial_backoff
+    for attempt in range(1, attempts + 1):
+        try:
+            return call()
+        except requests.exceptions.ConnectionError as e:
+            if attempt == attempts:
+                raise
+            logger.warning(
+                f"Hive API call '{description}' raised a connection error "
+                f"(attempt {attempt}/{attempts}), retrying in "
+                f"{backoff:.1f}s: {e}"
+            )
+            time.sleep(backoff)
+            backoff *= 2
+    raise AssertionError("unreachable")
+
+
+def end_test_and_count(
+    config: pytest.Config, test: HiveTest, result: HiveTestResult
+) -> None:
+    """End a hive test with retry and count the reported result."""
+    retry_on_connection_error(
+        f"end test {test.id}", lambda: test.end(result=result)
+    )
+    config.stash[reported_test_count_key] = (
+        config.stash.get(reported_test_count_key, 0) + 1
+    )
+
+
+def count_skipped_test(config: pytest.Config) -> None:
+    """
+    Count a test skipped during setup.
+
+    Such tests never start a hive test; counting them lets the
+    collected-vs-reported check account for them.
+    """
+    config.stash[skipped_test_count_key] = (
+        config.stash.get(skipped_test_count_key, 0) + 1
+    )
+
+
+def write_reported_test_count(
+    config: pytest.Config, session_temp_folder: Path
+) -> None:
+    """
+    Persist this process's reported/skipped test counts to the shared
+    session folder so that the last-finishing xdist worker can verify
+    the total against the number of collected tests.
+    """
+    worker_id = os.environ.get("PYTEST_XDIST_WORKER", "master")
+    counts = {
+        "reported": config.stash.get(reported_test_count_key, 0),
+        "skipped": config.stash.get(skipped_test_count_key, 0),
+    }
+    count_file = (
+        session_temp_folder / f"hive_reported_test_count_{worker_id}.json"
+    )
+    with open(count_file, "w") as f:
+        json.dump(counts, f)
+
+
+def verify_reported_test_count(
+    request: pytest.FixtureRequest,
+    suite: HiveTestSuite,
+    session_temp_folder: Path,
+) -> str | None:
+    """
+    Verify that every collected test reported a result to hive.
+
+    Return an error message if results were lost, `None` otherwise. On
+    loss, additionally report a failing meta-test case to hive so that
+    the loss is visible in the hive results, where the affected tests
+    are otherwise silently missing.
+    """
+    session = request.session
+    if session.shouldstop or session.shouldfail:
+        logger.warning(
+            "Test run was interrupted; skipping the collected-vs-reported "
+            "hive test count check."
+        )
+        return None
+    collected = session.testscollected
+    reported = 0
+    skipped = 0
+    worker_counts: Dict[str, Dict[str, int]] = {}
+    for count_file in sorted(
+        session_temp_folder.glob("hive_reported_test_count_*.json")
+    ):
+        with open(count_file, "r") as f:
+            counts = json.load(f)
+        worker_id = count_file.stem.removeprefix("hive_reported_test_count_")
+        worker_counts[worker_id] = counts
+        reported += counts["reported"]
+        skipped += counts["skipped"]
+        count_file.unlink()
+    if reported + skipped >= collected:
+        if reported + skipped > collected:
+            logger.warning(
+                f"More test results than collected tests: "
+                f"{reported} reported + {skipped} skipped > {collected} "
+                f"collected ({worker_counts})."
+            )
+        else:
+            logger.info(
+                f"All {collected} collected tests are accounted for "
+                f"({reported} reported to hive, {skipped} skipped)."
+            )
+        return None
+    missing = collected - reported - skipped
+    message = (
+        f"{missing} test result(s) were not reported to hive: collected "
+        f"{collected} tests, but only {reported} result(s) were reported "
+        f"and {skipped} test(s) were skipped (per-worker counts: "
+        f"{worker_counts}). Results are typically lost when hive API "
+        f"calls fail, e.g. due to client startup failures or ephemeral "
+        f"port exhaustion (EADDRNOTAVAIL) at high test throughput."
+    )
+    logger.error(message)
+    try:
+        check_test = suite.start_test(
+            name="reported-test-count-check",
+            description=(
+                "Meta-test verifying that every collected test reported "
+                "its result to hive; fails if test results were lost."
+            ),
+        )
+        check_test.end(result=HiveTestResult(test_pass=False, details=message))
+    except Exception as e:
+        logger.error(f"Failed to report the test count mismatch: {e}")
+    return message
diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py
new file mode 100644
index 00000000000..a74e7674cdb
--- /dev/null
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/__init__.py
@@ -0,0 +1 @@
+"""Tests for the pytest_hive plugin."""
diff --git a/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py
new file mode 100644
index 00000000000..c519c76fda3
--- /dev/null
+++ b/packages/testing/src/execution_testing/cli/pytest_commands/plugins/pytest_hive/tests/test_reporting.py
@@ -0,0 +1,187 @@
+"""Unit tests for the hive reporting guard helpers."""
+
+import json
+from pathlib import Path
+from typing import Any, List
+
+import pytest
+import requests
+from hive.testing import HiveTestResult
+
+from execution_testing.cli.pytest_commands.plugins.pytest_hive.reporting import (  # noqa: E501
+    count_skipped_test,
+    reported_test_count_key,
+    retry_on_connection_error,
+    skipped_test_count_key,
+    verify_reported_test_count,
+    write_reported_test_count,
+)
+
+
+class FakeSession:
+    """Session double exposing what `verify_reported_test_count` reads."""
+
+    def __init__(self, testscollected: int):  # noqa: D107
+        self.testscollected = testscollected
+        self.shouldstop: bool | str = False
+        self.shouldfail: bool | str = False
+
+
+class FakeRequest:
+    """FixtureRequest double carrying only a session."""
+
+    def __init__(self, session: FakeSession):  # noqa: D107
+        self.session = session
+
+
+class FakeHiveTest:
+    """HiveTest double recording its end result."""
+
+    def __init__(self, test_id: int):  # noqa: D107
+        self.id = test_id
+        self.result: HiveTestResult | None = None
+
+    def end(self, *, result: HiveTestResult) -> None:  # noqa: D102
+        self.result = result
+
+
+class FakeHiveTestSuite:
+    """HiveTestSuite double recording started meta-tests."""
+
+    def __init__(self) -> None:  # noqa: D107
+        self.started: List[FakeHiveTest] = []
+
+    def start_test(self, name: str, description: str) -> FakeHiveTest:  # noqa: D102
+        del name, description
+        test = FakeHiveTest(test_id=len(self.started))
+        self.started.append(test)
+        return test
+
+
+def write_count_file(folder: Path, worker_id: str, counts: dict) -> None:
+    """Write a per-worker reported-test count file."""
+    file = folder / f"hive_reported_test_count_{worker_id}.json"
+    with open(file, "w") as f:
+        json.dump(counts, f)
+
+
+def test_retry_returns_result_after_transient_connection_error() -> None:
+    """Transient connection errors are retried until success."""
+    calls: List[int] = []
+
+    def flaky() -> str:
+        calls.append(1)
+        if len(calls) < 3:
+            raise requests.exceptions.ConnectionError("EADDRNOTAVAIL")
+        return "ok"
+
+    result = retry_on_connection_error(
+        "flaky", flaky, attempts=5, initial_backoff=0.01
+    )
+    assert result == "ok"
+    assert len(calls) == 3
+
+
+def test_retry_raises_after_exhausting_attempts() -> None:
+    """The last connection error is raised once attempts are exhausted."""
+    calls: List[int] = []
+
+    def always_fails() -> None:
+        calls.append(1)
+        raise requests.exceptions.ConnectionError("EADDRNOTAVAIL")
+
+    with pytest.raises(requests.exceptions.ConnectionError):
+        retry_on_connection_error(
+            "fails", always_fails, attempts=3, initial_backoff=0.01
+        )
+    assert len(calls) == 3
+
+
+def test_retry_does_not_catch_other_exceptions() -> None:
+    """Non-connection errors (e.g. HTTP errors) propagate immediately."""
+    calls: List[int] = []
+
+    def http_error() -> None:
+        calls.append(1)
+        raise requests.exceptions.HTTPError("500 Server Error")
+
+    with pytest.raises(requests.exceptions.HTTPError):
+        retry_on_connection_error(
+            "http error", http_error, attempts=3, initial_backoff=0.01
+        )
+    assert len(calls) == 1
+
+
+def test_write_reported_test_count(
+    pytestconfig: pytest.Config, tmp_path: Path, monkeypatch: Any
+) -> None:
+    """Counts from the config stash are written to a per-worker file."""
+    monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw7")
+    monkeypatch.setitem(
+        pytestconfig.stash,
+        reported_test_count_key,
+        41,
+    )
+    monkeypatch.setitem(
+        pytestconfig.stash,
+        skipped_test_count_key,
+        1,
+    )
+    write_reported_test_count(pytestconfig, tmp_path)
+    file = tmp_path / "hive_reported_test_count_gw7.json"
+    with open(file, "r") as f:
+        assert json.load(f) == {"reported": 41, "skipped": 1}
+
+
+def test_count_skipped_test(
+    pytestconfig: pytest.Config, monkeypatch: Any
+) -> None:
+    """Skipped tests increment the stash counter."""
+    monkeypatch.setitem(
+        pytestconfig.stash,
+        skipped_test_count_key,
+        0,
+    )
+    count_skipped_test(pytestconfig)
+    assert pytestconfig.stash[skipped_test_count_key] == 1
+
+
+def test_verify_passes_when_all_tests_reported(tmp_path: Path) -> None:
+    """No error and no meta-test when all collected tests are reported."""
+    write_count_file(tmp_path, "gw0", {"reported": 6, "skipped": 0})
+    write_count_file(tmp_path, "gw1", {"reported": 3, "skipped": 1})
+    suite = FakeHiveTestSuite()
+    request = FakeRequest(FakeSession(testscollected=10))
+    error = verify_reported_test_count(request, suite, tmp_path)  # type: ignore[arg-type]
+    assert error is None
+    assert suite.started == []
+    # Count files are consumed by the check.
+    assert list(tmp_path.glob("hive_reported_test_count_*.json")) == []
+
+
+def test_verify_detects_lost_test_results(tmp_path: Path) -> None:
+    """A shortfall returns an error and reports a failing hive meta-test."""
+    write_count_file(tmp_path, "gw0", {"reported": 5, "skipped": 0})
+    write_count_file(tmp_path, "gw1", {"reported": 3, "skipped": 0})
+    suite = FakeHiveTestSuite()
+    request = FakeRequest(FakeSession(testscollected=10))
+    error = verify_reported_test_count(request, suite, tmp_path)  # type: ignore[arg-type]
+    assert error is not None
+    assert "2 test result(s) were not reported to hive" in error
+    assert len(suite.started) == 1
+    meta_test = suite.started[0]
+    assert meta_test.result is not None
+    assert meta_test.result.test_pass is False
+    assert error in meta_test.result.details
+
+
+def test_verify_skipped_when_run_interrupted(tmp_path: Path) -> None:
+    """No check (and no false positive) after -x/--maxfail or Ctrl-C."""
+    write_count_file(tmp_path, "gw0", {"reported": 1, "skipped": 0})
+    suite = FakeHiveTestSuite()
+    session = FakeSession(testscollected=10)
+    session.shouldfail = "stopping after 1 failures"
+    request = FakeRequest(session)
+    error = verify_reported_test_count(request, suite, tmp_path)  # type: ignore[arg-type]
+    assert error is None
+    assert suite.started == []