diff --git a/migrations/versions/2026_08_13_1000-fe4970567bb3_spool_tags.py b/migrations/versions/2026_08_13_1000-fe4970567bb3_spool_tags.py
new file mode 100644
index 00000000..f67bf486
--- /dev/null
+++ b/migrations/versions/2026_08_13_1000-fe4970567bb3_spool_tags.py
@@ -0,0 +1,50 @@
+"""spool tags.
+
+Revision ID: fe4970567bb3
+Revises: 9c1d5f2a7b31
+Create Date: 2026-08-13 10:00:00.000000
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "fe4970567bb3"
+down_revision = "9c1d5f2a7b31"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Create the spool_tag table linking physical NFC/RFID tags to spools."""
+    # Purely additive: one new table, no existing column touched, so lossless by
+    # construction. CREATE TABLE and its indexes ride the same revision, which is the
+    # pattern the extra-fields tables already established (b8881bdb716c) and which the
+    # 4-DB suite covers; the DDL/DML split cockroachdb needs (see 304a32906234) does not
+    # apply here because there is no data migration.
+    op.create_table(
+        "spool_tag",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("spool_id", sa.Integer(), nullable=False),
+        sa.Column("uid", sa.String(length=64), nullable=False),
+        sa.Column("format", sa.String(length=32), nullable=True),
+        sa.Column("added", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["spool_id"],
+            ["spool.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_spool_tag_id"), "spool_tag", ["id"], unique=False)
+    op.create_index(op.f("ix_spool_tag_spool_id"), "spool_tag", ["spool_id"], unique=False)
+    # The whole point of the table: one tag, one spool, enforced by the database rather
+    # than by whichever client happened to write last.
+    op.create_index(op.f("ix_spool_tag_uid"), "spool_tag", ["uid"], unique=True)
+
+
+def downgrade() -> None:
+    """Perform the downgrade."""
+    op.drop_index(op.f("ix_spool_tag_uid"), table_name="spool_tag")
+    op.drop_index(op.f("ix_spool_tag_spool_id"), table_name="spool_tag")
+    op.drop_index(op.f("ix_spool_tag_id"), table_name="spool_tag")
+    op.drop_table("spool_tag")
diff --git a/spoolman/api/v1/models.py b/spoolman/api/v1/models.py
index 5b038ea4..adf0d6a0 100644
--- a/spoolman/api/v1/models.py
+++ b/spoolman/api/v1/models.py
@@ -87,6 +87,17 @@ class Message(BaseModel):
     message: str = Field()
 
 
+class TagConflictMessage(Message):
+    """A tag UID is already linked to a different spool.
+
+    Subclasses Message so the `message` key is where it is in every other error body, and
+    adds the conflicting spool's ID so a client can offer to move the tag there instead of
+    making the user go and find it.
+    """
+
+    spool_id: int = Field(description="The spool the tag is already linked to.", examples=[42])
+
+
 class SettingResponse(BaseModel):
     value: str = Field(description="Setting value.")
     is_set: bool = Field(description="Whether the setting has been set. If false, 'value' contains the default value.")
@@ -282,6 +293,32 @@ class Filament(BaseModel):
         )
 
 
+class SpoolTag(BaseModel):
+    """A physical NFC/RFID tag linked to a spool."""
+
+    uid: str = Field(
+        description=(
+            "The tag's hardware UID, normalized to uppercase hexadecimal with separators "
+            "stripped. Unique across all spools: one tag identifies exactly one spool."
+        ),
+        examples=["04A2B3C4D5E6F7"],
+    )
+    format: str | None = Field(
+        None,
+        description=(
+            "What kind of tag this is, e.g. openprinttag, ntag, bambu, tigertag. "
+            "Informational, free-form, and not validated against a fixed list."
+        ),
+        examples=["ntag"],
+    )
+    added: SpoolmanDateTime = Field(description="When the tag was linked to the spool. UTC Timezone.")
+
+    @staticmethod
+    def from_db(item: models.SpoolTag) -> "SpoolTag":
+        """Create a new Pydantic spool tag object from a database spool tag object."""
+        return SpoolTag(uid=item.uid, format=item.format, added=item.added)
+
+
 class Spool(BaseModel):
     id: int = Field(description="Unique internal ID of this spool of filament.")
     registered: SpoolmanDateTime = Field(description="When the spool was registered in the database. UTC Timezone.")
@@ -362,6 +399,13 @@ class Spool(BaseModel):
     extra: dict[str, str] = Field(
         description=_extra_fields_description("spool"),
     )
+    tags: list[SpoolTag] = Field(
+        default_factory=list,
+        description=(
+            "NFC/RFID tags linked to this spool. A spool can carry more than one tag, e.g. when a "
+            "vendor tag has been copied onto a blank sticker. Empty if none are linked."
+        ),
+    )
 
     @staticmethod
     def from_db(item: models.Spool) -> "Spool":
@@ -410,6 +454,7 @@ class Spool(BaseModel):
             comment=item.comment,
             archived=item.archived if item.archived is not None else False,
             extra={field.key: field.value for field in item.extra},
+            tags=[SpoolTag.from_db(tag) for tag in item.tags],
         )
 
 
@@ -563,6 +608,69 @@ class SearchResults(BaseModel):
     )
 
 
+class TagScan(BaseModel):
+    """One tag read reported by a reader-side agent.
+
+    Scans are ephemeral: they are broadcast to the scan websockets and never stored. The
+    match is resolved server-side and included, so an agent that ignores websockets
+    entirely can use the scan endpoint as a one-shot lookup.
+    """
+
+    uid: str = Field(
+        description="The scanned tag's UID, normalized to uppercase hexadecimal with separators stripped.",
+        examples=["04A2B3C4D5E6F7"],
+    )
+    reader_id: str = Field(
+        description=(
+            "Which reader reported the scan. Either the id the agent sent, or one derived from its "
+            "network address when it sent none."
+        ),
+        examples=["printer-voron"],
+    )
+    name: str | None = Field(
+        None,
+        description="Human-readable name for the reader, if it sent one.",
+        examples=["Voron spool holder"],
+    )
+    format: str | None = Field(
+        None,
+        description="What kind of tag this is, if the agent could tell.",
+        examples=["ntag"],
+    )
+    payload_b64: str | None = Field(
+        None,
+        description=(
+            "The tag's raw contents, base64-encoded, if the agent read them. Carried through "
+            "untouched: Spoolman does not decode tag contents."
+        ),
+    )
+    matched_spool_id: int | None = Field(
+        None,
+        description="The spool this tag is linked to, or null if the tag is not known to Spoolman.",
+        examples=[42],
+    )
+    spool: Spool | None = Field(
+        None,
+        description="The matched spool, so a client needs no follow-up request. Null if the tag is unknown.",
+    )
+
+
+class TagReader(BaseModel):
+    """A reader that has reported a scan recently.
+
+    The registry is in-memory and is not persisted: a reader reappears the moment it
+    scans again, and the list is empty after a restart until one does.
+    """
+
+    reader_id: str = Field(description="The reader's id.", examples=["printer-voron"])
+    name: str | None = Field(
+        None,
+        description="Human-readable name for the reader, if it has sent one.",
+        examples=["Voron spool holder"],
+    )
+    last_seen: SpoolmanDateTime = Field(description="When this reader last reported a scan. UTC Timezone.")
+
+
 class Info(BaseModel):
     version: str = Field(examples=["0.7.0"])
     debug_mode: bool = Field(examples=[False])
@@ -607,6 +715,9 @@ class EventType(str, Enum):
     ADDED = "added"
     UPDATED = "updated"
     DELETED = "deleted"
+    # Only ever emitted on the dedicated scan websockets (see TagScanEvent), never on the
+    # entity ones, so no existing consumer can receive it.
+    SCANNED = "scanned"
 
 
 class Event(BaseModel):
@@ -644,3 +755,15 @@ class SettingEvent(Event):
 
     payload: SettingKV = Field(description="Updated setting.")
     resource: Literal["setting"] = Field(description="Resource type.")
+
+
+class TagScanEvent(Event):
+    """A tag was scanned by a reader.
+
+    Travels ONLY on the dedicated scan websockets, which are a separate subscription tree
+    from the entity ones. It therefore never reaches the root /api/v1/ websocket, whose
+    subscribers asked to hear about changes to data -- and a scan changes nothing.
+    """
+
+    payload: TagScan = Field(description="The scan.")
+    resource: Literal["tag_scan"] = Field(description="Resource type.")
diff --git a/spoolman/api/v1/router.py b/spoolman/api/v1/router.py
index 9813b3ac..332c3040 100644
--- a/spoolman/api/v1/router.py
+++ b/spoolman/api/v1/router.py
@@ -16,7 +16,7 @@ from spoolman.exceptions import ItemNotFoundError
 from spoolman.externaldb import get_external_db_name
 from spoolman.ws import websocket_manager
 
-from . import export, externaldb, field, filament, models, other, search, setting, spool, vendor
+from . import export, externaldb, field, filament, models, other, search, setting, spool, tag, vendor
 
 logger = logging.getLogger(__name__)
 
@@ -115,3 +115,4 @@ app.include_router(other.router)
 app.include_router(externaldb.router)
 app.include_router(export.router)
 app.include_router(search.router)
+app.include_router(tag.router)
diff --git a/spoolman/api/v1/spool.py b/spoolman/api/v1/spool.py
index d3790e91..f830a35b 100644
--- a/spoolman/api/v1/spool.py
+++ b/spoolman/api/v1/spool.py
@@ -5,7 +5,7 @@ import logging
 from datetime import datetime
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Path, Query, Request, WebSocket, WebSocketDisconnect
+from fastapi import APIRouter, Depends, Path, Query, Request, Response, WebSocket, WebSocketDisconnect
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field, field_validator
@@ -18,14 +18,20 @@ from spoolman.api.v1.models import (
     Spool,
     SpoolEvent,
     SpoolGroup,
+    SpoolTag,
+    TagConflictMessage,
     Vendor,
     extra_fields_request_description,
 )
+
+# Aliased: `tag` is taken by the find endpoint's query parameter, whose name is API surface.
 from spoolman.database import spool
+from spoolman.database import tag as tag_db
 from spoolman.database.database import get_db_session
 from spoolman.database.utils import parse_sort
-from spoolman.exceptions import ItemCreateError, SpoolMeasureError
+from spoolman.exceptions import ItemCreateError, SpoolMeasureError, TagConflictError
 from spoolman.extra_fields import EntityType, get_extra_fields, validate_extra_field_dict
+from spoolman.tags import FORMAT_MAX_LENGTH, KNOWN_FORMATS, UID_MAX_LENGTH
 from spoolman.ws import websocket_manager
 
 logger = logging.getLogger(__name__)
@@ -291,6 +297,20 @@ async def find(
             ),
         ),
     ] = None,
+    tag: Annotated[
+        str | None,
+        Query(
+            title="Tag UID",
+            description=(
+                "Match the spool that an NFC/RFID tag with this UID is linked to. Exact match on the "
+                "normalized UID: separators are ignored and case does not matter, so 04:a2:b3:c4, "
+                "04-A2-B3-C4 and 04a2b3c4 all find the same spool. A tag is linked to at most one "
+                "spool, so this returns either one spool or none. Returns 400 if the UID is not "
+                "hexadecimal."
+            ),
+            examples=["04A2B3C4D5E6F7", "04:a2:b3:c4:d5:e6:f7"],
+        ),
+    ] = None,
     allow_archived: Annotated[
         bool,
         Query(title="Allow Archived", description="Whether to include archived spools in the search results."),
@@ -343,6 +363,7 @@ async def find(
             vendor_id=filament_vendor_ids,
             location=location,
             lot_nr=lot_nr,
+            tag=tag,
             allow_archived=allow_archived,
             extra_field_filters=spool_extra or None,
             filament_extra_field_filters=filament_extra or None,
@@ -769,6 +790,94 @@ async def delete(
     return Message(message="Success!")
 
 
+class SpoolTagParameters(BaseModel):
+    uid: str = Field(
+        min_length=1,
+        max_length=UID_MAX_LENGTH * 2,  # room for separators; the normalized UID is what must fit
+        description=(
+            "The tag's hardware UID, in whatever shape the reader reports it. Separators (:, -, _, "
+            "spaces) are stripped and the result is uppercased before storing, so every spelling of "
+            "one physical tag resolves to the same tag."
+        ),
+        examples=["04:a2:b3:c4:d5:e6:f7", "04A2B3C4D5E6F7"],
+    )
+    format: str | None = Field(
+        None,
+        max_length=FORMAT_MAX_LENGTH,
+        description=(
+            "What kind of tag this is. Informational; not validated against a fixed list, because new "
+            f"tag types appear faster than releases do. Commonly one of: {', '.join(KNOWN_FORMATS)}."
+        ),
+        examples=["ntag"],
+    )
+
+
+@router.post(
+    "/{spool_id}/tag",
+    name="Link a tag to a spool",
+    description=(
+        "Link a physical NFC/RFID tag to this spool, so that the tag's UID identifies it. "
+        "A tag belongs to exactly one spool; linking a UID that another spool already holds "
+        "returns 409 with that spool's id, so a client can offer to move it instead. "
+        "Re-linking a tag to the spool that already holds it succeeds and changes nothing, "
+        "except that a format sent now refines one recorded earlier."
+    ),
+    status_code=201,
+    response_model_exclude_none=True,
+    response_model=SpoolTag,
+    responses={
+        400: {"model": Message},
+        404: {"model": Message},
+        409: {"model": TagConflictMessage},
+    },
+)
+async def link_tag(  # noqa: ANN201
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+    body: SpoolTagParameters,
+):
+    try:
+        db_item = await tag_db.link(db=db, spool_id=spool_id, uid=body.uid, tag_format=body.format)
+    except ValueError as e:
+        return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
+    except TagConflictError as e:
+        return JSONResponse(
+            status_code=409,
+            content=TagConflictMessage(message=str(e), spool_id=e.spool_id).dict(),
+        )
+    return SpoolTag.from_db(db_item)
+
+
+@router.delete(
+    "/{spool_id}/tag/{uid}",
+    name="Unlink a tag from a spool",
+    description=(
+        "Unlink a physical NFC/RFID tag from this spool. The UID is matched the same way it is "
+        "stored: separators are ignored and case does not matter. Deleting a spool unlinks its "
+        "tags on its own, so this is only for taking one tag off a spool that keeps existing."
+    ),
+    status_code=204,
+    responses={400: {"model": Message}, 404: {"model": Message}},
+)
+async def unlink_tag(
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    spool_id: int,
+    uid: Annotated[
+        str,
+        Path(
+            title="Tag UID",
+            description="The tag's UID, in any shape. Normalized before matching.",
+            examples=["04A2B3C4D5E6F7"],
+        ),
+    ],
+) -> Response:
+    try:
+        await tag_db.unlink(db=db, spool_id=spool_id, uid=uid)
+    except ValueError as e:
+        return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
+    return Response(status_code=204)
+
+
 @router.put(
     "/{spool_id}/use",
     name="Use spool filament",
diff --git a/spoolman/api/v1/tag.py b/spoolman/api/v1/tag.py
new file mode 100644
index 00000000..6d9a9615
--- /dev/null
+++ b/spoolman/api/v1/tag.py
@@ -0,0 +1,226 @@
+"""Tag scan relay endpoints.
+
+Spoolman never talks to reader hardware; the reader talks to Spoolman. The reader is almost
+never on the Spoolman host -- Spoolman runs in Docker on a NAS while the tag gets tapped at the
+printer -- so the contract between the two is one plain HTTP POST per tap, with no auth, no
+handshake, no inbound port on the device and no library. Anything that can POST JSON qualifies,
+which is why Node-RED, Home Assistant, a shell script and a microcontroller are all first-class
+here without Spoolman shipping code for any of them.
+
+What the relay adds on top of a lookup is decoupling *where the reader is* from *where the user
+is looking*: the scan is broadcast to browsers that subscribed to that reader, so tapping a tag
+at the printer can drive a page open on a desk across the house.
+"""
+
+import asyncio
+import logging
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, WebSocket, WebSocketDisconnect
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.api.v1.models import EventType, Message, Spool, TagReader, TagScan, TagScanEvent
+from spoolman.database import tag as tag_db
+from spoolman.database.database import get_db_session
+from spoolman.scanrelay import READER_ID_PATTERN, derive_reader_id, scan_relay
+from spoolman.tags import FORMAT_MAX_LENGTH, KNOWN_FORMATS, UID_MAX_LENGTH, normalize_uid
+from spoolman.ws import scan_websocket_manager
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/tag",
+    tags=["tag"],
+)
+
+# ruff: noqa: D103
+
+# Tag contents are carried through untouched and never decoded, so the only thing worth
+# constraining is size: an unbounded payload is an unbounded websocket broadcast to every
+# subscriber, from an unauthenticated endpoint.
+PAYLOAD_MAX_LENGTH = 8192
+
+
+class TagScanParameters(BaseModel):
+    uid: str = Field(
+        min_length=1,
+        max_length=UID_MAX_LENGTH * 2,  # room for separators; the normalized UID is what must fit
+        description=(
+            "The scanned tag's UID, in whatever shape the reader reports it. Any separator style, "
+            "any case: the server normalizes it, which is why an ESPHome device sending 04-a2-b3 "
+            "and a phone sending 04:A2:B3 identify the same tag."
+        ),
+        examples=["04-A2-B3-C4-D5-E6-F7"],
+    )
+    reader_id: str | None = Field(
+        None,
+        pattern=READER_ID_PATTERN,
+        description=(
+            "A stable, operator-chosen id for this reader, e.g. the device hostname. Browsers "
+            "subscribe by it, so it is what binds a screen to a reader. If omitted, one is derived "
+            "from the client's network address, which works but breaks on DHCP churn."
+        ),
+        examples=["printer-voron"],
+    )
+    name: str | None = Field(
+        None,
+        max_length=64,
+        description="Human-readable name for the reader, shown when choosing one.",
+        examples=["Voron spool holder"],
+    )
+    format: str | None = Field(
+        None,
+        max_length=FORMAT_MAX_LENGTH,
+        description=(
+            "What kind of tag this is, if the agent can tell. Informational; not validated against "
+            f"a fixed list. Commonly one of: {', '.join(KNOWN_FORMATS)}."
+        ),
+        examples=["ntag"],
+    )
+    payload_b64: str | None = Field(
+        None,
+        max_length=PAYLOAD_MAX_LENGTH,
+        description=(
+            "The tag's raw contents, base64-encoded, if the agent read them. Accepted and carried "
+            "into the broadcast untouched; Spoolman does not decode tag contents."
+        ),
+    )
+
+
+@router.post(
+    "/scan",
+    name="Report a tag scan",
+    description=(
+        "Report that a reader has scanned a tag. Spoolman resolves the tag to a spool and returns "
+        "the match, then broadcasts the scan on the tag scan websockets so a paired browser can "
+        "react to it.\n\n"
+        "The response is the whole integration for a device that only wants a lookup: "
+        "`matched_spool_id` is always present, null when the tag is not linked to anything. A "
+        "device may ignore the response entirely.\n\n"
+        "Scans are ephemeral -- broadcast only, never stored. Repeated identical scans from the "
+        "same reader within a few seconds are broadcast once, because readers re-detect a tag that "
+        "is sitting still; the response is unaffected, so a de-duplicated scan never looks to the "
+        "device like a failed lookup."
+    ),
+    responses={
+        200: {"model": TagScan},
+        400: {"model": Message},
+        299: {"model": TagScanEvent, "description": "Websocket message"},
+    },
+)
+async def scan(
+    *,
+    request: Request,
+    db: Annotated[AsyncSession, Depends(get_db_session)],
+    body: TagScanParameters,
+) -> JSONResponse:
+    reader_id = body.reader_id or derive_reader_id(request.client.host if request.client else None)
+
+    try:
+        # Normalized here so the response can echo the canonical form back -- an agent should be
+        # able to see the shape its tag is stored under without reimplementing the rules.
+        # database/tag.py normalizes again on the way to the query; doing so is idempotent, and
+        # that layer is what makes the guarantee hold for callers that never come through here.
+        uid = normalize_uid(body.uid)
+    except ValueError as e:
+        return JSONResponse(status_code=400, content=Message(message=str(e)).dict())
+
+    db_spool = await tag_db.find_spool_by_uid(db, uid)
+
+    scan_relay.register(reader_id, body.name)
+
+    result = TagScan(
+        uid=uid,
+        reader_id=reader_id,
+        name=body.name,
+        format=body.format,
+        payload_b64=body.payload_b64,
+        matched_spool_id=db_spool.id if db_spool is not None else None,
+        spool=Spool.from_db(db_spool) if db_spool is not None else None,
+    )
+
+    if scan_relay.should_broadcast(uid, reader_id):
+        await _broadcast(result)
+
+    content = jsonable_encoder(result, exclude_none=True)
+    # exclude_none strips a null match, but "no match" is the answer to the device's question
+    # and it has to be able to see it. Everything else may be omitted.
+    content.setdefault("matched_spool_id", None)
+    return JSONResponse(content=content)
+
+
+async def _broadcast(result: TagScan) -> None:
+    """Broadcast a scan to the reader's pool, and to anyone following every reader."""
+    try:
+        await scan_websocket_manager.send(
+            (result.reader_id,),
+            TagScanEvent(
+                type=EventType.SCANNED,
+                resource="tag_scan",
+                date=datetime.utcnow(),
+                payload=result,
+            ),
+        )
+    except Exception:
+        # Same reasoning as spool_changed: a websocket failure must never fail the request
+        # that triggered it.
+        logger.exception("Failed to send websocket message")
+
+
+@router.get(
+    "/reader",
+    name="List recently seen tag readers",
+    description=(
+        "List the readers that have reported a scan recently, so a client can offer a "
+        "'choose a reader' picker instead of asking someone to type an id.\n\n"
+        "The registry is in-memory and never persisted: it is empty after a restart, and a "
+        "reader reappears the moment it scans again."
+    ),
+    response_model_exclude_none=True,
+)
+async def readers() -> list[TagReader]:
+    return [
+        TagReader(reader_id=reader.reader_id, name=reader.name, last_seen=reader.last_seen)
+        for reader in scan_relay.readers()
+    ]
+
+
+@router.websocket(
+    "/scan",
+    name="Listen to tag scans from any reader",
+)
+async def notify_any(
+    websocket: WebSocket,
+) -> None:
+    await websocket.accept()
+    scan_websocket_manager.connect((), websocket)
+    try:
+        while True:
+            await asyncio.sleep(0.5)
+            if await websocket.receive_text():
+                await websocket.send_json({"status": "healthy"})
+    except WebSocketDisconnect:
+        scan_websocket_manager.disconnect((), websocket)
+
+
+@router.websocket(
+    "/scan/{reader_id}",
+    name="Listen to tag scans from one reader",
+)
+async def notify(
+    websocket: WebSocket,
+    reader_id: str,
+) -> None:
+    await websocket.accept()
+    scan_websocket_manager.connect((reader_id,), websocket)
+    try:
+        while True:
+            await asyncio.sleep(0.5)
+            if await websocket.receive_text():
+                await websocket.send_json({"status": "healthy"})
+    except WebSocketDisconnect:
+        scan_websocket_manager.disconnect((reader_id,), websocket)
diff --git a/spoolman/database/models.py b/spoolman/database/models.py
index bbb60cd5..f645d9d8 100644
--- a/spoolman/database/models.py
+++ b/spoolman/database/models.py
@@ -94,6 +94,14 @@ class Spool(Base):
         cascade="save-update, merge, delete, delete-orphan",
         lazy="selectin",
     )
+    # selectin for the same reason as `extra` above: a spool listing already joins
+    # filament and vendor, and a joined collection here would multiply against the
+    # extra-field rows.
+    tags: Mapped[list["SpoolTag"]] = relationship(
+        back_populates="spool",
+        cascade="save-update, merge, delete, delete-orphan",
+        lazy="selectin",
+    )
 
 
 class Setting(Base):
@@ -129,3 +137,27 @@ class SpoolField(Base):
     spool: Mapped["Spool"] = relationship(back_populates="extra")
     key: Mapped[str] = mapped_column(String(64), primary_key=True, index=True)
     value: Mapped[str] = mapped_column(Text())
+
+
+class SpoolTag(Base):
+    """A physical NFC/RFID tag linked to a spool.
+
+    Many tags per spool is a real case rather than a hypothetical: copying a Prusa NFC-V
+    tag's payload onto an NTAG215 so a PN532 can read it leaves one spool carrying two
+    physical tags with two different UIDs (#776).
+
+    The unique index on `uid` is the point of the table -- one tag, one spool, enforced.
+    It only means that because every write goes through `spoolman.tags.normalize_uid`;
+    see the module docstring there.
+    """
+
+    __tablename__ = "spool_tag"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    spool_id: Mapped[int] = mapped_column(ForeignKey("spool.id"), index=True)
+    spool: Mapped["Spool"] = relationship(back_populates="tags")
+    uid: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    # Free-ish string (openprinttag, ntag, bambu, ...) rather than an enum: informational
+    # in phase 1, and new tag types appear faster than migrations should.
+    format: Mapped[str | None] = mapped_column(String(32))
+    added: Mapped[datetime] = mapped_column()
diff --git a/spoolman/database/spool.py b/spoolman/database/spool.py
index c309480c..885717e9 100644
--- a/spoolman/database/spool.py
+++ b/spoolman/database/spool.py
@@ -34,6 +34,7 @@ from spoolman.database.utils import (
 from spoolman.exceptions import ItemCreateError, ItemNotFoundError, SpoolMeasureError
 from spoolman.extra_field_registry import EntityType, ExtraField, ExtraFieldType, get_extra_fields
 from spoolman.math import weight_from_length
+from spoolman.tags import normalize_uid
 from spoolman.ws import websocket_manager
 
 logger = logging.getLogger(__name__)
@@ -103,6 +104,11 @@ async def create(
         comment=comment,
         archived=archived,
         extra=[models.SpoolField(key=k, value=v) for k, v in (extra or {}).items() if v is not None],
+        # Explicitly empty rather than left unset: a selectin collection that was never
+        # populated is *unloaded* on the persistent object after the commit below, so
+        # Spool.from_db would emit a lazy SELECT from async code and raise MissingGreenlet.
+        # A new spool has no tags, and saying so costs nothing.
+        tags=[],
     )
     db.add(spool)
     await db.commit()
@@ -133,6 +139,7 @@ async def find(  # noqa: C901
     vendor_id: int | Sequence[int] | None = None,
     location: str | None = None,
     lot_nr: str | None = None,
+    tag: str | None = None,
     allow_archived: bool = False,
     extra_field_filters: dict[str, str] | None = None,
     filament_extra_field_filters: dict[str, str] | None = None,
@@ -158,6 +165,7 @@ async def find(  # noqa: C901
         vendor_id=vendor_id,
         location=location,
         lot_nr=lot_nr,
+        tag=tag,
         allow_archived=allow_archived,
     ).options(contains_eager(models.Spool.filament).contains_eager(models.Filament.vendor))
 
@@ -267,10 +275,19 @@ def _apply_spool_filters(
     vendor_id: int | Sequence[int] | None = None,
     location: str | None = None,
     lot_nr: str | None = None,
+    tag: str | None = None,
     allow_archived: bool = False,
 ) -> sqlalchemy.Select:
     """Apply the standard spool joins and where-clauses shared by find and find_groups."""
     stmt = stmt.join(models.Spool.filament, isouter=True).join(models.Filament.vendor, isouter=True)
+    if tag is not None:
+        # An inner join rather than a subquery, so the list query and the count query stay
+        # in sync automatically: they share this builder. `uid` is unique, so at most one
+        # spool_tag row can match and the join cannot multiply result rows -- which is why
+        # the contains_eager chain for filament/vendor is unaffected.
+        stmt = stmt.join(models.SpoolTag, models.SpoolTag.spool_id == models.Spool.id).where(
+            models.SpoolTag.uid == normalize_uid(tag),
+        )
     stmt = add_where_clause_int(stmt, models.Spool.filament_id, filament_id)
     stmt = add_where_clause_int_opt(stmt, models.Filament.vendor_id, vendor_id)
     stmt = add_where_clause_str(stmt, models.Vendor.name, vendor_name)
diff --git a/spoolman/database/tag.py b/spoolman/database/tag.py
new file mode 100644
index 00000000..ae8dc4f2
--- /dev/null
+++ b/spoolman/database/tag.py
@@ -0,0 +1,152 @@
+"""Helper functions for interacting with spool tag database objects.
+
+Kept out of `spool.py`, which is already the largest module in the tree, but following
+its conventions exactly: `AsyncSession` first, exceptions from `spoolman.exceptions`,
+and the websocket event emitted after the commit.
+
+This module is the ONE place a UID gets normalized, on the way in and on every lookup.
+Doing it in a Pydantic validator would read better but would only cover callers that
+arrive over HTTP -- and the unique constraint is worthless if any other path (an
+importer, a migration backfill, a future tag codec) can write a differently-shaped UID.
+"""
+
+import logging
+from datetime import datetime
+
+import sqlalchemy
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from spoolman.api.v1.models import EventType
+from spoolman.database import models, spool
+from spoolman.exceptions import ItemNotFoundError, TagConflictError
+from spoolman.tags import normalize_format, normalize_uid
+
+logger = logging.getLogger(__name__)
+
+
+async def _get_tag_by_uid(db: AsyncSession, uid: str) -> models.SpoolTag | None:
+    """Get the tag with this exact normalized UID, if any. Hits the unique index."""
+    stmt = sqlalchemy.select(models.SpoolTag).where(models.SpoolTag.uid == uid)
+    return (await db.execute(stmt)).scalar_one_or_none()
+
+
+async def link(
+    *,
+    db: AsyncSession,
+    spool_id: int,
+    uid: str,
+    tag_format: str | None = None,
+) -> models.SpoolTag:
+    """Link a physical tag to a spool.
+
+    Re-linking a UID to the spool that already holds it is idempotent; if the request
+    carries a format and the stored row has a different one, the stored one is refined,
+    since a later scan generally knows more about the tag than the first one did.
+
+    Args:
+        db: Database session.
+        spool_id: The spool to link the tag to.
+        uid: The tag UID in any shape; normalized here.
+        tag_format: Optional tag format name, e.g. "ntag".
+
+    Returns:
+        models.SpoolTag: The linked tag.
+
+    Raises:
+        ItemNotFoundError: If no spool with that ID exists.
+        TagConflictError: If the UID is already linked to a different spool.
+        ValueError: If the UID or format is not valid.
+
+    """
+    uid = normalize_uid(uid)
+    tag_format = normalize_format(tag_format)
+
+    db_spool = await spool.get_by_id(db, spool_id)
+
+    existing = await _get_tag_by_uid(db, uid)
+    if existing is not None:
+        if existing.spool_id != spool_id:
+            raise TagConflictError(
+                f"Tag {uid} is already linked to spool {existing.spool_id}.",
+                existing.spool_id,
+            )
+        if tag_format is not None and existing.format != tag_format:
+            existing.format = tag_format
+            await db.commit()
+            await spool.spool_changed(db_spool, EventType.UPDATED)
+        return existing
+
+    tag = models.SpoolTag(
+        uid=uid,
+        format=tag_format,
+        added=datetime.utcnow().replace(microsecond=0),
+    )
+    db_spool.tags.append(tag)
+    try:
+        await db.commit()
+    except IntegrityError:
+        # Two clients linked the same UID at the same time and the unique index caught
+        # the loser. The database, not the read above, is what makes "one tag, one spool"
+        # true; report the winner the same way a sequential conflict is reported.
+        await db.rollback()
+        winner = await _get_tag_by_uid(db, uid)
+        if winner is None:
+            raise
+        raise TagConflictError(
+            f"Tag {uid} is already linked to spool {winner.spool_id}.",
+            winner.spool_id,
+        ) from None
+
+    await spool.spool_changed(db_spool, EventType.UPDATED)
+    return tag
+
+
+async def unlink(*, db: AsyncSession, spool_id: int, uid: str) -> None:
+    """Unlink a tag from a spool.
+
+    Args:
+        db: Database session.
+        spool_id: The spool the tag is linked to.
+        uid: The tag UID in any shape; normalized here.
+
+    Raises:
+        ItemNotFoundError: If the spool does not exist, or does not hold that tag.
+        ValueError: If the UID is not valid.
+
+    """
+    uid = normalize_uid(uid)
+
+    db_spool = await spool.get_by_id(db, spool_id)
+    for tag in db_spool.tags:
+        if tag.uid == uid:
+            # delete-orphan on the relationship turns this into the DELETE.
+            db_spool.tags.remove(tag)
+            break
+    else:
+        raise ItemNotFoundError(f"Spool {spool_id} has no tag with UID {uid}.")
+
+    # Commit before notifying so the change is durable and visible to subsequent
+    # requests; post-commit notification must be the last, infallible step.
+    await db.commit()
+    await spool.spool_changed(db_spool, EventType.UPDATED)
+
+
+async def find_spool_by_uid(db: AsyncSession, uid: str) -> models.Spool | None:
+    """Find the spool a tag UID is linked to, or None if the tag is unknown.
+
+    Args:
+        db: Database session.
+        uid: The tag UID in any shape; normalized here.
+
+    Returns:
+        models.Spool | None: The spool holding that tag.
+
+    Raises:
+        ValueError: If the UID is not valid.
+
+    """
+    tag = await _get_tag_by_uid(db, normalize_uid(uid))
+    if tag is None:
+        return None
+    return await spool.get_by_id(db, tag.spool_id)
diff --git a/spoolman/exceptions.py b/spoolman/exceptions.py
index 3ebf26c8..60df8772 100644
--- a/spoolman/exceptions.py
+++ b/spoolman/exceptions.py
@@ -15,3 +15,17 @@ class ItemCreateError(Exception):
 
 class SpoolMeasureError(Exception):
     pass
+
+
+class TagConflictError(Exception):
+    """A tag UID is already linked to a different spool.
+
+    Carries the conflicting spool's ID rather than only a message, so the API can hand it
+    to the client and the client can offer "move the tag to this spool instead" without a
+    second lookup.
+    """
+
+    def __init__(self, message: str, spool_id: int) -> None:
+        """Initialize with the message and the ID of the spool already holding the tag."""
+        super().__init__(message)
+        self.spool_id = spool_id
diff --git a/spoolman/export.py b/spoolman/export.py
index 0b985260..bea5f962 100644
--- a/spoolman/export.py
+++ b/spoolman/export.py
@@ -51,6 +51,14 @@ async def flatten_sqlalchemy_object(obj: models.Base, parent_key: str = "", sep:
                     fields[f"{parent_key}extra.{v.key}"] = v.value
                 continue
 
+            if attr == "tags":
+                # A collection, like `extra`, so it needs collapsing into a cell rather
+                # than being left as a list of ORM objects: CSV would write their repr and
+                # JSON would stringify them. A spool's tags have no keys to spread across
+                # columns, so they become one comma-separated list of UIDs.
+                fields[f"{parent_key}tags"] = ",".join(v.uid for v in value)
+                continue
+
             # Handle nested SQLAlchemy objects
             if isinstance(value, models.Base):
                 nested_fields = await flatten_sqlalchemy_object(value, f"{parent_key}{attr}{sep}", sep=sep)
diff --git a/spoolman/scanrelay.py b/spoolman/scanrelay.py
new file mode 100644
index 00000000..d41993de
--- /dev/null
+++ b/spoolman/scanrelay.py
@@ -0,0 +1,148 @@
+"""In-memory state for the tag scan relay: scan debouncing and the reader registry.
+
+Relay state is neither database state nor HTTP state, so it lives on its own. Isolating it
+means the debounce window and the registry's eviction are testable without a socket or a
+session.
+
+Everything here is per-process and deliberately not persisted. `entrypoint.sh` runs a single
+uvicorn process with no `--workers`, so a module-level registry and debounce cache are correct
+as deployed; a user who adds `--workers 4` gets a partitioned registry and a per-worker
+debounce window. That is a documented limit, not a reason to add Redis or a table for state
+whose whole lifetime is "a few seconds" or "until the next restart".
+"""
+
+import re
+import threading
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+# Readers poll continuously -- nfc2klipper re-reads a tag for as long as it sits on the
+# reader -- so the same UID fires over and over. Suppressing repeats server-side, before the
+# broadcast, means every subscriber benefits and no device has to be well-behaved for the UI
+# to be.
+DEBOUNCE_WINDOW = timedelta(seconds=3)
+
+# How long a reader stays listed after its last scan. Long, because the list exists to fill a
+# "choose a reader" dropdown: a reader tapped this morning should still be pickable this
+# afternoon, and a stale entry is a much smaller annoyance than a missing one. The registry is
+# in-memory, so a restart clears it regardless and this is really "seen since startup".
+READER_TTL = timedelta(hours=24)
+
+# Hard caps. Spoolman has no authentication, so anything on the LAN can post scans; these keep
+# a broken or hostile agent from growing either structure without bound. Both evict oldest
+# first, so a working setup never notices them.
+MAX_READERS = 256
+MAX_DEBOUNCE_ENTRIES = 1024
+
+READER_ID_MAX_LENGTH = 64
+
+# What a reader_id may contain. It is operator-chosen, travels in a websocket path, and is
+# echoed back in events, so keep it to characters that are unambiguous in all three places.
+READER_ID_PATTERN = r"^[A-Za-z0-9._:-]{1,64}$"
+
+_NON_ID_CHARS = re.compile(r"[^A-Za-z0-9]+")
+
+
+def derive_reader_id(client_host: str | None) -> str:
+    """Derive a stable reader id from a client's network address.
+
+    Some setups are a dozen lines of ESPHome YAML and will never be rewritten to send a
+    reader_id. Deriving one from the address is ugly and breaks on DHCP churn, but it is
+    stable enough that tap-to-pair works against a completely unmodified agent, which is the
+    point.
+    """
+    sanitized = _NON_ID_CHARS.sub("-", client_host or "").strip("-")
+    if not sanitized:
+        return "unknown"
+    return f"ip-{sanitized}"[:READER_ID_MAX_LENGTH]
+
+
+@dataclass
+class Reader:
+    """A reader that has reported a scan."""
+
+    reader_id: str
+    name: str | None
+    last_seen: datetime
+
+
+class ScanRelay:
+    """Debounce cache and reader registry for the tag scan relay."""
+
+    def __init__(
+        self,
+        debounce_window: timedelta = DEBOUNCE_WINDOW,
+        reader_ttl: timedelta = READER_TTL,
+    ) -> None:
+        """Initialize an empty relay."""
+        self.debounce_window = debounce_window
+        self.reader_ttl = reader_ttl
+        # (uid, reader_id) -> when it was last broadcast.
+        self._last_broadcast: dict[tuple[str, str], datetime] = {}
+        self._readers: dict[str, Reader] = {}
+        # Scans arrive on the event loop, but the loop is not the only thing that can touch
+        # this: pruning walks the dicts while a scan may be inserting into them. A plain lock
+        # is enough and is never held across an await.
+        self._lock = threading.Lock()
+
+    def register(self, reader_id: str, name: str | None, now: datetime | None = None) -> None:
+        """Record that a reader reported a scan just now."""
+        now = now or datetime.utcnow()
+        with self._lock:
+            # Popping first re-inserts at the end, so the dict stays ordered least- to
+            # most-recently-seen and eviction can drop from the front.
+            existing = self._readers.pop(reader_id, None)
+            # A reader that stops sending a name keeps the one it last gave, so a scan from a
+            # trimmed-down agent doesn't blank the label in someone's dropdown.
+            self._readers[reader_id] = Reader(
+                reader_id=reader_id,
+                name=name if name is not None else (existing.name if existing else None),
+                last_seen=now,
+            )
+            self._prune_readers(now)
+            while len(self._readers) > MAX_READERS:
+                self._readers.pop(next(iter(self._readers)))
+
+    def should_broadcast(self, uid: str, reader_id: str, now: datetime | None = None) -> bool:
+        """Whether this scan is new enough to broadcast, recording it if so.
+
+        Only the fan-out is debounced. The caller still answers the device on every POST: a
+        de-duplicated scan must not look to the device like a failed lookup.
+        """
+        now = now or datetime.utcnow()
+        key = (uid, reader_id)
+        with self._lock:
+            self._prune_debounce(now)
+            last = self._last_broadcast.get(key)
+            if last is not None and now - last < self.debounce_window:
+                return False
+            self._last_broadcast[key] = now
+            while len(self._last_broadcast) > MAX_DEBOUNCE_ENTRIES:
+                self._last_broadcast.pop(next(iter(self._last_broadcast)))
+            return True
+
+    def readers(self, now: datetime | None = None) -> list[Reader]:
+        """List the readers seen within the TTL, most recently seen first.
+
+        Eviction happens here rather than on a timer: there is no background task to start, no
+        lifecycle hook to register, and nothing to shut down cleanly.
+        """
+        now = now or datetime.utcnow()
+        with self._lock:
+            self._prune_readers(now)
+            return sorted(self._readers.values(), key=lambda reader: reader.last_seen, reverse=True)
+
+    def _prune_readers(self, now: datetime) -> None:
+        """Drop readers past their TTL. Caller holds the lock."""
+        cutoff = now - self.reader_ttl
+        for reader_id in [rid for rid, reader in self._readers.items() if reader.last_seen < cutoff]:
+            del self._readers[reader_id]
+
+    def _prune_debounce(self, now: datetime) -> None:
+        """Drop debounce entries older than the window. Caller holds the lock."""
+        cutoff = now - self.debounce_window
+        for key in [k for k, seen in self._last_broadcast.items() if seen < cutoff]:
+            del self._last_broadcast[key]
+
+
+scan_relay = ScanRelay()
diff --git a/spoolman/tags.py b/spoolman/tags.py
new file mode 100644
index 00000000..ffbec3cb
--- /dev/null
+++ b/spoolman/tags.py
@@ -0,0 +1,96 @@
+"""NFC/RFID tag identity: UID normalization and the known tag formats.
+
+Spoolman keys a tag on its hardware UID and nothing else, which is the one thing every
+tag has -- a blank NTAG215 sticker, a Prusa NFC-V tag and a Bambu MIFARE tag whose
+contents we can never decrypt all report one. Readers, however, report that UID in
+whatever shape their library happens to use: Web NFC's ``serialNumber`` is
+colon-separated lowercase, ESPHome emits ``04-a2-b3``, nfcpy hands you raw bytes that
+most scripts ``.hex()`` into unseparated lowercase.
+
+``normalize_uid`` collapses all of those to one canonical spelling so that the unique
+index on ``spool_tag.uid`` actually means "one tag, one spool". Without it the same
+physical tag could be linked to three spools under three spellings and no constraint
+would notice.
+
+Pure module: no database, no FastAPI. Same shape as ``colors.py`` and ``math.py``.
+"""
+
+import re
+
+# Mirrors the SpoolTag.uid column width, so an over-long UID is a 400 and not a 500.
+UID_MAX_LENGTH = 64
+
+# Mirrors the SpoolTag.format column width.
+FORMAT_MAX_LENGTH = 32
+
+# Tag formats Spoolman knows the name of. Deliberately NOT an enum and NOT enforced:
+# `format` is informational in phase 1, new tag types appear faster than migrations
+# should, and rejecting an unknown one would make Spoolman the thing standing between a
+# user and their hardware. These exist to document the vocabulary and to keep clients
+# spelling the common ones the same way.
+KNOWN_FORMATS = (
+    "openprinttag",
+    "opentag3d",
+    "ntag",
+    "bambu",
+    "tigertag",
+    "qidi",
+    "creality",
+    "prusa",
+)
+
+# Separators readers put between UID bytes, plus the whitespace that survives a
+# copy-paste out of a log line.
+_SEPARATORS = re.compile(r"[\s:_.-]+")
+
+_HEX_ONLY = re.compile(r"^[0-9A-F]+$")
+
+
+def normalize_uid(uid: str) -> str:
+    """Normalize a tag UID to its canonical form: uppercase hex, no separators.
+
+    Accepts any separator style and any case, plus an optional ``0x`` prefix, so a
+    reader-side agent never has to care which shape its NFC library produces.
+
+    Args:
+        uid: The UID as reported by a reader, e.g. ``04:a2:b3:c4``, ``04-A2-B3-C4`` or
+            ``04a2b3c4``.
+
+    Returns:
+        str: The canonical UID, e.g. ``04A2B3C4``.
+
+    Raises:
+        ValueError: If the UID is empty, is not hexadecimal, or is longer than the
+            column can hold. A tag UID is a byte string by definition, so anything else
+            is a caller bug worth surfacing rather than a row worth storing -- a
+            junk-drawer uid column would make the uniqueness guarantee meaningless.
+
+    """
+    normalized = _SEPARATORS.sub("", uid).strip().upper().removeprefix("0X")
+    if not normalized:
+        raise ValueError("A tag UID must not be empty.")
+    if not _HEX_ONLY.match(normalized):
+        raise ValueError(f'"{uid}" is not a valid tag UID. A UID must be hexadecimal.')
+    if len(normalized) > UID_MAX_LENGTH:
+        raise ValueError(f"A tag UID can be at most {UID_MAX_LENGTH} hex characters.")
+    return normalized
+
+
+def normalize_format(tag_format: str | None) -> str | None:
+    """Normalize a tag format name, or return None if none was given.
+
+    Lowercased and trimmed so that ``NTAG`` and ``ntag`` are the same format. Unknown
+    values pass through unchanged -- see KNOWN_FORMATS.
+
+    Raises:
+        ValueError: If the name is longer than the column can hold.
+
+    """
+    if tag_format is None:
+        return None
+    normalized = tag_format.strip().lower()
+    if not normalized:
+        return None
+    if len(normalized) > FORMAT_MAX_LENGTH:
+        raise ValueError(f"A tag format can be at most {FORMAT_MAX_LENGTH} characters.")
+    return normalized
diff --git a/tests/test_scan_relay.py b/tests/test_scan_relay.py
new file mode 100644
index 00000000..ac292efe
--- /dev/null
+++ b/tests/test_scan_relay.py
@@ -0,0 +1,127 @@
+"""Tests for the scan relay's debounce cache and reader registry.
+
+Both are pure in-memory state, so they are tested here without a socket or a session. Time is
+injected rather than slept through, which is what keeps this file fast enough to be worth
+running on every change.
+"""
+
+import re
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from spoolman.scanrelay import MAX_READERS, READER_ID_PATTERN, ScanRelay, derive_reader_id
+
+# The relay compares datetimes it is given against each other only, so the tzinfo here is
+# just to keep the value unambiguous; the server passes naive UTC, as the rest of the code does.
+T0 = datetime(2026, 8, 13, 12, 0, 0, tzinfo=timezone.utc)
+
+
+@pytest.fixture
+def relay() -> ScanRelay:
+    return ScanRelay(debounce_window=timedelta(seconds=3), reader_ttl=timedelta(hours=1))
+
+
+def test_first_scan_broadcasts(relay: ScanRelay):
+    assert relay.should_broadcast("04A2", "desk", now=T0) is True
+
+
+def test_repeat_within_the_window_is_suppressed(relay: ScanRelay):
+    """A reader re-reads a tag that is sitting still; subscribers should see one event."""
+    relay.should_broadcast("04A2", "desk", now=T0)
+    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=1)) is False
+    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=2.9)) is False
+
+
+def test_repeat_after_the_window_broadcasts_again(relay: ScanRelay):
+    """Tapping the same tag again later is a new scan, not a duplicate."""
+    relay.should_broadcast("04A2", "desk", now=T0)
+    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=4)) is True
+
+
+def test_debounce_is_per_uid_and_per_reader(relay: ScanRelay):
+    """Two readers tapping at once, or one reader seeing two tags, are distinct scans."""
+    relay.should_broadcast("04A2", "desk", now=T0)
+    assert relay.should_broadcast("04A2", "printer", now=T0) is True
+    assert relay.should_broadcast("B1C2", "desk", now=T0) is True
+
+
+def test_debounce_entries_are_pruned(relay: ScanRelay):
+    """The cache does not grow without bound as tags come and go."""
+    for i in range(50):
+        relay.should_broadcast(f"UID{i:04X}", "desk", now=T0)
+    assert len(relay._last_broadcast) == 50  # noqa: SLF001
+    relay.should_broadcast("LATER", "desk", now=T0 + timedelta(seconds=30))
+    assert len(relay._last_broadcast) == 1  # noqa: SLF001
+
+
+def test_readers_are_listed_most_recently_seen_first(relay: ScanRelay):
+    relay.register("desk", "Desk reader", now=T0)
+    relay.register("printer", "Voron spool holder", now=T0 + timedelta(minutes=1))
+
+    readers = relay.readers(now=T0 + timedelta(minutes=2))
+    assert [r.reader_id for r in readers] == ["printer", "desk"]
+    assert readers[0].name == "Voron spool holder"
+    assert readers[1].last_seen == T0
+
+
+def test_registry_is_empty_before_anything_scans(relay: ScanRelay):
+    """The registry is not persisted, so this is also the state after a restart."""
+    assert relay.readers(now=T0) == []
+
+
+def test_reader_keeps_its_name_when_a_later_scan_omits_one(relay: ScanRelay):
+    """A trimmed-down agent must not blank the label in someone's picker."""
+    relay.register("desk", "Desk reader", now=T0)
+    relay.register("desk", None, now=T0 + timedelta(minutes=1))
+
+    readers = relay.readers(now=T0 + timedelta(minutes=2))
+    assert readers[0].name == "Desk reader"
+    assert readers[0].last_seen == T0 + timedelta(minutes=1)
+
+
+def test_a_new_name_replaces_the_old_one(relay: ScanRelay):
+    relay.register("desk", "Desk reader", now=T0)
+    relay.register("desk", "Bench reader", now=T0 + timedelta(minutes=1))
+    assert relay.readers(now=T0 + timedelta(minutes=2))[0].name == "Bench reader"
+
+
+def test_readers_past_their_ttl_are_evicted_on_read(relay: ScanRelay):
+    """Eviction happens as the response is built: no background task, nothing to shut down."""
+    relay.register("desk", None, now=T0)
+    assert len(relay.readers(now=T0 + timedelta(minutes=59))) == 1
+    assert relay.readers(now=T0 + timedelta(hours=2)) == []
+
+
+def test_registry_is_capped(relay: ScanRelay):
+    """An unauthenticated endpoint cannot be used to grow the registry without bound."""
+    for i in range(MAX_READERS + 20):
+        relay.register(f"reader-{i}", None, now=T0 + timedelta(seconds=i))
+
+    readers = relay.readers(now=T0 + timedelta(minutes=30))
+    assert len(readers) == MAX_READERS
+    # The least recently seen are the ones dropped.
+    assert readers[0].reader_id == f"reader-{MAX_READERS + 19}"
+    assert {r.reader_id for r in readers} == {f"reader-{i}" for i in range(20, MAX_READERS + 20)}
+
+
+@pytest.mark.parametrize(
+    ("host", "expected"),
+    [
+        ("192.168.1.50", "ip-192-168-1-50"),
+        ("10.0.0.1", "ip-10-0-0-1"),
+        ("fe80::1", "ip-fe80-1"),
+        ("testclient", "ip-testclient"),
+        (None, "unknown"),
+        ("", "unknown"),
+        ("...", "unknown"),
+    ],
+)
+def test_reader_id_is_derived_from_the_client_address(host: str | None, expected: str):
+    """An unmodified agent that sends no reader_id still gets a stable id to pair against."""
+    assert derive_reader_id(host) == expected
+
+
+def test_derived_reader_id_fits_the_pattern():
+    """Whatever comes out must be usable as a websocket path segment."""
+    assert re.match(READER_ID_PATTERN, derive_reader_id("a" * 200))
diff --git a/tests/test_tag_uid.py b/tests/test_tag_uid.py
new file mode 100644
index 00000000..61290166
--- /dev/null
+++ b/tests/test_tag_uid.py
@@ -0,0 +1,91 @@
+"""Tests for tag UID normalization.
+
+The unique index on `spool_tag.uid` only means "one tag, one spool" because every write goes
+through `normalize_uid`. These pin the shapes real readers produce, so a change that stopped
+collapsing one of them would show up here rather than as two spools quietly claiming the same
+physical tag.
+"""
+
+import pytest
+
+from spoolman.tags import UID_MAX_LENGTH, normalize_format, normalize_uid
+
+
+@pytest.mark.parametrize(
+    "reported",
+    [
+        "04A2B3C4D5E6F7",  # already canonical
+        "04a2b3c4d5e6f7",  # nfcpy: bytes.hex()
+        "04:a2:b3:c4:d5:e6:f7",  # Web NFC serialNumber
+        "04:A2:B3:C4:D5:E6:F7",
+        "04-a2-b3-c4-d5-e6-f7",  # ESPHome on_tag
+        "04-A2-B3-C4-D5-E6-F7",
+        "04 a2 b3 c4 d5 e6 f7",  # pasted out of a log line
+        "04_A2_B3_C4_D5_E6_F7",
+        "04.a2.b3.c4.d5.e6.f7",
+        "0x04a2b3c4d5e6f7",
+        "  04a2b3c4d5e6f7  ",
+    ],
+)
+def test_every_reader_shape_collapses_to_one_uid(reported: str):
+    """Every way a reader can spell one tag resolves to the same stored UID."""
+    assert normalize_uid(reported) == "04A2B3C4D5E6F7"
+
+
+def test_normalization_is_idempotent():
+    """Normalizing an already-normalized UID is a no-op, so layers may normalize twice."""
+    once = normalize_uid("04:a2:b3:c4")
+    assert normalize_uid(once) == once
+
+
+@pytest.mark.parametrize(
+    "bad",
+    [
+        "",
+        "   ",
+        ":::",
+        "----",
+        "not-a-uid",
+        "04A2B3G4",  # G is not hex
+        "04A2B3C4!",
+        "0x",
+    ],
+)
+def test_rejects_uids_that_are_not_hex(bad: str):
+    """A UID is a byte string by definition; anything else is a caller bug, not a row."""
+    with pytest.raises(ValueError, match="UID"):
+        normalize_uid(bad)
+
+
+def test_rejects_uid_longer_than_the_column():
+    """An over-long UID is a 400, not a database error."""
+    normalize_uid("A" * UID_MAX_LENGTH)  # the limit itself is fine
+    with pytest.raises(ValueError, match="at most"):
+        normalize_uid("A" * (UID_MAX_LENGTH + 1))
+
+
+def test_separators_do_not_count_towards_the_length_limit():
+    """The limit applies to the stored UID, not to what the reader happened to send."""
+    assert len(normalize_uid(":".join("AB" * (UID_MAX_LENGTH // 2)))) == UID_MAX_LENGTH
+
+
+@pytest.mark.parametrize(
+    ("reported", "expected"),
+    [
+        ("ntag", "ntag"),
+        ("NTAG", "ntag"),
+        ("  OpenPrintTag  ", "openprinttag"),
+        ("something-nobody-has-shipped-yet", "something-nobody-has-shipped-yet"),
+        (None, None),
+        ("", None),
+        ("   ", None),
+    ],
+)
+def test_format_is_normalized_but_not_restricted(reported: str | None, expected: str | None):
+    """Formats are lowercased for consistency, and unknown ones pass through unchanged."""
+    assert normalize_format(reported) == expected
+
+
+def test_rejects_format_longer_than_the_column():
+    with pytest.raises(ValueError, match="at most"):
+        normalize_format("f" * 33)
diff --git a/tests_integration/tests/spool/test_tags.py b/tests_integration/tests/spool/test_tags.py
new file mode 100644
index 00000000..a961394a
--- /dev/null
+++ b/tests_integration/tests/spool/test_tags.py
@@ -0,0 +1,298 @@
+"""Integration tests: linking physical NFC/RFID tags to spools, and looking a spool up by one.
+
+The point of the `spool_tag` table is that one tag identifies exactly one spool, enforced by the
+database rather than by whichever client wrote last. These tests pin that from both ends: the
+409 that stops a second spool claiming a tag, and the normalization that stops the same physical
+tag being linked twice under two spellings.
+"""
+
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+import httpx
+import pytest
+
+from ..conftest import URL, assert_httpx_code, assert_httpx_success
+
+
+def _uid() -> str:
+    """Build a UID no other test in the session will produce. Hex, as a real UID is."""
+    return uuid.uuid4().hex[:14].upper()
+
+
+@contextmanager
+def _spool(filament_id: int, **kwargs: object) -> Iterator[dict[str, Any]]:
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": filament_id, **kwargs})
+    assert_httpx_success(result)
+    spool = result.json()
+    try:
+        yield spool
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+
+def _get_spool(spool_id: int) -> dict[str, Any]:
+    result = httpx.get(f"{URL}/api/v1/spool/{spool_id}")
+    assert_httpx_success(result)
+    return result.json()
+
+
+def test_link_tag(random_filament: dict[str, Any]):
+    """Linking returns the stored tag and adds it to the spool."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        result = httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid, "format": "ntag"})
+        assert_httpx_code(result, 201)
+
+        tag = result.json()
+        assert tag["uid"] == uid
+        assert tag["format"] == "ntag"
+        assert "added" in tag
+
+        assert _get_spool(spool["id"])["tags"] == [tag]
+
+
+def test_spool_with_no_tags_has_an_empty_list(random_filament: dict[str, Any]):
+    """`tags` is always present, so a client never has to distinguish absent from empty."""
+    with _spool(random_filament["id"]) as spool:
+        assert spool["tags"] == []
+        assert _get_spool(spool["id"])["tags"] == []
+
+
+def test_link_tag_without_a_format(random_filament: dict[str, Any]):
+    """Format is optional; a reader that only knows the UID can still link."""
+    with _spool(random_filament["id"]) as spool:
+        result = httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": _uid()})
+        assert_httpx_code(result, 201)
+        # exclude_none omits it rather than sending an explicit null.
+        assert "format" not in result.json()
+
+
+@pytest.mark.parametrize(
+    "reported",
+    ["{uid}", "{lower}", "{colons}", "{dashes}", "{spaces}"],
+)
+def test_uid_is_normalized_on_write(random_filament: dict[str, Any], reported: str):
+    """Every shape a reader can report writes the same canonical UID."""
+    uid = _uid()
+    spelling = reported.format(
+        uid=uid,
+        lower=uid.lower(),
+        colons=":".join(uid[i : i + 2] for i in range(0, len(uid), 2)),
+        dashes="-".join(uid[i : i + 2] for i in range(0, len(uid), 2)).lower(),
+        spaces=" ".join(uid[i : i + 2] for i in range(0, len(uid), 2)),
+    )
+    with _spool(random_filament["id"]) as spool:
+        result = httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": spelling})
+        assert_httpx_code(result, 201)
+        assert result.json()["uid"] == uid
+
+
+def test_link_is_idempotent(random_filament: dict[str, Any]):
+    """Re-linking a tag to the spool that already holds it succeeds and adds nothing."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        for _ in range(3):
+            assert_httpx_code(httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}), 201)
+
+        tags = _get_spool(spool["id"])["tags"]
+        assert len(tags) == 1
+        assert tags[0]["uid"] == uid
+
+
+def test_relinking_with_a_format_refines_the_stored_one(random_filament: dict[str, Any]):
+    """A later scan generally knows more about the tag than the first one did."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+        result = httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid, "format": "openprinttag"})
+        assert_httpx_code(result, 201)
+
+        tags = _get_spool(spool["id"])["tags"]
+        assert len(tags) == 1
+        assert tags[0]["format"] == "openprinttag"
+
+
+def test_duplicate_uid_on_another_spool_is_a_conflict(random_filament: dict[str, Any]):
+    """One tag, one spool. The conflicting spool's id comes back so a client can offer to move it."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as first, _spool(random_filament["id"]) as second:
+        httpx.post(f"{URL}/api/v1/spool/{first['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        result = httpx.post(f"{URL}/api/v1/spool/{second['id']}/tag", json={"uid": uid})
+        assert_httpx_code(result, 409)
+        assert result.json()["spool_id"] == first["id"]
+        assert "message" in result.json()
+
+        # The losing spool got nothing.
+        assert _get_spool(second["id"])["tags"] == []
+
+
+def test_conflict_is_detected_across_uid_spellings(random_filament: dict[str, Any]):
+    """A different spelling of a linked tag is the same tag, so it conflicts.
+
+    Normalization is what makes the unique index mean anything: without it, one physical tag
+    could occupy three rows on three spools and no constraint would notice.
+    """
+    uid = _uid()
+    with _spool(random_filament["id"]) as first, _spool(random_filament["id"]) as second:
+        httpx.post(f"{URL}/api/v1/spool/{first['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        colons = ":".join(uid[i : i + 2] for i in range(0, len(uid), 2)).lower()
+        result = httpx.post(f"{URL}/api/v1/spool/{second['id']}/tag", json={"uid": colons})
+        assert_httpx_code(result, 409)
+        assert result.json()["spool_id"] == first["id"]
+
+
+def test_many_tags_on_one_spool(random_filament: dict[str, Any]):
+    """A spool can carry a vendor tag and a copy of it on a blank sticker; a real case (#776)."""
+    uids = sorted(_uid() for _ in range(3))
+    with _spool(random_filament["id"]) as spool:
+        for uid in uids:
+            httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        assert sorted(tag["uid"] for tag in _get_spool(spool["id"])["tags"]) == uids
+
+
+def test_unlink_tag(random_filament: dict[str, Any]):
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        assert_httpx_code(httpx.delete(f"{URL}/api/v1/spool/{spool['id']}/tag/{uid}"), 204)
+        assert _get_spool(spool["id"])["tags"] == []
+
+
+def test_unlink_accepts_any_uid_spelling(random_filament: dict[str, Any]):
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+        assert_httpx_code(httpx.delete(f"{URL}/api/v1/spool/{spool['id']}/tag/{uid.lower()}"), 204)
+        assert _get_spool(spool["id"])["tags"] == []
+
+
+def test_unlinking_frees_the_uid(random_filament: dict[str, Any]):
+    """After unlinking, another spool can take the tag -- the unique row really is gone."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as first, _spool(random_filament["id"]) as second:
+        httpx.post(f"{URL}/api/v1/spool/{first['id']}/tag", json={"uid": uid}).raise_for_status()
+        httpx.delete(f"{URL}/api/v1/spool/{first['id']}/tag/{uid}").raise_for_status()
+
+        assert_httpx_code(httpx.post(f"{URL}/api/v1/spool/{second['id']}/tag", json={"uid": uid}), 201)
+
+
+def test_unlink_unknown_tag_is_404(random_filament: dict[str, Any]):
+    with _spool(random_filament["id"]) as spool:
+        assert_httpx_code(httpx.delete(f"{URL}/api/v1/spool/{spool['id']}/tag/{_uid()}"), 404)
+
+
+def test_unlink_tag_belonging_to_another_spool_is_404(random_filament: dict[str, Any]):
+    """A tag is only unlinkable from the spool that holds it."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as first, _spool(random_filament["id"]) as second:
+        httpx.post(f"{URL}/api/v1/spool/{first['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        assert_httpx_code(httpx.delete(f"{URL}/api/v1/spool/{second['id']}/tag/{uid}"), 404)
+        assert len(_get_spool(first["id"])["tags"]) == 1
+
+
+def test_link_to_unknown_spool_is_404():
+    assert_httpx_code(httpx.post(f"{URL}/api/v1/spool/-1/tag", json={"uid": _uid()}), 404)
+
+
+@pytest.mark.parametrize("bad", ["", "not-a-uid", "04A2B3G4", "!!!"])
+def test_link_with_an_invalid_uid_is_rejected(random_filament: dict[str, Any], bad: str):
+    """A UID is hexadecimal by definition; junk is a client bug, not a row worth storing."""
+    with _spool(random_filament["id"]) as spool:
+        result = httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": bad})
+        # An empty UID fails Pydantic's min_length (422); the rest fail normalization (400).
+        assert result.status_code in (400, 422), result.text
+
+
+def test_deleting_a_spool_removes_its_tags(random_filament: dict[str, Any]):
+    """Cascade: the tag row must go with the spool, or its UID would be permanently unusable."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as doomed:
+        httpx.post(f"{URL}/api/v1/spool/{doomed['id']}/tag", json={"uid": uid}).raise_for_status()
+        httpx.delete(f"{URL}/api/v1/spool/{doomed['id']}").raise_for_status()
+
+    with _spool(random_filament["id"]) as fresh:
+        assert_httpx_code(httpx.post(f"{URL}/api/v1/spool/{fresh['id']}/tag", json={"uid": uid}), 201)
+
+
+def test_find_spool_by_tag(random_filament: dict[str, Any]):
+    """Answer the lookup the whole ecosystem has been working around: which spool is this tag."""
+    uid = _uid()
+    with _spool(random_filament["id"]) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        result = httpx.get(f"{URL}/api/v1/spool", params={"tag": uid})
+        assert_httpx_success(result)
+
+        found = result.json()
+        assert len(found) == 1
+        assert found[0]["id"] == spool["id"]
+        assert result.headers["x-total-count"] == "1"
+
+
+@pytest.mark.parametrize("style", ["lower", "colons", "dashes"])
+def test_find_by_tag_normalizes_the_query(random_filament: dict[str, Any], style: str):
+    """A reader querying with its own spelling finds the spool it linked."""
+    uid = _uid()
+    spellings = {
+        "lower": uid.lower(),
+        "colons": ":".join(uid[i : i + 2] for i in range(0, len(uid), 2)),
+        "dashes": "-".join(uid[i : i + 2] for i in range(0, len(uid), 2)).lower(),
+    }
+    with _spool(random_filament["id"]) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        result = httpx.get(f"{URL}/api/v1/spool", params={"tag": spellings[style]})
+        assert_httpx_success(result)
+        assert [s["id"] for s in result.json()] == [spool["id"]]
+
+
+def test_find_by_unknown_tag_is_empty():
+    """An unknown tag is not an error; it is a tag nobody has linked yet."""
+    result = httpx.get(f"{URL}/api/v1/spool", params={"tag": _uid()})
+    assert_httpx_success(result)
+    assert result.json() == []
+    assert result.headers["x-total-count"] == "0"
+
+
+@pytest.mark.parametrize("bad", ["not-a-uid", "04A2B3G4"])
+def test_find_by_invalid_tag_is_400(bad: str):
+    assert_httpx_code(httpx.get(f"{URL}/api/v1/spool", params={"tag": bad}), 400)
+
+
+def test_find_by_tag_respects_allow_archived(random_filament: dict[str, Any]):
+    """The tag filter behaves like every other filter: archived spools are hidden by default."""
+    uid = _uid()
+    with _spool(random_filament["id"], archived=True) as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        assert httpx.get(f"{URL}/api/v1/spool", params={"tag": uid}).json() == []
+
+        result = httpx.get(f"{URL}/api/v1/spool", params={"tag": uid, "allow_archived": True})
+        assert_httpx_success(result)
+        assert [s["id"] for s in result.json()] == [spool["id"]]
+
+
+def test_find_by_tag_combines_with_other_filters(random_filament: dict[str, Any]):
+    """The tag join goes through the shared filter builder, so it composes and the count agrees."""
+    uid = _uid()
+    with _spool(random_filament["id"], location="Shelf T") as spool:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        matching = httpx.get(f"{URL}/api/v1/spool", params={"tag": uid, "location": "Shelf T", "limit": 10})
+        assert_httpx_success(matching)
+        assert [s["id"] for s in matching.json()] == [spool["id"]]
+        assert matching.headers["x-total-count"] == "1"
+
+        # A limit makes the count a separate query; it must agree with the rows.
+        missing = httpx.get(f"{URL}/api/v1/spool", params={"tag": uid, "location": "Nowhere", "limit": 10})
+        assert_httpx_success(missing)
+        assert missing.json() == []
+        assert missing.headers["x-total-count"] == "0"
diff --git a/tests_integration/tests/tag/__init__.py b/tests_integration/tests/tag/__init__.py
new file mode 100644
index 00000000..4077b69e
--- /dev/null
+++ b/tests_integration/tests/tag/__init__.py
@@ -0,0 +1 @@
+"""Integration tests for the tag scan relay."""
diff --git a/tests_integration/tests/tag/test_relay_ws.py b/tests_integration/tests/tag/test_relay_ws.py
new file mode 100644
index 00000000..0a2c8cd1
--- /dev/null
+++ b/tests_integration/tests/tag/test_relay_ws.py
@@ -0,0 +1,221 @@
+"""Integration tests for the tag scan websockets.
+
+Two things are being pinned here. The first is the routing the pairing UX depends on: a scan
+must reach the browser that follows *that* reader, and only it. The second is the API v1
+compatibility guard -- scans live in their own subscription tree, so the existing `/api/v1/`
+root stream, whose subscribers asked to hear about changes to data, must never see one.
+"""
+
+import asyncio
+import contextlib
+import json
+import uuid
+from typing import Any
+
+import httpx
+import pytest
+from websockets.asyncio.client import ClientConnection, connect
+
+from ..conftest import URL
+
+WS_URL = URL.replace("http://", "ws://", 1)
+
+# How long to wait for an event we expect NOT to arrive. Long enough that a leak into the
+# wrong pool would reliably land inside it, short enough to keep the suite quick across all
+# four database backends.
+QUIET_PERIOD_S = 2.0
+
+# Long enough to absorb a slow container, short enough to fail rather than hang.
+RECV_TIMEOUT_S = 10.0
+
+
+def _uid() -> str:
+    return uuid.uuid4().hex[:14].upper()
+
+
+def _reader_id() -> str:
+    return f"reader-{uuid.uuid4().hex[:8]}"
+
+
+def _scan(uid: str, reader_id: str, **kwargs: object) -> httpx.Response:
+    return httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": uid, "reader_id": reader_id, **kwargs})
+
+
+async def _expect_nothing(ws: ClientConnection, why: str) -> None:
+    with contextlib.suppress(TimeoutError):
+        event = json.loads(await asyncio.wait_for(ws.recv(), timeout=QUIET_PERIOD_S))
+        pytest.fail(f"{why}: {event}")
+
+
+@pytest.mark.asyncio
+async def test_scan_reaches_the_readers_own_pool():
+    """The subscription a browser holds after tap-to-pair."""
+    uid, reader_id = _uid(), _reader_id()
+
+    async with connect(f"{WS_URL}/api/v1/tag/scan/{reader_id}") as ws:
+        # The subscription is registered synchronously right after the handshake,
+        # but give the server a beat before triggering the event.
+        await asyncio.sleep(0.2)
+        _scan(uid, reader_id, name="Voron spool holder").raise_for_status()
+
+        event = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+
+    assert event["type"] == "scanned"
+    assert event["resource"] == "tag_scan"
+    assert event["payload"]["uid"] == uid
+    assert event["payload"]["reader_id"] == reader_id
+    assert event["payload"]["name"] == "Voron spool holder"
+
+
+@pytest.mark.asyncio
+async def test_scan_does_not_reach_another_readers_pool():
+    """A wall tablet by printer A must not jump when someone taps a tag at printer B."""
+    uid, mine, theirs = _uid(), _reader_id(), _reader_id()
+
+    async with connect(f"{WS_URL}/api/v1/tag/scan/{mine}") as ws:
+        await asyncio.sleep(0.2)
+        _scan(uid, theirs).raise_for_status()
+        await _expect_nothing(ws, "a scan leaked into another reader's pool")
+
+
+@pytest.mark.asyncio
+async def test_root_scan_pool_follows_every_reader():
+    """Pairing a scanner works by listening to everything until the first tap arrives."""
+    reader_a, reader_b = _reader_id(), _reader_id()
+    uid_a, uid_b = _uid(), _uid()
+
+    async with connect(f"{WS_URL}/api/v1/tag/scan") as ws:
+        await asyncio.sleep(0.2)
+        _scan(uid_a, reader_a).raise_for_status()
+        first = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+        _scan(uid_b, reader_b).raise_for_status()
+        second = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+
+    assert (first["payload"]["uid"], first["payload"]["reader_id"]) == (uid_a, reader_a)
+    assert (second["payload"]["uid"], second["payload"]["reader_id"]) == (uid_b, reader_b)
+
+
+@pytest.mark.asyncio
+async def test_scan_carries_the_match(random_filament: dict[str, Any]):
+    """A subscriber navigates from the event alone, without a follow-up request."""
+    uid, reader_id = _uid(), _reader_id()
+
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": random_filament["id"]})
+    result.raise_for_status()
+    spool = result.json()
+    try:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        async with connect(f"{WS_URL}/api/v1/tag/scan/{reader_id}") as ws:
+            await asyncio.sleep(0.2)
+            _scan(uid, reader_id).raise_for_status()
+            event = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+
+        assert event["payload"]["matched_spool_id"] == spool["id"]
+        assert event["payload"]["spool"]["id"] == spool["id"]
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+
+@pytest.mark.asyncio
+async def test_scans_never_reach_the_entity_websockets():
+    """THE API v1 compatibility guard.
+
+    `/api/v1/` subscribes with pool () -- "listen to any changes" -- and SubscriptionTree
+    broadcasts to every level along a path. Had scans been added as a resource on the shared
+    manager, they would have appeared in every existing root consumer's stream. They are on a
+    separate manager precisely so they cannot.
+    """
+    uid, reader_id = _uid(), _reader_id()
+
+    async with connect(f"{WS_URL}/api/v1/") as root, connect(f"{WS_URL}/api/v1/spool") as spools:
+        await asyncio.sleep(0.2)
+        _scan(uid, reader_id).raise_for_status()
+
+        await _expect_nothing(root, "a scan reached the root entity websocket")
+        await _expect_nothing(spools, "a scan reached the spool websocket")
+
+
+@pytest.mark.asyncio
+async def test_repeated_scans_are_broadcast_once():
+    """Readers re-detect a tag that is sitting still; subscribers should see one event."""
+    uid, reader_id = _uid(), _reader_id()
+
+    async with connect(f"{WS_URL}/api/v1/tag/scan/{reader_id}") as ws:
+        await asyncio.sleep(0.2)
+        for _ in range(4):
+            _scan(uid, reader_id).raise_for_status()
+
+        event = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+        assert event["payload"]["uid"] == uid
+
+        await _expect_nothing(ws, "a debounced repeat scan was broadcast")
+
+
+@pytest.mark.asyncio
+async def test_debounce_suppresses_the_broadcast_not_the_response(random_filament: dict[str, Any]):
+    """A de-duplicated scan must not look to the device like a failed lookup."""
+    uid, reader_id = _uid(), _reader_id()
+
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": random_filament["id"]})
+    result.raise_for_status()
+    spool = result.json()
+    try:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        for _ in range(4):
+            scan = _scan(uid, reader_id)
+            scan.raise_for_status()
+            assert scan.json()["matched_spool_id"] == spool["id"]
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+
+@pytest.mark.asyncio
+async def test_distinct_tags_from_one_reader_are_not_debounced():
+    """Debouncing is per (uid, reader): moving from one spool to the next is two scans."""
+    reader_id = _reader_id()
+    uid_a, uid_b = _uid(), _uid()
+
+    async with connect(f"{WS_URL}/api/v1/tag/scan/{reader_id}") as ws:
+        await asyncio.sleep(0.2)
+        _scan(uid_a, reader_id).raise_for_status()
+        _scan(uid_b, reader_id).raise_for_status()
+
+        seen = {
+            json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))["payload"]["uid"],
+            json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))["payload"]["uid"],
+        }
+
+    assert seen == {uid_a, uid_b}
+
+
+@pytest.mark.asyncio
+async def test_linking_a_tag_emits_a_spool_event(random_filament: dict[str, Any]):
+    """Link and unlink ride the existing spool event.
+
+    They mutate a spool as far as any client is concerned, so an open inspector re-renders
+    with the new tag list without a single line of client-side wiring.
+    """
+    uid = _uid()
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": random_filament["id"]})
+    result.raise_for_status()
+    spool = result.json()
+    try:
+        async with connect(f"{WS_URL}/api/v1/spool/{spool['id']}") as ws:
+            await asyncio.sleep(0.2)
+
+            httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+            linked = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+
+            httpx.delete(f"{URL}/api/v1/spool/{spool['id']}/tag/{uid}").raise_for_status()
+            unlinked = json.loads(await asyncio.wait_for(ws.recv(), timeout=RECV_TIMEOUT_S))
+
+        assert linked["type"] == "updated"
+        assert linked["resource"] == "spool"
+        assert [tag["uid"] for tag in linked["payload"]["tags"]] == [uid]
+
+        assert unlinked["type"] == "updated"
+        assert unlinked["payload"]["tags"] == []
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
diff --git a/tests_integration/tests/tag/test_scan.py b/tests_integration/tests/tag/test_scan.py
new file mode 100644
index 00000000..fa388d44
--- /dev/null
+++ b/tests_integration/tests/tag/test_scan.py
@@ -0,0 +1,153 @@
+"""Integration tests for POST /tag/scan and GET /tag/reader.
+
+The scan endpoint is the whole device-side contract: one HTTP POST per tap, and a response a
+device can light an LED from. These tests drive it with plain POSTs, which is also how a user
+can demo the entire browser-side experience with no NFC hardware in the room.
+"""
+
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+import httpx
+import pytest
+
+from ..conftest import URL, assert_httpx_code, assert_httpx_success
+
+
+def _uid() -> str:
+    return uuid.uuid4().hex[:14].upper()
+
+
+def _reader_id() -> str:
+    return f"reader-{uuid.uuid4().hex[:8]}"
+
+
+@contextmanager
+def _spool_with_tag(filament_id: int, uid: str) -> Iterator[dict[str, Any]]:
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": filament_id})
+    assert_httpx_success(result)
+    spool = result.json()
+    try:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+        yield spool
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+
+def test_scan_of_a_known_tag_returns_the_spool(random_filament: dict[str, Any]):
+    """The lookup a device makes on every tap, answered in one round trip."""
+    uid = _uid()
+    with _spool_with_tag(random_filament["id"], uid) as spool:
+        result = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": uid, "reader_id": _reader_id()})
+        assert_httpx_success(result)
+
+        body = result.json()
+        assert body["matched_spool_id"] == spool["id"]
+        assert body["spool"]["id"] == spool["id"]
+        # The embedded spool is the standard object, so a device needs no follow-up request.
+        assert body["spool"]["filament"]["id"] == random_filament["id"]
+
+
+def test_scan_of_an_unknown_tag_reports_no_match():
+    """`matched_spool_id` is present and null -- "not known yet" is an answer, not a missing key."""
+    result = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": _uid(), "reader_id": _reader_id()})
+    assert_httpx_success(result)
+
+    body = result.json()
+    assert "matched_spool_id" in body
+    assert body["matched_spool_id"] is None
+    assert body.get("spool") is None
+
+
+def test_scan_normalizes_and_echoes_the_uid(random_filament: dict[str, Any]):
+    """A device sending its own spelling still matches, and is told the canonical form."""
+    uid = _uid()
+    dashed = "-".join(uid[i : i + 2] for i in range(0, len(uid), 2)).lower()
+    with _spool_with_tag(random_filament["id"], uid) as spool:
+        result = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": dashed, "reader_id": _reader_id()})
+        assert_httpx_success(result)
+
+        body = result.json()
+        assert body["uid"] == uid
+        assert body["matched_spool_id"] == spool["id"]
+
+
+def test_scan_finds_an_archived_spool(random_filament: dict[str, Any]):
+    """Tapping an archived spool's tag should say which spool it is, not pretend it is unknown."""
+    uid = _uid()
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": random_filament["id"], "archived": True})
+    assert_httpx_success(result)
+    spool = result.json()
+    try:
+        httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}).raise_for_status()
+
+        scan = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": uid, "reader_id": _reader_id()})
+        assert_httpx_success(scan)
+        assert scan.json()["matched_spool_id"] == spool["id"]
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+
+@pytest.mark.parametrize("bad", ["not-a-uid", "04A2B3G4", "!!!"])
+def test_scan_with_an_invalid_uid_is_400(bad: str):
+    assert_httpx_code(httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": bad, "reader_id": _reader_id()}), 400)
+
+
+def test_scan_with_a_malformed_reader_id_is_rejected():
+    """reader_id travels in a websocket path, so it is constrained rather than taken as-is."""
+    result = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": _uid(), "reader_id": "has spaces/and/slashes"})
+    assert_httpx_code(result, 422)
+
+
+def test_scanning_registers_the_reader():
+    """The registry backs a 'choose a reader' picker for people who prefer it to tap-to-pair."""
+    reader_id = _reader_id()
+    httpx.post(
+        f"{URL}/api/v1/tag/scan",
+        json={"uid": _uid(), "reader_id": reader_id, "name": "Voron spool holder"},
+    ).raise_for_status()
+
+    result = httpx.get(f"{URL}/api/v1/tag/reader")
+    assert_httpx_success(result)
+
+    readers = {reader["reader_id"]: reader for reader in result.json()}
+    assert reader_id in readers
+    assert readers[reader_id]["name"] == "Voron spool holder"
+    assert "last_seen" in readers[reader_id]
+
+
+def test_reader_id_is_derived_when_the_agent_sends_none():
+    """A dozen lines of ESPHome YAML that will never be rewritten still gets a stable id."""
+    result = httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": _uid()})
+    assert_httpx_success(result)
+
+    reader_id = result.json()["reader_id"]
+    assert reader_id.startswith("ip-")
+
+    listed = httpx.get(f"{URL}/api/v1/tag/reader")
+    assert_httpx_success(listed)
+    assert reader_id in {reader["reader_id"] for reader in listed.json()}
+
+
+def test_scan_stores_nothing(random_filament: dict[str, Any]):
+    """Scans are ephemeral. A scan must not link a tag, create a spool, or change one."""
+    uid = _uid()
+    before = httpx.get(f"{URL}/api/v1/spool", params={"limit": 1})
+    assert_httpx_success(before)
+
+    httpx.post(f"{URL}/api/v1/tag/scan", json={"uid": uid, "reader_id": _reader_id()}).raise_for_status()
+
+    # The scanned UID is still unlinked afterwards, so a spool can claim it.
+    result = httpx.post(f"{URL}/api/v1/spool", json={"filament_id": random_filament["id"]})
+    assert_httpx_success(result)
+    spool = result.json()
+    try:
+        assert_httpx_code(httpx.post(f"{URL}/api/v1/spool/{spool['id']}/tag", json={"uid": uid}), 201)
+    finally:
+        httpx.delete(f"{URL}/api/v1/spool/{spool['id']}")
+
+    after = httpx.get(f"{URL}/api/v1/spool", params={"limit": 1})
+    assert_httpx_success(after)
+    assert after.headers["x-total-count"] == before.headers["x-total-count"]