fix(runs): count cached PromptManager nodes fed by a runtime text link on every job (#347) (#423)

* fix(usage): count cached re-runs of runtime-linked prompt text (#347)

A PromptManager-family node whose text comes from a runtime link
(PreviewAny, an LLM, a list node) counted only the job that executed it;
later jobs served from ComfyUI's output cache counted nothing.

Each execution now records a receipt (digests of its STRING outputs and
hashes of the text it stores) per run key and queue item. On
execution_cached the receipt whose digests equal the cache entry's text is
replayed once for the job, with the node's role and what the queue hook
already saw recomputed from the job's own graph. The executing node and the
replay count through one shared own_counts. Unmatched cache entries count
nothing, as before; a node that executes after its replay in the same job
skips its own count. No schema change, no backfill.

* fix(usage): never replay a node twice in one job; dedupe receipts once complete (#347)

A repeated execution_cached id or message for the same job now counts the
node once. Receipts replace equal older ones only when their job finishes
(or the next job starts), so a list still recording [A] of [A, B] no longer
drops an older [A] receipt. Node-level tests check the receipt each
execution records.

* 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

@@ -5283,7 +5283,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
@@ -5297,6 +5297,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."""

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

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/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_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

@@ -34,15 +34,16 @@ test('execution script discovers only the new matrix and preserves all 78 origin
     assert.match(executions, /\[classic-start\]/);
     assert.match(executions, /\[nodes2-start\]/);
     // + 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)
-    assert.match(executions, /Total: 92 tests in 5 files/);
+    // + #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, 3);
-        assert.equal(cases.filter(line => line.includes('execution-text-link.spec.mjs')).length, 19);
+        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()