Merge remote-tracking branch 'origin/main' into fix/449-image-tags-section

TESTING.md

@@ -2471,13 +2471,15 @@ Category". It is built only from the concept's Images parts:
   Shift+Tab across windowed cards reach it (the card's controls are `button,
   input[type=checkbox]`); Home, End and the arrow keys still move between titles.
 - `div#library-bulkbar` sits between the toolbar and the caption, empty until something
-  is checked, then the concept's `.bulkbar`: "N selected", Set category, Clear. The
+  is checked, then the concept's `.bulkbar`: "N selected", Set category, Clear (D151
+  adds Add tags before Set category, in the Images bar's order). The
   selection is a set of prompt ids over the loaded shelf: windowed-out cards keep their
   state, Load more keeps it, and any shelf reload (search, category, sort, tags, a
   write) and leaving the page clear it. Clicking a checkbox toggles it and sets the
   anchor; Shift+click applies the anchor's range with the box's new state (the D135
   helpers). Escape in the bar and Clear clear it, focus to the heading. There is no
-  Ctrl/Cmd+A, Shift+arrow, long-press or "Select all N matching" on the Library.
+  Shift+arrow, long-press or "Select all N matching" on the Library (D151 adds
+  Ctrl/Cmd+A and "Select all N loaded").
 - Set category opens the shared `simpleInput` dialog ("Categorize selected prompts",
   `input#action-input` labelled Category, at most 100 characters, the Library's
   categories as suggestions). A blank value keeps it open; a control character is
@@ -2499,7 +2501,8 @@ Category". It is built only from the concept's Images parts:
   storage was restored since this page loaded. No category was changed." and a reload.
   No answer, 502, 503 or 504: "The category change could not be confirmed. The Library
   was reloaded." Any other refusal: "Category could not be set. Try again.", the
-  selection kept. Set category is `aria-disabled` while the write runs.
+  selection kept. Every bar action (and D151's select-all offer) is `aria-disabled` while
+  a write runs.
 - The bar's DOM and write loop live in `library-selection.mjs`; the toast text
   (`setOutcome` with the noun "prompt"), the range helpers, the count and the dialog are
   the Images bar's. One bar core shared with `images-selection.mjs` is a follow-up.
@@ -2513,6 +2516,43 @@ mutation); units `tests/js/workspace-library-selection.test.js`,
 `tests/js/workspace-library-state.test.js` (the epoch); Python
 `tests/test_bulk_library_category.py`. A Python change: ComfyUI needs a restart.
 
+### Adding tags to selected prompts and selecting all (#442)
+
+**D151** (beyond the concept, on D148's bar): the Prompt library adds tags to the selected
+prompts in one step, the legacy admin page's bulk "Add Tags", and selects every loaded
+prompt, as its select-all checkbox did (#444 ruling). The concept has neither.
+
+- The bar reads "N selected", Add tags, Set category, Clear: the Images bar's order.
+- Select all of the loaded shelf: Ctrl/Cmd+A in the grid (Images' key check; the search
+  box keeps its text meaning) selects every loaded prompt, windowed-out cards included,
+  and Escape in the grid clears it with focus kept on the card. "Select all N loaded"
+  follows the count while some loaded prompts are unchecked and at most 1,000 are loaded;
+  it selects them all and focus moves to Add tags; N follows each loaded batch. It sits
+  where D135's "Select all N matching" sits, but that offer appears once every loaded
+  image is checked; this one never reaches beyond the loaded prompts. No "all matching".
+- Add tags opens the shared `simpleInput` dialog ("Tag selected prompts", "Tags,
+  separated by commas", the tag vocabulary `GET /prompt_manager/tags` as suggestions, the
+  list the editor offers). The field refuses no tag, more than 20, a tag over 50
+  characters or with a control character, each with its sentence. Apply posts
+  `{prompt_ids, tags, expected_epoch}` to `POST /prompt_manager/bulk/tags`: prompt tags
+  only, never image tags, auto-tags or pair tags. Names are normalized as image tags are
+  (trim, no commas, casefold repeats folded); registry rows are found and inserted by
+  their exact name and only when some prompt gets them; a prompt gets only the names it
+  lacks by casefold. A prompt whose tags would pass 20 is skipped (`over_limit`), so the
+  editor can still save it. The answer is `{library_epoch, tags, updated, unchanged,
+  not_found, over_limit}`; the toast "Added “a, b” to N prompts." adds already-had,
+  skipped and not-found sentences. Storage restored: "Prompt storage was restored since
+  this page loaded. No tags were added."; unconfirmed: "Adding the tags could not be
+  confirmed. The Library was reloaded."; refused: "Tags could not be added. Try again."
+  The legacy `{prompt_ids, tags}` body is unchanged; both forms refuse a body that is not
+  a JSON object with 400.
+
+Tests: `tests/browser/workspace-library-selection.spec.mjs` (Add tags with its suggestions,
+field refusals and restore refusal; Ctrl/Cmd+A and Select all N loaded across batches and
+during a write); `tests/browser/library-concept.spec.mjs` (the bar comparison now includes
+Add tags); units `tests/js/workspace-library-selection.test.js`; Python
+`tests/test_bulk_library_tags.py`. A Python change: ComfyUI needs a restart.
+
 ### Prompt history page and recorded deviations (#249 PR F-a)
 
 The history page (`#/library/<id>`) takes the concept's heading, summary, negative
@@ -5894,8 +5934,26 @@ decide through the same `own_counts`.
 - 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).
+- No backfill: earlier cached jobs keep their counts.
+- History (#403): the receipt also keeps the raw, prepend and append text the node gave
+  the history capture. For a cached node whose graph inference leaves a field unknown,
+  the capture takes it from the receipt matching the cache entry
+  (`CachedRuns.recorded_fields`), as `inferred` / `graph_semantics`: the node did not run
+  in this job, the value follows from the run key (the whole upstream closure) plus the
+  cache proof. The job then joins the prompt's history scope, gets the
+  `positive_prompt_id` (and Pair) hint, and an image of it can be attributed through its
+  run, so the history page's "tracked runs" equals the Library count from the restart on.
+  Fail closed (fields stay unknown, as before): no matching receipt, a receipt that
+  disagrees with a field the graph knows, executions with equal outputs that recorded
+  different fields (a conflict that stays for that run key), a field over 64 KiB or
+  invalid Unicode, more than one run's history (1 MiB) in one receipt, or a full 8 MiB
+  field budget (least recently used receipts give up their fields first; their counting
+  stays). No new evidence vocabulary, so older builds still open the database.
+- Still different: jobs before the restart (no backfill); list batches (per item in the
+  Library, once per job in history); text with leading or trailing whitespace (the
+  Library stores it stripped, history matches the exact raw text, #438); a job in which the
+  node executes after its replay (RAM-pressure freed the entry: two samples, an
+  ambiguous capture); jobs while the history writer is fenced (restore, library move).
 - 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,
@@ -5903,7 +5961,10 @@ decide through the same `own_counts`.
   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.
+  covered by the Python tests. Since #403 these cases also check the history page's
+  tracked runs equal the Library count, and `tests/test_execution_history_hooks.py`
+  (`CachedRuntimeHistoryTests`) runs the real history hook and writer with a real
+  `CachedRuns`.
 
 ```sh
 python3 -m unittest discover -s tests -p "test_cached_runs.py"
@@ -6002,10 +6063,12 @@ Context cases verify A,A,B as three jobs/two combinations/one positive snapshot
 when every positive field matches; all five registered PM variants; exact raw
 whitespace and Unicode; known empty negatives; Text negative roles and crossed
 pairs; fully cached repeats after trigger configuration changes; successful source
-text before downstream failure/interruption; unknown cached dynamic raw; unsupported
+text before downstream failure/interruption; cached dynamic raw from the reused execution (#403); unsupported
 transforms and duplicate list invocation ambiguity. Fresh values use
 `runtime_return`, cached effective STRING outputs use `host_cache_output`, and
-submitted literals use graph inference. SQL NULL means unknown; empty text is
+submitted literals use graph inference. A cached dynamic raw is unknown unless the
+#347 receipt of the execution the cache entry comes from proves it; it is then
+`inferred` / `graph_semantics` (graph identity plus a same-process cache receipt, #403). SQL NULL means unknown; empty text is
 known. Text is never trimmed or normalized by capture. Cache availability and
 inferred sampler roles prove neither sampler execution nor consumption, and list
 indices do not identify generated images. External trigger configuration changes
@@ -6359,7 +6422,8 @@ Deploying this backend needs a ComfyUI restart (Python and route changes).
 `#/tags` ("Tags & categories", page title "Tags") is a native workspace route; it no
 longer hands off to the legacy tag manager. That page stays reachable from Overview's
 "Open existing tag manager" card for tag search and sort, the untagged entry and
-multi-tag AND/OR browsing; bulk prompt delete and tag stay on the legacy admin page. A
+multi-tag AND/OR browsing; bulk prompt delete stays on the legacy admin page (bulk tags:
+the Prompt library's selection bar, D151). A
 prompt's category is changed in the Prompt library editor, on many prompts at once with
 the Prompt library's selection bar (D148, #437), or with the legacy admin page's "Set
 Category" (both `POST /prompt_manager/bulk/category`);

database/library_commands.py

@@ -23,6 +23,8 @@ _TOKEN = re.compile(r"[0-9a-f]{64}")
 _PROMPT_COLUMNS = ("category", "rating", "notes")
 # Ids per IN (...) read: well under SQLite's oldest bound-parameter limit (999).
 _BULK_CHUNK = 500
+# The tags one prompt may carry: what the editor's save accepts (``validate_tags``).
+MAX_LIBRARY_TAGS = 20
 
 
 class LibraryCommandError(Exception):
@@ -363,3 +365,88 @@ class LibraryCommandOperations:
                 "unchanged": len(current) - len(changed),
                 "not_found": len(prompt_ids) - len(current),
             }
+
+    def bulk_add_library_tags(
+        self, prompt_ids: list[int], tags: list[str], *, expected_epoch: str
+    ) -> dict:
+        """Add tags to many Library parents, all or nothing (#442).
+
+        Names are normalized once (trim, the vocabulary name rules, casefold
+        repeats folded, at most 20). A parent is given only the names it lacks,
+        compared with ``casefold`` against its own tags; registry rows are
+        found and inserted by their exact name, and only for names some parent
+        gets. A parent whose tags would pass 20 is skipped (``over_limit``) so
+        the editor can still save it. Pair and image tags are not touched.
+        """
+        from .annotations import ImageTagLimitError, normalize_image_tags
+
+        _validate_epoch(expected_epoch)
+        _validate_bulk_ids(prompt_ids)
+        prompt_ids = list(dict.fromkeys(prompt_ids))
+        too_many = f"At most {MAX_LIBRARY_TAGS} tags can be added at once"
+        try:
+            tags = normalize_image_tags(tags)
+        except ImageTagLimitError:
+            # Its message names images; this is a prompt route.
+            raise ValueError(too_many) from None
+        if not tags:
+            raise ValueError("tags must name at least one tag")
+        if len(tags) > MAX_LIBRARY_TAGS:
+            raise ValueError(too_many)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            held = self._library_parent_tags(conn, prompt_ids)
+            additions, over_limit = {}, 0
+            for pid, names in held.items():
+                present = {name.casefold() for name in names}
+                missing = [name for name in tags if name.casefold() not in present]
+                if not missing:
+                    continue
+                if len(names) + len(missing) > MAX_LIBRARY_TAGS:
+                    over_limit += 1
+                    continue
+                additions[pid] = missing
+            wanted = list(
+                dict.fromkeys(n for names in additions.values() for n in names)
+            )
+            tag_ids = self._ensure_tags(conn, wanted)
+            now = utc_now_iso()
+            for pid, names in additions.items():
+                conn.executemany(
+                    "INSERT OR IGNORE INTO prompt_tags (prompt_id, tag_id) "
+                    "VALUES (?, ?)",
+                    [(pid, tag_ids[name]) for name in names],
+                )
+                self._ensure_library_parent(conn, pid)
+                conn.execute("UPDATE prompts SET updated_at=? WHERE id=?", (now, pid))
+            return {
+                "library_epoch": epoch,
+                "tags": tags,
+                "updated": len(additions),
+                "unchanged": len(held) - len(additions) - over_limit,
+                "not_found": len(prompt_ids) - len(held),
+                "over_limit": over_limit,
+            }
+
+    def _library_parent_tags(
+        self, conn: sqlite3.Connection, prompt_ids: list[int]
+    ) -> dict[int, list[str]]:
+        """The Library parents among ``prompt_ids`` and their own tag names."""
+        held: dict[int, list[str]] = {}
+        for start in range(0, len(prompt_ids), _BULK_CHUNK):
+            chunk = prompt_ids[start : start + _BULK_CHUNK]
+            marks = ",".join("?" for _ in chunk)
+            for (pid,) in conn.execute(
+                f"SELECT p.id FROM prompts p WHERE p.id IN ({marks}) "  # nosec B608
+                f"AND {_ELIGIBLE}",
+                chunk,
+            ):
+                held[pid] = []
+            for pid, name in conn.execute(
+                "SELECT pt.prompt_id,t.name FROM prompt_tags pt "  # nosec B608
+                f"JOIN tags t ON t.id=pt.tag_id WHERE pt.prompt_id IN ({marks})",
+                chunk,
+            ):
+                if pid in held:
+                    held[pid].append(name)
+        return held

prompt_manager.py

@@ -275,7 +275,12 @@ 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)
+        self._record_cached_item(
+            unique_id,
+            (encoding_text,),
+            storage_text,
+            fields=((text, prepend_text, append_text),),
+        )
         return result
 
     @classmethod

prompt_manager_base.py

@@ -243,14 +243,24 @@ class PromptManagerBase:
             return False
 
     def _record_cached_item(
-        self, unique_id: Any, outputs: tuple, text: Any, negative_text: Any = None
+        self,
+        unique_id: Any,
+        outputs: tuple,
+        text: Any,
+        negative_text: Any = None,
+        fields: Optional[tuple] = None,
     ) -> None:
-        """Remember this execution's outputs and stored text for cached jobs (#347)."""
+        """Remember this execution's outputs and stored text for cached jobs (#347).
+
+        ``fields`` are the ``(raw, prepend, append)`` texts this execution gave
+        ``record_runtime_context`` for each output, so a cached job's history
+        can record them too (#403).
+        """
         if unique_id is None:
             return
         try:
             self._cached_runs().record_item(
-                unique_id, outputs, stored_facts(text, negative_text)
+                unique_id, outputs, stored_facts(text, negative_text), fields
             )
         except Exception as error:
             self.logger.debug(f"Could not record the cached-run receipt: {error}")

prompt_manager_pair.py

@@ -198,7 +198,11 @@ class PromptManagerPair(PromptManagerFocused):
             type(self).__name__, negative_text, "", "", negative_text, role="negative"
         )
         self._record_cached_item(
-            unique_id, (positive, negative_text), text, negative_text
+            unique_id,
+            (positive, negative_text),
+            text,
+            negative_text,
+            fields=((text, prepend_text, append_text), (negative_text, "", "")),
         )
         return result
 

prompt_manager_text.py

@@ -246,7 +246,12 @@ 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)
+        self._record_cached_item(
+            unique_id,
+            (final_text,),
+            storage_text,
+            fields=((text, prepend_text, append_text),),
+        )
         return result
 
     @classmethod

py/api/prompts.py

@@ -1130,7 +1130,14 @@ class PromptRoutesMixin:
     async def bulk_add_tags(self, request):
         """Bulk add tags to prompts."""
         try:
-            data = await request.json()
+            try:
+                data = await request.json()
+            except ValueError:
+                return bad_request("Invalid JSON body")
+            if not isinstance(data, dict):
+                return bad_request("Expected a JSON object")
+            if "expected_epoch" in data:
+                return await self._bulk_library_tags(data)
             prompt_ids = data.get("prompt_ids", [])
             new_tags = data.get("tags", [])
 
@@ -1220,14 +1227,20 @@ class PromptRoutesMixin:
                 status=500,
             )
 
-    async def _bulk_library_category(self, data: dict) -> web.Response:
-        """The workspace Library's fenced form (#437).
+    async def _bulk_library_tags(self, data: dict) -> web.Response:
+        """The workspace Library's fenced form (#442).
 
-        ``{prompt_ids, category, expected_epoch}``: a non-blank category on
-        the selected Library parents, refused with 409 ``storage-changed``
-        when prompt storage was restored since the page read its epoch.
+        ``{prompt_ids, tags, expected_epoch}``: prompt tags only, never image tags.
         """
-        fields = {"prompt_ids", "category", "expected_epoch"}
+        return await self._fenced_library_write(
+            data, "tags", self.db.bulk_add_library_tags
+        )
+
+    async def _fenced_library_write(
+        self, data: dict, field: str, write
+    ) -> web.Response:
+        """One fenced Library bulk write: exact keys, checked ids, 409 on a restore."""
+        fields = {"prompt_ids", field, "expected_epoch"}
         if set(data) != fields:
             return bad_request(
                 "Request must contain exactly " + ", ".join(sorted(fields))
@@ -1235,10 +1248,7 @@ class PromptRoutesMixin:
         try:
             prompt_ids = parse_bulk_prompt_ids(data["prompt_ids"])
             result = await self._run_in_executor(
-                self.db.bulk_set_library_category,
-                prompt_ids,
-                data["category"],
-                expected_epoch=data["expected_epoch"],
+                write, prompt_ids, data[field], expected_epoch=data["expected_epoch"]
             )
         except LibraryCommandError as exc:
             return web.json_response(
@@ -1254,6 +1264,17 @@ class PromptRoutesMixin:
             return bad_request(str(exc))
         return web.json_response({"success": True, **result})
 
+    async def _bulk_library_category(self, data: dict) -> web.Response:
+        """The workspace Library's fenced form (#437).
+
+        ``{prompt_ids, category, expected_epoch}``: a non-blank category on
+        the selected Library parents, refused with 409 ``storage-changed``
+        when prompt storage was restored since the page read its epoch.
+        """
+        return await self._fenced_library_write(
+            data, "category", self.db.bulk_set_library_category
+        )
+
     async def _bulk_category_by_filter(self, data: dict) -> web.Response:
         """Set the category on every linked positive a listing filter admits.
 

tests/browser/library-concept.spec.mjs

@@ -317,7 +317,7 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`library geomet
 /**
  * D148 (#437): the selection checkbox overlay of `card` and the `.bulkbar` row inside `main`, with one item checked,
  * measured alike in the concept's Images page and production's Library: the overlay's inset in its card, the checkbox,
- * and the bar's box, tokens and the two actions both bars have (Set category, Clear).
+ * and the bar's box, tokens and the actions both bars have (Add tags, Set category, Clear; #442).
  */
 const measureSelection=(page,{card,main})=>page.evaluate(({card,main})=>{
     const css=(node,...keys)=>{const style=getComputedStyle(node);return Object.fromEntries(keys.map(key=>[key,style[key]]));};
@@ -327,7 +327,7 @@ const measureSelection=(page,{card,main})=>page.evaluate(({card,main})=>{
         'borderTopColor','borderTopLeftRadius','backgroundColor');
     const host=document.querySelector(card),overlay=host.querySelector('.image-overlays'),label=overlay.querySelector(':scope > label');
     const input=label.querySelector('input'),root=document.querySelector(main),bar=root.querySelector('.bulkbar');
-    const shared=['Set category','Clear'];
+    const shared=['Add tags','Set category','Clear'];
     return {
         overlay:{left:px(rect(overlay).left-rect(host).left),top:px(rect(overlay).top-rect(host).top),right:px(rect(host).right-rect(overlay).right),
             ...css(overlay,'position','display','justifyContent','pointerEvents')},
@@ -357,7 +357,7 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`the Library se
     const actual=await productionSelection(page,theme);
     await captureCanonicalState(page,testInfo,`production-library-selection-${width}-${theme}`);
     await testInfo.attach('library-selection',{body:JSON.stringify({reference,actual},null,2),contentType:'application/json'});
-    expect(reference.bar.buttons.map(button=>button.text),'the concept bar has both actions').toEqual(['Set category','Clear']);
+    expect(reference.bar.buttons.map(button=>button.text),'the concept bar has every Library action').toEqual(['Add tags','Set category','Clear']);
     expect(actual).toEqual(reference);
     // The comparison notices a moved checkbox (spec §7 mutation check).
     await mutate(page,{css:'.prompt-card .image-overlays { inset: 20px 9px auto }'});

tests/browser/workspace-library-selection.spec.mjs

@@ -11,6 +11,14 @@ const bar=page=>page.locator('#library-bulkbar > .bulkbar');
 const dialog=page=>page.getByRole('dialog',{name:'Categorize selected prompts',exact:true});
 const toast=page=>page.locator('.toast');
 const heading=page=>page.locator('#workspace-main h1');
+const tagsDialog=page=>page.getByRole('dialog',{name:'Tag selected prompts',exact:true});
+const chips=(page,id)=>card(page,id).locator('.tags > .chip');
+async function addTags(page,value) {
+    await bar(page).getByRole('button',{name:'Add tags',exact:true}).click();
+    await expect(tagsDialog(page)).toBeVisible();
+    await tagsDialog(page).locator('#action-input').fill(value);
+    await tagsDialog(page).getByRole('button',{name:'Apply',exact:true}).click();
+}
 
 async function shelf(request) {
     const body=await(await request.get('/prompt_manager/library?limit=100')).json();
@@ -53,7 +61,8 @@ test('Set category writes the selected prompts on the server, reloads the shelf
     await expect(box(page,first.id)).toHaveAccessibleName(new RegExp(`^Select `));
     await box(page,first.id).check();await expect(bar(page).locator('strong')).toHaveText('1 selected');
     await box(page,second.id).check();await expect(bar(page).locator('strong')).toHaveText('2 selected');
-    await expect(bar(page).getByRole('button')).toHaveText(['Set category','Clear']);
+    // Two of three loaded are checked, so Select all N loaded follows the count (D151, #444 ruling).
+    await expect(bar(page).getByRole('button')).toHaveText([`Select all ${items.length} loaded`,'Add tags','Set category','Clear']);
     await setCategory(page,'  Studio set  ');
     await expect(toast(page)).toHaveText('Category “Studio set” set on 2 prompts.');
     await expect(dialog(page)).toHaveCount(0);
@@ -175,3 +184,111 @@ test('an answer that may follow a committed write says so and reloads; a refused
     await expect(page.locator('#library-bulkbar')).toBeEmpty();
     await expect(card(page,items[0].id).locator('.card-title .chip')).toHaveText('Unconfirmed');
 });
+
+// #442: Add tags writes prompt tags (never image tags) on the selected prompts.
+function recordTagWrites(page) {
+    const writes=[];
+    page.on('request',item=>{if(item.method()==='POST'&&new URL(item.url()).pathname==='/prompt_manager/bulk/tags')writes.push(item.postDataJSON());});
+    return writes;
+}
+async function promptTags(request,id) {
+    return (await shelf(request)).find(item=>item.id===id).tags;
+}
+
+test('Add tags adds prompt tags to the selected prompts, suggests the vocabulary and reloads the shelf',async({page,request})=>{
+    const items=await shelf(request);const [first,second,third]=items;const writes=recordTagWrites(page);
+    const vocabulary=(await(await request.get('/prompt_manager/tags')).json()).tags;
+    await page.goto(route);await expect(cards(page)).toHaveCount(items.length);
+    await box(page,first.id).check();await box(page,second.id).check();
+    await bar(page).getByRole('button',{name:'Add tags',exact:true}).click();
+    await expect(tagsDialog(page).locator('datalist option')).toHaveCount(vocabulary.length);
+    await tagsDialog(page).getByRole('button',{name:'Cancel',exact:true}).click();await expect(tagsDialog(page)).toHaveCount(0);
+    await addTags(page,' studio-lit , Night ,studio-lit');
+    await expect(toast(page)).toHaveText('Added “studio-lit, Night” to 2 prompts.');
+    await expect(page.locator('#library-bulkbar')).toBeEmpty();
+    for(const item of [first,second]) {
+        await expect(chips(page,item.id).filter({hasText:'studio-lit'})).toHaveCount(1);
+        expect(await promptTags(request,item.id)).toEqual(expect.arrayContaining(['studio-lit','Night']));
+    }
+    expect(await promptTags(request,third.id)).not.toContain('studio-lit');
+    const epoch=(await(await request.get('/prompt_manager/library/options')).json()).library_epoch;
+    expect(writes).toEqual([{prompt_ids:[first.id,second.id],tags:['studio-lit','Night'],expected_epoch:epoch}]);
+    // A second add of the same names changes nothing and says so.
+    await box(page,first.id).check();await addTags(page,'Studio-Lit');
+    await expect(toast(page)).toHaveText('1 prompt already had “Studio-Lit”.');
+});
+
+test('the Add tags field states what it refuses, and a restore the page did not see adds nothing',async({page,request})=>{
+    await armUnseenRestores(page);
+    const items=await shelf(request);const writes=recordTagWrites(page);
+    await page.goto(route);await expect(cards(page)).toHaveCount(items.length);
+    await box(page,items[0].id).check();
+    await bar(page).getByRole('button',{name:'Add tags',exact:true}).click();
+    const field=tagsDialog(page).locator('#action-input');
+    for(const [value,message] of [[' , ','Enter at least one tag.'],['x'.repeat(51),'A tag can have at most 50 characters.'],
+        [Array.from({length:21},(_,i)=>`t${i}`).join(','),'Enter at most 20 tags.']]) {
+        await field.fill(value);await tagsDialog(page).getByRole('button',{name:'Apply',exact:true}).click();
+        await expect(tagsDialog(page)).toBeVisible();
+        expect(await field.evaluate(node=>node.validationMessage)).toBe(message);
+    }
+    await tagsDialog(page).getByRole('button',{name:'Cancel',exact:true}).click();
+    expect(writes).toHaveLength(0);
+    const backup=await request.get('/prompt_manager/backup');await expect(backup).toBeOK();const buffer=await backup.body();
+    const restore=await unseenRestore(page,()=>request.post('/prompt_manager/restore',{multipart:{database_file:{name:'library.db',mimeType:'application/octet-stream',buffer}}}));
+    await expect(restore).toBeOK();
+    await addTags(page,'after-restore');
+    await expect(toast(page)).toHaveText('Prompt storage was restored since this page loaded. No tags were added.');
+    await expect(page.locator('#library-bulkbar')).toBeEmpty();
+    expect((await(await request.get('/prompt_manager/tags')).json()).tags).not.toContain('after-restore');
+});
+
+// #444 ruling (D151): select every loaded prompt, as the legacy page's select-all checkbox did for its page.
+test('Ctrl/Cmd+A and Select all N loaded select every loaded prompt, windowed ones included',async({page,request,isMobile})=>{
+    await expect(await request.post('/__test__/reset',{data:{native_library:true,extra_prompts:200}})).toBeOK();
+    if(!isMobile)await page.setViewportSize({width:1600,height:1100});
+    const writes=recordTagWrites(page);
+    await page.goto(route);await expect(cards(page).first()).toBeVisible();
+    const more=page.getByRole('button',{name:'Load more prompts',exact:true});
+    for(let index=0;index<3;index++) {
+        await expect(more).toHaveAttribute('aria-disabled','false');
+        const response=page.waitForResponse(item=>{const url=new URL(item.url());return url.pathname==='/prompt_manager/library'&&url.searchParams.has('cursor');});
+        await more.evaluate(node=>node.click());await response;
+    }
+    await expect(more).toHaveAttribute('aria-disabled','false');
+    const loaded=160;expect(await cards(page).count()).toBeLessThan(loaded);
+    // In the search box Ctrl/Cmd+A keeps its text meaning: it selects the box's text, no prompt.
+    await page.locator('#search').focus();await page.keyboard.press('ControlOrMeta+a');
+    await expect(page.locator('#library-bulkbar')).toBeEmpty();
+    // One checked card offers the rest; the offer selects every loaded prompt and focus moves to the first action.
+    const firstId=Number(await cards(page).first().getAttribute('data-prompt-id'));
+    await box(page,firstId).check();
+    const offer=bar(page).getByRole('button',{name:`Select all ${loaded} loaded`,exact:true});
+    await expect(bar(page).locator(':scope > *').nth(1)).toHaveText(`Select all ${loaded} loaded`);
+    await offer.click();
+    await expect(bar(page).locator('strong')).toHaveText(`${loaded} selected`);
+    await expect(offer).toHaveCount(0);
+    await expect(bar(page).getByRole('button',{name:'Add tags',exact:true})).toBeFocused();
+    // Another batch arrives unchecked: the offer comes back with the new N.
+    {
+        const response=page.waitForResponse(item=>{const url=new URL(item.url());return url.pathname==='/prompt_manager/library'&&url.searchParams.has('cursor');});
+        await more.evaluate(node=>node.click());await response;
+    }
+    await expect(bar(page).getByRole('button',{name:`Select all ${loaded+40} loaded`,exact:true})).toBeVisible();
+    await expect(bar(page).locator('strong')).toHaveText(`${loaded} selected`);
+    // Escape in the grid clears; Ctrl/Cmd+A in the grid selects them all again.
+    const title=cards(page).first().locator('.card-title > button');
+    await title.focus();await page.keyboard.press('Escape');
+    await expect(page.locator('#library-bulkbar')).toBeEmpty();await expect(title).toBeFocused();
+    await page.keyboard.press('ControlOrMeta+a');
+    const all=loaded+40;
+    await expect(bar(page).locator('strong')).toHaveText(`${all} selected`);
+    // While the write runs, every bar action and the offer are aria-disabled.
+    await box(page,firstId).uncheck();
+    let release;const held=new Promise(resolve=>{release=resolve;});
+    await page.route('**/prompt_manager/bulk/tags',async route=>{await held;await route.continue();},{times:1});
+    await addTags(page,'everyone');
+    for(const name of ['Add tags','Set category',`Select all ${all} loaded`]) await expect(bar(page).getByRole('button',{name,exact:true})).toHaveAttribute('aria-disabled','true');
+    release();
+    await expect(toast(page)).toHaveText(`Added “everyone” to ${all-1} prompts.`);
+    expect(writes).toHaveLength(1);expect(new Set(writes[0].prompt_ids).size).toBe(all-1);
+});

tests/comfyui/execution-history.spec.mjs

@@ -263,15 +263,18 @@ test.describe('public chronological history with the actual acknowledged host br
     });
 });
 
-test('a cached dynamic unknown positive stays readable by run UUID and outside the known raw Library scope', async ({ execution }) => {
-    const text = `unknown positive ${randomUUID()}`, graph = recipeGraph({ text, dynamic: true });
+// #403: the graph cannot show a dynamic positive's raw text; a cached job takes it from the execution it reuses.
+test('a cached dynamic positive takes its raw text from the execution it reuses and joins the Library scope', async ({ execution }) => {
+    const text = `dynamic positive ${randomUUID()}`, graph = recipeGraph({ text, dynamic: true });
     const first = await submit(execution.request, graph);await history(execution.request, first);const known = (await capture(execution, first)).contexts[0];
-    const second = await submit(execution.request, graph);await history(execution.request, second);const unknown = (await capture(execution, second)).contexts[0];
-    expect(unknown.positive.raw).toBeNull();expect(unknown.positive.effective).toBe(known.positive.effective);
+    const second = await submit(execution.request, graph);await history(execution.request, second);const reused = await capture(execution, second);const cached = reused.contexts[0];
+    expect(cached.evidence.positive.effective.source).toBe('host_cache_output');expect(cached.positive.effective).toBe(known.positive.effective);
+    expect(cached.positive.raw).toBe(known.positive.raw);expect(cached.evidence.positive.raw).toMatchObject({ provenance: 'inferred', source: 'graph_semantics' });
+    expect(cached.positive_prompt_id).toBe(known.positive_prompt_id);
     const result = await readExecutionRuns(execution.databasePath, second), runId = result.rows[0].run_id;
-    const detail = await execution.request.get(`/prompt_manager/history/runs/${runId}`);expect(detail.ok()).toBe(true);expect((await detail.json()).contexts[0].positive.raw).toBeNull();
+    const detail = await execution.request.get(`/prompt_manager/history/runs/${runId}`);expect(detail.ok()).toBe(true);expect((await detail.json()).contexts[0].positive.raw).toBe(known.positive.raw);
     const response = await execution.request.get(`/prompt_manager/library/${known.positive_prompt_id}/history`);expect(response.ok()).toBe(true);const window = await response.json();
-    expect(window.recorded_run_count).toBe(1);expect(window.items.map(item => item.run_id)).not.toContain(runId);
+    expect(window.recorded_run_count).toBe(2);expect(window.items.map(item => item.run_id)).toContain(runId);
 });
 
 test('actual PM A,A,B jobs persist two combinations and one identical positive snapshot', async ({ execution }) => {
@@ -369,7 +372,7 @@ test('successful source capture survives downstream failure and real interruptio
     }
 });
 
-test('cached dynamic raw stays unknown while literal affixes and effective cache text survive', async ({ execution }) => {
+test('cached dynamic raw comes from the reused execution while literal affixes and effective cache text survive (#403)', async ({ execution }) => {
     const graph = recipeGraph({ dynamic: true, text: 'runtime dynamic recipe', prepend: ' pre ', append: ' post ' });
     const first = await submit(execution.request, graph);
     await history(execution.request, first);
@@ -377,8 +380,9 @@ test('cached dynamic raw stays unknown while literal affixes and effective cache
     const second = await submit(execution.request, graph);
     await history(execution.request, second);
     const value = (await capture(execution, second)).contexts[0];
-    expect(value.positive).toEqual({ raw: null, prepend: ' pre ', append: ' post ', effective: 'pre runtime dynamic recipe post' });
-    expect(value.evidence.positive.raw.provenance).toBe('unknown');
+    expect(value.positive).toEqual({ raw: 'runtime dynamic recipe', prepend: ' pre ', append: ' post ', effective: 'pre runtime dynamic recipe post' });
+    expect(value.evidence.positive.raw).toMatchObject({ provenance: 'inferred', source: 'graph_semantics' });
+    expect(value.evidence.positive.prepend.source).toBe('graph_literal');
     expect(value.evidence.positive.effective.source).toBe('host_cache_output');
 });
 

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

@@ -287,6 +287,14 @@ async function capturedContext(execution, id) {
     return result.captures[result.rows[0].run_id].contexts[0];
 }
 
+// #403: the prompt's history "tracked runs", once every job's capture reached SQLite.
+async function expectTrackedRuns(execution, promptId, expected) {
+    await expect.poll(async () => {
+        const response = await execution.request.get(`/prompt_manager/library/${promptId}/history`);
+        return response.ok() ? (await response.json()).recorded_run_count : null;
+    }, { timeout: 10000, message: 'history tracked runs equal the Library run count' }).toBe(expected);
+}
+
 async function firstRun(execution, page, testCase) {
     const prefix = `textlink/${randomUUID()}`;
     const built = await build(page, { ...testCase, prefix });
@@ -309,6 +317,8 @@ for (const kind of ['PromptManager', 'PromptManagerFocused', 'PromptManagerSplit
         const counts = await runCounts(execution, first.context);
         expect(counts.positive).toBe(3);
         if (pair) expect(counts).toMatchObject({ negative: 0, pair: 3 });
+        // #403: the cached jobs are tracked runs of the prompt too
+        await expectTrackedRuns(execution, first.context.positive_prompt_id, 3);
     });
 }
 
@@ -329,6 +339,7 @@ test.describe('cached runs in one renderer (#347)', () => {
                     else expect(again.cached).toContain(first.promptNode);
                 }
                 expect((await runCounts(execution, first.context)).positive).toBe(3);
+                await expectTrackedRuns(execution, first.context.positive_prompt_id, 3);
             } finally { await setCache(execution, 'RAM_PRESSURE'); }
         });
     }
@@ -352,6 +363,9 @@ test.describe('cached runs in one renderer (#347)', () => {
             const bContext = await capturedContext(execution, second.id);
             expect(bContext.positive.raw).toBe(b);
             expect((await runCounts(execution, bContext)).positive).toBe(1);
+            // #403: the third job reuses A's execution, not the newest (B's)
+            await expectTrackedRuns(execution, first.context.positive_prompt_id, 2);
+            await expectTrackedRuns(execution, bContext.positive_prompt_id, 1);
         } finally { await setCache(execution, 'RAM_PRESSURE'); }
     });
 
@@ -390,10 +404,14 @@ test.describe('cached runs in one renderer (#347)', () => {
         } });
         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
+        // #403: a cached job's history takes the raw text of the execution it reuses and names the prompt
         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(cachedContext.positive.raw).toBe(first.context.positive.raw);
+        expect(cachedContext.evidence.positive.raw).toMatchObject({ provenance: 'inferred', source: 'graph_semantics' });
+        expect(cachedContext.positive_prompt_id).toBe(first.context.positive_prompt_id);
         expect((await runCounts(execution, first.context)).positive).toBe(2);
+        // The rewired job used the prompt as its negative: neither number counts it
+        await expectTrackedRuns(execution, first.context.positive_prompt_id, 2);
     });
 });

tests/js/workspace-library-selection.test.js

@@ -42,3 +42,55 @@ test('the field refuses control characters as validate_category does', async ()
     assert.equal(categoryProblem('nul\u0000'), CONTROL_CHARACTERS);
     assert.equal(categoryProblem('unicode ✓ é'), null);
 });
+
+test('Add tags uses the Images dialog words for prompts, the tag routes and the editor limits (#442)', async () => {
+    const {PROMPT_TAGS_DIALOG, TAGS_URL, TAG_OPTIONS_URL, MAX_PROMPT_TAGS} = await source;
+    assert.deepEqual({...PROMPT_TAGS_DIALOG}, {title:'Tag selected prompts', label:'Tags, separated by commas', action:'Apply'});
+    assert.equal(TAGS_URL, '/prompt_manager/bulk/tags');
+    assert.equal(TAG_OPTIONS_URL, '/prompt_manager/tags');
+    assert.equal(MAX_PROMPT_TAGS, 20);
+});
+
+test('the Add tags toast says what was added, already present, skipped and missing', async () => {
+    const {tagsOutcome, writeOutcome, NONE_FOUND} = await source;
+    assert.equal(tagsOutcome({tags:['a', 'b'], updated:2, unchanged:0, over_limit:0, not_found:0}), 'Added “a, b” to 2 prompts.');
+    assert.equal(tagsOutcome({tags:['a'], updated:1, unchanged:2, over_limit:1, not_found:3}),
+        'Added “a” to 1 prompt. 2 prompts already had them. 1 prompt would pass 20 tags and was skipped. 3 could not be found.');
+    assert.equal(tagsOutcome({tags:['a'], updated:0, unchanged:2, over_limit:0, not_found:0}), '2 prompts already had “a”.');
+    // Every prompt skipped: found, so the skip sentence, never "none found"; nothing committed.
+    assert.deepEqual(writeOutcome({success:true, tags:['a'], updated:0, unchanged:0, over_limit:2, not_found:0}, null, 'tags'),
+        {toast:'2 prompts would pass 20 tags and were skipped.', committed:false, reload:true});
+    assert.deepEqual(writeOutcome({success:true, tags:['a'], updated:0, unchanged:0, over_limit:0, not_found:2}, null, 'tags'),
+        {toast:NONE_FOUND, committed:false, reload:true});
+});
+
+test('Add tags failures name tags, not the category', async () => {
+    const {writeOutcome, TAGS_STORAGE_CHANGED, TAGS_UNCONFIRMED, TAGS_FAILED} = await source;
+    assert.equal(writeOutcome(null, {status:409, code:'storage-changed'}, 'tags').toast, TAGS_STORAGE_CHANGED);
+    assert.equal(writeOutcome(null, {status:503}, 'tags').toast, TAGS_UNCONFIRMED);
+    assert.equal(writeOutcome(null, {status:400}, 'tags').toast, TAGS_FAILED);
+    assert.equal(TAGS_STORAGE_CHANGED, 'Prompt storage was restored since this page loaded. No tags were added.');
+});
+
+test('the Add tags field refuses what validate_tags refuses', async () => {
+    const {tagsProblem, TAG_PROBLEMS} = await source;
+    assert.equal(tagsProblem('red, blue'), null);
+    assert.equal(tagsProblem(' , ,'), TAG_PROBLEMS.none);
+    assert.equal(tagsProblem(Array.from({length:21}, (_, i) => `t${i}`).join(',')), TAG_PROBLEMS.many);
+    assert.equal(tagsProblem(Array.from({length:21}, () => 'same').join(',')), null);
+    assert.equal(tagsProblem('x'.repeat(51)), TAG_PROBLEMS.long);
+    assert.equal(tagsProblem('bad\ttag'), TAG_PROBLEMS.control);
+});
+
+test('Select all N loaded is offered while some loaded prompts are unchecked (#444 ruling)', async () => {
+    const {loadedOffer, offersLoaded} = await source;
+    assert.equal(loadedOffer(3), 'Select all 3 loaded');
+    assert.equal(loadedOffer(1203), 'Select all 1,203 loaded');
+    assert.equal(offersLoaded(new Set([1]), [1, 2, 3]), true);
+    assert.equal(offersLoaded(new Set([1, 2, 3]), [1, 2, 3]), false);
+    assert.equal(offersLoaded(new Set(), [1, 2]), false);
+    assert.equal(offersLoaded(new Set([9]), [9]), false);
+    // More loaded than any action accepts: never offered (Ctrl/Cmd+A still selects them).
+    assert.equal(offersLoaded(new Set([1]), Array.from({length:1001}, (_, i) => i + 1)), false);
+    assert.equal(offersLoaded(new Set([1]), Array.from({length:1000}, (_, i) => i + 1)), true);
+});

tests/test_bulk_library_tags.py

@@ -0,0 +1,286 @@
+"""#442: the workspace Library's fenced bulk "Add tags" (DB command and route)."""
+
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from database.execution_history_schema import rotate_history_epoch
+from database.library_commands import LibraryCommandError
+from database.operations import PromptDatabase
+from utils.hashing import generate_prompt_hash
+
+try:
+    from test_api_prompts import RouteCoverageCase
+except ImportError:
+    from tests.test_api_prompts import RouteCoverageCase
+
+URL = "/prompt_manager/bulk/tags"
+
+
+def save(db, text, **metadata):
+    return db.save_prompt(text, prompt_hash=generate_prompt_hash(text), **metadata)
+
+
+def tags_of(db, prompt_id):
+    with db.model.get_connection() as conn:
+        rows = conn.execute(
+            "SELECT t.name FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
+            "WHERE pt.prompt_id=? ORDER BY t.name",
+            (prompt_id,),
+        ).fetchall()
+    return [row[0] for row in rows]
+
+
+def updated_at(db, prompt_id):
+    with db.model.get_connection() as conn:
+        return conn.execute(
+            "SELECT updated_at FROM prompts WHERE id=?", (prompt_id,)
+        ).fetchone()[0]
+
+
+def tag_names(db):
+    with db.model.get_connection() as conn:
+        return sorted(row[0] for row in conn.execute("SELECT name FROM tags"))
+
+
+class TestBulkLibraryTags(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.db = PromptDatabase(str(Path(self.temp.name) / "tags.db"))
+        self.epoch = self.db.get_library_options()["library_epoch"]
+
+    def tearDown(self):
+        self.db.close_all()
+        self.temp.cleanup()
+
+    def run_bulk(self, ids, tags, epoch=None):
+        return self.db.bulk_add_library_tags(
+            ids, tags, expected_epoch=epoch or self.epoch
+        )
+
+    def test_adds_only_missing_tags_and_counts_each_outcome(self):
+        a = save(self.db, "Alpha", tags=["red"])
+        b = save(self.db, "Beta", tags=["red", "blue"])
+        c = save(self.db, "Gamma")
+        before_b = updated_at(self.db, b)
+        result = self.run_bulk([a, b, c, 99999], [" red ", "blue", "Blue"])
+        self.assertEqual(
+            result,
+            {
+                "library_epoch": self.epoch,
+                "tags": ["red", "blue"],
+                "updated": 2,
+                "unchanged": 1,
+                "not_found": 1,
+                "over_limit": 0,
+            },
+        )
+        self.assertEqual(tags_of(self.db, a), ["blue", "red"])
+        self.assertEqual(tags_of(self.db, c), ["blue", "red"])
+        self.assertEqual(updated_at(self.db, b), before_b)
+        self.assertTrue(updated_at(self.db, a).endswith("+00:00"))
+
+    def test_a_case_variant_of_a_held_tag_is_not_added_and_names_stay_exact(self):
+        a = save(self.db, "Alpha", tags=["Portrait"])
+        b = save(self.db, "Beta")
+        result = self.run_bulk([a, b], ["portrait"])
+        # Alpha already holds it (casefold); Beta gets the name exactly as typed.
+        self.assertEqual((result["updated"], result["unchanged"]), (1, 1))
+        self.assertEqual(tags_of(self.db, a), ["Portrait"])
+        self.assertEqual(tags_of(self.db, b), ["portrait"])
+        self.assertEqual(tag_names(self.db), ["Portrait", "portrait"])
+        c = save(self.db, "Gamma", tags=["STRASSE"])
+        self.assertEqual(self.run_bulk([c], ["straße"])["unchanged"], 1)
+
+    def test_names_nothing_receives_are_never_inserted(self):
+        full = save(self.db, "Full", tags=[f"t{i}" for i in range(20)])
+        self.run_bulk([full, 99999], ["unused"])
+        self.assertNotIn("unused", tag_names(self.db))
+
+    def test_a_prompt_already_over_twenty_that_gains_nothing_is_unchanged(self):
+        over = save(self.db, "Over", tags=[f"t{i}" for i in range(22)])
+        result = self.run_bulk([over], ["t1"])
+        self.assertEqual((result["unchanged"], result["over_limit"]), (1, 0))
+
+    def test_image_and_pair_tags_are_untouched(self):
+        pair = self.db.save_prompt_pair("Cat", "Blur", tags=["pairtag"])
+        positive = self.db.get_prompt_pair(pair)["positive_prompt_id"]
+        with self.db.model.get_connection() as conn:
+            before = conn.execute("SELECT COUNT(*) FROM prompt_pair_tags").fetchone()[0]
+        self.run_bulk([positive], ["pairtag", "own"])
+        self.assertEqual(tags_of(self.db, positive), ["own", "pairtag"])
+        with self.db.model.get_connection() as conn:
+            after = conn.execute("SELECT COUNT(*) FROM prompt_pair_tags").fetchone()[0]
+        self.assertEqual(after, before)
+
+    def test_a_prompt_that_would_pass_twenty_tags_is_skipped_untouched(self):
+        full = save(self.db, "Full", tags=[f"t{i}" for i in range(19)])
+        other = save(self.db, "Other")
+        before = updated_at(self.db, full)
+        result = self.run_bulk([full, other], ["x", "y"])
+        self.assertEqual((result["updated"], result["over_limit"]), (1, 1))
+        self.assertEqual(len(tags_of(self.db, full)), 19)
+        self.assertEqual(updated_at(self.db, full), before)
+        self.assertEqual(tags_of(self.db, other), ["x", "y"])
+        # Exactly twenty is allowed.
+        self.assertEqual(self.run_bulk([full], ["x"])["updated"], 1)
+
+    def test_changed_prompts_become_registered_library_parents(self):
+        a = save(self.db, "Alpha")
+        self.run_bulk([a], ["new"])
+        with self.db.model.get_connection() as conn:
+            self.assertIsNotNone(
+                conn.execute(
+                    "SELECT 1 FROM prompt_library_state WHERE prompt_id=?", (a,)
+                ).fetchone()
+            )
+
+    def test_negative_only_prompts_are_not_found(self):
+        pair = self.db.save_prompt_pair("Cat", "Blur")
+        negative = self.db.get_prompt_pair(pair)["negative_prompt_id"]
+        result = self.run_bulk([negative], ["x"])
+        self.assertEqual((result["updated"], result["not_found"]), (0, 1))
+        self.assertEqual(tags_of(self.db, negative), [])
+
+    def test_restored_storage_writes_nothing_not_even_a_tag_name(self):
+        a = save(self.db, "Alpha")
+        rotate_history_epoch(self.db.model.get_connection())
+        with self.assertRaises(LibraryCommandError) as caught:
+            self.run_bulk([a], ["fresh"])
+        self.assertEqual(caught.exception.code, "storage-changed")
+        self.assertEqual(tags_of(self.db, a), [])
+        self.assertNotIn("fresh", tag_names(self.db))
+
+    def test_a_failure_mid_write_leaves_every_prompt_as_it_was(self):
+        ids = [save(self.db, f"P{i}") for i in range(3)]
+        with self.db.model.get_connection() as conn:
+            conn.execute(
+                "CREATE TRIGGER fail_third BEFORE INSERT ON prompt_tags "
+                f"WHEN NEW.prompt_id={ids[2]} BEGIN SELECT RAISE(ABORT, 'boom'); END"
+            )
+        with self.assertRaises(sqlite3.DatabaseError):
+            self.run_bulk(ids, ["x"])
+        self.assertEqual([tags_of(self.db, pid) for pid in ids], [[], [], []])
+        self.assertNotIn("x", tag_names(self.db))
+
+    def test_entries_are_counted_after_dedupe_and_the_limit_names_prompts(self):
+        a = save(self.db, "Alpha")
+        self.assertEqual(self.run_bulk([a], ["dup"] * 30 + ["Dup"])["updated"], 1)
+        with self.assertRaises(ValueError) as caught:
+            self.run_bulk([a], [f"t{i}" for i in range(101)])
+        self.assertEqual(str(caught.exception), "At most 20 tags can be added at once")
+
+    def test_many_ids_are_read_in_chunks(self):
+        ids = [save(self.db, f"Many {i}") for i in range(1000)]
+        self.assertEqual(self.run_bulk(ids, ["bulk"])["updated"], 1000)
+
+    def test_rejects_invalid_arguments(self):
+        a = save(self.db, "Alpha")
+        for ids, tags, epoch in (
+            ([a], [], self.epoch),
+            ([a], ["  "], self.epoch),
+            ([a], "red", self.epoch),
+            ([a], ["x" * 51], self.epoch),
+            ([a], ["bad\x01"], self.epoch),
+            ([a], ["a,b"], self.epoch),
+            ([a], [f"t{i}" for i in range(21)], self.epoch),
+            ([a], ["ok"], "not-a-uuid"),
+            ([], ["ok"], self.epoch),
+        ):
+            with self.subTest(tags=tags, epoch=epoch, ids=ids):
+                with self.assertRaises(ValueError):
+                    self.run_bulk(ids, tags, epoch)
+
+
+class TestBulkLibraryTagsRoute(RouteCoverageCase):
+    def epoch(self):
+        return self.api.db.get_library_options()["library_epoch"]
+
+    async def test_fenced_form_adds_tags_and_reports_counts(self):
+        ids = [self._save_prompt(f"route {i}") for i in range(2)]
+        status, body = await self._json(
+            "POST",
+            URL,
+            json={"prompt_ids": ids, "tags": ["Fresh"], "expected_epoch": self.epoch()},
+        )
+        self.assertEqual(status, 200)
+        self.assertEqual(
+            body,
+            {
+                "success": True,
+                "library_epoch": self.epoch(),
+                "tags": ["Fresh"],
+                "updated": 2,
+                "unchanged": 0,
+                "not_found": 0,
+                "over_limit": 0,
+            },
+        )
+
+    async def test_restored_storage_answers_409_storage_changed(self):
+        pid = self._save_prompt("route restore")
+        stale = self.epoch()
+        new_epoch = rotate_history_epoch(self.api.db.model.get_connection())
+        status, body = await self._json(
+            "POST",
+            URL,
+            json={"prompt_ids": [pid], "tags": ["x"], "expected_epoch": stale},
+        )
+        self.assertEqual((status, body["code"]), (409, "storage-changed"))
+        self.assertEqual(body["library_epoch"], new_epoch)
+
+    async def test_fenced_form_validates_its_body(self):
+        pid = self._save_prompt("route validate")
+        epoch = self.epoch()
+        for body in (
+            {"prompt_ids": [pid], "tags": [], "expected_epoch": epoch},
+            {"prompt_ids": [pid], "tags": "x", "expected_epoch": epoch},
+            {"prompt_ids": [pid], "tags": ["x" * 51], "expected_epoch": epoch},
+            {"prompt_ids": [pid], "tags": ["ok"], "expected_epoch": "nope"},
+            {"prompt_ids": [True], "tags": ["ok"], "expected_epoch": epoch},
+            {
+                "prompt_ids": list(range(1, 1002)),
+                "tags": ["ok"],
+                "expected_epoch": epoch,
+            },
+            {"prompt_ids": [pid], "expected_epoch": epoch},
+            {"prompt_ids": [pid], "tags": ["ok"], "expected_epoch": epoch, "x": 1},
+        ):
+            with self.subTest(body=body):
+                status, answer = await self._json("POST", URL, json=body)
+                self.assertEqual(status, 400)
+                self.assertFalse(answer["success"])
+        self.assertEqual(tags_of(self.api.db, pid), [])
+
+    async def test_fenced_form_reports_an_unexpected_failure(self):
+        pid = self._save_prompt("route failure")
+        self._break("bulk_add_library_tags")
+        status, body = await self._json(
+            "POST",
+            URL,
+            json={"prompt_ids": [pid], "tags": ["ok"], "expected_epoch": self.epoch()},
+        )
+        self.assertEqual(status, 500)
+        self.assertFalse(body["success"])
+
+    async def test_legacy_form_refuses_a_body_that_is_not_an_object(self):
+        for raw in ("[1, 2]", "not json"):
+            with self.subTest(raw=raw):
+                status, body = await self._json(
+                    "POST", URL, data=raw, headers={"Content-Type": "application/json"}
+                )
+                self.assertEqual(status, 400)
+                self.assertFalse(body["success"])
+
+    async def test_legacy_form_is_unchanged(self):
+        pid = self._save_prompt("route legacy")
+        status, body = await self._json(
+            "POST", URL, json={"prompt_ids": [pid], "tags": ["Legacy"]}
+        )
+        self.assertEqual((status, body["updated_count"]), (200, 1))
+        self.assertNotIn("library_epoch", body)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_cached_runs.py

@@ -666,7 +666,7 @@ class TestOwnCounts(unittest.TestCase):
         return counts
 
 
-class TestReceipts(unittest.TestCase):
+class ReceiptCase(unittest.TestCase):
     def setUp(self):
         self.context = ("j1", None)
         self.runs = CachedRuns(
@@ -686,6 +686,8 @@ class TestReceipts(unittest.TestCase):
     def keys(self):
         return list(self.runs._receipts)
 
+
+class TestReceipts(ReceiptCase):
     def test_least_recently_used_run_keys_are_dropped(self):
         self.begin(1)
         for name in ("a", "b", "c"):
@@ -726,6 +728,196 @@ class TestReceipts(unittest.TestCase):
         self.assertEqual(list(self.runs._attempt.nodes), ["b"])
 
 
+class TestRecordedFields(ReceiptCase):
+    """The texts a receipt keeps for the history capture of a cached job (#403)."""
+
+    F = (("raw a", " pre ", ""),)
+
+    def record(self, item_id, name="a", output="x", fields=F, facts=None):
+        self.begin(item_id)
+        self.runs.record_item(name, (output,), facts or Facts("h"), fields)
+        self.runs.end_attempt(item_id)
+
+    def lookup(self, name="a", values=(("x",),), prompt_id="j9", item_id=9):
+        self.begin(item_id, prompt_id)
+        return self.runs.recorded_fields(prompt_id, name, values)
+
+    def test_a_cached_job_reads_the_fields_of_the_execution_it_reuses(self):
+        self.record(1)
+        self.assertEqual(self.lookup(), [self.F])
+
+    def test_list_items_return_their_own_fields_in_order(self):
+        self.begin(1)
+        for index, text in enumerate(("x", "y")):
+            self.context = ("j1", index)
+            self.runs.record_item("a", (text,), Facts("h"), ((text, "", ""),))
+        self.runs.end_attempt(1)
+        self.assertEqual(
+            self.lookup(values=(("x", "y"),)), [(("x", "", ""),), (("y", "", ""),)]
+        )
+
+    def test_fails_closed(self):
+        self.record(1)
+        for case, kwargs in (
+            ("other output", dict(values=(("changed",),))),
+            ("other node", dict(name="b")),
+            ("unknown node", dict(name="zzz")),
+            ("no values", dict(values=())),
+            ("malformed values", dict(values="x")),
+        ):
+            with self.subTest(case=case):
+                self.assertIsNone(self.lookup(**kwargs))
+        with self.subTest(case="another job"):
+            self.begin(9, "j9")
+            self.assertIsNone(self.runs.recorded_fields("j8", "a", (("x",),)))
+        with self.subTest(case="no job"):
+            self.runs.end_attempt(9)
+            self.assertIsNone(self.runs.recorded_fields("j9", "a", (("x",),)))
+
+    def test_an_execution_without_fields_gives_none(self):
+        for fields in (None, (("a", "b"),), ((1, "", ""),), (), "abc"):
+            with self.subTest(fields=fields):
+                self.runs = CachedRuns(context=lambda: self.context)
+                self.record(1, fields=fields)
+                self.assertIsNone(self.lookup())
+
+    def test_a_lookup_never_counts_or_marks_the_node_replayed(self):
+        self.record(1)
+        self.lookup()
+        self.assertFalse(self.runs.replayed("a"))
+
+    def test_equal_outputs_with_other_raw_text_conflict_and_still_count(self):
+        key = prompt_node_key(self.g, "a")
+        self.record(1, fields=(("x ", "", ""),))
+        self.record(2, fields=(("x", "", ""),))
+        # Counting dedupes by outputs as before; the fields cannot name either
+        self.assertEqual([r.item_id for r in self.runs._receipts[key]], [2])
+        self.assertIsNone(self.lookup())
+        self.assertIsNotNone(self.runs._match(key, (("x",),)))
+        # The conflict stays: a later execution cannot tell which text it was
+        self.record(3, fields=(("x", "", ""),))
+        self.assertIsNone(self.lookup())
+
+    def test_equal_outputs_with_other_facts_still_count_the_newest(self):
+        key = prompt_node_key(self.g, "a")
+        self.record(1, facts=Facts("old"), fields=(("a x", "", ""),))
+        self.record(2, facts=Facts("new"), fields=(("x", "a ", ""),))
+        items = self.runs._match(key, (("x",),))
+        self.assertEqual([item.facts for item in items], [Facts("new")])
+
+    def test_equal_outputs_and_fields_still_replace(self):
+        key = prompt_node_key(self.g, "a")
+        self.record(1)
+        self.record(2)
+        self.assertEqual([r.item_id for r in self.runs._receipts[key]], [2])
+        self.assertEqual(self.lookup(), [self.F])
+
+    def test_fields_dropped_by_the_budget_do_not_conflict(self):
+        self.runs = CachedRuns(context=lambda: self.context, max_field_bytes=0)
+        self.record(1)
+        self.runs._max_field_bytes = 100
+        self.record(2)
+        self.assertEqual(self.lookup(), [self.F])
+
+    def test_oversized_or_invalid_fields_are_not_kept(self):
+        for raw in ("x" * (module.MAX_FIELD_BYTES + 1), "\ud800"):
+            with self.subTest(size=len(raw)):
+                self.runs = CachedRuns(context=lambda: self.context)
+                self.record(1, fields=((raw, "", ""),))
+                self.assertIsNone(self.lookup())
+                self.assertEqual(self.runs._field_bytes, 0)
+
+    def test_old_receipts_give_up_their_fields_to_new_ones(self):
+        self.runs = CachedRuns(
+            context=lambda: self.context, max_keys=3, max_field_bytes=10
+        )
+        self.record(1, "a", fields=(("123456", "", ""),))
+        self.record(2, "b", fields=(("123456", "", ""),))
+        self.assertEqual(self.runs._field_bytes, 6)
+        self.assertIsNone(self.lookup("a"))
+        self.assertEqual(self.lookup("b"), [(("123456", "", ""),)])
+        # "a" still counts
+        self.assertIsNotNone(self.runs._match(prompt_node_key(self.g, "a"), (("x",),)))
+        # Larger than the whole budget: not kept, nothing else given up
+        self.record(3, "c", fields=(("12345678901", "", ""),))
+        self.assertEqual(self.runs._field_bytes, 6)
+        self.assertEqual(self.lookup("b"), [(("123456", "", ""),)])
+
+    def test_one_receipt_holds_at_most_one_runs_history(self):
+        with mock.patch.object(module, "MAX_RECEIPT_FIELD_BYTES", 8):
+            self.begin(1)
+            for index, raw in enumerate(("12345", "12345")):
+                self.context = ("j1", index)
+                self.runs.record_item("a", (raw,), Facts("h"), ((raw, "", ""),))
+            self.runs.end_attempt(1)
+        self.assertEqual(self.runs._field_bytes, 5)
+        self.assertIsNone(self.lookup(values=(("12345", "12345"),)))
+
+    def test_the_byte_count_follows_replacement_and_eviction(self):
+        self.begin(1)
+        self.runs.record_item("a", ("x",), Facts("h"), (("12345", "", ""),))
+        self.runs.record_item("a", ("x",), Facts("h"), (("123", "", ""),))
+        self.assertEqual(self.runs._field_bytes, 3)
+        self.runs.end_attempt(1)
+        for item_id, raw in ((2, "1"), (3, "22"), (4, "333")):
+            self.record(item_id, output=raw, fields=((raw, "", ""),))
+        # max_receipts=2: only the receipts of jobs 4 and 3 remain
+        self.assertEqual(self.runs._field_bytes, 5)
+        for item_id, name in ((5, "b"), (6, "c")):  # max_keys=2: "a" is dropped
+            self.record(item_id, name, fields=(("1", "", ""),))
+        self.assertEqual(self.runs._field_bytes, 2)
+        self.assertEqual(
+            self.runs._field_bytes,
+            sum(module._bytes(r) for r in self.runs._receipts.values()),
+        )
+
+
+class TestCachedJobsRecordFields(JobCase):
+    """The real nodes keep the texts they gave the history capture (#403)."""
+
+    def lookup(self, g, node_id="pm", prompt_id="job-x"):
+        self.item_id += 1
+        self.runs.begin_attempt(
+            ((self.item_id, prompt_id, g, {}, ["save"], {}), self.item_id)
+        )
+        values = cached_string_outputs(self.executor, node_id, g[node_id]["class_type"])
+        found = self.runs.recorded_fields(prompt_id, node_id, values)
+        self.runs.end_attempt(self.item_id)
+        return found
+
+    def test_each_node_kind_keeps_its_runtime_fields(self):
+        for kind in (
+            "PromptManager",
+            "PromptManagerFocused",
+            "PromptManagerSplit",
+            "PromptManagerText",
+        ):
+            with self.subTest(kind=kind):
+                g = graph(kind)
+                self.first(g, text=f" {LLM} ")
+                self.assertEqual(
+                    self.lookup(graph(kind, seed=2)), [((f" {LLM} ", "", ""),)]
+                )
+
+    def test_pair_keeps_positive_and_negative_fields(self):
+        g = graph("PromptManagerPair", negative_text="typed negative")
+        self.first(g, negative="typed negative")
+        self.assertEqual(
+            self.lookup(
+                graph("PromptManagerPair", negative_text="typed negative", seed=2)
+            ),
+            [((LLM, "", ""), ("typed negative", "", ""))],
+        )
+
+    def test_a_disabled_node_keeps_no_fields(self):
+        g = graph()
+        g["pm"]["inputs"]["save_to_library"] = False
+        self.first(g)
+        g2 = graph(seed=2)
+        g2["pm"]["inputs"]["save_to_library"] = False
+        self.assertIsNone(self.lookup(g2))
+
+
 class TestCacheReader(unittest.TestCase):
     def setUp(self):
         patcher = mock.patch.dict(sys.modules, host_modules())

tests/test_execution_context_host.py

@@ -100,14 +100,19 @@ class HostTests(unittest.TestCase):
         self.executor.caches = types.SimpleNamespace(outputs=HierarchicalCache())
 
     def begin(
-        self, kind="PromptManager", raw=" raw ", prepend=" pre ", append=" post "
+        self,
+        kind="PromptManager",
+        raw=" raw ",
+        prepend=" pre ",
+        append=" post ",
+        negative="negative",
     ):
         g = {
             "p": {
                 "class_type": kind,
                 "inputs": {
                     "text": raw,
-                    "negative_text": "negative",
+                    "negative_text": negative,
                     "prepend_text": prepend,
                     "append_text": append,
                 },
@@ -125,7 +130,7 @@ class HostTests(unittest.TestCase):
                 },
             },
         }
-        if type(raw) is list or type(prepend) is list:
+        if any(type(value) is list for value in (raw, prepend, negative)):
             g["dynamic"] = {"class_type": "RuntimeGenerator", "inputs": {}}
             g["stateful"] = {
                 "class_type": "PromptManagerText",
@@ -445,3 +450,138 @@ class HostTests(unittest.TestCase):
         ref = install_and_remove()
         gc.collect()
         self.assertIsNone(ref())
+
+    # -- #403: a cached runtime text takes its fields from the reused execution
+
+    class Receipts:
+        """A fake ``CachedRuns.recorded_fields``: fixed fields, calls recorded."""
+
+        def __init__(self, fields):
+            self.fields, self.calls = fields, []
+
+        def recorded_fields(self, prompt_id, node_id, values):
+            self.calls.append((prompt_id, node_id, values))
+            return self.fields
+
+    def observe_with(self, receipts, outputs, **begin):
+        self.begin(**begin)
+        self.executor.caches.outputs.entries["p"] = Entry(None, outputs)
+        self.observe("execution_start")
+        host.observe_cached_context(
+            self.executor,
+            "execution_cached",
+            {"prompt_id": JOB, "nodes": ["p"]},
+            receipts=receipts,
+        )
+        return self.seal()
+
+    def test_cached_runtime_raw_comes_from_the_matching_receipt(self):
+        receipts = self.Receipts([(("LLM raw", " pre ", " post "),)])
+        result = self.observe_with(
+            receipts, [object(), ["cached"]], raw=["dynamic", 0]
+        ).positive
+        self.assertEqual(receipts.calls, [(JOB, "p", (("cached",),))])
+        self.assertEqual(result.raw.value, "LLM raw")
+        self.assertEqual(
+            (result.raw.provenance, result.raw.source), ("inferred", "graph_semantics")
+        )
+        self.assertEqual(result.effective.value, "cached")
+        self.assertEqual(result.effective.source, "host_cache_output")
+        # Fields the graph already knows are never replaced
+        self.assertEqual(result.prepend.value, " pre ")
+        self.assertEqual(result.prepend.source, "graph_literal")
+        self.assertEqual(result.append.value, " post ")
+
+    def test_cached_runtime_prepend_comes_from_the_matching_receipt(self):
+        receipts = self.Receipts([((" raw ", "LLM pre", " post "),)])
+        result = self.observe_with(
+            receipts, [object(), ["cached"]], prepend=["dynamic", 0]
+        ).positive
+        self.assertEqual(result.prepend.value, "LLM pre")
+        self.assertEqual(result.prepend.source, "graph_semantics")
+        self.assertEqual(result.raw.value, " raw ")
+        self.assertEqual(result.raw.source, "graph_literal")
+
+    def test_a_receipt_disagreeing_with_a_known_field_fills_nothing(self):
+        receipts = self.Receipts([(("LLM raw", "other pre", " post "),)])
+        result = self.observe_with(
+            receipts, [object(), ["cached"]], raw=["dynamic", 0]
+        ).positive
+        self.assertIsNone(result.raw.value)
+        self.assertEqual(result.prepend.value, " pre ")
+
+    def test_pair_runtime_negative_takes_the_negative_fields(self):
+        receipts = self.Receipts(
+            [((" raw ", " pre ", " post "), ("LLM negative", "", ""))]
+        )
+        result = self.observe_with(
+            receipts,
+            [object(), object(), ["cached positive"], ["LLM negative"]],
+            kind="PromptManagerPair",
+            negative=["dynamic", 0],
+        )
+        self.assertEqual(result.negative.raw.value, "LLM negative")
+        self.assertEqual(result.negative.raw.source, "graph_semantics")
+        self.assertEqual(result.positive.raw.value, " raw ")
+        self.assertEqual(result.positive.raw.source, "graph_literal")
+
+    def test_known_typed_fields_are_never_replaced_by_a_receipt(self):
+        receipts = self.Receipts([(("other", "other", "other"),)])
+        result = self.observe_with(receipts, [object(), ["cached"]]).positive
+        self.assertEqual(
+            (result.raw.value, result.prepend.value, result.append.value),
+            (" raw ", " pre ", " post "),
+        )
+        self.assertEqual(result.raw.source, "graph_literal")
+
+    def test_no_matching_receipt_keeps_the_cached_raw_unknown(self):
+        for fields in (None, [None], [], "bad", [(("a", "b"),)], [((1, "", ""),)]):
+            with self.subTest(fields=fields):
+                result = self.observe_with(
+                    self.Receipts(fields), [object(), ["cached"]], raw=["dynamic", 0]
+                ).positive
+                self.assertIsNone(result.raw.value)
+                self.assertEqual(result.effective.value, "cached")
+
+    def test_a_failing_receipt_lookup_keeps_the_capture(self):
+        class Broken:
+            def recorded_fields(self, *args):
+                raise RuntimeError("boom")
+
+        result = self.observe_with(
+            Broken(), [object(), ["cached"]], raw=["dynamic", 0]
+        ).positive
+        self.assertIsNone(result.raw.value)
+        self.assertEqual(result.effective.value, "cached")
+
+    def test_pair_slots_take_their_own_receipt_fields(self):
+        receipts = self.Receipts(
+            [(("LLM positive", " pre ", " post "), ("LLM negative", "", ""))]
+        )
+        result = self.observe_with(
+            receipts,
+            [object(), object(), ["cached positive"], ["cached negative"]],
+            kind="PromptManagerPair",
+            raw=["dynamic", 0],
+        )
+        self.assertEqual(
+            receipts.calls[0], (JOB, "p", (("cached positive",), ("cached negative",)))
+        )
+        self.assertEqual(result.positive.raw.value, "LLM positive")
+        self.assertEqual(result.positive.effective.value, "cached positive")
+        # The negative editor is typed: the graph's value stays
+        self.assertEqual(result.negative.raw.value, "negative")
+        self.assertEqual(result.negative.raw.source, "graph_literal")
+
+    def test_default_receipts_are_the_process_cached_runs(self):
+        from utils import cached_runs
+
+        receipts = self.Receipts([(("LLM raw", " pre ", " post "),)])
+        with patch.object(cached_runs, "CACHED_RUNS", receipts):
+            self.begin(raw=["dynamic", 0])
+            self.executor.caches.outputs.entries["p"] = Entry(
+                None, [object(), ["cached"]]
+            )
+            self.observe("execution_start")
+            self.observe("execution_cached", nodes=["p"])
+        self.assertEqual(self.seal().positive.raw.value, "LLM raw")

tests/test_execution_history_hooks.py

@@ -54,7 +54,7 @@ def status(kind="execution_success", job=JOB):
     return Status("success", True, [(kind, {"prompt_id": job, "timestamp": 20})])
 
 
-class HistoryHooksTests(unittest.TestCase):
+class HooksCase(unittest.TestCase):
     def setUp(self):
         self.directory = tempfile.TemporaryDirectory()
         self.addCleanup(self.directory.cleanup)
@@ -122,6 +122,8 @@ class HistoryHooksTests(unittest.TestCase):
     def drain(self, registration):
         drain_writer(registration.writer, clock=time.monotonic)
 
+
+class HistoryHooksTests(HooksCase):
     def test_package_history_initializes_when_monitor_and_usage_setup_fail(self):
         try:
             from tests.test_init_package import (
@@ -767,3 +769,150 @@ class HistoryHooksTests(unittest.TestCase):
         self.assertEqual(
             calls, ["execution_start", "execution_cached", "execution_cached"]
         )
+
+
+class CachedRuntimeHistoryTests(HooksCase):
+    """A cached runtime-linked prompt is a tracked run, like an executed one (#403).
+
+    The real history hook and writer, a real ``CachedRuns`` and the real host
+    adapter over fakes of ComfyUI's executor, cache and executing context.
+    """
+
+    TEXT = "a lighthouse the LLM wrote"
+    GRAPH = {
+        "str": {"class_type": "PrimitiveString", "inputs": {"value": "seed"}},
+        "llm": {"class_type": "PreviewAny", "inputs": {"source": ["str", 0]}},
+        "p": {"class_type": "PromptManager", "inputs": {"text": ["llm", 0]}},
+        "s": {"class_type": "KSampler", "inputs": {"positive": ["p", 0]}},
+    }
+
+    def setUp(self):
+        super().setUp()
+        from utils import cached_runs
+        from utils.cached_runs import CachedRuns
+
+        self.context = None
+        self.runs = CachedRuns(context=lambda: self.context or (None, None))
+        self.cache_entries = {}
+        entries = self.cache_entries
+        execution = types.ModuleType("execution")
+        caching = types.ModuleType("comfy_execution.caching")
+        package = types.ModuleType("comfy_execution")
+        utils_module = types.ModuleType("comfy_execution.utils")
+        entry_type = namedtuple("CacheEntry", "ui outputs")
+
+        class PromptExecutor:
+            def add_message(self, event, data, broadcast):
+                return None
+
+        class HierarchicalCache:
+            def get_local(self, node_id):
+                return entries.get(node_id)
+
+        execution.PromptExecutor, execution.CacheEntry = PromptExecutor, entry_type
+        caching.HierarchicalCache = HierarchicalCache
+        package.caching, package.utils = caching, utils_module
+        utils_module.get_executing_context = lambda: (
+            SimpleNamespace(prompt_id=self.context[0], node_id="p", list_index=0)
+            if self.context
+            else None
+        )
+        self.entry_type = entry_type
+        modules = patch.dict(
+            sys.modules,
+            {
+                "execution": execution,
+                "comfy_execution": package,
+                "comfy_execution.caching": caching,
+                "comfy_execution.utils": utils_module,
+            },
+        )
+        modules.start()
+        self.addCleanup(modules.stop)
+        swap = patch.object(cached_runs, "CACHED_RUNS", self.runs)
+        swap.start()
+        self.addCleanup(swap.stop)
+        self.executor = PromptExecutor()
+        self.executor.server = self.server
+        self.executor.caches = SimpleNamespace(outputs=HierarchicalCache())
+        self.prompt_id = self.db.save_prompt(text=self.TEXT)
+        self.item_id = 0
+
+    def job(self, cached):
+        """One job; the node executes unless ``cached`` (ComfyUI's cache serves it)."""
+        from uuid import uuid4
+
+        from utils.cached_runs import stored_facts
+        from utils.execution_context_capture import record_runtime_context
+
+        self.item_id += 1
+        job = str(uuid4())
+        result = ((0, job, self.GRAPH, {}, ["s"]), self.item_id)
+        self.queue.results.append(result)
+        self.queue.get()
+        self.runs.begin_attempt(result)
+        self.context = (job, 0)
+        self.executor.add_message("execution_start", {"prompt_id": job}, False)
+        if cached:
+            self.executor.add_message(
+                "execution_cached", {"prompt_id": job, "nodes": ["p"]}, False
+            )
+        else:
+            # What PromptManager.encode_prompt records for this text
+            record_runtime_context("PromptManager", self.TEXT, "", "", self.TEXT)
+            self.runs.record_item(
+                "p", (self.TEXT,), stored_facts(self.TEXT), ((self.TEXT, "", ""),)
+            )
+            self.cache_entries["p"] = self.entry_type(None, [object(), [self.TEXT]])
+        self.context = None
+        self.queue.task_done(self.item_id, {}, status(job=job))
+        self.runs.end_attempt(self.item_id)
+
+    def tracked_runs(self):
+        with self.db.recorded_history_snapshot():
+            return self.db.list_history_window(self.prompt_id)["recorded_run_count"]
+
+    def test_three_jobs_of_runtime_text_are_three_tracked_runs(self):
+        registration = self.register()
+        for cached in (False, True, True):
+            self.job(cached)
+        self.drain(registration)
+        self.assertEqual(self.tracked_runs(), 3)
+        rows = self.rows()
+        for row in rows:
+            contexts = self.db.get_execution_contexts(row["run_id"])["contexts"]
+            self.assertEqual(contexts[0]["positive"]["raw"], self.TEXT)
+            self.assertEqual(contexts[0]["positive_prompt_id"], self.prompt_id)
+        cached = self.db.get_execution_contexts(rows[1]["run_id"])["contexts"][0]
+        self.assertEqual(
+            cached["evidence"]["positive"]["raw"],
+            {"provenance": "inferred", "source": "graph_semantics", "reason": None},
+        )
+        self.assertEqual(
+            cached["evidence"]["positive"]["effective"]["source"], "host_cache_output"
+        )
+        # The stored history passes the whole-database check older builds run too
+        from database.execution_history_schema import history_schema_error
+
+        self.assertIsNone(history_schema_error(self.db.model.get_connection()))
+        # An image of a cached job can now be attributed through its run (#342)
+        from database.run_attribution import named_prompt
+
+        self.assertEqual(
+            [named_prompt(self.db, row["run_id"]) for row in rows],
+            [self.prompt_id] * 3,
+        )
+
+    def test_without_a_receipt_a_cached_job_stays_untracked(self):
+        registration = self.register()
+        self.job(False)
+        self.runs._receipts.clear()  # as after a restart with a provider entry
+        self.job(True)
+        self.drain(registration)
+        self.assertEqual(self.tracked_runs(), 1)
+        cached = self.db.get_execution_contexts(self.rows()[1]["run_id"])["contexts"]
+        self.assertIsNone(cached[0]["positive"]["raw"])
+        self.assertIsNone(cached[0]["positive_prompt_id"])
+        from database.run_attribution import named_prompt
+
+        self.assertIsNone(named_prompt(self.db, self.rows()[1]["run_id"]))

tests/test_pair_usage_tracking.py

@@ -99,7 +99,10 @@ class TestPairUsageTracking(unittest.TestCase):
             unique_id="1",
         )
         node.cached_runs.record_item.assert_called_once_with(
-            "1", ("cat", ""), Facts(usage_tracking.generate_prompt_hash("cat"), None)
+            "1",
+            ("cat", ""),
+            Facts(usage_tracking.generate_prompt_hash("cat"), None),
+            (("cat", "", ""), ("", "", "")),
         )
 
     def test_validation_rejection_never_counts_or_saves(self):

tests/test_prompt_usage.py

@@ -308,7 +308,11 @@ class TestNodeRecordsCachedRunReceipt(UsageTestCase):
             "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"))
+            "1",
+            ("pre typed text",),
+            Facts(generate_prompt_hash("typed text")),
+            # #403: the texts the history capture got, for a cached job's history
+            (("typed text", "pre", ""),),
         )
         # Hook-visible text: the node counts nothing itself
         row = self.db.get_prompt_by_hash(generate_prompt_hash("typed text"))
@@ -319,7 +323,9 @@ class TestNodeRecordsCachedRunReceipt(UsageTestCase):
 
         node = self._node()
         node.process_text("  ", prompt=self.GRAPH, unique_id="1")
-        node.cached_runs.record_item.assert_called_once_with("1", ("  ",), Facts(None))
+        node.cached_runs.record_item.assert_called_once_with(
+            "1", ("  ",), Facts(None), (("  ", "", ""),)
+        )
 
 
 if __name__ == "__main__":

utils/cached_runs.py

@@ -15,9 +15,14 @@ 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.
 
+The receipt also keeps the raw, prepend and append text the node gave the
+execution history (#403), so the history capture of a cached job can record
+the raw text the graph cannot show (``recorded_fields``) and the job becomes a
+tracked run of its prompt, as it is a counted one.
+
 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.
+this process, nothing is counted (the behavior before #347) and the history
+fields stay unknown. In memory only and bounded; starts no threads.
 """
 
 import hashlib
@@ -39,6 +44,7 @@ try:
     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
+    from .execution_contexts import MAX_ATTEMPT_BYTES, MAX_FIELD_BYTES, text_size
 except ImportError:
     from utils.hashing import generate_prompt_hash
     from utils.logging_config import get_logger
@@ -53,6 +59,7 @@ except ImportError:
     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
+    from utils.execution_contexts import MAX_ATTEMPT_BYTES, MAX_FIELD_BYTES, text_size
 
 logger = get_logger("prompt_manager.cached_runs")
 
@@ -61,9 +68,15 @@ 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
+MAX_FIELD_TEXT_BYTES = 8 * 1024 * 1024  # history fields all receipts keep
+MAX_RECEIPT_FIELD_BYTES = MAX_ATTEMPT_BYTES  # one receipt: what one run's history holds
 _EMPTY_HASH = generate_prompt_hash("")
 
 Count = Tuple[Any, ...]  # ("prompt", hash) or ("pair", positive, negative or None)
+# (raw, prepend, append) per STRING slot, as given to record_runtime_context
+Fields = Tuple[Tuple[str, str, str], ...]
+# Executions with equal outputs recorded different fields: none can be named
+_CONFLICT: Fields = ()
 
 
 class Facts(NamedTuple):
@@ -96,6 +109,8 @@ class Seen(NamedTuple):
 class _Item(NamedTuple):
     outputs: Tuple[str, ...]
     facts: Facts
+    fields: Optional[Fields] = None
+    field_bytes: int = 0
 
 
 class _Receipt:
@@ -108,6 +123,18 @@ class _Receipt:
     def outputs(self) -> Dict[int, Tuple[str, ...]]:
         return {index: item.outputs for index, item in self.items.items()}
 
+    def field_bytes(self) -> int:
+        return sum(item.field_bytes for item in self.items.values())
+
+    def strip_fields(self) -> int:
+        """Drop the history fields (counting is unchanged); the bytes freed."""
+        freed = 0
+        for index, item in list(self.items.items()):
+            if item.field_bytes:
+                freed += item.field_bytes
+                self.items[index] = item._replace(fields=None, field_bytes=0)
+        return freed
+
 
 class _Attempt(NamedTuple):
     item_id: int
@@ -117,6 +144,22 @@ class _Attempt(NamedTuple):
     replayed: set
 
 
+def _field_size(fields: Any, slots: int) -> Optional[int]:
+    """UTF-8 size of well-formed fields for ``slots`` outputs, else None."""
+    if type(fields) is not tuple or len(fields) != slots:
+        return None
+    size = 0
+    for slot in fields:
+        if type(slot) is not tuple or len(slot) != 3:
+            return None
+        for value in slot:
+            try:
+                size += text_size(value, MAX_FIELD_BYTES)  # the capture's own check
+            except ValueError:
+                return None
+    return size
+
+
 def text_digest(value: str) -> str:
     """Digest of one returned STRING value, exactly as returned."""
     return hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()
@@ -245,11 +288,14 @@ class CachedRuns:
         max_keys: int = MAX_KEYS,
         max_receipts: int = MAX_RECEIPTS,
         max_items: int = MAX_ITEMS,
+        max_field_bytes: int = MAX_FIELD_TEXT_BYTES,
         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._max_field_bytes = max_field_bytes
+        self._field_bytes = 0  # UTF-8 bytes of the fields all receipts keep
         self._context = context
         self._receipts: "OrderedDict[str, List[_Receipt]]" = OrderedDict()
         self._attempt: Optional[_Attempt] = None
@@ -284,7 +330,10 @@ class CachedRuns:
                 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)
+                    before = _bytes(receipts)
+                    deduplicated = _deduplicated(receipts)
+                    self._field_bytes += _bytes(deduplicated) - before
+                    self._receipts[run_key] = deduplicated
 
     def _current(self) -> Tuple[Optional[_Attempt], Optional[int]]:
         """The attempt the executing node belongs to, and its list index."""
@@ -305,8 +354,19 @@ class CachedRuns:
         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."""
+    def record_item(
+        self,
+        node_id: Any,
+        outputs: Tuple[Any, ...],
+        facts: Facts,
+        fields: Optional[Fields] = None,
+    ) -> None:
+        """Remember what this execution returned and was entitled to count.
+
+        ``fields`` are the texts the node gave the execution history, one
+        ``(raw, prepend, append)`` per output; kept only when well formed,
+        each within the history's field limit and the store within its budget.
+        """
         try:
             attempt, index = self._current()
             node = attempt.nodes.get(str(node_id)) if attempt else None
@@ -317,14 +377,20 @@ class CachedRuns:
                 return
             if any(not isinstance(value, str) for value in outputs):
                 return
-            item = _Item(tuple(text_digest(value) for value in outputs), facts)
+            size = _field_size(fields, len(outputs)) if fields is not None else None
+            item = _Item(
+                tuple(text_digest(value) for value in outputs),
+                facts,
+                fields if size is not None else None,
+                size or 0,
+            )
             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, [])
+            receipts = self._take(run_key)
             if receipts and receipts[0].item_id == item_id:
                 receipt = receipts[0]
             else:
@@ -334,9 +400,44 @@ class CachedRuns:
                 receipt = _Receipt(item_id)
                 receipts.insert(0, receipt)
             receipt.items[index] = item
-            self._receipts[run_key] = receipts[: self._max_receipts]
+            receipts = receipts[: self._max_receipts]
+            if item.field_bytes and (
+                receipt.field_bytes() > MAX_RECEIPT_FIELD_BYTES
+                or not self._make_room(_bytes(receipts))
+            ):
+                # Counting is as before; this job's history fields stay unknown
+                receipt.items[index] = item._replace(fields=None, field_bytes=0)
+            self._put(run_key, receipts)
             while len(self._receipts) > self._max_keys:
-                self._receipts.popitem(last=False)
+                _, dropped = self._receipts.popitem(last=False)
+                self._field_bytes -= _bytes(dropped)
+
+    def _make_room(self, needed: int) -> bool:
+        """Free the fields of the least recently used receipts (under the lock).
+
+        Old receipts of workflows never run again must not hold the budget:
+        their counting stays, only their history fields go. False when
+        ``needed`` does not fit even then.
+        """
+        if needed > self._max_field_bytes:
+            return False
+        for receipts in self._receipts.values():  # least recently used first
+            if self._field_bytes + needed <= self._max_field_bytes:
+                break
+            for receipt in receipts:
+                self._field_bytes -= receipt.strip_fields()
+        return self._field_bytes + needed <= self._max_field_bytes
+
+    def _take(self, run_key: str) -> List[_Receipt]:
+        """Remove a key's receipts and their bytes from the store (under the lock)."""
+        receipts = self._receipts.pop(run_key, [])
+        self._field_bytes -= _bytes(receipts)
+        return list(receipts)
+
+    def _put(self, run_key: str, receipts: List[_Receipt]) -> None:
+        """Store a key's receipts as the most recently used (under the lock)."""
+        self._receipts[run_key] = receipts
+        self._field_bytes += _bytes(receipts)
 
     # -- cached nodes -------------------------------------------------------
 
@@ -393,25 +494,11 @@ class CachedRuns:
 
     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]
+            matches = self._matching(run_key, values)
             if not matches:
                 return None
-            self._receipts.move_to_end(run_key)
+            length = len(values[0])
             facts = {
                 tuple(receipt.items[index].facts for index in range(length))
                 for receipt in matches
@@ -420,13 +507,99 @@ class CachedRuns:
                 return None
             return [matches[0].items[index] for index in range(length)]
 
+    def _matching(self, run_key: str, values: Any) -> List[_Receipt]:
+        """Receipts of ``run_key`` whose outputs equal ``values`` (under the lock)."""
+        if type(values) not in (list, tuple) or not values:
+            return []
+        if any(type(slot) not in (list, tuple) for slot in values):
+            return []
+        length = len(values[0])
+        if length == 0 or length > self._max_items:
+            return []
+        if any(len(slot) != length for slot in values):
+            return []
+        if any(type(value) is not str for slot in values for value in slot):
+            return []
+        wanted = {
+            index: tuple(text_digest(slot[index]) for slot in values)
+            for index in range(length)
+        }
+        receipts = self._receipts.get(run_key)
+        if not receipts:
+            return []
+        matches = [receipt for receipt in receipts if receipt.outputs() == wanted]
+        if matches:
+            self._receipts.move_to_end(run_key)
+        return matches
+
+    # -- the history capture (#403) -------------------------------------------
+
+    def recorded_fields(
+        self, prompt_id: Any, node_id: Any, values: Any
+    ) -> Optional[List[Fields]]:
+        """The history fields of the execution a cached node reuses, per list item.
+
+        ``values`` are the node's cached STRING outputs (``cached_string_outputs``).
+        Only for a node of the job currently executing, and only when every
+        receipt matching ``values`` kept the same fields; None otherwise. Never
+        counts and never marks the node replayed.
+        """
+        try:
+            with self._lock:
+                attempt = self._attempt
+                if (
+                    attempt is None
+                    or type(prompt_id) is not str
+                    or prompt_id != attempt.prompt_id
+                    or type(node_id) is not str
+                    or node_id not in attempt.nodes
+                ):
+                    return None
+                matches = self._matching(attempt.nodes[node_id][1], values)
+                if not matches:
+                    return None
+                found = []
+                for index in range(len(values[0])):
+                    fields = {receipt.items[index].fields for receipt in matches}
+                    if len(fields) != 1 or None in fields or _CONFLICT in fields:
+                        return None
+                    found.append(fields.pop())
+                return found
+        except Exception as error:
+            logger.debug(f"Could not read a cached run's history fields: {error}")
+            return None
+
 
 def _deduplicated(receipts: List[_Receipt]) -> List[_Receipt]:
-    """The newest receipt replaces older ones with the same outputs."""
+    """The newest receipt replaces older ones with the same outputs.
+
+    Counting needs only the outputs. When a replaced receipt recorded other
+    history fields for equal outputs (a raw text the outputs do not show),
+    the newest item's fields become a conflict, so no cached job of it is
+    given either text (#403). A conflict stays until the run key is dropped.
+    """
     if not receipts:
         return receipts
-    outputs = receipts[0].outputs()
-    return [receipts[0]] + [r for r in receipts[1:] if r.outputs() != outputs]
+    newest, outputs = receipts[0], receipts[0].outputs()
+    kept = [newest]
+    for receipt in receipts[1:]:
+        if receipt.outputs() != outputs:
+            kept.append(receipt)
+            continue
+        for index, item in list(newest.items.items()):
+            older = receipt.items[index].fields
+            if older == _CONFLICT or (
+                older is not None and item.fields is not None and older != item.fields
+            ):
+                newest.items[index] = item._replace(fields=_CONFLICT, field_bytes=0)
+    return kept
+
+
+def _bytes(receipts: List[_Receipt]) -> int:
+    """UTF-8 bytes of the history fields ``receipts`` keep."""
+    return sum(
+        item.field_bytes for receipt in receipts for item in receipt.items.values()
+    )
 
 
 def _job_nodes(graph: Dict[str, Any], outputs: Any) -> Dict[str, Tuple[str, str]]:

utils/execution_context_host.py

@@ -151,10 +151,77 @@ def cached_string_outputs(
         return None
 
 
+def _receipt_fields(
+    receipts: object, executor: object, job_id: str, source: object
+) -> list | None:
+    """Per list index, the ``(raw, prepend, append)`` of the execution reused (#403).
+
+    From the #347 receipt whose digests equal every STRING slot of the cache
+    entry; None when there is none or the answer is malformed.
+    """
+    try:
+        if receipts is None:
+            try:
+                from .cached_runs import CACHED_RUNS as receipts
+            except ImportError:
+                from utils.cached_runs import CACHED_RUNS as receipts
+        values = cached_string_outputs(executor, source.node_id, source.node_type)
+        if values is None:
+            return None
+        found = receipts.recorded_fields(job_id, source.node_id, values)
+        if type(found) is not list or len(found) != len(values[0]):
+            return None
+        position = (
+            (1 if source.output_slot in (1, 3) else 0)
+            if source.node_type == "PromptManagerPair"
+            else 0
+        )
+        fields = []
+        for slots in found:
+            slot = slots[position] if type(slots) is tuple else None
+            if not (
+                type(slot) is tuple
+                and len(slot) == 3
+                and all(type(value) is str for value in slot)
+            ):
+                return None
+            fields.append(slot)
+        return fields
+    except Exception:
+        return None
+
+
+def _agrees(inferred: PromptSnapshot, fields: tuple) -> bool:
+    """Every field the graph knows equals the receipt's; else fill nothing."""
+    known = (inferred.raw, inferred.prepend, inferred.append)
+    return all(
+        field.value is None or field.value == value
+        for field, value in zip(known, fields)
+    )
+
+
+def _filled(inferred: TextEvidence, value: str) -> TextEvidence:
+    """A field the graph cannot show, taken from the reused execution (#403)."""
+    if inferred.value is not None:
+        return inferred
+    return TextEvidence(value, "inferred", "graph_semantics", None)
+
+
 def observe_cached_context(
-    executor: object, message_type: object, message_data: object
+    executor: object,
+    message_type: object,
+    message_data: object,
+    *,
+    receipts: object = None,
 ) -> None:
-    """Copy bounded local strings without cache/provider execution or retention."""
+    """Copy bounded local strings without cache/provider execution or retention.
+
+    Raw, prepend and append come from graph inference; a field it leaves
+    unknown (runtime-only text) is taken from the receipt of the execution the
+    cache entry is proven to come from (``CachedRuns.recorded_fields``), so the
+    cached job is a tracked run of its prompt (#403). ``receipts`` defaults to
+    the process ``CACHED_RUNS``.
+    """
     try:
         binding = active_capture()
         if binding is None or type(message_type) is not str:
@@ -198,18 +265,29 @@ def observe_cached_context(
             inferred = manager.source_snapshot(cap, source)
             if inferred is None:
                 continue
+            fields = None
+            if any(
+                field.value is None
+                for field in (inferred.raw, inferred.prepend, inferred.append)
+            ):
+                fields = _receipt_fields(receipts, executor, cap.job_id, source)
             for index, value in enumerate(values):
                 if type(value) is not str:
                     continue
+                raw, prepend, append = inferred.raw, inferred.prepend, inferred.append
+                if fields is not None and _agrees(inferred, fields[index]):
+                    raw = _filled(raw, fields[index][0])
+                    prepend = _filled(prepend, fields[index][1])
+                    append = _filled(append, fields[index][2])
                 manager.observe(
                     cap,
                     source,
                     index,
                     PromptSnapshot(
-                        inferred.raw,
+                        raw,
                         TextEvidence(value, "recorded", "host_cache_output", None),
-                        inferred.prepend,
-                        inferred.append,
+                        prepend,
+                        append,
                     ),
                 )
     except Exception:

web/js/workspace/library-selection.mjs

@@ -2,19 +2,23 @@ import { createButton } from '../shared/toolkit/core.js';
 import { selectionLabel, toggleSelection } from './images-state.mjs';
 import { applyKeys, rangeKeys } from './images-multiselect.mjs';
 import { CATEGORY_DIALOG, CATEGORY_FAILED, CATEGORY_MAX_LENGTH, setOutcome } from './images-category.mjs';
+import { NO_TAGS, parseBulkTags } from './images-tagging-bulk.mjs';
 import { openInputDialog } from './workspace-dialogs.mjs';
+import { countOf } from './plural.mjs';
 import { showWorkspaceToast } from './toast.mjs';
 
 /**
- * The Prompt library's selection and its one bulk action, "Set category" (#437, D148, beyond the concept): the
- * legacy admin page's "Set Category" in the workspace. Built from the concept's Images parts only: each card's
- * `.image-overlays` checkbox and the `.bulkbar` row ("N selected", the action, Clear). The selection is a set of
+ * The Prompt library's selection and its bulk actions (D148, D151, beyond the concept): "Add tags" (#442, D151) and "Set
+ * category" (#437), the legacy admin page's bulk Add Tags and Set Category in the workspace. Built from the concept's
+ * Images parts only: each card's `.image-overlays` checkbox and the `.bulkbar` row ("N selected", the actions in the
+ * Images bar's order, Clear). The selection is a set of
  * prompt ids over the shelf's loaded items, so windowed-out cards keep their state; Shift+click on a checkbox applies
  * the anchor's range (the Images helpers, D135). The dialog, the range helpers, the count and the toast text are the
  * Images bar's own; one bar core shared with images-selection.mjs is a follow-up (D148).
  *
- * Apply posts `{prompt_ids, category, expected_epoch}` to `POST /prompt_manager/bulk/category`: the epoch the shelf
- * was read under, so a restored storage answers 409 `storage-changed` and nothing is written.
+ * Apply posts `{prompt_ids, category|tags, expected_epoch}` to `POST /prompt_manager/bulk/category` or `/bulk/tags`:
+ * the epoch the shelf was read under, so a restored storage answers 409 `storage-changed` and nothing is written.
+ * Add tags writes prompt tags only, never image tags or auto-tags.
  */
 export const CATEGORY_URL = '/prompt_manager/bulk/category';
 export const PROMPT_CATEGORY_DIALOG = Object.freeze({ ...CATEGORY_DIALOG, title: 'Categorize selected prompts' });
@@ -26,6 +30,17 @@ export const UNCONFIRMED = 'The category change could not be confirmed. The Libr
 export const NONE_FOUND = 'None of the selected prompts could be found. The Library was reloaded.';
 export const CONTROL_CHARACTERS = 'A category cannot contain control characters.';
 export { CATEGORY_FAILED };
+export const TAGS_URL = '/prompt_manager/bulk/tags';
+export const TAG_OPTIONS_URL = '/prompt_manager/tags';
+export const PROMPT_TAGS_DIALOG = Object.freeze({ title: 'Tag selected prompts', label: 'Tags, separated by commas', action: 'Apply' });
+// What the editor's save accepts for one prompt (`validate_tags`): the route refuses more.
+export const MAX_PROMPT_TAGS = 20;
+export const MAX_TAG_LENGTH = 50;
+export const TAG_PROBLEMS = Object.freeze({ none: NO_TAGS, many: 'Enter at most 20 tags.',
+    long: 'A tag can have at most 50 characters.', control: 'A tag cannot contain control characters.' });
+export const TAGS_FAILED = 'Tags could not be added. Try again.';
+export const TAGS_STORAGE_CHANGED = 'Prompt storage was restored since this page loaded. No tags were added.';
+export const TAGS_UNCONFIRMED = 'Adding the tags could not be confirmed. The Library was reloaded.';
 // Answers that may follow a committed write (the Library editor's unknown outcome): no status, or a gateway's.
 const UNCONFIRMED_STATUSES = new Set([502, 503, 504]);
 
@@ -43,16 +58,56 @@ function el(tag, text, className) {
  * What a settled write means for the page: `toast` to show, whether the write may have `committed` (the page then
  * invalidates its cached Library views) and whether the shelf must `reload` (which clears the selection).
  */
-export function writeOutcome(result, failure) {
+export function writeOutcome(result, failure, kind = 'category') {
+    const copy = WRITES[kind];
     if (failure === null && result?.success) {
-        const found = (Number(result.updated) || 0) + (Number(result.unchanged) || 0);
+        const found = ['updated', 'unchanged', 'over_limit'].reduce((sum, key) => sum + (Number(result[key]) || 0), 0);
         if (!found) return { toast: NONE_FOUND, committed: false, reload: true };
-        return { toast: setOutcome(result, result.category, 'prompt'), committed: true, reload: true };
+        return { toast: copy.success(result), committed: (Number(result.updated) || 0) > 0, reload: true };
     }
-    if (failure?.code === 'storage-changed') return { toast: STORAGE_CHANGED, committed: false, reload: true };
+    if (failure?.code === 'storage-changed') return { toast: copy.storage, committed: false, reload: true };
     const status = failure?.status;
-    if (!Number.isInteger(status) || UNCONFIRMED_STATUSES.has(status)) return { toast: UNCONFIRMED, committed: true, reload: true };
-    return { toast: CATEGORY_FAILED, committed: false, reload: false };
+    if (!Number.isInteger(status) || UNCONFIRMED_STATUSES.has(status)) return { toast: copy.unconfirmed, committed: true, reload: true };
+    return { toast: copy.failed, committed: false, reload: false };
+}
+
+/** The Add tags toast (#442): what was added, then what already had them, was skipped or could not be found. */
+export function tagsOutcome(result) {
+    const updated = Number(result?.updated) || 0, unchanged = Number(result?.unchanged) || 0;
+    const skipped = Number(result?.over_limit) || 0, missing = Number(result?.not_found) || 0;
+    const names = Array.isArray(result?.tags) ? result.tags.join(', ') : '';
+    const parts = [];
+    if (updated) parts.push(`Added “${names}” to ${countOf(updated, 'prompt')}.`);
+    if (unchanged) parts.push(`${countOf(unchanged, 'prompt')} already had ${updated ? 'them' : `“${names}”`}.`);
+    if (skipped) parts.push(`${countOf(skipped, 'prompt')} would pass ${MAX_PROMPT_TAGS} tags and ${skipped === 1 ? 'was' : 'were'} skipped.`);
+    if (missing) parts.push(`${missing.toLocaleString('en-US')} could not be found.`);
+    return parts.join(' ');
+}
+
+const WRITES = Object.freeze({
+    category: { success: result => setOutcome(result, result.category, 'prompt'), storage: STORAGE_CHANGED,
+        unconfirmed: UNCONFIRMED, failed: CATEGORY_FAILED },
+    tags: { success: tagsOutcome, storage: TAGS_STORAGE_CHANGED, unconfirmed: TAGS_UNCONFIRMED, failed: TAGS_FAILED },
+});
+
+/**
+ * "Select all N loaded" (#444 ruling, D151; the legacy page's select-all checkbox): offered while some loaded prompts
+ * are unchecked and no more are loaded than an action accepts. It sits where D135's "Select all N matching" sits on
+ * Images, but that offer appears once every loaded image is checked; this one reaches only the loaded prompts.
+ */
+export function offersLoaded(selected, keys) {
+    return selected.size > 0 && keys.length <= MAX_CATEGORY_PROMPTS && keys.some(key => !selected.has(key));
+}
+export const loadedOffer = count => `Select all ${count.toLocaleString('en-US')} loaded`;
+
+/** The Add tags field's check before Apply: the parsed names, as `validate_tags` would judge them. */
+export function tagsProblem(text) {
+    const tags = parseBulkTags(text);
+    if (!tags.length) return TAG_PROBLEMS.none;
+    if (tags.length > MAX_PROMPT_TAGS) return TAG_PROBLEMS.many;
+    if (tags.some(tag => tag.length > MAX_TAG_LENGTH)) return TAG_PROBLEMS.long;
+    if (tags.some(tag => /[\x00-\x1f]/.test(tag))) return TAG_PROBLEMS.control;
+    return null;
 }
 
 /** The field's check before Apply: what `validate_category` refuses beyond the length the field already limits. */
@@ -63,7 +118,7 @@ export function categoryProblem(value) {
 /**
  * Mount the selection on the shelf `grid`. `getItems()` is the shelf's loaded items in order, `getEpoch()` the
  * storage epoch they were read under, `getCategories()` the Library's category list (the field's suggestions).
- * `onCommitted(receipt)` runs after a write that may have changed prompts, even once the page is gone (the page
+ * `onCommitted(kind, receipt)` (`category` or `tags`) runs after a write that may have changed prompts, even once the page is gone (the page
  * invalidates its cached Library views); `onReload()` then reloads the mounted shelf (which calls `clear`). Focus moves
  * to `focusFallback` when it was on the bar the reload hides.
  */
@@ -73,9 +128,19 @@ export function mountLibrarySelection({ scope, api, grid, getItems, getEpoch, ge
     const element = el('div');
     element.id = 'library-bulkbar';
     const bulk = el('div', undefined, 'bulkbar'), count = el('strong');
+    const addTags = createButton('Add tags');
     const setCategory = createButton('Set category');
     const clearButton = createButton('Clear');
-    bulk.append(count, setCategory, clearButton);
+    const selectLoaded = createButton('');
+    const actions = [addTags, setCategory, selectLoaded];
+    bulk.append(count, addTags, setCategory, clearButton);
+    scope.listen(selectLoaded, 'click', () => {
+        if (writing) return;
+        selectAll();
+        // The pressed offer left the bar: focus moves to its first action, as after D135's offer.
+        addTags.focus({ preventScroll: true });
+    });
+    scope.listen(addTags, 'click', () => openTags());
     scope.listen(setCategory, 'click', () => openCategory());
     scope.listen(clearButton, 'click', () => { clear(); focusFallback.focus({ preventScroll: true }); });
     // Escape in the bar clears the selection; the bar hides, so focus moves to the page heading as after Clear.
@@ -91,6 +156,12 @@ export function mountLibrarySelection({ scope, api, grid, getItems, getEpoch, ge
         if (!checked.size) { bulk.remove(); return; }
         const label = selectionLabel(checked.size);
         if (count.textContent !== label) count.textContent = label;
+        const keys = ids();
+        if (offersLoaded(checked, keys)) {
+            const offer = loadedOffer(keys.length);
+            if (selectLoaded.textContent !== offer) selectLoaded.textContent = offer;
+            if (count.nextElementSibling !== selectLoaded) count.after(selectLoaded);
+        } else selectLoaded.remove();
         if (bulk.parentNode !== element) element.append(bulk);
     }
     function paint() {
@@ -104,6 +175,10 @@ export function mountLibrarySelection({ scope, api, grid, getItems, getEpoch, ge
         anchor = null;
         if (checked.size) update(Object.freeze(new Set()));
     }
+    /** Every loaded prompt, windowed-out cards included; the anchor stays. */
+    function selectAll() {
+        update(applyKeys(checked, ids(), true));
+    }
     // The checkbox toggles natively; a Shift+click applies the anchor's range with the box's new state.
     scope.listen(grid, 'click', event => {
         const box = event.target.closest('.image-overlays > label');
@@ -134,32 +209,53 @@ export function mountLibrarySelection({ scope, api, grid, getItems, getEpoch, ge
         return overlays;
     }
 
-    function openCategory() {
-        if (writing) return;
+    /** The selected loaded prompts and the shelf epoch, or null after a toast says why nothing can be sent. */
+    function target(failed) {
+        if (writing) return null;
         const loaded = new Set(ids()), selected = [...checked].filter(id => loaded.has(id));
-        if (selected.length > MAX_CATEGORY_PROMPTS) { showWorkspaceToast(TOO_MANY_PROMPTS); return; }
+        if (selected.length > MAX_CATEGORY_PROMPTS) { showWorkspaceToast(TOO_MANY_PROMPTS); return null; }
         const epoch = getEpoch();
-        if (!selected.length || typeof epoch !== 'string') { showWorkspaceToast(CATEGORY_FAILED); return; }
+        if (!selected.length || typeof epoch !== 'string') { showWorkspaceToast(failed); return null; }
+        return { prompt_ids: selected, expected_epoch: epoch };
+    }
+    function openCategory() {
+        const chosen = target(CATEGORY_FAILED);
+        if (!chosen) return;
         openInputDialog({ scope, trigger: setCategory, title: PROMPT_CATEGORY_DIALOG.title, label: PROMPT_CATEGORY_DIALOG.label,
             action: PROMPT_CATEGORY_DIALOG.action, maxLength: CATEGORY_MAX_LENGTH, suggestions: getCategories(),
-            validate: categoryProblem, onApply: category => write({ prompt_ids: selected, category, expected_epoch: epoch }) });
+            validate: categoryProblem, onApply: category => write(CATEGORY_URL, { ...chosen, category }, 'category') });
+    }
+    /** Add tags (#442): the tag vocabulary as suggestions, the editor's list; a failed read opens without them. */
+    async function openTags() {
+        const chosen = target(TAGS_FAILED);
+        if (!chosen) return;
+        let suggestions = null;
+        try {
+            const result = await api.get(TAG_OPTIONS_URL, { signal: scope.signal, cache: 'no-store' });
+            if (Array.isArray(result?.tags)) suggestions = result.tags.filter(tag => typeof tag === 'string');
+        } catch {
+            // Suggestions are auxiliary; the field works without them.
+        }
+        if (scope.signal.aborted) return;
+        openInputDialog({ scope, trigger: addTags, ...PROMPT_TAGS_DIALOG, suggestions, validate: tagsProblem,
+            onApply: text => write(TAGS_URL, { ...chosen, tags: parseBulkTags(text) }, 'tags') });
     }
     /** The write carries no page signal: once sent it completes even if the page is left. A failure keeps the selection. */
-    async function write(body) {
+    async function write(url, body, kind) {
         writing = true;
-        setCategory.setAttribute('aria-disabled', 'true');
+        for (const button of actions) button.setAttribute('aria-disabled', 'true');
         let result = null, failure = null;
         try {
-            result = await api.post(CATEGORY_URL, body);
+            result = await api.post(url, body);
         } catch (error) {
             failure = error;
         } finally {
             writing = false;
-            setCategory.removeAttribute('aria-disabled');
+            for (const button of actions) button.removeAttribute('aria-disabled');
         }
-        const outcome = writeOutcome(result, failure);
+        const outcome = writeOutcome(result, failure, kind);
         // A left page still invalidates what it cached, so a return never shows the old categories.
-        if (outcome.committed) onCommitted(result?.success ? result : { phase: 'unconfirmed' });
+        if (outcome.committed) onCommitted(kind, result?.success ? result : { phase: 'unconfirmed' });
         if (scope.signal.aborted) return;
         const focused = element.contains(document.activeElement);
         showWorkspaceToast(outcome.toast);
@@ -173,7 +269,11 @@ export function mountLibrarySelection({ scope, api, grid, getItems, getEpoch, ge
         element,
         overlayFor,
         isChecked: id => checked.has(id),
+        hasSelection: () => checked.size > 0,
         clear,
+        selectAll,
+        /** A loaded batch: the count and the offer follow it. */
+        listingChanged: () => { if (checked.size) render(); },
         /** Whether a Set category write is still in flight. */
         writing: () => writing,
     };

web/js/workspace/library.mjs

@@ -178,7 +178,7 @@ export function mountLibrary(main,api,parentScope,{positiveId,savedState=null,on
         let categoryNames=[];
         const selection=mountLibrarySelection({scope,api,grid,getItems:()=>state?.items ?? [],getEpoch:()=>state?.epoch ?? null,
             getCategories:()=>categoryNames,focusFallback:{focus:()=>focusHeading(main)},
-            onCommitted:receipt=>invalidateLibraryAfterWrite({kind:'bulk-category',receipt}),onReload:()=>{loadCategories();reset();}});
+            onCommitted:(kind,receipt)=>invalidateLibraryAfterWrite({kind:`bulk-${kind}`,receipt}),onReload:()=>{loadCategories();reset();}});
         const message=el('p','','library-status');message.setAttribute('role','status');
         const continuation=el('div',undefined,'library-continuation');
         const more=createButton('Load more prompts');const retry=createButton('Retry prompt library');retry.hidden=true;
@@ -364,6 +364,9 @@ export function mountLibrary(main,api,parentScope,{positiveId,savedState=null,on
         scope.listen(grid,'keydown',event=>{
             const card=event.target.closest('article[data-prompt-id]');if(!card) return;
             const index=Number(card.dataset.index);
+            // D151 (#444 ruling): Ctrl/Cmd+A selects every loaded prompt and Escape clears, as on Images (focus stays).
+            if((event.ctrlKey || event.metaKey) && !event.altKey && event.key.toLowerCase()==='a') {event.preventDefault();selection.selectAll();return;}
+            if(event.key==='Escape' && selection.hasSelection()) {event.preventDefault();selection.clear();return;}
             if(event.target.matches('[data-history]') && ['Home','End','ArrowLeft','ArrowRight'].includes(event.key)) {event.preventDefault();focusCard(event.key==='Home'?0:event.key==='End'?state.items.length-1:index+(event.key==='ArrowLeft'?-1:1),false,true);}
             else if(event.key==='Tab') {
                 const controls=card.querySelectorAll(CARD_CONTROLS);const next=index+(event.shiftKey?-1:1);
@@ -372,7 +375,8 @@ export function mountLibrary(main,api,parentScope,{positiveId,savedState=null,on
             }
         });
         function render(next) {
-            if(items!==next.items) {
+            const batch=items!==next.items;
+            if(batch) {
                 pendingAnchor ||= captureAnchor();items=next.items;indexesById=new Map(items.map((item,index)=>[item.id,index]));dirty=true;
                 if(!items.length) {
                     anchor=null;pendingAnchor=null;heights.clear();layout={rows:[],height:0};
@@ -381,6 +385,8 @@ export function mountLibrary(main,api,parentScope,{positiveId,savedState=null,on
                 } else if(retainedAnchor) {pendingAnchor=retainedAnchor;retainedAnchor=null;}
             }
             state=next;
+            // A loaded batch: the selection bar's count and Select all N loaded follow it (D151).
+            if(batch) selection.listingChanged();
             if(next.variantTotal!==null) {
                 const count=`${countOf(next.total,'positive prompt')} · ${countOf(next.variantTotal,'negative variant')}`;
                 if(captionCount.textContent!==count) captionCount.textContent=count;caption.hidden=false;