feat: backend mirror of the junction limits (#329 §5, AC9)

custom_components/houseplan/junction_limits.py repeats П1-П4 for the write
barrier in websocket_api, counting per rule so an inherited violation still
round-trips, and raises JunctionLimitError with the stable code
junction_limit_<rule>.

П5 is deliberately not mirrored — it judges the rendered wall bodies, and a
second mitre/inset pipeline in Python would drift more dangerously than the
rule it guards. Optimize stays outside the check for the same reason migration
and import do: it repairs existing geometry.

test_parity_with_the_frontend_checks feeds identical fixtures to the TS
functions and to this module and demands the same verdict, so the two
implementations cannot silently diverge.

Issue: #329
User-Visible: no

custom_components/houseplan/junction_limits.py

@@ -0,0 +1,287 @@
+"""Issue #329 — wall junction limits, Python mirror of src/junction-limits.ts.
+
+The card refuses a WRITE that would ADD a junction violation; this module is
+the backend half of that contract, so a stale or hostile client cannot post
+what the editor refuses. It mirrors П1–П4 exactly, including the two legal
+incidences of П4 (a shared node and a T-joint into the middle of a foreign
+wall) and the П3 rule that measures the collinear same-thickness WALL RUN
+rather than one catalogue atom.
+
+П5 (the room keeps at least 25 cm² of interior) is deliberately NOT mirrored:
+it is a statement about the rendered wall bodies, and reproducing the mitre /
+inset pipeline in Python would be a second geometry implementation whose drift
+is a larger risk than the rule it guards. A document that violates only П5 is
+ugly, not corrupt — see docs/specs/329-junction-limits.md §5.
+"""
+
+from __future__ import annotations
+
+import math
+
+MIN_JUNCTION_ANGLE_DEG = 15.0
+MAX_JUNCTION_VALENCE = 6
+MIN_SEGMENT_LENGTH_CM = 20.0
+MIN_NODE_DISTANCE_CM = 5.0
+
+GRID_STEP_N = 1 / 240
+
+_EPS = 1e-9
+# Below this a node is ON the wall (T-joint), not near it.
+_INCIDENT_EPS = 1e-9
+
+
+class JunctionLimitError(ValueError):
+    """A write introduced a wall-junction violation (#329)."""
+
+    def __init__(self, space_id: str, rule: str, subject: str,
+                 actual: float, limit: float) -> None:
+        self.space_id = space_id
+        self.rule = rule
+        self.subject = subject
+        self.actual = actual
+        self.limit = limit
+        self.code = f"junction_limit_{rule}"
+        super().__init__(
+            f"space={space_id}; rule={rule}; subject={subject}; "
+            f"actual={actual:.12g}; limit={limit:.12g}"
+        )
+
+
+def _finite_point(point: object) -> bool:
+    return (
+        isinstance(point, (list, tuple))
+        and len(point) >= 2
+        and all(isinstance(value, (int, float)) and math.isfinite(value)
+                for value in point[:2])
+    )
+
+
+def _key(point) -> str:
+    return f"{point[0]:.6f},{point[1]:.6f}"
+
+
+def _length(a, b) -> float:
+    return math.hypot(b[0] - a[0], b[1] - a[1])
+
+
+def cm_to_units(cm: float, cell_cm: float, grid_pitch: float = GRID_STEP_N) -> float:
+    return (cm / (cell_cm or 1)) * grid_pitch
+
+
+def limit_segments(space: dict) -> list[dict]:
+    """Every wall the limits judge: contour atoms, partitions, draft segments."""
+    segments: list[dict] = []
+    for segment in space.get("wall_segments") or []:
+        if _finite_point(segment.get("a")) and _finite_point(segment.get("b")):
+            segments.append({
+                "id": str(segment.get("id") or ""),
+                "a": segment["a"], "b": segment["b"],
+                "cm": float(segment.get("cm") or 0),
+            })
+    for partition in space.get("partitions") or []:
+        if _finite_point(partition.get("a")) and _finite_point(partition.get("b")):
+            segments.append({
+                "id": str(partition.get("id") or ""),
+                "a": partition["a"], "b": partition["b"],
+                "cm": float(partition.get("cm") or 0),
+            })
+    for draft in space.get("room_drafts") or []:
+        points = draft.get("points") or []
+        drafted = draft.get("segments") or []
+        for index in range(max(len(points) - 1, 0)):
+            if not (_finite_point(points[index]) and _finite_point(points[index + 1])):
+                continue
+            piece = drafted[index] if index < len(drafted) else {}
+            segments.append({
+                "id": str((piece or {}).get("id")
+                          or f"{draft.get('id') or 'draft'}-{index}"),
+                "a": points[index], "b": points[index + 1],
+                "cm": float((piece or {}).get("cm") or 0),
+            })
+    return [segment for segment in segments
+            if _length(segment["a"], segment["b"]) > _EPS]
+
+
+def check_nodes(segments: list[dict]) -> list[tuple[str, str, float, float]]:
+    """П1 + П2: valence of a node and the narrowest wedge in it."""
+    rays: dict[str, list[float]] = {}
+    for segment in segments:
+        for start, end in ((segment["a"], segment["b"]), (segment["b"], segment["a"])):
+            rays.setdefault(_key(start), []).append(
+                math.atan2(end[1] - start[1], end[0] - start[0])
+            )
+    violations: list[tuple[str, str, float, float]] = []
+    for node, angles in rays.items():
+        if len(angles) > MAX_JUNCTION_VALENCE:
+            violations.append(
+                ("valence", node, float(len(angles)), float(MAX_JUNCTION_VALENCE))
+            )
+        if len(angles) < 2:
+            continue
+        ordered = sorted(angles)
+        smallest = math.inf
+        for index, value in enumerate(ordered):
+            delta = ordered[(index + 1) % len(ordered)] - value
+            if index == len(ordered) - 1:
+                delta += math.pi * 2
+            degrees = (delta * 180) / math.pi
+            # Collinear rays of one straight wall are a 180° pair, not a wedge.
+            if _EPS < degrees < smallest:
+                smallest = degrees
+        if smallest < MIN_JUNCTION_ANGLE_DEG - 1e-9:
+            violations.append(("angle", node, smallest, MIN_JUNCTION_ANGLE_DEG))
+    return violations
+
+
+def _axis_degrees(segment: dict) -> float:
+    degrees = math.degrees(math.atan2(
+        segment["b"][1] - segment["a"][1], segment["b"][0] - segment["a"][0]
+    ))
+    return (degrees % 180 + 180) % 180
+
+
+def _collinear(left: dict, right: dict, tolerance_deg: float = 1.0) -> bool:
+    delta = abs(_axis_degrees(left) - _axis_degrees(right))
+    return min(delta, 180 - delta) <= tolerance_deg
+
+
+def collinear_run_length_units(segment: dict, segments: list[dict]) -> float:
+    """Length of the WALL a segment belongs to, not of the atom.
+
+    Atomisation at a thickness step leaves pieces nobody drew — where a 30 cm
+    wall meets a 20 cm one, a (30−20)/2 = 5 cm piece continues the same wall.
+    They are collinear continuations at the same thickness, so П3 measures the
+    maximal collinear chain through the segment's nodes.
+    """
+    by_node: dict[str, list[dict]] = {}
+    for item in segments:
+        for point in (item["a"], item["b"]):
+            by_node.setdefault(_key(point), []).append(item)
+    visited = [segment]
+    total = _length(segment["a"], segment["b"])
+
+    def walk(current: dict, node: str) -> None:
+        nonlocal total
+        for candidate in by_node.get(node, []):
+            if any(candidate is seen for seen in visited):
+                continue
+            if not _collinear(candidate, current):
+                continue
+            if float(candidate.get("cm") or 0) != float(current.get("cm") or 0):
+                continue
+            visited.append(candidate)
+            total += _length(candidate["a"], candidate["b"])
+            walk(candidate, _key(candidate["b"])
+                 if _key(candidate["a"]) == node else _key(candidate["a"]))
+            return
+
+    walk(segment, _key(segment["a"]))
+    walk(segment, _key(segment["b"]))
+    return total
+
+
+def check_segment_lengths(
+    segments: list[dict], cell_cm: float, grid_pitch: float = GRID_STEP_N,
+) -> list[tuple[str, str, float, float]]:
+    """П3: a wall is at least 20 cm and never shorter than its own thickness."""
+    violations: list[tuple[str, str, float, float]] = []
+    for segment in segments:
+        units = collinear_run_length_units(segment, segments)
+        cm = (units / grid_pitch) * (cell_cm or 1)
+        thickness = float(segment.get("cm") or 0)
+        limit = max(MIN_SEGMENT_LENGTH_CM, thickness if thickness > 0 else 0.0)
+        if cm < limit - 1e-9:
+            violations.append((
+                "length", str(segment.get("id") or _key(segment["a"])), cm, limit,
+            ))
+    return violations
+
+
+def _distance_to_segment(point, a, b) -> float:
+    dx, dy = b[0] - a[0], b[1] - a[1]
+    length_sq = dx * dx + dy * dy
+    t = 0.0 if length_sq <= _EPS else max(0.0, min(1.0, (
+        (point[0] - a[0]) * dx + (point[1] - a[1]) * dy
+    ) / length_sq))
+    return math.hypot(point[0] - (a[0] + dx * t), point[1] - (a[1] + dy * t))
+
+
+def check_node_distances(
+    segments: list[dict], cell_cm: float, grid_pitch: float = GRID_STEP_N,
+) -> list[tuple[str, str, float, float]]:
+    """П4: non-incident nodes and node-to-foreign-wall clearance."""
+    nodes: dict[str, list] = {}
+    for segment in segments:
+        nodes[_key(segment["a"])] = segment["a"]
+        nodes[_key(segment["b"])] = segment["b"]
+    min_units = cm_to_units(MIN_NODE_DISTANCE_CM, cell_cm, grid_pitch)
+    violations: list[tuple[str, str, float, float]] = []
+    entries = list(nodes.items())
+    for i in range(len(entries)):
+        for j in range(i + 1, len(entries)):
+            distance = _length(entries[i][1], entries[j][1])
+            if distance < min_units - 1e-9:
+                violations.append((
+                    "distance", f"{entries[i][0]} ↔ {entries[j][0]}",
+                    (distance / grid_pitch) * (cell_cm or 1), MIN_NODE_DISTANCE_CM,
+                ))
+    for node_key, node in nodes.items():
+        for segment in segments:
+            # A node that belongs to the wall is a legal corner or T-joint.
+            if _key(segment["a"]) == node_key or _key(segment["b"]) == node_key:
+                continue
+            distance = _distance_to_segment(node, segment["a"], segment["b"])
+            # Sitting exactly ON the wall is the other legal incidence.
+            if distance <= _INCIDENT_EPS:
+                continue
+            if distance < min_units - 1e-9:
+                violations.append((
+                    "distance",
+                    f"{node_key} → {segment.get('id') or _key(segment['a'])}",
+                    (distance / grid_pitch) * (cell_cm or 1), MIN_NODE_DISTANCE_CM,
+                ))
+    return violations
+
+
+def space_violations(space: dict) -> list[tuple[str, str, float, float]]:
+    """П1–П4 over one space, in the frontend's order."""
+    segments = limit_segments(space)
+    cell_cm = float(space.get("cell_cm") or 1)
+    return [
+        *check_nodes(segments),
+        *check_segment_lengths(segments, cell_cm),
+        *check_node_distances(segments, cell_cm),
+    ]
+
+
+def validate_junction_limits(config: dict, previous: dict | None = None) -> None:
+    """Refuse a write that ADDS a junction violation; inherit the rest.
+
+    Counted per RULE, not per subject: a structural write re-atomises and
+    re-keys contour segments, so subject identity is not stable across the
+    barrier and matching by it would report an inherited violation as new
+    (the mistake that refused legitimate resizes on the frontend). Spec §3:
+    a write may keep existing violations, it may never add one.
+    """
+    old_spaces = {
+        str(space.get("id")): space
+        for space in (previous or {}).get("spaces") or []
+    }
+    for space in config.get("spaces") or []:
+        space_id = str(space.get("id", ""))
+        old_space = old_spaces.get(space_id)
+        if old_space is None:
+            # A brand-new space has nothing to inherit from — but neither is a
+            # first write allowed to arrive already broken.
+            before: dict[str, int] = {}
+        else:
+            before = {}
+            for rule, _subject, _actual, _limit in space_violations(old_space):
+                before[rule] = before.get(rule, 0) + 1
+        after: dict[str, list[tuple[str, str, float, float]]] = {}
+        for violation in space_violations(space):
+            after.setdefault(violation[0], []).append(violation)
+        for rule, items in after.items():
+            if len(items) > before.get(rule, 0):
+                _, subject, actual, limit = items[0]
+                raise JunctionLimitError(space_id, rule, subject, actual, limit)

custom_components/houseplan/websocket_api.py

@@ -66,6 +66,7 @@ from .wall_segment_model import (
     WallSegmentMigrationError,
     commit_wall_segment_model,
 )
+from .junction_limits import JunctionLimitError, validate_junction_limits
 from .validation import (
     CONFIG_SCHEMA, LAYOUT_SCHEMA, MAX_CONFIG_BYTES, MAX_PLAN_BYTES,
     PLAN_EXTENSIONS, POS_SCHEMA, MarkerControlError, OpeningPassageError,
@@ -1330,9 +1331,11 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
             validate_marker_value_badges(msg["config"], data.get("config"))
             validate_opening_passages(msg["config"], data.get("config"))
             validate_partition_opening_hosts(msg["config"], data.get("config"))
+            validate_junction_limits(msg["config"], data.get("config"))
         except (
-            MarkerControlError, OpeningPassageError, PartitionOpeningHostError,
-            PartitionOpeningJambMarginError, WallModelClientOutdatedError,
+            JunctionLimitError, MarkerControlError, OpeningPassageError,
+            PartitionOpeningHostError, PartitionOpeningJambMarginError,
+            WallModelClientOutdatedError,
         ) as err:
             connection.send_error(msg["id"], err.code, str(err))
             return
@@ -1679,8 +1682,9 @@ async def ws_plan_optimize(hass: HomeAssistant, connection, msg: dict[str, Any])
                 allow_optimize_rehost=True,
             )
         except (
-            MarkerControlError, OpeningPassageError, PartitionOpeningHostError,
-            PartitionOpeningJambMarginError, WallModelClientOutdatedError,
+            JunctionLimitError, MarkerControlError, OpeningPassageError,
+            PartitionOpeningHostError, PartitionOpeningJambMarginError,
+            WallModelClientOutdatedError,
             WallSegmentMigrationError,
         ) as err:
             connection.send_error(msg["id"], err.code, str(err))

docs/images/screenshots.json

@@ -2,7 +2,7 @@
   "version": 1,
   "fixture": "synthetic-only",
   "chromium": "151.0.7922.34",
-  "sourceFingerprint": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+  "sourceFingerprint": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
   "captureScriptSha256": "ce2e9542fed9dade3085be87d16f69adb2ac8262893ad78ad966b1b9673f2983",
   "command": "npm run build && node demo/docs/capture.mjs",
   "scenarios": {
@@ -14,7 +14,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "0ca982495a7749907c48ebbddfd089c54385a4347f79f64e94af2c5a8d5c03eb"
     },
     "view-touch": {
@@ -25,7 +25,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "1c8e28ff34736b936ba7c9ec047d1da27aaf6fbb01411155f2d5c05282c02c41"
     },
     "space-create": {
@@ -36,7 +36,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "a55b463055a565892ca139014e314a8f63fb5301c87fff2e48d7e69210fe9b6c"
     },
     "room-contour-close": {
@@ -47,7 +47,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "dfad9a27371cfa397a3fd8dce7005f42d24649743f5322105ee894067d3a9f16"
     },
     "plan-context-tray": {
@@ -58,7 +58,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "b75f348c632296da5c2fc142d53df613f757e1516acac315be50f5b4557729cc"
     },
     "device-editor": {
@@ -69,7 +69,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "6b9cd281450e7782cff292f11efc58d227bedd0df86afc3be987feb2edec9dc3"
     },
     "device-display-preview": {
@@ -80,7 +80,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "6f628abe6b89d7c4b352e6d73ac0b126526f6fdba5927bb774c5b884e484ca19"
     },
     "background-editor": {
@@ -91,7 +91,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "a1d1a905f1410f73231d2ea5d1dfbd8ae79c2fe8b0a79cff87ebb977ac7a3103"
     },
     "room-card": {
@@ -102,7 +102,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "48a5685039a5e0de2a28190d857b193c86e2e63606e3c7a8c40fc9d2b190da4e"
     },
     "device-info": {
@@ -113,7 +113,7 @@
       },
       "theme": "dark",
       "language": "en",
-      "sourceSha256": "d7dbfe2be5a7f7d20a59f281ad796428717854c7fcf8972568826c79f206cb99",
+      "sourceSha256": "eb7e35d0b47e9c3c3b60b0ddbca2e20a62e5eb1e9b516c4394b7617d72495502",
       "imageSha256": "29a342811aaedaebc54ba337792d60cb454a7cf4c3063b210f6c4bafe0a4d87b"
     }
   }

docs/specs/329-junction-limits.md

@@ -97,11 +97,25 @@
 ## 5. Бэкенд
 
 Семантическая дельта-валидация по образцу существующих (например, jamb
-margin): правила П1–П5 применяются к элементам, которых касается дельта
-записи (новые/изменённые сегменты, узлы, комнаты); унаследованные нарушения
-нетронутых элементов проходят. Ошибки — стабильные коды
-`junction_limit_<rule>` в общем формате `vol.Invalid`/семантических ошибок.
-Питоновское зеркало обязано давать решения, идентичные фронту.
+margin), модуль `custom_components/houseplan/junction_limits.py`. Считается
+так же, как на фронте: нарушения по ПРАВИЛУ, а не по носителю (структурная
+запись переатомизирует сегменты и меняет их id), унаследованные проходят,
+новое — отказ. Ошибка — `JunctionLimitError` со стабильным кодом
+`junction_limit_<rule>`, отдаётся тем же `connection.send_error`, что и
+остальные семантические ошибки записи конфига.
+
+Зеркалятся П1–П4. П5 (просвет комнаты) сознательно оставлен только на
+клиенте: это утверждение о ПОСТРОЕННЫХ телах стен, и повторение конвейера
+mitre/inset на Python было бы второй реализацией геометрии, расхождение
+которой опаснее самого правила. Документ, нарушающий только П5, некрасив, но
+не повреждён. Паритет П1–П4 закреплён тестом
+`tests_backend/test_junction_limits.py::test_parity_with_the_frontend_checks`:
+одни и те же фикстуры прогоняются через TS-функции и через питоновские, и
+вердикт обязан совпасть.
+
+Оптимизация планов проверку не проходит — это ремонтный путь того же класса,
+что миграция и импорт (§3): он приводит существующую геометрию в порядок и не
+должен упираться в правила, написанные для новых записей.
 
 ## 6. Затронутые поверхности и артефакты
 
@@ -112,8 +126,10 @@ margin): правила П1–П5 применяются к элементам,
   `junction.limit_angle`, `junction.limit_valence`, `junction.limit_length`,
   `junction.limit_distance`, `junction.limit_clearance` (с параметрами
   значений) + заголовок отказа.
-- `custom_components/houseplan/validation.py` — дельта-правила §5;
-  `tests_backend` — кейсы принятия унаследованного и отказа нового.
+- `custom_components/houseplan/junction_limits.py` — дельта-правила §5,
+  вызов в `websocket_api.py` рядом с остальными семантическими валидаторами;
+  `tests_backend/test_junction_limits.py` — границы правил, принятие
+  унаследованного, отказ нового и паритет с фронтом.
 - Тесты фронта: юниты на каждую формулу П1–П5 (границы: 14.9°/15°, 6/7 стен,
   19/20 см, 4/5 см, пустой/непустой просвет), мутанты в гейте на П1 и §4;
   смоки каналов отказа: текст объяснения ручки Resize и текст тоста
@@ -190,7 +206,9 @@ margin): правила П1–П5 применяются к элементам,
   #313). План байт-неизменен. Смок проверяет текст тоста.
 - **AC8.** Миграция/импорт/restore фикстуры с нарушениями проходят.
 - **AC9 (§5).** Бэкенд: эхо-запись унаследованного нарушения проходит;
-  запись, добавляющая новое нарушение, отклоняется стабильным кодом.
+  запись, добавляющая новое нарушение, отклоняется кодом
+  `junction_limit_<rule>`; вердикты П1–П4 совпадают с фронтом на общих
+  фикстурах (тест паритета).
 - **AC10.** Идемпотентность: повторный commit валидного конфига байт-иден-
   тичен; Optimize на легаси-фикстуре не создаёт новых нарушений (AC-пост-
   условие §3).

test/golden-matrix.test.mjs

@@ -367,7 +367,7 @@ test('sun-ray golden requires browser-painted light from a state-only sun entity
   assert.ok(scenario);
   const fixture = prepareGoldenFixture(scenario);
   const space = fixture.config.spaces.find((item) => item.id === scenario.space);
-  assert.equal(GOLDEN_MATRIX_VERSION, 47);
+  assert.equal(GOLDEN_MATRIX_VERSION, 48);
   assert.equal(space.settings.sun_rays, true);
   assert.equal(scenario.northDeg, 90,
     'the sign-sensitive golden must keep a non-zero north direction');

tests_backend/test_junction_limits.py

@@ -0,0 +1,189 @@
+"""Issue #329: the backend mirror of the wall-junction limits.
+
+Loaded by path, like the other pure backend tests, so Home Assistant is not
+needed. The parity test at the bottom is the important one: it feeds the same
+fixtures to the TypeScript checks and to this module and demands the same
+verdict, because two implementations of one rule are worth nothing if they can
+disagree.
+"""
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import types
+
+import pytest
+
+_ROOT = os.path.dirname(os.path.dirname(__file__))
+_PACKAGE_ROOT = os.path.join(_ROOT, "custom_components")
+_HOUSEPLAN_ROOT = os.path.join(_PACKAGE_ROOT, "houseplan")
+
+if "custom_components" not in sys.modules:
+    package = types.ModuleType("custom_components")
+    package.__path__ = [_PACKAGE_ROOT]
+    sys.modules["custom_components"] = package
+if "custom_components.houseplan" not in sys.modules:
+    package = types.ModuleType("custom_components.houseplan")
+    package.__path__ = [_HOUSEPLAN_ROOT]
+    sys.modules["custom_components.houseplan"] = package
+
+_PATH = os.path.join(_HOUSEPLAN_ROOT, "junction_limits.py")
+_spec = importlib.util.spec_from_file_location("hp_junction_limits", _PATH)
+jl = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(jl)
+
+CELL = 5.0
+
+
+def cm(value):
+    return jl.cm_to_units(value, CELL)
+
+
+def space(segments, space_id="s", cell_cm=CELL):
+    return {
+        "id": space_id,
+        "cell_cm": cell_cm,
+        "rooms": [],
+        "wall_segments": [
+            {"id": f"w{index}", "a": list(a), "b": list(b), "cm": thickness}
+            for index, (a, b, thickness) in enumerate(segments)
+        ],
+        "partitions": [],
+        "room_drafts": [],
+        "openings": [],
+    }
+
+
+def ray(degrees, length_cm=100.0):
+    radians = degrees * 3.141592653589793 / 180
+    import math
+    return ([0.0, 0.0], [math.cos(radians) * cm(length_cm),
+                         math.sin(radians) * cm(length_cm)])
+
+
+def rules(sp):
+    return sorted({item[0] for item in jl.space_violations(sp)})
+
+
+def test_min_angle_is_a_hard_15_degrees():
+    below = space([(*ray(0), 15), (*ray(14), 15)])
+    at = space([(*ray(0), 15), (*ray(15), 15)])
+    assert "angle" in rules(below)
+    assert "angle" not in rules(at)
+
+
+def test_valence_allows_six_walls_and_refuses_the_seventh():
+    six = space([(*ray(degree), 15) for degree in (0, 60, 120, 180, 240, 300)])
+    seven = space([(*ray(degree), 15) for degree in (0, 51, 102, 153, 204, 255, 306)])
+    assert "valence" not in rules(six)
+    assert "valence" in rules(seven)
+
+
+def test_length_measures_the_wall_run_not_the_atom():
+    # Two collinear 15 cm atoms of 10 cm each are ONE 20 cm wall — legal,
+    # exactly the thickness-step compensation case from the owner's fixture.
+    run = space([
+        ([0.0, 0.0], [cm(10), 0.0], 15),
+        ([cm(10), 0.0], [cm(20), 0.0], 15),
+    ])
+    assert "length" not in rules(run)
+    # A lone 19 cm wall is not.
+    short = space([([0.0, 0.0], [cm(19), 0.0], 15)])
+    assert "length" in rules(short)
+    # A wall may never be shorter than its own thickness.
+    thick = space([([0.0, 0.0], [cm(25), 0.0], 30)])
+    assert "length" in rules(thick)
+
+
+def test_distance_keeps_the_t_joint_legal():
+    near = space([
+        ([0.0, 0.0], [cm(300), 0.0], 15),
+        ([0.0, cm(4)], [cm(300), cm(4)], 15),
+    ])
+    assert "distance" in rules(near)
+    tee = space([
+        ([0.0, 0.0], [cm(300), 0.0], 15),
+        ([cm(150), 0.0], [cm(150), cm(300)], 15),
+    ])
+    assert "distance" not in rules(tee)
+
+
+def test_inherited_violation_does_not_block_an_unrelated_edit():
+    broken = space([(*ray(0), 15), (*ray(9), 15)])
+    previous = {"spaces": [broken]}
+    # The same broken space plus an unrelated, perfectly legal wall.
+    edited = json.loads(json.dumps(broken))
+    edited["wall_segments"].append({
+        "id": "unrelated", "a": [cm(1000), cm(1000)],
+        "b": [cm(1000), cm(1300)], "cm": 15,
+    })
+    jl.validate_junction_limits({"spaces": [edited]}, previous)
+
+
+def test_a_write_that_adds_a_violation_is_refused_with_a_stable_code():
+    clean = space([
+        ([0.0, 0.0], [cm(300), 0.0], 15),
+        ([cm(300), 0.0], [cm(300), cm(300)], 15),
+    ])
+    previous = {"spaces": [clean]}
+    broken = json.loads(json.dumps(clean))
+    broken["wall_segments"].append({
+        "id": "spike", "a": [0.0, 0.0],
+        "b": [cm(300), cm(300) * 0.15], "cm": 15,
+    })
+    with pytest.raises(jl.JunctionLimitError) as excinfo:
+        jl.validate_junction_limits({"spaces": [broken]}, previous)
+    assert excinfo.value.code == "junction_limit_angle"
+    assert excinfo.value.space_id == "s"
+
+
+def test_parity_with_the_frontend_checks():
+    """The same fixtures must get the same verdict on both sides."""
+    fixtures = {
+        "angle-14": [(*ray(0), 15), (*ray(14), 15)],
+        "angle-15": [(*ray(0), 15), (*ray(15), 15)],
+        "valence-7": [(*ray(degree), 15)
+                      for degree in (0, 51, 102, 153, 204, 255, 306)],
+        "length-19": [([0.0, 0.0], [cm(19), 0.0], 15)],
+        "length-run": [([0.0, 0.0], [cm(10), 0.0], 15),
+                       ([cm(10), 0.0], [cm(20), 0.0], 15)],
+        "length-thickness": [([0.0, 0.0], [cm(25), 0.0], 30)],
+        "distance-4": [([0.0, 0.0], [cm(300), 0.0], 15),
+                       ([0.0, cm(4)], [cm(300), cm(4)], 15)],
+        "tee": [([0.0, 0.0], [cm(300), 0.0], 15),
+                ([cm(150), 0.0], [cm(150), cm(300)], 15)],
+    }
+    payload = {name: space(segments, space_id=name)
+               for name, segments in fixtures.items()}
+    mine = {name: rules(sp) for name, sp in payload.items()}
+
+    script = """
+import { checkNodes, checkSegmentLengths, checkNodeDistances } from './test-build/junction-limits.js';
+import { GRID_STEP_N } from './test-build/space-geometry.js';
+let raw = '';
+process.stdin.on('data', (chunk) => { raw += chunk; });
+process.stdin.on('end', () => {
+  const spaces = JSON.parse(raw);
+  const out = {};
+  for (const [name, space] of Object.entries(spaces)) {
+    const segments = space.wall_segments;
+    const violations = [
+      ...checkNodes(segments),
+      ...checkSegmentLengths(segments, space.cell_cm, GRID_STEP_N),
+      ...checkNodeDistances(segments, space.cell_cm, GRID_STEP_N),
+    ];
+    out[name] = [...new Set(violations.map((item) => item.rule))].sort();
+  }
+  process.stdout.write(JSON.stringify(out));
+});
+"""
+    if not os.path.isdir(os.path.join(_ROOT, "test-build")):
+        pytest.skip("test-build/ is not compiled; run npx tsc -p tsconfig.test.json")
+    result = subprocess.run(
+        ["node", "--input-type=module", "--eval", script],
+        cwd=_ROOT, input=json.dumps(payload), capture_output=True, text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+    assert json.loads(result.stdout) == mine