docs: review document for #335

Issue: #335
User-Visible: no

custom_components/houseplan/__init__.py

@@ -239,12 +239,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: HouseplanConfigEntry) ->
         hass.bus.async_fire("houseplan_config_updated", {"rev": optimize_revs[0]})
         hass.bus.async_fire("houseplan_layout_updated", {"rev": optimize_revs[1]})
         if recovered_import:
-            await recorder.async_refresh()
             current = (await data.config_store.async_load() or {}).get("config") or {}
-            live_ids = {str(marker.get("id")) for marker in current.get("markers") or []}
-            for marker_id in list(recorder.book.data):
-                if marker_id not in live_ids:
-                    await recorder.async_delete(marker_id)
+            await recorder.async_purge_orphans(current)
+            await recorder.async_refresh()
 
     await async_check_plan_files(hass, entry)
 

custom_components/houseplan/trails.py

@@ -187,8 +187,11 @@ class TrailRecorder:
             # just finished calibrating) must start recording NOW, not at the
             # next state change — otherwise the first seconds of the path are
             # lost.
+            now = time.time()
+            changed = False
             for src in self.pairs:
-                self._sample(src, time.time())
+                changed |= self._sample(src, now)
+            self._handle_sample_change(changed, now)
 
     def _source_failure_reason(self, source: str) -> str | None:
         """Classify only refresh-time health evidence.
@@ -244,25 +247,70 @@ class TrailRecorder:
 
     async def async_delete(self, marker: str) -> bool:
         """Stop and erase one marker without racing subscription refresh/save."""
+        return bool(await self._async_delete_many({marker}))
+
+    async def async_purge_orphans(self, config: dict[str, Any]) -> int:
+        """Erase trails whose marker is absent or a removal tombstone.
+
+        A tombstone deliberately stays in config so discovery cannot resurrect
+        a deleted device.  For live tracking and trail ownership it is absent:
+        this is the same boundary used by ``async_refresh`` above.
+        """
+        live_marker_ids = {
+            str(marker.get("id"))
+            for marker in config.get("markers") or []
+            if marker.get("id") is not None and marker.get("removed") is not True
+        }
+        orphan_ids = set(self.book.data) - live_marker_ids
+        if not orphan_ids:
+            return 0
+        try:
+            return await self._async_delete_many(orphan_ids)
+        except Exception:  # noqa: BLE001 — config commit already succeeded
+            _LOGGER.exception(
+                "House Plan: removing orphan vacuum trails failed: markers=%s",
+                sorted(orphan_ids),
+            )
+            return 0
+
+    async def _async_delete_many(self, markers: set[str]) -> int:
+        """Delete one or more books with one subscription/store transaction."""
         async with self._refresh_lock:
             # The trail book owns deletion. When it has no such marker, this
             # is a no-op and must not silently damage the live tracking graph.
-            removed = self.book.delete(marker)
+            removed = {
+                marker: self.book.data.pop(marker)
+                for marker in markers
+                if marker in self.book.data
+            }
             if not removed:
-                return False
-            for src in list(self.pairs):
-                kept = [pair for pair in self.pairs[src] if pair[0] != marker]
-                if kept:
-                    self.pairs[src] = kept
-                else:
-                    del self.pairs[src]
-            self._resubscribe()
-            if self._unsub_save:
-                self._unsub_save()
-                self._unsub_save = None
-            await self.store.async_save(self.book.data)
+                return 0
+            previous_pairs = {src: list(pairs) for src, pairs in self.pairs.items()}
+            had_pending_save = self._unsub_save is not None
+            try:
+                for src in list(self.pairs):
+                    kept = [pair for pair in self.pairs[src] if pair[0] not in removed]
+                    if kept:
+                        self.pairs[src] = kept
+                    else:
+                        del self.pairs[src]
+                self._resubscribe()
+                if self._unsub_save:
+                    self._unsub_save()
+                    self._unsub_save = None
+                await self.store.async_save(self.book.data)
+            except Exception:
+                # The store is the durable authority. Restore the in-memory
+                # owner graph so the next successful config sync can retry
+                # instead of leaving an orphan on disk forever (#335).
+                self.book.data.update(removed)
+                self.pairs = previous_pairs
+                self._resubscribe()
+                if had_pending_save:
+                    self._schedule_save()
+                raise
         self.hass.bus.async_fire("houseplan_trail_updated", {})
-        return True
+        return len(removed)
 
     def _resubscribe(self) -> None:
         """Replace the state subscription for the current pair graph."""
@@ -344,6 +392,10 @@ class TrailRecorder:
         for src, pair_list in self.pairs.items():
             if eid == src or any(eid == vac for _, vac in pair_list):
                 changed |= self._sample(src, now)
+        self._handle_sample_change(changed, now)
+
+    def _handle_sample_change(self, changed: bool, now: float) -> None:
+        """Persist and announce one logical sampling pass when it changed."""
         if changed:
             self._schedule_save()
             if now - self._last_fire >= FIRE_THROTTLE_S:

custom_components/houseplan/websocket_api.py

@@ -493,24 +493,14 @@ async def ws_import_apply(hass: HomeAssistant, connection, msg: dict[str, Any])
 
     hass.bus.async_fire("houseplan_config_updated", {"rev": new_config_rev})
     hass.bus.async_fire("houseplan_layout_updated", {"rev": new_layout_rev})
-    _refresh_trail_recorder(hass)
     if kind == "full":
-        recorder = hass.data.get(DOMAIN, {}).get("trail_recorder")
-        live_marker_ids = {
-            str(marker.get("id")) for marker in target_config.get("markers") or []
-        }
-        if recorder is not None:
-            for marker_id in list(getattr(getattr(recorder, "book", None), "data", {})):
-                if marker_id not in live_marker_ids:
-                    try:
-                        await recorder.async_delete(marker_id)
-                    except Exception:  # noqa: BLE001
-                        _LOGGER.exception("House Plan: removing orphan import trail failed")
+        await _purge_trail_recorder(hass, target_config)
         entry = get_entry(hass)
         if entry is not None:
             from .repairs import async_check_plan_files
 
             hass.async_create_task(async_check_plan_files(hass, entry))
+    _refresh_trail_recorder(hass)
     connection.send_result(msg["id"], {
         "ok": True,
         "kind": kind,
@@ -1412,6 +1402,11 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) ->
             await hass.async_add_executor_job(_collect)
         except Exception:  # noqa: BLE001 — see above: the commit stands regardless
             _LOGGER.exception("House Plan: collecting superseded files failed")
+        # The config is already durable, so trail cleanup is best-effort and
+        # cannot turn this accepted write into a retryable client failure.
+        # Keep it under write_lock: a later config/set must not resurrect a
+        # marker between this commit and the ownership decision (#335).
+        await _purge_trail_recorder(hass, msg["config"])
     hass.bus.async_fire("houseplan_config_updated", {"rev": new_rev})
     _refresh_trail_recorder(hass)
     # refresh repair issues (broken plan references) without waiting for a restart
@@ -1867,24 +1862,14 @@ async def ws_plan_optimize_undo(hass: HomeAssistant, connection, msg: dict[str,
 
     hass.bus.async_fire("houseplan_config_updated", {"rev": new_config_rev})
     hass.bus.async_fire("houseplan_layout_updated", {"rev": new_layout_rev})
-    _refresh_trail_recorder(hass)
     if restored_kind == "import":
-        recorder = hass.data.get(DOMAIN, {}).get("trail_recorder")
-        live_marker_ids = {
-            str(marker.get("id")) for marker in restored_config.get("markers") or []
-        }
-        if recorder is not None:
-            for marker_id in list(getattr(getattr(recorder, "book", None), "data", {})):
-                if marker_id not in live_marker_ids:
-                    try:
-                        await recorder.async_delete(marker_id)
-                    except Exception:  # noqa: BLE001
-                        _LOGGER.exception("House Plan: removing orphan undo trail failed")
+        await _purge_trail_recorder(hass, restored_config)
         entry = get_entry(hass)
         if entry is not None:
             from .repairs import async_check_plan_files
 
             hass.async_create_task(async_check_plan_files(hass, entry))
+    _refresh_trail_recorder(hass)
     connection.send_result(msg["id"], {
         "ok": True,
         "config_rev": new_config_rev,
@@ -1968,6 +1953,12 @@ def _refresh_trail_recorder(hass: HomeAssistant) -> None:
         hass.async_create_task(rec.async_refresh())
 
 
+async def _purge_trail_recorder(hass: HomeAssistant, config: dict[str, Any]) -> int:
+    """Reconcile durable trails with the live marker set after a config commit."""
+    rec = hass.data.get(DOMAIN, {}).get("trail_recorder")
+    return await rec.async_purge_orphans(config) if rec else 0
+
+
 @websocket_api.websocket_command({vol.Required("type"): "houseplan/trail/get"})
 @websocket_api.async_response
 async def ws_trail_get(hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict) -> None:

docs/CHANGELOG.md

@@ -7,6 +7,12 @@
   the inner-face convergence accounts for both thicknesses, not just the
   larger one ([#339](https://github.com/Matysh/houseplan-card/issues/339)).
 
+- Vacuum trails now reconcile with ordinary plan edits on the server: deleting
+  a marker also removes its stored runs even if the browser-side cleanup was
+  interrupted, and a position sampled during Home Assistant startup is saved
+  instead of remaining memory-only
+  ([#335](https://github.com/Matysh/houseplan-card/issues/335)).
+
 ## v1.69.0-beta.1 — 2026-08-28
 
 - House Plan now has a complete German interface. `Deutsch` can be selected

docs/CHANGELOG.ru.md

@@ -13,6 +13,12 @@
   смыкания внутренних граней учитывает обе толщины, а не одну наибольшую
   ([#339](https://github.com/Matysh/houseplan-card/issues/339)).
 
+- Серверные трейлы пылесосов теперь согласуются с обычным редактированием
+  плана: удаление маркера удаляет и сохранённые маршруты, даже если клиентская
+  очистка прервалась, а точка, полученная при запуске Home Assistant,
+  сохраняется на диск и не остаётся только в памяти
+  ([#335](https://github.com/Matysh/houseplan-card/issues/335)).
+
 ## v1.69.0-beta.1 — 2026-08-28
 
 - В House Plan появилась полная немецкая локализация. `Deutsch` можно выбрать

docs/TESTING.md

@@ -2153,6 +2153,11 @@ require hands on real hardware — they remain for the human pass.
   source/vacuum pair stays subscribed. A successful delete removes only that
   marker's pairs and immediately rebuilds the subscription
   [backend: test_trail_recorder.py].
+- A successful config edit purges trails for both a missing marker and its
+  `removed: true` tombstone, but keeps live/hidden markers; a semantic no-op
+  performs no surprise cleanup. A startup refresh that samples a new point
+  schedules one save and one throttled update event
+  [backend: test_ha_websocket.py + test_trail_recorder.py].
 - Trail style: cartography casing (dark halo 2.25 + light core 0.9),
   readable over any room fill.
 - Hidden marker: neither puck nor trail. Uncalibrated active map: no puck.

docs/VACUUM.md

@@ -143,7 +143,11 @@ marker.vacuum = {
 All fields are optional and old plans remain readable. Hiding retains the
 configuration. Deleting a vacuum marker removes its layout and server trails,
 creates the normal removal tombstone and makes the HA device available for a
-fresh add without resurrecting old runs.
+fresh add without resurrecting old runs. The backend reconciles both a removal
+tombstone and a completely absent marker with the trail store after every
+successful config change, so an interrupted browser-side cleanup is repaired.
+An initial position sampled during integration startup follows the same
+debounced persistence and live-update path as a later state event.
 
 ## Troubleshooting
 

docs/reviews/CODE-REVIEW-335-r1.md

@@ -0,0 +1,239 @@
+# CODE-REVIEW-335-r1
+
+Issue: #335 «Трейлы пылесосов: осиротевшие маркеры навсегда остаются в
+store, точки после рестарта не сохраняются» · этап code · заход r1 ·
+блокирующих циклов израсходовано 0 из 2
+
+SHA под ревью: `1d818905` (после rebase на `origin/dev`, конфликт был
+только в двух changelog — см. комментарий владельца в issue).
+`origin/dev` на момент ревью: `dd093625`.
+
+## Скоуп
+
+Задача — light track (small), ТЗ живёт в теле issue #335, прошло
+SPEC-REVIEW r1 (жёлтый) → r2 (зелёный, `docs/reviews/SPEC-REVIEW-335-r2.md`).
+Один issue-коммит `1d818905` «Fix vacuum trail lifecycle persistence»
+(`Issue: #335`, `User-Visible: yes`), диапазон `origin/dev..HEAD`:
+
+```
+custom_components/houseplan/__init__.py      |   7 +-
+custom_components/houseplan/trails.py        |  82 ++++++++++++++----
+custom_components/houseplan/websocket_api.py |  39 ++++-----
+docs/CHANGELOG.md                            |   6 ++
+docs/CHANGELOG.ru.md                         |   6 ++
+docs/TESTING.md                              |   5 ++
+docs/VACUUM.md                               |   6 +-
+tests_backend/test_ha_websocket.py           |  68 +++++++++++++++
+tests_backend/test_trail_recorder.py         | 121 +++++++++++++++++++++++++++
+9 files changed, 295 insertions(+), 45 deletions(-)
+```
+
+Одна backend-поверхность (`TrailRecorder` + три websocket-хендлера +
+`__init__.py`), без миграции конфига, без изменений `src/**`, без
+геометрии. Соответствует заявленному small-треку.
+
+Продуктовая рамка (`docs/SCOPE.md`): серверные трейлы — часть J1
+(«live spatial overview») через `docs/VACUUM.md`. Исправление
+устраняет утечку хранения и потерю данных о живом положении —
+внутри J1, новых продуктовых поверхностей не добавляет.
+
+## Как проверялось
+
+Прочитан полный `git diff origin/dev...HEAD` (500 строк) построчно,
+включая `custom_components/houseplan/trails.py` целиком (414 строк) и
+контекст всех трёх мест вызова в `websocket_api.py` (`ws_config_set`,
+`ws_import_apply`, `ws_plan_optimize_undo`) и `__init__.py`
+(`recovered_import`). Сверено с `docs/VACUUM.md` (новый абзац) и с
+телом issue (AC1–AC4).
+
+### Гейты — что прогнано и результат
+
+| Гейт | Статус | Результат |
+|---|---|---|
+| `npx tsc --noEmit` | прогнан | чисто, без вывода |
+| `python -m pytest tests_backend/test_trail_recorder.py -q` (pure, без HA) | прогнан | 30 passed |
+| `python -m pytest tests_backend/test_ha_websocket.py -q` (HA harness, `pytest-homeassistant-custom-component` установлен в этом раунде) | прогнан | 61 passed, 1 error |
+| `npm test` | прогнан | 1461 passed, 0 failed, 1 skipped |
+| `npm run build` + сверка бандла | прогнан | `git status` после build чист — три копии бандла совпадают |
+| `git diff --check` | прогнан | чисто |
+
+**Про 1 error в `test_ha_websocket.py`.** Падает
+`test_issue_244_space_delete_is_authoritative_and_revision_guarded` —
+тест не про пылесосов и не тронут диффом (пространства/маркер
+`virtual`). Ошибка — `AssertionError` в teardown fixture `hass`
+(`pytest-homeassistant-custom-component`) о постороннем потоке
+`_run_safe_shutdown_loop`, не о ассертах теста. Перепроверено:
+тот же прогон на чистом `origin/dev` (`dd093625`, отдельный
+`git worktree`) даёт идентичную ошибку — `60 passed, 1 error`, тот же
+трейсбек. Это окружение-специфичный флейк текущего Linux-раннера, не
+регрессия этого диффа. Целевой новый тест
+`test_config_set_purges_tombstoned_and_absent_trails_durably` вошёл в
+61 passed.
+
+### Гейты — что не прогонялось и почему
+
+- `node scripts/check-docs.mjs` — не требуется: диф не трогает
+  `src/**` (только `custom_components/**/*.py` и docs).
+- `npm run invariants` — не требуется: диф не трогает геометрию,
+  `layout`, `marker.space`, `open_spans`, рёбра комнат.
+- Browser-smoke (`demo/smoke_*.mjs`) — проверено инструментом:
+  `node scripts/smoke-select.mjs --base origin/dev --head HEAD` →
+  «Исполняемого frontend-диффа нет (`src/**/*.ts` не тронут).
+  Browser-smoke этим диффом не выбираются». Смоки гоняют собранную
+  карточку, диф в неё не попадает.
+- `npm run golden:verify` — не требуется: диф не меняет рендер,
+  геометрию, стили или слои (чистый backend).
+- «Одно число — один источник»: диф не добавляет и не меняет ни
+  одной пользовательски видимой величины (никаких новых чисел на
+  экране), проверка неприменима.
+
+## Разбор по AC
+
+**AC1 — очистка при `config/set`.**
+Реализовано общим методом `TrailRecorder.async_purge_orphans(config)`
+(`trails.py:252-274`): живой marker-id = «`id` присутствует и
+`removed is not True»`, ровно граница, которую уже использует
+`async_refresh` (`trails.py:169-171`) и `import_export.live_layout`
+(`import_export.py:175-177`). Вызывается из `ws_config_set`
+(`websocket_api.py:1409`) **после** `async_save_config_state` (durable
+write уже совершён) и **всё ещё под `rt.write_lock`** — комментарий в
+коде (`websocket_api.py:1405-1408`) явно называет причину: не дать
+следующему `config/set` воскресить маркер между коммитом и решением
+об удалении трейла. Доказано тестом
+`test_config_set_purges_tombstoned_and_absent_trails_durably`
+(`test_ha_websocket.py:291+`): удалены и tombstone (`removed: true`),
+и полностью отсутствующий id (`hard_drop`), проверены и in-memory
+`recorder.book.data`, и долговечный `recorder.store` — обе половины
+контракта из ТЗ («marker-id отсутствует в памяти recorder и в
+долговечном trail store»). Тест умеет падать: без правки старый код
+проверял только полное отсутствие `id` (`live_marker_ids` без
+фильтра `removed`), tombstone остался бы в `recorder.book.data`, и
+`assert set(recorder.book.data) == {"live", "hidden"}` не прошёл бы.
+
+**AC2 — никаких побочных удалений.**
+`live`/`hidden` (без `removed: true`) сохраняются — то же тест
+подтверждает построчно. No-op write не запускает очистку: прочитано
+по коду, не на слово автора — семантический no-op в `ws_config_set`
+возвращается на `websocket_api.py:1375` **до** присвоения `new_rev` и
+до `async_save_config_state`, то есть до строки с покупкой purge
+(`1409`); ветка ошибки валидации/`missing_plan`/`conflict` возвращает
+раньше по коду ещё сильнее. Дополнительно доказано тем же
+websocket-тестом (вторая часть, `test_ha_websocket.py:344-356`):
+повторная отправка того же `candidate` с прежним `expected_rev`
+получает `noop["result"]["rev"] == removed["result"]["rev"]`, а
+искусственно добавленный `late_orphan` остаётся и в памяти, и в
+сторе — прямое доказательство, что no-op не чистит. Неуспешная запись
+(ошибка валидации) не покрыта отдельным websocket-тестом, но граница
+доказана чтением кода (задокументировано выше) — приемлемо, так как
+она структурная (ранний `return`), а не условная логика, которую
+легко сломать будущей правкой незаметно.
+`hidden`-без-`removed:true` не считается удалением — подтверждено тем
+же тестом (маркер `hidden` в обоих сравнениях). `disabled` в контракте
+относится к HA-статусу источника (`_source_failure_reason`), не к
+полю конфига маркера — `async_purge_orphans` его не читает вовсе,
+семантика верна по построению.
+
+**AC3 — долговечность refresh-точки.**
+`async_refresh()` теперь агрегирует `changed` по всем `src` в одном
+проходе и вызывает общий `_handle_sample_change(changed, now)`
+(`trails.py:190-194`, `397-403`) — тот же метод, которым пользуется
+`_on_state`. Доказано pure-тестом
+`test_refresh_persists_and_announces_a_new_startup_sample_once`
+(`test_trail_recorder.py:370-411`, входит в 30 passed): первый refresh
+с новой точкой планирует ровно одно сохранение (`SAVE_DELAY_S`) и
+шлёт ровно одно `houseplan_trail_updated`; повторный refresh без
+изменений не планирует второе и не шлёт второе событие. Тест умеет
+падать: до правки `async_refresh` вызывал `self._sample(...)` без
+использования результата и без вызова `_handle_sample_change` —
+`scheduled` и `hass.bus.fired` остались бы пустыми, оба ассерта не
+прошли бы.
+
+**AC4 — единая семантика и регрессии.**
+Общий helper `async_purge_orphans`/`_async_delete_many` используется
+во всех четырёх местах: `ws_config_set`, `ws_import_apply` (kind
+`full`), `ws_plan_optimize_undo` (`restored_kind == "import"`) и
+`__init__.py` recovered_import при старте (`async_purge_orphans` →
+`async_refresh`, тот же порядок, что и в websocket-хендлерах: purge
+до refresh — согласовано). Прежний дублирующийся цикл `for marker_id
+in ... if marker_id not in live_marker_ids: await
+recorder.async_delete(marker_id)` (по одному «store.async_save» на
+маркер, без единой границы `removed`) заменён везде на один вызов.
+Регрессии: `test_trail_delete_prunes_pair_and_replaces_subscription`
+(явный `houseplan/trail/delete`, `async_delete` теперь тонкая обёртка
+над `_async_delete_many`) и вся остальная сюита backend + frontend
+зелёные (см. таблицу гейтов). Откат/повтор при сбое стора покрыт
+новым `test_failed_orphan_store_write_rolls_back_and_can_be_retried`
+(`test_trail_recorder.py`, входит в 30 passed) — при ошибке
+`store.async_save` `book.data`/`pairs`/подписка откатываются, событие
+не летит, повторный вызов успешен. Это не было отдельным AC, но
+предотвращает конкретный сценарий регрессии («сбой записи стора при
+purge насовсем теряет живой маркер») — засчитано как часть «единой
+семантики», не отдельная находка.
+
+## Что проверено и корректно (не в составе AC, но задето диффом)
+
+- Единственность источника собранного конфига для purge:
+  `ws_config_set` передаёт в `_purge_trail_recorder` тот же
+  провалидированный объект `msg["config"]`, что и в
+  `async_save_config_state` (после `msg["config"].clear();
+  msg["config"].update(checked)` на `websocket_api.py:1326-1327`) —
+  проверено чтением, drift между сохранённым и очищаемым конфигом
+  невозможен структурно.
+  `ws_import_apply`/`ws_plan_optimize_undo` аналогично используют
+  `target_config`/`restored_config` — те же объекты, что были
+  закоммичены в `_commit_import_pair`.
+- `docs/VACUUM.md` обновлён в том же коммите ровно тем текстом,
+  который описывает новый контракт («backend reconciles both a
+  removal tombstone and a completely absent marker… after every
+  successful config change… An initial position sampled during
+  integration startup follows the same debounced persistence and
+  live-update path as a later state event») — соответствует коду.
+- `docs/TESTING.md` дополнение точно называет оба файла доказательства
+  (`test_ha_websocket.py` + `test_trail_recorder.py`) и описывает
+  ровно то поведение, которое тесты проверяют.
+- Трейлеры коммита: `Issue: #335`, `User-Visible: yes`; оба changelog
+  (`docs/CHANGELOG.md`, `docs/CHANGELOG.ru.md`) в том же коммите
+  `1d818905`, формулировки согласованы между языками и с телом ТЗ.
+- Порядок `purge → refresh` (не наоборот) согласован во всех четырёх
+  местах вызова — не было явно потребовано ТЗ, но убирает
+  потенциальный разнобой между вызовами.
+
+## Находки
+
+Нет находок серьёзности High или Medium. Ниже — только замеченная,
+но не блокирующая асимметрия, снятая без правки.
+
+- В `ws_config_set` purge держится под `rt.write_lock` с явным
+  комментарием о защите от гонки «commit → purge»; в
+  `ws_import_apply`/`ws_plan_optimize_undo` purge вызывается уже
+  **после** выхода из `write_lock` (как и было устроено в прежнем
+  коде — сам факт блокировки на время цикла удаления там никогда не
+  держался). Диф не увеличивает и не уменьшает это несоответствие
+  относительно `origin/dev`, ТЗ не называет его в скоупе (импорт/undo
+  — не «обычное редактирование» из «Проблема» №1), новых наблюдаемых
+  дефектов эта разница не создаёт при обычном однопользовательском
+  сценарии. Снимаю без правки: не регрессия этого диффа, вне
+  предмета AC1–AC4.
+
+## Чего не проверял
+
+- Ручной прогон в реальном Home Assistant (WSL/CI harness автора) —
+  недоступен в этом окружении; вместо него — полный HA-websocket-тест
+  через `pytest-homeassistant-custom-component`, установленный в этом
+  раунде (см. таблицу гейтов).
+- Многопользовательская гонка «второй клиент успевает воскресить
+  маркер между commit и purge» для `import`/`undo`-путей (см. находку
+  выше) — не воспроизводилась вручную, только прочитана по коду;
+  вне скоупа AC.
+- Perf-профили — не запрашивались AC, диф не касается путей,
+  чувствительных к производительности (debounce/throttle интервалы
+  прямо названы «вне скоупа» в ТЗ и не изменены).
+
+## Вердикт
+
+Зелёный. Все четыре AC доказаны — либо автотестом, который умеет
+падать без правки, либо чтением кода с явной пометкой. Регрессионный
+периметр (существующий `trail/delete`, живые/скрытые маркеры, полный
+frontend + pure backend) зелёный. Один флейк в
+`test_ha_websocket.py` подтверждён как окруженческий и не связанный с
+диффом (воспроизведён на `origin/dev` без изменений #335).

tests_backend/test_ha_websocket.py

@@ -367,6 +367,74 @@ async def test_trail_delete_rejects_non_admin_without_mutation(
     assert "m1" in recorder.book.data
 
 
+async def test_config_set_purges_tombstoned_and_absent_trails_durably(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """#335 AC1/AC2: the durable config owns the durable trail book."""
+    await _setup(hass)
+    client = await hass_ws_client(hass)
+    initial = {
+        "spaces": [],
+        "markers": [
+            {"id": "tombstone", "binding": "entity:vacuum.tombstone"},
+            {"id": "hard_drop", "binding": "entity:vacuum.hard_drop"},
+            {"id": "live", "binding": "entity:vacuum.live"},
+            {
+                "id": "hidden", "binding": "entity:vacuum.hidden",
+                "hidden": True,
+            },
+        ],
+        "settings": {},
+    }
+    await client.send_json_auto_id({
+        "type": "houseplan/config/set", "config": initial, "expected_rev": 0,
+    })
+    first = await client.receive_json()
+    assert first["success"]
+    await hass.async_block_till_done()
+
+    recorder = hass.data[DOMAIN]["trail_recorder"]
+    recorder.book.data = {
+        marker_id: {"current": {"points": [[index, index + 1]]}}
+        for index, marker_id in enumerate(("tombstone", "hard_drop", "live", "hidden"))
+    }
+    await recorder.store.async_save(copy.deepcopy(recorder.book.data))
+
+    candidate = copy.deepcopy(initial)
+    candidate["markers"] = [
+        {
+            "id": "tombstone", "binding": "entity:vacuum.tombstone",
+            "removed": True, "hidden": True,
+        },
+        {"id": "live", "binding": "entity:vacuum.live"},
+        {
+            "id": "hidden", "binding": "entity:vacuum.hidden",
+            "hidden": True,
+        },
+    ]
+    await client.send_json_auto_id({
+        "type": "houseplan/config/set", "config": candidate,
+        "expected_rev": first["result"]["rev"],
+    })
+    removed = await client.receive_json()
+    assert removed["success"]
+    assert set(recorder.book.data) == {"live", "hidden"}
+    assert set(await recorder.store.async_load() or {}) == {"live", "hidden"}
+
+    # A semantic no-op is not a lifecycle transition and must not perform a
+    # surprise cleanup. A later real config commit will reconcile this orphan.
+    recorder.book.data["late_orphan"] = {"current": {"points": [[9, 10]]}}
+    await recorder.store.async_save(copy.deepcopy(recorder.book.data))
+    await client.send_json_auto_id({
+        "type": "houseplan/config/set", "config": candidate,
+        "expected_rev": removed["result"]["rev"],
+    })
+    noop = await client.receive_json()
+    assert noop["success"] and noop["result"]["rev"] == removed["result"]["rev"]
+    assert "late_orphan" in recorder.book.data
+    assert "late_orphan" in (await recorder.store.async_load() or {})
+
+
 async def test_config_rev_conflict(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
     await _setup(hass)
     client = await hass_ws_client(hass)

tests_backend/test_trail_recorder.py

@@ -119,6 +119,49 @@ def test_sample_seeds_a_run_already_in_progress():
     assert rec.book.data["m1"]["current"]["points"] == [[1000.0, -500.0]]
 
 
+def test_refresh_persists_and_announces_a_new_startup_sample_once():
+    scheduled = []
+    saved = []
+    old_call_later = trails.async_call_later
+    trails.async_call_later = lambda hass, delay, cb: (
+        scheduled.append((delay, cb)) or (lambda: None)
+    )
+    try:
+        class CS:
+            async def async_load(self):
+                return {"config": {"markers": [{
+                    "id": "m1",
+                    "binding": "entity:vacuum.x50",
+                    "vacuum": {"source": "camera.map"},
+                }]}}
+
+        class RT:
+            config_store = CS()
+
+        class TrailStore:
+            async def async_save(self, data):
+                saved.append(json.loads(json.dumps(data)))
+
+        rec, hass, _states = _rec()
+        rec.rt = RT()
+        rec.store = TrailStore()
+        rec.pairs = {}
+
+        _run_isolated(rec.async_refresh())
+        assert len(scheduled) == 1
+        assert scheduled[0][0] == trails.SAVE_DELAY_S
+        assert hass.bus.fired == [("houseplan_trail_updated", {})]
+        _run_isolated(scheduled[0][1](None))
+        assert saved[-1]["m1"]["current"]["points"] == [[1000.0, -500.0]]
+
+        # The same point is a true no-op: no second save or live-card event.
+        _run_isolated(rec.async_refresh())
+        assert len(scheduled) == 1
+        assert hass.bus.fired == [("houseplan_trail_updated", {})]
+    finally:
+        trails.async_call_later = old_call_later
+
+
 def test_junk_position_ignored():
     rec, _hass, states = _rec()
     states["camera.map"] = S("idle", {"vacuum_position": {"x": "nope", "y": 1}})
@@ -173,6 +216,84 @@ def test_trail_delete_prunes_pair_and_replaces_subscription():
         trails.async_track_state_change_event = old_track
 
 
+def test_orphan_purge_treats_removed_as_absent_and_batches_one_store_write():
+    rec, hass, _states = _rec()
+    rec.book.data = {
+        "tombstone": {"current": {"points": [[1, 2]]}},
+        "hard_drop": {"current": {"points": [[3, 4]]}},
+        "live": {"current": {"points": [[5, 6]]}},
+        "hidden": {"current": {"points": [[7, 8]]}},
+    }
+    rec.pairs = {
+        "camera.map": [
+            ("tombstone", "vacuum.x50"),
+            ("hard_drop", "vacuum.x50"),
+            ("live", "vacuum.x50"),
+            ("hidden", "vacuum.x50"),
+        ]
+    }
+    saved = []
+
+    class TrailStore:
+        async def async_save(self, data):
+            saved.append(json.loads(json.dumps(data)))
+
+    rec.store = TrailStore()
+    removed = _run_isolated(rec.async_purge_orphans({"markers": [
+        {"id": "tombstone", "removed": True},
+        {"id": "live"},
+        {"id": "hidden", "hidden": True},
+    ]}))
+
+    assert removed == 2
+    assert set(rec.book.data) == {"live", "hidden"}
+    assert rec.pairs == {
+        "camera.map": [("live", "vacuum.x50"), ("hidden", "vacuum.x50")]
+    }
+    assert len(saved) == 1
+    assert set(saved[0]) == {"live", "hidden"}
+    assert hass.bus.fired == [("houseplan_trail_updated", {})]
+
+
+def test_failed_orphan_store_write_rolls_back_and_can_be_retried():
+    rec, hass, _states = _rec()
+    rec.book.data = {
+        "orphan": {"current": {"points": [[1, 2]]}},
+        "live": {"current": {"points": [[3, 4]]}},
+    }
+    rec.pairs = {
+        "camera.map": [
+            ("orphan", "vacuum.x50"),
+            ("live", "vacuum.x50"),
+        ]
+    }
+
+    class FailingStore:
+        async def async_save(self, _data):
+            raise OSError("disk full")
+
+    rec.store = FailingStore()
+    assert _run_isolated(rec.async_purge_orphans({"markers": [{"id": "live"}]})) == 0
+    assert set(rec.book.data) == {"orphan", "live"}
+    assert rec.pairs["camera.map"] == [
+        ("orphan", "vacuum.x50"),
+        ("live", "vacuum.x50"),
+    ]
+    assert hass.bus.fired == []
+
+    saved = []
+
+    class WorkingStore:
+        async def async_save(self, data):
+            saved.append(json.loads(json.dumps(data)))
+
+    rec.store = WorkingStore()
+    assert _run_isolated(rec.async_purge_orphans({"markers": [{"id": "live"}]})) == 1
+    assert set(rec.book.data) == {"live"}
+    assert set(saved[-1]) == {"live"}
+    assert hass.bus.fired == [("houseplan_trail_updated", {})]
+
+
 def test_object_style_position_is_read():
     # Tasshack in-memory attributes hold a Point OBJECT, not a dict
     class Point: