fix(vp): apply tray_exist_bits empty-slot cleanup to slicer-facing cache (#1726)

  VP bridges bound to a target printer (Proxy mode, Queue mode with
  specific target) forwarded the printer's raw AMS push_status to the
  slicer untouched. bambu_mqtt.py::_handle_ams_data applies a
  tray_exist_bits-driven cleanup to Bambuddy's internal state
  (promote empty slots to state=9, wipe stale tray_type / tray_color /
  tray_info_idx / tag_uid / tray_uuid / remain) so the AMS card renders
  empty slots as Empty, but the VP bridge cache never ran the same
  cleanup. Net result on real hardware: a printer with 3 loaded
  filaments and several previously-loaded-now-empty slots had Bambuddy's
  AMS card render those slots correctly as Empty, but BambuStudio after
  Sync painted them as phantom loaded filaments with stale color and
  material from before the slot went empty.

  Root cause: two consumers of the same payload, only one wired to the
  cleanup. _handle_ams_data ran it on every push; mqtt_bridge.py::
  _on_printer_raw merged the ams blob via _merge_ams_dict but copied
  tray_exist_bits through as an opaque scalar without acting on it.

  Fix: factored the bit-clear logic out of _handle_ams_data into a
  module-level helper apply_tray_exist_bits(units, tray_exist_bits_str,
  *, power_on_flag, log_label). Internal path replaced with a single
  call. Bridge calls it after _merge_ams_dict on the merged ams dict,
  before the merged state is stored as the 1 Hz cached-as-base source.

  Shared shutdown guard kept on both sides: all-zero bits +
  power_on_flag=False is the printer-off pattern (#765, would
  propagate phantom empties on every reconnect); nonzero bits +
  power-off is valid idle-printer state (#1365, X1C between prints)
  and still applies. AMS-HT units (id >= 128) skipped on both sides.

  Tests: new TestApplyTrayExistBitsHelper (10 cases) pins the helper
  contract directly. 3 new bridge regression tests reproduce the
  #1726 wire shape, the shutdown guard, and the AMS-HT skip on the
  cached slicer-facing state. Existing internal-state tests for the
  bit-clear logic (covers state=9 promotion, loaded-slot preserve,
  genuine-removal-with-power-on) continue to pass against the
  refactored path.

  One pre-existing bridge fixture had an inconsistent tray_exist_bits
  ('3' for 2 AMS units each with slot 0 loaded — bit 4 missing). The
  shared cleanup exposed it; corrected to '11' (bits 0 + 4) to match
  real-printer wire shape.

  Reported by @needo37 with full code-level analysis including the
  suggested fix shape and the BAMBUDDY_VP_DUMP_WIRE diagnostic to
  verify on a live system.

backend/app/services/bambu_mqtt.py

@@ -31,6 +31,108 @@ logger = logging.getLogger(__name__)
 _AMS_MODULE_PREFIXES = ("ams/", "n3f/", "n3s/")
 
 
+def apply_tray_exist_bits(
+    units: list,
+    tray_exist_bits_str: str | int | None,
+    *,
+    power_on_flag: bool = True,
+    log_label: str | None = None,
+) -> int:
+    """Wipe stale per-tray filament fields on slots whose `tray_exist_bits` bit is 0.
+
+    `tray_exist_bits` is firmware's canonical "which slots have a spool" bitmask
+    (BambuStudio uses it too). For every slot whose bit is 0, promote the tray
+    `state` to 9 (firmware's "no spool" code) and clear `tray_type` / `tray_color`
+    / `tray_info_idx` / `tag_uid` / `tray_uuid` / `remain` etc so downstream
+    readers (Bambuddy's AMS card, the VP slicer-facing cache, inventory short-
+    circuits keyed on `state in {9, 10}`) all see one canonical empty-slot signal
+    instead of guessing from payload shape (#1322, #147).
+
+    Two callers share this helper to keep their views consistent:
+
+    1. ``_handle_ams_data`` for Bambuddy's internal AMS state (printer card).
+    2. ``virtual_printer.mqtt_bridge._on_printer_raw`` for the cached slicer-
+       facing push_status (#1726 — without this the VP would forward stale
+       per-tray fields for empty slots, and BambuStudio's Sync would render
+       phantom loaded slots).
+
+    Skipped only on the printer-shutdown pattern: all-zero bits paired with
+    ``power_on_flag=False`` (#765). Non-zero bits with ``power_on_flag=False``
+    is valid idle-printer state (#1365 — X1C between prints) and MUST be applied
+    so spool removal is detected without requiring a manual reconnect.
+
+    AMS-HT units (``id >= 128``) use a separate addressing scheme and are
+    skipped here.
+
+    `tray_exist_bits_str` is expected as a hex string (firmware sends it that
+    way). Ints are tolerated for defensive symmetry but typically not seen
+    on the wire. ``None`` / empty / unparseable → no-op.
+
+    Mutates ``units`` in place. Returns the number of slots cleared.
+    """
+    if not tray_exist_bits_str:
+        return 0
+    try:
+        if isinstance(tray_exist_bits_str, int):
+            tray_exist_bits = tray_exist_bits_str
+        else:
+            tray_exist_bits = int(tray_exist_bits_str, 16)
+    except (ValueError, TypeError):
+        return 0
+    if tray_exist_bits == 0 and not power_on_flag:
+        return 0
+    if not isinstance(units, list):
+        return 0
+
+    cleared = 0
+    for ams_unit in units:
+        if not isinstance(ams_unit, dict):
+            continue
+        ams_id_raw = ams_unit.get("id")
+        if ams_id_raw is None:
+            continue
+        try:
+            ams_id = int(ams_id_raw) if isinstance(ams_id_raw, str) else ams_id_raw
+        except (ValueError, TypeError):
+            continue
+        if not isinstance(ams_id, int) or ams_id >= 128:
+            # Skip AMS-HT (id >= 128) — separate addressing scheme.
+            continue
+        for tray in ams_unit.get("tray", []):
+            if not isinstance(tray, dict):
+                continue
+            tray_id_raw = tray.get("id")
+            if tray_id_raw is None:
+                continue
+            try:
+                tray_id = int(tray_id_raw) if isinstance(tray_id_raw, str) else tray_id_raw
+            except (ValueError, TypeError):
+                continue
+            if not isinstance(tray_id, int):
+                continue
+            global_bit = ams_id * 4 + tray_id
+            slot_exists = (tray_exist_bits >> global_bit) & 1
+            if slot_exists:
+                continue
+            tray["state"] = 9
+            if tray.get("tray_type"):
+                if log_label:
+                    logger.debug(
+                        f"[{log_label}] Clearing empty slot: AMS {ams_id} slot {tray_id} "
+                        f"(tray_exist_bits bit {global_bit} = 0)"
+                    )
+                tray["tray_type"] = ""
+                tray["tray_sub_brands"] = ""
+                tray["tray_color"] = ""
+                tray["tray_id_name"] = ""
+                tray["tag_uid"] = "0000000000000000"
+                tray["tray_uuid"] = "00000000000000000000000000000000"
+                tray["tray_info_idx"] = ""
+                tray["remain"] = 0
+                cleared += 1
+    return cleared
+
+
 @dataclass
 class MQTTLogEntry:
     """Log entry for MQTT message debugging."""
@@ -1789,73 +1891,17 @@ class BambuMQTTClient:
         # Convert back to list, sorted by ID for consistent ordering
         merged_ams = sorted(existing_by_id.values(), key=lambda x: x.get("id", 0))
 
-        # Check tray_exist_bits to clear empty slots (Issue #147)
-        # New AMS models don't send empty tray data - they just update tray_exist_bits
-        # Each bit in tray_exist_bits represents a slot: bit=0 means empty, bit=1 means has spool
-        # Skip ONLY the printer-shutdown pattern: all-zero bits paired with
-        # power_on_flag=False (#765). On shutdown that combination would wipe all
-        # slot data and cause auto-unlink to remove spool assignments. Non-zero
-        # bits with power_on_flag=False are valid AMS state from an idle printer
-        # (#1365 — X1C reports power_on_flag=False between prints while the AMS
-        # keeps reporting its actual slot inventory); the update MUST be applied
-        # so spool removal is detected without requiring a manual reconnect.
-        tray_exist_bits_str = ams_data.get("tray_exist_bits") if isinstance(ams_data, dict) else None
-        power_on = ams_data.get("power_on_flag", True) if isinstance(ams_data, dict) else True
-        if tray_exist_bits_str:
-            try:
-                tray_exist_bits = int(tray_exist_bits_str, 16)
-            except (ValueError, TypeError) as e:
-                logger.debug("[%s] Could not parse tray_exist_bits: %s", self.serial_number, e)
-                tray_exist_bits = None
-
-            if tray_exist_bits is not None and not (tray_exist_bits == 0 and not power_on):
-                for ams_unit in merged_ams:
-                    ams_id_raw = ams_unit.get("id")
-                    if ams_id_raw is None:
-                        continue
-                    # Convert to int (may be string from JSON)
-                    ams_id = int(ams_id_raw) if isinstance(ams_id_raw, str) else ams_id_raw
-                    if ams_id >= 128:  # Skip HT AMS (id >= 128)
-                        continue
-                    # Bits for this AMS unit: bits (ams_id*4) to (ams_id*4 + 3)
-                    for tray in ams_unit.get("tray", []):
-                        tray_id_raw = tray.get("id")
-                        if tray_id_raw is None:
-                            continue
-                        # Convert to int (may be string from JSON)
-                        tray_id = int(tray_id_raw) if isinstance(tray_id_raw, str) else tray_id_raw
-                        global_bit = ams_id * 4 + tray_id
-                        slot_exists = (tray_exist_bits >> global_bit) & 1
-                        if not slot_exists:
-                            # #1322 follow-up (by @RosdasHH): the bitmask is
-                            # BambuStudio's canonical "no spool" signal, and
-                            # works across every firmware variant (P1S, A1
-                            # Mini, post-restart, post-Reset-Slot, steady-
-                            # state). Promote to state=9 (firmware's
-                            # explicit "no spool" code) so downstream
-                            # readers — printers.py's API serializer,
-                            # inventory.py's `tray_state in {9, 10}`
-                            # short-circuit, the AMS card — see one
-                            # canonical signal instead of guessing from
-                            # payload shape. Int (not "9") to match the
-                            # downstream `==` comparison.
-                            tray["state"] = 9
-                            if tray.get("tray_type"):
-                                # Stale data from before the slot went empty
-                                # — clear it so the AMS view doesn't render a
-                                # colour/material that's no longer there.
-                                logger.debug(
-                                    f"[{self.serial_number}] Clearing empty slot: AMS {ams_id} slot {tray_id} "
-                                    f"(tray_exist_bits bit {global_bit} = 0)"
-                                )
-                                tray["tray_type"] = ""
-                                tray["tray_sub_brands"] = ""
-                                tray["tray_color"] = ""
-                                tray["tray_id_name"] = ""
-                                tray["tag_uid"] = "0000000000000000"
-                                tray["tray_uuid"] = "00000000000000000000000000000000"
-                                tray["tray_info_idx"] = ""
-                                tray["remain"] = 0
+        # Empty-slot cleanup via tray_exist_bits (#147, #1322, #765, #1365).
+        # Shared with the VP bridge cache so the slicer-facing view stays in
+        # sync with Bambuddy's AMS card (#1726). See the helper's docstring
+        # for the full rationale and the printer-shutdown guard.
+        if isinstance(ams_data, dict):
+            apply_tray_exist_bits(
+                merged_ams,
+                ams_data.get("tray_exist_bits"),
+                power_on_flag=ams_data.get("power_on_flag", True),
+                log_label=self.serial_number,
+            )
 
         self.state.raw_data["ams"] = merged_ams
 

backend/app/services/virtual_printer/mqtt_bridge.py

@@ -41,6 +41,7 @@ import logging
 import socket
 from typing import TYPE_CHECKING
 
+from backend.app.services.bambu_mqtt import apply_tray_exist_bits
 from backend.app.services.virtual_printer._debug import append_event, dump_wire
 
 if TYPE_CHECKING:
@@ -651,6 +652,22 @@ class MQTTBridge:
                         merged = dict(prev_value)
                         merged.update(new_value)
                         new_state[key] = merged
+            # Apply empty-slot cleanup on the merged AMS so the slicer-facing
+            # cache mirrors what Bambuddy's AMS card shows internally. Without
+            # this the cached units carry stale per-tray filament fields for
+            # slots whose `tray_exist_bits` bit is 0, and BambuStudio's Sync
+            # paints those empty slots as phantom loaded filaments (#1726).
+            # Runs whether or not a prev cache existed — fresh pushalls also
+            # carry tray_exist_bits and benefit from the cleanup.
+            merged_ams_dict = new_state.get("ams")
+            if isinstance(merged_ams_dict, dict):
+                units = merged_ams_dict.get("ams")
+                apply_tray_exist_bits(
+                    units if isinstance(units, list) else [],
+                    merged_ams_dict.get("tray_exist_bits"),
+                    power_on_flag=merged_ams_dict.get("power_on_flag", True),
+                    log_label=self.vp_name,
+                )
             self._latest_print_state = new_state
             dump_wire(self.vp_name, "in", new_state)
             return

backend/tests/unit/services/test_bambu_mqtt.py

@@ -1197,6 +1197,140 @@ class TestAMSTrayStateClearning:
         assert tray0["remain"] == 75
 
 
+class TestApplyTrayExistBitsHelper:
+    """Direct contract pinning for the shared ``apply_tray_exist_bits`` helper.
+
+    The same logic is exercised end-to-end via ``_handle_ams_data`` in the
+    internal-state suite and via ``_on_printer_raw`` in the bridge suite,
+    but those go through the merge / cache layers — the helper itself
+    deserves direct coverage so future refactors don't silently change
+    the contract both callers depend on (#1726).
+    """
+
+    def test_returns_zero_on_missing_bits(self):
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]
+        assert apply_tray_exist_bits(units, None) == 0
+        assert apply_tray_exist_bits(units, "") == 0
+        # Untouched.
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+
+    def test_returns_zero_on_unparseable_bits(self):
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]
+        assert apply_tray_exist_bits(units, "garbage") == 0
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+
+    def test_shutdown_guard_zero_bits_with_power_off_skips(self):
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF"}]}]
+        cleared = apply_tray_exist_bits(units, "0", power_on_flag=False)
+        assert cleared == 0
+        # Slot preserved — wiping here would propagate phantom empties on
+        # every printer-off push.
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+
+    def test_zero_bits_with_power_on_still_clears(self):
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF"}]}]
+        cleared = apply_tray_exist_bits(units, "0", power_on_flag=True)
+        # Slot is genuinely empty per the printer's report.
+        assert cleared == 1
+        assert units[0]["tray"][0]["state"] == 9
+        assert units[0]["tray"][0]["tray_type"] == ""
+
+    def test_nonzero_bits_with_power_off_still_clears_removed_slot(self):
+        """#1365: X1C reports power_on_flag=False between prints while the
+        AMS keeps reporting its actual slot inventory. The guard must skip
+        ONLY the all-zero + power-off combination, not nonzero + power-off.
+        """
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [
+            {
+                "id": 0,
+                "tray": [
+                    {"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF"},
+                    {"id": 1, "tray_type": "PETG", "tray_color": "00FF00FF"},
+                ],
+            }
+        ]
+        # 0x1 = slot 0 loaded, slot 1 empty. Power off (steady-state idle).
+        cleared = apply_tray_exist_bits(units, "1", power_on_flag=False)
+        assert cleared == 1
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+        assert units[0]["tray"][1]["tray_type"] == ""
+
+    def test_promotes_state_to_int_nine(self):
+        """Downstream `tray_state in {9, 10}` uses `==` — int 9, not "9"."""
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "state": "11"}]}]
+        apply_tray_exist_bits(units, "0", power_on_flag=True)
+        assert units[0]["tray"][0]["state"] == 9
+        assert isinstance(units[0]["tray"][0]["state"], int)
+
+    def test_ams_ht_unit_skipped(self):
+        """AMS-HT (id >= 128) uses a different addressing scheme."""
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 128, "tray": [{"id": 0, "tray_type": "PLA"}]}]
+        cleared = apply_tray_exist_bits(units, "0", power_on_flag=True)
+        assert cleared == 0
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+
+    def test_string_ids_handled(self):
+        """Bridge cache stores ids as strings (JSON wire format)."""
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [
+            {
+                "id": "0",
+                "tray": [
+                    {"id": "0", "tray_type": "PLA"},
+                    {"id": "1", "tray_type": "PETG"},
+                ],
+            }
+        ]
+        # 0x1 = bit 0 set (slot 0), bit 1 clear (slot 1 empty).
+        cleared = apply_tray_exist_bits(units, "1", power_on_flag=True)
+        assert cleared == 1
+        assert units[0]["tray"][0]["tray_type"] == "PLA"
+        assert units[0]["tray"][1]["tray_type"] == ""
+
+    def test_multi_ams_global_bit_math(self):
+        """global_bit = ams_id * 4 + tray_id. Verify AMS 1 slots use
+        bits 4-7 of the mask, not bits 0-3."""
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [
+            {"id": 0, "tray": [{"id": i, "tray_type": "PLA"} for i in range(4)]},
+            {"id": 1, "tray": [{"id": i, "tray_type": "PETG"} for i in range(4)]},
+        ]
+        # 0x0f: all slots of AMS 0 loaded, all slots of AMS 1 empty.
+        cleared = apply_tray_exist_bits(units, "f", power_on_flag=True)
+        assert cleared == 4
+        for i in range(4):
+            assert units[0]["tray"][i]["tray_type"] == "PLA"
+            assert units[1]["tray"][i]["tray_type"] == ""
+
+    def test_state_promoted_even_when_no_stale_data(self):
+        """Slot without `tray_type` still gets state=9 — the bitmask is
+        authoritative, the field wipe just avoids extra log lines.
+        """
+        from backend.app.services.bambu_mqtt import apply_tray_exist_bits
+
+        units = [{"id": 0, "tray": [{"id": 0, "state": "11"}]}]
+        cleared = apply_tray_exist_bits(units, "0", power_on_flag=True)
+        # No tray_type to clear → cleared counter stays 0 but state is set.
+        assert cleared == 0
+        assert units[0]["tray"][0]["state"] == 9
+
+
 class TestNozzleRackData:
     """Tests for nozzle rack data parsing from H2 series device.nozzle.info."""
 

backend/tests/unit/test_vp_mqtt_bridge.py

@@ -670,7 +670,12 @@ class TestPushStatusCache:
                                 {"id": "0", "tray": [{"id": "0", "tray_type": "PLA"}]},
                                 {"id": "1", "tray": [{"id": "0", "tray_type": "PETG"}]},
                             ],
-                            "tray_exist_bits": "3",
+                            # bit 0 (AMS 0 slot 0) + bit 4 (AMS 1 slot 0) = 0x11.
+                            # `_on_printer_raw` now applies the #1726 bitmask
+                            # cleanup to the cached state, so the test fixture
+                            # must declare both loaded slots — same shape the
+                            # real printer sends.
+                            "tray_exist_bits": "11",
                         },
                     }
                 }
@@ -704,6 +709,181 @@ class TestPushStatusCache:
 
         await bridge.stop()
 
+    @pytest.mark.asyncio
+    async def test_tray_exist_bits_clears_empty_slots_in_slicer_cache(self):
+        """#1726 (reported by @needo37): the bridge cache forwards the real
+        printer's raw AMS payload to the slicer. Without the empty-slot
+        cleanup that bambu_mqtt.py applies to Bambuddy's internal state, the
+        cached units carried stale `tray_type` / `tray_color` /
+        `tray_info_idx` for slots whose `tray_exist_bits` bit was 0 — and
+        BambuStudio's Sync rendered those empty slots as phantom loaded
+        filaments. After the fix the bridge runs the same shared
+        ``apply_tray_exist_bits`` helper before storing the cache.
+        """
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        # Pushall: AMS 0 has slots 0/1/2/3; only slots 1, 2, 3 are loaded.
+        # Slot 0 carries stale data (RFID/color/material from a previously
+        # loaded spool). `tray_exist_bits` = 0xe = 0b1110 → bit 0 unset.
+        bridge._on_printer_raw(
+            f"device/{H2D_SERIAL}/report",
+            json.dumps(
+                {
+                    "print": {
+                        "command": "push_status",
+                        "ams": {
+                            "ams": [
+                                {
+                                    "id": "0",
+                                    "tray": [
+                                        {
+                                            "id": "0",
+                                            "tray_type": "PLA",
+                                            "tray_color": "FF0000FF",
+                                            "tray_info_idx": "GFL00",
+                                            "tag_uid": "1234567890abcdef",
+                                            "tray_uuid": "abcdef1234567890abcdef1234567890",
+                                            "remain": 75,
+                                            "state": "11",
+                                        },
+                                        {"id": "1", "tray_type": "PETG", "tray_color": "00FF00FF"},
+                                        {"id": "2", "tray_type": "ABS", "tray_color": "0000FFFF"},
+                                        {"id": "3", "tray_type": "TPU", "tray_color": "FFFF00FF"},
+                                    ],
+                                }
+                            ],
+                            "tray_exist_bits": "e",
+                        },
+                    }
+                }
+            ).encode(),
+        )
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        slot0 = cached["ams"]["ams"][0]["tray"][0]
+        # Empty slot: stale per-tray fields wiped, state promoted to 9.
+        assert slot0["state"] == 9, "empty slot must be promoted to state=9"
+        assert slot0["tray_type"] == ""
+        assert slot0["tray_color"] == ""
+        assert slot0["tray_info_idx"] == ""
+        assert slot0["tag_uid"] == "0000000000000000"
+        assert slot0["tray_uuid"] == "00000000000000000000000000000000"
+        assert slot0["remain"] == 0
+        # Loaded slots preserved.
+        assert cached["ams"]["ams"][0]["tray"][1]["tray_type"] == "PETG"
+        assert cached["ams"]["ams"][0]["tray"][2]["tray_type"] == "ABS"
+        assert cached["ams"]["ams"][0]["tray"][3]["tray_type"] == "TPU"
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_tray_exist_bits_shutdown_guard_preserves_cache(self):
+        """#765 shutdown guard mirrored at the bridge: when the printer
+        powers off it sends all-zero `tray_exist_bits` paired with
+        `power_on_flag=False`. Wiping the cache on that pattern would
+        propagate phantom empties to every slicer reconnect until the
+        printer powers back on and pushes a real state. Skip cleanup
+        on the shutdown-shaped payload."""
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        # 1. Normal pushall — all four slots loaded.
+        bridge._on_printer_raw(
+            f"device/{H2D_SERIAL}/report",
+            json.dumps(
+                {
+                    "print": {
+                        "command": "push_status",
+                        "ams": {
+                            "ams": [
+                                {
+                                    "id": "0",
+                                    "tray": [
+                                        {"id": str(i), "tray_type": "PLA", "tray_color": f"{i:02x}{i:02x}{i:02x}FF"}
+                                        for i in range(4)
+                                    ],
+                                }
+                            ],
+                            "tray_exist_bits": "f",
+                            "power_on_flag": True,
+                        },
+                    }
+                }
+            ).encode(),
+        )
+        await asyncio.sleep(0.01)
+
+        # 2. Shutdown-shaped push: tray_exist_bits=0 + power_on_flag=False.
+        bridge._on_printer_raw(
+            f"device/{H2D_SERIAL}/report",
+            json.dumps(
+                {
+                    "print": {
+                        "command": "push_status",
+                        "ams": {
+                            "tray_exist_bits": "0",
+                            "power_on_flag": False,
+                        },
+                    }
+                }
+            ).encode(),
+        )
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        for i in range(4):
+            assert cached["ams"]["ams"][0]["tray"][i]["tray_type"] == "PLA", f"slot {i} must survive the shutdown push"
+
+        await bridge.stop()
+
+    @pytest.mark.asyncio
+    async def test_tray_exist_bits_skips_ams_ht_units(self):
+        """AMS-HT units (id >= 128) use a separate addressing scheme and
+        must not be touched by the bitmask cleanup — bit math at
+        global_bit = ams_id * 4 + tray_id would overrun normal AMS bits.
+        Pin the skip so future AMS-HT support doesn't accidentally wipe
+        loaded HT slots.
+        """
+        server = _make_server()
+        bridge = _make_bridge(server)
+        await bridge.start()
+
+        bridge._on_printer_raw(
+            f"device/{H2D_SERIAL}/report",
+            json.dumps(
+                {
+                    "print": {
+                        "command": "push_status",
+                        "ams": {
+                            "ams": [
+                                {
+                                    "id": "128",
+                                    "tray": [
+                                        {"id": "0", "tray_type": "PLA", "tray_color": "FF0000FF"},
+                                    ],
+                                }
+                            ],
+                            "tray_exist_bits": "0",
+                            "power_on_flag": True,
+                        },
+                    }
+                }
+            ).encode(),
+        )
+        await asyncio.sleep(0.01)
+
+        cached = bridge.get_latest_print_state()
+        ht_slot = cached["ams"]["ams"][0]["tray"][0]
+        # tray_exist_bits="0" alone would normally wipe — but AMS-HT is
+        # skipped, so the HT slot keeps its loaded data.
+        assert ht_slot["tray_type"] == "PLA"
+
+        await bridge.stop()
+
     @pytest.mark.asyncio
     async def test_partial_ams_tray_update_preserves_other_trays(self):
         """Same shape as the unit-level test but at the tray level. AMS