[FIX]: Fail pytest when a RAMPART trial gate fails

rampart/pytest_plugin/plugin.py

@@ -676,6 +676,28 @@ def _enforce_incomplete_exit_status(
         )
 
 
+def _enforce_trial_gate_exit_status(
+    *,
+    session: pytest.Session,
+    rampart_session: RampartSession,
+) -> None:
+    """Force a non-zero exit status when an aggregate trial gate fails.
+
+    Args:
+        session (pytest.Session): The active pytest session.
+        rampart_session (RampartSession): The aggregated RAMPART session state.
+    """
+    if session.exitstatus != pytest.ExitCode.OK:
+        return
+    if all(group.passed for group in rampart_session.trial_groups.values()):
+        return
+
+    session.exitstatus = pytest.ExitCode.TESTS_FAILED
+    logger.warning(
+        "One or more RAMPART trial gates failed; forcing non-zero exit status.",
+    )
+
+
 def pytest_sessionfinish(
     session: pytest.Session,
     exitstatus: int,  # noqa: ARG001  — pytest hook signature
@@ -694,8 +716,8 @@ def pytest_sessionfinish(
       emit); hook sinks are added here when the fixture path was
       suppressed.
 
-    An incomplete run (a lost or crashed worker) is forced to a non-zero
-    exit status so a dropped shard cannot pass silently.
+    A failed aggregate trial gate or an incomplete run is forced to a
+    non-zero exit status so pytest reflects the final RAMPART verdict.
 
     Args:
         session (pytest.Session): The pytest session.
@@ -718,6 +740,10 @@ def pytest_sessionfinish(
 
     _aggregate_trial_results(rampart_session=rampart_session)
     _evaluate_gates(rampart_session=rampart_session)
+    _enforce_trial_gate_exit_status(
+        session=session,
+        rampart_session=rampart_session,
+    )
     _enforce_incomplete_exit_status(session=session, rampart_session=rampart_session)
 
     if is_xdist_controller(config=session.config):

tests/unit/pytest_plugin/test_plugin.py

@@ -887,3 +887,59 @@ def test_incomplete_run_does_not_mask_existing_failure(self) -> None:
             rampart_session=rampart_session,
         )
         assert session.exitstatus == pytest.ExitCode.INTERRUPTED
+
+
+class TestTrialGateExitStatus:
+    """Failed aggregate trial gates force a non-zero exit status."""
+
+    def test_failed_trial_gate_forces_tests_failed(self) -> None:
+        from rampart.pytest_plugin.plugin import _enforce_trial_gate_exit_status
+
+        session = MagicMock()
+        session.exitstatus = pytest.ExitCode.OK
+        rampart_session = RampartSession()
+        rampart_session.record_trial_group(
+            base_nodeid="test.py::test_trial",
+            clone_nodeids=["test.py::test_trial[trial-0]"],
+            threshold=1.0,
+        )
+
+        _enforce_trial_gate_exit_status(
+            session=cast("pytest.Session", session),
+            rampart_session=rampart_session,
+        )
+
+        assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
+
+    def test_passing_trial_gate_preserves_ok_status(self) -> None:
+        from rampart.pytest_plugin.plugin import _enforce_trial_gate_exit_status
+
+        session = MagicMock()
+        session.exitstatus = pytest.ExitCode.OK
+        rampart_session = RampartSession()
+
+        _enforce_trial_gate_exit_status(
+            session=cast("pytest.Session", session),
+            rampart_session=rampart_session,
+        )
+
+        assert session.exitstatus == pytest.ExitCode.OK
+
+    def test_failed_trial_gate_does_not_mask_existing_failure(self) -> None:
+        from rampart.pytest_plugin.plugin import _enforce_trial_gate_exit_status
+
+        session = MagicMock()
+        session.exitstatus = pytest.ExitCode.INTERRUPTED
+        rampart_session = RampartSession()
+        rampart_session.record_trial_group(
+            base_nodeid="test.py::test_trial",
+            clone_nodeids=["test.py::test_trial[trial-0]"],
+            threshold=1.0,
+        )
+
+        _enforce_trial_gate_exit_status(
+            session=cast("pytest.Session", session),
+            rampart_session=rampart_session,
+        )
+
+        assert session.exitstatus == pytest.ExitCode.INTERRUPTED