SpoolBuddy (PN5180): read the full tag UID through every cascade level (#3346)

backend/app/api/routes/spoolbuddy.py

@@ -42,9 +42,20 @@ from backend.app.schemas.spoolbuddy import (
     WriteTagRequest,
     WriteTagResultRequest,
 )
-from backend.app.services.spool_tag_matcher import get_spool_by_tag
+from backend.app.services.spool_tag_matcher import (
+    begins_with_cascade_level_1,
+    cascade_level_1_uid,
+    cascade_level_1_uid_start,
+    get_spool_by_tag,
+    warn_cascade_level_1_ambiguous,
+)
 from backend.app.services.spoolman import SpoolmanClientError, SpoolmanNotFoundError, SpoolmanUnavailableError
-from backend.app.utils.tag_normalization import is_bambu_tray_uuid, normalize_tag_uid, normalize_tray_uuid
+from backend.app.utils.tag_normalization import (
+    is_bambu_tray_uuid,
+    normalize_hex,
+    normalize_tag_uid,
+    normalize_tray_uuid,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -439,6 +450,130 @@ async def _backfill_spoolman_tray_uuid(client, sm_spool: dict, tag_uid: str, tra
         logger.exception("SpoolBuddy: could not store tray_uuid %s on Spoolman spool %d", tray_uuid, spool_id)
 
 
+def _spoolman_native_tag_uids(sm_spool: dict) -> set[str]:
+    """The UIDs a Spoolman spool holds as native tags (Spoolman 0.27+)."""
+    uids = {normalize_hex(t.get("uid")) for t in sm_spool.get("tags") or [] if isinstance(t, dict)}
+    uids.discard("")
+    return uids
+
+
+def _spoolman_extra_tag_uid(sm_spool: dict) -> str:
+    """The tag UID in a Spoolman spool's extra.tag, which holds it JSON-encoded; empty when none."""
+    extra = sm_spool.get("extra")
+    raw_tag = extra.get("tag") if isinstance(extra, dict) else None
+    return normalize_hex(raw_tag) if isinstance(raw_tag, str) else ""
+
+
+def _spoolman_tag_links(sm_spool: dict) -> set[str]:
+    """Every UID a Spoolman spool is linked by: its native tags and extra.tag."""
+    return _spoolman_native_tag_uids(sm_spool) | ({_spoolman_extra_tag_uid(sm_spool)} - {""})
+
+
+def _spoolman_spool_by_cascade_level_1(spools: list[dict], tag_uid: str, sak: int | None = None) -> dict | None:
+    """Match a tag UID and a Spoolman link stored from the tag's cascade level 1, either way round.
+
+    The Spoolman counterpart of the built-in inventory's fallback in
+    get_spool_by_tag (#3258), with the same rules. A full 7- or 10-byte read
+    finds the active spool linked as 88 + the UID's first three bytes, as a
+    native tag or in extra.tag. A level-1 read from a daemon not updated yet
+    finds the active spool linked by the 7-byte UID it begins. Three bytes can
+    begin more than one tag, so only a single fitting spool is matched.
+    """
+    level1_uid = cascade_level_1_uid(tag_uid)
+    uid_start = cascade_level_1_uid_start(tag_uid, sak)
+    if not level1_uid and not uid_start:
+        return None
+
+    matches: list[dict] = []
+    for sm_spool in spools:
+        if not isinstance(sm_spool, dict) or sm_spool.get("archived"):
+            continue
+        links = _spoolman_tag_links(sm_spool)
+        if level1_uid in links or (
+            uid_start is not None and any(begins_with_cascade_level_1(uid, uid_start) for uid in links)
+        ):
+            matches.append(sm_spool)
+
+    if len(matches) == 1:
+        logger.info(
+            "Matched Spoolman spool %s by cascade level 1 of tag %s: links %s",
+            matches[0].get("id"),
+            normalize_hex(tag_uid),
+            sorted(_spoolman_tag_links(matches[0])),
+        )
+        return matches[0]
+    if len(matches) > 1:
+        warn_cascade_level_1_ambiguous(normalize_hex(tag_uid), [sm_spool.get("id") for sm_spool in matches])
+    return None
+
+
+async def _upgrade_spoolman_cascade_level_1_link(client, sm_spool: dict, full_uid: str, level1_uid: str) -> None:
+    """Move a Spoolman spool's tag link stored from cascade level 1 over to the full UID (#3258).
+
+    The scan has linked the full UID as a native tag already (add_native_tags).
+    An extra.tag holding the 88 form is rewritten to the full UID, and the 88
+    native tag is unlinked once the spool is linked by the full UID, natively
+    or in extra.tag, so the spool is never left without a link.
+
+    Best-effort: a write Spoolman refuses or fails is logged and the scan keeps
+    its match. Whatever is left of the 88 form stays on the spool, and the next
+    full scan of the tag comes here again and finishes the move.
+    """
+    spool_id = sm_spool.get("id")
+    if not isinstance(spool_id, int):
+        return
+    changed = False
+
+    if _spoolman_extra_tag_uid(sm_spool) == level1_uid:
+        try:
+            await client.merge_spool_extra(spool_id, {"tag": json.dumps(full_uid)})
+            changed = True
+            logger.info(
+                "SpoolBuddy: extra.tag %s of Spoolman spool %d rewritten to the full UID %s",
+                level1_uid,
+                spool_id,
+                full_uid,
+            )
+        except Exception as exc:
+            logger.warning(
+                "SpoolBuddy: could not rewrite extra.tag %s of Spoolman spool %d to %s, trying again on the next scan: %s",
+                level1_uid,
+                spool_id,
+                full_uid,
+                exc,
+            )
+
+    if level1_uid in _spoolman_native_tag_uids(sm_spool):
+        try:
+            current = await client.get_spool(spool_id)
+            if full_uid in _spoolman_tag_links(current):
+                await client.unlink_native_tag(spool_id, level1_uid)
+                changed = True
+                logger.info(
+                    "SpoolBuddy: native tag %s unlinked from Spoolman spool %d, now linked by the full UID %s",
+                    level1_uid,
+                    spool_id,
+                    full_uid,
+                )
+            else:
+                logger.info(
+                    "SpoolBuddy: Spoolman spool %d keeps native tag %s until the full UID %s is linked",
+                    spool_id,
+                    level1_uid,
+                    full_uid,
+                )
+        except Exception as exc:
+            logger.warning(
+                "SpoolBuddy: could not unlink native tag %s from Spoolman spool %d, trying again on the next scan: %s",
+                level1_uid,
+                spool_id,
+                exc,
+            )
+
+    if changed:
+        await ws_manager.broadcast({"type": "inventory_changed"})
+
+
 @router.post("/nfc/tag-scanned")
 async def nfc_tag_scanned(
     req: TagScannedRequest,
@@ -509,13 +644,24 @@ async def nfc_tag_scanned(
                         cached_spools = await client.get_spools()
                     sm_spool = await client.find_spool_by_tag(uid, cached_spools=cached_spools)
 
+            # Last, a link stored from the tag's cascade level 1 (#3258), as
+            # get_spool_by_tag falls back to in the built-in inventory.
+            level1_read = cascade_level_1_uid_start(req.tag_uid, req.sak) is not None
+            level1_read_match = False
+            if sm_spool is None and (level1_read or cascade_level_1_uid(req.tag_uid)):
+                if cached_spools is None:
+                    cached_spools = await client.get_spools()
+                sm_spool = _spoolman_spool_by_cascade_level_1(cached_spools, req.tag_uid, req.sak)
+                level1_read_match = level1_read and sm_spool is not None
+
             if sm_spool is not None:
                 # What this scan read becomes a native tag of the spool: a spool found
                 # through extra.tag moves over on its first scan, and a Bambu spool
-                # collects the chip UID of each side as it is read.
-                await client.add_native_tags(
-                    sm_spool, [tray_uuid, req.tag_uid], "bambu" if tray_uuid else None, retry_refused=True
-                )
+                # collects the chip UID of each side as it is read. A level-1 read
+                # that matched through the fallback is not one: 88 + three bytes can
+                # begin more than one tag, and the spool is linked by the full UID.
+                scanned_uids = [tray_uuid] if level1_read_match else [tray_uuid, req.tag_uid]
+                await client.add_native_tags(sm_spool, scanned_uids, "bambu" if tray_uuid else None, retry_refused=True)
                 mapped = _map_spoolman_spool(sm_spool)
                 await ws_manager.broadcast(
                     {
@@ -546,6 +692,13 @@ async def nfc_tag_scanned(
                     }
                 )
                 logger.info("SpoolBuddy tag matched (Spoolman): %s -> spool %d", req.tag_uid, mapped["id"])
+                # A full read of a spool still linked by the 88 form, matched by the
+                # fallback or exactly after an earlier move stopped half-way.
+                level1_uid = cascade_level_1_uid(req.tag_uid)
+                if level1_uid and level1_uid in _spoolman_tag_links(sm_spool):
+                    await _upgrade_spoolman_cascade_level_1_link(
+                        client, sm_spool, normalize_hex(req.tag_uid), level1_uid
+                    )
                 if tray_uuid:
                     await _backfill_spoolman_tray_uuid(client, sm_spool, req.tag_uid, tray_uuid)
                 return {"status": "ok", "matched": True, "spool_id": mapped["id"]}
@@ -590,7 +743,7 @@ async def nfc_tag_scanned(
             return {"status": "ok", "matched": False, "spool_id": None}
     else:
         # Local mode — exclusive lookup, no Spoolman fallback.
-        spool = await get_spool_by_tag(db, req.tag_uid, tray_uuid or "")
+        spool = await get_spool_by_tag(db, req.tag_uid, tray_uuid or "", req.sak)
         if spool:
             await ws_manager.broadcast(
                 {

backend/app/services/spool_tag_matcher.py

@@ -14,6 +14,7 @@ from backend.app.services.color_catalog_lookup import resolve_bambu_color
 from backend.app.services.slot_nozzle import resolve_slot_nozzle
 from backend.app.services.spool_filament_preset import printer_safe_filament_id, resolve_spool_preset
 from backend.app.utils.tag_normalization import (
+    normalize_hex as _normalize_hex,
     normalize_tag_uid as _normalize_tag_uid,
     normalize_tray_uuid as _normalize_tray_uuid,
 )
@@ -24,6 +25,13 @@ logger = logging.getLogger(__name__)
 ZERO_TAG_UID = "0000000000000000"
 ZERO_TRAY_UUID = "00000000000000000000000000000000"
 
+# ISO 14443-3 cascade tag. A reader that stops after cascade level 1 reports a
+# 7- or 10-byte UID as this byte plus the UID's first three bytes (#3258).
+CASCADE_TAG = "88"
+# SAK bit that says the UID goes on at the next cascade level. Such a reader
+# reports the SAK of level 1, so its 88 form always comes with this bit set.
+SAK_UID_INCOMPLETE = 0x04
+
 # Spool catalog row describing the reusable plastic spool Bambu Lab ships filament
 # on, and the weight to assume when that row is absent. DEFAULT_SPOOL_CATALOG holds
 # three "Bambu Lab%" rows (High Temp 216, Low Temp 250, White 253); this is the one
@@ -429,10 +437,11 @@ async def tag_belongs_to_archived_spool(db: AsyncSession, tag_uid: str, tray_uui
     return found is not None
 
 
-async def get_spool_by_tag(db: AsyncSession, tag_uid: str, tray_uuid: str) -> Spool | None:
+async def get_spool_by_tag(db: AsyncSession, tag_uid: str, tray_uuid: str, sak: int | None = None) -> Spool | None:
     """Look up an active spool by RFID tag UID or Bambu Lab tray UUID.
 
-    Prefers tray_uuid match over tag_uid (more reliable).
+    Prefers tray_uuid match over tag_uid (more reliable). sak is the SAK the
+    reader reported with the UID, when it reported one.
     """
     tray_uuid_norm = _normalize_tray_uuid(tray_uuid)
     tag_uid_norm = _normalize_tag_uid(tag_uid)
@@ -521,6 +530,108 @@ async def get_spool_by_tag(db: AsyncSession, tag_uid: str, tray_uuid: str) -> Sp
                         )
                         return candidate
 
+        spool = await _spool_by_cascade_level_1(db, tag_uid, sak)
+        if spool:
+            return spool
+
+    return None
+
+
+def cascade_level_1_uid(tag_uid: str | None) -> str | None:
+    """The form a reader that stops after cascade level 1 reports this UID in, or None.
+
+    Only a 7- or 10-byte UID goes on past level 1; such a reader reports it as
+    the cascade tag plus its first three bytes (#3258).
+    """
+    uid = _normalize_hex(tag_uid)
+    return CASCADE_TAG + uid[:6] if len(uid) in (14, 20) else None
+
+
+def cascade_level_1_uid_start(tag_uid: str | None, sak: int | None = None) -> str | None:
+    """The first three bytes of the UID behind a read that stopped after cascade level 1, or None.
+
+    A SAK without the incomplete bit says the UID ended there: a 4-byte UID
+    that begins with 88 (a non-ISO clone card) is not such a read.
+    """
+    if sak is not None and not sak & SAK_UID_INCOMPLETE:
+        return None
+    uid = _normalize_hex(tag_uid)
+    return uid[len(CASCADE_TAG) :] if len(uid) == 8 and uid.startswith(CASCADE_TAG) else None
+
+
+def begins_with_cascade_level_1(stored_uid: str, uid_start: str) -> bool:
+    """Whether a stored tag link is a 7-byte UID that a cascade level 1 read of uid_start begins."""
+    return len(stored_uid) == 14 and stored_uid.startswith(uid_start)
+
+
+def warn_cascade_level_1_ambiguous(tag_uid: str, spool_ids: list) -> None:
+    """Log that more than one spool fits a tag at cascade level 1, so none is matched."""
+    logger.warning(
+        "Tag %s fits the tag links of spools %s at cascade level 1; not guessing between them",
+        tag_uid,
+        spool_ids,
+    )
+
+
+async def _spool_by_cascade_level_1(db: AsyncSession, tag_uid: str, sak: int | None = None) -> Spool | None:
+    """Match a tag UID and a link stored from the tag's cascade level 1 alone, either way round.
+
+    SpoolBuddy daemons before #3258 stopped after cascade level 1 and reported
+    the NTAG 04ECE673D32A81 as 8804ECE6, so spools linked on the scale carry
+    that form. A full read finds such a spool and rewrites its link to the full
+    UID, so the short form dies out. A level-1 read from a daemon not updated
+    yet finds the spool linked by the 7-byte UID it begins, so a link that was
+    rewritten, or stored by another reader, still works there. Three bytes can
+    begin more than one tag, so only a single fitting spool is matched.
+    """
+    loads = (selectinload(Spool.k_profiles), selectinload(Spool.assignments))
+    full_uid = _normalize_hex(tag_uid)
+    level1_uid = cascade_level_1_uid(full_uid)
+    uid_start = cascade_level_1_uid_start(full_uid, sak)
+    matches: list[Spool] = []
+
+    if level1_uid:
+        result = await db.execute(
+            select(Spool)
+            .options(*loads)
+            .where(func.upper(Spool.tag_uid) == level1_uid, Spool.archived_at.is_(None))
+            .limit(100)
+        )
+        matches = list(result.scalars().all())
+        if len(matches) == 1:
+            spool = matches[0]
+            spool.tag_uid = _normalize_tag_uid(full_uid)
+            await db.flush()
+            logger.info(
+                "Matched spool %d by its cascade level 1 tag link %s; link rewritten to the full UID %s",
+                spool.id,
+                level1_uid,
+                spool.tag_uid,
+            )
+            return spool
+    elif uid_start:
+        result = await db.execute(
+            select(Spool)
+            .options(*loads)
+            .where(func.upper(Spool.tag_uid).like(f"{uid_start}%"), Spool.archived_at.is_(None))
+            .limit(100)
+        )
+        matches = [
+            spool
+            for spool in result.scalars().all()
+            if begins_with_cascade_level_1(_normalize_tag_uid(spool.tag_uid), uid_start)
+        ]
+        if len(matches) == 1:
+            logger.info(
+                "Matched spool %d by cascade level 1 of its tag UID: scanned=%s, stored=%s",
+                matches[0].id,
+                full_uid,
+                matches[0].tag_uid,
+            )
+            return matches[0]
+
+    if len(matches) > 1:
+        warn_cascade_level_1_ambiguous(full_uid, [spool.id for spool in matches])
     return None
 
 

backend/tests/integration/test_spoolbuddy.py

@@ -647,12 +647,12 @@ class TestNfcEndpoints:
     # matches one by that exact UID saves the tray UUID read from the same tag.
     TRAY = "9E0B0717BEE94D7887EB1D8DFD1A14F3"
 
-    async def _scan(self, async_client: AsyncClient, tag_uid: str, tray_uuid: str | None):
+    async def _scan(self, async_client: AsyncClient, tag_uid: str, tray_uuid: str | None, sak: int | None = None):
         with patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws:
             mock_ws.broadcast = AsyncMock()
             resp = await async_client.post(
                 f"{API}/nfc/tag-scanned",
-                json={"device_id": "sb-1", "tag_uid": tag_uid, "tray_uuid": tray_uuid},
+                json={"device_id": "sb-1", "tag_uid": tag_uid, "tray_uuid": tray_uuid, "sak": sak},
             )
         assert resp.status_code == 200
         return resp.json()
@@ -710,6 +710,36 @@ class TestNfcEndpoints:
 
         assert await self._tray_uuid_of(db_session, spool.id) == other
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_full_uid_scan_saves_full_uid_over_cascade_level_1_link(
+        self, async_client: AsyncClient, spool_factory, db_session
+    ):
+        """A spool linked from a level-1 read (#3258) gets the full UID on its first full scan."""
+        from sqlalchemy import select
+
+        spool_id = (await spool_factory(tag_uid="8804ECE6")).id
+
+        data = await self._scan(async_client, "04ECE673D32A81", None)
+
+        assert data["spool_id"] == spool_id
+        db_session.expire_all()
+        stored = (await db_session.execute(select(Spool.tag_uid).where(Spool.id == spool_id))).scalar_one()
+        assert stored == "04ECE673D32A81"
+        # A daemon not updated yet still finds it.
+        assert (await self._scan(async_client, "8804ECE6", None))["spool_id"] == spool_id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_complete_uid_beginning_with_88_is_no_cascade_level_1_read(
+        self, async_client: AsyncClient, spool_factory
+    ):
+        """An old daemon's level-1 read comes with SAK 0x04; a 4-byte clone card's 88 UID with 0x08 (#3258)."""
+        spool_id = (await spool_factory(tag_uid="04ECE673D32A81")).id
+
+        assert (await self._scan(async_client, "8804ECE6", None, sak=0x04))["spool_id"] == spool_id
+        assert (await self._scan(async_client, "8804ECE6", None, sak=0x08))["matched"] is False
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_spoolman_exact_tag_match_stores_tray_uuid_as_tag(self, async_client: AsyncClient):

backend/tests/integration/test_spoolbuddy_spoolman_cascade_uid_3258.py

@@ -0,0 +1,362 @@
+"""SpoolBuddy scans in Spoolman mode against tag links stored from cascade level 1 (#3258).
+
+SpoolBuddy daemons before #3258 read an NTAG only to cascade level 1 and reported
+04ECE673D32A81 as 8804ECE6 (cascade tag 88 + its first three bytes), so Spoolman
+spools linked on the scale carry that form as a native tag or in extra.tag. These
+are the Spoolman-mode counterparts of the built-in inventory's tests in
+``test_spool_tag_matcher.py``, with the same rules.
+
+Most tests run the scan route against a mocked SpoolmanClient. The last class runs
+a real client against the fake Spoolman of ``backend/tests/_fixtures/spoolman_tags.py``.
+"""
+
+import json
+import logging
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, create_autospec, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.settings import Settings
+from backend.app.services.spoolman import SpoolmanClient, SpoolmanClientError, SpoolmanUnavailableError
+from backend.tests._fixtures.spoolman_tags import BASE_URL, FakeSpoolman, client_for
+
+SCAN = "/api/v1/spoolbuddy/nfc/tag-scanned"
+
+NTAG_UID = "04ECE673D32A81"
+NTAG_LEVEL1 = "8804ECE6"
+AMBIGUOUS = "fits the tag links of spools {} at cascade level 1; not guessing between them"
+
+
+@pytest.fixture
+async def spoolman_on(db_session: AsyncSession):
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    db_session.add(Settings(key="spoolman_url", value=BASE_URL))
+    await db_session.commit()
+
+
+def _spool(spool_id: int, *, tags=(), extra_tag: str | None = None, archived: bool = False) -> dict:
+    """A Spoolman spool as the API answers it; ``tags`` None for a server without native tags."""
+    spool = {
+        "id": spool_id,
+        "filament": {
+            "material": "PLA",
+            "name": "PLA Basic",
+            "color_hex": "FF0000",
+            "weight": 1000.0,
+            "spool_weight": 196.0,
+            "vendor": {"name": "Bambu Lab"},
+        },
+        "used_weight": 0.0,
+        "archived": archived,
+        "registered": "2026-10-01T00:00:00Z",
+        "extra": {"tag": json.dumps(extra_tag)} if extra_tag is not None else {},
+    }
+    if tags is not None:
+        spool["tags"] = [{"uid": uid} for uid in tags]
+    return spool
+
+
+def _client(spools: list[dict], *, tag_api: bool = True, current: dict | None = None):
+    """A mocked SpoolmanClient whose exact lookups miss and whose listing is ``spools``.
+
+    ``current`` is what get_spool answers after the scan linked its tags.
+    """
+    client = create_autospec(SpoolmanClient, instance=True)
+    client.base_url = BASE_URL
+    client.has_tag_api.return_value = tag_api
+    client.find_spool_by_native_tag.return_value = None
+    client.find_spool_by_tag.return_value = None
+    client.get_spools.return_value = spools
+    client.add_native_tags.return_value = 1 if tag_api else 0
+    client.get_spool.return_value = current
+    client.merge_spool_extra.return_value = {}
+    client.unlink_native_tag.return_value = None
+    return client
+
+
+@contextmanager
+def serving(client):
+    getter = AsyncMock(return_value=client)
+    with (
+        patch("backend.app.services.spoolman.get_spoolman_client", getter),
+        patch("backend.app.services.spoolman.init_spoolman_client", getter),
+        patch("backend.app.api.routes.spoolbuddy.ws_manager") as ws,
+    ):
+        ws.broadcast = AsyncMock()
+        yield ws
+
+
+async def _scan(async_client: AsyncClient, tag_uid: str, sak: int | None = None) -> dict:
+    body = {"device_id": "sb-1", "tag_uid": tag_uid}
+    if sak is not None:
+        body["sak"] = sak
+    resp = await async_client.post(SCAN, json=body)
+    assert resp.status_code == 200
+    return resp.json()
+
+
+def _broadcast_types(ws) -> list[str]:
+    return [call.args[0]["type"] for call in ws.broadcast.call_args_list]
+
+
+def _native_uids_added(client) -> list:
+    return [uid for call in client.add_native_tags.await_args_list for uid in call.args[1] if uid]
+
+
+class TestFullUidScan:
+    """A full 7- or 10-byte read finds the spool linked as 88 + its first three bytes."""
+
+    async def test_the_88_native_tag_is_unlinked_once_the_full_uid_is_linked(self, async_client, spoolman_on):
+        spool = _spool(7, tags=[NTAG_LEVEL1])
+        client = _client([spool], current=_spool(7, tags=[NTAG_LEVEL1, NTAG_UID]))
+
+        with serving(client) as ws:
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result == {"status": "ok", "matched": True, "spool_id": 7}
+        # The route's own add_native_tags call links the full UID.
+        assert _native_uids_added(client) == [NTAG_UID]
+        client.unlink_native_tag.assert_awaited_once_with(7, NTAG_LEVEL1)
+        client.merge_spool_extra.assert_not_awaited()
+        assert _broadcast_types(ws) == ["spoolbuddy_tag_matched", "inventory_changed"]
+
+    async def test_an_88_extra_tag_is_rewritten_to_the_full_uid(self, async_client, spoolman_on):
+        """A Spoolman without native tags: extra.tag is the only link."""
+        client = _client([_spool(7, tags=None, extra_tag=NTAG_LEVEL1)], tag_api=False)
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        client.merge_spool_extra.assert_awaited_once_with(7, {"tag": json.dumps(NTAG_UID)})
+        client.unlink_native_tag.assert_not_awaited()
+
+    async def test_both_88_links_move_to_the_full_uid(self, async_client, spoolman_on):
+        spool = _spool(7, tags=[NTAG_LEVEL1], extra_tag=NTAG_LEVEL1)
+        client = _client([spool], current=_spool(7, tags=[NTAG_LEVEL1, NTAG_UID], extra_tag=NTAG_UID))
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        assert _native_uids_added(client) == [NTAG_UID]
+        client.merge_spool_extra.assert_awaited_once_with(7, {"tag": json.dumps(NTAG_UID)})
+        client.unlink_native_tag.assert_awaited_once_with(7, NTAG_LEVEL1)
+
+    async def test_a_ten_byte_uid_finds_its_88_link(self, async_client, spoolman_on):
+        full = "04112233445566778899"
+        client = _client([_spool(7, tags=None, extra_tag="88041122")], tag_api=False)
+
+        with serving(client):
+            result = await _scan(async_client, full)
+
+        assert result["spool_id"] == 7
+        # Spoolman has no length limit, so the whole UID is stored.
+        client.merge_spool_extra.assert_awaited_once_with(7, {"tag": json.dumps(full)})
+
+    async def test_an_archived_spool_neither_matches_nor_makes_it_ambiguous(self, async_client, spoolman_on):
+        client = _client(
+            [_spool(1, extra_tag=NTAG_LEVEL1, archived=True), _spool(7, extra_tag=NTAG_LEVEL1)],
+            current=_spool(7, tags=[NTAG_UID], extra_tag=NTAG_UID),
+        )
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        client.merge_spool_extra.assert_awaited_once_with(7, {"tag": json.dumps(NTAG_UID)})
+
+    async def test_a_spool_matched_exactly_that_still_carries_the_88_form_finishes_the_move(
+        self, async_client, spoolman_on
+    ):
+        """An earlier move stopped half-way: linked by the full UID, the 88 forms still there."""
+        spool = _spool(7, tags=[NTAG_UID, NTAG_LEVEL1], extra_tag=NTAG_LEVEL1)
+        client = _client([], current=spool)
+        client.find_spool_by_native_tag.return_value = spool
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        client.get_spools.assert_not_awaited()
+        client.merge_spool_extra.assert_awaited_once_with(7, {"tag": json.dumps(NTAG_UID)})
+        client.unlink_native_tag.assert_awaited_once_with(7, NTAG_LEVEL1)
+
+    async def test_a_four_byte_uid_ignores_88_links(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=[NTAG_LEVEL1], extra_tag=NTAG_LEVEL1)])
+
+        with serving(client):
+            result = await _scan(async_client, "04ECE673")
+
+        assert result["matched"] is False
+        client.add_native_tags.assert_not_awaited()
+
+
+class TestLevel1Scan:
+    """A level-1 read from a daemon not updated yet finds the spool linked by the full UID."""
+
+    async def test_a_native_full_uid_is_matched_and_nothing_is_written(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=[NTAG_UID])])
+
+        with serving(client) as ws:
+            result = await _scan(async_client, NTAG_LEVEL1)
+
+        assert result == {"status": "ok", "matched": True, "spool_id": 7}
+        # The 88 form is not linked: three bytes can begin more than one tag.
+        assert _native_uids_added(client) == []
+        client.merge_spool_extra.assert_not_awaited()
+        client.unlink_native_tag.assert_not_awaited()
+        client.get_spool.assert_not_awaited()
+        assert _broadcast_types(ws) == ["spoolbuddy_tag_matched"]
+
+    async def test_a_full_uid_in_extra_tag_is_matched_and_nothing_is_written(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=None, extra_tag=NTAG_UID)], tag_api=False)
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_LEVEL1)
+
+        assert result["spool_id"] == 7
+        client.merge_spool_extra.assert_not_awaited()
+
+    async def test_an_old_daemon_reports_the_sak_of_level_1(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=[NTAG_UID])])
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_LEVEL1, sak=0x04)
+
+        assert result["spool_id"] == 7
+
+    async def test_a_complete_uid_beginning_with_88_is_no_level1_read(self, async_client, spoolman_on):
+        """A SAK without bit 0x04 says the UID ended there: a 4-byte clone card, not a level-1 read."""
+        client = _client([_spool(7, tags=[NTAG_UID])])
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_LEVEL1, sak=0x08)
+
+        assert result["matched"] is False
+
+    async def test_only_a_seven_byte_uid_can_be_what_it_begins(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=["04ECE673D32A8100"], extra_tag="04ECE673")])
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_LEVEL1)
+
+        assert result["matched"] is False
+
+
+class TestTwoCandidates:
+    """Two fitting spools may hold two different tags: no guess, nothing written."""
+
+    async def test_full_uid_scan_with_two_88_links(self, async_client, spoolman_on, caplog):
+        client = _client([_spool(3, tags=[NTAG_LEVEL1]), _spool(7, extra_tag=NTAG_LEVEL1)])
+
+        with serving(client) as ws, caplog.at_level(logging.WARNING):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result == {"status": "ok", "matched": False, "spool_id": None}
+        assert _broadcast_types(ws) == ["spoolbuddy_unknown_tag"]
+        assert f"Tag {NTAG_UID} " + AMBIGUOUS.format([3, 7]) in caplog.text
+        client.add_native_tags.assert_not_awaited()
+        client.merge_spool_extra.assert_not_awaited()
+        client.unlink_native_tag.assert_not_awaited()
+
+    async def test_level1_scan_with_two_full_uids_it_begins(self, async_client, spoolman_on, caplog):
+        client = _client([_spool(3, tags=[NTAG_UID]), _spool(7, extra_tag="04ECE6AABBCCDD")])
+
+        with serving(client) as ws, caplog.at_level(logging.WARNING):
+            result = await _scan(async_client, NTAG_LEVEL1)
+
+        assert result["matched"] is False
+        assert _broadcast_types(ws) == ["spoolbuddy_unknown_tag"]
+        assert f"Tag {NTAG_LEVEL1} " + AMBIGUOUS.format([3, 7]) in caplog.text
+        client.add_native_tags.assert_not_awaited()
+        client.merge_spool_extra.assert_not_awaited()
+
+
+class TestFailedWrites:
+    """A refused or failed write keeps the match; the next scan tries the move again."""
+
+    async def test_a_refused_extra_tag_rewrite_keeps_the_match_and_is_retried(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=None, extra_tag=NTAG_LEVEL1)], tag_api=False)
+        client.merge_spool_extra.side_effect = SpoolmanClientError("Spoolman rejected update", 400)
+
+        with serving(client) as ws:
+            first = await _scan(async_client, NTAG_UID)
+            second = await _scan(async_client, NTAG_UID)
+
+        assert first["spool_id"] == second["spool_id"] == 7
+        assert client.merge_spool_extra.await_count == 2
+        assert _broadcast_types(ws) == ["spoolbuddy_tag_matched", "spoolbuddy_tag_matched"]
+
+    async def test_the_88_native_tag_stays_when_the_full_uid_was_not_linked(self, async_client, spoolman_on):
+        """add_native_tags was refused: unlinking the 88 tag would leave the spool without a link."""
+        client = _client([_spool(7, tags=[NTAG_LEVEL1])], current=_spool(7, tags=[NTAG_LEVEL1]))
+        client.add_native_tags.return_value = 0
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        client.unlink_native_tag.assert_not_awaited()
+
+    async def test_a_failing_unlink_keeps_the_match(self, async_client, spoolman_on):
+        client = _client([_spool(7, tags=[NTAG_LEVEL1])], current=_spool(7, tags=[NTAG_LEVEL1, NTAG_UID]))
+        client.unlink_native_tag.side_effect = SpoolmanUnavailableError("Cannot reach Spoolman")
+
+        with serving(client):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result == {"status": "ok", "matched": True, "spool_id": 7}
+
+
+@contextmanager
+def serving_fake(fake: FakeSpoolman):
+    with serving(client_for(fake)) as ws:
+        yield ws
+
+
+class TestAgainstFakeSpoolman:
+    """The same move with a real SpoolmanClient, against a server that enforces the tag rules."""
+
+    async def test_both_88_links_move_and_the_next_scan_is_an_exact_match(self, async_client, spoolman_on):
+        fake = FakeSpoolman()
+        fake.add_spool(7, extra_tag=NTAG_LEVEL1, tags=[NTAG_LEVEL1])
+
+        with serving_fake(fake):
+            first = await _scan(async_client, NTAG_UID)
+            listings = fake.asked("GET /spool")
+            second = await _scan(async_client, NTAG_UID)
+
+        assert first["spool_id"] == second["spool_id"] == 7
+        assert fake.native(7) == [NTAG_UID]
+        assert fake.spools[7]["extra"]["tag"] == json.dumps(NTAG_UID)
+        assert fake.asked("GET /spool") == listings, "the second scan loaded the inventory again"
+
+    async def test_a_failed_extra_tag_write_is_finished_on_the_next_scan(self, async_client, spoolman_on):
+        fake = FakeSpoolman()
+        fake.add_spool(7, extra_tag=NTAG_LEVEL1, tags=[NTAG_LEVEL1])
+        fake.fail_patch.add(7)
+
+        with serving_fake(fake):
+            first = await _scan(async_client, NTAG_UID)
+            assert fake.spools[7]["extra"]["tag"] == json.dumps(NTAG_LEVEL1)
+            fake.fail_patch.clear()
+            second = await _scan(async_client, NTAG_UID)
+
+        assert first["spool_id"] == second["spool_id"] == 7
+        assert fake.native(7) == [NTAG_UID]
+        assert fake.spools[7]["extra"]["tag"] == json.dumps(NTAG_UID)
+
+    async def test_an_older_server_gets_extra_tag_rewritten_and_no_tag_writes(self, async_client, spoolman_on):
+        fake = FakeSpoolman(tag_api=False)
+        fake.add_spool(7, extra_tag=NTAG_LEVEL1)
+
+        with serving_fake(fake):
+            result = await _scan(async_client, NTAG_UID)
+
+        assert result["spool_id"] == 7
+        assert fake.spools[7]["extra"]["tag"] == json.dumps(NTAG_UID)
+        assert fake.tag_writes() == []

backend/tests/unit/services/test_spool_tag_matcher.py

@@ -439,6 +439,156 @@ async def test_get_spool_by_tag_no_false_positive_different_suffix(db_session):
     assert found is None, "Should not match when suffix differs"
 
 
+# -- get_spool_by_tag: cascade level 1 links (#3258) ------------------------
+#
+# SpoolBuddy daemons before #3258 read only cascade level 1 and reported the
+# NTAG 04ECE673D32A81 as 8804ECE6 (cascade tag 88 + its first three bytes).
+
+NTAG_UID = "04ECE673D32A81"
+NTAG_LEVEL1 = "8804ECE6"
+
+
+async def _add_spool(db_session, tag_uid: str, archived: bool = False) -> int:
+    from datetime import datetime
+
+    spool = Spool(
+        material="PLA",
+        tag_uid=tag_uid,
+        label_weight=1000,
+        core_weight=250,
+        archived_at=datetime.now() if archived else None,
+    )
+    db_session.add(spool)
+    await db_session.commit()
+    return spool.id
+
+
+async def _stored_tag_uid(db_session, spool_id: int) -> str | None:
+    db_session.expire_all()
+    return (await db_session.execute(select(Spool.tag_uid).where(Spool.id == spool_id))).scalar_one()
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_full_uid_rewrites_cascade_level_1_link(db_session):
+    """A full read finds the spool linked by a level-1 read and stores the full UID."""
+    spool_id = await _add_spool(db_session, NTAG_LEVEL1)
+
+    found = await get_spool_by_tag(db_session, NTAG_UID, "")
+    await db_session.commit()
+
+    assert found is not None
+    assert found.id == spool_id
+    assert await _stored_tag_uid(db_session, spool_id) == NTAG_UID
+    # The next read is an exact match.
+    assert (await get_spool_by_tag(db_session, NTAG_UID, "")).id == spool_id
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_full_uid_two_cascade_level_1_links_no_match(db_session):
+    """Two spools linked as 8804ECE6 may hold two different tags: no guess, nothing rewritten."""
+    first_id = await _add_spool(db_session, NTAG_LEVEL1)
+    second_id = await _add_spool(db_session, NTAG_LEVEL1)
+
+    found = await get_spool_by_tag(db_session, NTAG_UID, "")
+    await db_session.commit()
+
+    assert found is None
+    assert await _stored_tag_uid(db_session, first_id) == NTAG_LEVEL1
+    assert await _stored_tag_uid(db_session, second_id) == NTAG_LEVEL1
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_full_uid_exact_match_wins_over_cascade_level_1_link(db_session):
+    exact_id = await _add_spool(db_session, NTAG_UID)
+    linked_id = await _add_spool(db_session, NTAG_LEVEL1)
+
+    found = await get_spool_by_tag(db_session, NTAG_UID, "")
+    await db_session.commit()
+
+    assert found.id == exact_id
+    assert await _stored_tag_uid(db_session, linked_id) == NTAG_LEVEL1
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_cascade_level_1_skips_archived(db_session):
+    """An archived spool with the same link neither matches nor makes the match ambiguous."""
+    archived_id = await _add_spool(db_session, NTAG_LEVEL1, archived=True)
+    active_id = await _add_spool(db_session, NTAG_LEVEL1)
+
+    found = await get_spool_by_tag(db_session, NTAG_UID, "")
+    await db_session.commit()
+
+    assert found.id == active_id
+    assert await _stored_tag_uid(db_session, archived_id) == NTAG_LEVEL1
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_ten_byte_uid_rewrites_cascade_level_1_link(db_session):
+    spool_id = await _add_spool(db_session, "88041122")
+
+    found = await get_spool_by_tag(db_session, "04112233445566778899", "")
+    await db_session.commit()
+
+    assert found.id == spool_id
+    # tag_uid keeps the last 16 hex characters, as every lookup normalizes it.
+    assert await _stored_tag_uid(db_session, spool_id) == "2233445566778899"
+    assert (await get_spool_by_tag(db_session, "04112233445566778899", "")).id == spool_id
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_cascade_level_1_read_finds_full_uid(db_session):
+    """A daemon not updated yet still finds a spool linked by the full UID."""
+    spool_id = await _add_spool(db_session, NTAG_UID)
+
+    found = await get_spool_by_tag(db_session, NTAG_LEVEL1, "")
+    await db_session.commit()
+
+    assert found.id == spool_id
+    assert await _stored_tag_uid(db_session, spool_id) == NTAG_UID
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_cascade_level_1_read_with_its_sak_finds_full_uid(db_session):
+    """An old daemon reports the SAK of level 1, which says the UID goes on."""
+    spool_id = await _add_spool(db_session, NTAG_UID)
+
+    found = await get_spool_by_tag(db_session, NTAG_LEVEL1, "", sak=0x04)
+
+    assert found.id == spool_id
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_complete_uid_beginning_with_88_is_no_level_1_read(db_session):
+    """A 4-byte UID that begins with 88 (a non-ISO clone card) ended there: no fallback."""
+    await _add_spool(db_session, NTAG_UID)
+
+    assert await get_spool_by_tag(db_session, NTAG_LEVEL1, "", sak=0x08) is None
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_cascade_level_1_read_two_full_uids_no_match(db_session):
+    """Two stored 7-byte UIDs beginning with the same three bytes: no guess."""
+    await _add_spool(db_session, NTAG_UID)
+    await _add_spool(db_session, "04ECE6AABBCCDD")
+
+    assert await get_spool_by_tag(db_session, NTAG_LEVEL1, "") is None
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_cascade_level_1_read_ignores_other_uid_lengths(db_session):
+    """Only a stored 7-byte UID can be what a level-1 read begins."""
+    await _add_spool(db_session, "04ECE673D32A8100")
+
+    assert await get_spool_by_tag(db_session, NTAG_LEVEL1, "") is None
+
+
+@pytest.mark.asyncio
+async def test_get_spool_by_tag_four_byte_uid_ignores_cascade_level_1_links(db_session):
+    await _add_spool(db_session, "8804ECE6")
+
+    assert await get_spool_by_tag(db_session, "04ECE673", "") is None
+
+
 # -- auto_assign_spool (SpoolAssignment creation) ---------------------------
 
 

spoolbuddy/daemon/pn5180.py

@@ -65,6 +65,14 @@ BAMBU_CONTEXT = b"RFID-A\x00"  # 7 bytes including null terminator
 # the same on both tags of a spool and equal to the AMS's tray_uuid (#984).
 BAMBU_BLOCKS = [1, 2, 4, 5, 9]
 
+# ISO 14443-3 anticollision: SEL codes of cascade levels 1-3, the cascade tag
+# that opens a level's four bytes when the UID goes on at the next level, and
+# the SAK bit that says it does. A 4-byte UID ends at level 1, a 7-byte UID
+# (NTAG, Ultralight) at level 2, a 10-byte UID at level 3 (#3258).
+CASCADE_LEVELS = (0x93, 0x95, 0x97)
+CASCADE_TAG = 0x88
+SAK_UID_INCOMPLETE = 0x04
+
 
 def hkdf_derive_keys(uid: bytes) -> bytes:
     """Derive 96 bytes of MIFARE key material (16 sectors * 6 bytes each).
@@ -248,7 +256,11 @@ class PN5180:
     # -- ISO 14443A --
 
     def activate_type_a(self):
-        """Full Type A activation: WUPA -> Anticollision -> SELECT. Returns (uid, sak) or None."""
+        """Full Type A activation: WUPA -> Anticollision -> SELECT, for each cascade level.
+
+        Returns (uid, sak) with the complete 4-, 7- or 10-byte UID, cascade tags
+        dropped, and the SAK of the last SELECT; or None.
+        """
         # Crypto off, CRC off
         self.write_reg_and(0x00, 0xFFFFFFBF)
         self.write_reg_and(0x12, 0xFFFFFFFE)
@@ -285,46 +297,64 @@ class PN5180:
         if atqa[0] == 0xFF or atqa[0] == 0x00:
             return None
 
-        # Anti-collision Level 1
-        self.write_reg(0x03, 0xFFFFFFFF)
-        self.set_transceive_mode()
-        time.sleep(0.002)
+        uid = b""
+        for level, sel in enumerate(CASCADE_LEVELS):
+            if level > 0:
+                # The previous SELECT turned CRC on; anticollision frames carry none
+                self.write_reg_and(0x12, 0xFFFFFFFE)
+                self.write_reg_and(0x19, 0xFFFFFFFE)
 
-        self.send_data([0x93, 0x20])
-        time.sleep(0.010)
+            # Anti-collision
+            self.write_reg(0x03, 0xFFFFFFFF)
+            self.set_transceive_mode()
+            time.sleep(0.002)
 
-        rx_status = self.read_reg(0x13)
-        rx_len = rx_status & 0x1FF
-        if rx_len < 5 or rx_len > 64:
-            return None
+            self.send_data([sel, 0x20])
+            time.sleep(0.010)
 
-        uid_buf = self.read_data(5)
-        uid = uid_buf[:4]
-        bcc = uid[0] ^ uid[1] ^ uid[2] ^ uid[3]
-        if bcc != uid_buf[4]:
-            return None
+            rx_status = self.read_reg(0x13)
+            rx_len = rx_status & 0x1FF
+            if rx_len < 5 or rx_len > 64:
+                return None
 
-        # SELECT
-        self.write_reg(0x03, 0xFFFFFFFF)
-        self.set_transceive_mode()
-        time.sleep(0.002)
+            uid_buf = self.read_data(5)
+            part = uid_buf[:4]
+            bcc = part[0] ^ part[1] ^ part[2] ^ part[3]
+            if bcc != uid_buf[4]:
+                return None
 
-        # Enable CRC for SELECT
-        self.write_reg_or(0x19, 0x01)
-        self.write_reg_or(0x12, 0x01)
+            # SELECT
+            self.write_reg(0x03, 0xFFFFFFFF)
+            self.set_transceive_mode()
+            time.sleep(0.002)
 
-        self.send_data([0x93, 0x70, uid[0], uid[1], uid[2], uid[3], bcc])
-        time.sleep(0.010)
+            # Enable CRC for SELECT
+            self.write_reg_or(0x19, 0x01)
+            self.write_reg_or(0x12, 0x01)
 
-        rx_status = self.read_reg(0x13)
-        rx_len = rx_status & 0x1FF
-        if rx_len < 1:
-            return None
+            self.send_data([sel, 0x70, part[0], part[1], part[2], part[3], bcc])
+            time.sleep(0.010)
 
-        sak_buf = self.read_data(min(rx_len, 3))
-        sak = sak_buf[0]
+            rx_status = self.read_reg(0x13)
+            rx_len = rx_status & 0x1FF
+            if rx_len < 1:
+                return None
+
+            sak_buf = self.read_data(min(rx_len, 3))
+            sak = sak_buf[0]
+
+            if not sak & SAK_UID_INCOMPLETE:
+                return bytes(uid + part), sak
+
+            # UID goes on at the next level: these four bytes are the cascade
+            # tag and the next three bytes of the UID
+            if part[0] != CASCADE_TAG:
+                logger.debug("SAK 0x%02X at cascade level %d without cascade tag: %s", sak, level + 1, part.hex())
+                return None
+            uid += part[1:]
 
-        return bytes(uid), sak
+        logger.debug("UID still incomplete after cascade level 3: %s", uid.hex())
+        return None
 
     # -- MIFARE Classic --
 

spoolbuddy/scripts/read_tag.py

@@ -63,6 +63,14 @@ BAMBU_CONTEXT = b"RFID-A\x00"  # 7 bytes including null terminator
 # Blocks to read for Bambu tag data
 BAMBU_BLOCKS = [1, 2, 4, 5, 9]
 
+# ISO 14443-3 anticollision: SEL codes of cascade levels 1-3, the cascade tag
+# that opens a level's four bytes when the UID goes on at the next level, and
+# the SAK bit that says it does. A 4-byte UID ends at level 1, a 7-byte UID
+# (NTAG, Ultralight) at level 2, a 10-byte UID at level 3 (#3258).
+CASCADE_LEVELS = (0x93, 0x95, 0x97)
+CASCADE_TAG = 0x88
+SAK_UID_INCOMPLETE = 0x04
+
 
 def hkdf_derive_keys(uid: bytes) -> bytes:
     """Derive 96 bytes of MIFARE key material (16 sectors * 6 bytes each).
@@ -234,7 +242,11 @@ class PN5180:
     # -- ISO 14443A --
 
     def activate_type_a(self):
-        """Full Type A activation: WUPA -> Anticollision -> SELECT. Returns (uid, sak) or None."""
+        """Full Type A activation: WUPA -> Anticollision -> SELECT, for each cascade level.
+
+        Returns (uid, sak) with the complete 4-, 7- or 10-byte UID, cascade tags
+        dropped, and the SAK of the last SELECT; or None.
+        """
         # Crypto off, CRC off
         self.write_reg_and(0x00, 0xFFFFFFBF)
         self.write_reg_and(0x12, 0xFFFFFFFE)
@@ -271,46 +283,64 @@ class PN5180:
         if atqa[0] == 0xFF or atqa[0] == 0x00:
             return None
 
-        # Anti-collision Level 1
-        self.write_reg(0x03, 0xFFFFFFFF)
-        self.set_transceive_mode()
-        time.sleep(0.002)
+        uid = b""
+        for level, sel in enumerate(CASCADE_LEVELS):
+            if level > 0:
+                # The previous SELECT turned CRC on; anticollision frames carry none
+                self.write_reg_and(0x12, 0xFFFFFFFE)
+                self.write_reg_and(0x19, 0xFFFFFFFE)
 
-        self.send_data([0x93, 0x20])
-        time.sleep(0.010)
+            # Anti-collision
+            self.write_reg(0x03, 0xFFFFFFFF)
+            self.set_transceive_mode()
+            time.sleep(0.002)
 
-        rx_status = self.read_reg(0x13)
-        rx_len = rx_status & 0x1FF
-        if rx_len < 5 or rx_len > 64:
-            return None
+            self.send_data([sel, 0x20])
+            time.sleep(0.010)
 
-        uid_buf = self.read_data(5)
-        uid = uid_buf[:4]
-        bcc = uid[0] ^ uid[1] ^ uid[2] ^ uid[3]
-        if bcc != uid_buf[4]:
-            return None
+            rx_status = self.read_reg(0x13)
+            rx_len = rx_status & 0x1FF
+            if rx_len < 5 or rx_len > 64:
+                return None
 
-        # SELECT
-        self.write_reg(0x03, 0xFFFFFFFF)
-        self.set_transceive_mode()
-        time.sleep(0.002)
+            uid_buf = self.read_data(5)
+            part = uid_buf[:4]
+            bcc = part[0] ^ part[1] ^ part[2] ^ part[3]
+            if bcc != uid_buf[4]:
+                return None
 
-        # Enable CRC for SELECT
-        self.write_reg_or(0x19, 0x01)
-        self.write_reg_or(0x12, 0x01)
+            # SELECT
+            self.write_reg(0x03, 0xFFFFFFFF)
+            self.set_transceive_mode()
+            time.sleep(0.002)
 
-        self.send_data([0x93, 0x70, uid[0], uid[1], uid[2], uid[3], bcc])
-        time.sleep(0.010)
+            # Enable CRC for SELECT
+            self.write_reg_or(0x19, 0x01)
+            self.write_reg_or(0x12, 0x01)
 
-        rx_status = self.read_reg(0x13)
-        rx_len = rx_status & 0x1FF
-        if rx_len < 1:
-            return None
+            self.send_data([sel, 0x70, part[0], part[1], part[2], part[3], bcc])
+            time.sleep(0.010)
+
+            rx_status = self.read_reg(0x13)
+            rx_len = rx_status & 0x1FF
+            if rx_len < 1:
+                return None
+
+            sak_buf = self.read_data(min(rx_len, 3))
+            sak = sak_buf[0]
 
-        sak_buf = self.read_data(min(rx_len, 3))
-        sak = sak_buf[0]
+            if not sak & SAK_UID_INCOMPLETE:
+                return bytes(uid + part), sak
+
+            # UID goes on at the next level: these four bytes are the cascade
+            # tag and the next three bytes of the UID
+            if part[0] != CASCADE_TAG:
+                print(f"    SAK 0x{sak:02X} at cascade level {level + 1} without cascade tag: {part.hex().upper()}")
+                return None
+            uid += part[1:]
 
-        return bytes(uid), sak
+        print(f"    UID still incomplete after cascade level 3: {uid.hex().upper()}")
+        return None
 
     # -- MIFARE Classic --
 

spoolbuddy/tests/test_pn5180.py

@@ -0,0 +1,239 @@
+"""Tests for daemon.pn5180 — ISO 14443-3 cascade levels in activate_type_a() (#3258).
+
+The PN5180 is replaced by a fake of its register and transceive commands with
+one tag in the field, so the real anticollision and SELECT sequence runs.
+"""
+
+import importlib
+import importlib.util
+import sys
+import types
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+SEL = (0x93, 0x95, 0x97)
+CT = 0x88
+
+UID_4 = bytes.fromhex("1A2B3C4D")  # Bambu tag: MIFARE Classic 1K, single-size UID
+UID_7 = bytes.fromhex("04ECE673D32A81")  # NTAG215 from #3258, double-size UID
+UID_10 = bytes.fromhex("04112233445566778899")  # triple-size UID
+
+
+def _bcc(part: bytes) -> int:
+    return part[0] ^ part[1] ^ part[2] ^ part[3]
+
+
+def _cascade_parts(uid: bytes) -> list[bytes]:
+    """The four UID bytes a tag answers at each cascade level (ISO 14443-3)."""
+    if len(uid) == 4:
+        return [uid]
+    if len(uid) == 7:
+        return [bytes([CT]) + uid[:3], uid[3:]]
+    return [bytes([CT]) + uid[:3], bytes([CT]) + uid[3:6], uid[6:]]
+
+
+class FakeChip:
+    """Register and transceive commands of a PN5180 with one Type A tag in the field.
+
+    The tag answers WUPA/REQA, then anticollision and SELECT of the cascade level
+    it is at, and nothing else. Like a real tag, it ignores an anticollision
+    frame sent with CRC and a SELECT sent without one.
+    """
+
+    def __init__(self, uid: bytes, sak: int, parts: list[bytes] | None = None, bad_bcc_level=None, gone_at_level=None):
+        self.parts = parts if parts is not None else _cascade_parts(uid)
+        self.sak = sak
+        self.bad_bcc_level = bad_bcc_level
+        self.gone_at_level = gone_at_level
+        self.regs = {0x00: 0, 0x12: 0, 0x19: 0}
+        self.frames: list[bytes] = []
+        self.level = None  # cascade level the tag is at; None until woken
+        self.response = b""
+
+    # -- PN5180 commands --
+
+    def write_reg(self, reg, val):
+        self.regs[reg] = val
+
+    def write_reg_or(self, reg, mask):
+        self.regs[reg] = self.regs.get(reg, 0) | mask
+
+    def write_reg_and(self, reg, mask):
+        self.regs[reg] = self.regs.get(reg, 0) & mask
+
+    def read_reg(self, reg):
+        if reg == 0x13:  # RX_STATUS: received length in bits 0-8
+            return len(self.response)
+        return self.regs.get(reg, 0)
+
+    def read_data(self, length):
+        return self.response[:length]
+
+    def send_data(self, data, valid_bits=0x00):
+        frame = bytes(data)
+        self.frames.append(frame)
+        self.response = self._answer(frame)
+
+    # -- tag --
+
+    def _answer(self, frame: bytes) -> bytes:
+        crc = bool(self.regs[0x19] & 0x01) and bool(self.regs[0x12] & 0x01)
+        if frame in (b"\x52", b"\x26"):
+            self.level = 0
+            return b"\x44\x00"  # ATQA
+        if self.level is None or self.level >= len(self.parts) or self.level == self.gone_at_level:
+            return b""
+        if frame[0] != SEL[self.level]:
+            return b""
+        part = self.parts[self.level]
+        bcc = _bcc(part)
+        if frame[1:] == b"\x20":
+            if crc:
+                return b""
+            return part + bytes([bcc ^ 0xFF if self.level == self.bad_bcc_level else bcc])
+        if frame[1:] == b"\x70" + part + bytes([bcc]):
+            if not crc:
+                return b""
+            self.level += 1
+            return bytes([0x04 if self.level < len(self.parts) else self.sak])
+        return b""
+
+
+@pytest.fixture
+def pn5180_mod(monkeypatch):
+    """daemon.pn5180 imported without gpiod/spidev, its sleeps skipped."""
+    monkeypatch.setitem(sys.modules, "gpiod", MagicMock())
+    monkeypatch.setitem(sys.modules, "spidev", MagicMock())
+    monkeypatch.delitem(sys.modules, "daemon.pn5180", raising=False)
+    mod = importlib.import_module("daemon.pn5180")
+    monkeypatch.setattr(mod, "time", types.SimpleNamespace(sleep=lambda _s: None, monotonic=lambda: 0.0))
+    return mod
+
+
+@pytest.fixture
+def read_tag_mod(monkeypatch):
+    """scripts/read_tag.py, the Read Tag diagnostic, imported without gpiod/spidev, its sleeps skipped."""
+    monkeypatch.setitem(sys.modules, "gpiod", MagicMock())
+    monkeypatch.setitem(sys.modules, "spidev", MagicMock())
+    spec = importlib.util.spec_from_file_location(
+        "spoolbuddy_read_tag", Path(__file__).resolve().parents[1] / "scripts" / "read_tag.py"
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    monkeypatch.setattr(mod, "time", types.SimpleNamespace(sleep=lambda _s: None, monotonic=lambda: 0.0))
+    return mod
+
+
+def _reader(mod, chip: FakeChip):
+    """A PN5180 whose register and transceive commands go to ``chip``."""
+    nfc = mod.PN5180.__new__(mod.PN5180)
+    for name in ("write_reg", "write_reg_or", "write_reg_and", "read_reg", "read_data", "send_data"):
+        setattr(nfc, name, getattr(chip, name))
+    for name in ("reset", "load_rf_config", "rf_on", "rf_off"):
+        setattr(nfc, name, lambda *_args: None)
+    return nfc
+
+
+def _sels(chip: FakeChip) -> list[int]:
+    return [frame[0] for frame in chip.frames if len(frame) > 1]
+
+
+class TestActivateTypeA:
+    def test_single_size_uid_stops_at_level_1(self, pn5180_mod):
+        chip = FakeChip(UID_4, sak=0x08)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() == (UID_4, 0x08)
+        assert chip.frames == [
+            b"\x52",
+            b"\x93\x20",
+            b"\x93\x70" + UID_4 + bytes([_bcc(UID_4)]),
+        ]
+
+    def test_double_size_uid_is_read_to_level_2(self, pn5180_mod):
+        """An NTAG reports all seven bytes and its final SAK, not 88 + three bytes."""
+        chip = FakeChip(UID_7, sak=0x00)
+
+        uid, sak = _reader(pn5180_mod, chip).activate_type_a()
+
+        assert uid.hex().upper() == "04ECE673D32A81"
+        assert sak == 0x00
+        level2 = UID_7[3:]
+        assert chip.frames[-2:] == [b"\x95\x20", b"\x95\x70" + level2 + bytes([_bcc(level2)])]
+
+    def test_triple_size_uid_is_read_to_level_3(self, pn5180_mod):
+        chip = FakeChip(UID_10, sak=0x00)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() == (UID_10, 0x00)
+        assert _sels(chip) == [0x93, 0x93, 0x95, 0x95, 0x97, 0x97]
+
+    def test_double_size_mifare_classic_keeps_its_final_sak(self, pn5180_mod):
+        chip = FakeChip(UID_7, sak=0x08)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() == (UID_7, 0x08)
+
+    def test_crc_is_off_for_each_anticollision_and_on_for_each_select(self, pn5180_mod):
+        """The fake tag ignores frames with the wrong CRC setting, so a full read proves it."""
+        chip = FakeChip(UID_10, sak=0x00)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() is not None
+        assert chip.level == 3
+
+    def test_bad_bcc_at_level_2_is_no_tag(self, pn5180_mod):
+        chip = FakeChip(UID_7, sak=0x00, bad_bcc_level=1)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() is None
+        assert b"\x95\x20" in chip.frames
+        assert not any(frame.startswith(b"\x95\x70") for frame in chip.frames)
+
+    def test_no_answer_at_level_2_is_no_tag(self, pn5180_mod):
+        """A tag that leaves the field after level 1 is not reported as 88 + three bytes."""
+        chip = FakeChip(UID_7, sak=0x00, gone_at_level=1)
+
+        assert _reader(pn5180_mod, chip).activate_type_a() is None
+
+    def test_incomplete_sak_without_cascade_tag_is_no_tag(self, pn5180_mod):
+        chip = FakeChip(UID_7, sak=0x00, parts=[UID_7[:4], UID_7[3:]])
+
+        assert _reader(pn5180_mod, chip).activate_type_a() is None
+        assert 0x95 not in _sels(chip)
+
+    def test_uid_still_incomplete_after_level_3_is_no_tag(self, pn5180_mod):
+        part = bytes([CT, 0x01, 0x02, 0x03])
+        chip = FakeChip(UID_10, sak=0x00, parts=[part, part, part, part])
+
+        assert _reader(pn5180_mod, chip).activate_type_a() is None
+        assert _sels(chip) == [0x93, 0x93, 0x95, 0x95, 0x97, 0x97]
+
+
+class TestNfcReaderPoll:
+    def test_ntag_is_reported_with_its_full_uid(self, pn5180_mod):
+        from daemon.nfc_reader import NFCReader
+
+        reader = NFCReader()
+        reader._nfc = _reader(pn5180_mod, FakeChip(UID_7, sak=0x00))
+
+        event, data = reader.poll()
+
+        assert event == "tag_detected"
+        assert data["tag_uid"] == "04ECE673D32A81"
+        assert data["sak"] == 0x00
+        assert data["tag_type"] == "ntag"
+
+
+class TestReadTagDiagnostic:
+    """The Read Tag diagnostic has its own activate_type_a() and shows the UID the daemon reports."""
+
+    @pytest.mark.parametrize(
+        ("uid", "sak"), [(UID_4, 0x08), (UID_7, 0x00), (UID_10, 0x00)], ids=["4-byte", "7-byte", "10-byte"]
+    )
+    def test_the_full_uid_is_read(self, read_tag_mod, uid, sak):
+        chip = FakeChip(uid, sak=sak)
+
+        assert _reader(read_tag_mod, chip).activate_type_a() == (uid, sak)
+
+    def test_a_tag_gone_at_level_2_is_no_tag(self, read_tag_mod):
+        chip = FakeChip(UID_7, sak=0x00, gone_at_level=1)
+
+        assert _reader(read_tag_mod, chip).activate_type_a() is None