@@ -29,7 +29,10 @@
 from pkcs11_check.core.evidence.obligations import ObligationInventory
 from pkcs11_check.core.evidence.schema import LedgerHeader, LedgerRow, Outcome
 
-#: sha256 of ``b""`` -- the pinned digest of an empty (zero-record) shard.
+#: sha256 of ``b""`` -- the pinned digest of an empty input to
+#: ``shard_digest``. Direct ``shard_digest([])`` use only: zero-record roster
+#: shards are rejected at construction, so this digest is unreachable via
+#: ``admit_lane`` (an absent shard always fails ``roster_complete``).
 EMPTY_SHARD_SHA256 = hashlib.sha256(b"").hexdigest()
 
 
@@ -40,6 +43,15 @@ class ShardExpectation:
     records: int
     sha256: str
 
+    def __post_init__(self) -> None:
+        """Reject zero-record shards: every roster shard must carry rows."""
+        if self.records < 1:
+            raise ValueError(
+                "ShardExpectation records must be >= 1 "
+                f"(got {self.records}); absent shards fail roster_complete, "
+                "they are never predeclared empty"
+            )
+
 
 @dataclass(frozen=True)
 class PredeclaredRoster:
@@ -139,17 +151,9 @@ def _check_roster_complete(mappings: list[dict[str, Any]], roster: PredeclaredRo
     problems: list[str] = []
     for shard_id, expected in roster.shards.items():
         if shard_id not in groups:
-            if expected.records == 0:
-                if EMPTY_SHARD_SHA256 != expected.sha256:
-                    problems.append(
-                        f"roster_complete: empty shard {shard_id!r} digest mismatch: "
-                        f"rows hash {EMPTY_SHARD_SHA256}, roster expects {expected.sha256}"
-                    )
-            else:
-                problems.append(
-                    f"roster_complete: roster shard {shard_id!r} missing from rows "
-                    "-- no partial credit"
-                )
+            problems.append(
+                f"roster_complete: roster shard {shard_id!r} missing from rows -- no partial credit"
+            )
             continue
         present = groups[shard_id]
         if len(present) != expected.records:
 
@@ -117,11 +117,19 @@ def to_record(self) -> dict[str, Any]:
 
     @classmethod
     def from_record(cls, record: Mapping[str, Any], *, line_number: int) -> LedgerHeader:
-        """Build a header from a parsed ledger line, failing loud on bad types."""
+        """Build a header from a parsed ledger line.
+
+        Fails loud on bad types and on missing known counter keys (absent
+        counters are malformed evidence, never zero-filled); unknown extra
+        keys stay ignored for forward compatibility.
+        """
         label = f"ledger line {line_number}"
         counters = record.get("counters")
         if not isinstance(counters, Mapping):
             raise ValueError(f"{label}: header counters is not an object")
+        missing = [key for key in COUNTER_KEYS if key not in counters]
+        if missing:
+            raise ValueError(f"{label}: header counters missing keys {missing!r}")
         merged = zero_counters()
         for key, value in counters.items():
             if not isinstance(key, str) or type(value) is not int:
 
@@ -3,11 +3,14 @@
 from __future__ import annotations
 
 import hashlib
+import json
 from dataclasses import replace
+from pathlib import Path
 
 import pytest
 
 from pkcs11_check.core.evidence.admit import (
+    EMPTY_SHARD_SHA256,
     PredeclaredRoster,
     ShardExpectation,
     admit_lane,
@@ -232,3 +235,116 @@ def test_obligation_membership() -> None:
         compile_obligations({"a": {}}, {"cases": ["a", "zzz"]})
     with pytest.raises(ValueError):
         compile_obligations({"a": {}}, {})
+
+
+def test_roster_rejects_zero_records_at_construction() -> None:
+    """R19: a roster shard declaring records=0 is rejected when built."""
+    with pytest.raises(ValueError):
+        ShardExpectation(records=0, sha256=EMPTY_SHARD_SHA256)
+    with pytest.raises(ValueError):
+        ShardExpectation(records=-1, sha256=EMPTY_SHARD_SHA256)
+    # Control: records >= 1 still builds.
+    ShardExpectation(records=1, sha256=EMPTY_SHARD_SHA256)
+
+
+def test_roster_complete_absent_shard_always_fails() -> None:
+    """R19: every roster shard must be present in rows -- no unless-clause."""
+    obligations = ObligationInventory(members=frozenset())
+    roster = PredeclaredRoster(
+        lane="soft",
+        shards={"never_arrived": ShardExpectation(records=1, sha256="0" * 64)},
+    )
+    # Reviewer probe shape: empty rows + absent shard must fail, not admit.
+    verdict = admit_lane(_header([]), [], roster, obligations)
+    assert not verdict.admitted
+    assert _failed_with(verdict.failures, "roster_complete")
+
+    # Absent shard alongside otherwise-clean rows must also fail.
+    header, rows, clean_roster, _ = _clean_lane()
+    shards = dict(clean_roster.shards)
+    shards["ghost"] = ShardExpectation(records=2, sha256="1" * 64)
+    verdict = admit_lane(header, rows, replace(clean_roster, shards=shards), obligations)
+    assert not verdict.admitted
+    assert _failed_with(verdict.failures, "roster_complete")
+
+
+def test_loader_missing_incomplete_key_fails_closed(tmp_path: Path) -> None:
+    """R20: end-to-end -- header counters without `incomplete` fail at load."""
+    from pkcs11_check.core.evidence import load_ledger
+
+    header, rows, _, _ = _clean_lane()
+    ledger = tmp_path / "ledger.jsonl"
+    header_record = header.to_record()
+    del header_record["counters"]["incomplete"]
+    ledger.write_text(
+        "\n".join(
+            [
+                json.dumps(header_record, sort_keys=True),
+                *(json.dumps(row.to_record(), sort_keys=True) for row in rows),
+            ]
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    with pytest.raises(ValueError, match="incomplete"):
+        load_ledger(ledger)
+
+
+def test_loader_full_header_loads_and_admits(tmp_path: Path) -> None:
+    """R20 control: a complete header loads and the lane admits."""
+    from pkcs11_check.core.evidence import load_ledger
+
+    header, rows, _, obligations = _clean_lane()
+    ledger = tmp_path / "ledger.jsonl"
+    ledger.write_text(
+        "\n".join(
+            [
+                json.dumps(header.to_record(), sort_keys=True),
+                *(json.dumps(row.to_record(), sort_keys=True) for row in rows),
+            ]
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    loaded_header, loaded_rows = load_ledger(ledger)
+    verdict = admit_lane(loaded_header, loaded_rows, _roster(loaded_rows), obligations)
+    assert verdict.admitted
+    assert verdict.failures == []
+
+
+def test_shard_digest_golden_fixture() -> None:
+    """Independent golden pin of the P11Lab-reused canonicalization.
+
+    Canonical bytes are handwritten below; the digest literal was computed
+    with shell ``sha256sum`` over those exact bytes (no trailing newline),
+    independently of ``shard_digest``.
+    """
+    blob_low = b'{"attempt":1,"outcome":"passed"}'
+    blob_high = b'{"attempt":2,"outcome":"failed"}'
+    row_low = {"outcome": "passed", "attempt": 1}
+    row_high = {"outcome": "failed", "attempt": 2}
+    assert row_blob(row_low) == blob_low
+    assert row_blob(row_high) == blob_high
+    # Rows passed in reverse byte order: sorting + b"\\n" join are pinned too.
+    assert (
+        shard_digest([row_high, row_low])
+        == "7fe9e7768bcd63c10328e3d9daf94a351bd4de8cc3566e65609a3e3f40f5edf5"
+    )
+
+
+def test_occurrence_preserved_duplicate_attempts() -> None:
+    """Attempts [1, 1, 2] with attempts_total=3 fail occurrence_preserved."""
+    rows = [_row("tests/a.py::t", attempt=attempt, attempts_total=3) for attempt in (1, 1, 2)]
+    obligations = ObligationInventory(members=frozenset())
+    verdict = admit_lane(_header(rows), rows, _roster(rows), obligations)
+    assert not verdict.admitted
+    assert _failed_with(verdict.failures, "occurrence_preserved")
+
+
+def test_occurrence_preserved_off_by_one_attempts() -> None:
+    """Attempts [2, 3, 4] with attempts_total=3 fail occurrence_preserved."""
+    rows = [_row("tests/a.py::t", attempt=attempt, attempts_total=3) for attempt in (2, 3, 4)]
+    obligations = ObligationInventory(members=frozenset())
+    verdict = admit_lane(_header(rows), rows, _roster(rows), obligations)
+    assert not verdict.admitted
+    assert _failed_with(verdict.failures, "occurrence_preserved")
@@ -14,6 +14,7 @@
 
 from pkcs11_check.core.crash_codes import CTYPES_ACCESS_VIOLATION
 from pkcs11_check.core.evidence import export_ledger
+from pkcs11_check.core.evidence.schema import zero_counters
 from pkcs11_check.core.report_log import iter_report_log_records
 
 TRACEMALLOC_LIMIT_BYTES = 64 * 1024 * 1024
@@ -438,13 +439,20 @@ def _header_line(schema_v: Any = 1) -> str:
             "raw_stream_bytes": 0,
             "raw_stream_sha256": "",
             "malformed_lines": 0,
-            "counters": {},
+            "counters": zero_counters(),
             "created_by": "pkcs11-check",
         },
         sort_keys=True,
     )
 
 
+def _header_line_missing_counter() -> str:
+    """R20 pin: a header with one counter key removed is malformed evidence."""
+    header = json.loads(_header_line())
+    del header["counters"]["incomplete"]
+    return json.dumps(header, sort_keys=True)
+
+
 def _row_line(node_id: str, phase: str = "call", attempt: int = 1, **overrides: Any) -> str:
     row: dict[str, Any] = {
         "record": "case",
@@ -491,6 +499,7 @@ def _row_line(node_id: str, phase: str = "call", attempt: int = 1, **overrides:
             "duplicate",
         ),
         ([_header_line(), _row_line("tests/a.py::t", outcome="bogus")], "outcome"),
+        ([_header_line_missing_counter()], "missing keys"),
     ],
 )
 def test_load_rejects(tmp_path: Path, lines: list[str], match: str) -> None: