fix(validation): resolve PK identity-raise vs check-report tension (item 8j)

src/mountainash/datacontracts/contract.py

@@ -85,9 +85,14 @@ def to_typespec(cls) -> "TypeSpec":
 
     @classmethod
     def to_checks(cls) -> "list[ValidationCheck]":
+        from mountainash.datacontracts.compiler import primary_key_check
+
         checks: "list[ValidationCheck]" = []
         for name, contract_field in cls._contract_fields.items():
             checks.extend(contract_field.to_checks(name))
+        pk_check = primary_key_check(cls.to_typespec())
+        if pk_check is not None:
+            checks.append(pk_check)
         return checks
 
     @classmethod
@@ -100,13 +105,22 @@ def validate_datacontract(
         tail: int | None = None,
         sample: int | None = None,
         random_seed: int | None = None,
+        allow_imperfect_key: bool = False,
     ) -> "ValidationResult":
-        """Validate data against this contract; returns (never raises)."""
+        """Validate data against this contract; returns a ValidationResult.
+
+        Raises IdentityInvalidError if this contract declares a keyed identity
+        (Config.primary_key / Config.natural_key) and the data does not honour
+        it: always for key fields missing from the data (declaration-phase,
+        spec §7); for null-key rows or duplicate key tuples, unless
+        allow_imperfect_key=True — which lets the run proceed and reports the
+        duplicates via the primary_key_unique check instead (spec §9.3).
+        """
         from mountainash.datacontracts.validator import Validator
 
         return Validator(name=cls.contract_name(), contract=cls).validate(
             data, context=context, head=head, tail=tail, sample=sample,
-            random_seed=random_seed,
+            random_seed=random_seed, allow_imperfect_key=allow_imperfect_key,
         )
 
     @classmethod
@@ -119,11 +133,15 @@ def validate_datacontract_quick(
         tail: int | None = None,
         sample: int | None = None,
         random_seed: int | None = None,
+        allow_imperfect_key: bool = False,
     ) -> "ValidationResult":
-        """Quick validation — same runner, fail_fast=True (item 18 subsumed)."""
+        """Quick validation — same runner, fail_fast=True (item 18 subsumed).
+
+        See validate_datacontract for the allow_imperfect_key contract.
+        """
         from mountainash.datacontracts.validator import Validator
 
         return Validator(name=cls.contract_name(), contract=cls).validate_quick(
             data, context=context, head=head, tail=tail, sample=sample,
-            random_seed=random_seed,
+            random_seed=random_seed, allow_imperfect_key=allow_imperfect_key,
         )

src/mountainash/relations/dag/dag.py

@@ -615,17 +615,22 @@ def validate(
         context: dict[str, Any] | None = None,
         backend: Optional[str] = None,
         failure_sample: Optional[int] = None,
+        allow_imperfect_key: bool = False,
     ) -> "DAGValidationResult":
         """Full validation via the backend-agnostic ValidationRunner.
 
         Per-resource checks compile from each spec/contract; FK row-integrity
         checks are generated from constraint_metadata + spec foreign keys by
         validation.fk.build_fk_checks and compiled as relation anti-joins.
+        A resource's invalid keyed identity is isolated into that resource's
+        own failing result (check_id="__identity__") rather than raised out
+        of this call - every other resource still validates (spec item 8j §3.2).
         """
         from mountainash.relations.dag.validation import validate
 
         return validate(
-            self, specs, context=context, backend=backend, failure_sample=failure_sample
+            self, specs, context=context, backend=backend, failure_sample=failure_sample,
+            allow_imperfect_key=allow_imperfect_key,
         )
 
     def validate_quick(
@@ -635,12 +640,14 @@ def validate_quick(
         context: dict[str, Any] | None = None,
         backend: Optional[str] = None,
         failure_sample: Optional[int] = None,
+        allow_imperfect_key: bool = False,
     ) -> "DAGValidationResult":
         """Fast validation via the ValidationRunner (fail_fast=True; identical shapes)."""
         from mountainash.relations.dag.validation import validate_quick
 
         return validate_quick(
-            self, specs, context=context, backend=backend, failure_sample=failure_sample
+            self, specs, context=context, backend=backend, failure_sample=failure_sample,
+            allow_imperfect_key=allow_imperfect_key,
         )
 
     def _unknown_ref_error(self, missing: str) -> "UnknownRelationRef":

src/mountainash/relations/dag/validation.py

@@ -26,11 +26,13 @@ def validate(
     context: "dict[str, Any] | None" = None,
     backend: str | None = None,
     failure_sample: int | None = None,
+    allow_imperfect_key: bool = False,
 ) -> DAGValidationResult:
     """Full validation — all per-resource checks, then all FK checks."""
     return _run(
         dag, specs, context=context, backend=backend,
         fail_fast=False, failure_sample=failure_sample,
+        allow_imperfect_key=allow_imperfect_key,
     )
 
 
@@ -41,11 +43,13 @@ def validate_quick(
     context: "dict[str, Any] | None" = None,
     backend: str | None = None,
     failure_sample: int | None = None,
+    allow_imperfect_key: bool = False,
 ) -> DAGValidationResult:
     """Fast validation — same runner, fail_fast=True. Identical shapes."""
     return _run(
         dag, specs, context=context, backend=backend,
         fail_fast=True, failure_sample=failure_sample,
+        allow_imperfect_key=allow_imperfect_key,
     )
 
 
@@ -57,6 +61,7 @@ def _run(
     backend: str | None,
     fail_fast: bool,
     failure_sample: int | None,
+    allow_imperfect_key: bool = False,
 ) -> DAGValidationResult:
     from mountainash.datacontracts.compiler import compile_datacontract
     from mountainash.datacontracts.contract import BaseDataContract
@@ -104,4 +109,5 @@ def _run(
         failure_sample=failure_sample,
         backend=backend,
         fk_error_summaries=fk_errors,
+        allow_imperfect_key=allow_imperfect_key,
     )

src/mountainash/validation/runner.py

@@ -20,7 +20,7 @@
 
 from mountainash.expressions.core.expression_nodes import ScalarFunctionNode
 from mountainash.validation.checks import VERDICT_PASSING, check_kind
-from mountainash.validation.errors import UnknownCheckTypeError
+from mountainash.validation.errors import IdentityInvalidError, UnknownCheckTypeError
 from mountainash.validation.identity import RowIdentity, validate_keyed_identity
 from mountainash.validation.result import (
     CheckSummary,
@@ -434,6 +434,7 @@ def validate_dag(
         failure_sample: int | None = None,
         backend: str | None = None,
         fk_error_summaries: "list[CheckSummary] | None" = None,
+        allow_imperfect_key: bool = False,
     ) -> "DAGValidationResult":
         from mountainash.relations import relation as as_relation
         from mountainash.validation.checks import ForeignKeyRule
@@ -452,15 +453,43 @@ def _resolver(name: str) -> Any:
         for name, checks in checks_by_resource.items():
             intra = [c for c in checks if not isinstance(c, ForeignKeyRule)]
             fk_rules.extend(c for c in checks if isinstance(c, ForeignKeyRule))
-            result = self.validate_relation(
-                _resolver(name),
-                intra,
-                identity=identity_by_resource.get(name),
-                context=context,
-                fail_fast=fail_fast,
-                failure_sample=failure_sample,
-                validator_name=name,
-            )
+            resource_identity = identity_by_resource.get(name) or RowIdentity("none")
+            try:
+                result = self.validate_relation(
+                    _resolver(name),
+                    intra,
+                    identity=resource_identity,
+                    allow_imperfect_key=allow_imperfect_key,
+                    context=context,
+                    fail_fast=fail_fast,
+                    failure_sample=failure_sample,
+                    validator_name=name,
+                )
+            except IdentityInvalidError as exc:
+                # spec item 8j §3.2: a resource's invalid keyed identity never
+                # aborts the batch — isolate it into that resource's own failing
+                # result, same as every other exception in this loop already is
+                # (materialisation failures, runner.py:118-134). "__identity__"
+                # mirrors the existing "__fk__" synthetic-result naming
+                # (runner.py:470, the fail_fast early-return branch this snippet
+                # mirrors; runner.py:490, the fk_result construction).
+                summary = CheckSummary(
+                    check_id="__identity__",
+                    check_kind=None,
+                    status="error",
+                    severity="blocking",
+                    error=f"{type(exc).__name__}: {exc}",
+                )
+                result = ValidationResult(
+                    passes=False,
+                    validator_name=name,
+                    datacontract_name=None,
+                    context=dict(context or {}),
+                    check_summaries=summaries_frame([summary]),
+                    failure_cases=combine_failure_frames([], resource_identity),
+                    identity=resource_identity,
+                    identity_diagnostics={},
+                )
             results[name] = result
             if fail_fast and not result.passes:
                 return DAGValidationResult(

tests/datacontracts/test_native_contract.py

@@ -1,9 +1,11 @@
 """Native BaseDataContract: declaration collection, TypeSpec round trip, validate."""
 import polars as pl
+import pytest
 
 from mountainash.datacontracts.contract import BaseDataContract
 from mountainash.datacontracts.field import Field
 from mountainash.typespec.universal_types import UniversalType
+from mountainash.validation.errors import IdentityInvalidError
 
 
 class UserContract(BaseDataContract):
@@ -44,10 +46,34 @@ def test_natural_key_survives_as_primary_key(self):
     def test_to_checks_ids(self):
         ids = [c.id for c in UserContract.to_checks()]
         assert "id__not_null" in ids
-        assert "id__unique" in ids
+        assert "email__not_null" in ids
         assert "email__pattern" in ids
         assert "age__ge" in ids
 
+    def test_to_checks_includes_primary_key_unique(self):
+        ids = [c.id for c in UserContract.to_checks()]
+        assert "primary_key_unique" in ids  # UserContract: Config.natural_key = ["id"]
+
+    def test_to_checks_includes_primary_key_unique_for_primary_key_config(self):
+        class OrderContract(BaseDataContract):
+            order_id: int = Field(nullable=False)
+
+            class Config:
+                name = "orders"
+                primary_key = ["order_id"]
+
+        ids = [c.id for c in OrderContract.to_checks()]
+        assert "primary_key_unique" in ids
+
+
+def test_to_checks_matches_compile_datacontract_check_ids():
+    from mountainash.datacontracts.compiler import compile_datacontract, contract_from_typespec
+
+    spec = UserContract.to_typespec()
+    compiled_ids = {c.id for c in compile_datacontract(spec)}
+    contract_ids = {c.id for c in contract_from_typespec(spec).to_checks()}
+    assert compiled_ids == contract_ids
+
 
 class TestValidate:
     def test_valid_data_passes(self):
@@ -92,6 +118,35 @@ def test_quick_is_fail_fast_same_shapes(self):
         assert list(full.failure_cases.columns) == list(quick.failure_cases.columns)
         assert quick.check_summaries.height <= full.check_summaries.height
 
+    def test_validate_datacontract_raises_by_default_on_duplicate_key(self):
+        df = pl.DataFrame(
+            {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]}
+        )
+        with pytest.raises(IdentityInvalidError):
+            UserContract.validate_datacontract(df)
+
+    def test_validate_datacontract_allow_imperfect_key_reports_primary_key_unique(self):
+        df = pl.DataFrame(
+            {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]}
+        )
+        result = UserContract.validate_datacontract(df, allow_imperfect_key=True)
+        assert result.passes is False
+        failing = set(
+            result.check_summaries.filter(
+                result.check_summaries["status"] != "passed"
+            )["check_id"].to_list()
+        )
+        assert "primary_key_unique" in failing
+        assert result.identity_diagnostics["duplicate_key_tuples"] == 1
+
+    def test_validate_datacontract_quick_allow_imperfect_key_same_shape(self):
+        df = pl.DataFrame(
+            {"id": [1, 1], "email": ["a@b.c", "d@e.f"], "age": [30, 40], "note": ["x", "y"]}
+        )
+        result = UserContract.validate_datacontract_quick(df, allow_imperfect_key=True)
+        assert result.passes is False
+        assert result.identity_diagnostics["duplicate_key_tuples"] == 1
+
 
 class NoKeyContract(BaseDataContract):
     id: int = Field(unique=True)

tests/relations/dag/cross_backend/test_datapackage_validation_loop.py

@@ -15,7 +15,6 @@
 import pytest
 
 from mountainash.typespec.datapackage import DataPackage
-from mountainash.validation.errors import IdentityInvalidError
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -119,12 +118,14 @@ def test_conforming_data_passes_cross_backend(tmp_path, backend_name):
 
 @pytest.mark.cross_backend
 @pytest.mark.parametrize("backend_name", _COLLECT_BACKENDS)
-def test_duplicate_primary_key_raises_identity_error(tmp_path, backend_name):
+def test_duplicate_primary_key_isolates_identity_failure(tmp_path, backend_name):
     """Item 8j's characterization: a declared primary_key resolving to keyed
-    identity raises IdentityInvalidError and aborts the batch — it never
-    returns a DAGValidationResult for that call. This is the first check
-    that a descriptor-sourced to_typespec() produces a TypeSpec whose
-    primary_key still triggers that precondition."""
+    identity is isolated into that resource's own failing result
+    (check_id="__identity__") rather than raised out of dag.validate — the
+    batch still returns a DAGValidationResult (spec item 8j §3.2). This is
+    the first check that a descriptor-sourced to_typespec() produces a
+    TypeSpec whose primary_key still triggers that precondition, now
+    surfaced through the DAG's per-resource isolation instead of a raise."""
     pkg = _load_package(
         tmp_path,
         parents=[
@@ -138,8 +139,11 @@ def test_duplicate_primary_key_raises_identity_error(tmp_path, backend_name):
     dag = pkg.to_relation_dag()
     specs = {r.name: r.to_typespec() for r in pkg.resources}
 
-    with pytest.raises(IdentityInvalidError):
-        dag.validate(specs, backend=backend_name)
+    result = dag.validate(specs, backend=backend_name)  # must not raise
+
+    assert result.passes is False
+    parents_summaries = result.results["parents"].check_summaries
+    assert _status(parents_summaries, "__identity__") == "error"
 
 
 @pytest.mark.cross_backend