"""TrailRecorder wiring: subscription callback, dialects, run end.

Loads trails.py with STUBBED Home Assistant modules — but only inside a
snapshot of sys.modules that is restored immediately afterwards. Injecting
fake `homeassistant` modules globally poisons the real HA harness running in
the same pytest session (it did, once).
"""
import sys, types, pathlib, importlib.util, json

ROOT = pathlib.Path(__file__).resolve().parent.parent


def _load_trails():
    saved = {k: v for k, v in sys.modules.items() if k == "homeassistant" or k.startswith(("homeassistant.", "houseplan"))}
    try:
        for name in list(sys.modules):
            if name == "homeassistant" or name.startswith(("homeassistant.", "houseplan")):
                del sys.modules[name]
        ha = types.ModuleType("homeassistant"); sys.modules["homeassistant"] = ha
        core = types.ModuleType("homeassistant.core")
        core.HomeAssistant = object
        core.callback = lambda f: f
        sys.modules["homeassistant.core"] = core
        sys.modules["homeassistant.helpers"] = types.ModuleType("homeassistant.helpers")
        er = types.ModuleType("homeassistant.helpers.entity_registry")
        er.async_get = lambda hass: None
        er.async_entries_for_device = lambda reg, dev: []
        sys.modules["homeassistant.helpers.entity_registry"] = er
        ev = types.ModuleType("homeassistant.helpers.event")
        ev.async_call_later = lambda hass, delay, cb: (lambda: None)
        ev.async_track_state_change_event = lambda hass, ents, cb: (lambda: None)
        sys.modules["homeassistant.helpers.event"] = ev
        stm = types.ModuleType("homeassistant.helpers.storage")

        class _Store:
            def __init__(self, *a, **k): pass
            async def async_load(self): return None
            async def async_save(self, d): pass

        stm.Store = _Store
        sys.modules["homeassistant.helpers.storage"] = stm
        pkg = types.ModuleType("houseplan"); pkg.__path__ = [str(ROOT / "custom_components" / "houseplan")]
        sys.modules["houseplan"] = pkg
        c = types.ModuleType("houseplan.const"); c.DOMAIN = "houseplan"
        sys.modules["houseplan.const"] = c
        spec = importlib.util.spec_from_file_location(
            "houseplan.trails", str(ROOT / "custom_components" / "houseplan" / "trails.py")
        )
        mod = importlib.util.module_from_spec(spec)
        sys.modules["houseplan.trails"] = mod
        spec.loader.exec_module(mod)
        return mod
    finally:
        for name in list(sys.modules):
            if name == "homeassistant" or name.startswith(("homeassistant.", "houseplan")):
                del sys.modules[name]
        sys.modules.update(saved)


trails = _load_trails()


class S:
    def __init__(self, state, attrs): self.state, self.attributes = state, attrs


class States:
    def __init__(self, d): self.d = d
    def get(self, k): return self.d.get(k)


class Bus:
    def __init__(self): self.fired = []
    def async_fire(self, *a): self.fired.append(a)


class Hass:
    def __init__(self, states): self.states, self.bus, self.data = States(states), Bus(), {}


def _rec():
    states = {
        "vacuum.x50": S("cleaning", {"selected_map": "Первый этаж"}),
        "camera.map": S("idle", {"vacuum_position": {"x": 1000, "y": -500, "a": 90}, "map_index": 1}),
    }
    hass = Hass(states)
    rec = trails.TrailRecorder(hass, None)
    rec.pairs = {"camera.map": [("m1", "vacuum.x50")]}
    return rec, hass, states


class E:
    def __init__(self, eid): self.data = {"entity_id": eid}


def test_state_events_record_points_and_map_id():
    rec, hass, states = _rec()
    rec._on_state(E("camera.map"))
    states["camera.map"] = S("idle", {"vacuum_position": {"x": 1100, "y": -500}, "map_index": 1})
    rec._on_state(E("camera.map"))
    run = rec.book.data["m1"]["current"]
    assert run["points"] == [[1000.0, -500.0], [1100.0, -500.0]]
    assert run["map_id"] == "1"
    assert hass.bus.fired, "live cards must be notified"


def test_docking_ends_the_run():
    rec, hass, states = _rec()
    rec._on_state(E("camera.map"))
    states["vacuum.x50"] = S("docked", {})
    rec._on_state(E("vacuum.x50"))
    assert rec.book.data["m1"]["current"]["ended"] is not None


def test_sample_seeds_a_run_already_in_progress():
    # HA restarted mid-cleanup: the first point must not wait for an event
    rec, _hass, _states = _rec()
    assert rec._sample("camera.map", 123.0)
    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}})
    assert not rec._sample("camera.map", 1.0)
    states["camera.map"] = S("idle", {})
    assert not rec._sample("camera.map", 1.0)


def test_unknown_source_is_noop():
    rec, _hass, _states = _rec()
    assert not rec._sample("camera.other", 1.0)


def test_trail_book_delete_forgets_current_and_previous_runs():
    book = trails.TrailBook({
        "m1": {"current": {"points": [[1, 2]]}, "previous": {"points": [[3, 4]]}},
        "m2": {"current": {"points": [[5, 6]]}},
    })
    assert book.delete("m1") is True
    assert "m1" not in book.data
    assert "m2" in book.data
    assert book.delete("m1") is False


def test_missing_trail_delete_does_not_mutate_live_tracking_pairs():
    rec, _hass, _states = _rec()
    before = {src: list(pairs) for src, pairs in rec.pairs.items()}
    unsubscribed = []
    rec._unsub_track = lambda: unsubscribed.append(True)
    assert _run_isolated(rec.async_delete("m1")) is False
    assert rec.pairs == before
    assert unsubscribed == []


def test_trail_delete_prunes_pair_and_replaces_subscription():
    tracked = []
    old_track = trails.async_track_state_change_event
    trails.async_track_state_change_event = lambda hass, ents, cb: (
        tracked.append(list(ents)) or (lambda: None)
    )
    try:
        rec, _hass, _states = _rec()
        rec.pairs["other.source"] = [("m2", "vacuum.other")]
        rec.book.data["m1"] = {"current": {"points": [[1, 2]]}}
        unsubscribed = []
        rec._unsub_track = lambda: unsubscribed.append(True)
        assert _run_isolated(rec.async_delete("m1")) is True
        assert rec.pairs == {"other.source": [("m2", "vacuum.other")]}
        assert unsubscribed == [True]
        assert tracked == [["other.source", "vacuum.other"]]
    finally:
        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 _route_book():
    """One live marker: a run filed under a route that is about to vanish, one to keep."""
    return {
        "live": {
            "current": {"route_id": "vr_old", "map_id": "1", "points": [[1, 2]]},
            "previous": {"route_id": "vr_keep", "map_id": "1", "points": [[3, 4]]},
        },
    }


def _live_marker_without_vr_old():
    return {
        "id": "live", "space": "ground",
        "vacuum": {"map_routes": [
            {"id": "vr_keep", "source": "camera.map", "map_id": "1", "space": "ground"},
        ]},
    }


class _RecordingStore:
    def __init__(self):
        self.saved = []

    async def async_save(self, data):
        self.saved.append(json.loads(json.dumps(data)))


def test_issue_495_dropped_route_runs_reach_the_store_without_orphans():
    """#495 AC3: deleting a route persists the drop even when no marker is orphaned."""
    rec, hass, _states = _rec()
    rec.book.data = _route_book()
    rec.pairs = {"camera.map": [("live", "vacuum.x50")]}
    rec.store = _RecordingStore()
    pending_cancelled = []
    rec._unsub_save = lambda: pending_cancelled.append(True)

    removed = _run_isolated(rec.async_purge_orphans({"markers": [_live_marker_without_vr_old()]}))

    assert removed == 0, "the return value still counts markers, not runs"
    assert len(rec.store.saved) == 1
    assert rec.store.saved[0] == {
        "live": {"previous": {"route_id": "vr_keep", "map_id": "1", "points": [[3, 4]]}},
    }
    assert pending_cancelled == [True] and rec._unsub_save is None
    assert hass.bus.fired == [("houseplan_trail_updated", {})]
    # Restart model: a fresh book from what the store holds has no vr_old run.
    restarted = trails.TrailBook(json.loads(json.dumps(rec.store.saved[-1])))
    assert restarted.data["live"].get("current") is None
    assert restarted.data["live"]["previous"]["route_id"] == "vr_keep"
    # No change → no write, no event.
    _run_isolated(rec.async_purge_orphans({"markers": [_live_marker_without_vr_old()]}))
    assert len(rec.store.saved) == 1
    assert hass.bus.fired == [("houseplan_trail_updated", {})]


def test_issue_495_dropped_route_runs_roll_back_when_the_store_write_fails():
    """#495 AC4: memory follows the store — a failed write keeps the run for a retry."""
    rec, hass, _states = _rec()
    rec.book.data = _route_book()
    rec.pairs = {"camera.map": [("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": [_live_marker_without_vr_old()]})) == 0
    assert rec.book.data == _route_book()
    assert hass.bus.fired == []

    rec.store = _RecordingStore()
    assert _run_isolated(rec.async_purge_orphans({"markers": [_live_marker_without_vr_old()]})) == 0
    assert rec.store.saved[-1]["live"] == {
        "previous": {"route_id": "vr_keep", "map_id": "1", "points": [[3, 4]]},
    }
    assert hass.bus.fired == [("houseplan_trail_updated", {})]


def test_issue_495_dropped_route_runs_share_the_orphan_transaction():
    """#495 AC5: orphans and dropped routes leave in one write with one event."""
    rec, hass, _states = _rec()
    rec.book.data = {**_route_book(), "orphan": {"current": {"points": [[5, 6]]}}}
    rec.pairs = {"camera.map": [("live", "vacuum.x50"), ("orphan", "vacuum.x50")]}
    rec.store = _RecordingStore()

    removed = _run_isolated(rec.async_purge_orphans({"markers": [_live_marker_without_vr_old()]}))

    assert removed == 1
    assert len(rec.store.saved) == 1
    assert rec.store.saved[0] == {
        "live": {"previous": {"route_id": "vr_keep", "map_id": "1", "points": [[3, 4]]}},
    }
    assert rec.pairs == {"camera.map": [("live", "vacuum.x50")]}
    assert hass.bus.fired == [("houseplan_trail_updated", {})]


def test_issue_495_failed_orphan_transaction_restores_dropped_route_runs_too():
    """#495 AC4 in the orphan path: rollback covers both halves of the write."""
    rec, hass, _states = _rec()
    rec.book.data = {**_route_book(), "orphan": {"current": {"points": [[5, 6]]}}}
    rec.pairs = {"camera.map": [("live", "vacuum.x50"), ("orphan", "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": [_live_marker_without_vr_old()]})) == 0
    assert rec.book.data == {**_route_book(), "orphan": {"current": {"points": [[5, 6]]}}}
    assert hass.bus.fired == []


def test_object_style_position_is_read():
    # Tasshack in-memory attributes hold a Point OBJECT, not a dict
    class Point:
        def __init__(self, x, y): self.x, self.y = x, y
    rec, _hass, states = _rec()
    states["camera.map"] = S("idle", {"vacuum_position": Point(2020, 3096), "map_index": 1})
    assert rec._sample("camera.map", 5.0)
    assert rec.book.data["m1"]["current"]["points"] == [[2020.0, 3096.0]]


def test_unavailable_vacuum_is_no_verdict():
    # HA boot: the vacuum reads unavailable — the open run must NOT be ended
    rec, _hass, states = _rec()
    rec._sample("camera.map", 1.0)
    states["vacuum.x50"] = S("unavailable", {})
    assert not rec._sample("camera.map", 2.0)
    assert rec.book.data["m1"]["current"]["ended"] is None
    del states["vacuum.x50"]
    assert not rec._sample("camera.map", 3.0)
    assert rec.book.data["m1"]["current"]["ended"] is None


def test_short_available_stops_resume_one_run_and_neutral_states_do_not_shift_window():
    rec, _hass, states = _rec()
    assert rec._sample("camera.map", 100.0)
    states["vacuum.x50"] = S("docked", {})
    assert rec._sample("camera.map", 200.0)
    ended = rec.book.data["m1"]["current"]["ended"]
    assert ended == 200.0
    assert not rec._sample("camera.map", 300.0)  # repeated stop is idempotent
    assert rec.book.data["m1"]["current"]["ended"] == ended
    states["vacuum.x50"] = S("unknown", {})
    assert not rec._sample("camera.map", 400.0)
    del states["vacuum.x50"]
    assert not rec._sample("camera.map", 500.0)
    assert rec.book.data["m1"]["current"]["ended"] == ended

    states["vacuum.x50"] = S("cleaning", {"selected_map": "Первый этаж"})
    states["camera.map"] = S(
        "idle", {"vacuum_position": {"x": 1100, "y": -450}, "map_index": 1}
    )
    assert rec._sample("camera.map", 800.0)
    run = rec.book.data["m1"]["current"]
    assert run["ended"] is None
    assert run["points"] == [[1000.0, -500.0], [1100.0, -450.0]]
    assert "previous" not in rec.book.data["m1"]


def test_any_available_nonmoving_state_uses_the_same_grace_contract():
    for stopped in ("paused", "idle", "error", "washing", "docked"):
        rec, _hass, states = _rec()
        assert rec._sample("camera.map", 1.0)
        states["vacuum.x50"] = S(stopped, {})
        assert rec._sample("camera.map", 2.0), stopped
        states["vacuum.x50"] = S("cleaning", {})
        states["camera.map"] = S(
            "idle", {"vacuum_position": {"x": 1200, "y": -400}, "map_index": 1}
        )
        assert rec._sample("camera.map", 3.0), stopped
        assert rec.book.data["m1"]["current"]["points"] == [
            [1000.0, -500.0], [1200.0, -400.0],
        ]
        assert "previous" not in rec.book.data["m1"]


# ---------------- v1.54.0 audit regressions ----------------

def _run_isolated(coro):
    """Run a coroutine on a private loop WITHOUT touching the ambient one.

    asyncio.run() clears the thread's current-loop slot when it finishes; in
    the CI HA harness (pytest-asyncio keeps a session event loop) that
    poisoned the setup of every test that followed — 'There is no current
    event loop in thread MainThread' across whole files.
    """
    import asyncio
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()




def test_map_index_zero_matches_frontend_contract():
    # HP-1540-02: `map_index: 0` is a VALID first map. The old or-chain
    # dropped it and fell through to selected_map — the server stored the
    # run under a key the renderer never looked up.
    rec, _hass, states = _rec()
    states["camera.map"] = S("idle", {"vacuum_position": {"x": 10, "y": 20}, "map_index": 0})
    assert rec._sample("camera.map", 1.0)
    assert rec.book.data["m1"]["current"]["map_id"] == "0"


def test_vacuum_selected_map_zero_fallback_recorded_as_zero():
    # HP-1541-01: source names no map, vacuum reports selected_map: 0 — the
    # recorder must store the run under "0", the same id the fixed card-side
    # fallback (vacMapIdWithFallback) resolves. Before the fix the card asked
    # for calibration/trails under "default" and never found this run.
    rec, _hass, states = _rec()
    states["camera.map"] = S("idle", {"vacuum_position": {"x": 10, "y": 20}})
    states["vacuum.x50"] = S("cleaning", {"selected_map": 0})
    assert rec._sample("camera.map", 1.0)
    assert rec.book.data["m1"]["current"]["map_id"] == "0"


def test_map_id_contract_first_not_none_wins():
    # HP-1540-02: the shared contract — first NOT-None value, stringified
    cases = [
        ({"map_index": 0}, {"selected_map": "Floor"}, "0"),
        ({"map_index": "0"}, {}, "0"),
        ({"map_name": ""}, {"selected_map": "Floor"}, ""),
        ({"map_name": "A", "map_index": 0}, {}, "A"),
        ({"current_map": 2}, {}, "2"),
        ({"selected_map": "Src"}, {"selected_map": "Vac"}, "Src"),
        ({}, {"selected_map": "Vac"}, "Vac"),
        # HP-1541-01: the vacuum-entity fallback with a zero-ish id — must
        # match the card's vacMapIdWithFallback (test/vacuum.test.mjs)
        ({}, {"selected_map": 0}, "0"),
        ({}, {"selected_map": "0"}, "0"),
        ({}, {"selected_map": ""}, ""),
        ({}, {}, "default"),
    ]
    for src_attrs, vac_attrs, want in cases:
        assert trails.resolve_map_id(src_attrs, vac_attrs) == want, (src_attrs, vac_attrs)


def test_map_id_shared_fixture_ignores_vacuum_json_nonce():
    fixture = json.loads((ROOT / "test" / "fixtures" / "vacuum-attrs" / "map-id.json").read_text(
        encoding="utf-8"
    ))
    for row in fixture:
        assert trails.resolve_map_id(row["source"], row["vacuum"]) == row["expected"]
    assert fixture[-2]["expected"] == fixture[-1]["expected"]
    book = trails.TrailBook()
    for index, row in enumerate(fixture[-2:]):
        map_id = trails.resolve_map_id(row["source"], row["vacuum"])
        book.on_point("m1", map_id, float(index), 0.0, float(index))
    assert book.data["m1"]["current"]["points"] == [[0.0, 0.0], [1.0, 0.0]]
    assert "previous" not in book.data["m1"]


def test_one_source_two_floor_markers_both_record():
    # HP-1540-03: the multi-floor case — the same robot placed on two floors.
    # Both markers must receive the server-side run, on every map.
    states = {
        "vacuum.x50": S("cleaning", {}),
        "camera.map": S("idle", {"vacuum_position": {"x": 100, "y": 200}, "map_index": 0}),
    }
    hass = Hass(states)
    rec = trails.TrailRecorder(hass, None)
    rec.pairs = {"camera.map": [("m_floor1", "vacuum.x50"), ("m_floor2", "vacuum.x50")]}
    rec._on_state(E("camera.map"))
    # the robot moves to the second map: BOTH books rotate to the new run
    states["camera.map"] = S("idle", {"vacuum_position": {"x": 300, "y": 400}, "map_index": 1})
    rec._on_state(E("camera.map"))
    for marker in ("m_floor1", "m_floor2"):
        book = rec.book.data[marker]
        assert book["previous"]["map_id"] == "0", marker
        assert book["previous"]["points"] == [[100.0, 200.0]], marker
        assert book["current"]["map_id"] == "1", marker
        assert book["current"]["points"] == [[300.0, 400.0]], marker


def test_refresh_builds_pair_lists_and_dedups_subscription():
    # HP-1540-03: two markers over one source/vacuum → one entity set, both pairs
    import asyncio

    tracked = []

    def track(hass, ents, cb):
        tracked.append(list(ents))
        return lambda: None

    old_track = trails.async_track_state_change_event
    trails.async_track_state_change_event = track
    try:
        cfgm = [
            {"id": "m_f1", "binding": "entity:vacuum.x50", "vacuum": {"source": "camera.map"}},
            {"id": "m_f2", "binding": "entity:vacuum.x50", "vacuum": {"source": "camera.map"}},
        ]

        class CS:
            async def async_load(self):
                return {"config": {"markers": cfgm}}

        class RT:
            config_store = CS()

        hass = Hass({
            "vacuum.x50": S("docked", {}),
            "camera.map": S("idle", {}),
        })
        rec = trails.TrailRecorder(hass, RT())
        _run_isolated(rec.async_refresh())
        assert rec.pairs == {"camera.map": [("m_f1", "vacuum.x50"), ("m_f2", "vacuum.x50")]}
        assert tracked == [["camera.map", "vacuum.x50"]]
    finally:
        trails.async_track_state_change_event = old_track


def test_source_health_deduplicates_reason_changes_and_warns_after_recovery(caplog):
    rec, hass, states = _rec()
    key = {("m1", "camera.map")}

    class Entry:
        def __init__(self, disabled_by=None): self.disabled_by = disabled_by

    class Registry:
        row = None
        def async_get(self, _eid): return self.row

    registry = Registry()
    old_get = trails.er.async_get
    trails.er.async_get = lambda _hass: registry
    try:
        caplog.set_level("WARNING", logger=trails.__name__)
        rec._refresh_source_health(key)  # available baseline
        del states["camera.map"]
        rec._refresh_source_health(key)  # missing: warning 1
        registry.row = Entry("user")
        rec._refresh_source_health(key)  # disabled: reason update, no warning
        registry.row = None
        rec._refresh_source_health(key)  # missing again, no warning
        states["camera.map"] = S("unavailable", {})
        rec._refresh_source_health(key)  # proven existence: recovery
        del states["camera.map"]
        rec._refresh_source_health(key)  # missing after recovery: warning 2
        warnings = [record for record in caplog.records if record.levelname == "WARNING"]
        assert len(warnings) == 2
        assert rec._source_health[("m1", "camera.map")] == "missing"
    finally:
        trails.er.async_get = old_get


def test_source_health_lifecycle_clears_removed_or_rebound_marker(caplog):
    rec, _hass, states = _rec()
    del states["camera.map"]
    caplog.set_level("WARNING", logger=trails.__name__)
    old_get = trails.er.async_get
    trails.er.async_get = lambda _hass: type("Registry", (), {"async_get": lambda self, eid: None})()
    try:
        rec._refresh_source_health({("m1", "camera.map")})
        assert rec._source_health
        rec._refresh_source_health(set())
        assert rec._source_health == {}
    finally:
        trails.er.async_get = old_get


def test_source_health_unavailable_and_unsupported_are_proven_recovery(caplog):
    rec, _hass, states = _rec()
    key = {("m1", "camera.map")}
    caplog.set_level("WARNING", logger=trails.__name__)
    old_get = trails.er.async_get
    trails.er.async_get = lambda _hass: type("Registry", (), {"async_get": lambda self, eid: None})()
    try:
        del states["camera.map"]
        rec._refresh_source_health(key)
        assert len([r for r in caplog.records if r.levelname == "WARNING"]) == 1
        states["camera.map"] = S("unavailable", {})
        rec._refresh_source_health(key)
        assert rec._source_health == {}
        del states["camera.map"]
        rec._refresh_source_health(key)
        states["camera.map"] = S("idle", {})  # exists, but no supported attrs
        rec._refresh_source_health(key)
        assert rec._source_health == {}
        assert len([r for r in caplog.records if r.levelname == "WARNING"]) == 2
    finally:
        trails.er.async_get = old_get


def test_source_health_startup_and_source_change_lifecycle(caplog):
    rec, _hass, states = _rec()
    caplog.set_level("WARNING", logger=trails.__name__)
    old_get = trails.er.async_get
    trails.er.async_get = lambda _hass: type("Registry", (), {"async_get": lambda self, eid: None})()
    try:
        del states["camera.map"]
        rec._refresh_source_health({("m1", "camera.map")})
        rec._refresh_source_health({("m1", "camera.other")})
        assert ("m1", "camera.map") not in rec._source_health
        assert rec._source_health[("m1", "camera.other")] == "missing"
        assert len([r for r in caplog.records if r.levelname == "WARNING"]) == 2
    finally:
        trails.er.async_get = old_get


def test_source_health_startup_existing_states_do_not_warn(caplog):
    rec, _hass, states = _rec()
    caplog.set_level("WARNING", logger=trails.__name__)
    states["camera.map"] = S("unavailable", {})
    rec._refresh_source_health({("m1", "camera.map")})
    states["camera.map"] = S("idle", {})
    rec._refresh_source_health({("m1", "camera.map")})
    assert not [r for r in caplog.records if r.levelname == "WARNING"]


def test_source_health_disabled_unsupported_disabled_warns_twice(caplog):
    rec, _hass, states = _rec()
    key = {("m1", "camera.map")}

    class Entry:
        def __init__(self, disabled_by=None): self.disabled_by = disabled_by

    class Registry:
        row = Entry()
        def async_get(self, _eid): return self.row

    registry = Registry()
    old_get = trails.er.async_get
    trails.er.async_get = lambda _hass: registry
    try:
        caplog.set_level("WARNING", logger=trails.__name__)
        states["camera.map"] = S("idle", {})  # unsupported, but existing
        rec._refresh_source_health(key)
        registry.row = Entry("user")
        rec._refresh_source_health(key)
        registry.row = Entry()
        rec._refresh_source_health(key)  # proven unsupported recovery
        registry.row = Entry("user")
        rec._refresh_source_health(key)
        assert len([r for r in caplog.records if r.levelname == "WARNING"]) == 2
        assert rec._source_health[("m1", "camera.map")] == "disabled"
    finally:
        trails.er.async_get = old_get


def test_source_health_unverified_is_neutral(caplog):
    rec, _hass, states = _rec()
    key = {("m1", "camera.map")}

    class Registry:
        def async_get(self, _eid): return None

    registry = Registry()
    old_get = trails.er.async_get
    try:
        caplog.set_level("WARNING", logger=trails.__name__)
        del states["camera.map"]
        trails.er.async_get = lambda _hass: registry
        rec._refresh_source_health(key)  # missing: warning 1
        trails.er.async_get = lambda _hass: None
        rec._refresh_source_health(key)  # unverified: neutral
        trails.er.async_get = lambda _hass: registry
        rec._refresh_source_health(key)  # same incident, no warning
        assert len([r for r in caplog.records if r.levelname == "WARNING"]) == 1
        assert rec._source_health[("m1", "camera.map")] == "missing"
    finally:
        trails.er.async_get = old_get


def test_refresh_never_tracks_a_removed_marker_even_with_stale_vacuum_fields():
    class CS:
        async def async_load(self):
            return {"config": {"markers": [{
                "id": "m1",
                "binding": "entity:vacuum.x50",
                "removed": True,
                "vacuum": {"source": "camera.map"},
            }]}}

    class RT:
        config_store = CS()

    hass = Hass({
        "vacuum.x50": S("cleaning", {}),
        "camera.map": S("idle", {"vacuum_position": {"x": 1, "y": 2}}),
    })
    rec = trails.TrailRecorder(hass, RT())
    _run_isolated(rec.async_refresh())
    assert rec.pairs == {}
    assert rec.book.data == {}


def test_overlapping_refreshes_leave_one_subscription_teardown_zero():
    # HP-1540-05: two config/set refreshes racing across the awaited load used
    # to BOTH subscribe; teardown removed only the last handle and the other
    # callback leaked until HA restart.
    import asyncio

    active = []
    seq = {"n": 0}

    def track(hass, ents, cb):
        seq["n"] += 1
        hid = seq["n"]
        active.append(hid)
        return lambda: active.remove(hid)

    old_track = trails.async_track_state_change_event
    trails.async_track_state_change_event = track
    try:
        gate = asyncio.Event()

        class CS:
            async def async_load(self):
                await gate.wait()
                return {"config": {"markers": [
                    {"id": "m1", "binding": "entity:vacuum.x50", "vacuum": {"source": "camera.map"}},
                ]}}

        class RT:
            config_store = CS()

        hass = Hass({
            "vacuum.x50": S("docked", {}),
            "camera.map": S("idle", {}),
        })

        async def scenario():
            rec = trails.TrailRecorder(hass, RT())
            t1 = asyncio.ensure_future(rec.async_refresh())
            t2 = asyncio.ensure_future(rec.async_refresh())
            for _ in range(3):  # both tasks are launched; one parks on the gate
                await asyncio.sleep(0)
            gate.set()
            await t1
            await t2
            assert len(active) == 1, f"exactly one live subscription, got {active}"
            # teardown during an in-flight refresh must also end with zero
            gate.clear()
            t3 = asyncio.ensure_future(rec.async_refresh())
            for _ in range(3):
                await asyncio.sleep(0)
            rec.teardown()
            gate.set()
            await t3
            assert active == [], f"teardown must leave zero subscriptions, got {active}"

        _run_isolated(scenario())
    finally:
        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())
    _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_async_teardown_is_idempotent_and_never_writes_without_pending_state():
    rec, _hass, _states = _rec()
    saved = []

    class TrailStore:
        async def async_save(self, data):
            saved.append(json.loads(json.dumps(data)))

    rec.store = TrailStore()
    _run_isolated(rec.async_teardown())
    _run_isolated(rec.async_teardown())

    assert saved == []
    assert rec._closed is True


def test_refresh_watches_every_route_source_not_only_the_root(monkeypatch):
    """#162: карты одного робота могут идти через разные камеры."""
    import asyncio

    tracked = []
    old_track = trails.async_track_state_change_event
    trails.async_track_state_change_event = lambda hass, ents, cb: (
        tracked.append(sorted(ents)) or (lambda: None))
    try:
        markers = [{
            "id": "m1", "binding": "entity:vacuum.x50", "space": "floor1",
            "vacuum": {
                "source": "camera.floor1",
                "map_routes": [
                    {"id": "vr1", "source": "camera.floor1", "map_id": "a", "space": "floor1"},
                    {"id": "vr2", "source": "camera.floor2", "map_id": "b", "space": "floor2"},
                ],
            },
        }]

        class CS:
            async def async_load(self):
                return {"config": {"markers": markers}}

        class RT:
            config_store = CS()

        hass = Hass({
            "vacuum.x50": S("docked", {}),
            "camera.floor1": S("idle", {}),
            "camera.floor2": S("idle", {}),
        })
        rec = trails.TrailRecorder(hass, RT())
        _run_isolated(rec.async_refresh())
        assert sorted(rec.pairs) == ["camera.floor1", "camera.floor2"]
        assert rec.pairs["camera.floor2"] == [("m1", "vacuum.x50")]
        assert tracked == [["camera.floor1", "camera.floor2", "vacuum.x50"]]
        assert [r["id"] for r in rec.routes_by_marker["m1"]] == ["vr1", "vr2"]
    finally:
        trails.async_track_state_change_event = old_track


def test_sample_files_the_point_under_its_route():
    states = {
        "vacuum.x50": S("cleaning", {}),
        "camera.floor2": S("idle", {"vacuum_position": {"x": 5, "y": 6}, "map_name": "b"}),
    }
    rec = trails.TrailRecorder(Hass(states), None)
    rec.pairs = {"camera.floor2": [("m1", "vacuum.x50")]}
    rec.routes_by_marker = {"m1": [
        {"id": "vr1", "source": "camera.floor1", "map_id": "b", "space": "floor1"},
        {"id": "vr2", "source": "camera.floor2", "map_id": "b", "space": "floor2"},
    ]}
    rec._on_state(E("camera.floor2"))
    run = rec.book.data["m1"]["current"]
    assert run["route_id"] == "vr2", "источник, а не только id карты, выбирает маршрут"
    assert run["source"] == "camera.floor2"
    assert run["points"] == [[5.0, 6.0]]


def test_sample_without_a_matching_route_records_legacy_shaped_run():
    states = {
        "vacuum.x50": S("cleaning", {}),
        "camera.map": S("idle", {"vacuum_position": {"x": 1, "y": 2}, "map_name": "неизвестная"}),
    }
    rec = trails.TrailRecorder(Hass(states), None)
    rec.pairs = {"camera.map": [("m1", "vacuum.x50")]}
    rec.routes_by_marker = {"m1": [
        {"id": "vr1", "source": "camera.map", "map_id": "b", "space": "floor1"},
    ]}
    rec._on_state(E("camera.map"))
    run = rec.book.data["m1"]["current"]
    assert "route_id" not in run, "чужой маршрут прогону не приписывается"
    assert run["map_id"] == "неизвестная"
    assert run["points"] == [[1.0, 2.0]]