fix: validate optimize opening rehosts

Issue: #280
User-Visible: yes

custom_components/houseplan/validation.py

@@ -5,6 +5,7 @@ Kept separate so it can be covered by unit tests (only voluptuous is needed).
 from __future__ import annotations
 
 from collections import Counter
+import math
 import re
 
 import voluptuous as vol
@@ -80,10 +81,167 @@ class PartitionOpeningJambMarginError(ValueError):
 # sync with GRID_STEP_N/NORM_W in the frontend; it is a geometry scale, not a
 # user setting.
 NORMALIZED_CANVAS_CELLS = 240.0
+_DEFAULT_ROOM_WALL_CM = 15.0
+_OPTIMIZE_REHOST_EPSILON = 1e-8
+
+
+def _room_polygon(room: dict) -> list[list[float]]:
+    poly = room.get("poly")
+    if isinstance(poly, list) and len(poly) >= 3:
+        return poly
+    if all(key in room for key in ("x", "y", "w", "h")):
+        x, y = float(room["x"]), float(room["y"])
+        w, h = float(room["w"]), float(room["h"])
+        return [[x, y], [x + w, y], [x + w, y + h], [x, y + h]]
+    return []
+
+
+def _segment_metrics(a: list, b: list) -> tuple[float, float, float]:
+    dx, dy = float(b[0]) - float(a[0]), float(b[1]) - float(a[1])
+    return dx, dy, math.hypot(dx, dy)
+
+
+def _line_distance(point: list, a: list, b: list) -> float:
+    dx, dy, length = _segment_metrics(a, b)
+    if length <= _OPTIMIZE_REHOST_EPSILON:
+        return math.inf
+    return abs((float(point[0]) - float(a[0])) * dy
+               - (float(point[1]) - float(a[1])) * dx) / length
+
+
+def _projection(point: list, a: list, b: list) -> float:
+    dx, dy, length = _segment_metrics(a, b)
+    if length <= _OPTIMIZE_REHOST_EPSILON:
+        return math.inf
+    return ((float(point[0]) - float(a[0])) * dx
+            + (float(point[1]) - float(a[1])) * dy) / length
+
+
+def _segment_covers(a: list, b: list, target_a: list, target_b: list) -> bool:
+    """Return whether one exact collinear segment covers the target."""
+    _, _, length = _segment_metrics(a, b)
+    if length <= _OPTIMIZE_REHOST_EPSILON:
+        return False
+    if (_line_distance(target_a, a, b) > _OPTIMIZE_REHOST_EPSILON
+            or _line_distance(target_b, a, b) > _OPTIMIZE_REHOST_EPSILON):
+        return False
+    for point in (target_a, target_b):
+        along = _projection(point, a, b)
+        if along < -_OPTIMIZE_REHOST_EPSILON or along > length + _OPTIMIZE_REHOST_EPSILON:
+            return False
+    return True
+
+
+def _segments_overlap_on_axis(a: list, b: list, other_a: list, other_b: list) -> bool:
+    if (_line_distance(other_a, a, b) > _OPTIMIZE_REHOST_EPSILON
+            or _line_distance(other_b, a, b) > _OPTIMIZE_REHOST_EPSILON):
+        return False
+    _, _, length = _segment_metrics(a, b)
+    lo, hi = sorted((_projection(other_a, a, b), _projection(other_b, a, b)))
+    return min(length, hi) - max(0.0, lo) > _OPTIMIZE_REHOST_EPSILON
+
+
+def _angle_delta_mod_180(first: float, second: float) -> float:
+    return abs((first - second + 90.0) % 180.0 - 90.0)
+
+
+def _safe_optimize_partition_rehost(
+    space: dict, old_space: dict, opening: dict, old_opening: dict,
+) -> bool:
+    """Independently prove the exact #276 partition-to-room-wall transition."""
+    old_host = old_opening.get("host")
+    if not isinstance(old_host, dict) or old_host.get("kind") != "partition":
+        return False
+    partition_id = str(old_host.get("id", ""))
+    old_partition = next((item for item in old_space.get("partitions") or []
+                          if str(item.get("id", "")) == partition_id), None)
+    if old_partition is None or any(
+        str(item.get("id", "")) == partition_id
+        for item in space.get("partitions") or []
+    ):
+        return False
+    a, b = old_partition.get("a"), old_partition.get("b")
+    if not (isinstance(a, list) and len(a) == 2 and isinstance(b, list) and len(b) == 2):
+        return False
+    dx, dy, length = _segment_metrics(a, b)
+    if length <= _OPTIMIZE_REHOST_EPSILON:
+        return False
+
+    owners: set[str] = set()
+    for room in space.get("rooms") or []:
+        poly = _room_polygon(room)
+        if any(_segment_covers(poly[index], poly[(index + 1) % len(poly)], a, b)
+               for index in range(len(poly))):
+            owners.add(str(room.get("id", "")))
+    if len(owners) != 2:
+        return False
+    if any(_segments_overlap_on_axis(a, b, span["a"], span["b"])
+           for span in space.get("open_spans") or []):
+        return False
+
+    covering_walls = []
+    for wall in space.get("walls") or []:
+        wall_a, wall_b = wall.get("a"), wall.get("b")
+        if not (isinstance(wall_a, list) and isinstance(wall_b, list)):
+            continue
+        if _segment_covers(wall_a, wall_b, a, b):
+            covering_walls.append((_segment_metrics(wall_a, wall_b)[2], float(wall["cm"])))
+    effective_cm = min(covering_walls, key=lambda item: item[0])[1] \
+        if covering_walls else _DEFAULT_ROOM_WALL_CM
+    if effective_cm + _OPTIMIZE_REHOST_EPSILON < float(old_partition.get("cm", 0)):
+        return False
+
+    ignored = {"host", "x", "y", "angle"}
+    old_stable = {key: value for key, value in old_opening.items() if key not in ignored}
+    new_stable = {key: value for key, value in opening.items() if key not in ignored}
+    if old_stable != new_stable:
+        return False
+    try:
+        t = float(old_host["t"])
+        x, y = float(opening["x"]), float(opening["y"])
+        opening_angle = float(opening["angle"])
+        opening_length = float(opening["length"])
+    except (KeyError, TypeError, ValueError):
+        return False
+    if not (0 <= t <= 1 and opening_length > 0):
+        return False
+    expected_x, expected_y = float(a[0]) + dx * t, float(a[1]) + dy * t
+    expected_angle = math.degrees(math.atan2(dy, dx))
+    if expected_angle >= 90:
+        expected_angle -= 180
+    elif expected_angle < -90:
+        expected_angle += 180
+    if (abs(x - expected_x) > _OPTIMIZE_REHOST_EPSILON
+            or abs(y - expected_y) > _OPTIMIZE_REHOST_EPSILON
+            or _angle_delta_mod_180(opening_angle, expected_angle) > 1e-7):
+        return False
+    along = t * length
+    if (along - opening_length / 2 < -_OPTIMIZE_REHOST_EPSILON
+            or along + opening_length / 2 > length + _OPTIMIZE_REHOST_EPSILON):
+        return False
+
+    for other in space.get("openings") or []:
+        if other is opening or str(other.get("id", "")) == str(opening.get("id", "")):
+            continue
+        try:
+            center = [float(other["x"]), float(other["y"])]
+            other_angle = float(other["angle"])
+            other_length = float(other["length"])
+        except (KeyError, TypeError, ValueError):
+            continue
+        if (_line_distance(center, a, b) > _OPTIMIZE_REHOST_EPSILON
+                or _angle_delta_mod_180(other_angle, expected_angle) > 1e-7):
+            continue
+        other_along = _projection(center, a, b)
+        if (min(along + opening_length / 2, other_along + other_length / 2)
+                - max(along - opening_length / 2, other_along - other_length / 2)
+                > _OPTIMIZE_REHOST_EPSILON):
+            return False
+    return True
 
 
 def validate_partition_opening_hosts(
-    config: dict, previous: dict | None = None
+    config: dict, previous: dict | None = None, *, allow_optimize_rehost: bool = False
 ) -> None:
     """Validate hosted-opening write deltas without rejecting legacy reads.
 
@@ -114,9 +272,13 @@ def validate_partition_opening_hosts(
             opening_id = str(opening.get("id", ""))
             old = old_openings.get(opening_id)
             if old and old.get("host") is not None and opening.get("host") is None:
-                raise PartitionOpeningHostError(
-                    f"space={space_id}; opening={opening_id}; host removed"
-                )
+                if not (allow_optimize_rehost and old_space
+                        and _safe_optimize_partition_rehost(
+                            space, old_space, opening, old
+                        )):
+                    raise PartitionOpeningHostError(
+                        f"space={space_id}; opening={opening_id}; host removed"
+                    )
             host = opening.get("host")
             if host is None:
                 continue

custom_components/houseplan/websocket_api.py

@@ -1624,7 +1624,10 @@ async def ws_plan_optimize(hass: HomeAssistant, connection, msg: dict[str, Any])
             validate_marker_light_entities(msg["config"], config_data.get("config"))
             validate_marker_value_badges(msg["config"], config_data.get("config"))
             validate_opening_passages(msg["config"], config_data.get("config"))
-            validate_partition_opening_hosts(msg["config"], config_data.get("config"))
+            validate_partition_opening_hosts(
+                msg["config"], config_data.get("config"),
+                allow_optimize_rehost=True,
+            )
         except (
             MarkerControlError, OpeningPassageError, PartitionOpeningHostError,
             PartitionOpeningJambMarginError,

docs/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+- The backend now accepts a proved Optimize reconciliation of a coincident
+  partition: its opening remains on the same shared wall through Apply, reload
+  and Undo. Arbitrary host stripping by an ordinary write or crafted candidate
+  is still rejected ([#280](https://github.com/Matysh/houseplan-card/issues/280)).
 - A nearly perpendicular T-junction now preserves all physical masonry when
   its saved axis has a small valid slope. The two white wedges disappear in
   every mode without straightening stored coordinates, while deliberately

docs/CHANGELOG.ru.md

@@ -8,6 +8,11 @@
 
 ## Не выпущено
 
+- Backend теперь принимает доказанное преобразование совпадающей перегородки
+  из Optimize: её проём сохраняется на той же общей стене и корректно проходит
+  Apply/reload/Undo. Произвольное снятие host обычной записью или поддельным
+  candidate по-прежнему отклоняется
+  ([#280](https://github.com/Matysh/houseplan-card/issues/280)).
 - Почти перпендикулярный T-стык теперь сохраняет всю реальную кладку при
   небольшом допустимом уклоне оси. Два белых клина исчезают во всех режимах,
   сохранённые координаты не выпрямляются, а явно диагональная фаска остаётся

docs/CONFIG-COMPATIBILITY.md

@@ -135,6 +135,15 @@ hosted opening with an old bundle is unsupported. A missing/invalid host is not
 re-associated automatically: current renderers fail dark and Plan offers an
 explicit rebind.
 
+The sole host-removal exception is the explicit Optimize reconciliation from
+#276/#280. The server does not trust a client counter: it independently proves
+that the old partition was removed, its complete segment is a solid shared
+boundary of exactly two rooms, the replacement wall envelope is not narrower,
+the materialized centre/angle and every unrelated opening field are unchanged,
+and no new slot overlaps. This capability is enabled only by
+`houseplan/plan/optimize`; ordinary config writes and crafted candidates keep
+the fail-closed `invalid_partition_opening_host` result.
+
 New hosted openings and direct changes to `host.id`, `host.t`, `length`, host
 span or host thickness reserve a jamb at both endpoints equal to half the
 actual partition thickness. This is semantic delta validation, not a schema or

docs/TESTING.md

@@ -324,6 +324,12 @@
 - [ ] Backend rejects missing host references, out-of-range `t`, non-fitting or
       overlapping hosted openings and stale host stripping; exports round-trip
       the host. [auto: test_validation, test_ha_import_export]
+- [ ] The exact #276 Optimize candidate is shared by frontend and backend tests:
+      Python independently proves the removed partition, two-room solid wall,
+      envelope, opening identity and non-overlap; config/set and every partial
+      or mutated candidate remain rejected. Linux HA WS persists and reloads
+      the implicit opening, then Undo restores the partition and explicit host.
+      [auto: coincident-partitions, test_validation, test_ha_websocket]
 
 ## Independent-wall opening jamb margin (#186)
 

test/coincident-partitions.test.mjs

@@ -14,6 +14,10 @@ const fixture = JSON.parse(readFileSync(
   new URL('./fixtures/276-coincident-partition.json', import.meta.url),
   'utf8',
 ));
+const backendCandidate = JSON.parse(readFileSync(
+  new URL('./fixtures/280-optimize-rehost-candidate.json', import.meta.url),
+  'utf8',
+));
 const clone = (value) => structuredClone(value);
 
 const optimize = (config) => optimizePlans(config, {});
@@ -42,6 +46,11 @@ test('issue 276 reconciles the anonymized 5 cm offset fixture without moving its
   assert.equal(intervals.length, 2);
   assert.deepEqual(new Set(intervals.map((interval) => interval.cm)), new Set([20]));
   assert.equal(checkOptimizeGeometry(result.config).ok, true);
+  assert.deepEqual(
+    result.config,
+    backendCandidate,
+    'the frontend result must stay identical to the candidate proved by Python',
+  );
 
   const second = optimize(result.config);
   assert.equal(second.changed, false);

test/fixtures/280-optimize-rehost-candidate.json

@@ -0,0 +1,57 @@
+{
+  "model_version": 7,
+  "spaces": [
+    {
+      "id": "offset-shared-wall",
+      "title": "Offset shared wall",
+      "view_box": [0, 0, 1, 1],
+      "cell_cm": 5,
+      "rooms": [
+        {
+          "id": "left",
+          "name": "Left",
+          "area": null,
+          "poly": [
+            [0, 0], [0.5, 0], [0.5, 0.004166667],
+            [0.504166667, 0.004166667], [0.504166667, 0.995833333],
+            [0.5, 0.995833333], [0.5, 1], [0, 1]
+          ]
+        },
+        {
+          "id": "right",
+          "name": "Right",
+          "area": null,
+          "poly": [
+            [0.504166667, 0.004166667], [1, 0.004166667],
+            [1, 0.995833333], [0.504166667, 0.995833333]
+          ]
+        }
+      ],
+      "walls": [
+        {
+          "key": "0.504167,0.500000@1.5706",
+          "cm": 20,
+          "a": [0.504166667, 0.004166667],
+          "b": [0.504166667, 0.995833333]
+        }
+      ],
+      "openings": [
+        {
+          "id": "hosted-door",
+          "type": "door",
+          "x": 0.504166667,
+          "y": 0.5,
+          "angle": -90,
+          "length": 0.2,
+          "contact": "binary_sensor.test_door",
+          "lock": "lock.test_door",
+          "invert": true,
+          "flip_h": true,
+          "future_field": { "keep": true }
+        }
+      ]
+    }
+  ],
+  "markers": [],
+  "settings": {}
+}

tests_backend/test_ha_websocket.py

@@ -534,6 +534,53 @@ async def test_optimize_undo_restores_geometry_but_not_legacy_noisy_bits(
     assert restored_layout["lamp"]["x"] == -0.12345679
 
 
+async def test_optimize_accepts_proved_rehost_and_undo_restores_host(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """#280: the real WS boundary accepts exactly the #276 candidate."""
+    from custom_components.houseplan.store import get_data
+
+    await _setup(hass)
+    client = await hass_ws_client(hass)
+    runtime = get_data(hass)
+    fixture_dir = Path(__file__).parents[1] / "test" / "fixtures"
+    previous = json.loads(
+        (fixture_dir / "276-coincident-partition.json").read_text(encoding="utf-8")
+    )
+    candidate = json.loads(
+        (fixture_dir / "280-optimize-rehost-candidate.json").read_text(encoding="utf-8")
+    )
+    await runtime.config_store.async_save({"config": previous, "rev": 1})
+    await runtime.store.async_save({"layout": {}, "rev": 1})
+
+    await client.send_json_auto_id({
+        "type": "houseplan/plan/optimize",
+        "config": candidate,
+        "layout": {},
+        "expected_config_rev": 1,
+        "expected_layout_rev": 1,
+    })
+    optimized = await client.receive_json()
+    assert optimized["success"]
+    stored = (await runtime.config_store.async_load())["config"]
+    assert stored["spaces"][0].get("partitions") in (None, [])
+    assert "host" not in stored["spaces"][0]["openings"][0]
+
+    await client.send_json_auto_id({
+        "type": "houseplan/plan/optimize_undo",
+        "expected_config_rev": 2,
+        "expected_layout_rev": 2,
+    })
+    undone = await client.receive_json()
+    assert undone["success"] and undone["result"]["can_undo"] is False
+    restored = (await runtime.config_store.async_load())["config"]
+    restored_space = restored["spaces"][0]
+    assert restored_space["partitions"][0]["id"] == "redundant"
+    assert restored_space["openings"][0]["host"] == {
+        "kind": "partition", "id": "redundant", "t": 0.5,
+    }
+
+
 async def test_plan_optimize_persists_exact_storage_roundtrip_target(
     hass: HomeAssistant, hass_ws_client: WebSocketGenerator, monkeypatch
 ) -> None:

tests_backend/test_validation.py

@@ -4,12 +4,29 @@ import importlib.util
 import json
 import os
 import subprocess
+import sys
+import types
 
 import pytest
 import voluptuous as vol
 
+_ROOT = os.path.dirname(os.path.dirname(__file__))
+_PACKAGE_ROOT = os.path.join(_ROOT, "custom_components")
+_HOUSEPLAN_ROOT = os.path.join(_PACKAGE_ROOT, "houseplan")
+
+# Keep this pure test independent of Home Assistant even though Python normally
+# executes package __init__.py before resolving the validation submodule.
+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(
-    os.path.dirname(os.path.dirname(__file__)),
+    _ROOT,
     "custom_components", "houseplan", "validation.py",
 )
 _spec = importlib.util.spec_from_file_location("hp_validation", _PATH)
@@ -1717,6 +1734,108 @@ def test_partition_opening_host_schema_fit_overlap_and_downgrade_guard():
     v.validate_partition_opening_hosts({"spaces": [{**base, "openings": []}]}, previous)
 
 
+def test_optimize_accepts_only_proved_partition_to_room_wall_rehost():
+    root = os.path.dirname(os.path.dirname(__file__))
+    fixture_dir = os.path.join(root, "test", "fixtures")
+    with open(os.path.join(fixture_dir, "276-coincident-partition.json"),
+              encoding="utf-8") as stream:
+        previous = json.load(stream)
+    with open(os.path.join(fixture_dir, "280-optimize-rehost-candidate.json"),
+              encoding="utf-8") as stream:
+        candidate = json.load(stream)
+
+    with pytest.raises(v.PartitionOpeningHostError):
+        v.validate_partition_opening_hosts(candidate, previous)
+    v.validate_partition_opening_hosts(
+        candidate, previous, allow_optimize_rehost=True
+    )
+
+    def rejected(mutator):
+        changed = json.loads(json.dumps(candidate))
+        mutator(changed["spaces"][0])
+        with pytest.raises(v.PartitionOpeningHostError):
+            v.validate_partition_opening_hosts(
+                changed, previous, allow_optimize_rehost=True
+            )
+
+    rejected(lambda space: space.update(partitions=[{
+        "id": "redundant",
+        "a": [0.504166667, 0.004166667],
+        "b": [0.504166667, 0.995833333],
+        "cm": 20,
+    }]))
+    rejected(lambda space: space["rooms"].pop())
+    rejected(lambda space: space["walls"][0].update(cm=10))
+    rejected(lambda space: space["openings"][0].update(x=0.51))
+    rejected(lambda space: space["openings"][0].update(angle=-89))
+    rejected(lambda space: space["openings"][0].update(length=0.19))
+    rejected(lambda space: space["openings"][0].update(type="window"))
+    rejected(lambda space: space["openings"][0].update(
+        contact="binary_sensor.other"
+    ))
+    rejected(lambda space: space["openings"].append({
+        "id": "overlap", "type": "door",
+        "x": 0.504166667, "y": 0.5, "angle": -90, "length": 0.1,
+    }))
+    rejected(lambda space: space.update(open_spans=[{
+        "a": [0.504166667, 0.2], "b": [0.504166667, 0.8],
+    }]))
+
+
+def test_optimize_rehost_validation_is_atomic_across_the_batch():
+    root = os.path.dirname(os.path.dirname(__file__))
+    fixture_dir = os.path.join(root, "test", "fixtures")
+    previous = json.load(open(
+        os.path.join(fixture_dir, "276-coincident-partition.json"),
+        encoding="utf-8",
+    ))
+    candidate = json.load(open(
+        os.path.join(fixture_dir, "280-optimize-rehost-candidate.json"),
+        encoding="utf-8",
+    ))
+    old_space = previous["spaces"][0]
+    new_space = candidate["spaces"][0]
+    old_space["openings"].append({
+        **old_space["openings"][0], "id": "second", "length": 0.1,
+        "host": {"kind": "partition", "id": "redundant", "t": 0.75},
+    })
+    new_space["openings"].append({
+        **new_space["openings"][0], "id": "second", "length": 0.1,
+        "x": 0.6, "y": 0.747916667,
+    })
+    with pytest.raises(v.PartitionOpeningHostError) as raised:
+        v.validate_partition_opening_hosts(
+            candidate, previous, allow_optimize_rehost=True
+        )
+    assert "opening=second" in str(raised.value)
+
+
+def test_optimize_rehost_private_exact_fixture_when_available():
+    """Local owner acceptance; CI intentionally uses the anonymized contract."""
+    source = r"C:\Temp\44.json"
+    if not os.path.exists(source):
+        pytest.skip("private #280 fixture is not present")
+    script = (
+        "import {readFileSync} from 'node:fs';"
+        "import {optimizePlans} from './test-build/plan-optimizer.js';"
+        "const raw=JSON.parse(readFileSync(process.argv[1],'utf8'));"
+        "const previous=raw.payload?.config||raw.config||raw;"
+        "process.stdout.write(JSON.stringify(optimizePlans(previous,{}).config));"
+    )
+    completed = subprocess.run(
+        ["node", "--input-type=module", "-e", script, source],
+        cwd=_ROOT, capture_output=True, text=True, check=True,
+    )
+    candidate = json.loads(completed.stdout)
+    raw = json.load(open(source, encoding="utf-8"))
+    previous = raw.get("payload", {}).get("config", raw.get("config", raw))
+    with pytest.raises(v.PartitionOpeningHostError):
+        v.validate_partition_opening_hosts(candidate, previous)
+    v.validate_partition_opening_hosts(
+        candidate, previous, allow_optimize_rehost=True
+    )
+
+
 @pytest.mark.parametrize("cm,cell_cm", [(1, 5), (15, 5), (100, 2.5)])
 @pytest.mark.parametrize("reverse", [False, True])
 @pytest.mark.parametrize("opening_type", ["door", "window", "gate", "passage"])