fix(backend): укрепить I/O и инварианты хранилища (#625)

Issue: #625
User-Visible: yes

custom_components/houseplan/__init__.py

@@ -263,7 +263,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: HouseplanConfigEntry) -
     remove_panel_registration(hass)
     rec = hass.data.get(DOMAIN, {}).pop("trail_recorder", None)
     if rec:
-        rec.teardown()
+        await rec.async_teardown()
+    virtual_lights = getattr(entry.runtime_data, "virtual_lights", None)
+    if virtual_lights is not None:
+        await virtual_lights.async_flush()
     if entry.runtime_data.radar_coordinator:
         entry.runtime_data.radar_coordinator.teardown()
         entry.runtime_data.radar_coordinator = None

custom_components/houseplan/diagnostics.py

@@ -9,7 +9,7 @@ from homeassistant.core import HomeAssistant
 from .store import HouseplanConfigEntry
 
 # Marker metadata may contain personal notes, external links and manual filenames.
-TO_REDACT = {"link", "description", "pdfs", "name"}
+TO_REDACT = {"link", "description", "pdfs", "name", "binding", "settings"}
 
 
 async def async_get_config_entry_diagnostics(
@@ -37,6 +37,8 @@ async def async_get_config_entry_diagnostics(
             for s in config.get("spaces", [])
         ],
         "markers": async_redact_data(config.get("markers", []), TO_REDACT),
-        "settings": config.get("settings", {}),
+        "settings": async_redact_data(
+            {"settings": config.get("settings", {})}, TO_REDACT
+        )["settings"],
         "layout_entries": len(layout),
     }

custom_components/houseplan/http_api.py

@@ -255,13 +255,6 @@ class HouseplanDecorAssetUploadView(HomeAssistantView):
         if not filename:
             return web.json_response({"error": "invalid_format"}, status=400)
 
-        try:
-            validated = await hass.async_add_executor_job(
-                validate_asset, b"".join(blocks), filename, declared_mime,
-            )
-        except DecorAssetError as err:
-            return web.json_response({"error": err.code, "message": str(err)}, status=413 if err.code == "too_large" else 400)
-
         root = Path(hass.config.path(ASSETS_DIR))
 
         def _store() -> tuple[dict, bool]:
@@ -339,9 +332,19 @@ class HouseplanDecorAssetUploadView(HomeAssistantView):
 
         try:
             async with runtime.upload_lock:
+                # Pillow may consume far more RSS than the compressed body.
+                # Serialise decode/validation as well as quota+promotion so N
+                # parallel uploads cannot become N simultaneous decoders.
+                validated = await hass.async_add_executor_job(
+                    validate_asset, b"".join(blocks), filename, declared_mime,
+                )
                 row, reused = await hass.async_add_executor_job(_store)
         except DecorAssetError as err:
-            status = 507 if err.code == "capacity_exceeded" else 400
+            status = (
+                507 if err.code == "capacity_exceeded"
+                else 413 if err.code == "too_large"
+                else 400
+            )
             return web.json_response({"error": err.code, "message": str(err)}, status=status)
         except OSError as err:
             _LOGGER.warning("House Plan decor asset upload: store failed: %s", err)
@@ -362,6 +365,36 @@ class HouseplanUploadView(HomeAssistantView):
             return web.json_response({"error": "unauthorized"}, status=403)
 
         files_root = Path(hass.config.path(FILES_DIR))
+        runtime = get_data(hass)
+        if runtime is None:
+            return web.json_response({"error": "not_ready"}, status=503)
+
+        # Content-Length includes small multipart overhead, making it a safe
+        # conservative upper bound. Reject impossible requests before reading
+        # or creating a temporary file; the exact staged size is checked again
+        # under the same lock immediately before promotion.
+        declared_size = getattr(request, "content_length", None)
+        if declared_size is not None and declared_size > 0:
+            if declared_size > MAX_FILE_BYTES + _FLUSH_AT:
+                return web.json_response(
+                    {"error": "too_large", "max_mb": MAX_FILE_BYTES // 1024 // 1024},
+                    status=413,
+                )
+            try:
+                async with runtime.upload_lock:
+                    await hass.async_add_executor_job(
+                        partial(
+                            check_quota,
+                            files_root,
+                            declared_size,
+                            MAX_FILES_BYTES,
+                            MAX_FILES_COUNT,
+                        )
+                    )
+            except QuotaError as err:
+                return web.json_response(
+                    {"error": err.reason, "detail": err.detail}, status=507
+                )
         marker_id = "misc"
         filename: str | None = None
         # Every temporary file this request creates, promoted or not. The outer
@@ -444,21 +477,6 @@ class HouseplanUploadView(HomeAssistantView):
                 return web.json_response({"error": "no_file"}, status=400)
 
             tmp_path = temps[0]
-            try:
-                # The staged file already sits under files_root: hand it to
-                # the quota as `incoming` only, not as stored usage too (#498).
-                await hass.async_add_executor_job(
-                    partial(
-                        check_quota, files_root, tmp_path.stat().st_size,
-                        MAX_FILES_BYTES, MAX_FILES_COUNT, exclude=tmp_path,
-                        additional_disk_bytes=0,
-                    ),
-                )
-            except QuotaError as err:
-                _LOGGER.warning("House Plan upload refused: %s", err.detail)
-                return web.json_response({"error": err.reason, "detail": err.detail}, status=507)
-            except OSError:
-                pass
             target_dir = files_root / marker_id
             safe_name = filename
 
@@ -480,8 +498,27 @@ class HouseplanUploadView(HomeAssistantView):
                     raise
                 return name
 
+            def _check_and_promote() -> str:
+                # The staged file already sits under files_root: hand it to
+                # the quota as `incoming` only, not as stored usage too (#498).
+                check_quota(
+                    files_root,
+                    tmp_path.stat().st_size,
+                    MAX_FILES_BYTES,
+                    MAX_FILES_COUNT,
+                    exclude=tmp_path,
+                    additional_disk_bytes=0,
+                )
+                return _promote()
+
             try:
-                name = await hass.async_add_executor_job(_promote)
+                async with runtime.upload_lock:
+                    name = await hass.async_add_executor_job(_check_and_promote)
+            except QuotaError as err:
+                _LOGGER.warning("House Plan upload refused: %s", err.detail)
+                return web.json_response(
+                    {"error": err.reason, "detail": err.detail}, status=507
+                )
             except OSError as err:
                 _LOGGER.warning("House Plan upload: could not store the file: %s", err)
                 return web.json_response({"error": "io_error"}, status=500)

custom_components/houseplan/import_export.py

@@ -45,12 +45,14 @@ from .validation import (
     MAX_LAYOUT,
     MAX_MARKERS,
     MAX_SPACES,
+    DuplicateMarkerIdError,
     MarkerControlError,
     OpeningPassageError,
     PartitionOpeningHostError,
     PartitionOpeningJambMarginError,
     sanitize_filename,
     sanitize_marker_id,
+    validate_active_marker_ids,
     validate_marker_controls,
     validate_marker_light_entities,
     validate_marker_vacuum_routes,
@@ -1887,6 +1889,14 @@ def _materialize_import_candidate(
         config = CONFIG_SCHEMA(config)
     except vol.Invalid as err:
         raise ImportFailure("invalid_config", str(err)) from err
+    try:
+        validate_active_marker_ids(
+            config,
+            current_config if prepared["kind"] == "space" else None,
+            validate_all=prepared["kind"] == "full",
+        )
+    except DuplicateMarkerIdError as err:
+        raise ImportFailure(err.code, str(err)) from err
     try:
         layout = LAYOUT_SCHEMA(layout)
     except vol.Invalid as err:

custom_components/houseplan/plans.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import logging
 import os
+import tempfile
 import time
 from pathlib import Path
 from typing import Any
@@ -25,6 +26,21 @@ _LOGGER = logging.getLogger(__name__)
 TMP_PREFIX = ".upload-"
 
 
+def atomic_write(path: Path, data: bytes, *, prefix: str = ".upload-") -> None:
+    """Durably replace ``path`` without exposing a partial destination file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(prefix=prefix, dir=str(path.parent))
+    temp = Path(temp_name)
+    try:
+        with os.fdopen(fd, "wb") as stream:
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temp, path)
+    finally:
+        temp.unlink(missing_ok=True)
+
+
 def reserve_filename(directory: Path, name: str) -> str:
     """Atomically claim a free name inside `directory` and return it.
 

custom_components/houseplan/quality_scale.yaml

@@ -123,4 +123,7 @@ rules:
     comment: No dependencies.
   inject-websession:
     status: exempt
-    comment: No outgoing HTTP.
+    comment: >-
+      The optional user-confirmed support relay is the integration's only
+      outgoing HTTP path and creates its bounded ClientSession explicitly;
+      ordinary plan, device and asset operation remains local.

custom_components/houseplan/store.py

@@ -81,6 +81,7 @@ class HouseplanData:
     store: HouseplanStore
     config_store: HouseplanStore
     virtual_light_store: HouseplanStore
+    virtual_lights: Any | None = None
     # One lock for every load→modify→save cycle of both stores: prevents
     # lost updates from concurrent WS calls and makes the rev check atomic.
     write_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
@@ -118,7 +119,7 @@ HouseplanConfigEntry = ConfigEntry[HouseplanData]
 
 def create_data(hass: HomeAssistant) -> HouseplanData:
     """Create the stores for a config entry."""
-    return HouseplanData(
+    data = HouseplanData(
         store=HouseplanStore(hass, STORAGE_VERSION, STORAGE_KEY, minor_version=STORAGE_MINOR_VERSION),
         config_store=HouseplanStore(
             hass, STORAGE_VERSION, STORAGE_CONFIG_KEY, minor_version=STORAGE_MINOR_VERSION
@@ -130,6 +131,10 @@ def create_data(hass: HomeAssistant) -> HouseplanData:
             minor_version=STORAGE_MINOR_VERSION,
         ),
     )
+    from .virtual_lights import VirtualLightController
+
+    data.virtual_lights = VirtualLightController(data.virtual_light_store)
+    return data
 
 
 def get_data(hass: HomeAssistant) -> HouseplanData | None:
@@ -229,12 +234,17 @@ async def async_save_config_state(
     from .virtual_lights import async_reconcile_virtual_lights
 
     try:
+        controller = getattr(runtime, "virtual_lights", None)
+        if controller is not None:
+            await controller.async_flush()
         await async_reconcile_virtual_lights(
             runtime.virtual_light_store,
             canonical_config,
             rev,
             previous_config_rev=previous_rev,
         )
+        if controller is not None:
+            controller.reset()
     except Exception:  # noqa: BLE001 - config commit already stands
         _LOGGER.exception("House Plan: virtual-light state reconciliation failed")
     return payload

custom_components/houseplan/trails.py

@@ -455,16 +455,30 @@ class TrailRecorder:
                 self.hass, sorted(ents), self._on_state
             )
 
-    def teardown(self) -> None:
+    def _close_subscriptions(self) -> bool:
+        """Stop callbacks/timers and report whether a save was pending."""
         # HP-1540-05: flag FIRST — a refresh parked on its awaited load must
-        # not re-subscribe after this cleanup has already run
+        # not re-subscribe after this cleanup has already run.
         self._closed = True
         if self._unsub_track:
             self._unsub_track()
             self._unsub_track = None
+        pending = self._unsub_save is not None
         if self._unsub_save:
             self._unsub_save()
             self._unsub_save = None
+        return pending
+
+    async def async_teardown(self) -> None:
+        """Stop the recorder and durably flush a pending debounced save."""
+        async with self._refresh_lock:
+            pending = self._close_subscriptions()
+            if pending:
+                await self.store.async_save(self.book.data)
+
+    def teardown(self) -> None:
+        """Synchronous emergency cleanup for already-closing event loops."""
+        self._close_subscriptions()
 
     def _vacuum_entity(self, m: dict[str, Any]) -> str | None:
         b = str(m.get("binding") or "")

custom_components/houseplan/validation.py

@@ -42,6 +42,62 @@ class MarkerControlError(ValueError):
         self.code = code
 
 
+class DuplicateMarkerIdError(ValueError):
+    """A write introduced or retained a changed duplicate active marker id."""
+
+    code = "invalid_config"
+
+    def __init__(self) -> None:
+        # Marker ids may contain user-controlled entity/device identifiers. A
+        # stable message is enough and does not leak the value into logs.
+        super().__init__("duplicate active marker id")
+
+
+def _active_marker_groups(config: dict | None) -> dict[str, list[str]]:
+    """Return order-independent structural signatures grouped by live id."""
+    groups: dict[str, list[str]] = {}
+    for marker in (config or {}).get("markers") or []:
+        if not isinstance(marker, dict) or marker.get("removed") is True:
+            continue
+        marker_id = marker.get("id")
+        if not isinstance(marker_id, str) or not marker_id:
+            continue
+        groups.setdefault(marker_id, []).append(
+            json.dumps(marker, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
+        )
+    for signatures in groups.values():
+        signatures.sort()
+    return groups
+
+
+def validate_active_marker_ids(
+    config: dict,
+    previous: dict | None = None,
+    *,
+    validate_all: bool = False,
+) -> None:
+    """Reject new active-id ambiguity without stranding legacy documents.
+
+    Existing duplicate groups may round-trip unchanged so an unrelated save
+    remains possible. Once any member changes, the candidate must repair the
+    group down to at most one active marker. Authoritative imports pass
+    ``validate_all=True`` because they have no local legacy group to preserve.
+    """
+    candidate = _active_marker_groups(config)
+    duplicates = {
+        marker_id: signatures
+        for marker_id, signatures in candidate.items()
+        if len(signatures) > 1
+    }
+    if not duplicates:
+        return
+    if validate_all or previous is None:
+        raise DuplicateMarkerIdError()
+    old = _active_marker_groups(previous)
+    if any(old.get(marker_id) != signatures for marker_id, signatures in duplicates.items()):
+        raise DuplicateMarkerIdError()
+
+
 class OpeningPassageError(ValueError):
     """Semantic open-passage error with a stable public code and payload."""
 

custom_components/houseplan/virtual_lights.py

@@ -1,6 +1,9 @@
 """Persistent operational state for manual virtual lights."""
 from __future__ import annotations
 
+import asyncio
+import copy
+import logging
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -8,6 +11,8 @@ if TYPE_CHECKING:
 
 
 EVENT_VIRTUAL_LIGHT_UPDATED = "houseplan_virtual_light_updated"
+SAVE_DELAY_S = 0.5
+_LOGGER = logging.getLogger(__name__)
 
 
 def is_manual_virtual_light(marker: Any) -> bool:
@@ -57,6 +62,116 @@ def _wire(rev: int, config_rev: int, off: set[str]) -> dict[str, Any]:
     return {"rev": rev, "config_rev": config_rev, "off": sorted(off)}
 
 
+def _snapshot_payload(
+    stored: Any,
+    config: dict[str, Any],
+    config_rev: int,
+    *,
+    previous_config_rev: int | None = None,
+) -> dict[str, Any]:
+    rev, state_config_rev, stored_off = _read_state(stored)
+    eligible = eligible_virtual_light_ids(config)
+    expected_rev = config_rev if previous_config_rev is None else previous_config_rev
+    off = stored_off & eligible if state_config_rev == expected_rev else set()
+    if off != stored_off:
+        rev += 1
+    return _wire(rev, config_rev, off)
+
+
+class VirtualLightController:
+    """Runtime cache with coalesced durable writes for rapid toggles."""
+
+    def __init__(self, store: HouseplanStore) -> None:
+        self.store = store
+        self._state: dict[str, Any] | None = None
+        self._dirty = False
+        self._save_task: asyncio.Task[None] | None = None
+        self._flush_event = asyncio.Event()
+
+    async def async_snapshot(
+        self, config: dict[str, Any], config_rev: int,
+    ) -> dict[str, Any]:
+        source = self._state
+        if source is None:
+            source = await self.store.async_load() or {}
+        payload = _snapshot_payload(source, config, config_rev)
+        self._state = payload
+        if payload != source:
+            self._dirty = True
+            self._schedule_save()
+        return copy.deepcopy(payload)
+
+    async def async_toggle(
+        self,
+        config: dict[str, Any],
+        config_rev: int,
+        marker_id: str,
+    ) -> dict[str, Any] | None:
+        if marker_id not in eligible_virtual_light_ids(config):
+            return None
+        snapshot = await self.async_snapshot(config, config_rev)
+        off = set(snapshot["off"])
+        if marker_id in off:
+            off.remove(marker_id)
+        else:
+            off.add(marker_id)
+        payload = _wire(_integer(snapshot["rev"]) + 1, config_rev, off)
+        self._state = payload
+        self._dirty = True
+        self._schedule_save()
+        return {
+            "marker_id": marker_id,
+            "on": marker_id not in off,
+            "rev": payload["rev"],
+        }
+
+    def _schedule_save(self) -> None:
+        if self._save_task is None or self._save_task.done():
+            self._save_task = asyncio.create_task(self._delayed_save())
+
+    async def _delayed_save(self) -> None:
+        try:
+            while self._dirty:
+                if not self._flush_event.is_set():
+                    try:
+                        await asyncio.wait_for(
+                            self._flush_event.wait(), timeout=SAVE_DELAY_S,
+                        )
+                    except TimeoutError:
+                        pass
+                if not self._dirty or self._state is None:
+                    continue
+                payload = copy.deepcopy(self._state)
+                self._dirty = False
+                try:
+                    await self.store.async_save(payload)
+                except Exception:  # noqa: BLE001 - next toggle/unload retries
+                    self._dirty = True
+                    _LOGGER.exception("House Plan: virtual-light delayed save failed")
+                    return
+        finally:
+            self._save_task = None
+            if not self._dirty:
+                self._flush_event.clear()
+
+    async def async_flush(self) -> None:
+        """Persist the latest state before config transitions or unload."""
+        task = self._save_task
+        if task is not None:
+            self._flush_event.set()
+            await task
+        if self._dirty and self._state is not None:
+            payload = copy.deepcopy(self._state)
+            await self.store.async_save(payload)
+            self._dirty = False
+        if not self._dirty:
+            self._flush_event.clear()
+
+    def reset(self) -> None:
+        """Forget cache after an external reconciliation wrote the store."""
+        self._state = None
+
+
 async def async_virtual_light_snapshot(
     store: HouseplanStore,
     config: dict[str, Any],
@@ -70,12 +185,7 @@ async def async_virtual_light_snapshot(
     old off state for a marker whose role changed in the meantime.
     """
     stored = await store.async_load() or {}
-    rev, state_config_rev, stored_off = _read_state(stored)
-    eligible = eligible_virtual_light_ids(config)
-    off = stored_off & eligible if state_config_rev == config_rev else set()
-    if off != stored_off:
-        rev += 1
-    payload = _wire(rev, config_rev, off)
+    payload = _snapshot_payload(stored, config, config_rev)
     if payload != stored:
         await store.async_save(payload)
     return payload
@@ -90,12 +200,9 @@ async def async_reconcile_virtual_lights(
 ) -> dict[str, Any]:
     """Carry eligible state across one known configuration transition."""
     stored = await store.async_load() or {}
-    rev, state_config_rev, stored_off = _read_state(stored)
-    eligible = eligible_virtual_light_ids(config)
-    off = stored_off & eligible if state_config_rev == previous_config_rev else set()
-    if off != stored_off:
-        rev += 1
-    payload = _wire(rev, config_rev, off)
+    payload = _snapshot_payload(
+        stored, config, config_rev, previous_config_rev=previous_config_rev,
+    )
     if payload != stored:
         await store.async_save(payload)
     return payload

custom_components/houseplan/websocket_api.py

@@ -72,6 +72,7 @@ from .import_export import (
 from .junction_limits import JunctionLimitError, validate_junction_limits
 from .plans import (
     QuotaError,
+    atomic_write,
     check_quota,
     collect_attachments,
     collect_plans,
@@ -113,6 +114,7 @@ from .validation import (
     MAX_PLAN_BYTES,
     PLAN_EXTENSIONS,
     POS_SCHEMA,
+    DuplicateMarkerIdError,
     MarkerControlError,
     OpeningPassageError,
     PartitionOpeningHostError,
@@ -121,6 +123,7 @@ from .validation import (
     prepare_ordinary_summary_candidate,
     sanitize_filename,
     valid_space_id,
+    validate_active_marker_ids,
     validate_marker_controls,
     validate_marker_light_entities,
     validate_marker_vacuum_routes,
@@ -141,6 +144,19 @@ from .wall_segment_model import (
 )
 
 _LOGGER = logging.getLogger(__name__)
+_MISSING_REV_DEBUGGED: set[str] = set()
+
+
+def _debug_missing_revision_once(command: str, current_rev: int) -> None:
+    """Log one low-noise diagnostic per legacy write command."""
+    if command in _MISSING_REV_DEBUGGED:
+        return
+    _MISSING_REV_DEBUGGED.add(command)
+    _LOGGER.debug(
+        "House Plan: %s without expected_rev over rev %s; write rejected",
+        command,
+        current_rev,
+    )
 
 def _optimizer_backup_is_current(config_data: dict[str, Any], layout_data: dict[str, Any]) -> bool:
     """An optimization can be undone before any later ordinary plan edit."""
@@ -440,21 +456,23 @@ async def ws_export_create(hass: HomeAssistant, connection, msg: dict[str, Any])
         return
     try:
         async with rt.write_lock:
-            config_data = await rt.config_store.async_load() or {}
-            layout_data = await rt.store.async_load() or {}
-            document, filename = await hass.async_add_executor_job(
-                partial(
-                    create_export,
-                    rt,
-                    config_data,
-                    layout_data,
-                    kind=msg["kind"],
-                    space_id=msg.get("space_id"),
-                    plan_only=msg.get("plan_only", False),
-                    card_version=msg.get("card_version", ""),
-                    config_root=Path(hass.config.path("")),
-                )
+            # Copy one coherent pair while writers are excluded, then release
+            # the global lock before hashing assets and building the document.
+            config_data = copy.deepcopy(await rt.config_store.async_load() or {})
+            layout_data = copy.deepcopy(await rt.store.async_load() or {})
+        document, filename = await hass.async_add_executor_job(
+            partial(
+                create_export,
+                rt,
+                config_data,
+                layout_data,
+                kind=msg["kind"],
+                space_id=msg.get("space_id"),
+                plan_only=msg.get("plan_only", False),
+                card_version=msg.get("card_version", ""),
+                config_root=Path(hass.config.path("")),
             )
+        )
     except ImportFailure as err:
         _send_import_error(connection, msg["id"], err)
         return
@@ -575,8 +593,11 @@ async def ws_import_apply(hass: HomeAssistant, connection, msg: dict[str, Any])
                     "missing_plan",
                     "Plan file no longer exists: " + ", ".join(sorted(missing)),
                 )
-            missing_attachments = _missing_internal_attachments(
-                Path(hass.config.path("")), target_config, config_data.get("config")
+            missing_attachments = await hass.async_add_executor_job(
+                _missing_internal_attachments,
+                Path(hass.config.path("")),
+                target_config,
+                config_data.get("config"),
             )
             if missing_attachments:
                 raise ImportFailure(
@@ -751,11 +772,7 @@ async def ws_layout_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
             # indistinguishable from a stale writer.  Keep the schema field
             # optional only for rev-zero bootstrap and return the stable domain
             # conflict here rather than allowing canonical no-op to bypass CAS.
-            _LOGGER.warning(
-                "House Plan: layout/set without expected_rev over rev %s — "
-                "write rejected (outdated client?)",
-                current_rev,
-            )
+            _debug_missing_revision_once("layout/set", current_rev)
             connection.send_error(
                 msg["id"], "conflict",
                 f"Layout revision is required; reload the layout, or — for "
@@ -808,10 +825,6 @@ async def ws_layout_update(hass: HomeAssistant, connection, msg: dict[str, Any])
         config_data = resolved.config_data
         config = config_data.get("config") or {}
         markers = config.get("markers") or []
-        deleted = any(
-            str(m.get("id")) == msg["device_id"] and m.get("removed") is True
-            for m in markers
-        )
         live_virtual = any(
             str(m.get("id")) == msg["device_id"] and m.get("removed") is not True
             and m.get("binding") == "virtual"
@@ -821,6 +834,10 @@ async def ws_layout_update(hass: HomeAssistant, connection, msg: dict[str, Any])
             str(m.get("id")) == msg["device_id"] and m.get("removed") is not True
             for m in markers
         )
+        deleted = not live_explicit and any(
+            str(m.get("id")) == msg["device_id"] and m.get("removed") is True
+            for m in markers
+        )
         orphan_virtual = (
             msg["device_id"].startswith("v_")
             and not live_virtual
@@ -1429,10 +1446,10 @@ async def ws_config_get(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
         config = {**DEFAULT_CONFIG, **data.get("config", {})}
         config_rev = int(data.get("rev", 0))
         try:
-            virtual_lights = await async_virtual_light_snapshot(
-                rt.virtual_light_store,
-                config,
-                config_rev,
+            virtual_lights = await (
+                rt.virtual_lights.async_snapshot(config, config_rev)
+                if rt.virtual_lights is not None
+                else async_virtual_light_snapshot(rt.virtual_light_store, config, config_rev)
             )
         except Exception:  # noqa: BLE001 - config remains independently readable
             _LOGGER.exception("House Plan: reading virtual-light state failed")
@@ -1482,11 +1499,17 @@ async def ws_virtual_light_toggle(
     async with rt.write_lock:
         data = await rt.config_store.async_load() or {}
         config = {**DEFAULT_CONFIG, **data.get("config", {})}
-        result = await async_toggle_virtual_light(
-            rt.virtual_light_store,
-            config,
-            int(data.get("rev", 0)),
-            msg["marker_id"],
+        result = await (
+            rt.virtual_lights.async_toggle(
+                config, int(data.get("rev", 0)), msg["marker_id"]
+            )
+            if rt.virtual_lights is not None
+            else async_toggle_virtual_light(
+                rt.virtual_light_store,
+                config,
+                int(data.get("rev", 0)),
+                msg["marker_id"],
+            )
         )
         if result is None:
             connection.send_error(
@@ -1495,8 +1518,8 @@ async def ws_virtual_light_toggle(
                 "Marker is not an active virtual light with tap_action=toggle",
             )
             return
-    # Both the reply and event follow the durable Store write.  There is no
-    # optimistic client state, so all cards converge on this revision.
+    # The runtime revision is immediate; durable writes are coalesced and
+    # flushed before config transitions/unload.
     connection.send_result(msg["id"], result)
     hass.bus.async_fire(EVENT_VIRTUAL_LIGHT_UPDATED, result)
 
@@ -1627,11 +1650,7 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
             # empty-store bootstrap and so this path can return the same stable
             # domain error as an explicit stale revision.  Accepting it over a
             # saved document would bypass optimistic locking entirely.
-            _LOGGER.warning(
-                "House Plan: config/set without expected_rev over rev %s — "
-                "write rejected (outdated client?)",
-                current_rev,
-            )
+            _debug_missing_revision_once("config/set", current_rev)
             connection.send_error(
                 msg["id"], "conflict",
                 f"Configuration revision is required; reload the configuration, "
@@ -1670,6 +1689,7 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
             )
             msg["config"].clear()
             msg["config"].update(checked)
+            validate_active_marker_ids(msg["config"], data.get("config"))
             validate_marker_controls(msg["config"], data.get("config"))
             validate_marker_light_entities(msg["config"], data.get("config"))
             validate_marker_radars(
@@ -1686,7 +1706,7 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
         try:
             candidate_counts = await hass.async_add_executor_job(_validate_config_cpu)
         except (
-            JunctionLimitError, MarkerControlError, OpeningPassageError,
+            DuplicateMarkerIdError, JunctionLimitError, MarkerControlError, OpeningPassageError,
             PartitionOpeningHostError, PartitionOpeningJambMarginError,
             WallModelClientOutdatedError,
         ) as err:
@@ -1916,6 +1936,7 @@ async def ws_space_delete(hass: HomeAssistant, connection, msg: dict[str, Any])
                 )
                 return
             target_config = CONFIG_SCHEMA(target_config)
+            validate_active_marker_ids(target_config, current_config)
             target_layout = LAYOUT_SCHEMA(target_layout)
             new_config_rev = config_rev + 1
             new_layout_rev = layout_rev + 1
@@ -1952,6 +1973,9 @@ async def ws_space_delete(hass: HomeAssistant, connection, msg: dict[str, Any])
     except ImportFailure as err:
         _send_import_error(connection, msg["id"], err)
         return
+    except DuplicateMarkerIdError as err:
+        connection.send_error(msg["id"], err.code, str(err))
+        return
     except vol.Invalid as err:
         connection.send_error(msg["id"], "invalid_config", str(err))
         return
@@ -2058,6 +2082,7 @@ async def ws_plan_optimize(hass: HomeAssistant, connection, msg: dict[str, Any])
                 return None, migrated_size
             msg["config"].clear()
             msg["config"].update(checked)
+            validate_active_marker_ids(msg["config"], config_data.get("config"))
             validate_marker_controls(msg["config"], config_data.get("config"))
             validate_marker_light_entities(msg["config"], config_data.get("config"))
             validate_marker_radars(
@@ -2093,7 +2118,7 @@ async def ws_plan_optimize(hass: HomeAssistant, connection, msg: dict[str, Any])
                 )
                 return
         except (
-            JunctionLimitError, MarkerControlError, OpeningPassageError,
+            DuplicateMarkerIdError, JunctionLimitError, MarkerControlError, OpeningPassageError,
             PartitionOpeningHostError, PartitionOpeningJambMarginError,
             WallModelClientOutdatedError,
             WallSegmentMigrationError,
@@ -2347,8 +2372,7 @@ async def ws_plan_set(hass: HomeAssistant, connection, msg: dict[str, Any]) -> N
         # (HP-1490-02). A failed write reserves nothing — the file either
         # exists and is counted by the next scan, or does not and is not.
         check_quota(plans_dir, len(raw), MAX_PLANS_BYTES, MAX_PLANS_FILES)
-        plans_dir.mkdir(parents=True, exist_ok=True)
-        path.write_bytes(raw)
+        atomic_write(path, raw, prefix=".plan-upload-")
 
     data = _runtime(hass, connection, msg["id"])
     if data is None:

docs/ARCHITECTURE.md

@@ -1282,7 +1282,13 @@ is persisted; Apply still sends only the exact ordinary config/layout pair that
 was previewed.
 
 Manual attachments upload over HTTP (streaming, transactional staging), not WS —
-the old `houseplan/file/set` was removed in v1.10.0.
+the old `houseplan/file/set` was removed in v1.10.0. A usable
+`Content-Length` is checked conservatively against the hard limit, aggregate
+quota and free-space floor before multipart streaming begins; chunked uploads
+retain the streaming cap. The exact staged size is checked again under the
+runtime `upload_lock` immediately before promotion. Decor-image decoding is
+also serialized by that lock so compressed images cannot multiply peak Pillow
+memory across concurrent requests.
 
 Manual virtual-light state is operational data, not plan configuration. The
 integration owns a separate versioned `houseplan.virtual_lights` Store whose
@@ -1291,8 +1297,11 @@ write lock serializes config reconciliation and atomic toggles. Eligibility is
 always recalculated from server config; the toggle command accepts no desired
 state, entity id or service. It is intentionally available to every
 authenticated View user, while config writers remain governed by `may_write`.
-The durable save precedes both reply and event. A config-revision gap from an
-older writer clears manual off bits to the compatibility default `on`.
+The runtime revision, reply and event are immediate; rapid toggles are
+coalesced into one delayed durable write of the latest state. Config
+transitions and integration unload flush pending state before continuing. A
+config-revision gap from an older writer clears manual off bits to the
+compatibility default `on`.
 
 The first `config/get` frame carries the coherent operational snapshot. Full
 cards subscribe directly to the update event; all `houseplan-space-card`
@@ -1316,6 +1325,13 @@ geometry/presentation allowlists. The parser recomputes that projection and
 its placement manifest before showing a plan-only preview, so manually adding
 a private field while keeping `transfer.plan_only: true` is rejected.
 
+Export snapshots config and layout as one coherent deep copy while holding the
+shared write lock, then releases the lock before schema projection and content
+hashing in the executor. Thus an export reflects exactly one stored pair while
+ordinary reads and later writes do not wait for archive materialization.
+Import attachment/asset scans likewise run in the executor; only the paired
+revision check and commit remain serialized.
+
 The browser never parses imported configuration. Optimize, Optimize Undo, full
 import, space deletion and maintenance share the `optimize_pending`
 crash-recovery intent and the one-deep backup slot; the backup carries

docs/CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Large exports no longer hold up ordinary House Plan operations, and imports
+  no longer block Home Assistant while scanning attachments; uploads now
+  reject impossible sizes before streaming, image
+  validation is memory-bounded, rapid virtual-light toggles avoid redundant
+  disk writes, and ambiguous duplicate active devices are rejected without
+  stranding existing legacy configurations
+  ([#625](https://github.com/Matysh/houseplan-card/issues/625)).
 - Read-only household and kiosk views no longer show an administrator-only
   error when Home Assistant discovers a new device; config conflicts in View
   now refresh safely even though the editor has never been loaded

docs/CHANGELOG.ru.md

@@ -8,6 +8,13 @@
 
 ## Не выпущено
 
+- Большой экспорт больше не задерживает обычные операции House Plan, а импорт
+  не блокирует Home Assistant во время проверки вложений; заведомо невозможные
+  загрузки отклоняются до передачи файла,
+  проверка изображений ограничена по памяти, быстрые переключения виртуального
+  света не создают лишних записей на диск, а неоднозначные дубликаты активных
+  устройств отклоняются без блокировки старых конфигураций
+  ([#625](https://github.com/Matysh/houseplan-card/issues/625)).
 - Read-only экраны домочадцев и kiosk больше не показывают ошибку о правах
   администратора при появлении нового устройства в Home Assistant; конфликт
   конфига в режиме просмотра теперь безопасно обновляет данные, даже если

docs/CONFIG-COMPATIBILITY.md

@@ -706,6 +706,23 @@ full/space transfer preserve, remap or report/drop `derived_marker_state.ref`
 through the same reference seam as controls and value badges. Older clients
 ignore the field and may erase it if they reconstruct the marker.
 
+## Active marker ID uniqueness (#625)
+
+At every write boundary, a marker `id` may identify at most one active record
+(`removed` is not `true`). A tombstone and one active marker with the same id
+remain valid: the tombstone records lifecycle history and does not suppress the
+live marker's layout update.
+
+The structural `CONFIG_SCHEMA` remains permissive so an installation that
+already contains two active legacy records is still readable. The semantic
+validator compares the candidate with the previously stored document: an
+unchanged duplicate group may survive an unrelated save, but a new duplicate
+or any edit that leaves the group ambiguous is rejected as `invalid_config`
+with the stable detail `duplicate active marker id`. Removing/tombstoning enough
+records to leave one active marker is the supported repair. A full import is authoritative and therefore
+strict even when its source document contains legacy duplicates; there is no
+automatic deletion or migration.
+
 ## Atomic marker writes (#442)
 
 The Device editor builds a separate complete config candidate and treats a

tests_backend/test_ha_diagnostics.py

@@ -0,0 +1,57 @@
+"""Privacy contracts for House Plan diagnostics (#625)."""
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from custom_components.houseplan.diagnostics import async_get_config_entry_diagnostics
+
+
+@pytest.fixture(autouse=True)
+def _enable_custom_integrations(enable_custom_integrations):
+    yield
+
+
+class _Store:
+    def __init__(self, value):
+        self.value = value
+
+    async def async_load(self):
+        return self.value
+
+
+async def test_diagnostics_redact_marker_bindings_and_all_settings(hass) -> None:
+    config = {
+        "spaces": [{
+            "id": "floor", "aspect": 1.5, "plan_url": None,
+            "rooms": [{"area": "private-area"}],
+            "partitions": [], "wall_columns": [],
+        }],
+        "markers": [{
+            "id": "marker", "binding": "device:private-device",
+            "name": "Private name", "settings": {"token": "marker-secret"},
+        }],
+        "settings": {
+            "known_devices": ["private-device"],
+            "support_contact": "owner@example.invalid",
+            "nested": {"entity": "sensor.private"},
+        },
+    }
+    runtime = SimpleNamespace(
+        config_store=_Store({"config": config, "rev": 7}),
+        store=_Store({"layout": {"marker": {"x": 0.2, "y": 0.3}}}),
+    )
+    entry = SimpleNamespace(runtime_data=runtime, options={})
+
+    result = await async_get_config_entry_diagnostics(hass, entry)
+    wire = json.dumps(result, sort_keys=True)
+    for secret in (
+        "device:private-device", "Private name", "marker-secret",
+        "private-device", "owner@example.invalid", "sensor.private",
+    ):
+        assert secret not in wire
+    assert result["rev"] == 7
+    assert result["layout_entries"] == 1
+    assert result["spaces"][0]["rooms"] == 1

tests_backend/test_ha_import_export.py

@@ -10,6 +10,7 @@ import asyncio
 import copy
 import hashlib
 import json
+import threading
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
@@ -704,6 +705,22 @@ def test_full_preview_drops_dormant_broken_and_duplicate_links(tmp_path: Path) -
     assert controller["controls"] == ["marker:dumb", "switch.keep"]
 
 
+def test_full_import_rejects_duplicate_active_marker_ids(tmp_path: Path) -> None:
+    document = _document(tmp_path)
+    document["payload"]["config"]["markers"].append({
+        "id": "lamp", "binding": "virtual", "name": "Ambiguous duplicate",
+    })
+    with pytest.raises(ImportFailure) as invalid:
+        create_preview(
+            SimpleNamespace(instance_id="instance-a", import_previews={}),
+            json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip",
+            current_config_data={"config": _config(), "rev": 1},
+            current_layout_data={"layout": {}, "rev": 1}, config_root=tmp_path,
+        )
+    assert invalid.value.code == "invalid_config"
+    assert str(invalid.value) == "duplicate active marker id"
+
+
 @pytest.mark.parametrize("kind", ["self", "cycle"])
 def test_full_preview_still_rejects_self_and_cycle(kind: str, tmp_path: Path) -> None:
     document = _document(tmp_path)
@@ -2359,6 +2376,64 @@ async def test_full_export_waits_for_a_concurrent_paired_write(
     assert payload["layout"] == new_layout
 
 
+async def test_config_get_does_not_wait_for_slow_export_materialization(
+    hass: HomeAssistant, monkeypatch,
+) -> None:
+    await _setup(hass)
+    rt = get_data(hass)
+    assert rt is not None
+    await rt.config_store.async_save({"config": _config(), "rev": 1})
+    await rt.store.async_save({"layout": {}, "rev": 1})
+
+    started = threading.Event()
+    release = threading.Event()
+    original = wsapi.create_export
+
+    def slow_export(*args, **kwargs):
+        started.set()
+        assert release.wait(5), "test did not release the delayed export"
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(wsapi, "create_export", slow_export)
+    exported = _Connection()
+    export_task = asyncio.create_task(wsapi.ws_export_create.__wrapped__(hass, exported, {
+        "id": 45, "type": "houseplan/export/create", "kind": "full",
+        "card_version": "review",
+    }))
+    try:
+        assert await hass.async_add_executor_job(started.wait, 2)
+        read = _Connection()
+        await asyncio.wait_for(
+            wsapi.ws_config_get.__wrapped__(
+                hass, read, {"id": 46, "type": "houseplan/config/get"},
+            ),
+            timeout=1,
+        )
+        assert read.error is None and read.result is not None
+        assert not export_task.done()
+    finally:
+        release.set()
+        await export_task
+
+
+async def test_import_attachment_scan_runs_in_executor(
+    hass: HomeAssistant, tmp_path: Path, monkeypatch,
+) -> None:
+    await _setup(hass)
+    _rt, response, _document_value = await _candidate(hass, tmp_path)
+    loop_thread = threading.get_ident()
+    called_from: list[int] = []
+
+    def observed_scan(*_args):
+        called_from.append(threading.get_ident())
+        return set()
+
+    monkeypatch.setattr(wsapi, "_missing_internal_attachments", observed_scan)
+    connection = await _apply(hass, response)
+    assert connection.error is None and connection.result is not None
+    assert called_from and all(thread_id != loop_thread for thread_id in called_from)
+
+
 @pytest.mark.parametrize(("endpoint", "message"), [
     (wsapi.ws_export_create, {
         "id": 50, "type": "houseplan/export/create", "kind": "full", "card_version": "review",

tests_backend/test_ha_virtual_lights.py

@@ -111,6 +111,25 @@ async def test_read_only_authenticated_user_can_toggle(
     assert response["result"]["on"] is False
 
 
+async def test_unload_flushes_a_toggle_still_inside_the_debounce_window(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    monkeypatch,
+) -> None:
+    from custom_components.houseplan import virtual_lights as virtual_lights_module
+
+    monkeypatch.setattr(virtual_lights_module, "SAVE_DELAY_S", 3600)
+    entry = await _setup(hass)
+    client = await hass_ws_client(hass)
+    await _set_config(client, _config(_manual()), 0)
+    assert (await _toggle(client))["result"]["on"] is False
+    assert (await entry.runtime_data.virtual_light_store.async_load())["off"] == []
+
+    assert await hass.config_entries.async_reload(entry.entry_id)
+    restarted = await hass_ws_client(hass)
+    assert (await _get_config(restarted))["virtual_lights"]["off"] == ["lamp"]
+
+
 async def test_lifecycle_preserves_hidden_and_prunes_when_eligibility_ends(
     hass: HomeAssistant, hass_ws_client: WebSocketGenerator,
 ) -> None:
@@ -156,7 +175,7 @@ async def test_invalid_target_and_concurrent_toggles_are_server_atomic(
     assert (await _get_config(client))["virtual_lights"]["off"] == []
 
 
-async def test_failed_durable_save_emits_neither_success_nor_event(
+async def test_failed_delayed_save_keeps_immediate_result_and_is_flushable(
     hass: HomeAssistant,
     hass_ws_client: WebSocketGenerator,
     monkeypatch,
@@ -173,9 +192,21 @@ async def test_failed_durable_save_emits_neither_success_nor_event(
     async def fail_save(_data):
         raise OSError("disk full")
 
+    from custom_components.houseplan import virtual_lights as virtual_lights_module
+
+    original_save = entry.runtime_data.virtual_light_store.async_save
+    monkeypatch.setattr(virtual_lights_module, "SAVE_DELAY_S", 0)
     monkeypatch.setattr(entry.runtime_data.virtual_light_store, "async_save", fail_save)
     response = await _toggle(client)
     await hass.async_block_till_done()
+    assert response["success"]
+    assert events == [response["result"]]
+    assert entry.runtime_data.virtual_lights._dirty is True
+
+    monkeypatch.setattr(entry.runtime_data.virtual_light_store, "async_save", original_save)
+    await entry.runtime_data.virtual_lights.async_flush()
     remove()
-    assert not response["success"]
-    assert events == []
+    assert entry.runtime_data.virtual_lights._dirty is False
+    assert await entry.runtime_data.virtual_light_store.async_load() == {
+        "rev": 1, "config_rev": 1, "off": ["lamp"],
+    }

tests_backend/test_ha_websocket.py

@@ -3,6 +3,8 @@ import asyncio
 import copy
 import json
 import logging
+import threading
+import time
 from pathlib import Path
 
 import pytest
@@ -311,6 +313,8 @@ async def test_deleted_marker_rejects_a_stale_layout_update(
         "spaces": [],
         "markers": [
             {"id": "dev1", "binding": "device:dev1", "removed": True},
+            {"id": "dev_both", "binding": "device:old", "removed": True},
+            {"id": "dev_both", "binding": "device:current"},
             {"id": "v_live", "binding": "virtual", "name": "Still here"},
             {"id": "v_real", "binding": "device:real-device"},
         ],
@@ -329,6 +333,15 @@ async def test_deleted_marker_rejects_a_stale_layout_update(
     ignored = await client.receive_json()
     assert ignored["success"] and ignored["result"]["ignored"] == "removed"
 
+    await client.send_json_auto_id({
+        "type": "houseplan/layout/update",
+        "device_id": "dev_both",
+        "pos": {"s": "f1", "x": 0.15, "y": 0.25},
+    })
+    tombstone_and_live = await client.receive_json()
+    assert tombstone_and_live["success"]
+    assert "ignored" not in tombstone_and_live["result"]
+
     await client.send_json_auto_id({
         "type": "houseplan/layout/update",
         "device_id": "v_deleted",
@@ -350,6 +363,7 @@ async def test_deleted_marker_rejects_a_stale_layout_update(
 
     await client.send_json_auto_id({"type": "houseplan/layout/get"})
     assert (await client.receive_json())["result"]["layout"] == {
+        "dev_both": {"s": "f1", "x": 0.15, "y": 0.25},
         "v_real": {"s": "f1", "x": 0.2, "y": 0.25},
     }
 
@@ -364,7 +378,7 @@ async def test_deleted_marker_rejects_a_stale_layout_update(
             "rl_r1": {"s": "f1", "x": 0.35, "y": 0.45},
             "ordinary_auto_device": {"s": "f1", "x": 0.5, "y": 0.6},
         },
-        "expected_rev": 1,
+        "expected_rev": 2,
     })
     assert (await client.receive_json())["success"]
     await client.send_json_auto_id({"type": "houseplan/layout/get"})
@@ -548,8 +562,10 @@ async def test_issue_340_config_set_without_revision_is_bootstrap_only(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """A missing revision may initialise an empty store, never replace it."""
+    from custom_components.houseplan import websocket_api as wsapi
     from custom_components.houseplan.store import OPTIMIZE_BACKUP, get_data
 
+    wsapi._MISSING_REV_DEBUGGED.discard("config/set")
     await _setup(hass)
     first_client = await hass_ws_client(hass)
     stale_client = await hass_ws_client(hass)
@@ -592,7 +608,7 @@ async def test_issue_340_config_set_without_revision_is_bootstrap_only(
     })
     config_events.clear()
 
-    with caplog.at_level(logging.WARNING, logger="custom_components.houseplan.websocket_api"):
+    with caplog.at_level(logging.DEBUG, logger="custom_components.houseplan.websocket_api"):
         await stale_client.send_json_auto_id({
             "type": "houseplan/config/set", "config": copy.deepcopy(stale_config),
         })
@@ -608,15 +624,17 @@ async def test_issue_340_config_set_without_revision_is_bootstrap_only(
     assert "stale-secret" not in caplog.text
 
     # Even an exact semantic no-op may not be used to bypass the CAS guard.
-    await stale_client.send_json_auto_id({
-        "type": "houseplan/config/set", "config": copy.deepcopy(first_config),
-    })
-    noop_without_revision = await stale_client.receive_json()
+    with caplog.at_level(logging.DEBUG, logger="custom_components.houseplan.websocket_api"):
+        await stale_client.send_json_auto_id({
+            "type": "houseplan/config/set", "config": copy.deepcopy(first_config),
+        })
+        noop_without_revision = await stale_client.receive_json()
     await hass.async_block_till_done()
     assert not noop_without_revision["success"]
     assert noop_without_revision["error"]["code"] == "conflict"
     assert await runtime.config_store.async_load() == stored_before
     assert config_events == []
+    assert sum("config/set without expected_rev" in record.message for record in caplog.records) == 1
 
     # The same client succeeds after reading and returning the current rev.
     await stale_client.send_json_auto_id({
@@ -633,8 +651,10 @@ async def test_issue_356_layout_set_without_revision_is_bootstrap_only(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """A revision-less wholesale layout may initialise, never replace, a store."""
+    from custom_components.houseplan import websocket_api as wsapi
     from custom_components.houseplan.store import OPTIMIZE_BACKUP, get_data
 
+    wsapi._MISSING_REV_DEBUGGED.discard("layout/set")
     await _setup(hass)
     first_client = await hass_ws_client(hass)
     stale_client = await hass_ws_client(hass)
@@ -671,7 +691,7 @@ async def test_issue_356_layout_set_without_revision_is_bootstrap_only(
     layout_events.clear()
 
     with caplog.at_level(
-        logging.WARNING, logger="custom_components.houseplan.websocket_api",
+        logging.DEBUG, logger="custom_components.houseplan.websocket_api",
     ):
         await stale_client.send_json_auto_id({
             "type": "houseplan/layout/set", "layout": copy.deepcopy(stale_layout),
@@ -687,15 +707,17 @@ async def test_issue_356_layout_set_without_revision_is_bootstrap_only(
     assert "stale-secret" not in caplog.text
 
     # An equal body is still a write attempt and must not bypass the CAS guard.
-    await stale_client.send_json_auto_id({
-        "type": "houseplan/layout/set", "layout": copy.deepcopy(first_layout),
-    })
-    noop_without_revision = await stale_client.receive_json()
+    with caplog.at_level(logging.DEBUG, logger="custom_components.houseplan.websocket_api"):
+        await stale_client.send_json_auto_id({
+            "type": "houseplan/layout/set", "layout": copy.deepcopy(first_layout),
+        })
+        noop_without_revision = await stale_client.receive_json()
     await hass.async_block_till_done()
     assert not noop_without_revision["success"]
     assert noop_without_revision["error"]["code"] == "conflict"
     assert await runtime.store.async_load() == stored_before
     assert layout_events == []
+    assert sum("layout/set without expected_rev" in record.message for record in caplog.records) == 1
 
     # Reading and returning the current revision preserves the ordinary path.
     await stale_client.send_json_auto_id({
@@ -2673,9 +2695,28 @@ async def test_decor_asset_upload_deduplicates_and_rejects_mime_spoofing(
             return _Reader(self._mime)
 
     view = HouseplanDecorAssetUploadView()
+    real_validate = hp_http.validate_asset
+    validation_lock = threading.Lock()
+    active_validations = 0
+    max_active_validations = 0
+
+    def observed_validate(*args, **kwargs):
+        nonlocal active_validations, max_active_validations
+        with validation_lock:
+            active_validations += 1
+            max_active_validations = max(max_active_validations, active_validations)
+        try:
+            time.sleep(0.03)
+            return real_validate(*args, **kwargs)
+        finally:
+            with validation_lock:
+                active_validations -= 1
+
+    monkeypatch.setattr(hp_http, "validate_asset", observed_validate)
     first, duplicate = await asyncio.gather(
         view.post(_Request("image/png")), view.post(_Request("image/png")),
     )
+    assert max_active_validations == 1
     rows = [json.loads(first.text), json.loads(duplicate.text)]
     assert {row["reused"] for row in rows} == {False, True}
     assert rows[0]["asset"]["asset_id"] == rows[1]["asset"]["asset_id"]
@@ -3145,6 +3186,36 @@ async def test_upload_never_overwrites_an_existing_attachment(
     assert await got2.read() == b"TWO"
 
 
+async def test_attachment_upload_rejects_impossible_content_length_before_multipart(
+    hass: HomeAssistant,
+) -> None:
+    from custom_components.houseplan import http_api
+    from custom_components.houseplan.http_api import HouseplanUploadView
+
+    await _setup(hass)
+    multipart_called = False
+
+    class _User:
+        is_admin = True
+
+    class _Request:
+        app = {http_api.KEY_HASS: hass}
+        content_length = http_api.MAX_FILE_BYTES + http_api._FLUSH_AT + 1
+
+        def get(self, _key, default=None):
+            return _User()
+
+        async def multipart(self):
+            nonlocal multipart_called
+            multipart_called = True
+            raise AssertionError("multipart must not be read after early rejection")
+
+    response = await HouseplanUploadView().post(_Request())
+    assert response.status == 413
+    assert json.loads(response.text)["error"] == "too_large"
+    assert multipart_called is False
+
+
 async def test_upload_leaves_no_temporary_behind(
     hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_client, monkeypatch
 ) -> None:

tests_backend/test_trail_recorder.py

@@ -843,6 +843,30 @@ def test_overlapping_refreshes_leave_one_subscription_teardown_zero():
         trails.async_track_state_change_event = old_track
 
 
+def test_async_teardown_flushes_pending_debounced_state_and_closes_handles():
+    rec, _hass, _states = _rec()
+    saved = []
+    cancelled = []
+    untracked = []
+
+    class TrailStore:
+        async def async_save(self, data):
+            saved.append(json.loads(json.dumps(data)))
+
+    rec.store = TrailStore()
+    rec.book.data = {"m1": {"current": {"points": [[1, 2]]}}}
+    rec._unsub_save = lambda: cancelled.append(True)
+    rec._unsub_track = lambda: untracked.append(True)
+
+    _run_isolated(rec.async_teardown())
+
+    assert saved == [rec.book.data]
+    assert cancelled == [True]
+    assert untracked == [True]
+    assert rec._unsub_save is None and rec._unsub_track is None
+    assert rec._closed is True
+
+
 def test_refresh_watches_every_route_source_not_only_the_root(monkeypatch):
     """#162: карты одного робота могут идти через разные камеры."""
     import asyncio

tests_backend/test_validation.py

@@ -115,6 +115,35 @@ plans = _load_pure("plans")
 const = importlib.import_module("hp_pure.const")
 
 
+def test_active_marker_id_uniqueness_is_delta_safe_for_legacy_documents():
+    legacy = {"markers": [
+        {"id": "same", "binding": "virtual", "name": "A"},
+        {"id": "same", "binding": "virtual", "name": "B"},
+    ]}
+    # Structural validation deliberately remains migration-compatible; the
+    # semantic write boundary owns the uniqueness invariant.
+    assert len(v.CONFIG_SCHEMA({"spaces": [], **legacy})["markers"]) == 2
+    with pytest.raises(v.DuplicateMarkerIdError) as strict:
+        v.validate_active_marker_ids(legacy, validate_all=True)
+    assert strict.value.code == "invalid_config"
+
+    reordered = {"markers": list(reversed(legacy["markers"]))}
+    v.validate_active_marker_ids(reordered, legacy)
+
+    changed = {"markers": [legacy["markers"][0], {**legacy["markers"][1], "name": "C"}]}
+    with pytest.raises(v.DuplicateMarkerIdError):
+        v.validate_active_marker_ids(changed, legacy)
+
+    v.validate_active_marker_ids({"markers": [legacy["markers"][0]]}, legacy)
+
+
+def test_active_marker_id_allows_one_live_record_and_tombstones():
+    v.validate_active_marker_ids({"markers": [
+        {"id": "same", "binding": "virtual"},
+        {"id": "same", "binding": "virtual", "removed": True},
+    ]}, validate_all=True)
+
+
 def test_sanitize_marker_id():
     assert v.sanitize_marker_id("../etc/passwd") == "_etc_passwd"
     assert v.sanitize_marker_id("..") == "misc"       # pure traversal → misc
@@ -1067,6 +1096,23 @@ def test_every_room_fill_mode_the_editor_offers_is_accepted():
 # ---------- attachments & inner limits (HP-1454-02, -05) ----------
 
 
+def test_atomic_write_keeps_destination_and_cleans_temp_when_replace_fails(
+    tmp_path, monkeypatch,
+):
+    target = tmp_path / "plan.svg"
+    target.write_bytes(b"old")
+
+    def fail_replace(_source, _target):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(plans.os, "replace", fail_replace)
+    with pytest.raises(OSError, match="replace failed"):
+        plans.atomic_write(target, b"new", prefix=".plan-upload-")
+
+    assert target.read_bytes() == b"old"
+    assert list(tmp_path.glob(".plan-upload-*")) == []
+
+
 def test_reserve_filename_claims_the_name_atomically(tmp_path):
     """HP-1460-01: choosing a name and taking it must be one operation.
 

tests_backend/test_virtual_lights.py

@@ -23,6 +23,7 @@ async_reconcile_virtual_lights = _vl.async_reconcile_virtual_lights
 async_toggle_virtual_light = _vl.async_toggle_virtual_light
 async_virtual_light_snapshot = _vl.async_virtual_light_snapshot
 eligible_virtual_light_ids = _vl.eligible_virtual_light_ids
+VirtualLightController = _vl.VirtualLightController
 
 
 class FakeStore:
@@ -94,3 +95,23 @@ def test_toggle_accepts_only_an_id_and_inverts_server_current_state():
     assert first == {"marker_id": "lamp", "on": False, "rev": 1}
     assert second == {"marker_id": "lamp", "on": True, "rev": 2}
     assert _run(async_toggle_virtual_light(store, config, 1, "missing")) is None
+
+
+def test_runtime_controller_coalesces_rapid_toggles_into_one_durable_write():
+    async def exercise():
+        old_delay = _vl.SAVE_DELAY_S
+        _vl.SAVE_DELAY_S = 0.01
+        try:
+            store = FakeStore()
+            controller = VirtualLightController(store)
+            first = await controller.async_toggle(_config(_manual("lamp")), 1, "lamp")
+            second = await controller.async_toggle(_config(_manual("lamp")), 1, "lamp")
+            await controller.async_flush()
+            return first, second, store
+        finally:
+            _vl.SAVE_DELAY_S = old_delay
+
+    first, second, store = _run(exercise())
+    assert first == {"marker_id": "lamp", "on": False, "rev": 1}
+    assert second == {"marker_id": "lamp", "on": True, "rev": 2}
+    assert store.writes == [{"rev": 2, "config_rev": 1, "off": []}]