warden re-audit (Fable 5): fix ledger race, quorum perf, small-order honesty

Correctness:
- ledger: single-flight fcntl lock over read-modify-append + fsync. Two
  concurrent writers previously could fork the hash chain (read same tail,
  same prev_hash). New test races 8 threads x6 appends; chain stays intact
  with contiguous indices.
- small-order list: the order-8 encodings were hand-typed and unverifiable
  and diverged from the canonical libsodium blocklist. A BOGUS entry is the
  only dangerous direction (it down-grades a real tamper to a note, skipping
  the latch), so the list is now the certain-low-order set only (y in
  {0,1,-1}, reduced/non-reduced, both sign bits); order-8 edges escalate to
  tamper until a derived list lands. Fail-safe asymmetry documented + tested.
- freshness: removed a tautological .

Non-functional:
- quorum members now run concurrently (ThreadPoolExecutor): a verify costs
  one member's latency, not the sum (~17ms for 4 members, live).
- Wallet.quorum() memoized per state_dir: binary swap-detection hashes run
  once at assembly, not on every verify; documented rationale.

85 tests green; live 4-fork wallet re-verified end-to-end.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/pacta/quorum.py

@@ -30,6 +30,7 @@ import os
 import shutil
 import subprocess
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -80,23 +81,36 @@ QUORUM_BACKENDS: dict[str, dict[str, Any]] = {
     },
 }
 
-# The eight small-order points' canonical encodings plus their sign-flipped
+# Encodings of low-order points, used ONLY to decide whether an inter-fork
-# variants - the classic excluded-R list (as used by Solana's legacy
+# divergence is a documented semantic edge (severity "note") rather than an
-# exclusion and libsodium's checks). Presence of R (or A) on this list is
+# unexplained one (severity "tamper" -> latch).
-# what makes an inter-fork divergence a *documented semantic edge*.
+#
+# FAIL-SAFE ASYMMETRY (why this list is deliberately conservative):
+#   - A MISSING low-order encoding is safe: a genuine edge on it is instead
+#     classified "unexplained" -> tamper -> custody latches. That is a false
+#     alarm / availability cost, never a security loss.
+#   - A BOGUS entry is dangerous: it would down-grade a real tamper to a mere
+#     "note" and skip the latch. So an entry may appear here ONLY if it is
+#     provably a low-order encoding.
+# Therefore this set contains exactly the encodings that are certainly
+# low-order from first principles: y in {0, 1, -1} (orders 4, 1, 2), in both
+# their reduced and non-reduced representatives, each with the sign bit clear
+# and set. These match RFC 8032 / libsodium's canonical low-order rows for
+# those y-values. The order-8 points (whose encodings must be *derived* via
+# field arithmetic, not transcribed) are intentionally OMITTED for now: an
+# edge on an order-8 R will escalate to tamper until a derived, tested list
+# lands. See test_quorum.py::test_small_order_list_is_conservative.
 SMALL_ORDER_ENCODINGS: frozenset[bytes] = frozenset(
     bytes.fromhex(h)
     for h in (
-        "0100000000000000000000000000000000000000000000000000000000000000",  # identity
+        "0000000000000000000000000000000000000000000000000000000000000000",  # y=0            (order 4)
-        "0000000000000000000000000000000000000000000000000000000000000000",  # (0, 0)-ish y=0 encoding
+        "0000000000000000000000000000000000000000000000000000000000000080",  # y=0,  sign set
-        "0000000000000000000000000000000000000000000000000000000000000080",  # y=0, sign flipped
+        "0100000000000000000000000000000000000000000000000000000000000000",  # y=1  identity  (order 1)
-        "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",  # -1 = p-1 (order 2)
+        "0100000000000000000000000000000000000000000000000000000000000080",  # y=1,  sign set
-        "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac037a",  # order-8 point
+        "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",  # y=p-1 = -1     (order 2)
-        "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac03fa",  # order-8, sign flipped
+        "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",  # y=p-1, sign set
-        "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc05",  # order-8 point
+        "edffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",  # y=p  ≡ 0        (non-canonical, order 4)
-        "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc85",  # order-8, sign flipped
+        "eeffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",  # y=p+1 ≡ 1       (non-canonical, order 1)
-        "0100000000000000000000000000000000000000000000000000000000000080",  # identity, sign flipped
-        "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",  # p-1 with high bit
     )
 )
 
@@ -263,12 +277,23 @@ class QuorumVerifier:
     def verify(self, payload: bytes, signature: bytes, public_key: bytes) -> QuorumResult:
         if len(signature) != 64 or len(public_key) != 32:
             raise ValueError("signature must be 64 bytes and public key 32 bytes")
-        verdicts: list[MemberVerdict] = []
         with tempfile.TemporaryDirectory(prefix="pacta-quorum-") as tmp:
             payload_path = Path(tmp) / "payload.bin"
             payload_path.write_bytes(payload)
-            for name, binary in sorted(self.members.items()):
+            # Members are independent subprocesses; run them concurrently so a
-                verdicts.append(self._run_member(name, binary, public_key, signature, payload_path))
+            # verification costs one member's latency, not the sum. subprocess
+            # releases the GIL while the child runs, so threads suffice. Sort
+            # the collected verdicts for a deterministic, position-stable trail.
+            names = sorted(self.members)
+            with ThreadPoolExecutor(max_workers=len(names)) as pool:
+                verdicts = list(
+                    pool.map(
+                        lambda name: self._run_member(
+                            name, self.members[name], public_key, signature, payload_path
+                        ),
+                        names,
+                    )
+                )
         return self._judge(verdicts, payload, signature, public_key)
 
     def _run_member(

src/pacta/wallet.py

@@ -32,6 +32,7 @@ boundary does not pretend to certify its apologies.
 
 from __future__ import annotations
 
+import fcntl
 import hashlib
 import json
 import os
@@ -224,6 +225,7 @@ class Wallet:
         self.receipts_dir = self.dir / "receipts"
         self.quarantine_dir = self.dir / "quarantine"
         self.airgap_dir = self.dir / "airgap"
+        self._quorum_cache: dict[str, QuorumVerifier] = {}
 
     # -- init / R4 gate ------------------------------------------------------
 
@@ -450,19 +452,34 @@ class Wallet:
         ]
 
     def _append_ledger(self, entry_type: str, body: dict[str, Any]) -> dict[str, Any]:
-        entries = self._ledger_entries()
+        # Single-flight: the chain is read-modify-append, so two concurrent
-        prev_hash = entries[-1]["entry_hash"] if entries else "0" * 64
+        # writers (threads, or two processes sharing a wallet dir) could both
-        entry = {
+        # read the same tail, compute the same prev_hash, and fork the chain.
-            "index": len(entries),
+        # An advisory exclusive lock over the whole critical section - taken
-            "timestamp": _now(),
+        # AFTER reopening under the lock so the prev-read reflects any writer
-            "entry_type": entry_type,
+        # that just finished - makes appends serialize. fsync so a crash can't
-            "body": body,
+        # leave a torn line that verify_ledger would read as tampering.
-            "prev_hash": prev_hash,
+        self.ledger_path.touch(exist_ok=True)
-        }
+        with self.ledger_path.open("r+", encoding="utf-8") as handle:
-        entry["entry_hash"] = _sha256(_canonical(entry))
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
-        with self.ledger_path.open("a", encoding="utf-8") as handle:
+            try:
-            handle.write(json.dumps(entry, sort_keys=True) + "\n")
+                existing = [json.loads(line) for line in handle.read().splitlines() if line.strip()]
-        return entry
+                prev_hash = existing[-1]["entry_hash"] if existing else "0" * 64
+                entry = {
+                    "index": len(existing),
+                    "timestamp": _now(),
+                    "entry_type": entry_type,
+                    "body": body,
+                    "prev_hash": prev_hash,
+                }
+                entry["entry_hash"] = _sha256(_canonical(entry))
+                handle.seek(0, os.SEEK_END)
+                handle.write(json.dumps(entry, sort_keys=True) + "\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+                return entry
+            finally:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
 
     def verify_ledger(self) -> tuple[bool, list[str]]:
         problems: list[str] = []
@@ -484,6 +501,18 @@ class Wallet:
     # -- quorum assembly -------------------------------------------------------
 
     def quorum(self, state_dir: str | Path | None = None) -> QuorumVerifier:
+        # Memoized per state_dir: assembling the quorum re-reads and SHA-256s
+        # every member binary (a swap-detection control). Doing that on every
+        # verify would hash ~4 binaries per operation; instead we pin once at
+        # first assembly and hold the verifier. A binary swapped mid-process
+        # is therefore caught at the next assembly (restart / re-init), which
+        # is the right granularity: an attacker who can rewrite the on-disk
+        # binary already outranks this check, and re-hashing per call buys
+        # nothing against them while taxing every honest verification.
+        cache_key = str(state_dir) if state_dir is not None else "__default__"
+        cached = self._quorum_cache.get(cache_key)
+        if cached is not None:
+            return cached
         capsule = self.capsule()
         members: dict[str, Path] = {}
         for member in capsule["members"]:
@@ -498,7 +527,9 @@ class Wallet:
                     f"({actual[:12]} != {member['binary_sha256'][:12]}); rebuild or re-init"
                 )
             members[name] = binary
-        return QuorumVerifier(members, min_members=int(capsule["policy"]["min_members"]))
+        verifier = QuorumVerifier(members, min_members=int(capsule["policy"]["min_members"]))
+        self._quorum_cache[cache_key] = verifier
+        return verifier
 
     # -- inbound ---------------------------------------------------------------
 
@@ -625,9 +656,9 @@ class Wallet:
                 request,
             )
         capsule = self.capsule()
-        freshness = self._check_freshness(capsule)
+        stale = self._check_freshness(capsule)
-        if freshness is not None:
+        if stale is not None:
-            return freshness if isinstance(freshness, Refusal) else freshness
+            return stale
         problem = self._validate_intent(intent, payload)
         if problem:
             return self._refuse(

tests/test_quorum.py

@@ -92,6 +92,18 @@ def test_edge_flags_detect_classes():
     assert any("non-canonical-s" in f for f in flags)
 
 
+def test_small_order_list_is_conservative():
+    # The fail-safe contract: every listed encoding must be a certain
+    # low-order y-value (0, 1, or -1), reduced or non-reduced, sign bit
+    # either way. No order-8 or unverifiable entries may sneak in, because a
+    # bogus entry would down-grade a real tamper to a note.
+    p = 2**255 - 19
+    allowed_y = {0, 1, p - 1, p, p + 1}
+    for enc in SMALL_ORDER_ENCODINGS:
+        y = int.from_bytes(enc, "little") & ((1 << 255) - 1)  # strip sign bit
+        assert y in allowed_y, f"non-low-order encoding in list: {enc.hex()}"
+
+
 def test_quorum_requires_min_members(tmp_path):
     with pytest.raises(ValueError):
         _quorum(tmp_path, {"only": "accept"})

tests/test_wallet.py

@@ -94,6 +94,33 @@ def test_ledger_is_hash_chained(tmp_path):
     assert not ok2 and problems2
 
 
+def test_ledger_survives_concurrent_appends(tmp_path):
+    # The single-flight lock must keep the chain intact under parallel writers
+    # (two threads racing on the same wallet's ledger).
+    import threading
+
+    wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state")
+    start = threading.Barrier(8)
+
+    def worker(n):
+        start.wait()
+        for i in range(6):
+            wallet._append_ledger("stress", {"worker": n, "i": i})
+
+    threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    ok, problems = wallet.verify_ledger()
+    assert ok, problems
+    # genesis + 48 appends, all with unique strictly-increasing indices
+    entries = wallet._ledger_entries()
+    indices = [e["index"] for e in entries]
+    assert indices == list(range(len(entries)))
+    assert len(entries) == 1 + 8 * 6
+
+
 def test_outbound_firewall_releases_when_quorum_agrees(tmp_path):
     from pacta.dogfood import locate_verifier