Read parallelism and config_summary from node_types

src/submission_checker/checker.py

@@ -8,7 +8,7 @@
 
 import json
 from dataclasses import dataclass, replace
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import yaml
 
@@ -185,8 +185,24 @@ class _SystemFacts:
 
 
 def _system_desc_identity(data: dict[str, object]) -> dict[str, object]:
-    """Strip the fields that legitimately vary between points of the same frontier."""
-    return {k: v for k, v in data.items() if k not in _PER_POINT_SYSTEM_DESC_FIELDS}
+    """Strip the fields that legitimately vary between points of the same frontier.
+
+    §8.2.1 moved the parallelism fields and ``config_summary`` inside ``node_types``,
+    so stripping top-level keys alone no longer reaches them: a frontier that tunes
+    its replica size per point — which §4.5.3's C.3 example does — would differ on
+    ``node_types`` and fail system-description-consistency. They are stripped at both
+    depths, so a bundle in either placement compares on the same identity.
+    """
+    identity = {k: v for k, v in data.items() if k not in _PER_POINT_SYSTEM_DESC_FIELDS}
+    node_types = identity.get("node_types")
+    if isinstance(node_types, list):
+        identity["node_types"] = [
+            {k: v for k, v in node.items() if k not in _PER_POINT_SYSTEM_DESC_FIELDS}
+            if isinstance(node, dict)
+            else node
+            for node in node_types
+        ]
+    return identity
 
 
 def _read_json(path: Path) -> dict[str, object] | None:
@@ -268,32 +284,80 @@ def _declared_tps_utilization(system_desc_path: Path) -> float | None:
 def _point_parallelism(system_desc_path: Path) -> Parallelism | None:
     """A point's parallelism from its own ``system_desc.json``, or None if it declares none.
 
-    §8.2 places the fields at the top level; the structured ``config_summary`` object
-    is read as a fallback. A field left out is 1 — §8.2 defines "TP=1 means no
-    partitioning" and likewise for the others — but a description with none of them
-    says nothing about its deployment, so it yields ``None`` rather than a guess.
+    Two placements are accepted. §8.2.1 now declares the fields inside each
+    ``node_types`` entry, because §4.5.3's node-count declaration is per node type and
+    has to be consistent with them; older bundles carry them at the top level, with the
+    structured ``config_summary`` object as a further fallback. The node-type placement
+    wins where both appear, being the current format.
+
+    A field left out is 1 — §8.2 defines "TP=1 means no partitioning" and likewise for
+    the others — but a description with none of them says nothing about its deployment,
+    so it yields ``None`` rather than a guess.
     """
     data = _read_json(system_desc_path)
     if data is None:
         return None
     summary = data.get("config_summary")
     summary = summary if isinstance(summary, dict) else {}
+    node_types = [n for n in (data.get("node_types") or []) if isinstance(n, dict)]
+
     values: dict[str, int] = {}
     for name in _PARALLELISM_FIELDS:
+        declared = _declared_ints(node_types, name)
+        if len(set(declared)) > 1:
+            # §4.5.3 computes one A_replica = TP·PP·EP for the whole SUT, so a Parallelism
+            # cannot carry two values for a field. Rather than pick one and normalise
+            # against a replica size no node type has, decline: nodes-used is skipped,
+            # exactly as it already is for a bundle that declares no parallelism at all.
+            # Making §4.5.3's arithmetic per node type is the real fix and is its own job.
+            return None
+        if declared:
+            values[name] = declared[0]
+            continue
         value = data.get(name, summary.get(name))
         if isinstance(value, int) and not isinstance(value, bool):
             values[name] = value
     if not values:
         return None
     # §8.2: "Indicates whether the system is disaggregated (disaggregated > 1)", while
-    # the model reads it as a bool; either spelling of "yes" counts.
-    flag = data.get("disaggregated", summary.get("disaggregated"))
-    disaggregated = flag is True or (
-        isinstance(flag, int) and not isinstance(flag, bool) and flag > 1
+    # the model reads it as a bool; either spelling of "yes" counts. Any node type being
+    # disaggregated disaggregates the SUT.
+    flags = [
+        _node_value(n, "disaggregated")
+        for n in node_types
+        if _node_value(n, "disaggregated") is not None
+    ]
+    if not flags:
+        flags = [data.get("disaggregated", summary.get("disaggregated"))]
+    disaggregated = any(
+        flag is True or (isinstance(flag, int) and not isinstance(flag, bool) and flag > 1)
+        for flag in flags
     )
     return Parallelism(**values, disaggregated=disaggregated)
 
 
+def _node_value(node: dict[str, Any], name: str) -> object:
+    """*name* on a node type, or from that node type's own ``config_summary``.
+
+    §8.2.1 keeps ``config_summary`` beside the parallelism fields it concatenates, so
+    its object form is read per node type exactly as it used to be read SUT-wide.
+    """
+    if name in node:
+        return node[name]
+    summary = node.get("config_summary")
+    return summary.get(name) if isinstance(summary, dict) else None
+
+
+def _declared_ints(node_types: list[dict[str, Any]], name: str) -> list[int]:
+    """*name* as declared by each node type that declares it, in order."""
+    found = []
+    for node in node_types:
+        value = _node_value(node, name)
+        if isinstance(value, int) and not isinstance(value, bool):
+            found.append(value)
+    return found
+
+
 def _resolve_submission_root(path: Path) -> Path:
     """Return the ``<submission_id>/`` level, descending from an org root if needed.
 

src/submission_checker/models/file/system.py

@@ -77,6 +77,27 @@ class AcceleratorInfo(BaseModel):
     accelerator_host_interconnect: str | None = None
 
 
+class ConfigSummary(BaseModel):
+    """Structured parallelism/config summary — the object form of ``config_summary``.
+
+    Moved here from the retired ``run_metadata.json`` model: policies PR #119 deleted
+    that file and §8.2 absorbed its fields into the system description.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    disaggregated: bool | None = None
+    expert_parallel: int | None = None
+    tensor_parallel: int | None = None
+    pipeline_parallel: int | None = None
+    data_parallel: int | None = None
+    batch: int | None = None
+
+
+#: ``config_summary`` accepts either the structured object or free-form prose.
+_ConfigSummaryStr = Annotated[str, StringConstraints(min_length=4)]
+
+
 class NodeType(BaseModel):
     """Per-node hardware and software configuration (§8.2.1)."""
 
@@ -133,33 +154,24 @@ def _lift_flat_accelerator_fields(cls, data: object) -> object:
             data = {**data, "accelerator_info": [flat]}
         return data
 
-    @model_validator(mode="after")
-    def _require_core_or_vcpu_count(self) -> NodeType:
-        """A node must disclose at least one of physical core count or vCPU count."""
-        if self.host_processor_core_count is None and self.host_processor_vcpu_count is None:
-            raise Invalid("system-description-valid", "core-count-missing")
-        return self
-
-
-class ConfigSummary(BaseModel):
-    """Structured parallelism/config summary — the object form of ``config_summary``.
-
-    Moved here from the retired ``run_metadata.json`` model: policies PR #119 deleted
-    that file and §8.2 absorbed its fields into the system description.
-    """
-
-    model_config = ConfigDict(extra="allow")
-
+    # §8.2.1 declares the deployment configuration per node type: a heterogeneous SUT
+    # may partition differently on each, and §4.5.3's node-count declaration is per
+    # node type and must agree with these. Older bundles carry them at the top level
+    # of the system description instead; both placements are read.
     disaggregated: bool | None = None
     expert_parallel: int | None = None
     tensor_parallel: int | None = None
     pipeline_parallel: int | None = None
     data_parallel: int | None = None
-    batch: int | None = None
+    config_summary: ConfigSummary | _ConfigSummaryStr | None = None
 
+    @model_validator(mode="after")
+    def _require_core_or_vcpu_count(self) -> NodeType:
+        """A node must disclose at least one of physical core count or vCPU count."""
+        if self.host_processor_core_count is None and self.host_processor_vcpu_count is None:
+            raise Invalid("system-description-valid", "core-count-missing")
+        return self
 
-#: ``config_summary`` accepts either the structured object or free-form prose.
-_ConfigSummaryStr = Annotated[str, StringConstraints(min_length=4)]
 
 #: §8.2's availability field first, then the legacy spellings still read from older
 #: bundles: ``publication_status`` (written by earlier builds of this CLI) and

tests/submission_checker/test_checker.py

@@ -4,13 +4,18 @@
 
 import json
 from pathlib import Path
+from typing import Any
 from unittest.mock import patch
 
 import pytest
 import yaml
 
 from submission_checker import layout
-from submission_checker.checker import SubmissionChecker
+from submission_checker.checker import (
+    SubmissionChecker,
+    _point_parallelism,
+    _system_desc_identity,
+)
 from submission_checker.messages import Invalid
 from submission_checker.models import CheckResult, Report, Severity
 
@@ -849,3 +854,164 @@ def test_fixture_error_profile(request, fixture_name, expected):
         f"{fixture_name}: unexpected {sorted(actual - expected)}, "
         f"missing {sorted(expected - actual)}"
     )
+
+
+@pytest.mark.unit
+class TestParallelismPlacement:
+    """§8.2.1 declares the parallelism fields inside each ``node_types`` entry.
+
+    Older bundles carry them at the top level, with ``config_summary`` as a further
+    fallback, and both have to keep working.
+    """
+
+    @staticmethod
+    def _desc(tmp_path: Path, **top: Any) -> Path:
+        path = tmp_path / "system_desc.json"
+        path.write_text(json.dumps({"node_types": [], **top}))
+        return path
+
+    def test_node_types_placement_is_read(self, tmp_path: Path) -> None:
+        path = self._desc(
+            tmp_path,
+            node_types=[{"tensor_parallel": 8, "data_parallel": 2, "disaggregated": 0}],
+        )
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 8
+        assert para.data_parallel == 2
+        assert para.disaggregated is False
+        # §4.5.3: A_replica = TP·PP·EP, A_used = DP·A_replica.
+        assert para.replica == 8
+        assert para.accelerators_used == 16
+
+    def test_top_level_placement_still_works(self, tmp_path: Path) -> None:
+        path = self._desc(tmp_path, tensor_parallel=8, data_parallel=2)
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 8
+        assert para.accelerators_used == 16
+
+    def test_config_summary_is_still_the_last_fallback(self, tmp_path: Path) -> None:
+        path = self._desc(tmp_path, config_summary={"tensor_parallel": 4})
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 4
+
+    def test_node_types_wins_over_the_top_level(self, tmp_path: Path) -> None:
+        """The node-type placement is the current format, so it is the one believed."""
+        path = self._desc(
+            tmp_path,
+            tensor_parallel=2,
+            config_summary={"tensor_parallel": 3},
+            node_types=[{"tensor_parallel": 8}],
+        )
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 8
+
+    def test_the_two_placements_mix_per_field(self, tmp_path: Path) -> None:
+        """A bundle part-way through the move still resolves every field."""
+        path = self._desc(tmp_path, data_parallel=2, node_types=[{"tensor_parallel": 8}])
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert (para.tensor_parallel, para.data_parallel) == (8, 2)
+
+    def test_agreeing_node_types_resolve(self, tmp_path: Path) -> None:
+        path = self._desc(
+            tmp_path,
+            node_types=[{"tensor_parallel": 4}, {"tensor_parallel": 4}],
+        )
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 4
+
+    def test_disagreeing_node_types_decline_rather_than_pick(self, tmp_path: Path) -> None:
+        """§4.5.3 computes one A_replica for the whole SUT, so there is no honest
+        single answer. Normalising against a replica size no node type has would be
+        worse than skipping the check."""
+        path = self._desc(
+            tmp_path,
+            node_types=[{"tensor_parallel": 4}, {"tensor_parallel": 8}],
+        )
+
+        assert _point_parallelism(path) is None
+
+    def test_any_disaggregated_node_type_disaggregates_the_sut(self, tmp_path: Path) -> None:
+        path = self._desc(
+            tmp_path,
+            node_types=[
+                {"tensor_parallel": 4, "disaggregated": 0},
+                {"tensor_parallel": 4, "disaggregated": 2},
+            ],
+        )
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.disaggregated is True
+
+    def test_declaring_nothing_is_still_none(self, tmp_path: Path) -> None:
+        assert _point_parallelism(self._desc(tmp_path)) is None
+
+    def test_a_node_types_config_summary_is_read(self, tmp_path: Path) -> None:
+        """§8.2.1 keeps config_summary beside the fields it concatenates."""
+        path = self._desc(tmp_path, node_types=[{"config_summary": {"tensor_parallel": 4}}])
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 4
+
+    def test_a_node_types_own_field_beats_its_config_summary(self, tmp_path: Path) -> None:
+        path = self._desc(
+            tmp_path,
+            node_types=[{"tensor_parallel": 8, "config_summary": {"tensor_parallel": 4}}],
+        )
+
+        para = _point_parallelism(path)
+
+        assert para is not None
+        assert para.tensor_parallel == 8
+
+
+@pytest.mark.unit
+class TestSystemDescIdentity:
+    """Per-point fields are stripped at both depths, so §8.5's cross-point comparison
+    does not fire on a frontier that tunes its replica size per point.
+    """
+
+    def test_node_types_parallelism_is_not_part_of_identity(self) -> None:
+        a = {"system_name": "s", "node_types": [{"host_networking": "eth", "tensor_parallel": 4}]}
+        b = {"system_name": "s", "node_types": [{"host_networking": "eth", "tensor_parallel": 8}]}
+
+        assert _system_desc_identity(a) == _system_desc_identity(b)
+
+    def test_node_types_config_summary_is_not_part_of_identity(self) -> None:
+        a = {"node_types": [{"cooling": "air", "config_summary": "TP=4"}]}
+        b = {"node_types": [{"cooling": "air", "config_summary": "TP=8"}]}
+
+        assert _system_desc_identity(a) == _system_desc_identity(b)
+
+    def test_real_hardware_differences_still_count(self) -> None:
+        a = {"node_types": [{"cooling": "air", "tensor_parallel": 4}]}
+        b = {"node_types": [{"cooling": "liquid", "tensor_parallel": 4}]}
+
+        assert _system_desc_identity(a) != _system_desc_identity(b)
+
+    def test_a_malformed_node_types_is_passed_through(self) -> None:
+        """The identity helper must not be the thing that raises on bad input."""
+        data = {"node_types": ["not-an-object"]}
+
+        assert _system_desc_identity(data) == data