feat(annotations): persist per-image tag sets in the shared vocabulary

Add database/annotations.py (AnnotationOperations, mixed into
PromptDatabase): read_image_tags, replace_image_tags and add_image_tags
run in one _pair_transaction fenced by the storage epoch, own the byte
record by exact locator and digest, always supply UUID ids, rewrite the
ordered junction and bump tag_revision plus image_index_revision.
Revision mismatches raise AnnotationCommandError("tag-conflict") with
the current set; the identical set at the current or previous revision
is replayed without writes; a union over 100 names is "tag-limit".
normalize_image_tags validates each name with validate_vocabulary_tag_name
and folds case-insensitive repeats keeping the first spelling.
refresh_image_fingerprint, current_image_tags (chunked, fingerprint
matched, newest verification wins) and iter_image_tag_records serve the
S1 cheap reads.

Vocabulary participation: _move_tag_links also carries media_image_tags,
legacy merge_tags moves image-only sources, tag rows gain image_count
(files with any byte record carrying the tag), impact gains images and
rename/delete receipts gain images_changed. vocabulary-state.mjs accepts
the optional counts and names images in the delete text.

Browser fixtures clear the three annotation data tables on reset (never
the metadata row); TESTING.md documents the schema beside favorites.

TESTING.md

@@ -1344,6 +1344,31 @@ concept fixture stars the concept's favorites (each other prompt's first image).
   restore. Old backups without the table restore, and backup plus restore keeps
   WAL-resident favorites and closes every handle (`tests/test_database_backup.py`).
 
+Image annotations schema (#249, `database/annotation_schema.py`): the group
+`media_annotation_metadata` (one row: schema version, `image_index_revision`),
+`media_annotation_files` (one private locator per physical file,
+`normcase(join(realpath(root), relative))`, server-side only),
+`media_annotation_bytes` (one record per observed whole-file SHA-256 with the
+`size`/`mtime_ns` verified when those bytes were last seen and the set's
+`tag_revision`) and `media_image_tags` (the ordered tag set of one byte record,
+joined into the global `tags` registry). It installs like favorites: optional in
+old backups, required once current (`annotation_schema_error`, the single
+`require_library_state` flag), validated read-only before and after a SAVEPOINT
+install and on restore, never backfilled and never stat-ing media
+(`tests/test_annotation_schema.py`). Writers (`database/annotations.py`) always
+supply the UUID ids, run in `_pair_transaction` fenced by the storage epoch, and
+replace or union one byte record's set (`tag-conflict` / `tag-limit` with the
+current set, identical retries `replayed`); vocabulary rename, merge and delete
+carry `media_image_tags` through the one `_move_tag_links` loop, and tag rows,
+impact and command receipts report `image_count` / `images` /
+`images_changed` as files (`tests/test_annotations.py`,
+`tests/test_pair_tag_maintenance.py`, `tests/test_vocabulary.py`). Cheap reads
+(spec amendment S1) use a byte record only while the file's current
+`(size, mtime_ns)` equals its verified fingerprint; an in-place rewrite that
+preserves both keeps the old set matching until the file is next observed. The
+browser fixtures clear `media_image_tags`, `media_annotation_bytes` and
+`media_annotation_files` on reset and never the metadata row.
+
 Budgets: `test_filter_work_is_bounded` always runs. It asserts that a plain first
 page makes 1 link SELECT and a plain cursor page none, that every filter shape
 makes exactly 1, the covering-index plan, and that the per-file values are built

database/annotations.py

@@ -0,0 +1,450 @@
+"""Per-image tag sets (#249): byte-exact storage in the shared vocabulary.
+
+A tag set belongs to one byte record (``media_annotation_bytes``): the
+whole-file SHA-256 of one physical file, owned by the file's private locator
+(``media_annotation_files``). Callers pass verified facts (digest, size,
+``mtime_ns``); nothing here reads the filesystem or resolves an owner through
+prompt links, so tagging one image never changes another image, a prompt or a
+pair.
+
+Commands run in one ``_pair_transaction`` fenced by the storage epoch, like the
+native Library and vocabulary commands. A save is a full replacement guarded
+by the set's ``tag_revision``; the identical set sent again (a lost-response
+retry, or the current set at its own revision) answers ``replayed: true`` and
+writes nothing. ``add_image_tags`` unions into the existing set, keeping the
+stored spelling. Names enter the global ``tags`` registry exactly as spelled,
+so vocabulary rename, merge and delete carry image links with prompt and pair
+links (``PromptDatabase._move_tag_links``).
+
+Cheap reads (spec amendment S1) treat a byte record as *current* only while
+the file's present ``(size, mtime_ns)`` equals the record's verified
+fingerprint; a file without a current record shows no image tags.
+"""
+
+import re
+import sqlite3
+from contextlib import contextmanager
+from typing import Iterator
+from uuid import uuid4
+
+from .annotation_schema import (
+    BYTES_TABLE,
+    FILES_TABLE,
+    IMAGE_TAGS_TABLE,
+    MAX_LOCATOR_LENGTH,
+    METADATA_TABLE,
+)
+from .library_commands import _UUID, LibraryCommandError, _validate_epoch
+from .models import utc_now_iso
+from .vocabulary import validate_vocabulary_tag_name
+
+MAX_IMAGE_TAGS = 100
+_LOOKUP_CHUNK = 500
+_DIGEST = re.compile(r"[0-9a-f]{64}")
+_MAX_INT = 2**63 - 1
+_MIN_INT = -(2**63)
+# Only this module's table constants are interpolated into the SQL below.
+_OWNER = f"SELECT id FROM {FILES_TABLE} WHERE locator=?"  # nosec B608
+_INSERT_OWNER = (
+    f"INSERT INTO {FILES_TABLE}(id,locator,created_at) VALUES(?,?,?)"  # nosec B608
+)
+_BYTE = (
+    f"SELECT id,tag_revision FROM {BYTES_TABLE} "  # nosec B608
+    "WHERE file_id=? AND sha256=?"
+)
+_INSERT_BYTE = (
+    f"INSERT INTO {BYTES_TABLE}(id,file_id,sha256,size,mtime_ns,"  # nosec B608
+    "verified_at,first_seen_at,tag_revision) VALUES(?,?,?,?,?,?,?,0)"
+)
+_REFRESH_BYTE = (
+    f"UPDATE {BYTES_TABLE} SET size=?,mtime_ns=?,verified_at=? WHERE id=?"  # nosec B608
+)
+_REFRESH_BY_LOCATOR = (
+    f"UPDATE {BYTES_TABLE} SET size=?,mtime_ns=?,verified_at=? "  # nosec B608
+    f"WHERE sha256=? AND file_id=({_OWNER})"
+)
+_BYTE_TAGS = (
+    f"SELECT t.name FROM {IMAGE_TAGS_TABLE} mit "  # nosec B608
+    "JOIN tags t ON t.id=mit.tag_id WHERE mit.byte_id=? ORDER BY mit.position"
+)
+_CLEAR_TAGS = f"DELETE FROM {IMAGE_TAGS_TABLE} WHERE byte_id=?"  # nosec B608
+_INSERT_TAG = (
+    f"INSERT INTO {IMAGE_TAGS_TABLE}(byte_id,tag_id,position) "  # nosec B608
+    "VALUES(?,?,?)"
+)
+_BUMP_REVISION = (
+    f"UPDATE {BYTES_TABLE} SET tag_revision=tag_revision+1 WHERE id=?"  # nosec B608
+)
+_REVISION = f"SELECT tag_revision FROM {BYTES_TABLE} WHERE id=?"  # nosec B608
+_BUMP_INDEX = (
+    f"UPDATE {METADATA_TABLE} SET "  # nosec B608
+    "image_index_revision=image_index_revision+1 WHERE singleton_id=1"
+)
+_READ = (
+    "SELECT b.id,b.tag_revision,"  # nosec B608
+    f"(SELECT COUNT(*) FROM {BYTES_TABLE} o WHERE o.file_id=f.id AND o.sha256!=?) "
+    f"FROM {FILES_TABLE} f LEFT JOIN {BYTES_TABLE} b "
+    "ON b.file_id=f.id AND b.sha256=? WHERE f.locator=?"
+)
+_FINGERPRINTS = (
+    "SELECT f.locator,b.id,b.size,b.mtime_ns,b.verified_at "  # nosec B608
+    f"FROM {FILES_TABLE} f JOIN {BYTES_TABLE} b ON b.file_id=f.id "
+    "WHERE f.locator IN ({marks})"
+)
+_TAGS_OF = (
+    f"SELECT mit.byte_id,t.name FROM {IMAGE_TAGS_TABLE} mit "  # nosec B608
+    "JOIN tags t ON t.id=mit.tag_id WHERE mit.byte_id IN ({marks}) "
+    "ORDER BY mit.byte_id,mit.position"
+)
+_RECORDS = (
+    "SELECT f.locator,b.id,b.size,b.mtime_ns,t.name "  # nosec B608
+    f"FROM {FILES_TABLE} f JOIN {BYTES_TABLE} b ON b.file_id=f.id "
+    f"LEFT JOIN {IMAGE_TAGS_TABLE} mit ON mit.byte_id=b.id "
+    "LEFT JOIN tags t ON t.id=mit.tag_id "
+    "ORDER BY f.locator,b.first_seen_at,b.id,mit.position"
+)
+
+
+class AnnotationCommandError(LibraryCommandError):
+    """An image tag command lost a race or hit the set's bound.
+
+    ``code`` is ``tag-conflict`` (the set changed since it was read) or
+    ``tag-limit`` (the union would exceed ``MAX_IMAGE_TAGS``); ``current`` is
+    the stored set ``{sha256, tags, tag_revision}`` read inside the
+    transaction, so the caller can show it.
+    """
+
+    def __init__(self, code: str, epoch: str, current: dict | None = None):
+        super().__init__(code, epoch)
+        self.current = current
+
+
+def normalize_image_tags(names: object) -> list[str]:
+    """Trim and validate each name, fold case-insensitive repeats, cap at 100.
+
+    The first occurrence keeps its spelling and position. Raises ``ValueError``
+    for anything but a list of valid vocabulary names or for more than
+    ``MAX_IMAGE_TAGS`` distinct names.
+    """
+    if not isinstance(names, list):
+        raise ValueError("tags must be a list of strings")
+    result: list[str] = []
+    seen: set[str] = set()
+    for value in names:
+        name = validate_vocabulary_tag_name(value)
+        key = name.casefold()
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(name)
+    if len(result) > MAX_IMAGE_TAGS:
+        raise ValueError(f"An image can carry at most {MAX_IMAGE_TAGS} tags")
+    return result
+
+
+def _validate_locator(locator: object) -> None:
+    if not isinstance(locator, str) or not 0 < len(locator) <= MAX_LOCATOR_LENGTH:
+        raise ValueError(
+            f"locator must be a nonblank path of at most {MAX_LOCATOR_LENGTH} chars"
+        )
+
+
+def _validate_digest(sha256: object) -> None:
+    if not isinstance(sha256, str) or not _DIGEST.fullmatch(sha256):
+        raise ValueError("sha256 must be 64 lowercase hexadecimal characters")
+
+
+def _validate_int(value: object, name: str, minimum: int) -> None:
+    if type(value) is not int or not minimum <= value <= _MAX_INT:
+        raise ValueError(f"{name} must be an integer not below {minimum}")
+
+
+def _validate_target(locator: object, sha256: object, size: object, mtime_ns: object):
+    _validate_locator(locator)
+    _validate_digest(sha256)
+    _validate_int(size, "size", 0)
+    _validate_int(mtime_ns, "mtime_ns", _MIN_INT)
+
+
+def _validate_fingerprint(locator: object, fingerprint: object) -> None:
+    _validate_locator(locator)
+    if (
+        not isinstance(fingerprint, tuple)
+        or len(fingerprint) != 2
+        or any(type(value) is not int for value in fingerprint)
+    ):
+        raise ValueError("stats must map locators to (size, mtime_ns) integer pairs")
+
+
+def _new_id() -> str:
+    """A canonical UUID; the schema's CHECK lets NULL through, this does not."""
+    value = str(uuid4())
+    if not _UUID.fullmatch(value):
+        raise ValueError("Annotation ids must be canonical UUIDs")
+    return value
+
+
+def _receipt(epoch: str, sha256: str, tags: list, revision: int, replayed: bool):
+    return {
+        "library_epoch": epoch,
+        "sha256": sha256,
+        "tags": list(tags),
+        "tag_revision": revision,
+        "replayed": replayed,
+    }
+
+
+class AnnotationOperations:
+    """PromptDatabase mixin: image tag sets, fingerprints and cheap reads.
+
+    ``_`` helpers take the open ``_pair_transaction`` connection.
+    """
+
+    @contextmanager
+    def _annotation_read(self) -> Iterator[sqlite3.Connection]:
+        """One deferred read transaction, so paired SELECTs see one snapshot."""
+        conn = self.model.get_connection()
+        if conn.in_transaction:
+            raise sqlite3.ProgrammingError(
+                "Annotation reads require their own snapshot"
+            )
+        with conn:
+            conn.execute("BEGIN")
+            yield conn
+
+    @staticmethod
+    def _annotation_owner(conn: sqlite3.Connection, locator: str, now: str) -> str:
+        row = conn.execute(_OWNER, (locator,)).fetchone()
+        if row is not None:
+            return row[0]
+        file_id = _new_id()
+        conn.execute(_INSERT_OWNER, (file_id, locator, now))
+        return file_id
+
+    @staticmethod
+    def _annotation_byte(
+        conn: sqlite3.Connection,
+        file_id: str,
+        sha256: str,
+        size: int,
+        mtime_ns: int,
+        now: str,
+    ) -> tuple[str, int, bool]:
+        """The byte record for these bytes, refreshed or created.
+
+        Returns ``(id, tag_revision, created)``.
+        """
+        row = conn.execute(_BYTE, (file_id, sha256)).fetchone()
+        if row is None:
+            byte_id = _new_id()
+            conn.execute(
+                _INSERT_BYTE, (byte_id, file_id, sha256, size, mtime_ns, now, now)
+            )
+            return byte_id, 0, True
+        conn.execute(_REFRESH_BYTE, (size, mtime_ns, now, row[0]))
+        return row[0], row[1], False
+
+    @staticmethod
+    def _byte_tags(conn: sqlite3.Connection, byte_id: str) -> list[str]:
+        return [row[0] for row in conn.execute(_BYTE_TAGS, (byte_id,))]
+
+    def _write_image_tags(
+        self, conn: sqlite3.Connection, byte_id: str, tags: list[str]
+    ) -> int:
+        """Rewrite the junction in order, bump both revisions; return the new one."""
+        tag_ids = self._ensure_tags(conn, tags)
+        conn.execute(_CLEAR_TAGS, (byte_id,))
+        conn.executemany(
+            _INSERT_TAG,
+            [(byte_id, tag_ids[name], position) for position, name in enumerate(tags)],
+        )
+        conn.execute(_BUMP_REVISION, (byte_id,))
+        conn.execute(_BUMP_INDEX)
+        return conn.execute(_REVISION, (byte_id,)).fetchone()[0]
+
+    def _current_set(self, conn, sha256, byte_id, revision) -> dict:
+        return {
+            "sha256": sha256,
+            "tags": self._byte_tags(conn, byte_id),
+            "tag_revision": revision,
+        }
+
+    def read_image_tags(self, locator: str, sha256: str) -> dict:
+        """The stored set of these exact bytes, or an empty set at revision 0.
+
+        ``prior_count`` is the number of other byte records at the locator
+        (earlier or later bytes of the same file).
+        """
+        _validate_locator(locator)
+        _validate_digest(sha256)
+        with self._annotation_read() as conn:
+            row = conn.execute(_READ, (sha256, sha256, locator)).fetchone()
+            if row is None:
+                return {"tags": [], "tag_revision": 0, "prior_count": 0}
+            tags = self._byte_tags(conn, row[0]) if row[0] is not None else []
+            return {"tags": tags, "tag_revision": row[1] or 0, "prior_count": row[2]}
+
+    def replace_image_tags(
+        self,
+        *,
+        locator: str,
+        sha256: str,
+        size: int,
+        mtime_ns: int,
+        expected_epoch: str,
+        expected_tag_revision: int,
+        tags: list,
+    ) -> dict:
+        """Replace the set of these exact bytes, guarded by epoch and revision.
+
+        Creates the owner and byte record when the bytes are new (an explicit
+        empty set is a real save). The identical set at the current revision,
+        or one behind it (a lost-response retry), is ``replayed`` and writes
+        nothing; any other revision mismatch is ``tag-conflict`` with the
+        current set and no writes.
+        """
+        _validate_target(locator, sha256, size, mtime_ns)
+        _validate_epoch(expected_epoch)
+        _validate_int(expected_tag_revision, "expected_tag_revision", 0)
+        tags = normalize_image_tags(tags)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            now = utc_now_iso()
+            file_id = self._annotation_owner(conn, locator, now)
+            byte_id, revision, created = self._annotation_byte(
+                conn, file_id, sha256, size, mtime_ns, now
+            )
+            current = self._byte_tags(conn, byte_id)
+            if (
+                not created
+                and current == tags
+                and expected_tag_revision in (revision, revision - 1)
+            ):
+                return _receipt(epoch, sha256, current, revision, True)
+            if expected_tag_revision != revision:
+                raise AnnotationCommandError(
+                    "tag-conflict",
+                    epoch,
+                    {"sha256": sha256, "tags": current, "tag_revision": revision},
+                )
+            revision = self._write_image_tags(conn, byte_id, tags)
+            return _receipt(epoch, sha256, tags, revision, False)
+
+    def add_image_tags(
+        self,
+        *,
+        locator: str,
+        sha256: str,
+        size: int,
+        mtime_ns: int,
+        expected_epoch: str,
+        tags: list,
+    ) -> dict:
+        """Union ``tags`` into the set of these exact bytes, keeping stored spelling.
+
+        ``added`` counts the names that were new (case-insensitively). A union
+        over ``MAX_IMAGE_TAGS`` is ``tag-limit`` with the current set and no
+        writes; a union that adds nothing to an existing record is ``replayed``.
+        """
+        _validate_target(locator, sha256, size, mtime_ns)
+        _validate_epoch(expected_epoch)
+        tags = normalize_image_tags(tags)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            now = utc_now_iso()
+            file_id = self._annotation_owner(conn, locator, now)
+            byte_id, revision, created = self._annotation_byte(
+                conn, file_id, sha256, size, mtime_ns, now
+            )
+            current = self._byte_tags(conn, byte_id)
+            present = {name.casefold() for name in current}
+            added = [name for name in tags if name.casefold() not in present]
+            merged = current + added
+            if len(merged) > MAX_IMAGE_TAGS:
+                raise AnnotationCommandError(
+                    "tag-limit",
+                    epoch,
+                    {"sha256": sha256, "tags": current, "tag_revision": revision},
+                )
+            if not created and not added:
+                return {**_receipt(epoch, sha256, current, revision, True), "added": 0}
+            revision = self._write_image_tags(conn, byte_id, merged)
+            return {
+                **_receipt(epoch, sha256, merged, revision, False),
+                "added": len(added),
+            }
+
+    def refresh_image_fingerprint(
+        self, locator: str, sha256: str, size: int, mtime_ns: int
+    ) -> bool:
+        """Record that these exact bytes were just seen with this stat.
+
+        True when a byte record was refreshed; nothing is created and no tag
+        or tag revision changes. The image index revision moves because the
+        record's currency for cheap reads may have changed.
+        """
+        _validate_target(locator, sha256, size, mtime_ns)
+        with self.model.get_connection() as conn:
+            cursor = conn.execute(
+                _REFRESH_BY_LOCATOR, (size, mtime_ns, utc_now_iso(), sha256, locator)
+            )
+            if cursor.rowcount > 0:
+                conn.execute(_BUMP_INDEX)
+            return cursor.rowcount > 0
+
+    def current_image_tags(self, stats: dict) -> dict:
+        """Image tags of the files whose current ``(size, mtime_ns)`` is verified.
+
+        ``stats`` maps locators to their present fingerprint. A locator is
+        answered (possibly with an empty set) only when a byte record carries
+        exactly that verified fingerprint; the most recently verified record
+        wins when several do. Lookups go in chunks of 500 locators.
+        """
+        if not isinstance(stats, dict):
+            raise ValueError(
+                "stats must map locators to (size, mtime_ns) integer pairs"
+            )
+        for locator, fingerprint in stats.items():
+            _validate_fingerprint(locator, fingerprint)
+        locators = list(stats)
+        chosen: dict[str, tuple[tuple[str, str], str]] = {}
+        result: dict[str, list[str]] = {}
+        with self._annotation_read() as conn:
+            for start in range(0, len(locators), _LOOKUP_CHUNK):
+                chunk = locators[start : start + _LOOKUP_CHUNK]
+                marks = ",".join("?" * len(chunk))
+                for locator, byte_id, size, mtime_ns, verified_at in conn.execute(
+                    _FINGERPRINTS.format(marks=marks), chunk
+                ):
+                    if (size, mtime_ns) != stats[locator]:
+                        continue
+                    key = (verified_at, byte_id)
+                    if locator not in chosen or key > chosen[locator][0]:
+                        chosen[locator] = (key, byte_id)
+            owners = {byte_id: locator for locator, (_key, byte_id) in chosen.items()}
+            result = {locator: [] for locator in chosen}
+            byte_ids = list(owners)
+            for start in range(0, len(byte_ids), _LOOKUP_CHUNK):
+                chunk = byte_ids[start : start + _LOOKUP_CHUNK]
+                marks = ",".join("?" * len(chunk))
+                for byte_id, name in conn.execute(_TAGS_OF.format(marks=marks), chunk):
+                    result[owners[byte_id]].append(name)
+        return result
+
+    def iter_image_tag_records(self) -> Iterator[tuple[str, int, int, list[str]]]:
+        """Stream every byte record as ``(locator, size, mtime_ns, tags)``.
+
+        One SELECT ordered by locator, then by the bytes' first sighting, with
+        each set's names in position order.
+        """
+        conn = self.model.get_connection()
+        current: tuple | None = None
+        for locator, byte_id, size, mtime_ns, name in conn.execute(_RECORDS):
+            if current is None or current[0] != byte_id:
+                if current is not None:
+                    yield current[1]
+                current = (byte_id, (locator, size, mtime_ns, []))
+            if name is not None:
+                current[1][3].append(name)
+        if current is not None:
+            yield current[1]

database/operations.py

@@ -12,6 +12,7 @@ from .models import PromptModel, normalize_image_path, utc_now_iso
 from .pairs import PairDependencyError, PairOperations
 from .library import LibraryOperations, titled
 from .image_favorites import ImageFavoriteOperations
+from .annotations import AnnotationOperations
 from .saved_views import SavedViewOperations
 from .library_commands import LibraryCommandOperations
 from .vocabulary import VocabularyOperations
@@ -322,6 +323,7 @@ class PromptDatabase(
     ExecutionOutputOperations,
     HistoryReadOperations,
     ImageFavoriteOperations,
+    AnnotationOperations,
     SavedViewOperations,
 ):
     """Database operations class for managing prompts."""
@@ -1087,13 +1089,13 @@ class PromptDatabase(
     def _move_tag_links(
         conn: sqlite3.Connection, source_id: int, target_id: int
     ) -> None:
-        """Repoint component and pair links from source to target.
+        """Repoint component, pair and image links from source to target.
 
         Owners that already carry the target keep that one link; the source's
-        leftover duplicates are removed. The source registry row is left for the
-        caller to delete.
+        leftover duplicates are removed (an image set keeps the target's
+        position). The source registry row is left for the caller to delete.
         """
-        for junction in ("prompt_tags", "prompt_pair_tags"):
+        for junction in ("prompt_tags", "prompt_pair_tags", "media_image_tags"):
             conn.execute(
                 f"UPDATE OR IGNORE {junction} SET tag_id = ? WHERE tag_id = ?",
                 (target_id, source_id),
@@ -1239,7 +1241,10 @@ class PromptDatabase(
                 pair_count = conn.execute(
                     "SELECT COUNT(*) FROM prompt_pair_tags WHERE tag_id = ?", (src_id,)
                 ).fetchone()[0]
-                if src_count or pair_count:
+                image_count = conn.execute(
+                    "SELECT COUNT(*) FROM media_image_tags WHERE tag_id = ?", (src_id,)
+                ).fetchone()[0]
+                if src_count or pair_count or image_count:
                     self._move_tag_links(conn, src_id, target_id)
                     tags_merged += 1
                     affected += src_count

database/vocabulary.py

@@ -3,6 +3,9 @@
 A count is the number of eligible Library parents that carry the value
 directly or through one active saved pair, which is exactly what
 ``list_library(tags=[name])`` / ``list_library(category=value)`` return.
+A tag row's ``image_count`` is the number of files (private locators) with any
+byte record carrying the tag (#249), so a file counts once however many of its
+byte versions carry it, and equal bytes at two paths count twice.
 
 Tag commands run in one ``_pair_transaction``: the storage epoch is checked
 first, then the exact ``{id, name}`` identity, and the receipt is read on the
@@ -26,6 +29,11 @@ except ImportError:
     from utils.validators import validate_category, validate_tags
 
 _CATEGORY_DOMAIN = "prompt-manager-vocabulary-category:v1\n"
+# Files carrying a tag through any of their byte records (#249 image tags).
+_IMAGE_FILES = (
+    "SELECT mit.tag_id,COUNT(DISTINCT b.file_id) FROM media_image_tags mit "
+    "JOIN media_annotation_bytes b ON b.id=mit.byte_id"
+)
 
 
 def _tag_rows(page: str) -> str:
@@ -40,9 +48,13 @@ def _tag_rows(page: str) -> str:
         "WHERE pt.tag_id IN (SELECT id FROM page) AND " + _ELIGIBLE + " "
         "UNION SELECT ppt.tag_id,pp.positive_prompt_id FROM prompt_pairs pp "
         "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id "
-        "WHERE pp.archived=0 AND ppt.tag_id IN (SELECT id FROM page)) "
-        "SELECT page.id,page.name,COUNT(DISTINCT owners.pid) AS prompt_count "
+        "WHERE pp.archived=0 AND ppt.tag_id IN (SELECT id FROM page)),"
+        "images(tag_id,n) AS (" + _IMAGE_FILES + " "
+        "WHERE mit.tag_id IN (SELECT id FROM page) GROUP BY mit.tag_id) "
+        "SELECT page.id,page.name,COUNT(DISTINCT owners.pid) AS prompt_count,"
+        "COALESCE(MAX(images.n),0) AS image_count "
         "FROM page LEFT JOIN owners ON owners.tag_id=page.id "
+        "LEFT JOIN images ON images.tag_id=page.id "
         "GROUP BY page.id ORDER BY page.id"
     )
 
@@ -61,7 +73,10 @@ _TAG_IMPACT = (
     "AS negative_variants,"
     "(SELECT COUNT(*) FROM prompt_pairs pp JOIN prompt_pair_tags ppt "
     "ON ppt.pair_id=pp.id WHERE ppt.tag_id=:tag AND pp.archived!=0) "
-    "AS archived_negative_variants"
+    "AS archived_negative_variants,"
+    "(SELECT COUNT(DISTINCT b.file_id) FROM media_image_tags mit "
+    "JOIN media_annotation_bytes b ON b.id=mit.byte_id WHERE mit.tag_id=:tag) "
+    "AS images"
 )
 _MAX_ID = 2**63 - 1
 
@@ -205,7 +220,16 @@ class VocabularyOperations:
             "SELECT COUNT(DISTINCT pair_id) FROM prompt_pair_tags WHERE tag_id=?",
             (tag_id,),
         ).fetchone()[0]
-        return {"components_changed": components, "pairs_changed": pairs}
+        images = conn.execute(
+            "SELECT COUNT(DISTINCT b.file_id) FROM media_image_tags mit "
+            "JOIN media_annotation_bytes b ON b.id=mit.byte_id WHERE mit.tag_id=?",
+            (tag_id,),
+        ).fetchone()[0]
+        return {
+            "components_changed": components,
+            "pairs_changed": pairs,
+            "images_changed": images,
+        }
 
     def create_vocabulary_tag(self, name: str, *, expected_epoch: str) -> dict:
         """Insert an unused registry row, or reuse the exact existing name."""

tests/browser/server.py

@@ -189,6 +189,9 @@ def create_app(root: Path) -> web.Application:
             for table in (
                 "saved_views",
                 "image_favorites",
+                "media_image_tags",
+                "media_annotation_bytes",
+                "media_annotation_files",
                 "image_prompt_pairs",
                 "prompt_pair_tags",
                 "prompt_pairs",

tests/browser/workspace_concept_fixture.py

@@ -170,6 +170,9 @@ _TABLES = (
     "execution_context_texts",
     "saved_views",
     "image_favorites",
+    "media_image_tags",
+    "media_annotation_bytes",
+    "media_annotation_files",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_library_editor_fixture.py

@@ -28,6 +28,9 @@ EDITOR_METADATA = {
 _TABLES = (
     "saved_views",
     "image_favorites",
+    "media_image_tags",
+    "media_annotation_bytes",
+    "media_annotation_files",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_visual_fixture.py

@@ -114,6 +114,9 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None:
         for table in (
             "saved_views",
             "image_favorites",
+            "media_image_tags",
+            "media_annotation_bytes",
+            "media_annotation_files",
             "image_prompt_pairs",
             "prompt_pair_tags",
             "prompt_pairs",

tests/browser/workspace_vocabulary_fixture.py

@@ -81,6 +81,9 @@ MAX_EXTRA_CATEGORIES = 200
 _TABLES = (
     "saved_views",
     "image_favorites",
+    "media_image_tags",
+    "media_annotation_bytes",
+    "media_annotation_files",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/js/workspace-vocabulary-state.test.js

@@ -507,10 +507,49 @@ test('category preview text states the real impact and whether it renames or con
         '“Landscapes” is used by 2 other saved prompts. Apply renames it to “Scenery”.');
 });
 
+test('accepts image_count and names images in the delete text', async () => {
+    const {createVocabularyStream, createVocabularyCommand, readTagImpact, deleteTagText} = await helpers;
+    const rows = [{...tag(1, 'mist', 2), image_count: 3}, tag(2, 'dusk', 0)];
+    const stream = createVocabularyStream({api: queued([tagPage(rows)]).api, kind: 'tags'});
+    assert.equal(await stream.loadMore(), 2);
+    assert.deepEqual(stream.state.items.map(row => row.image_count), [3, undefined]);
+    for (const bad of [{...tag(1), image_count: -1}, {...tag(1), image_count: '3'}, {...tag(1), image_count: 1.5}]) {
+        const rejected = createVocabularyStream({api: queued([tagPage([bad])]).api, kind: 'tags'});
+        await rejected.loadMore();
+        assert.ok(rejected.state.error instanceof TypeError, JSON.stringify(bad));
+        assert.deepEqual(rejected.state.items, []);
+    }
+    const withImages = {...impact, images: 2};
+    const api = {get: async () => ({success: true, library_epoch: EPOCH, impact: withImages})};
+    assert.deepEqual((await readTagImpact(api, {id: 4, name: 'mist'})).impact, withImages);
+    await assert.rejects(readTagImpact({get: async () => ({success: true, library_epoch: EPOCH, impact: {...impact, images: -2}})},
+        {id: 4, name: 'mist'}), TypeError);
+    const command = createVocabularyCommand({api: {
+        patch: async () => renameReceipt({tag: {...tag(1, 'dusk', 2), image_count: 1}, images_changed: 1}),
+        post: async () => ({success: true, library_epoch: EPOCH, deleted: true, components_changed: 0, pairs_changed: 0, images_changed: 2}),
+    }});
+    await command.renameTag({id: 1, name: 'dawn', prompt_count: 2}, 'dusk', EPOCH);
+    assert.equal(command.state.phase, 'done');
+    assert.equal(command.state.receipt.images_changed, 1);
+    command.reset();
+    await command.deleteTag({id: 1, name: 'dusk'}, EPOCH);
+    assert.equal(command.state.phase, 'done');
+    const malformed = createVocabularyCommand({api: {post: async () => ({success: true, library_epoch: EPOCH, deleted: true,
+        components_changed: 0, pairs_changed: 0, images_changed: 'two'})}});
+    await malformed.deleteTag({id: 1, name: 'dusk'}, EPOCH);
+    assert.equal(malformed.state.phase, 'unknown-outcome');
+    assert.equal(deleteTagText('mist', withImages),
+        'Deleting “mist” removes the name and its links from 3 positive prompts, 3 negative variants (1 archived), 1 other saved prompt and 2 images. The prompts, negative variants, images and history themselves are kept.');
+    assert.equal(deleteTagText('mist', {...zero, images: 1}),
+        'Deleting “mist” removes the name and its links from 1 image. The prompts, negative variants, images and history themselves are kept.');
+    assert.equal(deleteTagText('unused', {...zero, images: 0}),
+        '“unused” is not linked to any saved prompt or image. Deleting it removes only the name.');
+});
+
 test('delete text names the tag, its real links and what stays unchanged', async () => {
     const {deleteTagText} = await helpers;
     assert.equal(deleteTagText('mist', impact),
         'Deleting “mist” removes the name and its links from 3 positive prompts, 3 negative variants (1 archived) and 1 other saved prompt. The prompts, negative variants, images and history themselves are kept.');
     assert.equal(deleteTagText('unused', zero),
-        '“unused” is not linked to any saved prompt. Deleting it removes only the name.');
+        '“unused” is not linked to any saved prompt or image. Deleting it removes only the name.');
 });

tests/test_annotations.py

@@ -0,0 +1,575 @@
+"""Per-image tag sets (#249 A2): byte-exact storage in the shared vocabulary."""
+
+import hashlib
+import os
+import re
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+try:
+    from tests.open_handles import assert_closed
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed
+
+from database.annotation_schema import annotation_schema_error
+from database.library_commands import LibraryCommandError
+from database.operations import PromptDatabase
+from utils.hashing import generate_prompt_hash
+
+try:  # guarded so a missing module fails assertions, not test loading
+    from database import annotations as annotations_module
+    from database.annotations import (
+        MAX_IMAGE_TAGS,
+        AnnotationCommandError,
+        normalize_image_tags,
+    )
+except ImportError:
+    annotations_module = None
+    MAX_IMAGE_TAGS = None
+    AnnotationCommandError = None
+    normalize_image_tags = None
+
+UUID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}")
+DIGEST_A = hashlib.sha256(b"bytes a").hexdigest()
+DIGEST_B = hashlib.sha256(b"bytes b").hexdigest()
+SIZE = 7
+MTIME = 1_700_000_000_123_456_789
+ANNOTATION_TABLES = (
+    "media_annotation_files",
+    "media_annotation_bytes",
+    "media_image_tags",
+    "media_annotation_metadata",
+)
+LEGACY_TABLES = (
+    "tags",
+    "prompt_tags",
+    "prompt_pair_tags",
+    "generated_images",
+    "image_prompt_pairs",
+)
+
+
+class AnnotationTestCase(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(
+            annotations_module, "database.annotations is not implemented"
+        )
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.addCleanup(assert_closed, self, directory.name)
+        self.root = directory.name
+        self.db = PromptDatabase(os.path.join(self.root, "annotations.db"))
+        self.addCleanup(self.db.close_all)
+        self.epoch = self.db.get_execution_epoch()
+
+    def locator(self, name: str) -> str:
+        return os.path.normcase(os.path.join(os.path.realpath(self.root), name))
+
+    def replace(self, loc, digest=DIGEST_A, tags=(), *, revision=0, **over):
+        options = {
+            "locator": loc,
+            "sha256": digest,
+            "size": SIZE,
+            "mtime_ns": MTIME,
+            "expected_epoch": self.epoch,
+            "expected_tag_revision": revision,
+            "tags": list(tags) if isinstance(tags, tuple) else tags,
+        }
+        options.update(over)
+        return self.db.replace_image_tags(**options)
+
+    def add(self, loc, digest=DIGEST_A, tags=(), **over):
+        options = {
+            "locator": loc,
+            "sha256": digest,
+            "size": SIZE,
+            "mtime_ns": MTIME,
+            "expected_epoch": self.epoch,
+            "tags": list(tags) if isinstance(tags, tuple) else tags,
+        }
+        options.update(over)
+        return self.db.add_image_tags(**options)
+
+    def rows(self, *tables):
+        """Every row of each table; ``verified_at`` is a fingerprint, not content."""
+        conn = self.db.model.get_connection()
+        result = []
+        for table in tables:
+            columns = [
+                row[1]
+                for row in conn.execute(f"PRAGMA table_info({table})")
+                if row[1] != "verified_at"
+            ]
+            result.append(
+                [
+                    tuple(row)
+                    for row in conn.execute(
+                        f"SELECT {','.join(columns)} FROM {table} ORDER BY 1,2"
+                    )
+                ]
+            )
+        return result
+
+    def counts(self):
+        conn = self.db.model.get_connection()
+        return {
+            table: conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+            for table in ANNOTATION_TABLES
+        }
+
+    def index_revision(self):
+        return (
+            self.db.model.get_connection()
+            .execute("SELECT image_index_revision FROM media_annotation_metadata")
+            .fetchone()[0]
+        )
+
+    def schema_clean(self):
+        self.assertIsNone(
+            annotation_schema_error(self.db.model.get_connection(), allow_missing=False)
+        )
+
+
+class NormalizeImageTagsTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(
+            normalize_image_tags, "database.annotations is not implemented"
+        )
+
+    def test_case_insensitive_dedupe_keeps_first_spelling_and_position(self):
+        self.assertEqual(
+            normalize_image_tags([" Forest ", "mist", "forest", "MIST", "Rain"]),
+            ["Forest", "mist", "Rain"],
+        )
+        self.assertEqual(normalize_image_tags([]), [])
+        self.assertEqual(MAX_IMAGE_TAGS, 100)
+
+    def test_limit_and_invalid_names_rejected(self):
+        for bad in (
+            None,
+            "forest",
+            {"a": 1},
+            [1],
+            [None],
+            ["a,b"],
+            [""],
+            ["   "],
+            ["x" * 101],
+            [f"t{i}" for i in range(MAX_IMAGE_TAGS + 1)],
+        ):
+            with self.subTest(bad=bad), self.assertRaises(ValueError):
+                normalize_image_tags(bad)
+        # Duplicates fold before the cap is applied.
+        many = [f"t{i}" for i in range(MAX_IMAGE_TAGS)]
+        self.assertEqual(normalize_image_tags(many + ["T0", "T99"]), many)
+
+
+class ReplaceImageTagsTests(AnnotationTestCase):
+    def test_first_save_creates_owner_and_byte_atomically_including_empty_set(self):
+        loc = self.locator("a.png")
+        receipt = self.replace(loc, tags=["Forest", "mist", "forest"])
+        self.assertEqual(receipt["tags"], ["Forest", "mist"])
+        self.assertEqual(receipt["tag_revision"], 1)
+        self.assertEqual(
+            receipt,
+            {
+                "library_epoch": self.epoch,
+                "sha256": DIGEST_A,
+                "tags": ["Forest", "mist"],
+                "tag_revision": 1,
+                "replayed": False,
+            },
+        )
+        conn = self.db.model.get_connection()
+        files = conn.execute("SELECT * FROM media_annotation_files").fetchall()
+        self.assertEqual(len(files), 1)
+        self.assertRegex(files[0]["id"], UUID)
+        self.assertEqual(files[0]["locator"], loc)
+        byte = conn.execute("SELECT * FROM media_annotation_bytes").fetchone()
+        self.assertRegex(byte["id"], UUID)
+        self.assertEqual(
+            (byte["file_id"], byte["sha256"], byte["size"], byte["mtime_ns"]),
+            (files[0]["id"], DIGEST_A, SIZE, MTIME),
+        )
+        self.assertEqual(byte["tag_revision"], 1)
+        self.assertEqual(byte["verified_at"], byte["first_seen_at"])
+        self.assertEqual(
+            [
+                tuple(row)
+                for row in conn.execute(
+                    "SELECT t.name, mit.position FROM media_image_tags mit "
+                    "JOIN tags t ON t.id = mit.tag_id ORDER BY mit.position"
+                )
+            ],
+            [("Forest", 0), ("mist", 1)],
+        )
+        self.assertEqual(self.index_revision(), 1)
+        # An explicit empty set is a real save: owner and byte record exist.
+        empty = self.replace(self.locator("b.png"), DIGEST_B, [])
+        self.assertEqual((empty["tags"], empty["tag_revision"]), ([], 1))
+        self.assertEqual(self.counts()["media_annotation_files"], 2)
+        self.assertEqual(self.counts()["media_annotation_bytes"], 2)
+        self.assertEqual(
+            self.db.read_image_tags(self.locator("b.png"), DIGEST_B),
+            {"tags": [], "tag_revision": 1, "prior_count": 0},
+        )
+        self.assertEqual(self.index_revision(), 2)
+        self.schema_clean()
+
+    def test_save_changes_only_target_byte_set(self):
+        a, b, c = self.locator("a.png"), self.locator("b.png"), self.locator("c.png")
+        prompt = self.db.save_prompt(
+            "Shared forest",
+            tags=["shared"],
+            prompt_hash=generate_prompt_hash("Shared forest"),
+        )
+        pair = self.db.save_prompt_pair("Shared forest", "blurry", tags=["pairtag"])
+        image_a = self.db.link_image_to_prompt(prompt, a)
+        image_b = self.db.link_image_to_prompt(prompt, b)
+        self.db.link_image_to_pair(pair, image_a, {"seed": 1})
+        self.db.link_image_to_pair(pair, image_b, {"seed": 2})
+        self.replace(b, DIGEST_A, ["sibling"])
+        self.replace(c, DIGEST_A, ["copy"])  # equal bytes at another path
+        before = self.rows(*LEGACY_TABLES)
+        sibling_before = self.db.read_image_tags(b, DIGEST_A)
+        copy_before = self.db.read_image_tags(c, DIGEST_A)
+
+        receipt = self.replace(a, DIGEST_A, ["only a"])
+
+        self.assertEqual((receipt["tags"], receipt["tag_revision"]), (["only a"], 1))
+        after = self.rows(*LEGACY_TABLES)
+        # The registry gained the new name; nothing else moved.
+        self.assertEqual(after[1:], before[1:])
+        self.assertEqual([row for row in after[0] if row[1] != "only a"], before[0])
+        self.assertEqual(self.db.read_image_tags(b, DIGEST_A), sibling_before)
+        self.assertEqual(self.db.read_image_tags(c, DIGEST_A), copy_before)
+        self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["shared"])
+        self.assertEqual(self.db.get_prompt_pair(pair)["tags"], ["pairtag"])
+
+    def test_changed_bytes_at_same_locator_start_empty_and_keep_prior(self):
+        loc = self.locator("a.png")
+        self.replace(loc, DIGEST_A, ["old"])
+        self.assertEqual(
+            self.db.read_image_tags(loc, DIGEST_B),
+            {"tags": [], "tag_revision": 0, "prior_count": 1},
+        )
+        receipt = self.replace(loc, DIGEST_B, ["new"], mtime_ns=MTIME + 5)
+        self.assertEqual((receipt["tags"], receipt["tag_revision"]), (["new"], 1))
+        self.assertEqual(
+            self.db.read_image_tags(loc, DIGEST_A),
+            {"tags": ["old"], "tag_revision": 1, "prior_count": 1},
+        )
+        self.assertEqual(
+            self.db.read_image_tags(loc, DIGEST_B),
+            {"tags": ["new"], "tag_revision": 1, "prior_count": 1},
+        )
+        self.assertEqual(self.counts()["media_annotation_files"], 1)
+        self.assertEqual(self.counts()["media_annotation_bytes"], 2)
+        self.assertEqual(
+            self.db.read_image_tags(self.locator("unseen.png"), DIGEST_A),
+            {"tags": [], "tag_revision": 0, "prior_count": 0},
+        )
+
+    def test_revision_conflict_returns_current_set(self):
+        loc = self.locator("a.png")
+        self.replace(loc, tags=["a"])
+        before = self.rows(*ANNOTATION_TABLES, "tags")
+        with self.assertRaises(AnnotationCommandError) as ctx:
+            self.replace(loc, tags=["b"], revision=0)
+        self.assertEqual(
+            (ctx.exception.code, ctx.exception.current["tag_revision"]),
+            ("tag-conflict", 1),
+        )
+        self.assertIsInstance(ctx.exception, LibraryCommandError)
+        self.assertEqual(ctx.exception.epoch, self.epoch)
+        self.assertEqual(ctx.exception.current["tags"], ["a"])
+        self.assertEqual(ctx.exception.current["sha256"], DIGEST_A)
+        with self.assertRaises(AnnotationCommandError) as ahead:
+            self.replace(loc, tags=["a"], revision=5)
+        self.assertEqual(ahead.exception.code, "tag-conflict")
+        self.assertEqual(self.rows(*ANNOTATION_TABLES, "tags"), before)
+        self.assertFalse(self.db.model.get_connection().in_transaction)
+
+    def test_identical_replay_after_commit_is_success_replayed(self):
+        loc = self.locator("a.png")
+        first = self.replace(loc, tags=["a", "b"])
+        self.assertEqual((first["tag_revision"], first["replayed"]), (1, False))
+        before = self.rows(*ANNOTATION_TABLES)
+        # The lost-response retry: same set, the revision it was sent against.
+        retry = self.replace(loc, tags=["a", "b"], revision=0)
+        self.assertEqual(retry, {**first, "replayed": True})
+        # Saving the current set at the current revision writes nothing either.
+        again = self.replace(loc, tags=["a", "b"], revision=1)
+        self.assertEqual(again, {**first, "replayed": True})
+        self.assertEqual(self.rows(*ANNOTATION_TABLES), before)
+        # Order is part of the set.
+        reordered = self.replace(loc, tags=["b", "a"], revision=1)
+        self.assertEqual(
+            (reordered["tags"], reordered["tag_revision"], reordered["replayed"]),
+            (["b", "a"], 2, False),
+        )
+        self.assertEqual(self.index_revision(), 2)
+
+    def test_epoch_checked_inside_transaction(self):
+        loc = self.locator("a.png")
+        seen = []
+        original = PromptDatabase._library_epoch
+
+        def spy(db, conn, expected_epoch):
+            seen.append(conn.in_transaction)
+            return original(db, conn, expected_epoch)
+
+        with patch.object(PromptDatabase, "_library_epoch", spy):
+            self.replace(loc, tags=["a"])
+            self.add(loc, tags=["b"])
+        self.assertEqual(seen, [True, True])
+        stale = "11111111-2222-4333-8444-555555555555"
+        for call in (
+            lambda: self.replace(
+                self.locator("x.png"), tags=["a"], expected_epoch=stale
+            ),
+            lambda: self.add(self.locator("x.png"), tags=["a"], expected_epoch=stale),
+        ):
+            with self.assertRaises(LibraryCommandError) as ctx:
+                call()
+            self.assertEqual(ctx.exception.code, "storage-changed")
+            self.assertEqual(ctx.exception.epoch, self.epoch)
+        self.assertEqual(self.counts()["media_annotation_files"], 1)
+        for bad_epoch in (None, 7, "not-a-uuid"):
+            with self.assertRaises(ValueError):
+                self.replace(loc, tags=["a"], expected_epoch=bad_epoch)
+
+    def test_limit_and_invalid_names_rejected_without_partial_write(self):
+        loc = self.locator("a.png")
+        before = self.rows(*ANNOTATION_TABLES, "tags")
+        for tags in (
+            ["ok", "a,b"],
+            ["ok", ""],
+            ["ok", 3],
+            "ok",
+            [f"t{i}" for i in range(101)],
+        ):
+            with self.subTest(tags=tags), self.assertRaises(ValueError):
+                self.replace(loc, tags=tags)
+            with self.subTest(tags=tags, command="add"), self.assertRaises(ValueError):
+                self.add(loc, tags=tags)
+        for field, value in (
+            ("locator", ""),
+            ("locator", None),
+            ("locator", "x" * 4097),
+            ("sha256", DIGEST_A.upper()),
+            ("sha256", DIGEST_A[:-1]),
+            ("sha256", None),
+            ("size", -1),
+            ("size", True),
+            ("size", "7"),
+            ("mtime_ns", 1.5),
+            ("mtime_ns", None),
+            ("expected_tag_revision", -1),
+            ("expected_tag_revision", False),
+        ):
+            with self.subTest(field=field, value=value), self.assertRaises(ValueError):
+                self.replace(loc, tags=["ok"], **{field: value})
+        self.assertEqual(self.rows(*ANNOTATION_TABLES, "tags"), before)
+        self.assertEqual(self.index_revision(), 0)
+
+    def test_writers_supply_uuid_ids_and_refuse_to_write_without_one(self):
+        loc = self.locator("a.png")
+        before = self.rows(*ANNOTATION_TABLES)
+        with patch.object(annotations_module, "uuid4", return_value=""):
+            with self.assertRaises(ValueError):
+                self.replace(loc, tags=["a"])
+        self.assertEqual(self.rows(*ANNOTATION_TABLES), before)
+        self.replace(loc, tags=["a"])
+        self.add(self.locator("b.png"), DIGEST_B, ["b"])
+        conn = self.db.model.get_connection()
+        for table in ("media_annotation_files", "media_annotation_bytes"):
+            for (value,) in conn.execute(f"SELECT id FROM {table}"):
+                self.assertRegex(value, UUID)
+        self.schema_clean()
+
+    def test_never_resolves_owner_through_prompt_lookup(self):
+        loc = self.locator("a.png")
+        prompt = self.db.save_prompt("Linked", tags=["prompt-tag"])
+        self.db.link_image_to_prompt(prompt, loc)
+        with patch.object(
+            PromptDatabase,
+            "get_prompt_id_for_image",
+            side_effect=AssertionError("prompt lookup is not an owner"),
+        ):
+            self.replace(loc, tags=["img"])
+            self.add(loc, tags=["more"])
+            self.assertEqual(
+                self.db.read_image_tags(loc, DIGEST_A)["tags"], ["img", "more"]
+            )
+            self.assertEqual(
+                self.db.current_image_tags({loc: (SIZE, MTIME)}),
+                {loc: ["img", "more"]},
+            )
+            self.assertTrue(
+                self.db.refresh_image_fingerprint(loc, DIGEST_A, SIZE, MTIME + 1)
+            )
+            list(self.db.iter_image_tag_records())
+        self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["prompt-tag"])
+
+
+class AddImageTagsTests(AnnotationTestCase):
+    def test_add_merges_without_removing_and_caps_at_100(self):
+        loc = self.locator("a.png")
+        self.replace(loc, tags=["Forest", "mist"])
+        receipt = self.add(loc, tags=["forest", "Rain", "rain"])
+        self.assertEqual(
+            receipt,
+            {
+                "library_epoch": self.epoch,
+                "sha256": DIGEST_A,
+                "tags": ["Forest", "mist", "Rain"],
+                "tag_revision": 2,
+                "replayed": False,
+                "added": 1,
+            },
+        )
+        nothing = self.add(loc, tags=["MIST"])
+        self.assertEqual(
+            (nothing["tags"], nothing["tag_revision"], nothing["replayed"]),
+            (["Forest", "mist", "Rain"], 2, True),
+        )
+        self.assertEqual(nothing["added"], 0)
+        before = self.rows(*ANNOTATION_TABLES, "tags")
+        with self.assertRaises(AnnotationCommandError) as ctx:
+            self.add(loc, tags=[f"t{i}" for i in range(98)])
+        self.assertEqual(ctx.exception.code, "tag-limit")
+        self.assertEqual(ctx.exception.current["tag_revision"], 2)
+        self.assertEqual(self.rows(*ANNOTATION_TABLES, "tags"), before)
+        exact = self.add(loc, tags=[f"t{i}" for i in range(97)])
+        self.assertEqual((len(exact["tags"]), exact["added"]), (100, 97))
+        # An unseen file: add creates the owner and byte record like a save.
+        fresh = self.add(self.locator("b.png"), DIGEST_B, ["one"])
+        self.assertEqual(
+            (fresh["tags"], fresh["tag_revision"], fresh["added"]), (["one"], 1, 1)
+        )
+        self.schema_clean()
+
+
+class CurrentImageTagsTests(AnnotationTestCase):
+    def test_current_tags_require_matching_fingerprint(self):
+        loc = self.locator("a.png")
+        other = self.locator("b.png")
+        self.replace(loc, tags=["a"])
+        self.replace(other, DIGEST_B, [], mtime_ns=MTIME + 9)
+        self.assertEqual(
+            self.db.current_image_tags({loc: (SIZE, MTIME), other: (SIZE, MTIME + 9)}),
+            {loc: ["a"], other: []},
+        )
+        self.assertEqual(self.db.current_image_tags({loc: (SIZE, MTIME + 1)}), {})
+        self.assertEqual(self.db.current_image_tags({loc: (SIZE + 1, MTIME)}), {})
+        self.assertEqual(self.db.current_image_tags({}), {})
+        self.assertEqual(
+            self.db.current_image_tags({self.locator("none.png"): (SIZE, MTIME)}), {}
+        )
+        # Refreshing the verified fingerprint makes the record current again.
+        self.assertTrue(
+            self.db.refresh_image_fingerprint(loc, DIGEST_A, SIZE, MTIME + 1)
+        )
+        self.assertFalse(self.db.refresh_image_fingerprint(loc, DIGEST_B, SIZE, MTIME))
+        self.assertFalse(
+            self.db.refresh_image_fingerprint(
+                self.locator("x.png"), DIGEST_A, SIZE, MTIME
+            )
+        )
+        self.assertEqual(
+            self.db.current_image_tags({loc: (SIZE, MTIME + 1)}), {loc: ["a"]}
+        )
+        self.assertEqual(self.db.current_image_tags({loc: (SIZE, MTIME)}), {})
+        self.assertEqual(
+            self.db.read_image_tags(loc, DIGEST_A),
+            {"tags": ["a"], "tag_revision": 1, "prior_count": 0},
+        )
+        for bad in (None, [], {loc: (SIZE,)}, {loc: (SIZE, 1.5)}, {3: (SIZE, MTIME)}):
+            with self.subTest(bad=bad), self.assertRaises(ValueError):
+                self.db.current_image_tags(bad)
+
+    def test_newest_verification_wins_on_equal_fingerprints(self):
+        loc = self.locator("a.png")
+        stamps = iter(f"2026-10-22T00:00:0{i}.000000+00:00" for i in range(10))
+        with patch.object(annotations_module, "utc_now_iso", side_effect=stamps):
+            self.replace(loc, DIGEST_A, ["old bytes"])
+            self.replace(loc, DIGEST_B, ["new bytes"])  # same size and mtime
+        self.assertEqual(
+            self.db.current_image_tags({loc: (SIZE, MTIME)}), {loc: ["new bytes"]}
+        )
+        with patch.object(
+            annotations_module,
+            "utc_now_iso",
+            return_value="2026-10-22T00:01:00.000000+00:00",
+        ):
+            self.assertTrue(
+                self.db.refresh_image_fingerprint(loc, DIGEST_A, SIZE, MTIME)
+            )
+        self.assertEqual(
+            self.db.current_image_tags({loc: (SIZE, MTIME)}), {loc: ["old bytes"]}
+        )
+
+    def test_lookup_is_chunked_and_records_stream_in_position_order(self):
+        locators = [self.locator(f"{i:04}.png") for i in range(503)]
+        self.replace(locators[0], tags=["z", "y"])
+        self.replace(locators[502], DIGEST_B, ["x"], mtime_ns=MTIME + 2)
+        self.replace(locators[250], DIGEST_B, [], mtime_ns=MTIME + 1)
+        statements = []
+        conn = self.db.model.get_connection()
+        conn.set_trace_callback(statements.append)
+        try:
+            current = self.db.current_image_tags(
+                {
+                    loc: (SIZE, MTIME + (2 if i == 502 else 0))
+                    for i, loc in enumerate(locators)
+                }
+            )
+        finally:
+            conn.set_trace_callback(None)
+        self.assertEqual(current, {locators[0]: ["z", "y"], locators[502]: ["x"]})
+        lookups = [s for s in statements if "media_annotation_files" in s]
+        self.assertEqual(len(lookups), 2)
+        self.assertEqual(
+            list(self.db.iter_image_tag_records()),
+            [
+                (locators[0], SIZE, MTIME, ["z", "y"]),
+                (locators[250], SIZE, MTIME + 1, []),
+                (locators[502], SIZE, MTIME + 2, ["x"]),
+            ],
+        )
+
+
+class AnnotationIsolationTests(AnnotationTestCase):
+    def test_cascade_from_vocabulary_delete_keeps_byte_records(self):
+        loc = self.locator("a.png")
+        self.replace(loc, tags=["gone", "kept"])
+        self.db.delete_tag_all_prompts("gone")
+        self.assertEqual(
+            self.db.read_image_tags(loc, DIGEST_A),
+            {"tags": ["kept"], "tag_revision": 1, "prior_count": 0},
+        )
+        self.schema_clean()
+
+    def test_junction_failure_leaves_nothing_behind(self):
+        loc = self.locator("a.png")
+        self.replace(loc, tags=["a"])
+        conn = self.db.model.get_connection()
+        conn.execute(
+            "CREATE TRIGGER reject_image_tag BEFORE INSERT ON media_image_tags "
+            "BEGIN SELECT RAISE(ABORT,'image tag failure'); END"
+        )
+        conn.commit()
+        before = self.rows(*ANNOTATION_TABLES, "tags")
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.replace(loc, tags=["b"], revision=1)
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.add(self.locator("new.png"), DIGEST_B, ["c"])
+        self.assertFalse(conn.in_transaction)
+        self.assertEqual(self.rows(*ANNOTATION_TABLES, "tags"), before)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_api_vocabulary.py

@@ -1,6 +1,8 @@
 """Vocabulary read routes page with signed, epoch-scoped cursors."""
 
+import hashlib
 import json
+import os
 import unittest
 from unittest.mock import patch
 
@@ -184,6 +186,7 @@ class TestVocabularyAPI(ImageAPITestCase):
                     "id": created["tag"]["id"],
                     "name": "sunset",
                     "prompt_count": 0,
+                    "image_count": 0,
                 },
             },
         )
@@ -202,6 +205,7 @@ class TestVocabularyAPI(ImageAPITestCase):
                     "other_prompts": 0,
                     "negative_variants": 1,
                     "archived_negative_variants": 0,
+                    "images": 0,
                 },
             },
         )
@@ -221,9 +225,11 @@ class TestVocabularyAPI(ImageAPITestCase):
                     "id": created["tag"]["id"],
                     "name": "sunset",
                     "prompt_count": 2,
+                    "image_count": 0,
                 },
                 "components_changed": 1,
                 "pairs_changed": 1,
+                "images_changed": 0,
             },
         )
         self.assertEqual(self.api.db.get_prompt_by_id(parent)["tags"], ["sunset"])
@@ -242,6 +248,7 @@ class TestVocabularyAPI(ImageAPITestCase):
                 "deleted": True,
                 "components_changed": 1,
                 "pairs_changed": 1,
+                "images_changed": 0,
             },
         )
         self.assertEqual(self.api.db.get_all_tags(), [])
@@ -253,6 +260,95 @@ class TestVocabularyAPI(ImageAPITestCase):
             self.assertEqual(response.status, 400, bad)
             self.assertNotIn("code", result)
 
+    def image(self, name, tags, digest="a", mtime_ns=1):
+        """Tag the bytes ``digest`` of the file ``name``; return its locator."""
+        locator = os.path.normcase(str(self.output_dir / name))
+        self.api.db.replace_image_tags(
+            locator=locator,
+            sha256=hashlib.sha256(digest.encode()).hexdigest(),
+            size=1,
+            mtime_ns=mtime_ns,
+            expected_epoch=self.api.db.get_execution_epoch(),
+            expected_tag_revision=0,
+            tags=tags,
+        )
+        return locator
+
+    async def test_tag_rows_report_image_count_by_locator(self):
+        self._save_prompt("Prompt only", tags=["prompt-only", "shared"])
+        self.image("one.png", ["twice", "shared"], digest="old")
+        self.image("one.png", ["twice"], digest="new", mtime_ns=2)
+        self.image("two.png", ["shared"], digest="new")
+        self.image("three.png", ["prior-only"], digest="first")
+        self.image("three.png", [], digest="second", mtime_ns=2)
+        response, page = await self.get("tags")
+        self.assertEqual(response.status, 200, page)
+        rows = {row["name"]: row for row in page["tags"]}
+        self.assertEqual(
+            {
+                name: (row["prompt_count"], row["image_count"])
+                for name, row in rows.items()
+            },
+            {
+                "prompt-only": (1, 0),
+                "shared": (1, 2),
+                "twice": (0, 1),
+                "prior-only": (0, 1),
+            },
+        )
+        row = rows["shared"]
+        self.assertEqual(row["image_count"], 2)
+        self.assertEqual(set(row), {"id", "name", "prompt_count", "image_count"})
+
+    async def test_impact_and_delete_counts_include_images(self):
+        epoch = self.api.db.get_execution_epoch()
+        self._save_prompt("Parent", tags=["moody"])
+        self.image("a.png", ["moody", "keep"])
+        self.image("b.png", ["moody"])
+        _, page = await self.get("tags")
+        moody = next(row for row in page["tags"] if row["name"] == "moody")
+        self.assertEqual(moody["image_count"], 2)
+        response, impact = await self.get(f"tags/{moody['id']}/impact")
+        self.assertEqual(response.status, 200, impact)
+        self.assertEqual(
+            impact["impact"],
+            {
+                "positive_prompts": 1,
+                "other_prompts": 0,
+                "negative_variants": 0,
+                "archived_negative_variants": 0,
+                "images": 2,
+            },
+        )
+        response, deleted = await self.send(
+            "POST",
+            f"tags/{moody['id']}/delete",
+            {"expected_epoch": epoch, "name": "moody"},
+        )
+        self.assertEqual(response.status, 200, deleted)
+        self.assertEqual(
+            deleted,
+            {
+                "success": True,
+                "library_epoch": epoch,
+                "deleted": True,
+                "components_changed": 1,
+                "pairs_changed": 0,
+                "images_changed": 2,
+            },
+        )
+        keep = next(
+            row for row in (await self.get("tags"))[1]["tags"] if row["name"] == "keep"
+        )
+        response, renamed = await self.send(
+            "PATCH",
+            f"tags/{keep['id']}",
+            {"expected_epoch": epoch, "name": "keep", "new_name": "kept"},
+        )
+        self.assertEqual(response.status, 200, renamed)
+        self.assertEqual(renamed["images_changed"], 1)
+        self.assertEqual(renamed["tag"]["image_count"], 1)
+
     async def test_create_route_validates_names(self):
         epoch = self.api.db.get_execution_epoch()
         before = self.tag_rows()

tests/test_pair_tag_maintenance.py

@@ -1,5 +1,6 @@
 """Global tag maintenance preserves pair and standalone tag relationships."""
 
+import hashlib
 import os
 import sqlite3
 import tempfile
@@ -18,6 +19,7 @@ class TestPairTagMaintenance(unittest.TestCase):
         directory = tempfile.TemporaryDirectory()
         self.addCleanup(directory.cleanup)
         self.addCleanup(assert_closed, self, directory.name)
+        self.root = directory.name
         self.db = PromptDatabase(os.path.join(directory.name, "pair-tags.db"))
         self.addCleanup(self.db.close_all)
 
@@ -27,6 +29,36 @@ class TestPairTagMaintenance(unittest.TestCase):
     def tags(self, pair_id: int) -> list:
         return self.db.get_prompt_pair(pair_id)["tags"]
 
+    def image(self, name: str, tags: list, digest: str = "a", revision: int = 0):
+        """Tag the bytes ``digest`` of the file ``name``; return its locator."""
+        locator = os.path.normcase(os.path.join(os.path.realpath(self.root), name))
+        replace = getattr(self.db, "replace_image_tags", None)
+        self.assertTrue(callable(replace), "replace_image_tags is missing")
+        replace(
+            locator=locator,
+            sha256=hashlib.sha256(digest.encode()).hexdigest(),
+            size=1,
+            mtime_ns=1,
+            expected_epoch=self.db.get_execution_epoch(),
+            expected_tag_revision=revision,
+            tags=tags,
+        )
+        return locator
+
+    def image_tags(self, locator: str, digest: str = "a") -> list:
+        return self.db.read_image_tags(
+            locator, hashlib.sha256(digest.encode()).hexdigest()
+        )["tags"]
+
+    def image_rows(self) -> list:
+        conn = self.db.model.get_connection()
+        return [
+            tuple(row)
+            for row in conn.execute(
+                "SELECT byte_id, tag_id, position FROM media_image_tags ORDER BY 1,3"
+            )
+        ]
+
     def test_rename_to_existing_tag_preserves_pair_links_and_deduplicates(self):
         first = self.pair("portrait", ["old", "new", "keep"])
         second = self.pair("landscape", ["old"])
@@ -136,6 +168,135 @@ class TestPairTagMaintenance(unittest.TestCase):
         self.assertEqual(self.tags(other), ["source"])
         self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["source"])
 
+    def test_rename_and_merge_move_image_links_with_dedup(self):
+        pair = self.pair("portrait", ["old", "new"])
+        both = self.image("both.png", ["new", "old", "keep"])
+        only_old = self.image("old.png", ["old"])
+        result = self.db.rename_tag_all_prompts("old", "new")
+        self.assertEqual(result["affected_count"], 0)  # prompts only, as before
+        self.assertEqual(self.tags(pair), ["new"])
+        self.assertEqual(self.image_tags(both), ["new", "keep"])
+        self.assertEqual(self.image_tags(only_old), ["new"])
+        self.assertNotIn("old", self.db.get_all_tags())
+        self.db.rename_tag_all_prompts("keep", "kept")
+        self.assertEqual(self.image_tags(both), ["new", "kept"])
+        merged = self.db.merge_tags(["kept", "new"], "final")
+        self.assertEqual(merged["tags_merged"], 2)
+        self.assertEqual(merged["affected_count"], 0)
+        self.assertEqual(self.image_tags(both), ["final"])
+        self.assertEqual(self.image_tags(only_old), ["final"])
+        self.assertEqual(self.tags(pair), ["final"])
+        epoch = self.db.get_execution_epoch()
+        conn = self.db.model.get_connection()
+        final = conn.execute("SELECT id FROM tags WHERE name='final'").fetchone()[0]
+        native = self.db.rename_vocabulary_tag(
+            final, expected_epoch=epoch, name="final", new_name="native"
+        )
+        self.assertEqual(native["images_changed"], 2)
+        self.assertEqual(self.image_tags(both), ["native"])
+        self.assertEqual(self.image_tags(only_old), ["native"])
+        self.image("again.png", ["native", "merge-me"])
+        source = conn.execute("SELECT id FROM tags WHERE name='merge-me'").fetchone()
+        receipt = self.db.rename_vocabulary_tag(
+            source[0], expected_epoch=epoch, name="merge-me", new_name="native"
+        )
+        self.assertTrue(receipt["merged"])
+        self.assertEqual(receipt["images_changed"], 1)
+        self.assertEqual(receipt["tag"]["image_count"], 3)
+        self.assertEqual(len(self.image_rows()), 3)
+
+    def test_delete_cascades_image_links_and_keeps_legacy_counts(self):
+        pair = self.pair("portrait", ["remove", "keep"])
+        prompt = self.db.save_prompt("standalone", tags=["remove", "keep"])
+        locator = self.image("a.png", ["keep", "remove"])
+        result = self.db.delete_tag_all_prompts("remove")
+        self.assertEqual(result["affected_count"], 1)
+        self.assertEqual(self.tags(pair), ["keep"])
+        self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["keep"])
+        self.assertEqual(self.image_tags(locator), ["keep"])
+        self.assertEqual(len(self.image_rows()), 1)
+        epoch = self.db.get_execution_epoch()
+        conn = self.db.model.get_connection()
+        keep = conn.execute("SELECT id FROM tags WHERE name='keep'").fetchone()[0]
+        impact = self.db.vocabulary_tag_impact(keep)["impact"]
+        self.assertEqual(impact["images"], 1)
+        deleted = self.db.delete_vocabulary_tag(keep, expected_epoch=epoch, name="keep")
+        self.assertEqual(
+            (
+                deleted["components_changed"],
+                deleted["pairs_changed"],
+                deleted["images_changed"],
+            ),
+            (1, 1, 1),
+        )
+        self.assertEqual(self.image_tags(locator), [])
+        self.assertEqual(self.image_rows(), [])
+        self.assertEqual(
+            conn.execute("SELECT COUNT(*) FROM media_annotation_bytes").fetchone()[0], 1
+        )
+
+    def test_image_only_and_prior_only_names_participate(self):
+        only = self.image("only.png", ["image-only"])
+        prior = self.image("prior.png", ["prior-only"], digest="old")
+        self.image("prior.png", [], digest="new")  # the file's bytes changed
+        self.db.save_prompt("standalone", tags=["target"])
+        merged = self.db.merge_tags(["image-only"], "target")
+        self.assertEqual((merged["tags_merged"], merged["affected_count"]), (1, 0))
+        self.assertEqual(self.image_tags(only), ["target"])
+        self.assertEqual(
+            self.db.rename_tag_all_prompts("prior-only", "renamed")["affected_count"], 0
+        )
+        self.assertEqual(self.image_tags(prior, "old"), ["renamed"])
+        self.assertEqual(self.image_tags(prior, "new"), [])
+        self.assertEqual(set(self.db.get_all_tags()), {"target", "renamed"})
+        epoch = self.db.get_execution_epoch()
+        conn = self.db.model.get_connection()
+        renamed = conn.execute("SELECT id FROM tags WHERE name='renamed'").fetchone()[0]
+        receipt = self.db.rename_vocabulary_tag(
+            renamed, expected_epoch=epoch, name="renamed", new_name="target"
+        )
+        self.assertTrue(receipt["merged"])
+        self.assertEqual(receipt["images_changed"], 1)
+        self.assertEqual(self.image_tags(prior, "old"), ["target"])
+        self.assertEqual(receipt["tag"]["image_count"], 2)
+
+    def test_image_junction_failure_rolls_back_prompt_pair_and_name_changes(self):
+        pair = self.pair("portrait", ["source", "target"])
+        prompt = self.db.save_prompt("standalone", tags=["source"])
+        locator = self.image("a.png", ["source"])
+        conn = self.db.model.get_connection()
+        tables = ("tags", "prompt_tags", "prompt_pair_tags", "media_image_tags")
+        before = [
+            conn.execute(f"SELECT * FROM {table} ORDER BY 1,2").fetchall()
+            for table in tables
+        ]
+        source = conn.execute("SELECT id FROM tags WHERE name='source'").fetchone()[0]
+        conn.execute(
+            "CREATE TRIGGER reject_image_tag_change BEFORE UPDATE "
+            "ON media_image_tags BEGIN SELECT RAISE(ABORT,'image tag failure'); END"
+        )
+        conn.commit()
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.db.rename_vocabulary_tag(
+                source,
+                expected_epoch=self.db.get_execution_epoch(),
+                name="source",
+                new_name="target",
+            )
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.db.merge_tags(["source"], "target")
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.db.rename_tag_all_prompts("source", "target")
+        self.assertFalse(conn.in_transaction)
+        after = [
+            conn.execute(f"SELECT * FROM {table} ORDER BY 1,2").fetchall()
+            for table in tables
+        ]
+        self.assertEqual(after, before)
+        self.assertEqual(self.tags(pair), ["source", "target"])
+        self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["source"])
+        self.assertEqual(self.image_tags(locator), ["source"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_vocabulary.py

@@ -1,5 +1,7 @@
 """Bounded vocabulary reads count exactly what Library Browse shows."""
 
+import hashlib
+import os
 import re
 import sqlite3
 import tempfile
@@ -61,7 +63,48 @@ class TestVocabularyReads(unittest.TestCase):
         by_name = {row["name"]: row for row in page["tags"]}
         self.assertEqual(by_name["z"]["prompt_count"], 0)
         self.assertEqual(by_name["a"]["prompt_count"], 2)
-        self.assertEqual(set(by_name["c"]), {"id", "name", "prompt_count"})
+        self.assertEqual(
+            set(by_name["c"]), {"id", "name", "prompt_count", "image_count"}
+        )
+        self.assertEqual({row["image_count"] for row in page["tags"]}, {0})
+
+    def image(self, name, tags, digest="a", mtime_ns=1):
+        """Tag the bytes ``digest`` of the file ``name``; return its locator."""
+        locator = os.path.normcase(os.path.join(os.path.realpath(self.temp.name), name))
+        self.read(
+            "replace_image_tags",
+            locator=locator,
+            sha256=hashlib.sha256(digest.encode()).hexdigest(),
+            size=1,
+            mtime_ns=mtime_ns,
+            expected_epoch=self.db.get_execution_epoch(),
+            expected_tag_revision=0,
+            tags=tags,
+        )
+        return locator
+
+    def test_tag_rows_report_image_count_by_locator(self):
+        self.save("Prompt only", tags=["prompt-only"])
+        self.image("one.png", ["twice", "shared"], digest="old")
+        self.image(
+            "one.png", ["twice"], digest="new", mtime_ns=2
+        )  # two bytes, one file
+        self.image("two.png", ["shared"], digest="new")  # equal digest, other locator
+        self.image("three.png", ["prior-only"], digest="first")
+        self.image("three.png", [], digest="second", mtime_ns=2)
+        rows = {row["name"]: row for row in self.all_tags()}
+        self.assertEqual(rows["twice"]["image_count"], 1)
+        row = rows["shared"]
+        self.assertEqual(row["image_count"], 2)
+        self.assertEqual(rows["prior-only"]["image_count"], 1)
+        self.assertEqual(rows["prompt-only"]["image_count"], 0)
+        self.assertEqual(
+            {name: r["prompt_count"] for name, r in rows.items()},
+            {"prompt-only": 1, "twice": 0, "shared": 0, "prior-only": 0},
+        )
+        self.assertEqual(
+            self.db.list_library(tags=["shared"])["total"], row["prompt_count"]
+        )
 
     def test_prompt_count_equals_library_browse_total_for_every_tag(self):
         self.save("Parent only", tags=["parent-only"])
@@ -302,7 +345,12 @@ class TestVocabularyTagCommands(unittest.TestCase):
         self.assertTrue(created["created"])
         self.assertEqual(
             created["tag"],
-            {"id": self.tag_id("sunset"), "name": "sunset", "prompt_count": 0},
+            {
+                "id": self.tag_id("sunset"),
+                "name": "sunset",
+                "prompt_count": 0,
+                "image_count": 0,
+            },
         )
         after = self.junction_rows()
         self.assertEqual(after[1:], before[1:])
@@ -345,9 +393,15 @@ class TestVocabularyTagCommands(unittest.TestCase):
             {
                 "library_epoch": self.epoch,
                 "merged": False,
-                "tag": {"id": tag_id, "name": "dramatic", "prompt_count": 2},
+                "tag": {
+                    "id": tag_id,
+                    "name": "dramatic",
+                    "prompt_count": 2,
+                    "image_count": 0,
+                },
                 "components_changed": 2,
                 "pairs_changed": 2,
+                "images_changed": 0,
             },
         )
         self.assertIsNone(self.tag_id("moody"))
@@ -397,7 +451,8 @@ class TestVocabularyTagCommands(unittest.TestCase):
         )
         self.assertFalse(result["merged"])
         self.assertEqual(
-            result["tag"], {"id": tag_id, "name": "cat", "prompt_count": 1}
+            result["tag"],
+            {"id": tag_id, "name": "cat", "prompt_count": 1, "image_count": 0},
         )
         self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["cat"])
 
@@ -467,6 +522,7 @@ class TestVocabularyTagCommands(unittest.TestCase):
                     "other_prompts": 1,
                     "negative_variants": 1,
                     "archived_negative_variants": 1,
+                    "images": 0,
                 },
             },
         )
@@ -482,6 +538,7 @@ class TestVocabularyTagCommands(unittest.TestCase):
                 "deleted": True,
                 "components_changed": 2,
                 "pairs_changed": 2,
+                "images_changed": 0,
             },
         )
         self.assertIsNone(self.tag_id("moody"))
@@ -494,6 +551,69 @@ class TestVocabularyTagCommands(unittest.TestCase):
             )
         self.assertEqual(caught.exception.code, "vocabulary-conflict")
 
+    def image(self, name, tags, digest="a", mtime_ns=1):
+        locator = os.path.normcase(os.path.join(os.path.realpath(self.temp.name), name))
+        self.command(
+            "replace_image_tags",
+            locator=locator,
+            sha256=hashlib.sha256(digest.encode()).hexdigest(),
+            size=1,
+            mtime_ns=mtime_ns,
+            expected_epoch=self.epoch,
+            expected_tag_revision=0,
+            tags=tags,
+        )
+        return locator
+
+    def test_impact_and_delete_counts_include_images(self):
+        self.owners()
+        self.image("a.png", ["moody", "keep"])
+        self.image("a.png", ["moody"], digest="prior", mtime_ns=2)
+        self.image("b.png", ["moody"])
+        tag_id = self.tag_id("moody")
+        impact = self.command("vocabulary_tag_impact", tag_id)["impact"]
+        self.assertEqual(impact["images"], 2)
+        self.assertEqual(
+            self.command("vocabulary_tag_impact", self.tag_id("keep"))["impact"],
+            {
+                "positive_prompts": 0,
+                "other_prompts": 0,
+                "negative_variants": 0,
+                "archived_negative_variants": 0,
+                "images": 1,
+            },
+        )
+        result = self.command(
+            "delete_vocabulary_tag", tag_id, expected_epoch=self.epoch, name="moody"
+        )
+        self.assertEqual(
+            (
+                result["components_changed"],
+                result["pairs_changed"],
+                result["images_changed"],
+            ),
+            (2, 2, 2),
+        )
+        with self.db.model.get_connection() as conn:
+            self.assertEqual(
+                conn.execute("SELECT COUNT(*) FROM media_image_tags").fetchone()[0], 1
+            )
+            self.assertEqual(
+                conn.execute("SELECT COUNT(*) FROM media_annotation_bytes").fetchone()[
+                    0
+                ],
+                3,
+            )
+        renamed = self.command(
+            "rename_vocabulary_tag",
+            self.tag_id("keep"),
+            expected_epoch=self.epoch,
+            name="keep",
+            new_name="kept",
+        )
+        self.assertEqual(renamed["images_changed"], 1)
+        self.assertEqual(renamed["tag"]["image_count"], 1)
+
 
 _OLD_STAMP = "2000-01-01T00:00:00+00:00"
 _DML = re.compile(r"\s*(INSERT|UPDATE|DELETE|REPLACE)\b", re.IGNORECASE)

web/js/workspace/vocabulary-state.mjs

@@ -16,7 +16,8 @@ import { countOf } from './plural.mjs';
 const LIMIT = 100;
 const BASE = '/prompt_manager/vocabulary';
 const KINDS = {
-    tags: {key: row => row.id, valid: row => id(row.id) && typeof row.name === 'string' && row.name.length > 0},
+    // image_count (#249 image tags) is optional until every server reports it.
+    tags: {key: row => row.id, valid: row => id(row.id) && typeof row.name === 'string' && row.name.length > 0 && optionalCount(row.image_count)},
     categories: {key: row => row.value, valid: row => typeof row.value === 'string' && row.value.length > 0},
 };
 const RECOVERY = {'storage-changed': 'storage-changed', 'vocabulary-conflict': 'conflict'};
@@ -29,6 +30,7 @@ const IMPACT_FIELDS = ['positive_prompts', 'other_prompts', 'negative_variants',
 const object = value => value !== null && typeof value === 'object' && !Array.isArray(value);
 const id = value => Number.isSafeInteger(value) && value > 0;
 const count = value => Number.isSafeInteger(value) && value >= 0;
+const optionalCount = value => value === undefined || count(value);
 const epochOf = value => typeof value === 'string' && /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/.test(value);
 const hexToken = value => typeof value === 'string' && /^[0-9a-f]{64}$/.test(value);
 
@@ -108,7 +110,7 @@ export function createVocabularyStream({api, kind, onChange = () => {}}) {
 }
 
 function validateTag(value) {return object(value) && KINDS.tags.valid(value) && count(value.prompt_count);}
-function validateImpact(value) {return object(value) && IMPACT_FIELDS.every(name => count(value[name]));}
+function validateImpact(value) {return object(value) && IMPACT_FIELDS.every(name => count(value[name])) && optionalCount(value.images);}
 function requireRow(row) {requireValue(object(row) && id(row.id) && typeof row.name === 'string', 'Invalid tag row');}
 function requireEpoch(epoch) {requireValue(epochOf(epoch), 'Invalid library epoch');}
 function requireText(value, name) {requireValue(typeof value === 'string', `${name} must be a string`);}
@@ -116,8 +118,9 @@ function requireText(value, name) {requireValue(typeof value === 'string', `${na
 const RECEIPTS = {
     'create-tag': value => typeof value.created === 'boolean' && validateTag(value.tag),
     'rename-tag': value => typeof value.merged === 'boolean' && validateTag(value.tag) &&
-        count(value.components_changed) && count(value.pairs_changed),
-    'delete-tag': value => value.deleted === true && count(value.components_changed) && count(value.pairs_changed),
+        count(value.components_changed) && count(value.pairs_changed) && optionalCount(value.images_changed),
+    'delete-tag': value => value.deleted === true && count(value.components_changed) && count(value.pairs_changed) &&
+        optionalCount(value.images_changed),
     'apply-category': (value, body) => value.source === body.source && value.target === body.target &&
         typeof value.consolidated === 'boolean' && validateImpact(value.impact),
 };
@@ -270,6 +273,7 @@ function impactList(impact) {
         parts.push(`${countOf(negatives, 'negative variant')}${archived}`);
     }
     if (impact.other_prompts) parts.push(countOf(impact.other_prompts, 'other saved prompt'));
+    if (impact.images) parts.push(countOf(impact.images, 'image'));
     return parts.length > 1 ? `${parts.slice(0, -1).join(', ')} and ${parts.at(-1)}` : parts.join('');
 }
 
@@ -282,6 +286,6 @@ export function categoryPreviewText(shown) {
 /** The Delete tag confirmation sentence (decision D14). */
 export function deleteTagText(name, impact) {
     const list = impactList(impact);
-    if (!list) return `“${name}” is not linked to any saved prompt. Deleting it removes only the name.`;
+    if (!list) return `“${name}” is not linked to any saved prompt or image. Deleting it removes only the name.`;
     return `Deleting “${name}” removes the name and its links from ${list}. The prompts, negative variants, images and history themselves are kept.`;
 }