Trail recorder: zero map ids, one source per N markers, serialized refresh

Audit HP-1540-02: the recorder chose the map id with an or-chain, so a
valid numeric map_index=0 fell through to selected_map (or 'default') and
the server stored the run under a key the renderer never looks up. The
choice is now resolve_map_id() — the explicit backend half of the contract
shared with vacMapIdFromAttrs in src/vacuum.ts: the first value that is
not None wins, zero and empty string included.

HP-1540-03: pairs was a plain source -> (marker, vacuum) dict, so the
second placement of the same robot (the documented two-floor case) evicted
the first and its server history silently stopped. A source now maps to a
list of pairs, every marker gets its own copy of the run, and the state
subscription set is deduplicated.

HP-1540-05: every config/set spawns async_refresh as a detached task; two
of them interleaving across the awaited config load could both subscribe,
overwriting one unsub handle — a callback leak until HA restart. Refresh
is serialized with an asyncio.Lock and teardown flags the recorder closed
first, so a refresh parked on its await can never resubscribe afterwards.

Regressions cover map_index 0/'0'/''/selected_map cross-checks, one
source feeding two floor markers across a map switch, pair-list refresh
with a deduplicated entity set, and two overlapping refreshes leaving
exactly one live subscription (zero after teardown). On the v1.54.0
recorder 11 of these tests fail.

tests_backend/test_trail_recorder.py

@@ -85,7 +85,7 @@ def _rec():
     }
     hass = Hass(states)
     rec = trails.TrailRecorder(hass, None)
-    rec.pairs = {"camera.map": ("m1", "vacuum.x50")}
+    rec.pairs = {"camera.map": [("m1", "vacuum.x50")]}
     return rec, hass, states
 
 
@@ -152,3 +152,151 @@ def test_unavailable_vacuum_is_no_verdict():
     del states["vacuum.x50"]
     assert not rec._sample("camera.map", 3.0)
     assert rec.book.data["m1"]["current"]["ended"] is None
+
+
+# ---------------- v1.54.0 audit regressions ----------------
+
+
+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_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"),
+        ({}, {}, "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_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())
+        asyncio.run(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_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}"
+
+        asyncio.run(scenario())
+    finally:
+        trails.async_track_state_change_event = old_track