Add @reproduce_failure support

.github/scripts/release.py

@@ -2,7 +2,7 @@
 import os
 import re
 import subprocess
-from datetime import datetime, timezone
+from datetime import datetime
 from pathlib import Path
 
 SOURCE_DIRS = ["src/"]
@@ -61,7 +61,7 @@ def set_version(pyproject: Path, new_version: str) -> None:
 
 
 def add_changelog(path: Path, *, version: str, content: str) -> None:
-    date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    date = datetime.now(datetime.UTC).strftime("%Y-%m-%d")
     entry = f"## {version} - {date}\n\n{content}"
 
     existing = path.read_text()

RELEASE.md

@@ -0,0 +1,3 @@
+RELEASE_TYPE: patch
+
+Add protocol support for reporting failure blobs back to the client. These are strings that can be used to reproduce a specific failure exactly.

src/hegel/server.py

@@ -10,6 +10,7 @@
 import cbor2
 from hypothesis import HealthCheck, settings
 from hypothesis.control import BuildContext
+from hypothesis.core import decode_failure, encode_failure
 from hypothesis.database import DirectoryBasedExampleDatabase
 from hypothesis.errors import (
     FailedHealthCheck,
@@ -297,6 +298,7 @@ def run_server_on_connection(connection: Connection) -> None:
                             test_cases=message["test_cases"],
                             database_key=message.get("database_key"),
                             seed=message.get("seed"),
+                            failure_blob=message.get("failure_blob"),
                             suppress_health_check=message.get(
                                 "suppress_health_check", []
                             ),
@@ -328,6 +330,7 @@ def _run_test(
     test_cases: int,
     database_key: bytes | None,
     seed: int | None,
+    failure_blob: bytes | None = None,
     suppress_health_check: list[str] | None,
     derandomize: bool,
     database: str | UniqueIdentifier | None,
@@ -398,7 +401,47 @@ def _run_test(
             database_key=database_key,
         )
         try:
-            runner.run()
+            if failure_blob is not None:
+                choices = decode_failure(failure_blob)
+                data = ConjectureData.for_choices(choices)
+                with contextlib.suppress(StopTest):
+                    state.test_function(data)
+
+                is_interesting = data.status is Status.INTERESTING
+                result = {
+                    "passed": not is_interesting,
+                    "test_cases": 1,
+                    "valid_test_cases": 0,
+                    "invalid_test_cases": 0,
+                    "interesting_test_cases": int(is_interesting),
+                }
+                if is_interesting:
+                    result["failure_blobs"] = [failure_blob]
+                    interesting_choices = [choices]
+                else:
+                    result["failure_blobs"] = []
+                    interesting_choices = []
+            else:
+                runner.run()
+
+                result = {
+                    "passed": len(runner.interesting_examples) == 0,
+                    "test_cases": runner.call_count,
+                    "valid_test_cases": runner.valid_examples,
+                    "invalid_test_cases": runner.invalid_examples,
+                    "interesting_test_cases": len(runner.interesting_examples),
+                    "seed": str(seed),
+                }
+                interesting_examples = sorted(
+                    runner.interesting_examples.values(),
+                    key=lambda d: sort_key(d.nodes),
+                )
+
+                interesting_choices = [v.choices for v in interesting_examples]
+
+                result["failure_blobs"] = [
+                    encode_failure(choices) for choices in interesting_choices
+                ]
         except FailedHealthCheck as e:
             result = {
                 "passed": False,
@@ -416,15 +459,6 @@ def _run_test(
             channel.send_request({"event": "test_done", "results": result}).get()
             return result
 
-        result = {
-            "passed": len(runner.interesting_examples) == 0,
-            "test_cases": runner.call_count,
-            "valid_test_cases": runner.valid_examples,
-            "invalid_test_cases": runner.invalid_examples,
-            "interesting_test_cases": len(runner.interesting_examples),
-            "seed": str(seed),
-        }
-
         # Check for flaky behavior detected during test execution
         flaky_error = state.flaky_error
         if flaky_error is not None:
@@ -435,14 +469,12 @@ def _run_test(
             result["flaky"] = FLAKY_TEST_RESULT_MSG
 
         channel.send_request({"event": "test_done", "results": result}).get()
+
         final_state = HegelState(connection, channel, is_final=True)
 
-        for v in sorted(
-            runner.interesting_examples.values(),
-            key=lambda d: sort_key(d.nodes),
-        ):
+        for choices in interesting_choices:
             with contextlib.suppress(StopTest):
-                final_state.test_function(ConjectureData.for_choices(v.choices))
+                final_state.test_function(ConjectureData.for_choices(choices))
 
         return result
     except Exception:

tests/client/client.py

@@ -49,13 +49,15 @@ def __init__(self, connection: ClientConnection):
         self.connection = connection
         self._control = connection.control_channel
         self.__lock = threading.Lock()
+        self.last_result: dict | None = None
 
     def run_test(
         self,
         test_fn: Callable[[], None],
         *,
         test_cases: int = 100,
         seed: int | None = None,
+        failure_blob: bytes | None = None,
         suppress_health_check: list[str] | None = None,
         database_key: bytes | None = None,
         derandomize: bool = False,
@@ -72,6 +74,7 @@ def run_test(
             "channel_id": test_channel.channel_id,
             "database_key": database_key,
             "derandomize": derandomize,
+            "failure_blob": failure_blob,
         }
         if database is not not_set:
             message["database"] = database
@@ -105,6 +108,7 @@ def run_test(
                 )
 
         assert result_data is not None
+        self.last_result = result_data
 
         if "error" in result_data:
             raise ValueError(result_data["error"])
@@ -117,6 +121,8 @@ def run_test(
 
         n_interesting = result_data["interesting_test_cases"]
 
+        if result_data["passed"] and failure_blob:
+            raise AssertionError("failure blob did not reproduce")
         if n_interesting == 0:
             return
 

tests/test_server.py

@@ -25,6 +25,11 @@
 )
 from tests.client.client import _request
 
+try:
+    ExceptionGroup
+except NameError:  # pragma: no cover
+    from exceptiongroup import ExceptionGroup
+
 
 def test_start_and_stop_span(client):
     def test():
@@ -323,6 +328,64 @@ def test():
     client.run_test(test, test_cases=10)
 
 
+def test_reproduce_failure(client):
+    def test():
+        assert (
+            generate_from_schema({"type": "integer", "min_value": 0, "max_value": 1000})
+            <= 10
+        )
+
+    with pytest.raises(AssertionError):
+        client.run_test(test, test_cases=100)
+
+    blob = client.last_result["failure_blobs"][0]
+    assert isinstance(blob, bytes)
+
+    with pytest.raises(AssertionError):
+        client.run_test(test, failure_blob=blob)
+
+
+def test_reproduce_failure_blob_no_longer_fails(client):
+    """When a blob no longer reproduces, the client raises RuntimeError."""
+
+    def failing_test():
+        assert (
+            generate_from_schema({"type": "integer", "min_value": 0, "max_value": 1000})
+            <= 10
+        )
+
+    with pytest.raises(AssertionError):
+        client.run_test(failing_test, test_cases=100)
+
+    blob = client.last_result["failure_blobs"][0]
+
+    # The blob was for failing_test, but we replay with a test that always passes.
+    with pytest.raises(AssertionError, match="failure blob did not reproduce"):
+        client.run_test(lambda: None, failure_blob=blob)
+
+
+def test_reproduce_failure_result_not_in_passing_test(client):
+    def test():
+        x = generate_from_schema({"type": "integer", "min_value": 0, "max_value": 100})
+        assert x >= 0
+
+    client.run_test(test, test_cases=50)
+    assert client.last_result["failure_blobs"] == []
+
+
+def test_multiple_blobs(client):
+    def test():
+        x = generate_from_schema({"type": "integer", "min_value": 0, "max_value": 100})
+        assert x <= 10
+
+        y = generate_from_schema({"type": "integer", "min_value": -10, "max_value": -1})
+        assert y >= 0
+
+    with pytest.raises(ExceptionGroup):
+        client.run_test(test, test_cases=50)
+    assert len(client.last_result["failure_blobs"]) == 2
+
+
 def test_derandomize_with_database_key(client):
     """Tests that derandomize=True derives seed from database_key."""