test(comfyui): count #347's cached-run cases in the execution suite pin (#347)

The pin listed 19 execution-text-link cases per project; #347 adds five node
kinds and eight one-renderer cases, so the suite lists 32 per project and 118
in total with #332's image-attribution case from main.

TESTING.md

@@ -4676,6 +4676,70 @@ 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 categories (#332, store and routes): an image file may carry its own
+category; without one it shows the category its link supplies.
+
+- The table is `image_categories(root_key, relative_path, category, updated_at)`
+  with primary key `(root_key, relative_path)` and the index
+  `idx_image_categories_category`. `migrate_image_categories` creates both empty
+  inside a savepoint, with no backfill. It reads only `sqlite_master` and its own
+  table's PRAGMAs, and writes only its two objects. `tests/test_image_categories.py`
+  checks this with an SQLite authorizer on a 23,000-link library, and checks that
+  `prompts` and `prompt_pairs` `(id, category, updated_at)` are byte-for-byte equal
+  afterwards.
+- The **file key** is `image_category_key(sources, absolute path)` in
+  `py/api/images.py`: the deepest output source that contains the path (compared
+  lexically, never resolving the path, so a symlinked subfolder keeps its listed
+  spelling) and the path below it, keyed with the favorites helpers. A source
+  matches by its real path or its configured spelling (a detected output folder
+  can be a symlink). Without nested sources it equals a favorite's key. Every write and read uses it. A path
+  that no source contains has no key, so it has no own category.
+- `image_categories_schema_error` checks the structure at startup (a value saved
+  under older rules never stops the library opening; a table without its index,
+  which this code never leaves, is repaired by the install). Restore also checks the
+  values: text columns, keys within the favorites bound, `category` 1 to 100
+  characters with no character below U+0020 and no leading or trailing space, and
+  `updated_at` of at most 64 bytes. Backups made before #332 restore without the
+  table; a migrated restore stage must have it (`require_library_state`).
+- `PUT /prompt_manager/images/category` `{items, category}` answers
+  `{success, category, updated, unchanged, not_found}`.
+  `DELETE /prompt_manager/images/category` `{items}` answers `{success, cleared}`.
+  `GET /prompt_manager/images/categories` answers `{success, categories}`: the
+  sorted distinct own values whose file is in some source's listing.
+- `items` are 1 to 500 Images targets `{root_index, root_scope, path}`, checked by
+  `py/api/media_target.py` (shared with Collections, #331). Refusals:
+  - 400: a malformed body, a field other than `items` and `category`, a
+    `root_index` outside the output sources (a LoRA folder), or a category that is
+    blank after stripping, over 100 characters, or holds a control character;
+  - 403: traversal, a drive, a backslash, a non-media name ("Only media files can
+    have a category") or a key over the stored bound;
+  - 409 `source-changed`: a stale `root_scope`;
+  - 413: a body over 1 MiB.
+  Any invalid item fails the whole request before a write. Targets that fold to one
+  key are written and counted once. Instead of `items` a request may send `filter`,
+  the Images listing's own query ("Select all N matching", #392, keys as in
+  `parse_listing_filter`), resolved now: over 500 matches is a 400
+  `too-many-images` ("Select at most 500 images."), a renumbered source a 409
+  `source-changed`.
+- PUT writes only files the source's listing holds and counts the rest in
+  `not_found`; when none is listed it answers 404 `files-not-found` and writes
+  nothing. A file that already has the value counts as `unchanged` and keeps its
+  `updated_at`. DELETE accepts any well-formed target, so a missing file's
+  category can still be removed. An unexpected failure is a logged 500 ("The
+  category could not be changed." / "Image categories could not be read.") without
+  the exception text. The routes never read or write `prompts`, `prompt_pairs` or
+  `generated_images`; `POST /prompt_manager/bulk/category` is unchanged.
+- Duplicate cleanup (System › Maintenance): the kept copy's own category wins. A
+  kept copy without one receives the removed copy's (`COPY_SQL`, part of the
+  cleanup's query-plan check). The removed copy's row is never deleted, so a
+  give-back (the kept copy changed) and a failed delete lose nothing. The job log
+  says "Kept N image categories on the kept copies".
+  `tests/test_duplicate_image_categories.py` covers nested sources in both orders.
+- A moved, renamed or deleted file's row stays and is hidden; a file that later
+  appears at the same path shows it. The real-host check is the last test in
+  `tests/comfyui/image-attribution.spec.mjs`: PUT, list, DELETE on a generated
+  output.
+
 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,
@@ -5205,7 +5269,7 @@ which used to save `null` under a link.
   Primitive or through `PreviewAny`, for PromptManager, PromptManagerText, Focused, Split
   and the Pair (positive, negative, both, neither linked, and a Primitive negative). A
   wildcard Primitive on PromptManager (queued as written) and Focused (one choice) saves
-  and counts what was queued. The linked text is saved, counted once, recorded as
+  and counts what was queued. The linked text is saved, counted once per job, recorded as
   the run's raw positive/negative, and after a scan the run's image links to it. Text
   only known at run time (`PreviewAny`) cannot be read from the image; the case only
   requires that it is never linked to another prompt (its attribution from the recorded
@@ -5219,6 +5283,39 @@ PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:comfyui
 PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:executions -- execution-text-link
 ```
 
+## Cached runtime-linked runs (#347)
+
+Text only known at run time (a `PreviewAny` or LLM link, a list node) is counted by the
+node when it executes; the queue hook cannot read it. When ComfyUI serves such a node from
+its output cache, `utils/cached_runs.py` counts the job instead: every execution records
+a receipt (digests of its STRING outputs and hashes of the text it stores, per run key and
+queue item), and on `execution_cached` the receipt whose digests equal the cache entry's
+text is replayed once. The role (positive, negative, which Pair outputs are used) and what
+the hook already saw are recomputed from the job's own graph, and the node and replay
+decide through the same `own_counts`.
+
+- Fail closed: no receipt in this process (after a restart, an external cache provider),
+  changed cached text, more than 64 list values, a node the job's outputs do not need, or
+  a host without the queue/executor interfaces: nothing is counted, as before #347.
+- A cached node counts at job start like the hook: a job failing later still counts it.
+  A node that executes after its replay in the same job (RAM-pressure freed the entry)
+  skips its own count but still consumes the pending first-use catch-up.
+- No backfill: earlier cached jobs keep their counts. The history page's "tracked runs"
+  can stay below the Library count for these prompts (#403).
+- Tests: `tests/test_cached_runs.py` (whole jobs with the real nodes, a real database,
+  fake host caches); `tests/comfyui/execution-text-link.spec.mjs` "#347" cases on the
+  managed host: PreviewAny into each node kind, 3 jobs changing only the sampler seed,
+  count 3 in both renderers; in one renderer the Classic, LRU, RAM-pressure and no-cache
+  modes, A, B, A from the owned `PMCycledText` under LRU, typed text, a downstream
+  failure, a "run selected output" job that does not need the node, and a rewire to the
+  negative input. The RAM-pressure eviction path cannot be forced on a real host and is
+  covered by the Python tests.
+
+```sh
+python3 -m unittest discover -s tests -p "test_cached_runs.py"
+PM_COMFY_ROOT=... PM_TEST_PYTHON=... npm run --prefix tools/vendor test:executions -- execution-text-link --grep "#347"
+```
+
 ## Durable execution ledger
 
 `npm run --prefix tools/vendor test:executions` runs 52 real execution cases

__init__.py

@@ -195,6 +195,26 @@ except Exception as e:
             f"Failed to register usage hook: {e}"
         )
 
+# A cached node counts what its last execution of the same text counted (#347).
+try:
+    from .py.config import server_instance
+    from .utils.cached_runs import register_cached_runs
+
+    register_cached_runs(server_instance, _db_factory)
+except Exception as e:
+    try:
+        from .utils.logging_config import get_logger
+
+        get_logger("prompt_manager.init").error(
+            f"Failed to register cached-run counting: {e}"
+        )
+    except Exception:
+        import logging
+
+        logging.getLogger("prompt_manager.init").error(
+            f"Failed to register cached-run counting: {e}"
+        )
+
 
 def _history_db_factory() -> "PromptDatabase":
     """Initialize history on its worker even if monitoring initialization failed."""

database/duplicate_copies.py

@@ -11,6 +11,8 @@ IMMEDIATE`` transaction, so no row ever names a deleted file:
 - its favorite;
 - its image tags (``media_image_tags`` of its byte record), joined onto the
   kept copy's byte record as their union, in order;
+- its own image category (#332), copied to the kept copy when that copy has
+  none; the removed copy's row is kept, so a give-back loses nothing;
 - its pending Needs review items, settled: only an item still ``pending`` at
   revision 0 with no decision row is removed (``review._SETTLE_ITEM``), so a
   decided item is never touched.
@@ -22,6 +24,8 @@ import os
 
 from .annotation_schema import BYTES_TABLE, FILES_TABLE
 from .annotations import _BUMP_INDEX
+from .image_categories import COPY_SQL as COPY_CATEGORY_SQL
+from .image_categories import copy_image_category
 from .models import normalize_image_path, utc_now_iso
 from .review import _BUMP_REVIEW_INDEX, _PENDING_OF_LOCATOR, _SETTLE_ITEM
 from .review_changes import notify_review_change
@@ -76,6 +80,7 @@ FILE_LINK_SQL = (
     COPY_BYTE_SQL,
     COPY_TAG_COUNT_SQL,
     PENDING_COUNT_SQL,
+    COPY_CATEGORY_SQL,
 )
 
 
@@ -137,8 +142,10 @@ class DuplicateCopyOperations:
         removed key, kept root key, kept key)`` per listed root holding the
         file), ``removed_locator``,
         ``kept_locator``, ``sha256`` and the kept copy's ``kept_size`` and
-        ``kept_mtime_ns``. Returns ``{moved, merged, favorites, tags,
-        review}``.
+        ``kept_mtime_ns``; ``categories``: ``(removed file key, kept file
+        key)`` pairs of the image categories (#332). Returns ``{moved,
+        merged, favorites, tags, review, categories}`` (``categories``: the
+        image categories copied to the kept copy).
         """
         counts = dict.fromkeys(("moved", "merged", "favorites", "tags", "review"), 0)
         kept_keys = [normalize_image_path(path) for path in move["kept"]]
@@ -174,6 +181,10 @@ class DuplicateCopyOperations:
             for favorite in move.get("favorites", ()):
                 counts["favorites"] += _move_favorite(conn, *favorite)
             counts["tags"] = self._move_copy_tags(conn, move)
+            counts["categories"] = sum(
+                copy_image_category(conn, removed, kept)
+                for removed, kept in move.get("categories", ())
+            )
             for (item_id,) in conn.execute(
                 _PENDING_OF_LOCATOR, (move["removed_locator"],)
             ).fetchall():

database/image_categories.py

@@ -0,0 +1,249 @@
+"""Image categories (#332): an image file's own category.
+
+An image shows its own category when it has one, otherwise the category its
+link supplies (its recipe's, else its positive prompt's); this table holds
+only the own values. A row is keyed by the file key: the deepest configured
+output source that contains the file (its real, case-normalized path) and the
+path below it, case-folded where the OS folds case
+(``py/api/images.py`` ``image_category_key``). Without nested sources the key
+equals a favorite's key. Nothing follows a moved file: its row stays and is
+hidden until a file appears at the same path again.
+
+Writes validate strictly (the route strips the value and applies
+``validate_category``). Startup checks the structure only, so a value saved
+under older rules never stops the library opening; restore also checks the
+values. Optional in old backups, required once a database is current.
+"""
+
+import re
+import sqlite3
+from typing import Iterable
+
+TABLE = "image_categories"
+INDEX = "idx_image_categories_category"
+MAX_KEY_LENGTH = 4096
+MAX_CATEGORY_LENGTH = 100
+_DDL = (
+    f"CREATE TABLE {TABLE} (root_key TEXT NOT NULL, relative_path TEXT NOT NULL, "
+    "category TEXT NOT NULL, updated_at TEXT NOT NULL, "
+    "PRIMARY KEY (root_key, relative_path))"
+)
+_INDEX_DDL = f"CREATE INDEX {INDEX} ON {TABLE}(category)"
+_AUTOINDEX = f"sqlite_autoindex_{TABLE}_1"
+_COLUMNS = (
+    ("root_key", "TEXT", 1, 1),
+    ("relative_path", "TEXT", 1, 2),
+    ("category", "TEXT", 1, 0),
+    ("updated_at", "TEXT", 1, 0),
+)
+# A category may not hold a character below U+0020 (validate_category).
+_CONTROL_GLOB = "*[" + "".join(chr(code) for code in range(1, 32)) + "]*"
+
+CATEGORY_OF_SQL = (
+    f"SELECT category FROM {TABLE} WHERE root_key = ? AND relative_path = ?"
+)
+CATEGORIES_OF_ROOT_SQL = (
+    f"SELECT relative_path, category FROM {TABLE} WHERE root_key = ?"
+)
+UPSERT_SQL = (
+    f"INSERT INTO {TABLE} (root_key, relative_path, category, updated_at) "
+    "VALUES (?, ?, ?, ?) ON CONFLICT(root_key, relative_path) "
+    "DO UPDATE SET category = excluded.category, updated_at = excluded.updated_at"
+)
+DELETE_SQL = f"DELETE FROM {TABLE} WHERE root_key = ? AND relative_path = ?"
+# The duplicate cleanup's copy (#332 MC11): the removed copy's row is copied
+# to the kept copy only when the kept copy has none; it is never deleted.
+COPY_SQL = (
+    f"INSERT OR IGNORE INTO {TABLE} (root_key, relative_path, category, updated_at) "
+    f"SELECT ?, ?, category, updated_at FROM {TABLE} "
+    "WHERE root_key = ? AND relative_path = ?"
+)
+STATEMENTS = (CATEGORY_OF_SQL, CATEGORIES_OF_ROOT_SQL, UPSERT_SQL, DELETE_SQL, COPY_SQL)
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql or "").lower()
+
+
+def _structure_error(
+    conn: sqlite3.Connection, allow_missing: bool, allow_missing_index: bool = False
+) -> str | None:
+    objects = conn.execute(
+        "SELECT type,name,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing image categories schema"
+    tables = [sql for kind, _name, sql in objects if kind == "table"]
+    others = {(kind, name) for kind, name, _sql in objects if kind != "table"}
+    expected = {("index", _AUTOINDEX), ("index", INDEX)}
+    # Before its install, a table without the category index is repaired by
+    # CREATE INDEX IF NOT EXISTS rather than refused.
+    if allow_missing_index and others == {("index", _AUTOINDEX)}:
+        expected = others
+    if len(tables) != 1 or others != expected:
+        return "Image categories name is reserved by an unsupported schema object"
+    if _compact(tables[0]) != _compact(_DDL):
+        return "Image categories has an invalid table definition"
+    index_sql = next((sql for kind, name, sql in objects if name == INDEX), None)
+    if ("index", INDEX) in expected and _compact(index_sql) != _compact(_INDEX_DDL):
+        return "Image categories has an invalid index"
+    autoindex_sql = next(sql for kind, name, sql in objects if name == _AUTOINDEX)
+    if autoindex_sql is not None:
+        return "Image categories has an invalid index"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Image categories has invalid columns"
+    return None
+
+
+def _values_error(conn: sqlite3.Connection) -> str | None:
+    # Bytes bound characters from above (UTF-8 is at most 4 bytes each).
+    limit = MAX_KEY_LENGTH * 4
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(root_key) != 'text' "  # nosec B608
+        "OR typeof(relative_path) != 'text' OR typeof(category) != 'text' "
+        "OR typeof(updated_at) != 'text' "
+        f"OR length(CAST(root_key AS BLOB)) > {limit} "
+        f"OR length(CAST(relative_path AS BLOB)) > {limit} "
+        "OR length(CAST(updated_at AS BLOB)) > 64 "
+        "OR instr(CAST(category AS BLOB), x'00') > 0 "
+        f"OR length(category) < 1 OR length(category) > {MAX_CATEGORY_LENGTH} "
+        "OR category != TRIM(category) OR category GLOB ? LIMIT 1",
+        (_CONTROL_GLOB,),
+    ).fetchone():
+        return "Image categories contains invalid values"
+    for root, path in conn.execute(
+        f"SELECT CAST(root_key AS BLOB), CAST(relative_path AS BLOB) "  # nosec B608
+        f"FROM {TABLE} WHERE length(CAST(root_key AS BLOB)) > {MAX_KEY_LENGTH} "
+        f"OR length(CAST(relative_path AS BLOB)) > {MAX_KEY_LENGTH}"
+    ):
+        for raw in (root, path):
+            if len(bytes(raw).decode("utf-8", "replace")) > MAX_KEY_LENGTH:
+                return "Image categories contains invalid values"
+    return None
+
+
+def image_categories_schema_error(
+    conn: sqlite3.Connection,
+    *,
+    allow_missing: bool = True,
+    check_values: bool = True,
+) -> str | None:
+    """Reject an unsupported image categories schema (and, optionally, values).
+
+    Reads only. A missing table is acceptable for backups that predate it
+    unless ``allow_missing`` is false. ``check_values`` also refuses a stored
+    value the write path could never have written.
+    """
+    error = _structure_error(conn, allow_missing)
+    if error is not None or not check_values:
+        return error
+    exists = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)
+    ).fetchone()
+    return _values_error(conn) if exists else None
+
+
+def migrate_image_categories(conn: sqlite3.Connection) -> None:
+    """Install the table and its index atomically; structure is checked
+    before and after. Reads only ``sqlite_master`` and this table's PRAGMAs
+    and writes only its own two objects (no backfill)."""
+    conn.execute("SAVEPOINT image_categories_schema")
+    try:
+        error = _structure_error(conn, True, allow_missing_index=True)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        conn.execute(
+            _INDEX_DDL.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS", 1)
+        )
+        error = image_categories_schema_error(
+            conn, allow_missing=False, check_values=False
+        )
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE image_categories_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO image_categories_schema")
+        conn.execute("RELEASE image_categories_schema")
+        raise
+
+
+def copy_image_category(conn: sqlite3.Connection, removed_key, kept_key) -> int:
+    """Give the kept copy the removed copy's category when it has none.
+
+    Runs inside the caller's transaction; never deletes the removed copy's
+    row, so a give-back or a failed delete loses nothing. Returns 1 when a
+    row was inserted, else 0.
+    """
+    if tuple(removed_key) == tuple(kept_key):
+        return 0
+    return max(conn.execute(COPY_SQL, (*kept_key, *removed_key)).rowcount, 0)
+
+
+class ImageCategoryOperations:
+    """PromptDatabase mixin: set, clear and read images' own categories.
+
+    Never reads or writes ``prompts``, ``prompt_pairs`` or
+    ``generated_images``.
+    """
+
+    def image_category(self, root_key: str, relative_key: str) -> str | None:
+        """One file's own category, or None."""
+        row = (
+            self.model.get_connection()
+            .execute(CATEGORY_OF_SQL, (root_key, relative_key))
+            .fetchone()
+        )
+        return row[0] if row else None
+
+    def own_categories_of(self, root_keys: Iterable[str]) -> dict:
+        """``{(root key, relative key): category}`` under the given sources.
+
+        One primary-key range read per source.
+        """
+        found = {}
+        conn = self.model.get_connection()
+        for root_key in dict.fromkeys(root_keys):
+            for relative, category in conn.execute(CATEGORIES_OF_ROOT_SQL, (root_key,)):
+                found[(root_key, relative)] = category
+        return found
+
+    def set_image_categories(self, file_keys, category: str) -> dict:
+        """Set *category* on every file key, in one transaction.
+
+        ``file_keys``: ``(root key, relative key)`` pairs from any number of
+        sources. Returns ``{updated, unchanged}``; a file that already had
+        exactly this value is ``unchanged`` and keeps its ``updated_at``.
+        """
+        from .models import utc_now_iso  # models imports this module's schema
+
+        keys = list(dict.fromkeys(tuple(key) for key in file_keys))
+        counts = {"updated": 0, "unchanged": 0}
+        now = utc_now_iso()
+        with self._pair_transaction() as conn:
+            for key in keys:
+                row = conn.execute(CATEGORY_OF_SQL, key).fetchone()
+                if row is not None and row[0] == category:
+                    counts["unchanged"] += 1
+                    continue
+                conn.execute(UPSERT_SQL, (*key, category, now))
+                counts["updated"] += 1
+        return counts
+
+    def clear_image_categories(self, file_keys) -> int:
+        """Remove the own category of every file key (one transaction).
+
+        Returns the number of rows removed.
+        """
+        keys = list(dict.fromkeys(tuple(key) for key in file_keys))
+        removed = 0
+        with self._pair_transaction() as conn:
+            for key in keys:
+                removed += conn.execute(DELETE_SQL, key).rowcount
+        return removed

database/models.py

@@ -34,6 +34,7 @@ from .review_schema import (
     review_without_annotations_error,
 )
 from .saved_views import migrate_saved_views, saved_views_schema_error
+from .image_categories import image_categories_schema_error, migrate_image_categories
 from .workflow_documents import (
     DOCUMENT_ID_COLUMNS,
     migrate_workflow_documents,
@@ -350,6 +351,8 @@ class PromptModel:
         migrate_annotation_state(conn)
         migrate_review_state(conn)
         migrate_autotag_state(conn)
+        # Image categories (#332): independent of every table above.
+        migrate_image_categories(conn)
         # Last: the older generated_images rebuilds run only before file_path.
         migrate_workflow_documents(conn)
 
@@ -1155,8 +1158,9 @@ class PromptModel:
         Args:
             path: File to inspect
             require_library_state: Also require the tables every current
-                database has (Library state, image favorites, saved views,
-                image annotations and the review queue), as a migrated
+                database has (Library state, image favorites, image
+                categories, saved views, image annotations and the review
+                queue), as a migrated
                 restore stage must; old backups may lack them.
             documents: Read every workflow document and refuse a damaged
                 one. A restore checks its upload without it and repairs
@@ -1220,6 +1224,11 @@ class PromptModel:
             )
             if favorites_error:
                 return False, favorites_error
+            categories_error = image_categories_schema_error(
+                conn, allow_missing=not require_library_state
+            )
+            if categories_error:
+                return False, categories_error
             views_error = saved_views_schema_error(
                 conn, allow_missing=not require_library_state
             )

database/operations.py

@@ -27,6 +27,7 @@ from .health import HealthOperations, iter_consistency_issues
 from .duplicate_copies import DuplicateCopyOperations
 from .workflow_storage import WorkflowStorageOperations
 from .category_import import CategoryImportOperations, EmptyReplacement  # noqa: F401
+from .image_categories import ImageCategoryOperations
 from .workflow_documents import (
     DocumentReader,
     documents_of_rows,
@@ -368,6 +369,7 @@ class PromptDatabase(
     DuplicateCopyOperations,
     WorkflowStorageOperations,
     CategoryImportOperations,
+    ImageCategoryOperations,
 ):
     """Database operations class for managing prompts."""
 

prompt_manager.py

@@ -268,6 +268,7 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
         record_runtime_context(
             type(self).__name__, text, prepend_text, append_text, encoding_text
         )
+        self._record_cached_item(unique_id, (encoding_text,), storage_text)
         return result
 
     @classmethod

prompt_manager_base.py

@@ -26,6 +26,13 @@ try:
     from .utils.prompt_graph import resolve_text, run_prompt_nodes
     from .utils.run_keys import executing_run_id, prompt_node_key
     from .utils.usage_tracking import PENDING_FIRST_USE
+    from .utils.cached_runs import (
+        CACHED_RUNS,
+        Role,
+        Seen,
+        own_counts,
+        stored_facts,
+    )
 except ImportError:
     import sys
 
@@ -38,6 +45,13 @@ except ImportError:
     from utils.prompt_graph import resolve_text, run_prompt_nodes
     from utils.run_keys import executing_run_id, prompt_node_key
     from utils.usage_tracking import PENDING_FIRST_USE
+    from utils.cached_runs import (
+        CACHED_RUNS,
+        Role,
+        Seen,
+        own_counts,
+        stored_facts,
+    )
 
 
 class PromptManagerBase:
@@ -71,8 +85,10 @@ class PromptManagerBase:
         """Save the prompt to the SQLite database.
 
         Runs are normally counted by the queue hook (utils/usage_tracking.py), so a
-        save only counts one when asked to (batch items whose text the hook could not
-        see) or when the hook queued this prompt before it existed.
+        save only counts one when asked to (text the hook could not see, unless
+        this job already counted the node from ComfyUI's cache) or when the hook
+        queued this prompt before it existed. The two are independent: the
+        pending catch-up is written even when ``count_run`` is False.
 
         Args:
             text: The prompt text
@@ -167,8 +183,19 @@ class PromptManagerBase:
         batch_item = (
             role["known"] and role["is_run_prompt"] and not role["seen_at_queue"]
         )
+        counts = own_counts(
+            type(self).__name__,
+            stored_facts(text),
+            Role(
+                frozenset({"text"} if role["known"] and role["is_run_prompt"] else ())
+            ),
+            Seen(text=role["seen_at_queue"]),
+        )
         prompt_id = self._save_prompt_to_database(
-            text=text, category=category, tags=tags, count_run=batch_item
+            text=text,
+            category=category,
+            tags=tags,
+            count_run=bool(counts) and not self._replayed(unique_id),
         )
         if role["is_run_prompt"] and not prompt_id:
             # The save failed: an older entry of this key must not stand in (#342)
@@ -182,6 +209,34 @@ class PromptManagerBase:
             )
         return prompt_id
 
+    def _cached_runs(self) -> Any:
+        return getattr(self, "cached_runs", CACHED_RUNS)
+
+    def _replayed(self, unique_id: Any) -> bool:
+        """This job already counted the node from ComfyUI's cache (#347).
+
+        A RAM-pressure cache can free the entry after ``execution_cached``, so
+        the node executes in a job that replayed it: its own count is skipped.
+        """
+        try:
+            return unique_id is not None and self._cached_runs().replayed(unique_id)
+        except Exception as error:
+            self.logger.debug(f"Could not check the cached-run replay: {error}")
+            return False
+
+    def _record_cached_item(
+        self, unique_id: Any, outputs: tuple, text: Any, negative_text: Any = None
+    ) -> None:
+        """Remember this execution's outputs and stored text for cached jobs (#347)."""
+        if unique_id is None:
+            return
+        try:
+            self._cached_runs().record_item(
+                unique_id, outputs, stored_facts(text, negative_text)
+            )
+        except Exception as error:
+            self.logger.debug(f"Could not record the cached-run receipt: {error}")
+
     @staticmethod
     def _run_identity(prompt_graph: Any, unique_id: Any) -> Dict[str, Any]:
         """The tracker's proof that an image belongs to this execution (#342)."""

prompt_manager_pair.py

@@ -12,6 +12,14 @@ else:
 
 try:
     from .prompt_manager import IO, InputTypeDict, PromptManagerFocused
+    from .utils.cached_runs import (
+        PAIR,
+        Role,
+        Seen,
+        own_counts,
+        pair_components,
+        stored_facts,
+    )
     from .utils.hashing import generate_prompt_hash
     from .utils.pair_execution import record_pair_execution
     from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
@@ -23,6 +31,14 @@ try:
     )
 except ImportError:
     from prompt_manager import IO, InputTypeDict, PromptManagerFocused
+    from utils.cached_runs import (
+        PAIR,
+        Role,
+        Seen,
+        own_counts,
+        pair_components,
+        stored_facts,
+    )
     from utils.hashing import generate_prompt_hash
     from utils.pair_execution import record_pair_execution
     from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
@@ -165,6 +181,9 @@ class PromptManagerPair(PromptManagerFocused):
         record_runtime_context(
             type(self).__name__, negative_text, "", "", negative_text, role="negative"
         )
+        self._record_cached_item(
+            unique_id, (positive, negative_text), text, negative_text
+        )
         return result
 
     def _save_and_track_pair(
@@ -209,36 +228,35 @@ class PromptManagerPair(PromptManagerFocused):
             and generate_prompt_hash(resolved_negative)
             == generate_prompt_hash(negative_text)
         )
+        negative_seen = resolved_negative is not None and generate_prompt_hash(
+            resolved_negative
+        ) == generate_prompt_hash(negative_text)
         positive_pending = getattr(self, "pending_first_use", PENDING_FIRST_USE)
         pair_pending = getattr(self, "pending_pair_first_use", PENDING_PAIR_FIRST_USE)
-        counted = set()
-        for field, component_text, resolved_text, component_id in (
-            ("text", text, resolved_positive, prompt_id),
-            (
-                "negative_text",
-                negative_text,
-                resolved_negative,
-                pair["negative_prompt_id"],
-            ),
-        ):
-            component_hash = generate_prompt_hash(component_text)
-            if (
-                field not in used_fields
-                or not component_id
-                or component_hash in counted
-            ):
+        facts = stored_facts(text, negative_text)
+        role = Role(frozenset(used_fields), pair_used)
+        # Own counts are skipped when this job already counted the node from
+        # ComfyUI's cache (#347); the pending first-use catch-up never is.
+        counts = (
+            []
+            if self._replayed(unique_id)
+            else own_counts(
+                PAIR, facts, role, Seen(positive_seen, negative_seen, pair_seen)
+            )
+        )
+        component_ids = {"text": prompt_id, "negative_text": pair["negative_prompt_id"]}
+        for field, component_hash in pair_components(facts, role):
+            component_id = component_ids[field]
+            if not component_id:
                 continue
-            counted.add(component_hash)
-            seen_at_queue = (
-                resolved_text is not None
-                and generate_prompt_hash(resolved_text) == component_hash
+            runs = positive_pending.consume(component_hash) + int(
+                ("prompt", component_hash) in counts
             )
-            runs = positive_pending.consume(component_hash) + int(not seen_at_queue)
             if runs:
                 self.db.record_prompt_use(component_id, times=runs)
         if pair_used:
             runs = pair_pending.consume(pair_run_key(text, negative_text)) + int(
-                not pair_seen
+                ("pair", facts.positive_hash, facts.negative_hash) in counts
             )
             if runs:
                 self.db.record_prompt_pair_use(pair_id, times=runs)

prompt_manager_text.py

@@ -239,6 +239,7 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
         record_runtime_context(
             type(self).__name__, text, prepend_text, append_text, final_text
         )
+        self._record_cached_item(unique_id, (final_text,), storage_text)
         return result
 
     @classmethod

py/api/__init__.py

@@ -72,6 +72,7 @@ from .diagnostics_report import DiagnosticsReportMixin
 from .autotag_routes import AutotagRoutesMixin
 from .lora_integration import LoraIntegrationMixin
 from .lora_import_routes import LoraImportRoutesMixin
+from .image_categories import ImageCategoryRoutesMixin
 
 try:
     from ...database.operations import PromptDatabase
@@ -336,6 +337,7 @@ class PromptManagerAPI(
     AutotagRoutesMixin,
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
+    ImageCategoryRoutesMixin,
 ):
     """REST API handler for PromptManager operations and web interface.
 
@@ -700,6 +702,7 @@ class PromptManagerAPI(
         self._register_autotag_routes(routes)
         self._register_lora_routes(routes)
         self._register_lora_import_routes(routes)
+        self._register_image_category_routes(routes)
 
         # Register gzip compression middleware (once)
         global _gzip_registered

py/api/annotations.py

@@ -27,7 +27,12 @@ from .annotation_suggest import (
     validate_options,
 )
 from .autotag_routes import model_busy_response
-from .images import _forbidden, _has_traversal, _unavailable_link, _verified_title
+
+# The target check is shared with the collection routes (#331); both names
+# stay importable from here.
+from .media_target import MAX_TARGET_PATH  # noqa: F401
+from .media_target import media_target as _target
+from .images import _unavailable_link, _verified_title
 from .prompts import bad_request
 from .. import model_gate
 
@@ -40,7 +45,7 @@ try:
     )
     from ...database.execution_history import subscribe_history_restore
     from ...database.library_commands import LibraryCommandError
-    from ...utils.media_capture import CAPTURE_EXTENSIONS, CaptureError, CaptureStore
+    from ...utils.media_capture import CaptureError, CaptureStore
 except ImportError:
     from database.annotations import (
         MAX_IMAGE_TAGS,
@@ -50,15 +55,13 @@ except ImportError:
     )
     from database.execution_history import subscribe_history_restore
     from database.library_commands import LibraryCommandError
-    from utils.media_capture import CAPTURE_EXTENSIONS, CaptureError, CaptureStore
+    from utils.media_capture import CaptureError, CaptureStore
 
 # Request bodies of the annotation routes are small (a target plus a tag set).
 MAX_BODY_BYTES = 64 * 1024
-MAX_TARGET_PATH = 1024
 CAPTURE_URL = "/prompt_manager/image-annotations/capture/"
 _PREVIEW_CHUNK = 64 * 1024
 
-_TARGET_FIELDS = frozenset({"root_index", "root_scope", "path"})
 _OBSERVE_FIELDS = frozenset({"target"})
 _SAVE_FIELDS = frozenset(
     {
@@ -188,37 +191,6 @@ def _capture_error(exc: CaptureError) -> web.Response:
     return _coded(exc.code, _MESSAGES.get(exc.code) or str(exc))
 
 
-def _target(data: object):
-    """``((index, scope, relative), None)`` for a target, else ``(None, error)``.
-
-    Only the listing identity is accepted: an int root index, the root's
-    64-hex scope and a bounded root-relative POSIX media path, never an
-    absolute, drive-qualified, backslash or ``..`` path. Nothing here touches
-    the filesystem or the database.
-    """
-    if not isinstance(data, dict) or set(data) != _TARGET_FIELDS:
-        return None, bad_request("target must name root_index, root_scope and path")
-    index, scope, relative = data["root_index"], data["root_scope"], data["path"]
-    if type(index) is not int or index < 0:
-        return None, bad_request("Invalid gallery root")
-    if not isinstance(scope, str) or _HEX64.fullmatch(scope) is None:
-        return None, bad_request("Invalid gallery root scope")
-    if not isinstance(relative, str) or not relative or len(relative) > MAX_TARGET_PATH:
-        return None, bad_request("path must be a relative media path")
-    parts = relative.split("/")
-    if (
-        "\0" in relative
-        or "\\" in relative
-        or relative.startswith("/")
-        or _has_traversal(relative)
-        or any(part in ("", ".", "..") or ":" in part for part in parts)
-    ):
-        return None, _forbidden()
-    if os.path.splitext(parts[-1])[1].lower() not in CAPTURE_EXTENSIONS:
-        return None, _forbidden("Only media files can be tagged")
-    return (index, scope, relative), None
-
-
 def _expires_at(seconds: float | None) -> str:
     moment = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
         seconds=max(0.0, seconds or 0.0)

py/api/duplicate_files.py

@@ -589,6 +589,7 @@ class DuplicateFilesMixin:
         spellings = await self._run_in_executor(self._source_spellings)
         removed, failure, keepers = 0, None, {}
         totals = dict.fromkeys(("moved", "merged", "favorites", "tags", "review"), 0)
+        totals["categories"] = 0  # image categories kept on the kept copies (#332)
         for done, target in enumerate(targets, start=1):
             if failure is not None:
                 self._record_target(job, "kept", target, NOT_ATTEMPTED)
@@ -635,6 +636,9 @@ class DuplicateFilesMixin:
             job.log_line(
                 f"Added {_plural(totals['tags'], 'image tag')} to the kept copies"
             )
+        if totals.get("categories"):
+            kept = _plural(totals["categories"], "image category", "image categories")
+            job.log_line(f"Kept {kept} on the kept copies")
         if totals["review"]:
             job.log_line(
                 f"Settled {_plural(totals['review'], 'Needs review item')} "

py/api/duplicate_spellings.py

@@ -18,7 +18,7 @@ Blocking: each lookup stats the file's folders once (cached).
 import os
 from pathlib import Path
 
-from .images import _favorite_relative_key, _favorite_root_key
+from .images import _favorite_relative_key, _favorite_root_key, image_category_key
 
 
 def _identity(path):
@@ -78,6 +78,15 @@ class Spellings:
                 )
         return spelled
 
+    @property
+    def listed_roots(self):
+        """The listed roots (the output sources), in order."""
+        return [root for root, _identity in self._listed]
+
+    def category_key(self, path):
+        """*path*'s image category file key (#332), or None outside every source."""
+        return image_category_key(self.listed_roots, path)
+
     def favorites(self, path):
         """``{listed root: path below it (POSIX)}`` for each listed root holding it."""
         path = os.path.abspath(str(path))
@@ -125,9 +134,23 @@ def copy_move(target, kept_stat, spellings):
         "targets": targets,
         "kept_path": target["keeper_abs"],
         "favorites": favorites,
+        "categories": _category_pairs(target, spellings),
         "removed_locator": target["locator"],
         "kept_locator": target["keeper_locator"],
         "sha256": target["hash"],
         "kept_size": kept_stat.st_size,
         "kept_mtime_ns": kept_stat.st_mtime_ns,
     }
+
+
+def _category_pairs(target, spellings):
+    """``[(removed file key, kept file key)]`` for the image category (#332).
+
+    From the two absolute paths, not the favorites tuples: those follow the
+    listed root, which differs from the file key when sources nest.
+    """
+    removed = spellings.category_key(target["abs"])
+    kept = spellings.category_key(target["keeper_abs"])
+    if removed is None or kept is None:
+        return []
+    return [(removed, kept)]

py/api/image_categories.py

@@ -0,0 +1,361 @@
+"""Image category routes (#332): an image file's own category.
+
+``PUT /prompt_manager/images/category`` sets the own category of up to 500
+listed files; ``DELETE`` removes it (listed or not, so a missing file's
+category can still be removed); ``GET /prompt_manager/images/categories``
+lists the own-category values of files the sources list. Targets are the
+Images listing identity (``{root_index, root_scope, path}``, checked by
+``media_target``) of an output source; LoRA folders are not sources here.
+Instead of ``items`` a request may name ``filter``, the Images listing's own
+query ("Select all N matching", #392), resolved now up to 500 images.
+Each file is stored under its file key (``images.image_category_key``).
+These routes never read or write prompts, recipes or image links.
+"""
+
+import os
+from pathlib import Path
+
+from aiohttp import web
+
+from .image_tag_jobs import MAX_JOB_TARGETS, _parse_targets
+from .images import (
+    _favorite_relative_key,
+    _favorite_root_key,
+    _forbidden,
+    _json_body_or_error,
+    image_category_key,
+)
+from .prompts import bad_request
+
+try:
+    from ...database.image_categories import MAX_CATEGORY_LENGTH, MAX_KEY_LENGTH
+    from ...utils.validators import validate_category
+except ImportError:
+    from database.image_categories import MAX_CATEGORY_LENGTH, MAX_KEY_LENGTH
+    from utils.validators import validate_category
+
+CATEGORY_MEDIA_MESSAGE = "Only media files can have a category"
+_PUT_FIELDS = frozenset({"category"})
+_DELETE_FIELDS = frozenset()
+TOO_MANY = f"Select at most {MAX_JOB_TARGETS} images."
+SOURCE_CHANGED_MESSAGE = "The sources changed. Refresh the images and try again."
+WRITE_FAILED = "The category could not be changed."
+READ_FAILED = "Image categories could not be read."
+
+
+def _coded(status: int, code: str, message: str) -> web.Response:
+    return web.json_response(
+        {"success": False, "code": code, "error": message},
+        status=status,
+        headers={"Cache-Control": "no-store"},
+    )
+
+
+def _category_value(raw) -> tuple[str | None, web.Response | None]:
+    """The stripped category, or a 400 (an empty value is a 400: use DELETE)."""
+    if not isinstance(raw, str):
+        return None, bad_request("category must be a string")
+    value = raw.strip()
+    if not value:
+        return None, bad_request("category must not be empty")
+    if len(value) > MAX_CATEGORY_LENGTH:
+        return None, bad_request(
+            f"Category cannot exceed {MAX_CATEGORY_LENGTH} characters"
+        )
+    try:
+        value.encode("utf-8")
+        validate_category(value)
+    except UnicodeEncodeError:
+        return None, bad_request("category must be valid text")
+    except ValueError as exc:
+        return None, bad_request(str(exc))
+    return value, None
+
+
+class _KeyRefused(Exception):
+    """A target has no storable file key (outside every source, or too long)."""
+
+    def __init__(self, message: str = "Access denied"):
+        super().__init__(message)
+        self.message = message
+
+
+def _file_key(roots, root, relative) -> tuple[str, str]:
+    """The target's file key (blocking: resolves the sources)."""
+    key = image_category_key(roots, Path(root) / relative)
+    if key is None:
+        raise _KeyRefused()
+    if max(len(key[0]), len(key[1])) > MAX_KEY_LENGTH:
+        raise _KeyRefused("Path is too long")
+    return key
+
+
+def _nested_sources(root: Path, sources) -> list[tuple[str, str]]:
+    """``[(relative key prefix, source root key)]`` of sources inside *root*.
+
+    Deepest first, so the first prefix that matches is the file's source.
+    """
+    base = os.path.join(os.path.normcase(os.path.realpath(str(root))), "")
+    nested = []
+    for source in sources:
+        real = os.path.realpath(str(source))
+        if not os.path.normcase(real).startswith(base):
+            continue
+        relative = os.path.relpath(real, os.path.realpath(str(root)))
+        nested.append(
+            (
+                _favorite_relative_key(relative.replace(os.sep, "/")),
+                _favorite_root_key(source),
+            )
+        )
+    nested.sort(key=lambda item: item[0].count("/"), reverse=True)
+    return nested
+
+
+def _source_prefix(relative_key: str, nested) -> str | None:
+    """The prefix of the deepest nested source holding *relative_key*."""
+    for prefix, _source_key in nested:
+        if relative_key.startswith(prefix + "/"):
+            return prefix
+    return None
+
+
+class ImageCategoryRoutesMixin:
+    """Routes for images' own categories (#332)."""
+
+    def _register_image_category_routes(self, routes):
+        # Fixed paths beside DELETE /prompt_manager/images/{image_id}, as the
+        # favorite routes are. Registered after it, they still win: aiohttp
+        # (3.10+) looks fixed paths up before dynamic resources
+        # (test_delete_category_never_reaches_delete_image).
+        @routes.put("/prompt_manager/images/category")
+        async def set_image_category_route(request):
+            return await self.set_image_category(request)
+
+        @routes.delete("/prompt_manager/images/category")
+        async def clear_image_category_route(request):
+            return await self.clear_image_category(request)
+
+        @routes.get("/prompt_manager/images/categories")
+        async def list_image_categories_route(request):
+            return await self.list_image_categories(request)
+
+    # -- reading ---------------------------------------------------------------
+
+    def _own_category_map(self, sources, listings) -> dict:
+        """``{root / listed path: own category}`` for the listed files (blocking).
+
+        One primary-key range read per source; each stored row is matched to
+        the listing of every selected root that holds its source, so the work
+        follows the stored rows, not the files. A file under a nested source
+        is keyed by the deepest one (``image_category_key``). A row whose file
+        is gone matches nothing. A failed read raises.
+        """
+        sources = list(sources)
+        rows = self.db.own_categories_of(_favorite_root_key(s) for s in sources)
+        if not rows:
+            return {}
+        by_source: dict[str, dict[str, str]] = {}
+        for (root_key, relative), category in rows.items():
+            by_source.setdefault(root_key, {})[relative] = category
+        found = {}
+        for _index, root, files in listings:
+            listing = self._favorite_listing(root, files)
+            root_key = _favorite_root_key(root)
+            nested = [
+                item for item in _nested_sources(root, sources) if item[1] != root_key
+            ]
+            for relative, category in by_source.get(root_key, {}).items():
+                listed = listing.get(relative)
+                if listed is not None and _source_prefix(relative, nested) is None:
+                    found[root / listed] = category
+            for prefix, source_key in nested:
+                for relative, category in by_source.get(source_key, {}).items():
+                    full = f"{prefix}/{relative}"
+                    listed = listing.get(full)
+                    if listed is not None and _source_prefix(full, nested) == prefix:
+                        found[root / listed] = category
+        return found
+
+    async def _source_listings(self, roots) -> list[tuple]:
+        return [
+            (index, root, await self._get_gallery_files(root))
+            for index, root in enumerate(roots)
+        ]
+
+    async def list_image_categories(self, request):
+        """GET: the own-category values of listed files, sorted."""
+        try:
+            roots = list(self._get_all_output_dirs())
+            listings = await self._source_listings(roots)
+            found = await self._run_in_executor(self._own_category_map, roots, listings)
+        except Exception:
+            self.logger.exception("Image categories read failed")
+            return _coded(500, "categories-failed", READ_FAILED)
+        return web.json_response(
+            {"success": True, "categories": sorted(set(found.values()))},
+            headers={"Cache-Control": "no-store"},
+        )
+
+    # -- writing ---------------------------------------------------------------
+
+    async def _filter_targets(self, raw):
+        """``(targets, None)`` for an Images listing filter, else ``(None, error)``.
+
+        "Select all N matching" (#392) sends the listing's own query; it is
+        resolved now, up to 500 images (a 400 ``too-many-images`` above).
+        """
+        from .images import ListingSourceChanged, parse_listing_filter
+
+        try:
+            selection = await self.resolve_listing_filter(
+                parse_listing_filter(raw), MAX_JOB_TARGETS
+            )
+        except ListingSourceChanged:
+            return None, _coded(409, "source-changed", SOURCE_CHANGED_MESSAGE)
+        except ValueError as exc:
+            return None, bad_request(str(exc))
+        if selection.truncated:
+            return None, _coded(400, "too-many-images", TOO_MANY)
+        return tuple(selection.targets), None
+
+    async def _category_request(self, request, fields: frozenset):
+        """``(data, targets, roots, None)`` or ``(None, None, None, error)``.
+
+        The images are ``items`` (1 to 500 targets) or ``filter`` (all images
+        an Images listing filter admits), besides *fields*.
+        """
+        data, error = await _json_body_or_error(request)
+        if error is not None:
+            return None, None, None, error
+        names = set(data) - fields
+        if set(data) & fields != fields or names not in ({"items"}, {"filter"}):
+            return (
+                None,
+                None,
+                None,
+                bad_request(
+                    "Request must contain "
+                    + ", ".join(sorted(fields))
+                    + (" and " if fields else "")
+                    + "either items or filter"
+                ),
+            )
+        if "filter" in data:
+            targets, error = await self._filter_targets(data["filter"])
+            if error is not None:
+                return None, None, None, error
+            return data, targets, list(self._get_all_output_dirs()), None
+        items = data["items"]
+        if not isinstance(items, list) or not 1 <= len(items) <= MAX_JOB_TARGETS:
+            return (
+                None,
+                None,
+                None,
+                bad_request(f"items must list 1 to {MAX_JOB_TARGETS} images"),
+            )
+        targets, error = _parse_targets(items, media_message=CATEGORY_MEDIA_MESSAGE)
+        if error is not None:
+            return None, None, None, error
+        try:
+            for _index, _scope, relative in targets:
+                relative.encode("utf-8")
+        except UnicodeEncodeError:
+            return None, None, None, bad_request("path must be valid text")
+        roots = list(self._get_all_output_dirs())
+        for index, scope in dict.fromkeys((t[0], t[1]) for t in targets):
+            if index >= len(roots):
+                return None, None, None, bad_request("Invalid gallery root")
+            error = self._gallery_scope_error(
+                {"root_scope": scope, "root": str(index)}, roots
+            )
+            if error is not None:
+                if error.status == 409:
+                    error = _coded(409, "source-changed", SOURCE_CHANGED_MESSAGE)
+                return None, None, None, error
+        return data, targets, roots, None
+
+    def _listed_file_keys(self, roots, targets, files) -> tuple[list, list]:
+        """``(listed keys, unlisted keys)`` of the targets (blocking).
+
+        A target counts as listed when its source's listing holds it; its key
+        then follows the listed spelling. Targets folding to one key count
+        once, and a key that is listed through any target is listed.
+        """
+        listings, listed, missing = {}, {}, {}
+        for index, _scope, relative in targets:
+            root = roots[index]
+            if index not in listings:
+                listings[index] = self._favorite_listing(root, files[index])
+            spelled = listings[index].get(_favorite_relative_key(relative))
+            key = _file_key(roots, root, spelled or relative)
+            (missing if spelled is None else listed)[key] = True
+        return list(listed), [key for key in missing if key not in listed]
+
+    async def set_image_category(self, request):
+        """PUT ``{items, category}``: set the own category of listed files."""
+        data, targets, roots, error = await self._category_request(request, _PUT_FIELDS)
+        if error is not None:
+            return error
+        category, error = _category_value(data["category"])
+        if error is not None:
+            return error
+        try:
+            files = {}
+            for index in dict.fromkeys(t[0] for t in targets):
+                files[index] = await self._get_gallery_files(roots[index])
+            listed_keys, missing_keys = await self._run_in_executor(
+                self._listed_file_keys, roots, targets, files
+            )
+        except _KeyRefused as refused:
+            return _forbidden(refused.message)
+        except Exception:
+            self.logger.exception("Image category update failed")
+            return _coded(500, "category-failed", WRITE_FAILED)
+        try:
+            if not listed_keys:
+                return _coded(
+                    404,
+                    "files-not-found",
+                    "None of the selected images could be found.",
+                )
+            counts = await self._run_in_executor(
+                self.db.set_image_categories, listed_keys, category
+            )
+        except Exception:
+            self.logger.exception("Image category update failed")
+            return _coded(500, "category-failed", WRITE_FAILED)
+        return web.json_response(
+            {
+                "success": True,
+                "category": category,
+                "updated": counts["updated"],
+                "unchanged": counts["unchanged"],
+                "not_found": len(missing_keys),
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+
+    async def clear_image_category(self, request):
+        """DELETE ``{items}``: remove the own category, listed or not."""
+        _data, targets, roots, error = await self._category_request(
+            request, _DELETE_FIELDS
+        )
+        if error is not None:
+            return error
+        try:
+            keys = await self._run_in_executor(
+                lambda: list(
+                    dict.fromkeys(_file_key(roots, roots[i], r) for i, _s, r in targets)
+                )
+            )
+            cleared = await self._run_in_executor(self.db.clear_image_categories, keys)
+        except _KeyRefused as refused:
+            return _forbidden(refused.message)
+        except Exception:
+            self.logger.exception("Image category removal failed")
+            return _coded(500, "category-failed", WRITE_FAILED)
+        return web.json_response(
+            {"success": True, "cleared": cleared},
+            headers={"Cache-Control": "no-store"},
+        )

py/api/image_tag_jobs.py

@@ -193,13 +193,17 @@ class _Spec:
         return scope
 
 
-def _parse_targets(raw):
-    """Unique validated targets in request order, or ``(None, error)``."""
+def _parse_targets(raw, **target_options):
+    """Unique validated targets in request order, or ``(None, error)``.
+
+    ``target_options`` reach the target check (the image category routes
+    pass their own ``media_message``, #332).
+    """
     if not isinstance(raw, list) or not 1 <= len(raw) <= MAX_JOB_TARGETS:
         return None, bad_request(f"targets must list 1 to {MAX_JOB_TARGETS} images")
     seen, targets = set(), []
     for item in raw:
-        target, error = _target(item)
+        target, error = _target(item, **target_options)
         if error is not None:
             return None, error
         if target not in seen:

py/api/images.py

@@ -802,6 +802,53 @@ def _favorite_root_key(root) -> str:
     return _favorite_relative_key(os.path.realpath(str(root)))
 
 
+def _source_depth(source: str) -> int:
+    return len([part for part in re.split(r"[\\/]+", source) if part])
+
+
+def deepest_source(sources, absolute_path):
+    """``(source, POSIX path below it)`` of the deepest source holding the path.
+
+    A source matches by its real path or by its spelling as configured (a
+    detected output folder can be a symlink), and the deepest real path wins.
+    The path is compared lexically after ``normpath`` and never resolved, so a
+    symlinked subfolder keeps its listed spelling. Containment folds case
+    where the OS does. *source* is the real path. None when no source
+    contains the path.
+    """
+    path = os.path.normpath(str(absolute_path))
+    folded = os.path.normcase(path)
+    best = None
+    for source in sources:
+        real = os.path.normpath(os.path.realpath(str(source)))
+        for spelling in dict.fromkeys((real, os.path.normpath(str(source)))):
+            prefix = os.path.join(os.path.normcase(spelling), "")
+            if not folded.startswith(prefix):
+                continue
+            if best is None or _source_depth(real) > _source_depth(best[0]):
+                best = (real, spelling)
+            break
+    if best is None:
+        return None
+    relative = os.path.relpath(path, best[1])
+    return best[0], relative.replace(os.sep, "/")
+
+
+def image_category_key(sources, absolute_path):
+    """The file key of an image's own category (#332), or None.
+
+    ``(root key, relative key)``: the deepest output source that contains the
+    file and the path below it, keyed with the favorites helpers. Without
+    nested sources it equals the file's favorite key. A path no source
+    contains has no key (and so no own category).
+    """
+    found = deepest_source(sources, absolute_path)
+    if found is None:
+        return None
+    source, relative = found
+    return _favorite_root_key(source), _favorite_relative_key(relative)
+
+
 def _name_suffix(name):
     """``Path(name).suffix`` as Python 3.10-3.13 compute it.
 

py/api/media_target.py

@@ -0,0 +1,54 @@
+"""The listing identity of one media file, as a request names it.
+
+Shared by the image-annotation routes and the collection routes (#331): a
+target is ``{root_index, root_scope, path}``, checked here without touching
+the filesystem or the database. The scope is compared with the current roots
+by the caller.
+"""
+
+import os
+import re
+
+from .images import _forbidden, _has_traversal
+from .prompts import bad_request
+
+try:
+    from ...utils.media_capture import CAPTURE_EXTENSIONS
+except ImportError:
+    from utils.media_capture import CAPTURE_EXTENSIONS
+
+MAX_TARGET_PATH = 1024
+TARGET_FIELDS = frozenset({"root_index", "root_scope", "path"})
+_HEX64 = re.compile(r"[0-9a-f]{64}")
+TAGGING_MEDIA_MESSAGE = "Only media files can be tagged"
+
+
+def media_target(data: object, *, media_message: str = TAGGING_MEDIA_MESSAGE):
+    """``((index, scope, relative), None)`` for a target, else ``(None, error)``.
+
+    Only the listing identity is accepted: an int root index, the root's
+    64-hex scope and a bounded root-relative POSIX media path, never an
+    absolute, drive-qualified, backslash or ``..`` path (a 400 for a malformed
+    shape, a 403 for an unsafe or non-media path, carrying *media_message*).
+    """
+    if not isinstance(data, dict) or set(data) != TARGET_FIELDS:
+        return None, bad_request("target must name root_index, root_scope and path")
+    index, scope, relative = data["root_index"], data["root_scope"], data["path"]
+    if type(index) is not int or index < 0:
+        return None, bad_request("Invalid gallery root")
+    if not isinstance(scope, str) or _HEX64.fullmatch(scope) is None:
+        return None, bad_request("Invalid gallery root scope")
+    if not isinstance(relative, str) or not relative or len(relative) > MAX_TARGET_PATH:
+        return None, bad_request("path must be a relative media path")
+    parts = relative.split("/")
+    if (
+        "\0" in relative
+        or "\\" in relative
+        or relative.startswith("/")
+        or _has_traversal(relative)
+        or any(part in ("", ".", "..") or ":" in part for part in parts)
+    ):
+        return None, _forbidden()
+    if os.path.splitext(parts[-1])[1].lower() not in CAPTURE_EXTENSIONS:
+        return None, _forbidden(media_message)
+    return (index, scope, relative), None

tests/browser/server.py

@@ -268,6 +268,7 @@ def create_app(root: Path) -> web.Application:
                 "media_image_tags",
                 "media_annotation_bytes",
                 "media_annotation_files",
+                "image_categories",
                 "image_prompt_pairs",
                 "prompt_pair_tags",
                 "prompt_pairs",

tests/browser/workspace_concept_fixture.py

@@ -181,6 +181,7 @@ _TABLES = (
     "media_image_tags",
     "media_annotation_bytes",
     "media_annotation_files",
+    "image_categories",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_library_editor_fixture.py

@@ -33,6 +33,7 @@ _TABLES = (
     "media_image_tags",
     "media_annotation_bytes",
     "media_annotation_files",
+    "image_categories",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_review_fixture.py

@@ -85,6 +85,7 @@ _TABLES = (
     "media_image_tags",
     "media_annotation_bytes",
     "media_annotation_files",
+    "image_categories",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_tagging_fixture.py

@@ -51,6 +51,7 @@ _TABLES = (
     "media_image_tags",
     "media_annotation_bytes",
     "media_annotation_files",
+    "image_categories",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/browser/workspace_visual_fixture.py

@@ -119,6 +119,7 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None:
             "media_image_tags",
             "media_annotation_bytes",
             "media_annotation_files",
+            "image_categories",
             "image_prompt_pairs",
             "prompt_pair_tags",
             "prompt_pairs",

tests/browser/workspace_vocabulary_fixture.py

@@ -86,6 +86,7 @@ _TABLES = (
     "media_image_tags",
     "media_annotation_bytes",
     "media_annotation_files",
+    "image_categories",
     "image_prompt_pairs",
     "prompt_pair_tags",
     "prompt_pairs",

tests/comfyui/execution-text-link.spec.mjs

@@ -11,7 +11,7 @@ import { installHostHooks } from './host-fixture.mjs';
 
 // The worker's owned host must be up before the shared host hooks open the page.
 test.beforeEach(async ({ execution }) => { expect(execution.databasePath).toBeTruthy(); });
-installHostHooks(test, ['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerPair', 'PMHistoryInputs']);
+installHostHooks(test, ['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerPair', 'PMHistoryInputs', 'PMCycledText']);
 
 // `primitive`: a core string node the queue-time resolver follows. `legacy`: the frontend's virtual Primitive node,
 // whose value is queued as literal text. `preview`: PrimitiveString -> PreviewAny, the shape of the maintainer's
@@ -37,6 +37,7 @@ function build(page, { kind, positive, negative, prefix }) {
                 primitive.widgets[0].value = side.text;
                 return;
             }
+            if (side.via === 'cycled') { connect(add('PMCycledText', 0, y), 0, prompt, name); return; }
             const text = add('PrimitiveString', 0, y);
             text.widgets.find(widget => widget.name === 'value').value = side.text;
             let source = text;
@@ -75,9 +76,9 @@ function build(page, { kind, positive, negative, prefix }) {
     }, { kind, positive, negative, prefix });
 }
 
-async function run(execution, page, testCase) {
+async function run(execution, page, testCase, built) {
     const prefix = `textlink/${randomUUID()}`;
-    const { output, workflow, promptNode } = await build(page, { ...testCase, prefix });
+    const { output, workflow, promptNode } = built ?? await build(page, { ...testCase, prefix });
     const id = randomUUID();
     const response = await execution.request.post('/prompt', { data: { prompt: output, prompt_id: id, extra_data: { extra_pnginfo: { workflow } } } });
     expect(response.ok(), await response.text()).toBe(true);
@@ -193,3 +194,164 @@ for (const [label, linked, negativeVia] of [['positive', ['text'], 'primitive'],
         await expectImagesLinked(execution, evidence, sides.text.via === 'preview');
     });
 }
+
+// #347: a cached PromptManager node counts every job, as an executing one does. Jobs after the first change only
+// the sampler seed, so ComfyUI serves the PromptManager node (and its PreviewAny source) from its output cache.
+const COUNTS = `import json, sqlite3, sys, pathlib
+db = sqlite3.connect(pathlib.Path(sys.argv[1]).resolve().as_uri() + "?mode=ro", uri=True)
+one = lambda sql, key: (db.execute(sql, (int(key),)).fetchone() or [None])[0] if key != "-" else None
+print(json.dumps({"positive": one("SELECT run_count FROM prompts WHERE id=?", sys.argv[2]),
+    "negative": one("SELECT run_count FROM prompts WHERE id=?", sys.argv[3]),
+    "pair": one("SELECT run_count FROM prompt_pairs WHERE id=?", sys.argv[4])}))`;
+async function runCounts(execution, context) {
+    const key = value => value == null ? '-' : String(value);
+    const { stdout } = await promisify(execFile)(process.env.PM_TEST_PYTHON || 'python',
+        ['-c', COUNTS, execution.databasePath, key(context.positive_prompt_id), key(context.negative_prompt_id), key(context.pair_id)], { timeout: 5000 });
+    return JSON.parse(stdout);
+}
+
+async function setCache(execution, mode) {
+    expect((await execution.request.post('/pm_execution_fixture/cache', { data: { mode } })).ok()).toBe(true);
+}
+
+// Queue an already built workflow again with another sampler seed; returns the job's history item and cached nodes.
+async function queueAgain(execution, first, { seed, edit, targets, status = 'success' } = {}) {
+    const output = structuredClone(first.output);
+    const sampler = Object.values(output).find(node => node.class_type === 'KSampler');
+    sampler.inputs.seed = seed;
+    edit?.(output);
+    const id = randomUUID();
+    const data = { prompt: output, prompt_id: id, extra_data: { extra_pnginfo: { workflow: first.workflow } } };
+    if (targets) data.partial_execution_targets = targets;
+    const response = await execution.request.post('/prompt', { data });
+    expect(response.ok(), await response.text()).toBe(true);
+    let item;
+    await expect.poll(async () => {
+        item = (await (await execution.request.get(`/history/${id}`)).json())[id];
+        return item?.status?.status_str ?? null;
+    }, { timeout: 20000 }).toBe(status);
+    const cached = item.status.messages.find(([event]) => event === 'execution_cached')?.[1]?.nodes ?? [];
+    return { id, item, cached };
+}
+
+// The recorded context of one job, once its capture reached SQLite.
+async function capturedContext(execution, id) {
+    let result;
+    await expect.poll(async () => {
+        result = await readExecutionRuns(execution.databasePath, id);
+        const row = result.rows[0];
+        return row ? result.captures[row.run_id].state : null;
+    }, { timeout: 10000 }).toMatch(/^(complete|partial)$/);
+    return result.captures[result.rows[0].run_id].contexts[0];
+}
+
+async function firstRun(execution, page, testCase) {
+    const prefix = `textlink/${randomUUID()}`;
+    const built = await build(page, { ...testCase, prefix });
+    const evidence = await run(execution, page, testCase, built);
+    return { ...evidence, workflow: built.workflow };
+}
+
+for (const kind of ['PromptManager', 'PromptManagerFocused', 'PromptManagerSplit', 'PromptManagerText', 'PromptManagerPair']) {
+    test(`${kind} counts each cached re-run of text linked from PreviewAny (#347)`, async ({ execution, page }) => {
+        test.setTimeout(90000);
+        await setCache(execution, 'RAM_PRESSURE');
+        const pair = kind === 'PromptManagerPair';
+        const testCase = { kind, positive: { typed: 'Typed under the link', text: unique(`Cached ${kind}`), via: 'preview' },
+            negative: pair ? { typed: unique('Typed negative') } : undefined };
+        const first = await firstRun(execution, page, testCase);
+        for (const seed of [2, 3]) {
+            const again = await queueAgain(execution, first, { seed });
+            expect(again.cached, 'ComfyUI served the PromptManager node from its cache').toContain(first.promptNode);
+        }
+        const counts = await runCounts(execution, first.context);
+        expect(counts.positive).toBe(3);
+        if (pair) expect(counts).toMatchObject({ negative: 0, pair: 3 });
+    });
+}
+
+test.describe('cached runs in one renderer (#347)', () => {
+    // Cache behavior does not depend on the renderer: one project runs these.
+    test.beforeEach(({}, testInfo) => { test.skip(testInfo.project.name !== 'classic-start', 'covered in classic-start'); });
+    const linked = () => ({ kind: 'PromptManager', positive: { typed: 'Typed under the link', text: unique('Cached'), via: 'preview' } });
+
+    for (const mode of ['CLASSIC', 'LRU', 'RAM_PRESSURE', 'NONE']) {
+        test(`counts 3 jobs under the ${mode} cache`, async ({ execution, page }) => {
+            test.setTimeout(90000);
+            await setCache(execution, mode);
+            try {
+                const first = await firstRun(execution, page, linked());
+                for (const seed of [2, 3]) {
+                    const again = await queueAgain(execution, first, { seed });
+                    if (mode === 'NONE') expect(again.cached).not.toContain(first.promptNode);
+                    else expect(again.cached).toContain(first.promptNode);
+                }
+                expect((await runCounts(execution, first.context)).positive).toBe(3);
+            } finally { await setCache(execution, 'RAM_PRESSURE'); }
+        });
+    }
+
+    test('alternating upstream values A, B, A under LRU each count their own prompt', async ({ execution, page }) => {
+        test.setTimeout(90000);
+        await setCache(execution, 'LRU');
+        try {
+            const [a, b] = [unique('Cycled A'), unique('Cycled B')];
+            const cycle = async value => expect((await execution.request.post('/pm_execution_fixture/cycled', { data: { value } })).ok()).toBe(true);
+            await cycle(a);
+            const first = await firstRun(execution, page, { kind: 'PromptManager', positive: { typed: 'Typed under the link', text: a, via: 'cycled' } });
+            expect(first.context.positive.raw).toBe(a);
+            await cycle(b);
+            const second = await queueAgain(execution, first, { seed: 2 });
+            expect(second.cached).not.toContain(first.promptNode);
+            await cycle(a);
+            const third = await queueAgain(execution, first, { seed: 3 });
+            expect(third.cached, 'LRU kept A').toContain(first.promptNode);
+            expect((await runCounts(execution, first.context)).positive).toBe(2);
+            const bContext = await capturedContext(execution, second.id);
+            expect(bContext.positive.raw).toBe(b);
+            expect((await runCounts(execution, bContext)).positive).toBe(1);
+        } finally { await setCache(execution, 'RAM_PRESSURE'); }
+    });
+
+    test('typed text served from the cache still counts once per job', async ({ execution, page }) => {
+        test.setTimeout(90000);
+        await setCache(execution, 'RAM_PRESSURE');
+        const first = await firstRun(execution, page, { kind: 'PromptManager', positive: { typed: unique('Typed cached') } });
+        for (const seed of [2, 3]) expect((await queueAgain(execution, first, { seed })).cached).toContain(first.promptNode);
+        expect((await runCounts(execution, first.context)).positive).toBe(3);
+    });
+
+    test('a job failing downstream of a cached node counts it once; a job that does not need it, never', async ({ execution, page }) => {
+        test.setTimeout(90000);
+        await setCache(execution, 'RAM_PRESSURE');
+        const first = await firstRun(execution, page, linked());
+        const failing = await queueAgain(execution, first, { seed: 2, status: 'error', edit: output => {
+            output.pm_failure = { class_type: 'PMHistoryOutput', inputs: { value: [Object.keys(output).find(id => output[id].class_type === 'KSampler'), 0], mode: 'failure' } };
+        } });
+        expect(failing.cached).toContain(first.promptNode);
+        expect((await runCounts(execution, first.context)).positive).toBe(2);
+        // "Run selected output": only an unrelated output node; ComfyUI still reports the cached PromptManager node
+        const unrelated = await queueAgain(execution, first, { seed: 2, targets: ['pm_unrelated'], edit: output => {
+            output.pm_unrelated = { class_type: 'PMHistoryCached', inputs: { value: unique('unrelated') } };
+        } });
+        expect(unrelated.cached).toContain(first.promptNode);
+        expect((await runCounts(execution, first.context)).positive).toBe(2);
+    });
+
+    test('a cached node rewired to the negative input does not count the positive prompt', async ({ execution, page }) => {
+        test.setTimeout(90000);
+        await setCache(execution, 'RAM_PRESSURE');
+        const first = await firstRun(execution, page, linked());
+        const rewired = await queueAgain(execution, first, { seed: 2, edit: output => {
+            const sampler = Object.values(output).find(node => node.class_type === 'KSampler');
+            [sampler.inputs.positive, sampler.inputs.negative] = [sampler.inputs.negative, sampler.inputs.positive];
+        } });
+        expect(rewired.cached).toContain(first.promptNode);
+        expect((await runCounts(execution, first.context)).positive).toBe(1);
+        // The history capture of a cached job is unchanged: effective text from the cache, no stored-prompt hint
+        const cachedContext = await capturedContext(execution, (await queueAgain(execution, first, { seed: 3 })).id);
+        expect(cachedContext.positive.effective).toBe(first.context.positive.effective);
+        expect(cachedContext.positive_prompt_id).toBeNull();
+        expect((await runCounts(execution, first.context)).positive).toBe(2);
+    });
+});

tests/comfyui/fixtures/ledger_nodes.py

@@ -201,6 +201,39 @@ NODE_CLASS_MAPPINGS.update(
     {"PMHistoryImages": PMHistoryImages, "PMHistoryImageOutput": PMHistoryImageOutput}
 )
 
+# #347: text whose value (and cache signature) follows a value the test sets, so
+# one queued graph can serve A, then B, then A again (each cached under LRU).
+_CYCLED = {"value": "cycled"}
+
+
+class PMCycledText:
+    @classmethod
+    def INPUT_TYPES(cls) -> dict:
+        return {"required": {}}
+
+    RETURN_TYPES = ("STRING",)
+    FUNCTION = "execute"
+    CATEGORY = "testing/execution-history"
+
+    @classmethod
+    def IS_CHANGED(cls, **kwargs: object) -> str:
+        return _CYCLED["value"]
+
+    def execute(self) -> tuple:
+        return (_CYCLED["value"],)
+
+
+NODE_CLASS_MAPPINGS["PMCycledText"] = PMCycledText
+
+
+@PromptServer.instance.routes.post("/pm_execution_fixture/cycled")
+async def cycled_text(request: web.Request) -> web.Response:
+    value = (await request.json())["value"]
+    if type(value) is not str:
+        raise web.HTTPBadRequest()
+    _CYCLED["value"] = value
+    return web.json_response({"value": value})
+
 
 @PromptServer.instance.routes.post("/pm_execution_fixture/metadata")
 async def metadata(request: web.Request) -> web.Response:

tests/comfyui/image-attribution.spec.mjs

@@ -140,3 +140,26 @@ test('an output saved without metadata links once its recorded run names a store
         expect((await request.post('/pm_execution_fixture/metadata', { data: { disabled: false } })).ok()).toBe(true);
     }
 });
+
+// #332 on the real host: the migration ran at startup and the image category routes
+// set, list and remove an own category on a generated output.
+test('an output gets its own image category and loses it again', async ({ execution }, testInfo) => {
+    const request = execution.request;
+    const filename = await saved(request, graph({ text: `category host ${randomUUID()}`, prefix: 'pm332/category' }));
+    let image;
+    await expect.poll(async () => {
+        const listing = await (await request.get('/prompt_manager/images/output?mode=workspace&q=pm332&rescan=1')).json();
+        image = (listing.images || []).find(entry => entry.filename === filename || entry.path?.endsWith(`/${filename}`));
+        return Boolean(image);
+    }, { timeout: 20000, message: `${filename} is listed by Images` }).toBe(true);
+    const items = [{ root_index: image.root_index, root_scope: image.root_scope, path: image.path }];
+    const value = `PM332 host ${testInfo.project.name}`;
+    const set = await request.put('/prompt_manager/images/category', { data: { items, category: value } });
+    expect(set.status()).toBe(200);
+    expect(await set.json()).toEqual({ success: true, category: value, updated: 1, unchanged: 0, not_found: 0 });
+    expect((await (await request.get('/prompt_manager/images/categories')).json()).categories).toContain(value);
+    const cleared = await request.delete('/prompt_manager/images/category', { data: { items } });
+    expect(cleared.status()).toBe(200);
+    expect(await cleared.json()).toEqual({ success: true, cleared: 1 });
+    expect((await (await request.get('/prompt_manager/images/categories')).json()).categories).not.toContain(value);
+});

tests/fixtures/legacy_contract_snapshot.json

@@ -2,6 +2,7 @@
   "routes": [
     "DELETE /prompt_manager/delete/{prompt_id}",
     "DELETE /prompt_manager/image-annotations/capture/{token}",
+    "DELETE /prompt_manager/images/category",
     "DELETE /prompt_manager/images/favorite",
     "DELETE /prompt_manager/images/{image_id}",
     "DELETE /prompt_manager/pairs/{pair_id}",
@@ -26,6 +27,7 @@
     "GET /prompt_manager/history/runs/{run_id}/outputs",
     "GET /prompt_manager/image-annotations/capture/{token}",
     "GET /prompt_manager/images/all",
+    "GET /prompt_manager/images/categories",
     "GET /prompt_manager/images/inspect/{image_path}",
     "GET /prompt_manager/images/output",
     "GET /prompt_manager/images/prompt/{image_path}",
@@ -156,6 +158,7 @@
     "POST /prompt_manager/vocabulary/tags",
     "POST /prompt_manager/vocabulary/tags/{tag_id}/delete",
     "PUT /prompt_manager/image-annotations/tags",
+    "PUT /prompt_manager/images/category",
     "PUT /prompt_manager/images/favorite",
     "PUT /prompt_manager/prompts/{prompt_id}",
     "PUT /prompt_manager/prompts/{prompt_id}/rating",

tests/test_api_image_categories.py

@@ -0,0 +1,375 @@
+"""Image category routes (#332): PUT/DELETE /images/category, GET /images/categories."""
+
+import json
+import ntpath
+import sqlite3
+from unittest.mock import patch
+
+try:
+    from test_api_images import ImageAPITestCase, make_png
+    from open_handles import assert_closed
+    from test_image_categories import category_checksum
+except ImportError:
+    from tests.test_api_images import ImageAPITestCase, make_png
+    from tests.open_handles import assert_closed
+    from tests.test_image_categories import category_checksum
+
+from database.image_categories import MAX_KEY_LENGTH, TABLE
+from py.api.images import _favorite_root_key, image_category_key
+
+CATEGORY = "/prompt_manager/images/category"
+CATEGORIES = "/prompt_manager/images/categories"
+
+
+class ImageCategoryTestCase(ImageAPITestCase):
+    async def tearDownAsync(self):
+        self.api.db.close_all()
+        assert_closed(self, self._temp_db.name, self._output_tmp.name)
+        await super().tearDownAsync()
+
+    def target(self, path, index=0):
+        roots = self.api._get_all_output_dirs()
+        return {
+            "root_index": index,
+            "root_scope": self.api._gallery_root_scope(roots[index], index),
+            "path": path,
+        }
+
+    async def send(self, method, body, status=200):
+        response = await self.client.request(method, CATEGORY, json=body)
+        text = await response.text()
+        self.assertEqual(response.status, status, text)
+        return json.loads(text)
+
+    async def put(self, paths, category, status=200):
+        items = [p if isinstance(p, dict) else self.target(p) for p in paths]
+        return await self.send("PUT", {"items": items, "category": category}, status)
+
+    async def delete(self, paths, status=200):
+        items = [p if isinstance(p, dict) else self.target(p) for p in paths]
+        return await self.send("DELETE", {"items": items}, status)
+
+    async def listed(self, status=200):
+        response = await self.client.get(CATEGORIES)
+        text = await response.text()
+        self.assertEqual(response.status, status, text)
+        return json.loads(text)
+
+    def rows(self):
+        return [
+            tuple(row)
+            for row in self.api.db.model.get_connection().execute(
+                f"SELECT root_key, relative_path, category FROM {TABLE} "
+                "ORDER BY root_key, relative_path"
+            )
+        ]
+
+    def key(self, relative, root=None):
+        roots = self.api._get_all_output_dirs()
+        return image_category_key(roots, (root or self.output_dir) / relative)
+
+
+class TestSetAndClear(ImageCategoryTestCase):
+    async def test_put_sets_listed_and_counts_missing(self):
+        make_png(self.output_dir / "a.png")
+        make_png(self.output_dir / "sub" / "b.png")
+        data = await self.put(["a.png", "sub/b.png", "missing.png"], "  Portraits ")
+        self.assertEqual(
+            data,
+            {
+                "success": True,
+                "category": "Portraits",
+                "updated": 2,
+                "unchanged": 0,
+                "not_found": 1,
+            },
+        )
+        root_key = _favorite_root_key(self.output_dir)
+        self.assertEqual(
+            self.rows(),
+            [(root_key, "a.png", "Portraits"), (root_key, "sub/b.png", "Portraits")],
+        )
+        data = await self.put(["a.png", "sub/b.png"], "Portraits")
+        self.assertEqual((data["updated"], data["unchanged"]), (0, 2))
+        data = await self.put(["a.png"], "Scans")
+        self.assertEqual((data["updated"], data["unchanged"]), (1, 0))
+        self.assertEqual(self.api.db.image_category(*self.key("a.png")), "Scans")
+
+    async def test_put_with_no_listed_file_is_404_and_writes_nothing(self):
+        data = await self.put(["missing.png", "gone/x.png"], "Scans", status=404)
+        self.assertEqual(data["code"], "files-not-found")
+        self.assertEqual(self.rows(), [])
+
+    async def test_duplicate_targets_are_written_and_counted_once(self):
+        make_png(self.output_dir / "a.png")
+        item = self.target("a.png")
+        data = await self.put([item, dict(item)], "Scans")
+        self.assertEqual((data["updated"], data["not_found"]), (1, 0))
+
+    async def test_delete_clears_listed_and_missing_files(self):
+        make_png(self.output_dir / "a.png")
+        await self.put(["a.png"], "Scans")
+        self.api.db.set_image_categories([self.key("gone.png")], "Scans")
+        data = await self.delete(["a.png", "gone.png", "never.png"])
+        self.assertEqual(data, {"success": True, "cleared": 2})
+        self.assertEqual(self.rows(), [])
+        self.assertEqual((await self.delete(["a.png"]))["cleared"], 0)
+
+    async def test_put_never_writes_prompts(self):
+        make_png(self.output_dir / "a.png")
+        make_png(self.output_dir / "b.png")
+        positive = self._save_prompt("castle at dusk", category="Landscapes")
+        self.api.db.link_image_to_prompt(positive, str(self.output_dir / "a.png"))
+        negative = self._save_prompt("blurry", category="Negatives")
+        self.api.db.save_prompt_pair("castle at dusk", "blurry", category="Recipes")
+        conn = self.api.db.model.get_connection()
+        before = category_checksum(conn)
+        links = conn.execute("SELECT * FROM generated_images").fetchall()
+        await self.put(["a.png", "b.png"], "Portraits")
+        await self.listed()
+        await self.delete(["a.png"])
+        self.assertEqual(category_checksum(conn), before)
+        self.assertEqual(
+            conn.execute("SELECT * FROM generated_images").fetchall(), links
+        )
+        self.assertIsNotNone(negative)
+
+    async def test_two_sources_in_one_request(self):
+        second = self.comfy_dir / "renders"
+        make_png(self.output_dir / "a.png")
+        make_png(second / "a.png")
+        self.api._get_all_output_dirs = lambda: [self.output_dir, second]
+        data = await self.put([self.target("a.png", 0), self.target("a.png", 1)], "X")
+        self.assertEqual(data["updated"], 2)
+        self.assertEqual(
+            {row[0] for row in self.rows()},
+            {_favorite_root_key(self.output_dir), _favorite_root_key(second)},
+        )
+        self.assertEqual((await self.delete([self.target("a.png", 1)]))["cleared"], 1)
+
+    async def test_nested_sources_key_by_the_deepest(self):
+        inner = self.output_dir / "nested"
+        make_png(inner / "a.png")
+        self.api._get_all_output_dirs = lambda: [self.output_dir, inner]
+        # Named through the outer source, stored under the inner one.
+        data = await self.put([self.target("nested/a.png", 0)], "Scans")
+        self.assertEqual(data["updated"], 1)
+        self.assertEqual(self.rows(), [(_favorite_root_key(inner), "a.png", "Scans")])
+        # The same file through the inner source is the same key.
+        data = await self.put([self.target("a.png", 1)], "Scans")
+        self.assertEqual(data["unchanged"], 1)
+        self.assertEqual((await self.listed())["categories"], ["Scans"])
+        self.assertEqual((await self.delete([self.target("a.png", 1)]))["cleared"], 1)
+
+    async def test_video_files_can_have_a_category(self):
+        (self.output_dir / "clip.mp4").write_bytes(b"\x00\x00\x00\x18ftypmp42")
+        data = await self.put(["clip.mp4"], "Clips")
+        self.assertEqual(data["updated"], 1)
+
+    async def test_windows_case_folds_the_key(self):
+        make_png(self.output_dir / "Sub" / "A.png")
+        with patch("py.api.images._path_flavor", ntpath):
+            data = await self.put(["Sub/A.png"], "Scans")
+            self.assertEqual(data["updated"], 1)
+            data = await self.put(["sub/a.PNG"], "Scans")
+            self.assertEqual(data["unchanged"], 1)
+            self.assertEqual(self.rows()[0][1], "sub/a.png")
+            self.assertEqual((await self.delete(["SUB/A.png"]))["cleared"], 1)
+
+    async def test_a_key_beyond_the_stored_bound_is_403(self):
+        make_png(self.output_dir / "a.png")
+        too_long = "/" + "r" * MAX_KEY_LENGTH
+        with patch("py.api.images._favorite_root_key", return_value=too_long):
+            data = await self.put(["a.png"], "Scans", status=403)
+            self.assertNotIn(too_long, json.dumps(data))
+            await self.delete(["a.png"], status=403)
+        self.assertEqual(self.rows(), [])
+
+    async def test_an_unexpected_failure_is_a_logged_500(self):
+        make_png(self.output_dir / "a.png")
+
+        def broken(*_args, **_kwargs):
+            raise sqlite3.OperationalError("secret detail")
+
+        self.api.db.set_image_categories = broken
+        self.api.db.clear_image_categories = broken
+        with self.assertLogs("prompt_manager", level="ERROR"):
+            data = await self.put(["a.png"], "Scans", status=500)
+        self.assertEqual(data["error"], "The category could not be changed.")
+        self.assertNotIn("secret", json.dumps(data))
+        with self.assertLogs("prompt_manager", level="ERROR"):
+            data = await self.delete(["a.png"], status=500)
+        self.assertNotIn("secret", json.dumps(data))
+
+    async def test_a_symlinked_output_folder_keys_like_its_favorites(self):
+        """A detected output folder can be a symlink: keys use its real path."""
+        real = self.comfy_dir / "real-output"
+        make_png(real / "a.png")
+        link = self.comfy_dir / "linked-output"
+        self._make_symlink(real, link, target_is_directory=True)
+        self.api._get_all_output_dirs = lambda: [link]
+        data = await self.put(["a.png"], "Scans")
+        self.assertEqual(data["updated"], 1)
+        self.assertEqual(self.rows(), [(_favorite_root_key(link), "a.png", "Scans")])
+        self.assertEqual((await self.listed())["categories"], ["Scans"])
+        self.assertEqual((await self.delete(["a.png"]))["cleared"], 1)
+
+    async def test_lone_surrogates_are_400_not_500(self):
+        make_png(self.output_dir / "a.png")
+        await self.send(
+            "PUT", {"items": [self.target("a.png")], "category": "x\ud800"}, 400
+        )
+        await self.send("DELETE", {"items": [self.target("a\ud800.png")]}, 400)
+        self.assertEqual(self.rows(), [])
+
+    async def test_delete_category_never_reaches_delete_image(self):
+        make_png(self.output_dir / "a.png")
+        calls = []
+
+        async def recorded(request):
+            calls.append(request.match_info)
+            raise AssertionError("delete_image reached")
+
+        self.api.delete_image = recorded
+        await self.put(["a.png"], "Scans")
+        data = await self.delete(["a.png"])
+        self.assertEqual(data["cleared"], 1)
+        self.assertEqual(calls, [])
+
+
+class TestRefusals(ImageCategoryTestCase):
+    async def test_malformed_bodies_are_400(self):
+        make_png(self.output_dir / "a.png")
+        item = self.target("a.png")
+        cases = [
+            {"items": [item]},
+            {"category": "x"},
+            {"items": [item], "category": "x", "extra": 1},
+            {"items": item, "category": "x"},
+            {"items": [], "category": "x"},
+            {"items": [item] * 501, "category": "x"},
+            {"items": [item], "category": 3},
+            {"items": [item], "category": None},
+            {"items": [item], "category": " \t "},
+            {"items": [item], "category": "x" * 101},
+            {"items": [item], "category": "a\x07b"},
+            {"items": [{**item, "root_index": "0"}], "category": "x"},
+            {"items": [{**item, "root_scope": "nothex"}], "category": "x"},
+            {"items": [{"root_index": 0, "path": "a.png"}], "category": "x"},
+            {"items": [{**item, "root_index": 1}], "category": "x"},
+        ]
+        for body in cases:
+            with self.subTest(body=str(body)[:80]):
+                data = await self.send("PUT", body, status=400)
+                self.assertFalse(data["success"])
+        await self.send("DELETE", {"items": [item], "category": "x"}, status=400)
+        await self.send("DELETE", {"items": []}, status=400)
+        response = await self.client.request(
+            "PUT", CATEGORY, data="not json", headers={"Content-Type": "text/plain"}
+        )
+        self.assertEqual(response.status, 400)
+        self.assertEqual(self.rows(), [])
+
+    async def test_unsafe_paths_are_403(self):
+        (self.output_dir / "notes.txt").write_text("not media")
+        for path in ("../outside.png", "/a.png", "C:/a.png", "a\\b.png", "a/./b.png"):
+            with self.subTest(path=path):
+                await self.put([self.target(path)], "x", status=403)
+        data = await self.put([self.target("notes.txt")], "x", status=403)
+        self.assertEqual(data["error"], "Only media files can have a category")
+        await self.delete([self.target("notes.txt")], status=403)
+
+    async def test_a_stale_scope_is_409(self):
+        make_png(self.output_dir / "a.png")
+        item = {**self.target("a.png"), "root_scope": "0" * 64}
+        data = await self.put([item], "x", status=409)
+        self.assertEqual(data["code"], "source-changed")
+        data = await self.delete([item], status=409)
+        self.assertEqual(data["code"], "source-changed")
+
+    async def test_an_oversized_body_is_413(self):
+        response = await self.client.request(
+            "PUT",
+            CATEGORY,
+            data=b'{"items": [], "category": "' + b"x" * (2 * 1024 * 1024) + b'"}',
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(response.status, 413)
+
+
+class TestCategoryList(ImageCategoryTestCase):
+    async def test_lists_only_values_whose_file_is_listed(self):
+        make_png(self.output_dir / "a.png")
+        make_png(self.output_dir / "c.png")
+        await self.put(["a.png"], "B")
+        await self.put(["c.png"], "A")
+        self.api.db.set_image_categories([self.key("gone.png")], "Z")
+        self.assertEqual(
+            await self.listed(), {"success": True, "categories": ["A", "B"]}
+        )
+
+    async def test_empty_library_lists_nothing(self):
+        self.assertEqual((await self.listed())["categories"], [])
+
+    async def test_a_failed_read_is_500(self):
+        make_png(self.output_dir / "a.png")
+
+        def broken(*_args, **_kwargs):
+            raise sqlite3.OperationalError("secret detail")
+
+        self.api.db.own_categories_of = broken
+        with self.assertLogs("prompt_manager", level="ERROR"):
+            data = await self.listed(status=500)
+        self.assertEqual(data["error"], "Image categories could not be read.")
+        self.assertNotIn("secret", json.dumps(data))
+
+
+class TestAllMatchingFilter(ImageCategoryTestCase):
+    """#392 "Select all N matching": the listing filter, resolved now."""
+
+    async def test_filter_sets_and_clears_every_match(self):
+        for name in ("bulk/a.png", "bulk/b.png", "other/c.png"):
+            make_png(self.output_dir / name)
+        data = await self.send(
+            "PUT", {"filter": {"subfolder": "bulk"}, "category": "Scans"}
+        )
+        self.assertEqual((data["updated"], data["not_found"]), (2, 0))
+        self.assertEqual(
+            sorted(row[1] for row in self.rows()), ["bulk/a.png", "bulk/b.png"]
+        )
+        data = await self.send("DELETE", {"filter": {"subfolder": "bulk"}})
+        self.assertEqual(data["cleared"], 2)
+        data = await self.send(
+            "PUT", {"filter": {"subfolder": "none"}, "category": "Scans"}, 404
+        )
+        self.assertEqual(data["code"], "files-not-found")
+        self.assertEqual(
+            (await self.send("DELETE", {"filter": {"q": "zz"}}))["cleared"], 0
+        )
+
+    async def test_filter_refusals(self):
+        make_png(self.output_dir / "a.png")
+        item = self.target("a.png")
+        for body in (
+            {"filter": {}, "items": [item], "category": "x"},
+            {"filter": {"unknown": "1"}, "category": "x"},
+            {"filter": "q=a", "category": "x"},
+            {"filter": {"root": "0", "root_scope": "nothex"}, "category": "x"},
+        ):
+            with self.subTest(body=body):
+                await self.send("PUT", body, 400)
+        data = await self.send(
+            "PUT",
+            {"filter": {"root": "0", "root_scope": "0" * 64}, "category": "x"},
+            409,
+        )
+        self.assertEqual(data["code"], "source-changed")
+        self.assertEqual(self.rows(), [])
+
+    async def test_more_than_500_matching_is_refused(self):
+        for number in range(501):
+            make_png(self.output_dir / f"many/{number:03}.png", size=(1, 1))
+        data = await self.send("PUT", {"filter": {}, "category": "x"}, 400)
+        self.assertEqual(data["code"], "too-many-images")
+        self.assertEqual(data["error"], "Select at most 500 images.")
+        self.assertEqual(self.rows(), [])

tests/test_cached_runs.py

@@ -0,0 +1,823 @@
+"""Cached re-runs of runtime-linked prompt text count once per job (#347).
+
+Jobs are simulated end to end: the queue hook (submit + dequeue), the
+``execution_cached`` replay, the real node executions and ComfyUI's output
+cache (fakes of ``execution`` and ``comfy_execution.caching``, as in
+tests/test_execution_context_host.py).
+"""
+
+import logging
+import os
+import sys
+import tempfile
+import threading
+import types
+import unittest
+from collections import namedtuple
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+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.operations import PromptDatabase
+from utils import cached_runs as module
+from utils.cached_runs import (
+    CachedRuns,
+    Facts,
+    Role,
+    Seen,
+    graph_role,
+    graph_seen,
+    own_counts,
+    register_cached_runs,
+    stored_facts,
+)
+from utils.execution_context_host import cached_string_outputs
+from utils.hashing import generate_prompt_hash
+from utils.run_keys import prompt_node_key, upstream_closure, upstream_nodes
+from utils.usage_tracking import (
+    PendingFirstUse,
+    QueuedRuns,
+    find_pair_id,
+    find_pair_id_by_hashes,
+    handle_queued_prompt,
+    on_dequeue,
+)
+
+Entry = namedtuple("CacheEntry", "ui outputs")
+LLM = "a lighthouse the LLM wrote"
+
+
+class _Cache:
+    def __init__(self):
+        self.entries = {}
+
+    def get_local(self, node_id):
+        return self.entries.get(node_id)
+
+
+class HierarchicalCache(_Cache):
+    pass
+
+
+class LRUCache(_Cache):
+    pass
+
+
+class RAMPressureCache(LRUCache):
+    pass
+
+
+class NullCache(_Cache):
+    pass
+
+
+def host_modules():
+    """Fake ``execution`` and ``comfy_execution.caching`` for the cache reader."""
+    execution = types.ModuleType("execution")
+    execution.CacheEntry = Entry
+
+    class PromptExecutor:
+        def add_message(self, event, data, broadcast):
+            return None
+
+    execution.PromptExecutor = PromptExecutor
+    caching = types.ModuleType("comfy_execution.caching")
+    for cls in (HierarchicalCache, LRUCache, RAMPressureCache, NullCache):
+        setattr(caching, cls.__name__, cls)
+    package = types.ModuleType("comfy_execution")
+    package.caching = caching
+    return {
+        "execution": execution,
+        "comfy_execution": package,
+        "comfy_execution.caching": caching,
+    }
+
+
+def graph(
+    kind="PromptManager",
+    text=None,
+    role="positive",
+    sampler=True,
+    negative_text="",
+    seed=1,
+):
+    """PrimitiveString -> PreviewAny -> ``kind`` -> KSampler (runtime text).
+
+    ``text`` set: typed text instead of the runtime link. ``role`` "negative":
+    the node feeds the sampler's negative input. Pair: its negative editor holds
+    ``negative_text`` and it feeds both sampler inputs.
+    """
+    g = {
+        "str": {"class_type": "PrimitiveString", "inputs": {"value": "seed text"}},
+        "llm": {"class_type": "PreviewAny", "inputs": {"source": ["str", 0]}},
+        "ckpt": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt": "m"}},
+        "pm": {
+            "class_type": kind,
+            "inputs": {"text": ["llm", 0] if text is None else text},
+        },
+        "other": {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": "", "clip": ["ckpt", 1]},
+        },
+        "save": {"class_type": "SaveImage", "inputs": {"images": ["ks", 0]}},
+    }
+    if kind != "PromptManagerText":
+        g["pm"]["inputs"]["clip"] = ["ckpt", 1]
+    if kind == "PromptManagerPair":
+        g["pm"]["inputs"]["negative_text"] = negative_text
+    source = ["pm", 0]
+    if kind == "PromptManagerText":
+        g["enc"] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": ["pm", 0], "clip": ["ckpt", 1]},
+        }
+        source = ["enc", 0]
+    positive, negative = (source, ["other", 0])
+    if role == "negative":
+        positive, negative = negative, positive
+    if kind == "PromptManagerPair":
+        positive, negative = ["pm", 0], ["pm", 1]
+    if sampler:
+        g["ks"] = {
+            "class_type": "KSampler",
+            "inputs": {
+                "model": ["ckpt", 0],
+                "positive": positive,
+                "negative": negative,
+                "seed": seed,
+            },
+        }
+    else:
+        g["save"]["inputs"]["images"] = positive
+    return g
+
+
+class JobCase(unittest.TestCase):
+    """A database, the queue hook stashes and one ``CachedRuns`` per test."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.addCleanup(assert_closed, self, self.tmp.name)
+        self.db = PromptDatabase(os.path.join(self.tmp.name, "runs.db"))
+        self.addCleanup(self.db.close_all)
+        patcher = mock.patch.dict(sys.modules, host_modules())
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.queued, self.pair_queued = QueuedRuns(), QueuedRuns()
+        self.pending, self.pair_pending = PendingFirstUse(), PendingFirstUse()
+        self.context = (None, None)
+        self.runs = CachedRuns(context=lambda: self.context)
+        self.queue = object()
+        self.executor = types.SimpleNamespace(
+            server=types.SimpleNamespace(prompt_queue=self.queue),
+            caches=types.SimpleNamespace(outputs=HierarchicalCache()),
+        )
+        self.item_id = 0
+        self.trackers = []
+
+    # -- nodes ----------------------------------------------------------------
+
+    def node(self, kind):
+        if kind == "PromptManagerPair":
+            from prompt_manager_pair import PromptManagerPair as cls
+        elif kind == "PromptManagerText":
+            from prompt_manager_text import PromptManagerText as cls
+        else:
+            import prompt_manager
+
+            cls = getattr(prompt_manager, kind)
+        node = cls.__new__(cls)
+        node.db = self.db
+        node.logger = logging.getLogger("test.cached_runs")
+        node.prompt_tracker = mock.Mock()
+        self.trackers.append(node.prompt_tracker)
+        node.comfyui_integration = mock.Mock()
+        node._inject_lora_trigger_words = lambda text: text
+        node.pending_first_use = self.pending
+        node.pending_pair_first_use = self.pair_pending
+        node.cached_runs = self.runs
+        return node
+
+    def execute(self, g, node_id="pm", text=LLM, negative="", index=None):
+        """Run the node as ComfyUI would; returns its STRING outputs."""
+        kind = g[node_id]["class_type"]
+        node = self.node(kind)
+        self.context = (self.context[0], index)
+        clip = mock.Mock()
+        if kind == "PromptManagerText":
+            return node.process_text(text, prompt=g, unique_id=node_id)
+        if kind == "PromptManagerPair":
+            return node.encode_prompt(
+                clip, text, negative_text=negative, prompt=g, unique_id=node_id
+            )
+        return node.encode_prompt(clip, text, prompt=g, unique_id=node_id)
+
+    def cache(self, node_id, *results):
+        """Store executed results (one per list index) as ComfyUI's cache entry."""
+        width = len(results[0])
+        outputs = [[result[slot] for result in results] for slot in range(width)]
+        self.executor.caches.outputs.entries[node_id] = Entry({}, outputs)
+
+    # -- jobs -----------------------------------------------------------------
+
+    def job(self, g, cached=(), run=None, outputs=("save",), prompt_id=None):
+        """One queued job: hook, dequeue, ``execution_cached``, executions, done."""
+        self.item_id += 1
+        prompt_id = prompt_id or f"job-{self.item_id}"
+        request = handle_queued_prompt(
+            {"prompt": g, "prompt_id": prompt_id},
+            self.queued,
+            pair_queued=self.pair_queued,
+        )
+        on_dequeue(
+            request["prompt_id"],
+            self.queued,
+            lambda: self.db,
+            self.pending,
+            pair_queued=self.pair_queued,
+            pair_pending=self.pair_pending,
+        )
+        self.runs.begin_attempt(
+            ((self.item_id, prompt_id, g, {}, list(outputs), {}), self.item_id)
+        )
+        self.context = (prompt_id, None)
+        self.runs.replay(self.executor, prompt_id, list(cached), lambda: self.db)
+        result = run() if run else None
+        self.runs.end_attempt(self.item_id)
+        self.context = (None, None)
+        return result
+
+    def first(self, g, text=LLM, negative="", node_id="pm"):
+        """A job that executes the node and fills the cache."""
+        result = self.job(
+            g, run=lambda: self.execute(g, node_id, text=text, negative=negative)
+        )
+        self.cache(node_id, result)
+        return result
+
+    def count(self, text):
+        row = self.db.get_prompt_by_hash(generate_prompt_hash(text))
+        return row["run_count"] if row else None
+
+    def pair_count(self, positive, negative):
+        pair_id = find_pair_id(self.db, positive, negative)
+        return self.db.get_prompt_pair(pair_id)["run_count"] if pair_id else None
+
+
+class TestCachedJobsCount(JobCase):
+    def test_runtime_text_counts_every_cached_job_for_each_node_kind(self):
+        for kind in (
+            "PromptManager",
+            "PromptManagerFocused",
+            "PromptManagerSplit",
+            "PromptManagerText",
+        ):
+            with self.subTest(kind=kind):
+                text = f"{LLM} for {kind}"
+                g = graph(kind)
+                self.first(g, text=text)
+                self.job(graph(kind, seed=2), cached=["pm"])
+                self.job(graph(kind, seed=3), cached=["pm"])
+                self.assertEqual(self.count(text), 3)
+
+    def test_typed_text_still_counts_once_per_job(self):
+        g = graph(text="typed castle")
+        self.first(g, text="typed castle")
+        self.job(g, cached=["pm"])
+        self.job(g, cached=["pm"])
+        self.assertEqual(self.count("typed castle"), 3)
+
+    def test_rewired_to_negative_while_cached_counts_nothing(self):
+        self.first(graph())
+        self.job(graph(role="negative"), cached=["pm"])
+        self.assertEqual(self.count(LLM), 1)
+        # Rewired back to positive: the receipt still proves the text
+        self.job(graph(), cached=["pm"])
+        self.assertEqual(self.count(LLM), 2)
+
+    def test_no_sampler_counts_every_node_on_replay(self):
+        g = graph(sampler=False)
+        self.first(g)
+        self.job(g, cached=["pm"])
+        self.assertEqual(self.count(LLM), 2)
+
+    def test_empty_editor_replays_nothing(self):
+        g = graph()
+        self.first(g, text="")
+        self.job(g, cached=["pm"])
+        self.assertEqual(self.db.search_prompts(), [])
+
+    def test_alternating_upstream_values_each_count_their_own_prompt(self):
+        g = graph()
+        self.first(g, text="value A")
+        self.first(g, text="value B")
+        self.cache("pm", self.execute_outputs("value A"))
+        self.job(g, cached=["pm"])
+        self.assertEqual((self.count("value A"), self.count("value B")), (2, 1))
+
+    def execute_outputs(self, text):
+        return (mock.ANY, text)
+
+    def test_list_items_replay_each_item_in_order(self):
+        g = graph()
+        results = []
+
+        def run():
+            for index, text in enumerate(("one", "two", "three")):
+                results.append(self.execute(g, text=text, index=index))
+
+        self.job(g, run=run)
+        self.cache("pm", *results)
+        self.job(g, cached=["pm"])
+        self.assertEqual([self.count(t) for t in ("one", "two", "three")], [2, 2, 2])
+
+    def test_a_repeated_id_or_message_never_counts_a_node_twice(self):
+        g = graph()
+        self.first(g)
+        self.job(g, cached=["pm", "pm"])
+        self.assertEqual(self.count(LLM), 2)
+        self.item_id += 1
+        self.runs.begin_attempt(((1, "again", g, {}, ["save"], {}), self.item_id))
+        for _ in range(2):
+            self.runs.replay(self.executor, "again", ["pm"], lambda: self.db)
+        self.assertEqual(self.count(LLM), 3)
+
+    def test_a_shorter_list_keeps_its_receipt_after_a_longer_one(self):
+        g = graph()
+        self.first(g, text="alpha")
+        results = []
+
+        def run():
+            for index, text in enumerate(("alpha", "beta")):
+                results.append(self.execute(g, text=text, index=index))
+
+        self.job(g, run=run)
+        self.cache("pm", (mock.ANY, "alpha"))
+        self.job(g, cached=["pm"])
+        self.assertEqual((self.count("alpha"), self.count("beta")), (3, 1))
+
+    def test_a_job_reusing_a_prompt_id_counts_and_executes_normally(self):
+        g = graph()
+        self.first(g)
+        self.job(g, cached=["pm"], prompt_id="same")
+        self.job(g, run=lambda: self.execute(g), prompt_id="same")
+        self.job(g, cached=["pm"], prompt_id="same")
+        self.assertEqual(self.count(LLM), 4)
+
+
+class TestFailClosed(JobCase):
+    def setUp(self):
+        super().setUp()
+        self.g = graph()
+        self.first(self.g)
+
+    def assertNotCounted(self):
+        self.assertEqual(self.count(LLM), 1)
+
+    def test_changed_cached_text_is_not_counted(self):
+        self.cache("pm", (mock.ANY, LLM + " edited"))
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_no_receipt_in_this_process_is_not_counted(self):
+        self.runs = CachedRuns(context=lambda: self.context)
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_missing_cache_entry_is_not_counted(self):
+        self.executor.caches.outputs.entries.clear()
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_unsupported_cache_is_not_counted(self):
+        entries = self.executor.caches.outputs.entries
+        self.executor.caches.outputs = NullCache()
+        self.executor.caches.outputs.entries = entries
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_more_list_values_than_the_bound_are_not_counted(self):
+        self.cache("pm", *[(mock.ANY, LLM)] * 65)
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_missing_list_index_is_not_counted(self):
+        self.cache("pm", (mock.ANY, LLM), (mock.ANY, LLM))
+        self.job(self.g, cached=["pm"])
+        self.assertNotCounted()
+
+    def test_node_the_job_does_not_run_is_not_counted(self):
+        self.job(self.g, cached=["pm"], outputs=("other",))
+        self.assertNotCounted()
+
+    def test_message_for_another_job_is_not_counted(self):
+        self.item_id += 1
+        self.runs.begin_attempt(((1, "this", self.g, {}, ["save"], {}), self.item_id))
+        self.runs.replay(self.executor, "another", ["pm"], lambda: self.db)
+        self.assertNotCounted()
+
+    def test_unkeyable_node_is_not_counted(self):
+        g = graph()
+        g["pm"]["inputs"]["extra"] = ["missing", 0]
+        self.job(g, cached=["pm"])
+        self.assertIsNone(prompt_node_key(g, "pm"))
+        self.assertNotCounted()
+
+    def test_deleted_prompt_is_skipped(self):
+        self.db.delete_prompt(
+            self.db.get_prompt_by_hash(generate_prompt_hash(LLM))["id"]
+        )
+        self.job(self.g, cached=["pm"])
+        self.assertIsNone(self.count(LLM))
+
+    def test_database_errors_never_raise(self):
+        def broken():
+            raise RuntimeError("database unavailable")
+
+        self.item_id += 1
+        self.runs.begin_attempt(((1, "j", self.g, {}, ["save"], {}), self.item_id))
+        self.runs.replay(self.executor, "j", ["pm"], broken)
+        self.assertNotCounted()
+
+    def test_malformed_messages_and_results_are_ignored(self):
+        self.runs.begin_attempt(None)
+        self.runs.begin_attempt((("x",), "not an int"))
+        self.runs.replay(self.executor, None, ["pm"], lambda: self.db)
+        self.item_id += 1
+        self.runs.begin_attempt(((1, "j", self.g, {}, ["save"], {}), self.item_id))
+        self.runs.replay(self.executor, "j", "pm", lambda: self.db)
+        self.runs.replay(self.executor, "j", ["pm"] * 501, lambda: self.db)
+        self.runs.replay(self.executor, "j", [7, None], lambda: self.db)
+        self.assertNotCounted()
+
+
+class TestReplayThenExecution(JobCase):
+    def test_execution_after_replay_in_the_same_job_is_not_counted_again(self):
+        # RAM-pressure frees the entry after execution_cached: the node executes
+        g = graph()
+        self.first(g)
+        self.job(g, cached=["pm"], run=lambda: self.execute(g))
+        self.assertEqual(self.count(LLM), 2)
+        # The suppressed execution still leaves a receipt: the next job counts
+        self.job(g, cached=["pm"])
+        self.assertEqual(self.count(LLM), 3)
+
+    def test_suppression_keeps_the_pending_catch_up_and_the_tracker_queue(self):
+        g = graph()
+        self.first(g)
+        prompt_id = self.db.get_prompt_by_hash(generate_prompt_hash(LLM))["id"]
+
+        def run():
+            # The replay counted, then the prompt was deleted and re-saved
+            self.db.delete_prompt(prompt_id)
+            self.pending.add(generate_prompt_hash(LLM))
+            return self.execute(g)
+
+        self.job(g, cached=["pm"], run=run)
+        self.assertEqual(self.count(LLM), 1)  # the catch-up, not the own count
+        push = self.trackers[-1].set_current_prompt.call_args.kwargs["push_to_queue"]
+        self.assertTrue(push)
+
+    def test_without_host_context_the_node_counts_as_before(self):
+        g = graph()
+        self.runs = CachedRuns(context=lambda: (None, None))
+        self.first(g)
+        self.job(g, cached=["pm"])
+        self.job(g, run=lambda: self.execute(g))
+        self.assertEqual(self.count(LLM), 2)
+
+
+class TestPairReplay(JobCase):
+    def test_runtime_positive_counts_positive_and_pair_every_job(self):
+        g = graph("PromptManagerPair", negative_text="blurry")
+        self.first(g, negative="blurry")
+        self.job(graph("PromptManagerPair", negative_text="blurry", seed=2), ["pm"])
+        self.job(graph("PromptManagerPair", negative_text="blurry", seed=3), ["pm"])
+        self.assertEqual(self.count(LLM), 3)
+        self.assertEqual(self.pair_count(LLM, "blurry"), 3)
+        self.assertEqual(self.count("blurry"), 0)
+
+    def test_empty_negative_pair_replays_its_pair_count(self):
+        g = graph("PromptManagerPair")
+        self.first(g)
+        self.job(g, cached=["pm"])
+        self.assertEqual(self.pair_count(LLM, ""), 2)
+        self.assertEqual(self.count(LLM), 2)
+
+    def test_pair_execution_after_replay_is_not_counted_again(self):
+        g = graph("PromptManagerPair", negative_text="blurry")
+        self.first(g, negative="blurry")
+        self.job(g, ["pm"], run=lambda: self.execute(g, negative="blurry"))
+        self.assertEqual(self.count(LLM), 2)
+        self.assertEqual(self.pair_count(LLM, "blurry"), 2)
+
+    def test_pair_role_follows_the_current_graph(self):
+        g = graph("PromptManagerPair", negative_text="blurry")
+        self.first(g, negative="blurry")
+        rewired = graph("PromptManagerPair", negative_text="blurry")
+        rewired["ks"]["inputs"]["positive"] = ["other", 0]
+        self.job(rewired, cached=["pm"])
+        self.assertEqual(self.count(LLM), 1)
+        self.assertEqual(self.pair_count(LLM, "blurry"), 1)
+
+
+class TestOwnCounts(unittest.TestCase):
+    """The node path and replay decide through the same function."""
+
+    def test_single_node_table(self):
+        facts = Facts("p")
+        for used, seen, expected in (
+            ({"text"}, False, [("prompt", "p")]),
+            ({"text"}, True, []),
+            (set(), False, []),
+        ):
+            with self.subTest(used=used, seen=seen):
+                counts = own_counts(
+                    "PromptManager", facts, Role(frozenset(used)), Seen(text=seen)
+                )
+                self.assertEqual(counts, expected)
+        self.assertEqual(
+            own_counts("PromptManager", Facts(None), Role(frozenset({"text"})), Seen()),
+            [],
+        )
+
+    def test_pair_table(self):
+        both = Role(frozenset({"text", "negative_text"}), True)
+        self.assertEqual(
+            own_counts("PromptManagerPair", Facts("p", "n"), both, Seen()),
+            [("prompt", "p"), ("prompt", "n"), ("pair", "p", "n")],
+        )
+        self.assertEqual(
+            own_counts("PromptManagerPair", Facts("p", "p"), both, Seen()),
+            [("prompt", "p"), ("pair", "p", "p")],
+        )
+        self.assertEqual(
+            own_counts("PromptManagerPair", Facts("p"), both, Seen(text=True)),
+            [("pair", "p", None)],
+        )
+        self.assertEqual(
+            own_counts(
+                "PromptManagerPair", Facts("p", "n"), both, Seen(True, True, True)
+            ),
+            [],
+        )
+
+    def test_node_and_replay_agree_on_role_and_seen(self):
+        for g, node_id in (
+            (graph(), "pm"),
+            (graph(text="typed"), "pm"),
+            (graph(role="negative"), "pm"),
+            (graph(sampler=False), "pm"),
+            (graph("PromptManagerPair", negative_text="n"), "pm"),
+            (graph("PromptManagerPair", text="typed", negative_text="n"), "pm"),
+        ):
+            kind = g[node_id]["class_type"]
+            text = "typed" if isinstance(g[node_id]["inputs"]["text"], str) else LLM
+            facts = stored_facts(text, g[node_id]["inputs"].get("negative_text"))
+            with self.subTest(kind=kind, inputs=g[node_id]["inputs"]):
+                replay = own_counts(
+                    kind,
+                    facts,
+                    graph_role(g, node_id, kind),
+                    graph_seen(g, node_id, kind, facts),
+                )
+                self.assertEqual(replay, self.node_counts(g, node_id, kind, text))
+
+    @staticmethod
+    def node_counts(g, node_id, kind, text):
+        """What the executing node counted itself, from a recording database."""
+        db = mock.Mock()
+        db.get_prompt_by_hash.side_effect = lambda h: {"id": h}
+        db.save_prompt_pair.return_value = 1
+        negative = g[node_id]["inputs"].get("negative_text", "")
+        db.get_prompt_pair.return_value = {
+            "positive_prompt_id": generate_prompt_hash(text),
+            "negative_prompt_id": (
+                generate_prompt_hash(negative) if negative.strip() else None
+            ),
+        }
+        case = JobCase()
+        case.db = db
+        case.pending, case.pair_pending = PendingFirstUse(), PendingFirstUse()
+        case.runs = CachedRuns(context=lambda: (None, None))
+        case.trackers = []
+        case.context = (None, None)
+        if kind == "PromptManagerPair":
+            node = case.node(kind)
+            node.encode_prompt(
+                mock.Mock(), text, negative_text=negative, prompt=g, unique_id=node_id
+            )
+        else:
+            node = case.node(kind)
+            node._track_prompt_execution(text, text, None, None, {}, g, node_id)
+        counts = [("prompt", c.args[0]) for c in db.record_prompt_use.call_args_list]
+        if kind == "PromptManagerPair":
+            counts += [
+                ("pair", generate_prompt_hash(text), db.get_prompt_pair()[k])
+                for k in ["negative_prompt_id"]
+                if db.record_prompt_pair_use.called
+            ]
+        return counts
+
+
+class TestReceipts(unittest.TestCase):
+    def setUp(self):
+        self.context = ("j1", None)
+        self.runs = CachedRuns(
+            context=lambda: self.context, max_keys=2, max_receipts=2, max_items=3
+        )
+        self.g = {
+            name: {"class_type": "PromptManager", "inputs": {"text": name}}
+            for name in ("a", "b", "c")
+        }
+
+    def begin(self, item_id, prompt_id="j1"):
+        self.context = (prompt_id, None)
+        self.runs.begin_attempt(
+            ((item_id, prompt_id, self.g, {}, ["a", "b", "c"], {}), item_id)
+        )
+
+    def keys(self):
+        return list(self.runs._receipts)
+
+    def test_least_recently_used_run_keys_are_dropped(self):
+        self.begin(1)
+        for name in ("a", "b", "c"):
+            self.runs.record_item(name, (name,), Facts("h"))
+        self.assertEqual(self.keys(), [prompt_node_key(self.g, n) for n in ("b", "c")])
+
+    def test_receipts_per_key_are_bounded_and_equal_outputs_replace(self):
+        key = prompt_node_key(self.g, "a")
+        for item_id, text in ((1, "x"), (2, "y"), (3, "z"), (4, "y")):
+            self.begin(item_id)
+            self.runs.record_item("a", (text,), Facts(text))
+            self.runs.end_attempt(item_id)
+        receipts = self.runs._receipts[key]
+        self.assertEqual([r.item_id for r in receipts], [4, 3])
+
+    def test_list_indices_past_the_bound_are_not_recorded(self):
+        self.begin(1)
+        self.context = ("j1", 3)
+        self.runs.record_item("a", ("x",), Facts("x"))
+        self.assertEqual(self.keys(), [])
+
+    def test_non_string_outputs_and_unknown_nodes_are_not_recorded(self):
+        self.begin(1)
+        self.runs.record_item("a", (None,), Facts("x"))
+        self.runs.record_item("zzz", ("x",), Facts("x"))
+        self.assertEqual(self.keys(), [])
+
+    def test_task_done_clears_only_its_own_attempt(self):
+        self.begin(1)
+        self.runs.end_attempt(2)
+        self.runs.record_item("a", ("x",), Facts("x"))
+        self.runs.end_attempt(1)
+        self.runs.record_item("b", ("x",), Facts("x"))
+        self.assertEqual(self.keys(), [prompt_node_key(self.g, "a")])
+
+    def test_job_nodes_follow_the_outputs_to_execute(self):
+        self.runs.begin_attempt(((1, "j1", self.g, {}, ["b"], {}), 1))
+        self.assertEqual(list(self.runs._attempt.nodes), ["b"])
+
+
+class TestCacheReader(unittest.TestCase):
+    def setUp(self):
+        patcher = mock.patch.dict(sys.modules, host_modules())
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.cache = LRUCache()
+        self.executor = types.SimpleNamespace(
+            caches=types.SimpleNamespace(outputs=self.cache)
+        )
+
+    def read(self, outputs, kind="PromptManager"):
+        self.cache.entries["n"] = Entry({}, outputs)
+        return cached_string_outputs(self.executor, "n", kind)
+
+    def test_reads_each_string_slot_in_order(self):
+        self.assertEqual(self.read([["c"], ["a", "b"]]), (("a", "b"),))
+        self.assertEqual(self.read([["t"]], "PromptManagerText"), (("t",),))
+        self.assertEqual(
+            self.read([[1], [2], ["p"], ["n"]], "PromptManagerPair"), (("p",), ("n",))
+        )
+
+    def test_fails_closed(self):
+        for outputs, kind in (
+            ([["c"], ["a", 1]], "PromptManager"),
+            ([["c"]], "PromptManager"),
+            ([["c"], ["a"] * 65], "PromptManager"),
+            ([[1], [2], ["p", "q"], ["n"]], "PromptManagerPair"),
+            ([["c"], "text"], "PromptManager"),
+            ([["c"], ["a"]], "CLIPTextEncode"),
+            ([[]] * 5, "PromptManager"),
+        ):
+            with self.subTest(outputs=outputs, kind=kind):
+                self.assertIsNone(self.read(outputs, kind))
+        self.cache.entries["m"] = ({}, [["c"], ["a"]])  # not a CacheEntry
+        self.assertIsNone(cached_string_outputs(self.executor, "m", "PromptManager"))
+        self.executor.caches.outputs = NullCache()
+        self.assertIsNone(self.read([["c"], ["a"]]))
+
+
+class TestGraphHelpers(unittest.TestCase):
+    def test_upstream_nodes_and_closure(self):
+        g = graph()
+        self.assertEqual(
+            set(upstream_nodes(g, ["save"])),
+            {"save", "ks", "pm", "llm", "str", "ckpt", "other"},
+        )
+        self.assertEqual(upstream_nodes(g, ["missing"]), [])
+        self.assertEqual(upstream_nodes(None, ["save"]), [])
+        self.assertIsNone(upstream_closure(g, ["save"], limit=2))
+        self.assertIsNone(upstream_closure({"a": {"inputs": {"x": ["b", 0]}}}, ["a"]))
+
+    def test_find_pair_by_hashes_matches_the_text_lookup(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db = PromptDatabase(os.path.join(tmp, "p.db"))
+            try:
+                pair_id = db.save_prompt_pair("cat", "")
+                self.assertEqual(
+                    find_pair_id_by_hashes(db, generate_prompt_hash("cat"), None),
+                    pair_id,
+                )
+                self.assertIsNone(
+                    find_pair_id_by_hashes(
+                        db, generate_prompt_hash("cat"), generate_prompt_hash("x")
+                    )
+                )
+            finally:
+                db.close_all()
+            assert_closed(self, tmp)
+
+
+class TestRegistration(unittest.TestCase):
+    def setUp(self):
+        self.modules = host_modules()
+        patcher = mock.patch.dict(sys.modules, self.modules)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        flag = mock.patch.object(module, "_registered", threading.Event())
+        flag.start()
+        self.addCleanup(flag.stop)
+
+    class Queue:
+        def __init__(self):
+            self.done = []
+
+        def get(self, timeout=None):
+            return self.result
+
+        def task_done(self, item_id, *args, **kwargs):
+            self.done.append(item_id)
+
+    def test_registers_once_and_follows_the_job(self):
+        queue = self.Queue()
+        server = types.SimpleNamespace(prompt_queue=queue)
+        runs = mock.Mock()
+        self.assertTrue(register_cached_runs(server, lambda: "db", runs))
+        self.assertFalse(register_cached_runs(server, lambda: "db", runs))
+        queue.result = (("item",), 5)
+        queue.get()
+        runs.begin_attempt.assert_called_once_with((("item",), 5))
+        executor = self.modules["execution"].PromptExecutor()
+        executor.server = server
+        executor.add_message("execution_start", {"prompt_id": "j"}, broadcast=False)
+        executor.add_message(
+            "execution_cached", {"prompt_id": "j", "nodes": ["n"]}, broadcast=False
+        )
+        runs.replay.assert_called_once()
+        self.assertEqual(runs.replay.call_args.args[1:3], ("j", ["n"]))
+        other = self.modules["execution"].PromptExecutor()
+        other.server = types.SimpleNamespace(prompt_queue=object())
+        other.add_message("execution_cached", {"prompt_id": "j", "nodes": []}, False)
+        self.assertEqual(runs.replay.call_count, 1)
+        queue.task_done(5, None, status=None)
+        runs.end_attempt.assert_called_once_with(5)
+
+    def test_hosts_without_the_interfaces_register_nothing(self):
+        class NoTaskDone:
+            def get(self):
+                return None
+
+        self.assertFalse(
+            register_cached_runs(
+                types.SimpleNamespace(prompt_queue=NoTaskDone()), lambda: None
+            )
+        )
+        self.assertFalse(register_cached_runs(types.SimpleNamespace(), lambda: None))
+        del self.modules["execution"].PromptExecutor.add_message
+        self.assertFalse(
+            register_cached_runs(
+                types.SimpleNamespace(prompt_queue=self.Queue()), lambda: None
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_duplicate_image_categories.py

@@ -0,0 +1,151 @@
+"""Duplicate cleanup keeps images' own categories (#332 MC11).
+
+The kept copy's own category wins; a kept copy without one receives the
+removed copy's. The removed copy's row is never deleted, so a give-back
+(the kept copy changed) and a failed delete lose nothing.
+"""
+
+import asyncio
+import os
+from unittest.mock import patch
+
+try:
+    from tests.test_api_duplicate_files import WAIT, DuplicateFilesTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_duplicate_files import WAIT, DuplicateFilesTestCase
+
+from py.api import duplicate_files  # noqa: E402
+from py.api.images import image_category_key  # noqa: E402
+from py.config import GalleryConfig  # noqa: E402
+
+
+class TestDuplicateCleanupKeepsCategories(DuplicateFilesTestCase):
+    def key(self, path):
+        return image_category_key(self.api._get_all_output_dirs(), path)
+
+    def category(self, path):
+        return self.api.db.image_category(*self.key(path))
+
+    def set_category(self, path, value):
+        self.api.db.set_image_categories([self.key(path)], value)
+
+    async def run_cleanup(self, keep="a.png"):
+        job, group = await self.found_group()
+        plan = await self.review(job.id, [self.keep_of(group, keep)])
+        body = await self.apply(plan["token"])
+        cleanup = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        return cleanup
+
+    async def test_kept_copy_without_a_category_receives_the_removed_one(self):
+        self.trio()
+        b = self.output_dir / "b.png"
+        self.set_category(b, "Portraits")
+        cleanup = await self.run_cleanup()
+        self.assertFalse(b.exists())
+        self.assertEqual(self.category(self.output_dir / "a.png"), "Portraits")
+        # The removed copy's row stays (hidden while its file is gone).
+        self.assertEqual(self.category(b), "Portraits")
+        self.assertIn(
+            "Kept 1 image category on the kept copies", "\n".join(cleanup.log)
+        )
+
+    async def test_kept_copy_keeps_its_own_category(self):
+        self.trio()
+        self.set_category(self.output_dir / "a.png", "Landscapes")
+        self.set_category(self.output_dir / "b.png", "Portraits")
+        self.set_category(self.output_dir / "sub" / "c.png", "Scans")
+        cleanup = await self.run_cleanup()
+        self.assertEqual(self.category(self.output_dir / "a.png"), "Landscapes")
+        self.assertNotIn("image categor", "\n".join(cleanup.log))
+
+    async def test_two_removed_copies_give_the_first_category(self):
+        self.trio()
+        self.set_category(self.output_dir / "b.png", "Portraits")
+        self.set_category(self.output_dir / "sub" / "c.png", "Scans")
+        cleanup = await self.run_cleanup()
+        self.assertIn(self.category(self.output_dir / "a.png"), {"Portraits", "Scans"})
+        self.assertIn(
+            "Kept 1 image category on the kept copies", "\n".join(cleanup.log)
+        )
+
+    async def test_give_back_loses_no_category(self):
+        self.trio()
+        self.set_category(self.output_dir / "b.png", "Portraits")
+        with patch.object(duplicate_files, "still_kept", return_value=False):
+            cleanup = await self.run_cleanup()
+        self.assertEqual(cleanup.counters["removed"], 0)
+        rows = {row["path"]: row["detail"] for row in cleanup.outcomes["kept"]}
+        self.assertEqual(
+            rows["b.png"], duplicate_files.ROW_DETAILS[duplicate_files.KEEPER_GONE]
+        )
+        self.assertTrue((self.output_dir / "b.png").is_file())
+        self.assertEqual(self.category(self.output_dir / "b.png"), "Portraits")
+
+    async def test_a_failed_delete_keeps_both_categories(self):
+        self.trio()
+        self.set_category(self.output_dir / "b.png", "Portraits")
+        real = duplicate_files._unlink
+
+        def unlink(path):
+            if os.path.basename(path) == "b.png":
+                raise PermissionError(13, "denied")
+            real(path)
+
+        with patch.object(duplicate_files, "_unlink", side_effect=unlink):
+            await self.run_cleanup()
+        self.assertTrue((self.output_dir / "b.png").is_file())
+        self.assertEqual(self.category(self.output_dir / "b.png"), "Portraits")
+        self.assertEqual(self.category(self.output_dir / "a.png"), "Portraits")
+
+    async def test_nested_sources_use_the_deepest_source_keys(self):
+        self.write("sub/a.png", b"same", mtime=1)
+        self.write("sub/b.png", b"same", mtime=2)
+        # The subfolder is listed first, so the cleanup's listed root is the
+        # file key's source; the outer source still holds the files.
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir / "sub"),
+            self._root(self.output_dir),
+        ]
+        sub_key = duplicate_files.favorite_root_key(self.output_dir / "sub")
+        self.api.db.set_image_categories([(sub_key, "b.png")], "Portraits")
+        await self.run_cleanup()
+        self.assertFalse((self.output_dir / "sub" / "b.png").exists())
+        self.assertEqual(self.api.db.image_category(sub_key, "a.png"), "Portraits")
+        outer_key = duplicate_files.favorite_root_key(self.output_dir)
+        self.assertIsNone(self.api.db.image_category(outer_key, "sub/a.png"))
+
+    async def test_outer_source_listed_first_still_keys_by_the_deepest(self):
+        self.write("sub/a.png", b"same", mtime=1)
+        self.write("sub/b.png", b"same", mtime=2)
+        GalleryConfig.MONITORING_DIRECTORIES = [
+            self._root(self.output_dir),
+            self._root(self.output_dir / "sub"),
+        ]
+        sub_key = duplicate_files.favorite_root_key(self.output_dir / "sub")
+        self.api.db.set_image_categories([(sub_key, "b.png")], "Portraits")
+        job, group = await self.found_group()
+        [kept] = [f for f in group["files"] if f["path"].endswith("a.png")][:1]
+        plan = await self.review(
+            job.id, [self.keep_of(group, kept["path"], kept["root_index"])]
+        )
+        body = await self.apply(plan["token"])
+        cleanup = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(cleanup.task), WAIT)
+        self.assertFalse((self.output_dir / "sub" / "b.png").exists())
+        self.assertEqual(self.api.db.image_category(sub_key, "a.png"), "Portraits")
+
+    async def test_a_symlinked_output_folder_still_moves_the_category(self):
+        link = self.output_dir.parent / "linked-output"
+        try:
+            os.symlink(self.output_dir, link, target_is_directory=True)
+        except (OSError, NotImplementedError):
+            self.skipTest("symlinks are not available")
+        self.trio()
+        with patch.object(self.api, "_get_all_output_dirs", return_value=[link]):
+            key = image_category_key([link], link / "b.png")
+            self.api.db.set_image_categories([key], "Portraits")
+            await self.run_cleanup()
+            kept = image_category_key([link], link / "a.png")
+        self.assertFalse((self.output_dir / "b.png").exists())
+        self.assertEqual(self.api.db.image_category(*kept), "Portraits")

tests/test_image_categories.py

@@ -0,0 +1,523 @@
+"""Image categories (#332): schema, migration, restore check, store and file key."""
+
+import hashlib
+import ntpath
+import os
+import sqlite3
+import tempfile
+import unittest
+from contextlib import closing
+from pathlib import Path
+from unittest.mock import patch
+
+try:
+    from open_handles import assert_closed
+except ImportError:
+    from tests.open_handles import assert_closed
+
+from database.image_categories import (
+    COPY_SQL,
+    INDEX,
+    STATEMENTS,
+    TABLE,
+    copy_image_category,
+    image_categories_schema_error,
+    migrate_image_categories,
+)
+from database.models import PromptModel
+from database.operations import PromptDatabase
+from py.api.images import image_category_key
+
+_DDL = (
+    f"CREATE TABLE {TABLE} (root_key TEXT NOT NULL, relative_path TEXT NOT NULL, "
+    "category TEXT NOT NULL, updated_at TEXT NOT NULL, "
+    "PRIMARY KEY (root_key, relative_path))"
+)
+_INDEX_DDL = f"CREATE INDEX {INDEX} ON {TABLE}(category)"
+
+
+def category_checksum(conn) -> tuple:
+    """sha256 of prompts and prompt_pairs ``(id, category, updated_at)``."""
+    sums = []
+    for table in ("prompts", "prompt_pairs"):
+        digest = hashlib.sha256()
+        for row in conn.execute(  # nosec B608 - fixed table names
+            f"SELECT id, category, updated_at FROM {table} ORDER BY id"
+        ):
+            digest.update(repr(tuple(row)).encode())
+        sums.append(digest.hexdigest())
+    return tuple(sums)
+
+
+def table_checksums(conn) -> dict:
+    """``{name: (schema sql, sha256 of rows)}`` of every table but ours."""
+    found = {}
+    for name, sql in conn.execute(
+        "SELECT name, sql FROM sqlite_master WHERE type='table' AND name != ?",
+        (TABLE,),
+    ).fetchall():
+        digest = hashlib.sha256()
+        rows = conn.execute(f'SELECT * FROM "{name}"')  # nosec B608
+        for text in sorted(repr(tuple(row)) for row in rows):
+            digest.update(text.encode())
+        found[name] = (sql, digest.hexdigest())
+    return found
+
+
+class _TempDir(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.path = os.path.join(self.tmpdir, "categories.db")
+
+    def tearDown(self):
+        PromptModel.close_all_instances()
+        assert_closed(self, self.tmpdir)
+        for name in os.listdir(self.tmpdir):
+            os.unlink(os.path.join(self.tmpdir, name))
+        os.rmdir(self.tmpdir)
+
+
+class TestImageCategoriesSchema(_TempDir):
+    def test_migration_creates_table_and_index(self):
+        db = PromptDatabase(self.path)
+        db.close_all()
+        PromptModel.close_all_instances()
+        with closing(sqlite3.connect(self.path)) as conn:
+            names = {
+                (kind, name)
+                for kind, name in conn.execute(
+                    "SELECT type, name FROM sqlite_master WHERE tbl_name=?", (TABLE,)
+                )
+            }
+            self.assertEqual(
+                names,
+                {
+                    ("table", TABLE),
+                    ("index", INDEX),
+                    ("index", f"sqlite_autoindex_{TABLE}_1"),
+                },
+            )
+            self.assertEqual(
+                conn.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0], 0
+            )
+            self.assertIsNone(image_categories_schema_error(conn, allow_missing=False))
+
+    def test_migration_leaves_prompt_categories_unchanged(self):
+        """23,000 links; the migration reads and writes only its own objects."""
+        db = PromptDatabase(self.path)
+        db.close_all()
+        PromptModel.close_all_instances()
+        with closing(sqlite3.connect(self.path)) as conn:
+            conn.execute(f"DROP TABLE {TABLE}")  # a library from before #332
+            categories = ["Landscapes", "Portraits", None, "  ", "Scans"]
+            conn.executemany(
+                "INSERT INTO prompts (text, category, hash, updated_at) "
+                "VALUES (?, ?, ?, ?)",
+                [
+                    (f"p{n}", categories[n % 5], f"h{n}", f"2026-01-01 00:00:{n % 60}")
+                    for n in range(200)
+                ],
+            )
+            conn.executemany(
+                "INSERT INTO prompt_pairs (positive_prompt_id, negative_prompt_id, "
+                "category) VALUES (?, ?, ?)",
+                [(n, n + 100, categories[(n + 1) % 5]) for n in range(1, 50)],
+            )
+            conn.executemany(
+                "INSERT INTO generated_images (prompt_id, image_path, filename, "
+                "file_path) VALUES (?, ?, ?, ?)",
+                [
+                    (n % 200 + 1, f"/out/{n}.png", f"{n}.png", f"/out/{n}.png")
+                    for n in range(23000)
+                ],
+            )
+            conn.commit()
+            before = table_checksums(conn)
+            prompts_before = category_checksum(conn)
+            seen = []
+
+            def authorize(action, arg1, arg2, _db, _trigger):
+                seen.append((action, arg1, arg2))
+                return sqlite3.SQLITE_OK
+
+            conn.set_authorizer(authorize)
+            migrate_image_categories(conn)
+            conn.commit()
+            conn.set_authorizer(None)
+            own = {TABLE, "sqlite_master", "sqlite_schema", INDEX}
+            touched = set()
+            for action, arg1, arg2 in seen:
+                if action in (sqlite3.SQLITE_READ, sqlite3.SQLITE_INSERT):
+                    touched.add(arg1)
+                if action == sqlite3.SQLITE_READ:
+                    self.assertIn(arg1, own, (action, arg1, arg2))
+                elif action in (
+                    sqlite3.SQLITE_INSERT,
+                    sqlite3.SQLITE_UPDATE,
+                    sqlite3.SQLITE_DELETE,
+                ):
+                    self.assertIn(arg1, own, (action, arg1, arg2))
+                elif action == sqlite3.SQLITE_CREATE_TABLE:
+                    self.assertEqual(arg1, TABLE)
+                elif action == sqlite3.SQLITE_CREATE_INDEX:
+                    self.assertIn(arg1, (INDEX, f"sqlite_autoindex_{TABLE}_1"))
+                    self.assertEqual(arg2, TABLE)
+                elif action == sqlite3.SQLITE_PRAGMA:
+                    self.assertEqual((arg1, arg2), ("table_info", TABLE))
+            self.assertIn(TABLE, touched)  # the authorizer saw the work
+            self.assertEqual(table_checksums(conn), before)
+            self.assertEqual(category_checksum(conn), prompts_before)
+            schema = conn.execute(
+                "SELECT sql FROM sqlite_master ORDER BY name"
+            ).fetchall()
+            migrate_image_categories(conn)  # a second run changes nothing
+            conn.commit()
+            self.assertEqual(
+                conn.execute("SELECT sql FROM sqlite_master ORDER BY name").fetchall(),
+                schema,
+            )
+        # The whole startup sequence keeps them too.
+        PromptDatabase(self.path).close_all()
+        PromptModel.close_all_instances()
+        with closing(sqlite3.connect(self.path)) as conn:
+            self.assertEqual(category_checksum(conn), prompts_before)
+            self.assertEqual(
+                conn.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0], 0
+            )
+
+    def test_structure_accepted_and_refused(self):
+        variants = {
+            "missing": None,
+            "extra column": (
+                f"CREATE TABLE {TABLE} (root_key TEXT NOT NULL, relative_path TEXT "
+                "NOT NULL, category TEXT NOT NULL, updated_at TEXT NOT NULL, x TEXT, "
+                "PRIMARY KEY (root_key, relative_path))",
+                _INDEX_DDL,
+            ),
+            "wrong index": (_DDL, f"CREATE INDEX {INDEX} ON {TABLE}(updated_at)"),
+            "extra index": (
+                _DDL,
+                _INDEX_DDL,
+                f"CREATE INDEX extra ON {TABLE}(root_key)",
+            ),
+            "trigger": (
+                _DDL,
+                _INDEX_DDL,
+                f"CREATE TRIGGER t AFTER INSERT ON {TABLE} BEGIN SELECT 1; END",
+            ),
+            "wrong ddl": (
+                f"CREATE TABLE {TABLE} (root_key TEXT, relative_path TEXT, "
+                "category TEXT, updated_at TEXT, "
+                "PRIMARY KEY (root_key, relative_path))",
+                _INDEX_DDL,
+            ),
+            "view": (f"CREATE VIEW {TABLE} AS SELECT 1",),
+        }
+        for label, statements in variants.items():
+            with self.subTest(label), closing(sqlite3.connect(":memory:")) as conn:
+                for sql in statements or ():
+                    conn.execute(sql)
+                error = image_categories_schema_error(conn)
+                if statements is None:
+                    self.assertIsNone(error)
+                    self.assertIsNotNone(
+                        image_categories_schema_error(conn, allow_missing=False)
+                    )
+                    continue
+                self.assertIsNotNone(error)
+                self.assertIsNotNone(
+                    image_categories_schema_error(conn, check_values=False)
+                )
+                with self.assertRaises(sqlite3.DatabaseError):
+                    migrate_image_categories(conn)
+        # A table without its category index (never written by this code) is
+        # refused by the check but repaired by the install.
+        with closing(sqlite3.connect(":memory:")) as conn:
+            conn.execute(_DDL)
+            conn.execute(f"INSERT INTO {TABLE} VALUES ('r', 'a.png', 'Scans', 'now')")
+            self.assertIsNotNone(image_categories_schema_error(conn))
+            migrate_image_categories(conn)
+            self.assertIsNone(image_categories_schema_error(conn, allow_missing=False))
+            self.assertEqual(
+                conn.execute(f"SELECT COUNT(*) FROM {TABLE}").fetchone()[0], 1
+            )
+        with closing(sqlite3.connect(":memory:")) as conn:
+            migrate_image_categories(conn)
+            migrate_image_categories(conn)  # idempotent
+            self.assertIsNone(image_categories_schema_error(conn, allow_missing=False))
+
+    def test_values_refused_but_startup_checks_structure_only(self):
+        bad = {
+            "blank": "",
+            "101 characters": "x" * 101,
+            "control character": "a\tb",
+            "nul": "a\x00b",
+            "leading space": " Landscapes",
+            "trailing space": "Landscapes ",
+        }
+        for label, value in bad.items():
+            with self.subTest(label), closing(sqlite3.connect(":memory:")) as conn:
+                migrate_image_categories(conn)
+                conn.execute(
+                    f"INSERT INTO {TABLE} VALUES ('r', 'a.png', ?, 'now')", (value,)
+                )
+                self.assertIsNotNone(image_categories_schema_error(conn))
+                self.assertIsNone(
+                    image_categories_schema_error(conn, check_values=False)
+                )
+                migrate_image_categories(conn)  # startup still opens it
+        with closing(sqlite3.connect(":memory:")) as conn:
+            migrate_image_categories(conn)
+            conn.execute(
+                f"INSERT INTO {TABLE} VALUES ('r', ?, 'Scans', 'now')", ("x" * 5000,)
+            )
+            self.assertIsNotNone(image_categories_schema_error(conn))
+            conn.execute(f"DELETE FROM {TABLE}")
+            conn.execute(f"INSERT INTO {TABLE} VALUES ('r', x'00ff', 'Scans', 'now')")
+            self.assertIsNotNone(image_categories_schema_error(conn))
+            conn.execute(f"DELETE FROM {TABLE}")
+            for value in ("x" * 100, "Scans", "Café · 風景", "a b"):
+                conn.execute(
+                    f"INSERT OR REPLACE INTO {TABLE} VALUES ('r', 'a.png', ?, 'now')",
+                    (value,),
+                )
+                self.assertIsNone(image_categories_schema_error(conn), value)
+
+    def test_restore_check(self):
+        PromptDatabase(self.path).close_all()
+        PromptModel.close_all_instances()
+        ok, reason = PromptModel.verify_database_file(
+            self.path, require_library_state=True
+        )
+        self.assertTrue(ok, reason)
+        with closing(sqlite3.connect(self.path)) as conn:
+            conn.execute(f"DROP TABLE {TABLE}")
+            conn.commit()
+        # A backup from before #332 is accepted ...
+        ok, reason = PromptModel.verify_database_file(self.path)
+        self.assertTrue(ok, reason)
+        # ... but a current database must have the table.
+        ok, reason = PromptModel.verify_database_file(
+            self.path, require_library_state=True
+        )
+        self.assertFalse(ok)
+        self.assertIn("image categories", reason.lower())
+        with closing(sqlite3.connect(self.path)) as conn:
+            conn.execute(_DDL)
+            conn.execute(_INDEX_DDL)
+            conn.execute(f"INSERT INTO {TABLE} VALUES ('r', 'a.png', ' x', 'now')")
+            conn.commit()
+        ok, reason = PromptModel.verify_database_file(self.path)
+        self.assertFalse(ok)
+        self.assertIn("image categories", reason.lower())
+
+
+class TestImageCategoryStore(_TempDir):
+    def setUp(self):
+        super().setUp()
+        self.db = PromptDatabase(self.path)
+
+    def tearDown(self):
+        self.db.close_all()
+        super().tearDown()
+
+    def rows(self):
+        return (
+            self.db.model.get_connection()
+            .execute(
+                f"SELECT root_key, relative_path, category, updated_at FROM {TABLE} "
+                "ORDER BY root_key, relative_path"
+            )
+            .fetchall()
+        )
+
+    def test_set_counts_updated_and_unchanged(self):
+        keys = [("/out", "a.png"), ("/other", "a.png"), ("/out", "a.png")]
+        self.assertEqual(
+            self.db.set_image_categories(keys, "Portraits"),
+            {"updated": 2, "unchanged": 0},
+        )
+        stamps = {(r[0], r[1]): r[3] for r in self.rows()}
+        self.assertEqual(
+            self.db.set_image_categories(
+                [("/out", "a.png"), ("/out", "b.png")], "Portraits"
+            ),
+            {"updated": 1, "unchanged": 1},
+        )
+        # An unchanged file keeps its updated_at.
+        self.assertEqual(
+            {(r[0], r[1]): r[3] for r in self.rows()}[("/out", "a.png")],
+            stamps[("/out", "a.png")],
+        )
+        self.assertEqual(
+            self.db.set_image_categories([("/out", "a.png")], "Scans"),
+            {"updated": 1, "unchanged": 0},
+        )
+        self.assertEqual(self.db.image_category("/out", "a.png"), "Scans")
+        self.assertEqual(self.db.image_category("/other", "a.png"), "Portraits")
+        self.assertIsNone(self.db.image_category("/out", "missing.png"))
+        self.assertEqual(
+            self.db.own_categories_of(["/out", "/out", "/none"]),
+            {("/out", "a.png"): "Scans", ("/out", "b.png"): "Portraits"},
+        )
+
+    def test_clear_counts_removed_rows(self):
+        self.db.set_image_categories([("/out", "a.png"), ("/b", "c.png")], "Scans")
+        self.assertEqual(
+            self.db.clear_image_categories(
+                [("/out", "a.png"), ("/b", "c.png"), ("/out", "gone.png")]
+            ),
+            2,
+        )
+        self.assertEqual(self.rows(), [])
+        self.assertEqual(self.db.clear_image_categories([("/out", "a.png")]), 0)
+
+    def test_a_failed_write_writes_nothing(self):
+        self.db.set_image_categories([("/out", "a.png")], "Scans")
+        keys = [("/out", "b.png"), ("/out", None)]  # NOT NULL fails the second
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.db.set_image_categories(keys, "Portraits")
+        self.assertEqual([r[:3] for r in self.rows()], [("/out", "a.png", "Scans")])
+
+    def test_store_never_touches_prompt_categories(self):
+        conn = self.db.model.get_connection()
+        with conn:
+            conn.execute(
+                "INSERT INTO prompts (text, category, hash) VALUES ('p', 'L', 'h')"
+            )
+        before = category_checksum(conn)
+        seen = set()
+        conn.set_authorizer(
+            lambda action, arg1, *_: seen.add(arg1) or sqlite3.SQLITE_OK
+        )
+        try:
+            self.db.set_image_categories([("/out", "a.png")], "Scans")
+            self.db.own_categories_of(["/out"])
+            self.db.image_category("/out", "a.png")
+            self.db.clear_image_categories([("/out", "a.png")])
+        finally:
+            conn.set_authorizer(None)
+        self.assertFalse(seen & {"prompts", "prompt_pairs", "generated_images"}, seen)
+        self.assertEqual(category_checksum(conn), before)
+
+    def test_copy_keeps_kept_copy_category(self):
+        self.db.set_image_categories([("/out", "removed.png")], "Scans")
+        self.db.set_image_categories([("/out", "kept.png")], "Portraits")
+        with self.db._pair_transaction() as conn:
+            inserted = copy_image_category(
+                conn, ("/out", "removed.png"), ("/out", "kept.png")
+            )
+        self.assertEqual(inserted, 0)
+        self.assertEqual(self.db.image_category("/out", "kept.png"), "Portraits")
+        with self.db._pair_transaction() as conn:
+            inserted = copy_image_category(
+                conn, ("/out", "removed.png"), ("/out", "new.png")
+            )
+        self.assertEqual(inserted, 1)
+        self.assertEqual(self.db.image_category("/out", "new.png"), "Scans")
+        # The removed copy's row is never deleted.
+        self.assertEqual(self.db.image_category("/out", "removed.png"), "Scans")
+        with self.db._pair_transaction() as conn:
+            self.assertEqual(
+                copy_image_category(conn, ("/out", "none.png"), ("/out", "x.png")), 0
+            )
+            self.assertEqual(
+                copy_image_category(
+                    conn, ("/out", "removed.png"), ("/out", "removed.png")
+                ),
+                0,
+            )
+
+    def test_statements_never_scan_the_table(self):
+        conn = self.db.model.get_connection()
+        for sql in STATEMENTS:
+            with self.subTest(sql):
+                params = (None,) * sql.count("?")
+                plan = " ".join(
+                    row[-1] for row in conn.execute(f"EXPLAIN QUERY PLAN {sql}", params)
+                )
+                self.assertNotRegex(plan, rf"SCAN {TABLE}\b", plan)
+        plan = " ".join(
+            row[-1]
+            for row in conn.execute(
+                f"EXPLAIN QUERY PLAN SELECT rowid FROM {TABLE} WHERE category=? "
+                "ORDER BY rowid",
+                ("x",),
+            )
+        )
+        self.assertIn(INDEX, plan)
+        self.assertIn(COPY_SQL, STATEMENTS)
+
+
+class TestImageCategoryKey(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.base = Path(os.path.realpath(self._tmp.name))
+        self.outer = self.base / "output"
+        self.inner = self.outer / "nested"
+        (self.inner / "sub").mkdir(parents=True)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_equals_the_favorite_key_without_nesting(self):
+        from py.api.images import _favorite_relative_key, _favorite_root_key
+
+        key = image_category_key([self.outer], self.outer / "sub" / "a.png")
+        self.assertEqual(
+            key, (_favorite_root_key(self.outer), _favorite_relative_key("sub/a.png"))
+        )
+
+    def test_deepest_source_wins_in_any_order(self):
+        from py.api.images import _favorite_root_key
+
+        path = self.inner / "sub" / "a.png"
+        for sources in ([self.outer, self.inner], [self.inner, self.outer]):
+            with self.subTest(sources=sources):
+                self.assertEqual(
+                    image_category_key(sources, path),
+                    (_favorite_root_key(self.inner), "sub/a.png"),
+                )
+        # A file of the outer source only is keyed by it.
+        self.assertEqual(
+            image_category_key([self.outer, self.inner], self.outer / "b.png"),
+            (_favorite_root_key(self.outer), "b.png"),
+        )
+
+    def test_outside_every_source_has_no_key(self):
+        self.assertIsNone(image_category_key([self.inner], self.outer / "a.png"))
+        self.assertIsNone(image_category_key([], self.outer / "a.png"))
+        # A sibling sharing a name prefix is not inside.
+        self.assertIsNone(
+            image_category_key([self.outer], self.base / "output2" / "a.png")
+        )
+        self.assertIsNone(image_category_key([self.outer], self.outer))
+
+    def test_symlinked_subfolder_is_not_resolved(self):
+        target = self.base / "elsewhere"
+        target.mkdir()
+        link = self.outer / "linked"
+        try:
+            os.symlink(target, link, target_is_directory=True)
+        except (OSError, NotImplementedError):
+            self.skipTest("symlinks are not available")
+        key = image_category_key([self.outer], link / "a.png")
+        self.assertEqual(key[1], "linked/a.png")
+
+    def test_windows_folds_case_of_the_relative_key(self):
+        with patch("py.api.images._path_flavor", ntpath):
+            upper = image_category_key([self.outer], self.outer / "Sub" / "A.PNG")
+            lower = image_category_key([self.outer], self.outer / "sub" / "a.png")
+        self.assertEqual(upper[1], "sub/a.png")
+        self.assertEqual(upper, lower)
+
+    def test_posix_keeps_case_distinct(self):
+        if os.path.normcase("A") == "a":
+            self.skipTest("this OS folds case")
+        upper = image_category_key([self.outer], self.outer / "A.png")
+        lower = image_category_key([self.outer], self.outer / "a.png")
+        self.assertNotEqual(upper, lower)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pair_usage_tracking.py

@@ -85,6 +85,23 @@ class TestPairUsageTracking(unittest.TestCase):
             self.db.get_prompt_by_id(pair["negative_prompt_id"])["run_count"], 0
         )
 
+    def test_execution_records_a_cached_run_receipt_for_both_editors(self):
+        from utils.cached_runs import Facts
+
+        node = self.node()
+        node.cached_runs = Mock()
+        node.cached_runs.replayed.return_value = False
+        node.encode_prompt(
+            Mock(),
+            "cat",
+            negative_text="",
+            prompt=pair_graph(negative=""),
+            unique_id="1",
+        )
+        node.cached_runs.record_item.assert_called_once_with(
+            "1", ("cat", ""), Facts(usage_tracking.generate_prompt_hash("cat"), None)
+        )
+
     def test_validation_rejection_never_counts_or_saves(self):
         self.submit(pair_graph())
         self.assertEqual(self.db.list_prompt_pairs()["total"], 0)

tests/test_prompt_usage.py

@@ -278,5 +278,49 @@ class TestNodeRecordsUse(UsageTestCase):
         self.assertEqual(self.db.get_prompt_by_id(pid)["run_count"], 2)
 
 
+class TestNodeRecordsCachedRunReceipt(UsageTestCase):
+    """Each execution leaves a receipt for cached jobs (#347)."""
+
+    GRAPH = {
+        "1": {"class_type": "PromptManagerText", "inputs": {"text": "typed text"}},
+    }
+
+    def _node(self):
+        from unittest import mock
+
+        from prompt_manager_text import PromptManagerText
+
+        node = PromptManagerText.__new__(PromptManagerText)
+        node.db = self.db
+        node.logger = __import__("logging").getLogger("test.prompt_usage")
+        node.prompt_tracker = mock.Mock()
+        node.comfyui_integration = mock.Mock()
+        node._inject_lora_trigger_words = lambda text: text
+        node.cached_runs = mock.Mock()
+        node.cached_runs.replayed.return_value = False
+        return node
+
+    def test_receipt_holds_the_returned_text_and_the_stored_hash(self):
+        from utils.cached_runs import Facts
+
+        node = self._node()
+        node.process_text(
+            "typed text", prepend_text="pre", prompt=self.GRAPH, unique_id="1"
+        )
+        node.cached_runs.record_item.assert_called_once_with(
+            "1", ("pre typed text",), Facts(generate_prompt_hash("typed text"))
+        )
+        # Hook-visible text: the node counts nothing itself
+        row = self.db.get_prompt_by_hash(generate_prompt_hash("typed text"))
+        self.assertEqual(row["run_count"], 0)
+
+    def test_empty_editor_records_empty_facts(self):
+        from utils.cached_runs import Facts
+
+        node = self._node()
+        node.process_text("  ", prompt=self.GRAPH, unique_id="1")
+        node.cached_runs.record_item.assert_called_once_with("1", ("  ",), Facts(None))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/vendor/comfyui-config.test.mjs

@@ -33,15 +33,17 @@ test('execution script discovers only the new matrix and preserves all 78 origin
     const executions = list('tools/vendor/playwright.execution.config.mjs');
     assert.match(executions, /\[classic-start\]/);
     assert.match(executions, /\[nodes2-start\]/);
-    // + the fresh-host library location case, + #342 image attribution, + #343 linked text
-    assert.match(executions, /Total: 90 tests in 5 files/);
+    // + the fresh-host library location case, + #342 image attribution, + #343 linked text,
+    // + #332 image category routes on a generated output (image-attribution.spec.mjs, 2 projects),
+    // + #347 cached runtime-linked runs: five node kinds and eight one-renderer cases (execution-text-link.spec.mjs, 2 projects)
+    assert.match(executions, /Total: 118 tests in 5 files/);
     for (const project of ['classic-start', 'nodes2-start']) {
         const cases = executions.split('\n').filter(line => line.includes(`[${project}]`));
         assert.equal(cases.filter(line => line.includes('execution-history.spec.mjs')).length, 15);
         assert.equal(cases.filter(line => line.includes('execution-outputs.spec.mjs')).length, 8);
         assert.equal(cases.filter(line => line.includes('library-location.spec.mjs')).length, 1);
-        assert.equal(cases.filter(line => line.includes('image-attribution.spec.mjs')).length, 2);
-        assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 19);
+        assert.equal(cases.filter(line => line.includes('image-attribution.spec.mjs')).length, 3);
+        assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 32);
     }
     assert.doesNotMatch(executions, /prompt-manager.spec/);
 });

utils/cached_runs.py

@@ -0,0 +1,532 @@
+"""Count a cached PromptManager node like the execution it reuses (#347).
+
+The queue hook (``usage_tracking``) counts text it can read from the queued
+graph. Text only known at run time (a PreviewAny or LLM link, a list node) is
+counted by the node itself when it executes. When ComfyUI later serves the node
+from its output cache the node does not execute, so without this module that
+job is not counted.
+
+When a PromptManager-family node executes it records a *receipt item*: the
+role-independent facts its own count is made from (the stored text hashes)
+and digests of the STRING outputs it returned. When ComfyUI reports the node
+as cached for a later job (``execution_cached``), the receipt whose digests
+equal the text in the cache entry is replayed once for that job. The node's
+role (positive, negative, which Pair outputs are used) and what the hook
+already saw are recomputed from that job's own graph, and both paths count
+through the same ``own_counts`` so they cannot drift.
+
+Fail closed: when the cached text cannot be matched to a receipt recorded in
+this process, nothing is counted (the behavior before #347). In memory only and
+bounded; starts no threads.
+"""
+
+import hashlib
+import threading
+from collections import OrderedDict
+from typing import Any, Callable, Dict, FrozenSet, List, NamedTuple, Optional, Tuple
+
+try:
+    from .hashing import generate_prompt_hash
+    from .logging_config import get_logger
+    from .prompt_graph import (
+        PROMPT_MANAGER_TYPES,
+        intrinsic_pair_nodes,
+        resolve_text,
+        run_prompt_components,
+        run_prompt_nodes,
+    )
+    from .prompt_pair_resolver import resolve_text_input
+    from .queue_observers import subscribe_dequeue, subscribe_task_done
+    from .run_keys import prompt_node_key, upstream_nodes
+    from .usage_tracking import find_pair_id_by_hashes, survive_switch
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+    from utils.logging_config import get_logger
+    from utils.prompt_graph import (
+        PROMPT_MANAGER_TYPES,
+        intrinsic_pair_nodes,
+        resolve_text,
+        run_prompt_components,
+        run_prompt_nodes,
+    )
+    from utils.prompt_pair_resolver import resolve_text_input
+    from utils.queue_observers import subscribe_dequeue, subscribe_task_done
+    from utils.run_keys import prompt_node_key, upstream_nodes
+    from utils.usage_tracking import find_pair_id_by_hashes, survive_switch
+
+logger = get_logger("prompt_manager.cached_runs")
+
+PAIR = "PromptManagerPair"
+MAX_KEYS = 256  # run keys remembered (the same bound as MAX_KEYED_RUNS)
+MAX_RECEIPTS = 4  # receipts per run key: a recurring upstream value (A, B, A)
+MAX_ITEMS = 64  # list items per receipt (the cache reader's bound)
+MAX_CACHED_NODES = 500  # the execution_cached guard the history capture uses
+_EMPTY_HASH = generate_prompt_hash("")
+
+Count = Tuple[Any, ...]  # ("prompt", hash) or ("pair", positive, negative or None)
+
+
+class Facts(NamedTuple):
+    """What one execution stored: hashes of the stored text, None when nothing.
+
+    ``positive_hash`` is None for an empty editor (or saving disabled);
+    ``negative_hash`` (Pair only) is None for an empty negative, which the pair
+    stores as ``negative_prompt_id IS NULL``.
+    """
+
+    positive_hash: Optional[str]
+    negative_hash: Optional[str] = None
+
+
+class Role(NamedTuple):
+    """How a job's graph uses the node: which editors reach a sampler's positive."""
+
+    used_fields: FrozenSet[str]
+    pair_used: bool = False
+
+
+class Seen(NamedTuple):
+    """Which of the node's texts the queue hook resolved, and so already counted."""
+
+    text: bool = False
+    negative_text: bool = False
+    pair: bool = False
+
+
+class _Item(NamedTuple):
+    outputs: Tuple[str, ...]
+    facts: Facts
+
+
+class _Receipt:
+    __slots__ = ("item_id", "items")
+
+    def __init__(self, item_id: int) -> None:
+        self.item_id = item_id
+        self.items: Dict[int, _Item] = {}
+
+    def outputs(self) -> Dict[int, Tuple[str, ...]]:
+        return {index: item.outputs for index, item in self.items.items()}
+
+
+class _Attempt(NamedTuple):
+    item_id: int
+    prompt_id: str
+    graph: Dict[str, Any]
+    nodes: Dict[str, Tuple[str, str]]  # node id -> (class, run key)
+    replayed: set
+
+
+def text_digest(value: str) -> str:
+    """Digest of one returned STRING value, exactly as returned."""
+    return hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()
+
+
+def stored_facts(text: Any, negative_text: Any = None) -> Facts:
+    """The facts of an execution that stores ``text`` (and a Pair's negative)."""
+    positive = text if isinstance(text, str) and text.strip() else None
+    negative = (
+        negative_text
+        if isinstance(negative_text, str) and negative_text.strip()
+        else None
+    )
+    return Facts(
+        generate_prompt_hash(positive) if positive is not None else None,
+        generate_prompt_hash(negative) if negative is not None else None,
+    )
+
+
+def pair_components(facts: Facts, role: Role) -> List[Tuple[str, str]]:
+    """``(field, hash)`` of each used, stored Pair component; each hash once."""
+    found, counted = [], set()
+    for field, prompt_hash in (
+        ("text", facts.positive_hash),
+        ("negative_text", facts.negative_hash),
+    ):
+        if field not in role.used_fields or prompt_hash is None:
+            continue
+        if prompt_hash in counted:
+            continue
+        counted.add(prompt_hash)
+        found.append((field, prompt_hash))
+    return found
+
+
+def own_counts(node_type: str, facts: Facts, role: Role, seen: Seen) -> List[Count]:
+    """What one execution of a node counts itself: text the hook could not see.
+
+    Shared by the executing node and by replay. Excludes the pending first-use
+    catch-up, which only an execution that saves the prompt can consume.
+    """
+    if facts.positive_hash is None:
+        return []
+    if node_type != PAIR:
+        if "text" in role.used_fields and not seen.text:
+            return [("prompt", facts.positive_hash)]
+        return []
+    counts: List[Count] = [
+        ("prompt", prompt_hash)
+        for field, prompt_hash in pair_components(facts, role)
+        if not getattr(seen, field)
+    ]
+    if role.pair_used and not seen.pair:
+        counts.append(("pair", facts.positive_hash, facts.negative_hash))
+    return counts
+
+
+def graph_role(graph: Any, node_id: str, node_type: str) -> Role:
+    """The node's role in a queued graph, by the functions the nodes use."""
+    if node_type != PAIR:
+        used = {"text"} if node_id in run_prompt_nodes(graph) else set()
+        return Role(frozenset(used))
+    used = {
+        field for source, field in run_prompt_components(graph) if source == node_id
+    }
+    return Role(frozenset(used), node_id in intrinsic_pair_nodes(graph))
+
+
+def _same(resolved: Optional[str], prompt_hash: Optional[str]) -> bool:
+    return resolved is not None and generate_prompt_hash(resolved) == prompt_hash
+
+
+def graph_seen(graph: Any, node_id: str, node_type: str, facts: Facts) -> Seen:
+    """Which texts the queue hook resolved from ``graph`` (as the node decides)."""
+    if node_type != PAIR:
+        return Seen(text=_same(resolve_text(graph, node_id), facts.positive_hash))
+    positive = _same(resolve_text_input(graph, node_id, "text"), facts.positive_hash)
+    negative = _same(
+        resolve_text_input(graph, node_id, "negative_text"),
+        facts.negative_hash or _EMPTY_HASH,
+    )
+    return Seen(positive, negative, positive and negative)
+
+
+def _host_context() -> Tuple[Optional[str], Optional[int]]:
+    """(prompt id, list index) of the node executing on this thread, if known."""
+    try:
+        from comfy_execution.utils import get_executing_context
+    except Exception:
+        return None, None
+    try:
+        context = get_executing_context()
+    except Exception:
+        return None, None
+    prompt_id = getattr(context, "prompt_id", None)
+    index = getattr(context, "list_index", None)
+    return (
+        prompt_id if isinstance(prompt_id, str) and prompt_id else None,
+        index if type(index) is int else None,
+    )
+
+
+def write_count(db: Any, count: Count) -> bool:
+    """Record one counted run; False when the prompt or pair no longer exists."""
+
+    def write() -> bool:
+        if count[0] == "prompt":
+            row = db.get_prompt_by_hash(count[1])
+            if row:
+                db.record_prompt_use(row["id"])
+            return bool(row)
+        pair_id = find_pair_id_by_hashes(db, count[1], count[2])
+        if pair_id:
+            db.record_prompt_pair_use(pair_id)
+        return bool(pair_id)
+
+    return bool(survive_switch(db, write))
+
+
+class CachedRuns:
+    """Receipts of executed nodes and the one job currently executing."""
+
+    def __init__(
+        self,
+        *,
+        max_keys: int = MAX_KEYS,
+        max_receipts: int = MAX_RECEIPTS,
+        max_items: int = MAX_ITEMS,
+        context: Callable[[], Tuple[Optional[str], Optional[int]]] = _host_context,
+    ) -> None:
+        self._max_keys = max_keys
+        self._max_receipts = max_receipts
+        self._max_items = max_items
+        self._context = context
+        self._receipts: "OrderedDict[str, List[_Receipt]]" = OrderedDict()
+        self._attempt: Optional[_Attempt] = None
+        self._lock = threading.Lock()
+
+    # -- the job ------------------------------------------------------------
+
+    def begin_attempt(self, result: Any) -> None:
+        """A job was dequeued: ``result`` is ``PromptQueue.get``'s return value."""
+        attempt = None
+        try:
+            item, item_id = result[0], result[1]
+            prompt_id, graph, outputs = item[1], item[2], item[4]
+            if type(item_id) is int and isinstance(prompt_id, str):
+                if isinstance(graph, dict):
+                    attempt = _Attempt(
+                        item_id, prompt_id, graph, _job_nodes(graph, outputs), set()
+                    )
+        except Exception as error:
+            logger.debug(f"Dequeued job not tracked for cached runs: {error}")
+        with self._lock:
+            self._attempt = attempt
+
+    def end_attempt(self, item_id: Any) -> None:
+        """The job finished: its receipts are complete and replace equal older ones.
+
+        Deduplicating only complete receipts keeps a list execution that is
+        still recording ([A] so far of [A, B]) from dropping an older [A].
+        """
+        with self._lock:
+            if self._attempt is not None and self._attempt.item_id == item_id:
+                self._attempt = None
+            for run_key, receipts in self._receipts.items():
+                if receipts and receipts[0].item_id == item_id:
+                    self._receipts[run_key] = _deduplicated(receipts)
+
+    def _current(self) -> Tuple[Optional[_Attempt], Optional[int]]:
+        """The attempt the executing node belongs to, and its list index."""
+        prompt_id, index = self._context()
+        with self._lock:
+            attempt = self._attempt
+        if attempt is None or prompt_id is None or prompt_id != attempt.prompt_id:
+            return None, None
+        return attempt, index
+
+    # -- node execution -----------------------------------------------------
+
+    def replayed(self, node_id: Any) -> bool:
+        """Whether this job already counted the executing node by replay."""
+        attempt, _ = self._current()
+        if attempt is None:
+            return False
+        with self._lock:
+            return str(node_id) in attempt.replayed
+
+    def record_item(self, node_id: Any, outputs: Tuple[Any, ...], facts: Facts) -> None:
+        """Remember what this execution returned and was entitled to count."""
+        try:
+            attempt, index = self._current()
+            node = attempt.nodes.get(str(node_id)) if attempt else None
+            if node is None:
+                return
+            index = 0 if index is None else index
+            if not 0 <= index < self._max_items:
+                return
+            if any(not isinstance(value, str) for value in outputs):
+                return
+            item = _Item(tuple(text_digest(value) for value in outputs), facts)
+            self._store(node[1], attempt.item_id, index, item)
+        except Exception as error:
+            logger.debug(f"Could not record a cached-run receipt: {error}")
+
+    def _store(self, run_key: str, item_id: int, index: int, item: _Item) -> None:
+        with self._lock:
+            receipts = self._receipts.pop(run_key, [])
+            if receipts and receipts[0].item_id == item_id:
+                receipt = receipts[0]
+            else:
+                # The previous job's receipt is complete (its task_done may
+                # never have fired): it replaces older receipts it equals.
+                receipts = _deduplicated(receipts)
+                receipt = _Receipt(item_id)
+                receipts.insert(0, receipt)
+            receipt.items[index] = item
+            self._receipts[run_key] = receipts[: self._max_receipts]
+            while len(self._receipts) > self._max_keys:
+                self._receipts.popitem(last=False)
+
+    # -- cached nodes -------------------------------------------------------
+
+    def replay(
+        self,
+        executor: Any,
+        prompt_id: Any,
+        nodes: Any,
+        db_factory: Callable[[], Any],
+        reader: Optional[Callable[[Any, str, str], Any]] = None,
+    ) -> None:
+        """Count the job's cached nodes whose cached text a receipt proves."""
+        if reader is None:
+            reader = _host().cached_string_outputs
+        with self._lock:
+            attempt = self._attempt
+        if attempt is None or prompt_id != attempt.prompt_id:
+            return
+        if type(nodes) not in (list, tuple) or len(nodes) > MAX_CACHED_NODES:
+            return
+        db = None
+        for node_id in nodes:
+            node = attempt.nodes.get(node_id) if type(node_id) is str else None
+            if node is None:
+                continue
+            node_type, run_key = node
+            items = self._match(run_key, reader(executor, node_id, node_type))
+            if items is None:
+                logger.debug(
+                    f"Cached node {node_id} not counted: no matching execution"
+                )
+                continue
+            with self._lock:
+                if node_id in attempt.replayed:
+                    continue  # a repeated message or id never counts twice
+                attempt.replayed.add(node_id)
+            counts: List[Count] = []
+            role = graph_role(attempt.graph, node_id, node_type)
+            for item in items:
+                seen = graph_seen(attempt.graph, node_id, node_type, item.facts)
+                counts.extend(own_counts(node_type, item.facts, role, seen))
+            if not counts:
+                continue
+            try:
+                db = db if db is not None else db_factory()
+                for count in counts:
+                    if not write_count(db, count):
+                        logger.info(
+                            f"Cached node {node_id}: its {count[0]} no longer exists, "
+                            "not counted"
+                        )
+            except Exception as error:
+                logger.warning(f"Could not count cached node {node_id}: {error}")
+
+    def _match(self, run_key: str, values: Any) -> Optional[List[_Item]]:
+        """The items of the one receipt whose outputs equal ``values``, in order."""
+        if not values:
+            return None
+        length = len(values[0])
+        if length == 0 or length > self._max_items:
+            return None
+        if any(len(slot) != length for slot in values):
+            return None
+        wanted = {
+            index: tuple(text_digest(slot[index]) for slot in values)
+            for index in range(length)
+        }
+        with self._lock:
+            receipts = self._receipts.get(run_key)
+            if not receipts:
+                return None
+            matches = [receipt for receipt in receipts if receipt.outputs() == wanted]
+            if not matches:
+                return None
+            self._receipts.move_to_end(run_key)
+            facts = {
+                tuple(receipt.items[index].facts for index in range(length))
+                for receipt in matches
+            }
+            if len(facts) != 1:
+                return None
+            return [matches[0].items[index] for index in range(length)]
+
+
+def _deduplicated(receipts: List[_Receipt]) -> List[_Receipt]:
+    """The newest receipt replaces older ones with the same outputs."""
+    if not receipts:
+        return receipts
+    outputs = receipts[0].outputs()
+    return [receipts[0]] + [r for r in receipts[1:] if r.outputs() != outputs]
+
+
+def _job_nodes(graph: Dict[str, Any], outputs: Any) -> Dict[str, Tuple[str, str]]:
+    """PromptManager-family nodes the job's outputs need, with their run keys."""
+    starts = (
+        [str(node) for node in outputs] if isinstance(outputs, (list, tuple)) else []
+    )
+    needed = upstream_nodes(graph, starts)
+    nodes = {}
+    for node_id in needed:
+        node = graph.get(node_id)
+        node_type = node.get("class_type") if isinstance(node, dict) else None
+        if node_type not in PROMPT_MANAGER_TYPES:
+            continue
+        run_key = prompt_node_key(graph, node_id)
+        if run_key is not None:
+            nodes[node_id] = (node_type, run_key)
+    return nodes
+
+
+def _host() -> Any:
+    """The host adapter module (imported late: it needs ComfyUI's modules)."""
+    try:
+        from . import execution_context_host
+    except ImportError:
+        from utils import execution_context_host
+    return execution_context_host
+
+
+# Shared between the nodes and the host observers
+CACHED_RUNS = CachedRuns()
+
+_registered = threading.Event()
+_register_lock = threading.Lock()
+
+
+def register_cached_runs(
+    server: Any, db_factory: Callable[[], Any], runs: Optional[CachedRuns] = None
+) -> bool:
+    """Observe dequeue, ``execution_cached`` and ``task_done``; once per process.
+
+    Registers nothing when the host lacks any of them: counting is then as
+    before #347.
+    """
+    runs = CACHED_RUNS if runs is None else runs
+    with _register_lock:
+        if _registered.is_set():
+            return False
+        queue = getattr(server, "prompt_queue", None)
+        if not callable(getattr(queue, "get", None)) or not callable(
+            getattr(queue, "task_done", None)
+        ):
+            logger.info("Cached runs not counted: the prompt queue is not observable")
+            return False
+        try:
+            from execution import PromptExecutor
+        except Exception:
+            logger.info("Cached runs not counted: no ComfyUI executor")
+            return False
+        if not callable(getattr(PromptExecutor, "add_message", None)):
+            logger.info("Cached runs not counted: executor messages unavailable")
+            return False
+        subscribe_executor_messages = _host().subscribe_executor_messages
+
+        def on_message(executor: Any, args: tuple, kwargs: dict) -> None:
+            try:
+                event = args[0] if args else kwargs.get("event")
+                if event != "execution_cached":
+                    return
+                server_of = getattr(executor, "server", None)
+                if getattr(server_of, "prompt_queue", None) is not queue:
+                    return
+                data = args[1] if len(args) > 1 else kwargs.get("data")
+                if type(data) is not dict:
+                    return
+                runs.replay(
+                    executor, data.get("prompt_id"), data.get("nodes"), db_factory
+                )
+            except Exception as error:
+                logger.warning(f"Could not count cached nodes: {error}")
+
+        unsubscribers = []
+        try:
+            unsubscribers.append(
+                subscribe_dequeue(queue, "cached_runs", runs.begin_attempt)
+            )
+            unsubscribers.append(
+                subscribe_task_done(
+                    queue,
+                    "cached_runs",
+                    lambda args, kwargs: runs.end_attempt(
+                        args[0] if args else kwargs.get("item_id")
+                    ),
+                )
+            )
+            unsubscribers.append(subscribe_executor_messages("cached_runs", on_message))
+        except Exception:
+            for unsubscribe in unsubscribers:
+                unsubscribe()
+            raise
+        _registered.set()
+        logger.info("Counting cached runs of runtime-linked prompt text")
+        return True

utils/execution_context_host.py

@@ -75,6 +75,82 @@ def subscribe_executor_messages(key: str, callback: Callable) -> Callable[[], No
         return state.subscribe(key, callback)
 
 
+# STRING output slots of each PromptManager-family node, in output order
+STRING_SLOTS = {
+    "PromptManager": (1,),
+    "PromptManagerFocused": (1,),
+    "PromptManagerSplit": (1,),
+    "PromptManagerText": (0,),
+    "PromptManagerPair": (2, 3),
+}
+MAX_CACHED_VALUES = 64
+
+
+def _local_cache_reader(executor: object) -> Callable | None:
+    """``node_id -> outputs`` of the executor's local output cache, or None.
+
+    Only the supported cache classes, only ``get_local`` (never a provider or
+    an async lookup), only a real ``CacheEntry`` with at most four outputs.
+    """
+    from comfy_execution import caching
+    from execution import CacheEntry
+
+    cache = executor.caches.outputs
+    supported = tuple(
+        cls
+        for name in ("HierarchicalCache", "LRUCache", "RAMPressureCache")
+        if isinstance(cls := getattr(caching, name, None), type)
+    )
+    if type(cache) not in supported:
+        return None
+    local = getattr(cache, "get_local", None)
+    if not callable(local):
+        return None
+
+    def read(node_id: str) -> list | tuple | None:
+        entry = local(node_id)
+        if type(entry) is not CacheEntry:
+            return None
+        outputs = entry.outputs
+        if type(outputs) not in (list, tuple) or not 0 < len(outputs) <= 4:
+            return None
+        return outputs
+
+    return read
+
+
+def cached_string_outputs(
+    executor: object, node_id: str, node_type: str
+) -> tuple[tuple[str, ...], ...] | None:
+    """One tuple per STRING slot of a cached node's local entry, or None (#347).
+
+    None when the cache or entry is not supported, a slot is missing or holds
+    more than 64 values, any value is not a ``str``, or the slots differ in
+    length. Never runs a provider or an async lookup.
+    """
+    try:
+        slots = STRING_SLOTS.get(node_type)
+        if slots is None or type(node_id) is not str:
+            return None
+        local = _local_cache_reader(executor)
+        outputs = local(node_id) if local is not None else None
+        if outputs is None or not max(slots) < len(outputs):
+            return None
+        found = []
+        for slot in slots:
+            values = outputs[slot]
+            if type(values) not in (list, tuple) or len(values) > MAX_CACHED_VALUES:
+                return None
+            if any(type(value) is not str for value in values):
+                return None
+            found.append(tuple(values))
+        if len({len(values) for values in found}) != 1:
+            return None
+        return tuple(found)
+    except Exception:
+        return None
+
+
 def observe_cached_context(
     executor: object, message_type: object, message_data: object
 ) -> None:
@@ -101,19 +177,8 @@ def observe_cached_context(
         nodes = message_data.get("nodes")
         if type(nodes) not in (list, tuple) or len(nodes) > 500:
             return
-        from comfy_execution import caching
-        from execution import CacheEntry
-
-        cache = executor.caches.outputs
-        supported = tuple(
-            cls
-            for name in ("HierarchicalCache", "LRUCache", "RAMPressureCache")
-            if isinstance(cls := getattr(caching, name, None), type)
-        )
-        if type(cache) not in supported:
-            return
-        local = getattr(cache, "get_local", None)
-        if not callable(local):
+        local = _local_cache_reader(executor)
+        if local is None:
             return
         # Only source IDs retained by the bounded submitted graph are queried.
         for source in cap.sources:
@@ -124,11 +189,8 @@ def observe_cached_context(
                 if source.node_type == "PromptManagerPair"
                 else 0 if source.node_type == "PromptManagerText" else 1
             )
-            entry = local(source.node_id)
-            if type(entry) is not CacheEntry:
-                continue
-            outputs = entry.outputs
-            if type(outputs) not in (list, tuple) or not slot < len(outputs) <= 4:
+            outputs = local(source.node_id)
+            if outputs is None or not slot < len(outputs):
                 continue
             values = outputs[slot]
             if type(values) not in (list, tuple) or len(values) > 64:

utils/run_keys.py

@@ -51,6 +51,53 @@ def _links(inputs: Any) -> List[str]:
     return [value[0] for value in inputs.values() if is_link(value)]
 
 
+def upstream_closure(
+    graph: Dict[str, Any], starts: List[str], limit: Optional[int] = None
+) -> Optional[Dict[str, Any]]:
+    """``{node_id: [class_type, inputs]}`` of ``starts`` and everything upstream.
+
+    None when a node or an upstream link's node is missing, or the closure
+    exceeds ``limit`` nodes.
+    """
+    closure: Dict[str, Any] = {}
+    stack = list(starts)
+    while stack:
+        current = stack.pop()
+        if current in closure:
+            continue
+        node = graph.get(current)
+        if not isinstance(node, dict):
+            return None
+        if limit is not None and len(closure) >= limit:
+            return None
+        inputs = node.get("inputs")
+        closure[current] = [node.get("class_type"), inputs]
+        stack.extend(_links(inputs))
+    return closure
+
+
+def upstream_nodes(graph: Any, starts: List[str]) -> List[str]:
+    """Ids of ``starts`` and every node upstream of them that exists in ``graph``.
+
+    Used to tell which nodes a job's outputs to execute need ("run selected
+    output" jobs leave the rest of the graph out). Missing nodes are skipped.
+    """
+    if not isinstance(graph, dict):
+        return []
+    found: Dict[str, None] = {}
+    stack = [str(node) for node in starts]
+    while stack:
+        current = stack.pop()
+        if current in found:
+            continue
+        node = graph.get(current)
+        if not isinstance(node, dict):
+            continue
+        found[current] = None
+        stack.extend(_links(node.get("inputs")))
+    return list(found)
+
+
 def prompt_node_key(graph: Any, node_id: Any) -> Optional[str]:
     """Digest of ``node_id`` and its whole upstream closure, or None.
 
@@ -62,20 +109,9 @@ def prompt_node_key(graph: Any, node_id: Any) -> Optional[str]:
     if not isinstance(graph, dict) or node_id is None:
         return None
     start = str(node_id)
-    closure: Dict[str, Any] = {}
-    stack = [start]
-    while stack:
-        current = stack.pop()
-        if current in closure:
-            continue
-        node = graph.get(current)
-        if not isinstance(node, dict):
-            return None
-        if len(closure) >= MAX_VISITS:
-            return None
-        inputs = node.get("inputs")
-        closure[current] = [node.get("class_type"), inputs]
-        stack.extend(_links(inputs))
+    closure = upstream_closure(graph, [start], MAX_VISITS)
+    if closure is None:
+        return None
     try:
         encoded = json.dumps(
             [RUN_KEY_VERSION, start, closure],

utils/usage_tracking.py

@@ -16,7 +16,9 @@ it never counts. Servers without a ``prompt_queue`` fall back to counting at
 submit time.
 
 Text from known pure string nodes is resolved at queue time too. Anything else
-(PromptSearchList batches, unknown nodes) is counted by the node as it executes.
+(PromptSearchList batches, unknown nodes) is counted by the node as it executes,
+and when ComfyUI serves such a node from its cache, the job replays what that
+execution counted (``cached_runs``, #347).
 """
 
 import threading
@@ -167,13 +169,22 @@ def queued_pair_runs(graph: Any) -> List[Tuple[str, str, str]]:
 
 def find_pair_id(db: Any, positive: str, negative: str) -> Optional[int]:
     """Find a saved pair without creating or overwriting its metadata."""
-    positive_row = db.get_prompt_by_hash(generate_prompt_hash(positive))
-    negative_row = (
-        db.get_prompt_by_hash(generate_prompt_hash(negative))
-        if negative.strip()
-        else None
+    return find_pair_id_by_hashes(
+        db,
+        generate_prompt_hash(positive),
+        generate_prompt_hash(negative) if negative.strip() else None,
     )
-    if not positive_row or (negative.strip() and not negative_row):
+
+
+def find_pair_id_by_hashes(
+    db: Any, positive_hash: str, negative_hash: Optional[str]
+) -> Optional[int]:
+    """The saved pair of two prompt hashes; ``None`` negative is the empty one."""
+    positive_row = db.get_prompt_by_hash(positive_hash)
+    negative_row = (
+        db.get_prompt_by_hash(negative_hash) if negative_hash is not None else None
+    )
+    if not positive_row or (negative_hash is not None and not negative_row):
         return None
     row = (
         db.model.get_connection()