Add concurrent stateful testing

.claude/CLAUDE.md

@@ -171,26 +171,56 @@ public `Generator`/`TestCase`/`Generators`/`Hegel`/`Stateful` surface stays in `
   `@HegelTest`
   annotation + `HegelTestExtension` (a JUnit 5 `TestTemplateInvocationContextProvider` that drives
   the engine loop and invokes the user method per case).
-- **Stateful testing** — `Stateful.run(machine, tc)` reflects `@Rule`/`@Invariant` methods (sorted
-  by name for determinism) and registers them via `hegel_new_state_machine` as a *sequential*
-  machine (every rule in group 0, concurrency fixed at `1, 1`, worker index 0, and a per-machine
-  `step_count` — `Stateful.DEFAULT_STEP_COUNT` = 50, or the `run(machine, tc, stepCount)` overload;
-  the engine has no default). `@Rule(weight = w)` becomes the `rule_weights` array (validated
-  finite and positive in Java; `null` — the engine's all-equal default — when every rule is at 1);
-  the engine samples proportionally among the *enabled* rules, so weights are hints, and tests
-  check them per case, not in aggregate. It then drives the
-  engine's round protocol: `hegel_state_machine_next_group` opens each round (or reports
-  `HEGEL_STATE_MACHINE_DONE`, which is `INT64_MIN`), `hegel_state_machine_next_rule` hands out the
-  round's rules until the join point, a rule that fails its own assumption is reported with
-  `hegel_state_machine_rule_rejected` (and the round's STATEFUL_RULE span discarded), and at each
-  join point `hegel_state_machine_should_check_invariant` decides which invariants run — always
-  for `@Invariant(alwaysRun = true)`, sampled otherwise; the initial and final checks run every
-  invariant unconditionally. The machine handle is freed in a `finally`. An engine-level
-  `E_ASSUME` inside a rule (the data source is aborted) unwinds the whole case as invalid instead
-  of being reported as the rule's rejection. (Since engine 0.44 concurrency alone no longer marks
-  a run nondeterministic, and `HEGEL_RUN_STATUS_FAILED_NONDETERMINISTIC` is retired; this binding
-  drives machines sequentially anyway.) `Pool<T>` tracks previously generated values over
-  the engine's pool primitives so rules can reuse or consume them.
+- **Stateful testing** — `Stateful.run(machine, tc[, options])` reflects `@Rule`/`@Invariant`
+  methods (sorted by name for determinism) and registers them via `hegel_new_state_machine`:
+  `@Rule(group = "g")` names become `rule_groups` ids (0.. in first-appearance order; rules that
+  name none share `Stateful.ANONYMOUS_GROUP`, so a plain machine is all-zeros), `@Rule(weight = w)`
+  becomes the `rule_weights` array (validated finite and positive in Java; `null` — the engine's
+  all-equal default — when every rule is at 1; the engine samples proportionally among the
+  *enabled* rules, so weights are hints, and tests check them per case, not in aggregate), and
+  `Stateful.Options` carries the per-machine `step_count` (`Stateful.DEFAULT_STEP_COUNT` = 50; the
+  engine has no default) and the concurrency bounds (default `1, 1`). The engine draws the
+  concurrency level and the driver runs exactly that many workers. Two drivers share the
+  registration and the round protocol — `hegel_state_machine_next_group` opens each round on the
+  root handle (or reports `HEGEL_STATE_MACHINE_DONE`, which is `INT64_MIN`),
+  `hegel_state_machine_next_rule(worker)` hands out the round's rules until the worker's join point,
+  a rule that fails its own assumption is reported with `hegel_state_machine_rule_rejected(worker)`,
+  and at each join point `hegel_state_machine_should_check_invariant` (root handle) decides which
+  invariants run — always for `@Invariant(alwaysRun = true)`, sampled otherwise; the initial and
+  final checks run every invariant unconditionally:
+  - `maxConcurrency == 1` → the **sequential** driver, unchanged from before concurrency existed:
+    everything on the calling thread and the root handle as worker 0, one `STATEFUL_RULE` span
+    per round (discarded on rejection), `Step N: rule` notes.
+  - `maxConcurrency > 1` → the **concurrent** driver, modelled on hegel-rust's `run_concurrent_machine`
+    (even when the drawn level is 1, so a run's output format is uniform). Persistent daemon
+    worker threads (`hegel-worker-N`) live for the test case; per round the root thread notes the
+    header (`---------------- Round k: group "name" ----------------`), makes a **fresh clone per
+    worker** (`hegel_test_case_clone` + `hegel_test_case_set_worker`; cloning is a draw and can
+    return `E_STOP_TEST` on a replay, hence the raw-rc convention for `Libhegel.testCaseClone`),
+    hands each worker its clone through a queue, waits for one event per worker (the join point),
+    absorbs the workers' buffered draws/notes into the root in worker order (each line stamped
+    `[worker N +X.XXXms]` client-side — Java notes never go through `hegel_note`, so the engine's
+    own attribution is invisible here; draw names stay unique family-wide through the root's
+    synchronized name counter and are recorded plain), frees the clones, and resolves the round in
+    Rust's precedence: a control error (`LibhegelException`/`IllegalArgumentException`, or a worker
+    that exited without reporting) is rethrown first; then an overrun or an engine-level `E_ASSUME`
+    concludes the case, dropping any failures found alongside with a `Dropped concurrent failure
+    from worker N` note; otherwise the lowest worker's failure is rethrown as-is (its stack trace
+    is the worker's, so `originOf` still works) and the rest are noted as dropped. There is no
+    cancellation: a failing worker ends its own round and the others finish theirs (an overrun
+    aborts the family engine-side anyway). No spans are opened in the concurrent path (none of the
+    bindings do; the engine owns rule structure). The `Reporter` is only ever called from the
+    driving thread: verbose output of worker lines is delivered at the join. Workers are stopped
+    and joined in a `finally` before any leftover clone is freed.
+  The machine handle is freed in a `finally`. `Pool<T>` (sequential; bound to its creating handle,
+  unsynchronised) and `ConcurrentPool<T>` (one monitor across the engine call and the map update;
+  `add(tc, value)` draws through the calling rule's handle) track previously generated values over
+  the engine's pool primitives so rules can reuse or consume them. `LiveDataSource` maps
+  `E_CONCURRENT_USE` to a `HegelException` naming the fix (draw through the rule's own `TestCase`,
+  use `ConcurrentPool`). `FakeLibhegel.concurrentRounds` scripts per-worker rule queues so the
+  driver's join-point layout and every resolution path are tested deterministically
+  (`ConcurrentStatefulDriverTest`); `ConcurrentStatefulTest` covers the real engine (parallelism,
+  group exclusion, a found lost-update race, pools, blobs).
 - **Derivation** — `dev.hegel.generators.Derive` + `RecordGenerator` build generators from records,
   enums, scalars, and generic `List`/`Set`/`Optional`/`Map` by reflection.
 

README.md

@@ -97,6 +97,34 @@ Hegel reports the minimal example showing that our sort is incorrectly dropping
 
 The optional `"xs"` label passed to `draw` names the value in the falsifying-example output. See the [API documentation](https://javadoc.io/doc/dev.hegel/hegel) for a full tour of generators, combinators, control functions, and settings.
 
+## Stateful testing
+
+Model-based tests drive a state machine whose `@Rule` methods are the actions and whose `@Invariant`
+methods are the properties, with the engine choosing and shrinking the action sequence:
+
+```java
+Stateful.run(new IntegerStack(), tc);
+```
+
+Run the same machine with a concurrency bound to look for races: the engine draws how many worker
+threads apply the rules at once, `@Rule(group = "...")` says which rules may overlap, and a
+`ConcurrentPool` passes generated values between concurrent rules.
+
+```java
+Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4));
+```
+
+See the `Stateful` Javadoc for complete sequential and concurrent examples. The concurrent one, a
+counter that loses updates under contention, is checked in as a runnable demonstration that is
+skipped unless asked for; run it from a checkout with
+
+```bash
+mvn test -pl hegel -am -Dtest=ConcurrentCounterExample -Dhegel.examples=true -Dsurefire.failIfNoSpecifiedTests=false
+```
+
+(or `-pl hegel-jna -am` for the JNA frontend) and it fails with a report of the rounds and the
+worker-stamped rule and draw lines that led to the lost update.
+
 ## Using Hegel as a library
 
 Frontends for other JVM languages (or custom runners) use `Hegel.run` instead of `Hegel.test`. It

RELEASE.md

@@ -0,0 +1,88 @@
+RELEASE_TYPE: minor
+
+This release adds concurrent stateful testing. Running a state machine with
+`Stateful.Options.maxConcurrency` above 1 makes the engine draw a concurrency level for each test
+case and the driver run that many worker threads, which apply the machine's rules at the same time
+on the same machine object, so races and lost updates surface as invariant failures like any other
+bug:
+
+```java
+class Counter {
+  private final Map<String, Integer> store = new ConcurrentHashMap<>();
+  private final AtomicInteger increments = new AtomicInteger();
+  private final ConcurrentPool<String> keys;
+
+  Counter(TestCase tc) {
+    keys = new ConcurrentPool<>(tc);
+  }
+
+  @Rule(group = "ops")
+  void register(TestCase tc) {
+    String key = tc.draw(text().minSize(1).maxSize(3));
+    store.putIfAbsent(key, 0);
+    keys.add(tc, key);
+  }
+
+  @Rule(group = "ops", weight = 3)
+  void increment(TestCase tc) {
+    String key = tc.draw(keys.reusable());
+    store.put(key, store.get(key) + 1); // racy: a lost update
+    increments.incrementAndGet();
+  }
+
+  @Rule(group = "audit")
+  void audit(TestCase tc) {
+    tc.note("store holds " + store.size() + " keys");
+  }
+
+  @Invariant
+  void noLostUpdates(TestCase tc) {
+    assertEquals(increments.get(), store.values().stream().mapToInt(Integer::intValue).sum());
+  }
+}
+
+@HegelTest
+void counterUnderContention(TestCase tc) {
+  Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4));
+}
+```
+
+Execution proceeds in rounds. Each round the engine picks one concurrency group (`@Rule(group =
+"...")`; rules that name none share the anonymous group), every worker applies a few of that group's
+rules concurrently, and once all workers have finished the round the sampled invariants run on the
+driving thread. Rules in the same group may therefore overlap in time, rules in different groups
+never do, and invariants never overlap a rule. Rules run on the same machine object from several
+threads, so its state must be safe for concurrent access; each rule receives its worker's own
+`TestCase` and must draw only through it. The new `ConcurrentPool` is the thread-safe counterpart of
+`Pool` for passing generated values between concurrent rules; its `add` takes the calling rule's test
+case. A failure report groups each round's draws and notes by worker under the round's header, every
+line stamped `[worker N +X.XXXms]` with the time since the machine started:
+
+```
+Concurrency level: 2
+---------------- Round 1: group "ops" ----------------
+[worker 0 +0.412ms] Rule: increment
+[worker 0 +0.437ms] draw_1 = "a";
+[worker 1 +0.415ms] Rule: increment
+[worker 1 +0.440ms] draw_2 = "a";
+
+java.lang.AssertionError: expected: <2> but was: <1>
+```
+
+When a rule fails in one worker the other workers finish their round before the failure is
+reported, and when several workers fail in the same round the lowest-numbered worker's exception is
+reported with the others noted as dropped. A concurrency bug that depends on thread scheduling may
+not reproduce on every replay; the engine then confirms it by repeated replay and reports it with a
+caveat, as for any nondeterministic failure.
+
+`Stateful.options()` is the new way to configure a run:
+`Stateful.run(machine, tc, Stateful.options().stepCount(30).minConcurrency(2).maxConcurrency(4))`.
+The existing `run(machine, tc)` and `run(machine, tc, stepCount)` overloads are unchanged and keep
+running the machine sequentially on the calling thread, with the same output and choice sequences
+as before, so stored failures for existing stateful tests remain valid.
+
+In `dev.hegel:hegel-lowlevel`, `Libhegel` gains `testCaseClone` (`hegel_test_case_clone`; a draw,
+so it returns the raw return code and reports `E_STOP_TEST` on a replay of a shorter sequence) and
+`testCaseSetWorker` (`hegel_test_case_set_worker`). Bindings implementing `Libhegel` must add the
+two methods, which is what makes this a minor release; the `dev.hegel:hegel` and
+`dev.hegel:hegel-jna` frontends only gain API.

hegel-jna/src/main/java/dev/hegel/JnaLibhegel.java

@@ -122,6 +122,10 @@ int hegel_test_case_from_blob(
 
         int hegel_test_case_should_capture(Pointer ctx, Pointer tc, ByteByReference out);
 
+        int hegel_test_case_clone(Pointer ctx, Pointer tc, PointerByReference out);
+
+        int hegel_test_case_set_worker(Pointer ctx, Pointer tc, long workerIndex);
+
         int hegel_generate_boolean(Pointer ctx, Pointer tc, double p, byte hasForced, byte forced, ByteByReference out);
 
         int hegel_generate_integer(Pointer ctx, Pointer tc, long min, long max, LongByReference out);
@@ -548,6 +552,21 @@ public boolean testCaseShouldCapture(long tc) {
         return out.getValue() != 0;
     }
 
+    @Override
+    public int testCaseClone(long tc, long[] out) {
+        PointerByReference ref = new PointerByReference();
+        int code = lib.hegel_test_case_clone(ctx(), pointer(tc), ref);
+        if (code == Abi.OK) {
+            out[0] = address(ref.getValue());
+        }
+        return code;
+    }
+
+    @Override
+    public void testCaseSetWorker(long tc, long workerIndex) {
+        check("hegel_test_case_set_worker", lib.hegel_test_case_set_worker(ctx(), pointer(tc), workerIndex));
+    }
+
     // --- draws ---
 
     @Override

hegel-jna/src/test/java/dev/hegel/JnaLibhegelTest.java

@@ -74,6 +74,10 @@ void infrastructureCallsReportNullHandles() {
         assertThrows(HegelException.class, () -> lib.runStart(0, null));
         assertThrows(HegelException.class, () -> lib.runStartBlob(0, "blob", null));
         assertThrows(HegelException.class, () -> lib.testCaseShouldCapture(0));
+        assertThrows(HegelException.class, () -> lib.testCaseSetWorker(0, 0));
+        long[] clone = {7};
+        assertEquals(Abi.E_INVALID_HANDLE, lib.testCaseClone(0, clone));
+        assertEquals(7, clone[0]);
     }
 
     @Test

hegel-lowlevel/src/main/java/dev/hegel/lowlevel/Libhegel.java

@@ -200,6 +200,28 @@ static Libhegel load(Path library, Iterable<? extends LibhegelBackend> backends)
      */
     boolean testCaseShouldCapture(long tc);
 
+    /**
+     * {@code hegel_test_case_clone}: a caller-owned handle onto an independent choice stream of
+     * the same test case. The clone shares the case's outcome and budgets (and its collections,
+     * pools and state machines, which are family-wide) but draws from its own sequence, so a clone
+     * and its source can be driven concurrently from different threads while staying
+     * deterministic under replay. Release it with {@link #testCaseFree}.
+     *
+     * <p>Cloning consumes one choice position on the source stream, so it is a draw: it returns
+     * the raw rc — {@link Abi#E_STOP_TEST} when the choice budget is exhausted (as on a replay of
+     * a shorter sequence), {@link Abi#E_CONCURRENT_USE} if the source is mid-operation on another
+     * thread, {@link Abi#E_ALREADY_COMPLETE} once the case has completed — and on OK writes the
+     * handle into {@code out[0]}.
+     */
+    int testCaseClone(long tc, long[] out);
+
+    /**
+     * {@code hegel_test_case_set_worker}: attribute the handle's engine-side output (notes and
+     * printer lines) to concurrent worker {@code workerIndex}, which must be non-negative. Blocks
+     * and clones derived from the handle afterwards inherit the attribution.
+     */
+    void testCaseSetWorker(long tc, long workerIndex);
+
     // Per-test-case draws. Each returns the raw rc.
     int generateBoolean(long tc, double p, boolean[] out);
 

hegel/src/main/java/dev/hegel/RealLibhegel.java

@@ -123,6 +123,8 @@ private static MethodHandle findEmitLine() {
     private final MethodHandle testCaseFromBlob;
     private final MethodHandle testCaseFree;
     private final MethodHandle testCaseShouldCapture;
+    private final MethodHandle testCaseClone;
+    private final MethodHandle testCaseSetWorker;
     private final MethodHandle generateBoolean;
     private final MethodHandle generateInteger;
     private final MethodHandle generateFloat;
@@ -288,6 +290,13 @@ private static MethodHandle findEmitLine() {
                 lookup,
                 "hegel_test_case_should_capture",
                 FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS));
+        this.testCaseClone =
+                h(linker, lookup, "hegel_test_case_clone", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS));
+        this.testCaseSetWorker = h(
+                linker,
+                lookup,
+                "hegel_test_case_set_worker",
+                FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_LONG));
         this.generateBoolean = h(
                 linker,
                 lookup,
@@ -736,6 +745,21 @@ public boolean testCaseShouldCapture(long tc) {
         return out.get(JAVA_BOOLEAN, 0);
     }
 
+    @Override
+    public int testCaseClone(long tc, long[] out) {
+        MemorySegment seg = Arena.ofAuto().allocate(ADDRESS);
+        int code = rc(testCaseClone, segment(tc), seg);
+        if (code == Abi.OK) {
+            out[0] = seg.get(ADDRESS, 0).address();
+        }
+        return code;
+    }
+
+    @Override
+    public void testCaseSetWorker(long tc, long workerIndex) {
+        check("hegel_test_case_set_worker", rc(testCaseSetWorker, segment(tc), workerIndex));
+    }
+
     // --- draws ---
 
     @Override

hegel/src/test/java/dev/hegel/RealLibhegelTest.java

@@ -83,6 +83,10 @@ void infrastructureCallsReportNullHandles() {
         assertThrows(HegelException.class, () -> lib.runStart(0, null));
         assertThrows(HegelException.class, () -> lib.runStartBlob(0, "blob", null));
         assertThrows(HegelException.class, () -> lib.testCaseShouldCapture(0));
+        assertThrows(HegelException.class, () -> lib.testCaseSetWorker(0, 0));
+        long[] clone = {7};
+        assertEquals(Abi.E_INVALID_HANDLE, lib.testCaseClone(0, clone));
+        assertEquals(7, clone[0]);
     }
 
     @Test