fix(ingestion): the switch walk also skips any input named with "negative"

#90's rule matches only whole underscore-separated words, so negative2,
negativeprompt and NegativePrompt counted as positive inputs and an off
node wired only to one of them blocked the file. The walk skips those too;
follows_input is unchanged.

.github/workflows/viewerjs-check.yml

@@ -27,6 +27,15 @@ concurrency:
 # coverage shards 7m34s/6m04s/5m22s (tests 6m38s/4m53s/4m04s), merge 54s, firefox 7m57s/5m21s (tests
 # 7m08s/4m34s), webkit 8m16s/5m03s (tests 7m26s/3m58s); a cache miss adds up to 5m to npm ci, which
 # the 20-minute limits absorb.
+# Hangs (#425): a job killed at its limit keeps no log and skips its upload steps, so each browser
+# test step has its own timeout-minutes (18) below the job's 20. At the step limit, SIGTERM makes
+# tools/vendor/running-tests-reporter.mjs print the tests still running and hand over to Playwright's
+# interrupt, which writes the blob report; the always() uploads then run.
+# Measured on main (runs 40725289094, 40719375442, 40700439399): setup before the test step <= 52s,
+# uploads <= 35s; test steps up to 14m42s (webkit shard 1/2) and 13m05s for coverage (shard 1/3:
+# 12m53s, node units and the viewer suite before 11.8m of workflows), so coverage is the tight one
+# in its 18 minutes. This holds only when the npm cache hits: a cold cache adds up to 5m to npm ci,
+# and then the job limit can still end the job before the step limit does.
 jobs:
   frontend-coverage:
     name: frontend coverage (shard ${{ matrix.shard }}/3)
@@ -79,6 +88,7 @@ jobs:
       - if: matrix.shard == 3
         run: npm run --prefix tools/vendor test
       - name: Page workflows with coverage, shard ${{ matrix.shard }}/3
+        timeout-minutes: 18
         run: npm run --prefix tools/vendor test:coverage:shard -- ${{ matrix.shard }}/3
       - name: Upload raw coverage
         uses: actions/upload-artifact@v7
@@ -87,6 +97,7 @@ jobs:
           path: coverage/js/raw/
           if-no-files-found: error
           retention-days: 3
+      # Traces and the blob report (test-results/blob/) of a failed, timed-out or cancelled step.
       - name: Upload Chromium and touch traces
         if: always()
         uses: actions/upload-artifact@v7
@@ -211,7 +222,9 @@ jobs:
           key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
       - run: npm ci --prefix tools/vendor --ignore-scripts
       - name: Page workflows in ${{ matrix.browser }}, shard ${{ matrix.shard }}/2
+        timeout-minutes: 18
         run: npm run --prefix tools/vendor test:workflows -- --project ${{ matrix.browser }} --shard ${{ matrix.shard }}/2
+      # Traces and the blob report (test-results/blob/) of a failed, timed-out or cancelled step.
       - name: Upload browser traces
         if: always()
         uses: actions/upload-artifact@v7

README.md

@@ -312,6 +312,12 @@ Both nodes support the same metadata fields:
 
 Both nodes will automatically save prompts to the database and link any generated images to the prompt.
 
+- **Save to library** (`save_to_library`, on by default; "Save to library" in the Focused, Split and Pair editors):
+  turn it off while iterating and PromptManager ignores that node's runs. The prompt is not saved, the run is not
+  counted and its images are not linked (a scan skips them too), while the node's output is unchanged. Workflows
+  saved before this switch existed load with it on. If any PromptManager node feeding a sampler's positive input
+  has it off, none of that run's images are linked, because the image cannot show which sampler produced it.
+
 ### 🖼️ Using the Output-Wide Gallery
 
 The comprehensive gallery system provides complete access to all your ComfyUI output:

TESTING.md

@@ -262,6 +262,31 @@ watchdog dump, is a failure like any other: read the dump, find the blocking
 call, fix it. Never raise a limit or rerun to get green; a limit changes only
 with new measurements written next to it.
 
+Browser jobs (#425): GitHub keeps no log of a job killed at its limit and skips
+its upload steps, so the test step in each browser job in `viewerjs-check.yml`
+has its own `timeout-minutes` (18, below the job's 20). At the step limit
+GitHub sends SIGTERM, which would end Playwright without any report; under `CI`
+`tools/vendor/running-tests-reporter.mjs` catches the first one, prints
+`Still running when the run ended (SIGTERM): N` and one line per unfinished
+test (path relative to the suite's `rootDir`), then sends Playwright its own
+SIGINT, so the workers stop, the run ends interrupted and the blob report is
+written to `tools/vendor/test-results/blob/`. In the coverage job npm signals
+only `frontend-coverage.mjs`, which forwards SIGTERM and SIGINT to the child it
+is running (started in its own process group, so a group-wide signal still
+arrives once). Delivery is proven locally in the pinned image for a SIGTERM to
+the whole process group (`timeout -s TERM`). A SIGTERM to npm alone
+(`timeout --foreground`) reaches nothing: npm forwards it to the `sh -c` that
+runs the script, and that shell exits without passing it on. Delivery in CI,
+where the step runs through `docker exec`, is not yet observed; the next real
+step timeout will show it. That directory and the traces
+upload with `if: always()`; `npx playwright merge-reports` opens the blob.
+Each test keeps the 30 s default and there is no `globalTimeout`. Two limits
+of this: it assumes the npm cache hits (a cold cache adds up to 5 minutes to
+`npm ci`, and the job limit can then still win), and coverage shard 1 (12m53s
+measured) is the tight one in its 18-minute step. Local runs keep the `list`
+reporter only; reproduce with `CI=1 timeout -s TERM 40 npm run --prefix
+tools/vendor test:workflows -- --project=firefox`.
+
 ### After a push
 
 Follow the run for the commit you pushed, never an older one:
@@ -5249,6 +5274,75 @@ same as a file the scan could not attribute, so the file stays on disk and in th
 gallery as unlinked, and its favorite and tags (kept by file) are untouched. Every
 run prints this. Queue-linked batch images can't be proven after the fact and read
 `unlink`, so review before applying. Stop ComfyUI first.
+
+### Foreign workflows (#247): the shared per-file decision
+
+The output scan decides each file through `utils/file_prompt.py`
+(`decide_file_prompt`, the ladder), which the monitor will share for outputs of
+workflows without a PromptManager node. It does no file or database I/O.
+
+- **Save to library (#90).** Before anything else, a file whose PromptManager
+  node had Save to library off is `not_saved`, and one whose switch is wired
+  from a node the file cannot show is `not_saved_wired`: nothing is written, no
+  recorded-run fallback, no Needs review item. A resolved pair whose negative's
+  node is off stores the positive alone (source `opted-out-negative`). The
+  #90 graph walk follows only prompt-like inputs, so the ladder also walks
+  every input upstream of each sampler or guider positive (not negative, model,
+  clip, latent or image inputs; every PromptManager node of a graph without a
+  sampler) and checks the text later steps name: a resolved or inspected
+  positive whose path passes such a node (`side_opted_out`), or a legacy answer
+  that is such a node's text (resolved, else literal). A switch that is off
+  makes the file `not_saved`; one wired from a node the graph cannot read makes
+  it `not_saved_wired`, as #90's direct check does.
+- **Order.** The default profile decides first, exactly as before (resolved
+  pair, A1111, sampler-less). Only a graph it leaves ambiguous or unresolved
+  reaches the inspector's resolve (`inspection_resolution` in
+  `utils/prompt_inspector.py`, the inspection profile, then the recorded saver
+  anchor of #386; `name_anchor=False`: the file-name anchor never decides a
+  link; the scan's legacy floor may still link the file, as before #247, and a
+  file nothing links is queued with the anchor's choice per #390), and only that positive's `storage_text` is stored (never
+  prepend or append text, never `recorded_texts`). Then the #170 browser parser
+  (`graph_prompt_text`), the legacy floor, then the recorded run's stored
+  prompt. The legacy answer is refused when a PromptManager-family node has a
+  prepend or append part and the answer is not that node's own `text` (the
+  parser joins the parts, so the stored text cannot be told from it). History is asked once per file and only at that step; the scan's
+  batch lookup stays lazy and makes one `recorded_runs` pass for the stored
+  prompt and the saver (`_BatchHistory`).
+- **Policies.** `SCAN` keeps the legacy floor for every undecided file.
+  `MONITOR` waits for history inside the output folder (`history`) and stores
+  nothing for a file the inspector calls ambiguous; an untraceable or
+  malformed one still gets the legacy floor.
+- **Junk.** `junk_text` refuses `blank`, `number` (the #252 `_numeric` rule),
+  `lora-only` (only `<lora:...>` tags and separators) and `model-file` (one
+  token ending in a model extension) at whichever step found the text and
+  nothing later is tried; the file reports `found`, except at the undecided
+  graph step, which keeps its own status (an ambiguous file is still queued).
+- **One prompt per file.** `database/scanned_prompts.py` stores a positive in
+  one `BEGIN IMMEDIATE` transaction: a file already linked to another prompt
+  (either spelling, `file_link_rows`/`link_conflict`, the pair ingester's rule)
+  is `conflict` and nothing is written; a failed link rolls the new prompt back
+  (`found`); two writers of one text store one prompt. A recorded run's stored
+  prompt links the same way (`link_stored_prompt_or_raise`). A rescan therefore
+  never adds a second link to a file an older scan linked to the legacy answer;
+  the row reads "Kept: already linked to a different prompt". Link rows are
+  matched under the path and its `realpath` spelling (`file_link_rows`), so a
+  file the scan stored through the resolved folder is found when it arrives
+  through a symlinked or junction folder. The reverse order (a row stored under
+  the watcher's raw spelling, then a scan whose root resolves differently) is
+  matched only when that raw spelling equals the realpath. Every scan write
+  runs through `survive_library_switch`, so a library move reruns it.
+- **Marks.** Every prompt ingestion creates, pair components included, carries
+  category `scanned`, tag `auto-scanned` and the note `Auto-scanned from
+  <file>` (`SCAN_MARKS`); a reused prompt keeps its fields. Run counts are
+  never touched.
+
+Tests: `tests/test_file_prompt.py` (one test per ladder step and junk class),
+`tests/test_scanned_prompts.py` (store, conflict, rollback, concurrency with
+`FirstWriterHold`, pair marks), `tests/test_pair_scan.py` `TestScanDecision`,
+`tests/test_run_attribution.py` (dataset branches linked by the recorded
+saver, a legacy-linked file left as it is, `_BatchHistory.recorded`) and
+`tests/test_recorded_saver_anchor.py` `TestInspectionResolution`.
+
 ## Linked prompt text (#343)
 
 A link drives `text` on every PromptManager node, and `negative_text` on the Pair,

database/pair_ingestion.py

@@ -1,7 +1,14 @@
-"""Atomic metadata enrichment for a single known generation context."""
+"""Atomic metadata enrichment for a single known generation context.
+
+Also the helpers both ingestion store paths share (#247): the file's link
+rows and the one-prompt-per-file rule (``file_link_rows``,
+``link_conflict``) and the marks a prompt created by ingestion carries
+(``Marks``, ``mark_created``).
+"""
 
 import os
-from typing import Any, Dict
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
 
 from .models import normalize_image_path
 from .operations import _coerce_file_info
@@ -37,16 +44,85 @@ def blank_positive(resolution: Dict[str, Any]) -> bool:
     return not (isinstance(text, str) and text.strip())
 
 
+@dataclass(frozen=True)
+class Marks:
+    """What a prompt ingestion creates carries, so it can be found and removed.
+
+    ``note`` is formatted with ``name``, the file's name.
+    """
+
+    category: str
+    tags: Tuple[str, ...]
+    note: str
+
+
+SCAN_MARKS = Marks("scanned", ("auto-scanned",), "Auto-scanned from {name}")
+RUN_MARKS = Marks("scanned", ("auto-scanned",), "Saved from a workflow run: {name}")
+
+
+def mark_created(conn: Any, db: Any, prompt_id: int, marks: Marks, name: str) -> None:
+    """Apply ``marks`` to a prompt row the caller just created on ``conn``."""
+    conn.execute(
+        "UPDATE prompts SET category = ?, notes = ? WHERE id = ?",
+        (marks.category, marks.note.format(name=name), prompt_id),
+    )
+    db._sync_prompt_tags(conn, prompt_id, list(marks.tags))
+
+
+def file_link_rows(conn: Any, image_path: str) -> List[Any]:
+    """The file's link rows, each with its prompt's hash.
+
+    Matched under the path and its resolved spelling (an output folder
+    reached through a symlink or junction), each normalized and raw;
+    deduplicated by row id.
+    """
+    rows: Dict[Any, Any] = {}
+    for spelling in dict.fromkeys((image_path, _resolved(image_path))):
+        for row in conn.execute(
+            "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi "
+            "JOIN prompts p ON p.id=gi.prompt_id "
+            "WHERE gi.file_path=? OR gi.image_path=?",
+            (normalize_image_path(spelling), spelling),
+        ).fetchall():
+            rows.setdefault(row["id"], row)
+    return list(rows.values())
+
+
+def _resolved(path: str) -> str:
+    try:
+        return os.path.realpath(path)
+    except (OSError, ValueError):
+        return path
+
+
+def link_conflict(rows: List[Any], prompt_hash: str) -> bool:
+    """One prompt per file: more than one row, or a row of another prompt."""
+    return len(rows) > 1 or bool(rows and rows[0]["positive_hash"] != prompt_hash)
+
+
 def ingest_resolved_pair(
-    db: Any, image_path: str, metadata: Dict[str, Any], resolution: Dict[str, Any]
+    db: Any,
+    image_path: str,
+    metadata: Dict[str, Any],
+    resolution: Dict[str, Any],
+    *,
+    marks: Optional[Marks] = None,
 ) -> Dict[str, Any]:
+    """Store a resolved pair and link the file to it in one transaction.
+
+    With ``marks`` (ingestion from files), each pair component this call
+    creates carries them; without (the PromptManager node path) components
+    are created bare, as always.
+    """
     state = resolution.get("state", "unresolved")
     if state == "resolved" and blank_positive(resolution):
         state = "unresolved"
     if state != "resolved":
         return {"status": state, "pairs_added": 0, "links_added": 0}
     with db._pair_transaction() as conn:
-        return _ingest_resolved_pair_in(conn, db, image_path, metadata, resolution)
+        return _ingest_resolved_pair_in(
+            conn, db, image_path, metadata, resolution, marks=marks
+        )
 
 
 def _conflict() -> Dict[str, Any]:
@@ -61,6 +137,7 @@ def _ingest_resolved_pair_in(
     resolution: Dict[str, Any],
     *,
     provenance: Dict[str, Any] | None = None,
+    marks: Optional[Marks] = None,
 ) -> Dict[str, Any]:
     """``ingest_resolved_pair`` inside the caller's open ``_pair_transaction``.
 
@@ -69,7 +146,8 @@ def _ingest_resolved_pair_in(
     negative is a ``conflict`` and nothing is written. The caller owns the
     transaction, so its own later failure rolls the link back too.
     ``provenance`` replaces the link's default resolver provenance (Needs
-    review records the user's selection there).
+    review records the user's selection there). ``marks`` as in
+    ``ingest_resolved_pair``.
     """
     positive, negative = resolution["positive"], resolution["negative"]
     positive_text = positive.get("storage_text", positive["text"])
@@ -82,15 +160,8 @@ def _ingest_resolved_pair_in(
             "negative": negative,
         }
     encoded = dumps_strict(provenance, sort_keys=True, separators=(",", ":"))
-    images = conn.execute(
-        "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi "
-        "JOIN prompts p ON p.id=gi.prompt_id "
-        "WHERE gi.file_path=? OR gi.image_path=?",
-        (normalized, image_path),
-    ).fetchall()
-    if len(images) > 1 or (
-        images and images[0]["positive_hash"] != generate_prompt_hash(positive_text)
-    ):
+    images = file_link_rows(conn, image_path)
+    if link_conflict(images, generate_prompt_hash(positive_text)):
         return _conflict()
     existing = images[0] if images else None
     if existing:
@@ -115,7 +186,12 @@ def _ingest_resolved_pair_in(
             generate_prompt_hash(negative_text) if negative_text.strip() else None,
         ),
     ).fetchone()
-    pair_id = db._save_pair(conn, positive_text, negative_text)
+    if marks is None:
+        pair_id = db._save_pair(conn, positive_text, negative_text)
+    else:
+        pair_id = _save_marked_pair(
+            conn, db, positive_text, negative_text, marks, os.path.basename(image_path)
+        )
     pair = conn.execute("SELECT * FROM prompt_pairs WHERE id=?", (pair_id,)).fetchone()
     pairs_added = int(prior_pair is None)
     file_size, width, height, image_format = _coerce_file_info(metadata)
@@ -197,6 +273,26 @@ def _ingest_resolved_pair_in(
     }
 
 
+def _save_marked_pair(
+    conn: Any,
+    db: Any,
+    positive_text: str,
+    negative_text: str,
+    marks: Marks,
+    name: str,
+) -> int:
+    """``_save_pair`` that marks each component it creates."""
+    texts = [positive_text] + ([negative_text] if negative_text.strip() else [])
+    ids = []
+    for text in texts:
+        prompt_id, created = db._pair_component_outcome(conn, text)
+        if created:
+            mark_created(conn, db, prompt_id, marks, name)
+        ids.append(prompt_id)
+    negative_id = ids[1] if len(ids) > 1 else None
+    return db._save_pair_components(conn, ids[0], negative_id)[0]
+
+
 def _document_form(conn: Any, values: Dict[str, Any]) -> Dict[str, Any]:
     """``values`` with the workflow and prompt graph texts stored once.
 

database/run_attribution.py

@@ -164,7 +164,7 @@ def _runs_by_location(db: Any, names: List[str]) -> Optional[Dict[tuple, dict]]:
     return found
 
 
-def _recorded_runs(
+def recorded_runs(
     db: Any,
     image_paths: Iterable[str],
     output_root: Optional[str],
@@ -201,7 +201,8 @@ def _recorded_runs(
     return attributed
 
 
-def _named_prompt(db: Any, run_id: str) -> Optional[int]:
+def named_prompt(db: Any, run_id: str) -> Optional[int]:
+    """The one stored prompt every positive context of ``run_id`` names, else None."""
     contexts = db.get_execution_contexts(run_id).get("contexts") or []
     named = {
         context.get("positive_prompt_id")
@@ -228,11 +229,11 @@ def stored_run_prompts(
     paths = list(dict.fromkeys(path for path in image_paths if path))
     result: Dict[str, Optional[int]] = dict.fromkeys(paths)
     prompts: Dict[str, Optional[int]] = {}
-    for path, (run_id, _nodes) in _recorded_runs(
+    for path, (run_id, _nodes) in recorded_runs(
         db, paths, output_root, path_style
     ).items():
         if run_id not in prompts:
-            prompts[run_id] = _named_prompt(db, run_id)
+            prompts[run_id] = named_prompt(db, run_id)
         result[path] = prompts[run_id]
     return result
 
@@ -275,7 +276,7 @@ def recorded_output(
 ) -> Optional[RecordedOutput]:
     """The recorded run facts for ``image_path``, else None (#386).
 
-    The file must be attributed to one run (``_recorded_runs``: one run
+    The file must be attributed to one run (``recorded_runs``: one run
     reported its location and it was written during that run). Read-only; it
     never links anything.
     """
@@ -286,7 +287,7 @@ def recorded_output(
     if own_transaction:  # outputs and texts from one state of history
         conn.execute("BEGIN")
     try:
-        found = _recorded_runs(db, [image_path], output_root, path_style)
+        found = recorded_runs(db, [image_path], output_root, path_style)
         if image_path not in found:
             return None
         run_id, nodes = found[image_path]

database/scanned_prompts.py

@@ -0,0 +1,87 @@
+"""Link an output file to the prompt of one text, creating it when new (#247).
+
+The store step of ingestion from file metadata when the decision is a
+positive text (``utils.file_prompt``). The prompt insert and the link run in
+one ``BEGIN IMMEDIATE`` transaction: two writers never store one text twice,
+a failed link leaves no new prompt behind, and a file stays linked to one
+prompt (a file already linked to another prompt is a ``conflict``).
+"""
+
+import logging
+import os
+from typing import Any, Dict, Optional, Tuple
+
+from .pair_ingestion import Marks, file_link_rows, link_conflict, mark_created
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+_LOGGER = logging.getLogger("prompt_manager.database.scanned_prompts")
+
+
+def store_scanned_prompt_or_raise(
+    db: Any, path: str, canonical: Dict[str, Any], text: str, marks: Marks
+) -> Tuple[str, Optional[int]]:
+    """``(status, prompt id)`` after linking ``path`` to ``text``'s prompt.
+
+    ``added`` (prompt created, carrying ``marks``), ``linked`` (stored text
+    reused), ``unchanged`` (already linked to it) or ``conflict`` (linked to
+    another prompt; nothing written, id None). Database errors are raised
+    and roll the whole store back, so a caller can run it again.
+    """
+    with db._pair_transaction() as conn:
+        rows = file_link_rows(conn, path)
+        if link_conflict(rows, generate_prompt_hash(text)):
+            return "conflict", None
+        if rows:
+            return "unchanged", rows[0]["prompt_id"]
+        prompt_id, created = db._pair_component_outcome(conn, text)
+        if created:
+            mark_created(conn, db, prompt_id, marks, os.path.basename(path))
+        db._insert_image_link(conn, prompt_id, path, canonical)
+        return ("added" if created else "linked"), prompt_id
+
+
+def store_scanned_prompt(
+    db: Any, path: str, canonical: Dict[str, Any], text: str, marks: Marks
+) -> Tuple[str, Optional[int]]:
+    """The scan's form: an error is logged and reported as ``("found", None)``.
+
+    A write that failed because Move library switched runs again on the new
+    library (``survive_library_switch``); the rerun re-checks the link rows.
+    """
+    try:
+        return db.survive_library_switch(
+            lambda: store_scanned_prompt_or_raise(db, path, canonical, text, marks)
+        )
+    except Exception as error:
+        _LOGGER.error(
+            f"Failed to store the prompt of {os.path.basename(path)}: {error}"
+        )
+        return "found", None
+
+
+def link_stored_prompt_or_raise(
+    db: Any, path: str, canonical: Dict[str, Any], prompt_id: int
+) -> Optional[str]:
+    """Link ``path`` to the stored prompt ``prompt_id`` (recorded history, #342).
+
+    ``linked``, ``unchanged`` (already linked to it), ``conflict`` (linked to
+    another prompt; nothing written) or None when the prompt is gone. One
+    ``BEGIN IMMEDIATE`` transaction; database errors are raised.
+    """
+    with db._pair_transaction() as conn:
+        rows = file_link_rows(conn, path)
+        if len(rows) > 1 or (rows and rows[0]["prompt_id"] != prompt_id):
+            return "conflict"
+        if rows:
+            return "unchanged"
+        exists = conn.execute(
+            "SELECT 1 FROM prompts WHERE id = ?", (prompt_id,)
+        ).fetchone()
+        if exists is None:
+            return None
+        db._insert_image_link(conn, prompt_id, path, canonical)
+        return "linked"

prompt_manager.py

@@ -26,13 +26,13 @@ except ImportError:
     InputTypeDict = dict
 
 try:
-    from .prompt_manager_base import PromptManagerBase
+    from .prompt_manager_base import SAVE_TO_LIBRARY_INPUT, PromptManagerBase
 except ImportError:
     import os
     import sys
 
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-    from prompt_manager_base import PromptManagerBase
+    from prompt_manager_base import SAVE_TO_LIBRARY_INPUT, PromptManagerBase
 
 
 class PromptManager(PromptManagerBase, ComfyNodeABC):
@@ -115,6 +115,7 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
                         )
                     },
                 ),
+                "save_to_library": SAVE_TO_LIBRARY_INPUT,
             },
             "hidden": {
                 # The queued graph and this node's id: used to tell the positive
@@ -152,6 +153,7 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
         append_text: str = "",
         prompt: Optional[Dict[str, Any]] = None,
         unique_id: Optional[str] = None,
+        save_to_library: bool = True,
     ) -> Tuple[Any]:
         """
         Encode the text prompt and save it to the database.
@@ -164,6 +166,7 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
             search_text: Text to search for in past prompts
             prepend_text: Text to prepend to the main prompt
             append_text: Text to append to the main prompt
+            save_to_library: Off: nothing is saved, counted or tracked (#90)
 
         Returns:
             Tuple containing the conditioning for the diffusion model and the final
@@ -205,7 +208,9 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
         # Save prompt to database and set execution context for gallery tracking
         prompt_id = None
         extended_tags = []
-        if storage_text and storage_text.strip():
+        if not save_to_library:
+            self._skip_not_saved(prompt, unique_id)
+        elif storage_text and storage_text.strip():
             self.logger.debug(f"Processing prompt text: {storage_text[:100]}...")
 
             # Add prepend/append info to tags if they exist
@@ -249,6 +254,11 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
         tokens = clip.tokenize(encoding_text)
         conditioning = clip.encode_from_tokens_scheduled(tokens)
 
+        self.logger.info(f"CLIP encoding completed, text: {repr(encoding_text)[:80]}")
+        result = (conditioning, encoding_text)
+        if not save_to_library:
+            return result  # #90: no metadata registration, no history text
+
         # Register with ComfyUI integration for standard metadata compatibility
         node_id = f"promptmanager_{int(time.time() * 1000)}"
         self.comfyui_integration.register_prompt(
@@ -262,9 +272,6 @@ class PromptManager(PromptManagerBase, ComfyNodeABC):
                 "append_text": append_text.strip() if append_text else None,
             },
         )
-
-        self.logger.info(f"CLIP encoding completed, text: {repr(encoding_text)[:80]}")
-        result = (conditioning, encoding_text)
         record_runtime_context(
             type(self).__name__, text, prepend_text, append_text, encoding_text
         )

prompt_manager_base.py

@@ -23,7 +23,7 @@ try:
     from .utils.image_monitor import get_image_monitor
     from .utils.prompt_tracker import get_prompt_tracker
     from .utils.hashing import generate_prompt_hash
-    from .utils.prompt_graph import resolve_text, run_prompt_nodes
+    from .utils.prompt_graph import resolve_text, run_prompt_nodes, save_setting
     from .utils.run_keys import executing_run_id, prompt_node_key
     from .utils.usage_tracking import PENDING_FIRST_USE
     from .utils.cached_runs import (
@@ -42,7 +42,7 @@ except ImportError:
     from utils.image_monitor import get_image_monitor
     from utils.prompt_tracker import get_prompt_tracker
     from utils.hashing import generate_prompt_hash
-    from utils.prompt_graph import resolve_text, run_prompt_nodes
+    from utils.prompt_graph import resolve_text, run_prompt_nodes, save_setting
     from utils.run_keys import executing_run_id, prompt_node_key
     from utils.usage_tracking import PENDING_FIRST_USE
     from utils.cached_runs import (
@@ -54,6 +54,22 @@ except ImportError:
     )
 
 
+# The optional input every PromptManager node declares last (#90): declared
+# last so no widget value of a workflow saved before it shifts.
+SAVE_TO_LIBRARY_INPUT = (
+    "BOOLEAN",
+    {
+        "default": True,
+        "label_on": "save",
+        "label_off": "skip",
+        "tooltip": (
+            "Save this node's prompt to the library, count its runs and link its "
+            "images. Off: PromptManager ignores the run; the output is unchanged."
+        ),
+    },
+)
+
+
 class PromptManagerBase:
     """Mixin providing shared prompt management logic for ComfyUI nodes.
 
@@ -156,7 +172,9 @@ class PromptManagerBase:
         resolved = resolve_text(prompt_graph, node_id)
         return {
             "is_run_prompt": node_id in run_prompt_nodes(prompt_graph),
+            # The hook skips a switch it cannot read (#90): the node counts
             "seen_at_queue": resolved is not None
+            and save_setting(prompt_graph, node_id) is True
             and generate_prompt_hash(resolved) == generate_prompt_hash(text),
             "known": True,
         }
@@ -245,6 +263,24 @@ class PromptManagerBase:
             "run_id": executing_run_id(),
         }
 
+    def _skip_not_saved(self, prompt_graph: Any = None, unique_id: Any = None) -> None:
+        """``save_to_library`` is off: nothing is saved, counted or tracked (#90).
+
+        A literal off makes the node no run prompt (``utils.prompt_graph``), so
+        no image looks up its run key. A switch wired from a node the graph
+        cannot read keeps the node a run prompt: its run key is marked like an
+        empty editor's, so an earlier run's entry of that key never links this
+        run's images.
+        """
+        message = (
+            "Save to library is off for this node: its prompt is not saved, the "
+            "run is not counted and its images are not linked"
+        )
+        if isinstance(prompt_graph, dict) and unique_id is not None:
+            self._mark_untracked(prompt_graph, unique_id, message)
+        else:
+            self.logger.info(message)
+
     def _skip_empty_prompt(self, prompt_graph: Any, unique_id: Any) -> None:
         """The ``text`` editor is empty: save and track nothing (#342).
 

prompt_manager_pair.py

@@ -22,7 +22,11 @@ try:
     )
     from .utils.hashing import generate_prompt_hash
     from .utils.pair_execution import record_pair_execution
-    from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
+    from .utils.prompt_graph import (
+        intrinsic_pair_nodes,
+        run_prompt_components,
+        save_setting,
+    )
     from .utils.prompt_pair_resolver import resolve_text_input
     from .utils.usage_tracking import (
         PENDING_FIRST_USE,
@@ -41,7 +45,11 @@ except ImportError:
     )
     from utils.hashing import generate_prompt_hash
     from utils.pair_execution import record_pair_execution
-    from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
+    from utils.prompt_graph import (
+        intrinsic_pair_nodes,
+        run_prompt_components,
+        save_setting,
+    )
     from utils.prompt_pair_resolver import resolve_text_input
     from utils.usage_tracking import (
         PENDING_FIRST_USE,
@@ -90,8 +98,13 @@ class PromptManagerPair(PromptManagerFocused):
         prompt: Optional[Dict[str, Any]] = None,
         unique_id: Optional[str] = None,
         extra_pnginfo: Optional[Dict[str, Any]] = None,
+        save_to_library: bool = True,
     ) -> Tuple[Any, Any, str, str]:
-        """Encode both sides once; database and tracking failures never block CLIP."""
+        """Encode both sides once; database and tracking failures never block CLIP.
+
+        With ``save_to_library`` off nothing is saved, counted, tracked or
+        recorded (#90); the four outputs are the same.
+        """
         if clip is None:
             raise RuntimeError("ERROR: clip input is invalid: None")
         if not isinstance(text, str) or not isinstance(negative_text, str):
@@ -109,6 +122,10 @@ class PromptManagerPair(PromptManagerFocused):
         negative_conditioning = clip.encode_from_tokens_scheduled(
             clip.tokenize(negative_text)
         )
+        result = positive_conditioning, negative_conditioning, positive, negative_text
+        if not save_to_library:
+            self._skip_not_saved(prompt, unique_id)
+            return result
         record_pair_execution(
             extra_pnginfo,
             unique_id,
@@ -169,7 +186,6 @@ class PromptManagerPair(PromptManagerFocused):
             )
         except Exception as error:
             self.logger.warning(f"Could not register paired metadata: {error}")
-        result = positive_conditioning, negative_conditioning, positive, negative_text
         record_runtime_context(
             type(self).__name__,
             text,
@@ -219,6 +235,9 @@ class PromptManagerPair(PromptManagerFocused):
         resolved_negative = (
             resolve_text_input(prompt, node_id, "negative_text") if known else None
         )
+        if known and save_setting(prompt, node_id) is not True:
+            # The hook skips a switch it cannot read (#90): the node counts
+            resolved_positive = resolved_negative = None
         positive_seen = resolved_positive is not None and generate_prompt_hash(
             resolved_positive
         ) == generate_prompt_hash(text)

prompt_manager_text.py

@@ -26,13 +26,13 @@ except ImportError:
     InputTypeDict = dict
 
 try:
-    from .prompt_manager_base import PromptManagerBase
+    from .prompt_manager_base import SAVE_TO_LIBRARY_INPUT, PromptManagerBase
 except ImportError:
     import os
     import sys
 
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-    from prompt_manager_base import PromptManagerBase
+    from prompt_manager_base import SAVE_TO_LIBRARY_INPUT, PromptManagerBase
 
 
 class PromptManagerText(PromptManagerBase, ComfyNodeABC):
@@ -115,6 +115,7 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
                         ),
                     },
                 ),
+                "save_to_library": SAVE_TO_LIBRARY_INPUT,
             },
             "hidden": {
                 # The queued graph and this node's id: used to tell the positive
@@ -149,6 +150,7 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
         append_text: str = "",
         prompt: Optional[Dict[str, Any]] = None,
         unique_id: Optional[str] = None,
+        save_to_library: bool = True,
     ) -> Tuple[str]:
         """
         Process the text prompt and save it to the database.
@@ -160,6 +162,7 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
             search_text: Text to search for in past prompts
             prepend_text: Text to prepend to the main prompt
             append_text: Text to append to the main prompt
+            save_to_library: Off: nothing is saved, counted or tracked (#90)
 
         Returns:
             Tuple containing the final processed text string
@@ -184,7 +187,9 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
         # Save prompt to database and set execution context for gallery tracking
         prompt_id = None
         extended_tags = []
-        if storage_text and storage_text.strip():
+        if not save_to_library:
+            self._skip_not_saved(prompt, unique_id)
+        elif storage_text and storage_text.strip():
             self.logger.debug(f"Processing prompt text: {storage_text[:100]}...")
 
             # Add prepend/append info to tags if they exist
@@ -220,6 +225,11 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
         else:
             self._skip_empty_prompt(prompt, unique_id)
 
+        self.logger.debug(f"Text processing completed: {final_text[:100]}...")
+        result = (final_text,)
+        if not save_to_library:
+            return result  # #90: no metadata registration, no history text
+
         # Register with ComfyUI integration for standard metadata compatibility
         node_id = f"promptmanagertext_{int(time.time() * 1000)}"
         self.comfyui_integration.register_prompt(
@@ -233,9 +243,6 @@ class PromptManagerText(PromptManagerBase, ComfyNodeABC):
                 "append_text": append_text.strip() if append_text else None,
             },
         )
-
-        self.logger.debug(f"Text processing completed: {final_text[:100]}...")
-        result = (final_text,)
         record_runtime_context(
             type(self).__name__, text, prepend_text, append_text, final_text
         )

py/api/__init__.py

@@ -76,6 +76,7 @@ from .image_categories import ImageCategoryRoutesMixin
 
 try:
     from ...database.operations import PromptDatabase
+    from ...utils.file_prompt import parse_comfyui_prompt
     from ...utils.logging_config import get_logger
     from ...utils.video_metadata import is_video_path, read_video_metadata
 except ImportError:
@@ -85,6 +86,7 @@ except ImportError:
         0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     )
     from database.operations import PromptDatabase
+    from utils.file_prompt import parse_comfyui_prompt
     from utils.logging_config import get_logger
     from utils.video_metadata import is_video_path, read_video_metadata
 
@@ -1028,85 +1030,4 @@ class PromptManagerAPI(
 
     def _parse_comfyui_prompt(self, metadata):
         """Parse ComfyUI and A1111 prompt data from metadata."""
-        result = {
-            "prompt": None,
-            "workflow": None,
-            "parameters": {},
-            "positive_prompt": None,
-            "negative_prompt": None,
-        }
-
-        # Check for A1111/Forge style parameters first.
-        # The chunk is one block: a possibly multi-line positive prompt, then a
-        # "Negative prompt:" section, then a trailing "Steps: ..." settings line.
-        # Taking lines[0] as the positive prompt truncates any multi-line prompt
-        # to one line, so split on the settings line instead.
-        if "parameters" in metadata:
-            import re as _re
-
-            params = metadata["parameters"]
-            body = params
-            settings = ""
-            _sm = _re.search(r"\nSteps:\s", body)
-            if _sm:
-                settings = body[_sm.start() + 1 :].strip()
-                body = body[: _sm.start()]
-            negative = None
-            _nm = _re.search(r"(?im)^\s*negative prompt:\s*", body)
-            if _nm:
-                negative = body[_nm.end() :].strip()
-                body = body[: _nm.start()]
-            elif body.lower().lstrip().startswith("negative prompt:"):
-                negative = body.lower().lstrip()[len("negative prompt:") :].strip()
-                body = ""
-            result["positive_prompt"] = body.strip() or None
-            if negative:
-                result["negative_prompt"] = negative
-            # Store raw parameters too
-            result["parameters"]["parameters"] = params
-            if settings:
-                for _f in ("Steps", "Sampler", "Schedule type", "CFG scale", "Seed"):
-                    _fm = _re.search(
-                        r"(?i)(?:^|,\s*)" + _re.escape(_f) + r":\s*([^,]+)", settings
-                    )
-                    if _fm:
-                        result["parameters"][_f.lower().replace(" ", "_")] = _fm.group(
-                            1
-                        ).strip()
-
-        # If no A1111 format found, proceed with ComfyUI parsing
-        if result["positive_prompt"] is None:
-            # Check for direct prompt field
-            if "prompt" in metadata:
-                try:
-                    prompt_data = json.loads(metadata["prompt"])
-                    result["prompt"] = prompt_data
-                except json.JSONDecodeError:
-                    result["prompt"] = metadata["prompt"]
-
-            # Check for workflow
-            if "workflow" in metadata:
-                try:
-                    workflow_data = json.loads(metadata["workflow"])
-                    result["workflow"] = workflow_data
-                except json.JSONDecodeError:
-                    result["workflow"] = metadata["workflow"]
-
-            # Check for other common ComfyUI fields
-            common_fields = [
-                "positive",
-                "negative",
-                "steps",
-                "cfg",
-                "sampler",
-                "scheduler",
-                "seed",
-            ]
-            for field in common_fields:
-                if field in metadata:
-                    try:
-                        result["parameters"][field] = json.loads(metadata[field])
-                    except json.JSONDecodeError:
-                        result["parameters"][field] = metadata[field]
-
-        return result
+        return parse_comfyui_prompt(metadata)

py/api/admin.py

@@ -40,43 +40,51 @@ except ImportError:
     from utils import exit_signal
 
 try:
-    from ...database.pair_ingestion import blank_positive, ingest_resolved_pair
+    from ...database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from ...database.run_attribution import (
+        RecordedOutput,
         host_output_directory,
-        stored_run_prompt,
-        stored_run_prompts,
+        named_prompt,
+        recorded_runs,
     )
-    from ...utils.generation_params import (
-        PLACEHOLDERS,
-        GenerationParamsError,
-        extract_generation_params,
-        parse_page_json,
+    from ...database.scanned_prompts import (
+        link_stored_prompt_or_raise,
+        store_scanned_prompt,
+    )
+    from ...utils.file_prompt import (
+        NEVER,
+        NOT_SAVED,
+        NOT_SAVED_WIRED,
+        PAIR,
+        Policy,
+        decide_file_prompt,
     )
-    from ...utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt
-    from ...utils.review_capture import reviewable
 except ImportError:
-    from database.pair_ingestion import blank_positive, ingest_resolved_pair
+    from database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from database.run_attribution import (
+        RecordedOutput,
         host_output_directory,
-        stored_run_prompt,
-        stored_run_prompts,
+        named_prompt,
+        recorded_runs,
     )
-    from utils.generation_params import (
-        PLACEHOLDERS,
-        GenerationParamsError,
-        extract_generation_params,
-        parse_page_json,
+    from database.scanned_prompts import (
+        link_stored_prompt_or_raise,
+        store_scanned_prompt,
+    )
+    from utils.file_prompt import (
+        NEVER,
+        NOT_SAVED,
+        NOT_SAVED_WIRED,
+        PAIR,
+        Policy,
+        decide_file_prompt,
     )
-    from utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt
-    from utils.review_capture import reviewable
 
 
 try:
     from ...database.library import display_title
-    from ...utils.hashing import generate_prompt_hash
 except ImportError:
     from database.library import display_title
-    from utils.hashing import generate_prompt_hash
 
 IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff")
 VIDEO_EXTENSIONS = (".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".wmv")
@@ -256,10 +264,15 @@ SCAN_DETAILS = {
     "skipped": "Skipped · already linked",
     "ambiguous": "Needs review: more than one prompt could match",
     "unresolved": "Needs review: the prompt could not be traced in the workflow",
-    "conflict": "Needs review: already linked to a different prompt",
+    "conflict": "Kept: already linked to a different prompt",
     "failed": "Error: the prompt could not be stored",
     "error": "Error: the file could not be processed (see the server log)",
     "found": "Metadata found, but no prompt text to store",
+    "not_saved": "Skipped · its PromptManager node does not save to the library",
+    "not_saved_wired": (
+        "Skipped · its Save to library switch is wired from another node, so only "
+        "the run itself can link it"
+    ),
     None: "No prompt metadata in the file",
 }
 
@@ -318,22 +331,6 @@ def _file_size(path):
         return None
 
 
-def _with_file_size(canonical, media_file):
-    """Record the file's size in ``file_info`` unless the metadata carries one.
-
-    Stored as ``generated_images.file_size``; the "new" scan scope skips a known
-    file only while its size is unchanged.
-    """
-    info = canonical.get("file_info")
-    info = dict(info) if isinstance(info, dict) else {}
-    if info.get("size") is None:
-        size = _file_size(media_file)
-        if size is not None:
-            info["size"] = size
-    if info:
-        canonical["file_info"] = info
-
-
 def _scan_roots(output_dirs):
     """(absolute root, public label, listing index), the most specific first.
 
@@ -589,18 +586,56 @@ class _BatchHistory:
     """A scan batch's recorded-history lookup, made on the first file needing it.
 
     Most files link through their metadata and never ask; the first one that
-    does triggers one lookup for the whole batch (``_recorded_runs_for``).
+    does triggers one ``recorded_runs`` pass for the whole batch, from which
+    both the stored prompt (``get``) and the recorded saver (``recorded``)
+    of every file are read. A failure is logged and names nothing.
     """
 
     def __init__(self, api, media_files):
         self._api = api
-        self._files = media_files
-        self._found = None
+        self._files = [str(path) for path in media_files]
+        self._runs = None
+        self._prompts = {}
+
+    def _found(self):
+        if self._runs is None:
+            try:
+                self._runs = recorded_runs(
+                    self._api.db, self._files, host_output_directory(), None
+                )
+            except Exception:
+                self._api.logger.warning(
+                    "History attribution failed for a scan batch", exc_info=True
+                )
+                self._runs = {}
+        return self._runs
 
     def get(self, path):
-        if self._found is None:
-            self._found = self._api._recorded_runs_for(self._files)
-        return self._found.get(path)
+        """The stored prompt id history attributes to ``path``, else None."""
+        found = self._found().get(path)
+        if found is None:
+            return None
+        run_id = found[0]
+        if run_id not in self._prompts:
+            try:
+                self._prompts[run_id] = named_prompt(self._api.db, run_id)
+            except Exception:
+                self._api.logger.warning(
+                    "History attribution failed for a scan batch", exc_info=True
+                )
+                self._prompts[run_id] = None
+        return self._prompts[run_id]
+
+    def recorded(self, path):
+        """``RecordedOutput(saver, {})`` for an attributed file, else None.
+
+        ``saver`` is set only when the run reported the file from one node.
+        """
+        found = self._found().get(path)
+        if found is None:
+            return None
+        nodes = found[1]
+        return RecordedOutput(next(iter(nodes)) if len(nodes) == 1 else None, {})
 
 
 class AdminRoutesMixin:
@@ -1665,24 +1700,6 @@ class AdminRoutesMixin:
                 note(media_file, status)
         return counts
 
-    @staticmethod
-    def _graph_prompt_text(metadata):
-        """The positive prompt the browser parser traces from the graphs, or None.
-
-        Records parse as the legacy pages parsed them (NaN reads as null).
-        """
-        records = {}
-        for key in ("prompt", "workflow"):
-            value = metadata.get(key)
-            records[key] = parse_page_json(value) if isinstance(value, str) else value
-        try:
-            positive = extract_generation_params(records)["positivePrompt"]
-        except GenerationParamsError:
-            return None
-        if not isinstance(positive, str) or positive == PLACEHOLDERS["positivePrompt"]:
-            return None
-        return positive.strip() or None
-
     def _ingest_scanned_file(
         self,
         media_file,
@@ -1693,124 +1710,81 @@ class AdminRoutesMixin:
         records=None,
         history=None,
     ):
-        """Enrich exact pairs; retain legacy positive-only import for absent pairs.
+        """Store the prompt the shared per-file decision names (#247).
 
-        Returns the file's scan status, or None without prompt metadata. With
-        ``detailed``, a stored or linked file returns a dict instead: its
-        ``status``, the prompt's display ``title`` and, for a resolved pair,
-        the pair counts. When the resolver finds the file ambiguous (whether
-        or not the browser parser then links it) and ``ambiguous`` is a list,
-        ``media_file`` is appended to it; only a file that then returns
-        ``"ambiguous"`` (nothing linked it) is a Needs review case. An
-        ambiguous result with nothing to choose (``reviewable`` is False: fewer
-        than two candidates carry a prompt) that nothing links returns
-        ``"unresolved"`` instead, and ``media_file`` goes to ``untraced``.
-        The parsed records of either ambiguous file go to
-        ``records["canonical"]`` when ``records`` is a dict.
+        ``utils.file_prompt.decide_file_prompt`` (scan policy) decides; this
+        stores the result. Returns the file's scan status, or None without
+        prompt metadata. With ``detailed``, a stored or linked file returns a
+        dict instead: its ``status``, the prompt's display ``title`` and, for
+        a resolved pair, the pair counts. When the resolver finds the file
+        ambiguous (whether or not a later step then links it) and
+        ``ambiguous`` is a list, ``media_file`` is appended to it; only a file
+        that then returns ``"ambiguous"`` (nothing linked it) is a Needs
+        review case. An ambiguous result with nothing to choose
+        (``reviewable`` is False: fewer than two candidates carry a prompt)
+        goes to ``untraced`` instead and returns ``"unresolved"`` when nothing
+        links it. The parsed records of either ambiguous file go to
+        ``records["canonical"]`` when ``records`` is a dict. ``history`` is
+        the batch's lazy recorded-history lookup.
         """
-        if not metadata:
-            # No prompt metadata (--disable-metadata, bare savers): only the
-            # recorded run can say which prompt made it (#342)
-            return self._link_recorded_run(media_file, {}, detailed, history)
-        canonical = dict(metadata)
-        for key in ("prompt", "workflow"):
-            if isinstance(canonical.get(key), str):
-                try:
-                    canonical[key] = json.loads(canonical[key])
-                except (ValueError, TypeError):
-                    # Preserve malformed metadata for conservative resolver rejection.
-                    pass
-        if not canonical.get("prompt") and not canonical.get("parameters"):
-            return self._link_recorded_run(media_file, canonical, detailed, history)
-        _with_file_size(canonical, media_file)
-        resolution = resolve_prompt_pair(canonical)
-        if blank_positive(resolution):
-            # Only prepend/append text: no stored prompt can own it (#342)
-            resolution = {**resolution, "state": "unresolved"}
-        # Ambiguous without a choice: unresolved unless something links it.
-        unchoosable = resolution["state"] == "ambiguous" and not reviewable(resolution)
-        if unchoosable:
-            if untraced is not None:
-                untraced.append(media_file)
-            if records is not None:
-                records["canonical"] = canonical
-        elif resolution["state"] == "ambiguous" and ambiguous is not None:
-            ambiguous.append(media_file)
-            if records is not None:
-                records["canonical"] = canonical
-        if resolution["state"] == "resolved":
-            result = ingest_resolved_pair(
-                self.db, str(media_file), canonical, resolution
+        path = str(media_file)
+        if history is None:  # one file: history is still asked once
+            history = _BatchHistory(self, [media_file])
+        decision = decide_file_prompt(
+            metadata,
+            file_name=Path(path).name,
+            file_size=functools.partial(_file_size, media_file),
+            in_output=False,  # only the monitor policy reads it
+            lookup=functools.partial(history.recorded, path),
+            policy=Policy.SCAN,
+        )
+        if decision.kind in (NOT_SAVED, NOT_SAVED_WIRED):
+            # Save to library off (#90): no write, recorded-run fallback or
+            # Needs review item
+            return decision.kind
+        self._note_review(media_file, decision, ambiguous, untraced, records)
+        if decision.kind == NEVER:
+            if decision.try_recorded_run:
+                linked = self._link_recorded_run(
+                    media_file, decision.canonical, detailed, history
+                )
+                if linked is not None:
+                    return linked
+            return decision.unlinked_status
+        if decision.kind == PAIR:
+            # A rerun after a library move re-checks the link rows.
+            result = self.db.survive_library_switch(
+                lambda: ingest_resolved_pair(
+                    self.db,
+                    path,
+                    decision.canonical,
+                    decision.resolution,
+                    marks=SCAN_MARKS,
+                )
             )
             if not detailed:
                 return result["status"]
-            return {
-                **result,
-                "title": self._scan_title(media_file, resolution["positive"]["text"]),
-            }
-        if canonical.get("prompt") and resolution["state"] in (
-            "ambiguous",
-            "unresolved",
-        ):
-            # The resolver could not decide: link the prompt the inspector and the
-            # legacy pages show for this file (the browser parser, in Python).
-            # Graphs without a recognized sampler keep the legacy extractor below.
-            prompt_text = self._graph_prompt_text(metadata)
-            if not prompt_text:
-                linked = self._link_recorded_run(
-                    media_file, canonical, detailed, history
-                )
-                if linked is not None:
-                    return linked
-                return "unresolved" if unchoosable else resolution["state"]
-        elif resolution["state"] == "ambiguous":
-            return "unresolved" if unchoosable else "ambiguous"
-        elif resolution["state"] == "unresolved" and not isinstance(
-            canonical.get("parameters"), str
-        ):
-            return resolution["state"]
-        elif canonical.get("prompt"):
-            # No node takes a positive or negative input (a guider or wrapper
-            # graph): the one wired text, never a guess (#252).
-            found = sampler_less_prompt(canonical["prompt"])
-            prompt_text = found["text"]
-            if prompt_text is None and isinstance(canonical.get("parameters"), str):
-                # A metadata saver also wrote A1111 text: its positive prompt.
-                parsed = self._parse_comfyui_prompt(metadata)["positive_prompt"]
-                prompt_text = parsed.strip() if parsed and parsed.strip() else None
-            if prompt_text is None:
-                linked = self._link_recorded_run(
-                    media_file, canonical, detailed, history
-                )
-                if linked is not None:
-                    return linked
-                return "unresolved" if found["state"] == "ambiguous" else "found"
-        else:
-            prompt_text = self._parse_comfyui_prompt(metadata)["positive_prompt"]
-            if not (prompt_text and prompt_text.strip()):
-                return "found"
-            prompt_text = prompt_text.strip()
-
-        status = self._store_scanned_prompt(media_file, canonical, prompt_text)
+            title_text = decision.resolution["positive"]["text"]
+            return {**result, "title": self._scan_title(media_file, title_text)}
+        status, _prompt_id = store_scanned_prompt(
+            self.db, path, decision.canonical, decision.text, SCAN_MARKS
+        )
         if not detailed or status not in SCAN_TITLED_DETAILS:
             return status
-        return {"status": status, "title": self._scan_title(media_file, prompt_text)}
+        return {"status": status, "title": self._scan_title(media_file, decision.text)}
 
-    def _recorded_runs_for(self, media_files):
-        """``{path: prompt id or None}`` from recorded history for a scan batch.
-
-        One output lookup per chunk of names (``stored_run_prompts``), so a
-        batch never queries history once per file. An empty dict on failure.
-        """
-        try:
-            return stored_run_prompts(
-                self.db, [str(path) for path in media_files], host_output_directory()
-            )
-        except Exception:
-            self.logger.warning(
-                "History attribution failed for a scan batch", exc_info=True
-            )
-            return {}
+    @staticmethod
+    def _note_review(media_file, decision, ambiguous, untraced, records):
+        """Hand an ambiguous file to the batch's Needs review lists."""
+        if decision.review == "untraced":
+            if untraced is not None:
+                untraced.append(media_file)
+            if records is not None:
+                records["canonical"] = decision.canonical
+        elif decision.review == "ambiguous" and ambiguous is not None:
+            ambiguous.append(media_file)
+            if records is not None:
+                records["canonical"] = decision.canonical
 
     def _link_recorded_run(self, media_file, canonical, detailed=False, history=None):
         """Link a file recorded history attributes to one stored prompt (#342).
@@ -1819,31 +1793,25 @@ class AdminRoutesMixin:
         ``{"status", "title"}``), or None when history names no single stored
         prompt for the file (``database.run_attribution``). Never creates a
         prompt. Works without any prompt metadata in the file. ``history`` is
-        the batch's precomputed lookup; without it the file is looked up alone.
+        the batch's lookup (``_BatchHistory``). A file already linked to
+        another prompt is a ``conflict``: one prompt per file.
         """
-        if history is not None:
-            prompt_id = history.get(str(media_file))
-        else:
-            try:
-                prompt_id = stored_run_prompt(
-                    self.db, str(media_file), host_output_directory()
-                )
-            except Exception:
-                self.logger.warning(
-                    f"History attribution failed for {media_file.name}",
-                    exc_info=True,
-                )
-                return None
+        if history is None:
+            history = _BatchHistory(self, [media_file])
+        prompt_id = history.get(str(media_file))
         if prompt_id is None:
             return None
         try:
-            image_id = self.db.link_image_to_prompt(
-                prompt_id, str(media_file), metadata=canonical
+            status = self.db.survive_library_switch(
+                lambda: link_stored_prompt_or_raise(
+                    self.db, str(media_file), canonical, prompt_id
+                )
             )
         except Exception as e:
             self.logger.error(f"Failed to link {media_file.name} from history: {e}")
             return "found"
-        status = "linked" if image_id else "unchanged"
+        if status is None:
+            return None
         if not detailed or status not in SCAN_TITLED_DETAILS:
             return status
         prompt = self.db.get_prompt_by_id(prompt_id) or {}
@@ -1863,35 +1831,3 @@ class AdminRoutesMixin:
         except Exception:
             self.logger.warning("Scan row title could not be read; using its text")
         return display_title((), text)
-
-    def _store_scanned_prompt(self, media_file, canonical, prompt_text):
-        """Link ``media_file`` to ``prompt_text``'s prompt, saving it when new."""
-        prompt_hash = generate_prompt_hash(prompt_text)
-        existing = self.db.get_prompt_by_hash(prompt_hash)
-        if existing:
-            try:
-                image_id = self.db.link_image_to_prompt(
-                    existing["id"], str(media_file), metadata=canonical
-                )
-            except Exception as e:
-                self.logger.error(
-                    f"Failed to link {media_file.name} to existing prompt: {e}"
-                )
-                return "found"
-            return "linked" if image_id else "unchanged"
-
-        prompt_id = self.db.save_prompt(
-            prompt_text,
-            "scanned",
-            ["auto-scanned"],
-            None,
-            f"Auto-scanned from {media_file.name}",
-            prompt_hash,
-        )
-        if not prompt_id:
-            return "found"
-        try:
-            self.db.link_image_to_prompt(prompt_id, str(media_file), metadata=canonical)
-        except Exception as e:
-            self.logger.error(f"Failed to link {media_file.name} to new prompt: {e}")
-        return "added"

py/api/review_recording.py

@@ -3,7 +3,7 @@
 The rule (maintainer test of #284/#288): a file is queued only when its
 final outcome is ambiguous and unlinked. The strict resolver must find it
 ``ambiguous``, the scan's legacy fallback (the browser parser,
-``_graph_prompt_text``) must not link it, and no saved row may link it
+``graph_prompt_text``) must not link it, and no saved row may link it
 already. A file the fallback links, an exact pair the ingester stored and an
 already linked file are never queued. Neither is an ambiguous file with
 nothing to choose: fewer than two of its candidates carry a prompt

tests/browser/classic-node.spec.mjs

@@ -19,7 +19,7 @@ test.beforeEach(async ({page, browserName}) => {
         await import('/extensions/promptmanager/prompt_manager.js');
         class Node {
             constructor(){
-                this.widgets=['text','category','tags','search_text'].map(name=>({name,value:'',inputEl:document.createElement('textarea')}));
+                this.widgets=[...['text','category','tags','search_text'].map(name=>({name,value:'',inputEl:document.createElement('textarea')})),{name:'save_to_library',type:'toggle',value:true}];
                 this.widgets_values=[]; this.inputs=[];
             }
             addDOMWidget(name,type,element){document.body.append(element);return {element};}
@@ -36,6 +36,14 @@ test.afterEach(async ({page,browserName}) => {
     expect(page.classicErrors).toEqual([]);
 });
 
+test('Classic loads an older workflow with Save to library on and keeps a saved off (#90)',async({page})=>{
+    // A workflow saved before the switch held the search UI's slot ('') at the switch's index
+    const loaded=await page.evaluate(()=>{const widget=node.widgets.find(w=>w.name==='save_to_library');widget.value='';node.configure({});return widget.value;});
+    expect(loaded).toBe(true);
+    const off=await page.evaluate(()=>{const widget=node.widgets.find(w=>w.name==='save_to_library');widget.value=false;node.configure({});return [widget.value,node.serialize().widgets_values[4]];});
+    expect(off).toEqual([false,false]);
+});
+
 test('Classic keeps immediate selection, search query and metadata serialization',async({page})=>{
     await page.route('**/prompt_manager/search?**', route=>route.fulfill({json:{success:true,results:[{text:'A new portrait',category:'Portrait',tags:['film grain'],rating:5}]}}));
     await page.evaluate(()=>{node.widgets.find(w=>w.name==='search_text').value='tag:"film grain" portrait';});

tests/browser/focused-node.spec.mjs

@@ -26,7 +26,7 @@ test.beforeEach(async({page,browserName})=>{
         window.fixtureApp={registerExtension(extension){this.extension=extension;},ui:{settings:new EventTarget()},graph:{setDirtyCanvas(){}},canvas:{}};
         await import('/extensions/promptmanager/prompt_manager.js');
         class Node {
-            constructor(){this.widgets=['text','category','tags','search_text'].map(name=>({name,type:'text',value:'',options:{},inputEl:document.createElement('textarea')}));this.inputs=[...['clip','prepend_text','append_text'].map(name=>({name,type:name==='clip'?'CLIP':'STRING',link:null})),...['text','category','tags','search_text'].map(name=>({name,type:'STRING',link:null,widget:{name}}))];this.properties={};this.size=[550,700];}
+            constructor(){this.widgets=[...['text','category','tags','search_text'].map(name=>({name,type:'text',value:'',options:{},inputEl:document.createElement('textarea')})),{name:'save_to_library',type:'toggle',value:true,options:{}}];this.inputs=[...['clip','prepend_text','append_text'].map(name=>({name,type:name==='clip'?'CLIP':'STRING',link:null})),...['text','category','tags','search_text'].map(name=>({name,type:'STRING',link:null,widget:{name}})),{name:'save_to_library',type:'BOOLEAN',link:null,widget:{name:'save_to_library'}}];this.properties={};this.size=[550,700];}
             findInputByType(type){const index=this.inputs.findIndex(input=>input.type===type&&input.link==null);return index<0?undefined:{index,slot:this.inputs[index]};}
             addWidget(type,name,value,callback,options){const widget={type,name,value,callback,options};this.widgets.push(widget);return widget;}
             addDOMWidget(name,type,element){const host=document.createElement('div');host.style.height='650px';host.className='fixture-dom-widget';host.append(element);document.body.append(host);return {element};}
@@ -41,6 +41,32 @@ test.beforeEach(async({page,browserName})=>{
 });
 test.afterEach(async({page,browserName})=>{await saveCoverage(page,browserName);expect(page.focusedErrors).toEqual([]);});
 
+test(`${variant} Save to library switch saves its value and loads older workflows on (#90)`, async ({page}) => {
+    const toggle=page.getByRole('checkbox',{name:'Save to library',exact:true});
+    await expect(toggle).toBeChecked();
+    expect(await page.evaluate(()=>node.widgets.map(widget=>widget.name))).toEqual(['text','category','tags','search_text','prepend_text','append_text','save_to_library']);
+    // The native switch is drawn in the toolbar only: the host keeps it hidden and zero-height, like the text fallbacks
+    expect(await page.evaluate(()=>{const widget=node.widgets.find(item=>item.name==='save_to_library');return {type:widget.type,hidden:widget.hidden,size:widget.computeSize()};})).toEqual({type:'hidden',hidden:true,size:[0,-4]});
+    expect(await page.evaluate(()=>node.widgets.find(item=>item.name==='text').computeSize())).toEqual([0,-4]);
+    await toggle.uncheck();
+    await expect(page.locator('.pm-focused > .pm-editor-status[role=status]')).toHaveText('Save to library is off: runs of this node are not added to the library.');
+    const saved=await page.evaluate(()=>node.serialize());
+    expect(saved.widgets_values[6]).toBe(false);
+    await page.evaluate(saved=>node.configure(saved),saved);
+    await expect(toggle).not.toBeChecked();
+    await page.getByLabel('Main prompt',{exact:true}).fill('Typed while off');
+    await expect(page.locator('.pm-focused > .pm-editor-status[role=status]')).toHaveText('Save to library is off: runs of this node are not added to the library.');
+    // A workflow saved before the switch held the editor's own slot ('') at its index
+    await page.evaluate(()=>node.configure({widgets_values:['Old text','','','','','',''],properties:{}}));
+    await expect(toggle).toBeChecked();
+    expect(await page.evaluate(()=>node.widgets.find(widget=>widget.name==='save_to_library').value)).toBe(true);
+    await toggle.check();
+    await page.getByLabel('Main prompt',{exact:true}).fill('Typed while on');
+    await expect(page.locator('.pm-focused > .pm-editor-status[role=status]')).toHaveText('Main prompt is saved when this node runs.');
+    await page.evaluate(()=>{node.inputs.find(input=>input.name==='save_to_library').link=9;node.onConnectionsChange?.();});
+    await expect(toggle).toBeDisabled();
+});
+
 test(`${variant} library autocompletes filters and searches without changing editor metadata`, async ({page}) => {
     const requests=[];
     await page.route('**/prompt_manager/categories',route=>route.fulfill({json:{success:true,categories:['Portrait']}}));
@@ -257,7 +283,7 @@ test(`${variant} restores workflow values and disclosures and protects connected
 test(`${variant} offers text as a link socket, keeps typed text while linked and through a reload (#343)`,async({page})=>{
     const sockets=()=>page.evaluate(()=>Object.fromEntries(node.inputs.map(input=>[input.name,input.widget?.name??null])));
     // The hidden text widget no longer owns the input, so the host draws it as a socket; the other widget inputs are unchanged.
-    expect(await sockets()).toEqual({clip:null,prepend_text:null,append_text:null,text:null,category:'category',tags:'tags',search_text:'search_text'});
+    expect(await sockets()).toEqual({clip:null,prepend_text:null,append_text:null,text:null,category:'category',tags:'tags',search_text:'search_text',save_to_library:'save_to_library'});
     expect(await page.evaluate(()=>node.findInputByType('STRING').slot.name)).toBe('text');
     expect(await page.evaluate(()=>node.findInputByType('CLIP').slot.name)).toBe('clip');
     const main=page.getByRole('textbox',{name:'Main prompt',exact:true});

tests/browser/pair-node.spec.mjs

@@ -17,7 +17,7 @@ test.beforeEach(async({page,browserName})=>{
         window.fixtureApp={registerExtension(extension){this.extension=extension;},ui:{settings},graph:{_nodes:[],setDirtyCanvas(){}},canvas:{},graphToPrompt:async()=>({})};
         await import('/extensions/promptmanager/prompt_manager.js');
         class Node {
-            constructor(){this.id=1;this.type='PromptManagerPair';this.widgets=['text','negative_text','category','tags','search_text'].map(name=>({name,type:'text',value:'',options:{},inputEl:document.createElement('textarea')}));this.inputs=[...['clip','prepend_text','append_text'].map(name=>({name,type:name==='clip'?'CLIP':'STRING',link:null})),...['text','negative_text','category','tags','search_text'].map(name=>({name,type:'STRING',link:null,widget:{name}}))];this.properties={};this.size=[1100,700];}
+            constructor(){this.id=1;this.type='PromptManagerPair';this.widgets=[...['text','negative_text','category','tags','search_text'].map(name=>({name,type:'text',value:'',options:{},inputEl:document.createElement('textarea')})),{name:'save_to_library',type:'toggle',value:true,options:{}}];this.inputs=[...['clip','prepend_text','append_text'].map(name=>({name,type:name==='clip'?'CLIP':'STRING',link:null})),...['text','negative_text','category','tags','search_text'].map(name=>({name,type:'STRING',link:null,widget:{name}})),{name:'save_to_library',type:'BOOLEAN',link:null,widget:{name:'save_to_library'}}];this.properties={};this.size=[1100,700];}
             findInputByType(type){const index=this.inputs.findIndex(input=>input.type===type&&input.link==null);return index<0?undefined:{index,slot:this.inputs[index]};}
             addWidget(type,name,value,callback,options){const widget={type,name,value,callback,options};this.widgets.push(widget);return widget;}
             addDOMWidget(name,type,element){const host=document.createElement('div');host.style.cssText='height:700px;width:1100px;max-width:100%';host.className='fixture-dom-widget';host.append(element);document.body.append(host);return {element};}
@@ -34,6 +34,18 @@ test.afterEach(async({page,browserName})=>{await saveCoverage(page,browserName);
 async function selectPair(page){await page.getByRole('button',{name:'Recent',exact:true}).click();await page.locator('.pm-editor-result').click();}
 async function draft(page,positive='Draft positive',negative='Draft negative'){await page.getByRole('textbox',{name:'Positive prompt',exact:true}).fill(positive);await page.getByRole('textbox',{name:'Negative prompt',exact:true}).fill(negative);}
 
+test('paired editor Save to library switch keeps the widget order and loads older workflows on (#90)',async({page})=>{
+    await expect(page.locator('.pm-pair')).toBeVisible();
+    const toggle=page.getByRole('checkbox',{name:'Save to library',exact:true});
+    await expect(toggle).toBeChecked();
+    expect(await page.evaluate(()=>node.widgets.map(widget=>widget.name))).toEqual(['text','negative_text','category','tags','search_text','prepend_text','append_text','save_to_library']);
+    await toggle.uncheck();
+    expect(await page.evaluate(()=>node.serialize().widgets_values[7])).toBe(false);
+    await expect(page.locator('.pm-pair > .pm-editor-status[role=status]').last()).toHaveText('Save to library is off: runs of this node are not added to the library.');
+    await page.evaluate(()=>node.configure({widgets_values:['P','N','','','','','',''],properties:{}}));
+    await expect(toggle).toBeChecked();
+});
+
 test('paired editor previews both texts before explicitly replacing both',async({page})=>{
     await expect(page.locator('.pm-pair')).toBeVisible();
     await draft(page);

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

@@ -16,8 +16,8 @@ installHostHooks(test, ['PromptManager', 'PromptManagerText', 'PromptManagerFocu
 // `primitive`: a core string node the queue-time resolver follows. `legacy`: the frontend's virtual Primitive node,
 // whose value is queued as literal text. `preview`: PrimitiveString -> PreviewAny, the shape of the maintainer's
 // TextGenerate -> PreviewAny chain, whose text is only known when the node runs.
-function build(page, { kind, positive, negative, prefix }) {
-    return page.evaluate(async ({ kind, positive, negative, prefix }) => {
+function build(page, { kind, positive, negative, prefix, saveToLibrary }) {
+    return page.evaluate(async ({ kind, positive, negative, prefix, saveToLibrary }) => {
         const { app } = await import('/scripts/app.js');
         app.graph.clear();
         const add = (type, x, y) => { const node = LiteGraph.createNode(type); app.graph.add(node); node.pos = [x, y]; return node; };
@@ -28,6 +28,7 @@ function build(page, { kind, positive, negative, prefix }) {
         const set = (name, value) => { const widget = prompt.widgets.find(item => item.name === name); widget.value = value; if (widget.inputEl) widget.inputEl.value = value; };
         set('text', positive.typed);
         if (negative) set('negative_text', negative.typed);
+        if (saveToLibrary !== undefined) prompt.widgets.find(item => item.name === 'save_to_library').value = saveToLibrary;
         const drive = (name, side, y) => {
             if (!side.via) return;
             if (side.via === 'legacy') {
@@ -73,7 +74,7 @@ function build(page, { kind, positive, negative, prefix }) {
         await prompt._pmTextSerializerReady;
         const { output, workflow } = await app.graphToPrompt();
         return { output, workflow, promptNode: String(prompt.id) };
-    }, { kind, positive, negative, prefix });
+    }, { kind, positive, negative, prefix, saveToLibrary });
 }
 
 async function run(execution, page, testCase, built) {
@@ -195,6 +196,47 @@ for (const [label, linked, negativeVia] of [['positive', ['text'], 'primitive'],
     });
 }
 
+// #90: with Save to library off the run still produces its image, but PromptManager saves, counts and links nothing,
+// and the run's history records no prompt text for it.
+for (const kind of ['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerSplit', 'PromptManagerPair']) {
+    test(`${kind} with Save to library off saves, counts and links nothing (#90)`, async ({ execution, page }) => {
+        test.setTimeout(90000);
+        const text = unique(`Not saved ${kind}`);
+        const negative = kind === 'PromptManagerPair' ? { typed: unique('Not saved negative') } : undefined;
+        const prefix = `notsaved/${randomUUID()}`;
+        const { output, workflow, promptNode } = await build(page, { kind, positive: { typed: text }, negative, prefix, saveToLibrary: false });
+        expect(output[promptNode].inputs.save_to_library).toBe(false);
+        const id = randomUUID();
+        const response = await execution.request.post('/prompt', { data: { prompt: output, prompt_id: id, extra_data: { extra_pnginfo: { workflow } } } });
+        expect(response.ok(), await response.text()).toBe(true);
+        let item;
+        await expect.poll(async () => {
+            item = (await (await execution.request.get(`/history/${id}`)).json())[id];
+            return item?.status?.status_str ?? null;
+        }, { timeout: 20000 }).toBe('success');
+        let result;
+        await expect.poll(async () => {
+            result = await readExecutionRuns(execution.databasePath, id);
+            const run = result.rows[0];
+            return Boolean(run && result.captures[run.run_id]);
+        }, { timeout: 10000, message: 'the run reaches SQLite' }).toBe(true);
+        const run = result.rows[0];
+        for (const context of result.captures[run.run_id].contexts || []) {
+            expect(context.positive_prompt_id ?? null).toBeNull();
+            expect(context.pair_id ?? null).toBeNull();
+        }
+        expect((result.usage[run.run_id] || []).every(entry => entry.run_count === 0)).toBe(true);
+        const files = Object.values(item.outputs).flatMap(value => value.images || []).map(image => image.filename);
+        expect(files.length).toBeGreaterThan(0);
+        await scan(execution);
+        expect(Object.values(await linkedPrompts(execution, files))).toEqual(files.map(() => null));
+        for (const query of [text, negative?.typed].filter(Boolean)) {
+            const found = await (await execution.request.get(`/prompt_manager/search?text=${encodeURIComponent(query)}`)).json();
+            expect(found.results ?? [], query).toEqual([]);
+        }
+    });
+}
+
 // #347: a cached PromptManager node counts every job, as an executing one does. Jobs after the first change only
 // the sampler seed, so ComfyUI serves the PromptManager node (and its PreviewAny source) from its output cache.
 const COUNTS = `import json, sqlite3, sys, pathlib

tests/comfyui/text-link.spec.mjs

@@ -370,6 +370,7 @@ test('workflows saved before #343 load unchanged and their linked text keeps wor
         const nodes = Object.fromEntries(app.graph._nodes.map(node => [node.id, node]));
         const reloaded = app.graph.serialize();
         return {
+            saves: Object.fromEntries(Object.entries(output).filter(([, node]) => node.class_type.startsWith('PromptManager')).map(([id, node]) => [id, node.inputs.save_to_library])),
             values: Object.fromEntries(reloaded.nodes.map(node => [node.id, node.widgets_values])),
             links: Object.fromEntries(Object.entries(output).map(([id, node]) => [id, { text: node.inputs.text, negative_text: node.inputs.negative_text }])),
             widgetBound: Object.fromEntries(Object.values(nodes).filter(node => node.type.startsWith('PromptManager')).map(node => [node.id, node.inputs.filter(input => ['text', 'negative_text'].includes(input.name) && input.widget).map(input => input.name)])),
@@ -377,8 +378,16 @@ test('workflows saved before #343 load unchanged and their linked text keeps wor
     }, workflow);
     for (const node of workflow.nodes) {
         // Every saved widget value comes back as it was saved, except the typed text main lost under a link (null).
-        if (node.widgets_values?.[0] !== null) expect(result.values[node.id], `${node.type} ${node.id}`).toEqual(node.widgets_values);
+        // #90: the last saved value is the editor's own slot (''), where the Save to library switch is now: it loads on.
+        if (node.widgets_values?.[0] === null) continue;
+        if (!node.type.startsWith('PromptManager')) { expect(result.values[node.id], `${node.type} ${node.id}`).toEqual(node.widgets_values); continue; }
+        const saved = node.widgets_values.length - 1;
+        expect(result.values[node.id].slice(0, saved), `${node.type} ${node.id}`).toEqual(node.widgets_values.slice(0, saved));
+        expect(result.values[node.id][saved], `${node.type} ${node.id} save_to_library`).toBe(true);
     }
+    // Every PromptManager node of the old workflow queues with Save to library on.
+    for (const save of Object.values(result.saves)) expect(save).toBe(true);
+    expect(Object.keys(result.saves).length).toBeGreaterThan(0);
     expect(result.links).toMatchObject({
         1: { text: 'Saved PromptManager text' }, 2: { text: 'Saved PromptManagerText text' }, 3: { text: 'Saved PromptManagerFocused text' },
         4: { text: 'Saved PromptManagerSplit text' }, 5: { text: 'Saved PromptManagerPair text', negative_text: 'Saved pair negative' },

tests/js/save-to-library.test.js

@@ -0,0 +1,68 @@
+// #90: the node's save_to_library switch (web/js/save-to-library.js).
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const source = import('../../web/js/save-to-library.js');
+
+function classicNode(saved) {
+    const names = ['text', 'category', 'tags', 'search_text', 'prepend_text', 'append_text', 'save_to_library', 'prompt_manager_search_ui'];
+    const widgets = names.map(name => ({name, value: name === 'save_to_library' ? true : ''}));
+    const node = {widgets, widgets_values: saved ? [...saved] : undefined};
+    // What a host configure does: each saved value goes to the widget at its index.
+    (saved || []).forEach((value, index) => { if (widgets[index]) widgets[index].value = value; });
+    return node;
+}
+
+test('an older workflow stored the editor slot where the switch is now: it loads on', async () => {
+    const {normalizeSaveWidget, savesToLibrary} = await source;
+    const node = classicNode(['a cat', 'c', 't', '', '', '', '']);
+    assert.equal(node.widgets[6].value, '', 'the host put the old DOM slot value into the switch');
+    assert.equal(normalizeSaveWidget(node), true);
+    assert.equal(node.widgets[6].value, true);
+    assert.equal(node.widgets_values[6], true);
+    assert.equal(savesToLibrary(node), true);
+    assert.deepEqual(node.widgets_values.slice(0, 6), ['a cat', 'c', 't', '', '', ''], 'no earlier value moved');
+});
+
+test('null, undefined and missing slots load on; saved booleans are kept', async () => {
+    const {normalizeSaveWidget, savesToLibrary} = await source;
+    for (const value of [null, undefined, 0, 'false']) {
+        const node = classicNode(['a', '', '', '', '', '', value]);
+        normalizeSaveWidget(node);
+        assert.equal(savesToLibrary(node), true, String(value));
+    }
+    const short = classicNode(['a', '', '', '', '', '']);
+    assert.equal(normalizeSaveWidget(short), false);
+    assert.equal(savesToLibrary(short), true);
+    const off = classicNode(['a', '', '', '', '', '', false]);
+    assert.equal(normalizeSaveWidget(off), false);
+    assert.equal(savesToLibrary(off), false);
+});
+
+test('a node without the widget (an older backend) saves', async () => {
+    const {normalizeSaveWidget, savesToLibrary} = await source;
+    assert.equal(savesToLibrary({widgets: [{name: 'text', value: ''}]}), true);
+    assert.equal(normalizeSaveWidget({widgets: []}), false);
+    assert.equal(savesToLibrary(undefined), true);
+});
+
+test('the switch moves after the editor fallbacks so every frontend has one order', async () => {
+    const {placeSaveWidgetLast} = await source;
+    const widgets = ['text', 'category', 'tags', 'search_text', 'save_to_library', 'prepend_text', 'append_text'].map(name => ({name}));
+    const node = {widgets};
+    assert.equal(placeSaveWidgetLast(node), true);
+    assert.deepEqual(node.widgets.map(widget => widget.name), ['text', 'category', 'tags', 'search_text', 'prepend_text', 'append_text', 'save_to_library']);
+    assert.equal(placeSaveWidgetLast(node), false, 'already last');
+    assert.equal(placeSaveWidgetLast({widgets: [{name: 'text'}]}), false);
+});
+
+test('setting the switch updates the widget, its saved value and notifies the host', async () => {
+    const {setSaveToLibrary, savesToLibrary} = await source;
+    const calls = [];
+    const node = classicNode(['a', '', '', '', '', '', true]);
+    node.widgets[6].callback = value => calls.push(value);
+    setSaveToLibrary(node, false);
+    assert.equal(savesToLibrary(node), false);
+    assert.equal(node.widgets_values[6], false);
+    assert.deepEqual(calls, [false]);
+    setSaveToLibrary({widgets: []}, false);
+});

tests/test_api_admin.py

@@ -2205,24 +2205,30 @@ class TestOutputScanInternals(AdminAPITestCase):
         self.assertEqual(self.api._ingest_scanned_file(f, meta), "unchanged")
 
     def test_ingest_link_failures_are_tolerated(self):
+        """A failed link rolls the new prompt back: found, nothing left behind."""
         f = Path("gen.png")
         meta = {"prompt": json.dumps(_wired_prompt("a dog"))}
-        self.api.db.link_image_to_prompt = _raise
-        self.assertEqual(self.api._ingest_scanned_file(f, meta), "added")
-        self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
+        self.api.db._insert_image_link = _raise
+        with self.assertLogs("prompt_manager.database.scanned_prompts", "ERROR"):
+            self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
+            self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
+        self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
 
     def test_ingest_save_failure_counts_as_found(self):
         f = Path("gen.png")
         meta = {"prompt": json.dumps(_wired_prompt("a bird"))}
-        self.api.db.save_prompt = lambda *a, **k: None
-        self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
+        with patch(
+            "py.api.admin.store_scanned_prompt", return_value=("found", None)
+        ) as store:
+            self.assertEqual(self.api._ingest_scanned_file(f, meta), "found")
+        self.assertEqual(store.call_args.args[3], "a bird")
 
     def test_batch_counts_and_per_file_errors(self):
         meta = {"prompt": json.dumps(_wired_prompt("a fish"))}
-        self.api.db.get_prompt_by_hash = _raise
-        counts = self.api._ingest_scan_batch_sync(
-            [(Path("a.png"), meta), (Path("b.png"), {})]
-        )
+        with patch("py.api.admin.store_scanned_prompt", side_effect=_raise):
+            counts = self.api._ingest_scan_batch_sync(
+                [(Path("a.png"), meta), (Path("b.png"), {})]
+            )
         self.assertEqual(
             counts,
             {

tests/test_cached_runs.py

@@ -210,13 +210,24 @@ class JobCase(unittest.TestCase):
         node = self.node(kind)
         self.context = (self.context[0], index)
         clip = mock.Mock()
+        # The node receives its switch as the queued graph holds it (#90)
+        save = g[node_id]["inputs"].get("save_to_library", True)
         if kind == "PromptManagerText":
-            return node.process_text(text, prompt=g, unique_id=node_id)
+            return node.process_text(
+                text, prompt=g, unique_id=node_id, save_to_library=save
+            )
         if kind == "PromptManagerPair":
             return node.encode_prompt(
-                clip, text, negative_text=negative, prompt=g, unique_id=node_id
+                clip,
+                text,
+                negative_text=negative,
+                prompt=g,
+                unique_id=node_id,
+                save_to_library=save,
             )
-        return node.encode_prompt(clip, text, prompt=g, unique_id=node_id)
+        return node.encode_prompt(
+            clip, text, prompt=g, unique_id=node_id, save_to_library=save
+        )
 
     def cache(self, node_id, *results):
         """Store executed results (one per list index) as ComfyUI's cache entry."""
@@ -313,6 +324,35 @@ class TestCachedJobsCount(JobCase):
         self.job(g, cached=["pm"])
         self.assertEqual(self.db.search_prompts(), [])
 
+    def test_disabled_node_then_cached_jobs_counts_0(self):
+        # #90 x #347: a node with Save to library off records no receipt and is
+        # no run prompt, so its cached jobs replay nothing. The prompt exists
+        # (an enabled run saved it), so any replayed count would show.
+        for kind in (
+            "PromptManager",
+            "PromptManagerFocused",
+            "PromptManagerSplit",
+            "PromptManagerText",
+            "PromptManagerPair",
+        ):
+            with self.subTest(kind=kind):
+                text = f"{LLM} not saved by {kind}"
+                negative = "blurry" if kind == "PromptManagerPair" else ""
+                self.first(graph(kind, negative_text=negative), text, negative)
+                before = (self.count(text), self.pair_count(text, negative))
+
+                def off(seed):
+                    g = graph(kind, negative_text=negative, seed=seed)
+                    g["pm"]["inputs"]["save_to_library"] = False
+                    return g
+
+                self.first(off(1), text, negative)
+                self.job(off(2), cached=["pm"])
+                self.job(off(3), cached=["pm"])
+                after = (self.count(text), self.pair_count(text, negative))
+                self.assertEqual(before[0], 1)
+                self.assertEqual(after, before)
+
     def test_alternating_upstream_values_each_count_their_own_prompt(self):
         g = graph()
         self.first(g, text="value A")

tests/test_file_prompt.py

@@ -0,0 +1,705 @@
+"""The per-file prompt decision shared by the scan and the monitor (#247).
+
+One test per case, named by the ladder step of the design
+(``docs/egm/247-foreign-workflows``). The ladder does no file or database
+I/O: history comes in through ``lookup``.
+"""
+
+import json
+import unittest
+from unittest import mock
+
+from database.run_attribution import RecordedOutput
+from tests.test_pair_scan import flux_graph, sampled_graph
+from tests.test_prompt_attribution import issue_graph, pair_evidence
+from tests import test_save_to_library as save_tests
+from utils import file_prompt
+from utils.file_prompt import Policy, decide_file_prompt, junk_text
+from utils.prompt_inspector import InspectionResult
+
+try:
+    from tests import inspection_parity as parity
+    from tests import saver_anchor_graphs as shapes
+except ImportError:  # pragma: no cover - discover -s tests puts tests/ first
+    import inspection_parity as parity
+    import saver_anchor_graphs as shapes
+
+SHARED_NAME = "IMG_GEN_00371_.png"
+LORA = "<lora:style:1.00>"
+
+
+def _decide(
+    metadata,
+    *,
+    policy=Policy.SCAN,
+    lookup=None,
+    in_output=True,
+    file_name=SHARED_NAME,
+    file_size=None,
+):
+    return decide_file_prompt(
+        metadata,
+        file_name=file_name,
+        file_size=file_size,
+        in_output=in_output,
+        lookup=lookup,
+        policy=policy,
+    )
+
+
+def _graph(graph):
+    return {"prompt": json.dumps(graph)}
+
+
+class _Lookup:
+    """A history lookup that counts its calls."""
+
+    def __init__(self, saver=None):
+        self.calls = 0
+        self.found = RecordedOutput(saver, {}) if saver is not None else None
+
+    def __call__(self):
+        self.calls += 1
+        return self.found
+
+
+def _inspected(state, malformed=False):
+    resolution = {"state": state, "positive": None, "negative": None}
+    return InspectionResult(resolution, malformed=malformed)
+
+
+def _untraced_by_both():
+    """A sampled graph whose positive no profile traces; the legacy parser does."""
+    return mock.patch.object(
+        file_prompt, "inspection_resolution", return_value=_inspected("unresolved")
+    )
+
+
+def _gated(graph):
+    """``graph`` past the default profile's node gate, so it is unresolved."""
+    graph = dict(graph)
+    for index in range(510):
+        graph[f"pad{index}"] = {"class_type": "Note", "inputs": {}}
+    return graph
+
+
+def _prompt_manager_branches(prepend="PRE", append="APP"):
+    """Two KSampler branches, each fed by a PromptManagerText string."""
+    graph = sampled_graph()
+    del graph["1"]
+    sampler = graph.pop("3")
+    for index, text in enumerate(("a cat", "a dog"), start=1):
+        base = index * 10
+        graph[str(base)] = {
+            "class_type": "PromptManagerText",
+            "inputs": {"text": text, "prepend_text": prepend, "append_text": append},
+        }
+        graph[str(base + 1)] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": [str(base), 0]},
+        }
+        graph[str(base + 2)] = {
+            **sampler,
+            "inputs": {**sampler["inputs"], "positive": [str(base + 1), 0]},
+        }
+    return graph
+
+
+def _numeric_first_branches():
+    """Three branches; the legacy parser's answer (the first) is a number."""
+    graph = sampled_graph("2048")
+    for index, text in ((5, "a cat"), (6, "a dog")):
+        graph[str(index * 10)] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": text},
+        }
+        graph[str(index * 10 + 1)] = {
+            **graph["3"],
+            "inputs": {**graph["3"]["inputs"], "positive": [str(index * 10), 0]},
+        }
+    return graph
+
+
+class TestStepsBeforeResolving(unittest.TestCase):
+    def test_step_1_no_metadata_tries_the_recorded_run(self):
+        for metadata in ({}, None):
+            decision = _decide(metadata)
+            self.assertEqual(decision.kind, "never")
+            self.assertTrue(decision.try_recorded_run)
+            self.assertIsNone(decision.unlinked_status)
+            self.assertEqual(decision.canonical, {})
+
+    def test_step_3_neither_prompt_nor_parameters_tries_the_recorded_run(self):
+        decision = _decide({"workflow": json.dumps({"nodes": []})})
+        self.assertEqual(decision.kind, "never")
+        self.assertTrue(decision.try_recorded_run)
+        self.assertIsNone(decision.unlinked_status)
+        self.assertEqual(decision.canonical, {"workflow": {"nodes": []}})
+
+    def test_strings_are_parsed_and_malformed_ones_kept(self):
+        decision = _decide({"prompt": "{not json", "workflow": "{}"})
+        self.assertEqual(decision.canonical["prompt"], "{not json")
+        self.assertEqual(decision.canonical["workflow"], {})
+
+    def test_the_file_size_is_recorded_unless_the_metadata_has_one(self):
+        meta = _graph(sampled_graph())
+        self.assertEqual(
+            _decide(meta, file_size=42).canonical["file_info"], {"size": 42}
+        )
+        meta["file_info"] = {"size": 7, "width": 3}
+        self.assertEqual(
+            _decide(meta, file_size=42).canonical["file_info"], {"size": 7, "width": 3}
+        )
+        self.assertNotIn("file_info", _decide(_graph(sampled_graph())).canonical)
+
+    def test_a_file_size_reader_is_called_only_for_prompt_metadata(self):
+        reads = []
+
+        def size():
+            reads.append(1)
+            return 9
+
+        for metadata in ({}, {"workflow": "{}"}):
+            _decide(metadata, file_size=size)
+        self.assertEqual(reads, [])
+        decision = _decide(_graph(sampled_graph()), file_size=size)
+        self.assertEqual((reads, decision.canonical["file_info"]), ([1], {"size": 9}))
+
+
+def _off_behind(middle, key):
+    """A switched-off PromptManagerText reaching the encoder through
+    ``middle``'s ``key`` input."""
+    graph = save_tests.run_graph()
+    graph["1"] = {
+        "class_type": "PromptManagerText",
+        "inputs": {"text": "SECRET OFF TEXT", "save_to_library": False},
+    }
+    graph["7"] = {"class_type": middle, "inputs": {key: ["1", 0]}}
+    graph["8"] = {
+        "class_type": "CLIPTextEncode",
+        "inputs": {"text": ["7", 0], "clip": ["4", 1]},
+    }
+    graph["3"]["inputs"]["positive"] = ["8", 0]
+    return graph
+
+
+class TestSaveToLibrarySteps(unittest.TestCase):
+    """#90: a PromptManager node with Save to library off stores nothing."""
+
+    def test_step_2_an_opted_out_file_is_not_saved(self):
+        lookup = _Lookup("9")
+        graph = save_tests.run_graph(save=False)
+        graph["5"] = {**graph["3"], "inputs": {**graph["3"]["inputs"]}}
+        graph["5"]["inputs"]["positive"] = ["2", 0]  # ambiguous otherwise
+        for policy in Policy:
+            decision = _decide(_graph(graph), policy=policy, lookup=lookup)
+            self.assertEqual(decision.kind, "not_saved")
+            self.assertFalse(decision.try_recorded_run)
+            self.assertIsNone(decision.review)
+        self.assertEqual(lookup.calls, 0)
+
+    def test_step_2_a_wired_switch_is_not_saved_wired(self):
+        graph = save_tests.run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        lookup = _Lookup("9")
+        decision = _decide(_graph(graph), lookup=lookup)
+        self.assertEqual(decision.kind, "not_saved_wired")
+        self.assertFalse(decision.try_recorded_run)
+        self.assertIsNone(decision.review)
+        self.assertEqual(lookup.calls, 0)
+
+    def test_step_8_1_an_off_node_behind_a_pass_through_is_not_saved(self):
+        """The #90 graph walk does not follow ``anything`` or ``value``; the
+        inspector does, so its positive is checked against the switch."""
+        for middle, key in (("easy showAnything", "anything"), ("MyNode", "value")):
+            with self.subTest(middle=middle):
+                graph = _off_behind(middle, key)
+                for policy in Policy:
+                    decision = _decide(_graph(graph), policy=policy, lookup=_Lookup())
+                    self.assertEqual(decision.kind, "not_saved")
+                    self.assertIsNone(decision.text)
+
+    def test_step_8_4_a_legacy_answer_from_an_off_node_is_not_saved(self):
+        graph = _off_behind("MyNode", "value")
+        graph["9"] = {"class_type": "CLIPTextEncode", "inputs": {"text": "on text"}}
+        graph["5"] = {**graph["3"], "inputs": {**graph["3"]["inputs"]}}
+        graph["5"]["inputs"]["positive"] = ["9", 0]
+        decision = _decide(_graph(graph), lookup=_Lookup())
+        self.assertEqual(decision.kind, "not_saved")
+        self.assertIsNone(decision.text)
+
+    def test_step_2_an_off_node_on_any_positive_input_is_not_saved(self):
+        """Any input name upstream of a positive, not only prompt-like ones."""
+        guider = _off_behind("MyNode", "value")
+        guider["3"] = {
+            "class_type": "BasicGuider",
+            "inputs": {"model": ["4", 0], "conditioning": ["8", 0]},
+        }
+        prepended = _off_behind("MyNode", "value")
+        prepended["1"]["inputs"]["prepend_text"] = "pre"
+        pair = _off_behind("MyNode", "value")
+        pair["1"] = {
+            "class_type": "PromptManagerPair",
+            "inputs": {"text": "x", "negative_text": "y", "save_to_library": False},
+        }
+        pair["2"]["inputs"]["text"] = ["1", 1]
+        cases = {"sampler-less guider": guider, "prepend": prepended, "pair": pair}
+        for name, graph in cases.items():
+            with self.subTest(name=name):
+                lookup = _Lookup("9")
+                decision = _decide(_graph(graph), lookup=lookup)
+                self.assertEqual(decision.kind, "not_saved")
+                self.assertFalse(decision.try_recorded_run)
+                self.assertEqual(lookup.calls, 0)
+
+    def test_step_2_a_wired_switch_behind_a_pass_through_is_not_saved_wired(self):
+        graph = _off_behind("MyNode", "value")
+        graph["1"]["inputs"]["save_to_library"] = ["20", 0]
+        graph["20"] = {"class_type": "SomeBoolNode", "inputs": {}}
+        self.assertEqual(_decide(_graph(graph)).kind, "not_saved_wired")
+
+    def test_a_switched_on_node_behind_a_pass_through_is_stored(self):
+        """Regression guard: on, or no switch at all (older files)."""
+        for value in (True, None):
+            with self.subTest(value=value):
+                graph = _off_behind("MyNode", "value")
+                if value is None:
+                    del graph["1"]["inputs"]["save_to_library"]
+                else:
+                    graph["1"]["inputs"]["save_to_library"] = value
+                decision = _decide(_graph(graph), lookup=_Lookup())
+                self.assertEqual(decision.kind, "positive")
+                self.assertEqual(decision.text, "SECRET OFF TEXT")
+
+    def test_an_off_node_on_a_negative_input_name_never_blocks(self):
+        """The walk skips negative inputs: #90's own rule (``neg``,
+        ``neg_text``, ``text_neg``) and any name containing "negative"."""
+        for key in (
+            "neg",
+            "neg_text",
+            "text_neg",
+            "negative_text",
+            "negative2",
+            "negativeprompt",
+            "NegativePrompt",
+        ):
+            with self.subTest(key=key):
+                graph = {
+                    "1": {
+                        "class_type": "PromptManagerText",
+                        "inputs": {"text": "bad hands", "save_to_library": False},
+                    },
+                    "9": {
+                        "class_type": "PromptManagerText",
+                        "inputs": {"text": "a cat", "save_to_library": True},
+                    },
+                    "7": {
+                        "class_type": "PipeEncode",
+                        "inputs": {"pos_text": ["9", 0], key: ["1", 0]},
+                    },
+                    "4": {"class_type": "CheckpointLoaderSimple", "inputs": {}},
+                    "3": {
+                        "class_type": "KSampler",
+                        "inputs": {
+                            "model": ["4", 0],
+                            "positive": ["7", 0],
+                            "negative": ["7", 1],
+                        },
+                    },
+                }
+                self.assertIsNone(file_prompt._switched_off_upstream(graph))
+                self.assertNotIn(
+                    _decide(_graph(graph)).kind, ("not_saved", "not_saved_wired")
+                )
+
+    def test_the_guards_accept_malformed_nodes(self):
+        graph = {
+            "1": {"class_type": "PromptManagerText", "inputs": None},
+            "2": {"class_type": "PromptManagerText"},
+        }
+        self.assertIsNone(file_prompt._off_text(graph, "x"))
+        side = {"provenance": ["not", "a", "dict"]}
+        with mock.patch.object(file_prompt, "side_opted_out", return_value=True):
+            self.assertIsNone(file_prompt._side_switch(graph, side))
+
+    def test_the_off_text_guard_reads_a_nodes_resolved_text(self):
+        graph = {
+            "1": {
+                "class_type": "PromptManagerText",
+                "inputs": {"text": ["30", 0], "save_to_library": False},
+            },
+            "30": {
+                "class_type": "PrimitiveStringMultiline",
+                "inputs": {"value": "SECRET OFF TEXT"},
+            },
+            "2": {
+                "class_type": "PromptManagerText",
+                "inputs": {"text": "WIRED", "save_to_library": ["9", 0]},
+            },
+            "9": {"class_type": "SomeBoolNode", "inputs": {}},
+        }
+        self.assertEqual(file_prompt._off_text(graph, " SECRET OFF TEXT"), "not_saved")
+        self.assertEqual(file_prompt._off_text(graph, "WIRED"), "not_saved_wired")
+        self.assertIsNone(file_prompt._off_text(graph, "other"))
+
+    def test_steps_5_and_7_check_the_resolved_positive_too(self):
+        """A resolved positive through an off node the #90 walk missed."""
+        cases = {
+            "pair": save_tests.run_graph(save=False),
+            "negative off": save_tests.run_graph(save=False, negative_save=False),
+        }
+        for name, graph in cases.items():
+            with (
+                self.subTest(name=name),
+                mock.patch.object(
+                    file_prompt, "metadata_opted_out", return_value=False
+                ),
+            ):
+                self.assertEqual(_decide(_graph(graph)).kind, "not_saved")
+
+    def test_step_5_a_junk_positive_is_never_stored(self):
+        """Pinned: the positive alone still passes the junk gate, and a blank
+        positive (never storable) stores nothing either."""
+        graph = save_tests.run_graph(save=True, negative_save=False)
+        graph["1"]["inputs"]["text"] = "2048"
+        decision = _decide(_graph(graph))
+        self.assertEqual((decision.kind, decision.source), ("never", "junk:number"))
+        self.assertEqual(decision.unlinked_status, "found")
+        graph["1"]["inputs"]["text"] = ""
+        decision = _decide(_graph(graph))
+        self.assertEqual((decision.kind, decision.text), ("never", None))
+
+    def test_step_2_a_workflow_only_opt_out_comes_before_step_3(self):
+        workflow = {
+            "nodes": [{"id": 1, "type": "PromptManager", "widgets_values": [False]}]
+        }
+        decision = _decide({"workflow": json.dumps(workflow)})
+        self.assertEqual(decision.kind, "not_saved")
+
+    def test_step_5_an_opted_out_negative_stores_the_positive_alone(self):
+        graph = save_tests.run_graph(save=True, negative_save=False)
+        decision = _decide(_graph(graph))
+        self.assertEqual(decision.kind, "positive")
+        self.assertEqual(decision.source, "opted-out-negative")
+        self.assertEqual(decision.text, save_tests.POSITIVE)
+
+
+class TestResolvedSteps(unittest.TestCase):
+    def test_step_6_a_blank_positive_counts_as_unresolved(self):
+        meta = {
+            "prompt": json.dumps(issue_graph()),
+            "comfyassets_prompt_pairs": json.dumps(pair_evidence()),
+        }
+        decision = _decide(meta, policy=Policy.MONITOR, in_output=False)
+        self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.resolution["state"], "unresolved")
+
+    def test_step_7_a_resolved_pair(self):
+        lookup = _Lookup("9")
+        decision = _decide(_graph(sampled_graph("cat", "blur")), lookup=lookup)
+        self.assertEqual(decision.kind, "pair")
+        self.assertEqual(decision.source, "resolver")
+        self.assertEqual(decision.text, "cat")
+        self.assertEqual(decision.resolution["negative"]["text"], "blur")
+        self.assertIsNone(decision.review)
+        self.assertEqual(lookup.calls, 0, "decided before step 8: history not asked")
+
+    def test_step_7_a_blank_negative_stays_a_pair(self):
+        decision = _decide(_graph(sampled_graph("cat", "")))
+        self.assertEqual(decision.kind, "pair")
+        self.assertEqual(decision.resolution["negative"]["text"], "")
+
+
+class TestUndecidedGraph(unittest.TestCase):
+    def test_step_8_1_the_inspector_alone_decides(self):
+        decision = _decide({"prompt": parity.build("f03")["prompt"]})
+        self.assertEqual(decision.kind, "positive")
+        self.assertEqual(decision.source, "inspector")
+        self.assertEqual(decision.text, parity.POS)
+        self.assertEqual(decision.resolution["state"], "unresolved")
+
+    def test_step_8_1_the_recorded_saver_names_its_branch(self):
+        lookup = _Lookup("23")
+        decision = _decide(_graph(shapes.shared_prefix()), lookup=lookup)
+        self.assertEqual(decision.kind, "positive")
+        self.assertEqual(decision.source, "recorded-run")
+        self.assertEqual(decision.text, shapes.BRANCH_TEXTS[1])
+        self.assertEqual(decision.review, "ambiguous")
+        self.assertEqual(lookup.calls, 1)
+
+    def test_step_8_1_stores_the_storage_text_without_prepend_or_append(self):
+        graph = shapes.shared_prefix()
+        graph["20"]["inputs"]["text"] = ["25", 1]
+        graph["25"] = {
+            "class_type": "PromptManagerSplit",
+            "inputs": {
+                "text": shapes.BRANCH_TEXTS[1],
+                "prepend_text": "trigger",
+                "append_text": "detail",
+            },
+        }
+        decision = _decide(_graph(graph), lookup=_Lookup("23"))
+        self.assertEqual(decision.source, "recorded-run")
+        self.assertEqual(decision.text, shapes.BRANCH_TEXTS[1])
+
+    def test_step_8_1_a_file_name_naming_another_saver_anchors_nothing(self):
+        graph = shapes.branch_bench()
+        for index in (1, 3, 4):
+            graph[str(index * 10 + 3)]["inputs"]["filename_prefix"] = "bench/shared"
+        decision = _decide(
+            _graph(graph), lookup=_Lookup("43"), file_name="s45_branch_2_00001_.png"
+        )
+        # The name contradicts history's branch 4: no recorded anchor.
+        self.assertEqual(decision.source, "legacy-parser")
+
+    def test_step_8_1_the_file_name_anchor_alone_never_decides(self):
+        """#390: the file-name anchor never decides a link. The scan's legacy
+        floor may still link the file (as before #247), and a file nothing
+        links is queued with the anchor's choice per #390; the monitor
+        stores nothing."""
+        name = "s45_branch_2_00001_.png"
+        scan = _decide(_graph(shapes.branch_bench()), lookup=_Lookup(), file_name=name)
+        self.assertEqual((scan.kind, scan.source), ("positive", "legacy-parser"))
+        self.assertEqual(scan.review, "ambiguous")
+        monitor = _decide(
+            _graph(shapes.branch_bench()),
+            policy=Policy.MONITOR,
+            lookup=_Lookup(),
+            file_name=name,
+        )
+        self.assertEqual(monitor.kind, "never")
+        recorded = _decide(
+            _graph(shapes.branch_bench()), lookup=_Lookup("43"), file_name=name
+        )
+        # The recorded saver still decides: history says branch 4, but the
+        # file name names branch 2, which contradicts it.
+        self.assertNotEqual(recorded.source, "recorded-run")
+        recorded = _decide(_graph(shapes.branch_bench()), lookup=_Lookup("43"))
+        self.assertEqual(recorded.source, "recorded-run")
+        self.assertEqual(recorded.text, shapes.BRANCH_TEXTS[3])
+
+    def test_step_8_2_the_monitor_waits_for_history_inside_the_output_folder(self):
+        decision = _decide(_graph(shapes.shared_prefix()), policy=Policy.MONITOR)
+        self.assertEqual(decision.kind, "history")
+
+    def test_step_8_2_outside_the_output_folder_steps_8_3_and_8_4_apply(self):
+        decision = _decide(
+            _graph(shapes.shared_prefix()), policy=Policy.MONITOR, in_output=False
+        )
+        self.assertEqual(decision.kind, "never")
+        with _untraced_by_both():
+            decision = _decide(
+                _graph(_gated(sampled_graph())), policy=Policy.MONITOR, in_output=False
+            )
+        self.assertEqual(
+            (decision.kind, decision.source), ("positive", "legacy-parser")
+        )
+
+    def test_step_8_3_an_ambiguous_file_is_left_unlinked_by_the_monitor(self):
+        lookup = _Lookup("missing")
+        decision = _decide(
+            _graph(shapes.shared_prefix()), policy=Policy.MONITOR, lookup=lookup
+        )
+        self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertFalse(decision.try_recorded_run)
+        self.assertEqual(lookup.calls, 1)
+
+    def test_step_8_4_joined_prompt_manager_text_is_never_the_legacy_answer(self):
+        """The legacy parser returns prepend + text + append; the stored text
+        of a PromptManager node cannot be told from it."""
+        decision = _decide(_graph(_prompt_manager_branches()), lookup=_Lookup())
+        self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertEqual(decision.review, "ambiguous")
+        self.assertTrue(decision.try_recorded_run)
+        # A linked part the parser does not follow: its answer is the node's
+        # own text, which is what the node stores (#170 parity).
+        linked = _prompt_manager_branches(prepend=["99", 0], append="")
+        decision = _decide(_graph(linked), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.text), ("positive", "a cat"))
+
+    def test_step_8_4_prompt_manager_text_without_parts_keeps_the_floor(self):
+        """Regression guard (passes on the base): no prepend or append."""
+        graph = _prompt_manager_branches(prepend="", append=" ")
+        decision = _decide(_graph(graph), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.text), ("positive", "a cat"))
+
+    def test_step_8_4_an_ambiguous_file_gets_the_legacy_answer_in_the_scan(self):
+        decision = _decide(_graph(shapes.shared_prefix()), lookup=_Lookup())
+        self.assertEqual(decision.kind, "positive")
+        self.assertEqual(decision.source, "legacy-parser")
+        self.assertEqual(decision.text, shapes.BRANCH_TEXTS[0])
+        self.assertEqual(decision.review, "ambiguous")
+
+    def test_step_8_4_an_untraceable_file_gets_the_legacy_answer_in_both(self):
+        for policy in Policy:
+            for inspected in (_inspected("unresolved"), _inspected("x", True)):
+                with (
+                    self.subTest(policy=policy, malformed=inspected.malformed),
+                    mock.patch.object(
+                        file_prompt, "inspection_resolution", return_value=inspected
+                    ),
+                ):
+                    decision = _decide(
+                        _graph(_gated(sampled_graph("cat"))),
+                        policy=policy,
+                        lookup=_Lookup(),
+                    )
+                    self.assertEqual(decision.kind, "positive")
+                    self.assertEqual(decision.source, "legacy-parser")
+                    self.assertEqual(decision.text, "cat")
+
+    def test_step_8_4_nothing_traced_tries_the_recorded_run(self):
+        graph = sampled_graph()
+        graph["5"] = {**graph["3"]}
+        for sampler in ("3", "5"):
+            graph[sampler]["inputs"]["positive"] = ["9", 0]
+        graph["9"] = {"class_type": "BasicGuider", "inputs": {"width": 2048}}
+        decision = _decide(_graph(graph), lookup=_Lookup())
+        self.assertEqual(decision.kind, "never")
+        self.assertTrue(decision.try_recorded_run)
+        # Ambiguous without a choice is reported unresolved.
+        self.assertEqual(decision.unlinked_status, "unresolved")
+        self.assertEqual(decision.review, "untraced")
+
+
+class TestStepsWithoutAnUndecidedGraph(unittest.TestCase):
+    def test_step_9_ambiguous_parameters_only(self):
+        meta = {
+            "parameters": (
+                "cat\nNegative prompt: blur\nNegative prompt: noise"
+                "\nSteps: 20, Sampler: Euler"
+            )
+        }
+        decision = _decide(meta)
+        self.assertEqual(decision.kind, "never")
+        self.assertFalse(decision.try_recorded_run)
+        self.assertEqual(decision.unlinked_status, "unresolved")
+
+    def test_step_10_unresolved_without_a1111_text(self):
+        decision = _decide({"prompt": {}, "parameters": 5})
+        self.assertEqual(decision.kind, "never")
+        self.assertFalse(decision.try_recorded_run)
+        self.assertEqual(decision.unlinked_status, "unresolved")
+
+    def test_step_11_sampler_less_with_one_text(self):
+        decision = _decide(_graph(flux_graph()))
+        self.assertEqual((decision.kind, decision.source), ("positive", "sampler-less"))
+        self.assertEqual(decision.text, "a red fox")
+
+    def test_step_11_sampler_less_with_several_texts(self):
+        graph = flux_graph()
+        graph["50"] = {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {"model": ["1", 0], "clip": ["2", 0], "text": "a lora note"},
+        }
+        graph["22"]["inputs"]["model"] = ["50", 0]
+        graph["6"]["inputs"]["clip"] = ["50", 1]
+        decision = _decide(_graph(graph))
+        self.assertEqual(decision.kind, "never")
+        self.assertTrue(decision.try_recorded_run)
+        self.assertEqual(decision.unlinked_status, "unresolved")
+        meta = {**_graph(graph), "parameters": "a red fox\nNegative prompt: x"}
+        decision = _decide(meta)
+        self.assertEqual((decision.kind, decision.source), ("positive", "a1111"))
+        self.assertEqual(decision.text, "a red fox")
+
+    def test_step_11_sampler_less_without_a_text(self):
+        graph = {"40": {"class_type": "easy showAnything", "inputs": {"text": "x"}}}
+        decision = _decide(_graph(graph))
+        self.assertEqual(decision.kind, "never")
+        self.assertTrue(decision.try_recorded_run)
+        self.assertEqual(decision.unlinked_status, "found")
+
+    def test_step_12_a1111_only(self):
+        decision = _decide({"parameters": "a cat"})
+        self.assertEqual((decision.kind, decision.source), ("positive", "a1111"))
+        self.assertEqual(decision.text, "a cat")
+        decision = _decide({"parameters": "Negative prompt: blur"})
+        self.assertEqual((decision.kind, decision.unlinked_status), ("never", "found"))
+        self.assertFalse(decision.try_recorded_run)
+
+    def test_step_12_parsed_metadata_never_raises(self):
+        # The monitor hands parsed values: a dict prompt, non-string parameters.
+        decision = _decide({"prompt": {}, "parameters": "a cat"})
+        self.assertEqual(decision.text, "a cat")
+        decision = _decide({"parameters": 5})
+        self.assertEqual((decision.kind, decision.unlinked_status), ("never", "found"))
+
+
+class TestJunkGate(unittest.TestCase):
+    JUNK = {
+        "blank": "  ",
+        "number": "2048",
+        "lora-only": f"{LORA}, <lora:detail:0.5> |",
+        "model-file": "models/checkpoints/Flux-Dev.SafeTensors",
+    }
+
+    def test_each_junk_class_is_named(self):
+        for name, text in self.JUNK.items():
+            with self.subTest(name=name):
+                self.assertEqual(junk_text(text), name)
+        for ending in ("ckpt", "pt", "pth", "bin", "gguf", "sft"):
+            self.assertEqual(junk_text(f"model.{ending}"), "model-file")
+
+    def test_near_misses_are_not_junk(self):
+        for text in (
+            f"a red fox {LORA}",
+            "a.safetensors b.safetensors",
+            "nan",
+            "a portrait saved as final.safetensors",
+            "4k",
+            "C:\\My Models\\x.safetensors",
+        ):
+            with self.subTest(text=text):
+                self.assertIsNone(junk_text(text))
+
+    def test_step_13_junk_from_any_step_is_never_stored(self):
+        cases = {
+            "resolver": _graph(sampled_graph("2048", "blur")),
+            "sampler-less": _graph(flux_graph(text=LORA)),
+            "a1111": {"parameters": "model.safetensors"},
+        }
+        for step, meta in cases.items():
+            with self.subTest(step=step):
+                decision = _decide(meta)
+                self.assertEqual(decision.kind, "never")
+                self.assertTrue(decision.source.startswith("junk:"), decision.source)
+                self.assertTrue(decision.try_recorded_run)
+                self.assertEqual(decision.unlinked_status, "found")
+
+    def test_step_13_a_junk_legacy_answer_is_never_stored(self):
+        with _untraced_by_both():
+            decision = _decide(_graph(_gated(sampled_graph(LORA))), lookup=_Lookup())
+        self.assertEqual(decision.source, "junk:lora-only")
+        self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.unlinked_status, "unresolved")
+
+    def test_step_13_junk_at_step_8_keeps_that_steps_status_and_review(self):
+        # An ambiguous file whose legacy answer is a number is still a
+        # Needs review case: its status stays "ambiguous".
+        decision = _decide(_graph(_numeric_first_branches()), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.source), ("never", "junk:number"))
+        self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertEqual(decision.review, "ambiguous")
+        self.assertTrue(decision.try_recorded_run)
+
+
+class TestParseComfyuiPrompt(unittest.TestCase):
+    def test_a_parsed_prompt_and_non_string_values_are_kept(self):
+        parsed = file_prompt.parse_comfyui_prompt(
+            {"prompt": {"1": {}}, "workflow": {"nodes": []}, "steps": 20}
+        )
+        self.assertEqual(parsed["prompt"], {"1": {}})
+        self.assertEqual(parsed["workflow"], {"nodes": []})
+        self.assertEqual(parsed["parameters"]["steps"], 20)
+        self.assertIsNone(parsed["positive_prompt"])
+
+    def test_non_string_parameters_are_ignored(self):
+        parsed = file_prompt.parse_comfyui_prompt({"parameters": {"a": 1}})
+        self.assertIsNone(parsed["positive_prompt"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ingestion_resolver_profile.py

@@ -1,9 +1,11 @@
-"""Ingestion keeps the default resolver profile; only inspection routes moved (D1).
+"""Ingestion decides with the default resolver profile first (D1, #247 D-Q1).
 
 The live monitor and the admin scan write pairs and links to the user's
-database, so their resolver semantics stay byte-for-byte today's. f03 (506
-executable nodes) separates the profiles: the default profile's 500-node gate
-leaves it unresolved, while the inspection profile resolves it.
+database, so the default profile decides every file it can, byte-for-byte as
+before. f03 (506 executable nodes) separates the profiles: the default
+profile's 500-node gate leaves it unresolved, while the inspection profile
+resolves it. Only such an undecided file reaches the inspector's resolve
+(``inspection_resolution``), and only its positive is stored (#247).
 """
 
 import logging
@@ -68,24 +70,26 @@ class TestIngestionKeepsDefaultProfile(unittest.TestCase):
         spy.assert_default_and_gated(self)
         handler.db_manager.link_image_to_prompt_or_raise.assert_called_once()
 
-    def test_admin_scan_ingestion_uses_the_default_profile(self):
+    def test_admin_scan_ingestion_uses_the_default_profile_first(self):
         from py.api import admin
+        from utils import file_prompt
 
         mixin = admin.AdminRoutesMixin.__new__(admin.AdminRoutesMixin)
         mixin.db = Mock()
-        mixin.db.get_prompt_by_hash.return_value = None
-        mixin.db.save_prompt.return_value = 7
         spy = _Spy()
-        with patch.object(admin, "resolve_prompt_pair", spy):
+        with (
+            patch.object(file_prompt, "resolve_prompt_pair", spy),
+            patch.object(
+                admin, "store_scanned_prompt", return_value=("added", 7)
+            ) as store,
+        ):
             status = mixin._ingest_scanned_file(Path("/out/f03.png"), _f03_metadata())
         spy.assert_default_and_gated(self)
-        # The resolver leaves f03 unresolved; the scan links the prompt the
-        # browser parser traces, as the inspector shows it.
+        # The default profile leaves f03 unresolved; the inspector's resolve
+        # names the prompt the inspector shows, and the scan stores it.
         self.assertEqual(status, "added")
-        self.assertEqual(
-            mixin.db.save_prompt.call_args.args[0], "a red fox in snow, detailed fur"
-        )
-        mixin.db.link_image_to_prompt.assert_called_once()
+        self.assertEqual(store.call_args.args[3], "a red fox in snow, detailed fur")
+        mixin.db.link_image_to_prompt.assert_not_called()
 
 
 if __name__ == "__main__":

tests/test_pair_scan.py

@@ -1,9 +1,12 @@
 """Metadata scanning enriches known pairs and conservatively leaves conflicts."""
 
 import json
+import sqlite3
 from pathlib import Path
+from unittest.mock import patch
 
 from tests.test_api_admin import AdminAPITestCase
+from utils.hashing import generate_prompt_hash
 
 
 def graph(positive="cat", negative="blur"):
@@ -449,3 +452,159 @@ class TestSamplerLessScan(AdminAPITestCase):
         path = self.output_dir / "saver.png"
         self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
         self.assertEqual(self._texts(), ["a red fox"])
+
+
+class TestScanDecision(AdminAPITestCase):
+    """The scan applies the shared per-file decision (#247): one prompt per
+    file, no junk text, marks on every prompt it creates."""
+
+    def _path(self, name):
+        path = self.output_dir / name
+        path.write_bytes(b"png")
+        return path
+
+    def _links(self, path):
+        rows = self.api.db.model.get_connection().execute(
+            "SELECT prompt_id FROM generated_images WHERE image_path=?", (str(path),)
+        )
+        return [row[0] for row in rows]
+
+    def _prompt(self, text):
+        return self.api.db.get_prompt_by_hash(generate_prompt_hash(text))
+
+    def test_a_file_linked_to_another_prompt_is_a_conflict(self):
+        path = self._path("fox.png")
+        other = self.api.db.save_prompt(
+            "a dog", prompt_hash=generate_prompt_hash("a dog")
+        )
+        self.api.db.link_image_to_prompt(other, str(path))
+        meta = {"prompt": json.dumps(flux_graph())}
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "conflict")
+        self.assertEqual(self._links(path), [other])
+        self.assertIsNone(self._prompt("a red fox"))
+
+    def test_junk_text_is_never_stored(self):
+        cases = {
+            "lora.png": {"prompt": json.dumps(flux_graph(text="<lora:style:1.00>"))},
+            "model.png": {"parameters": "models/flux-dev.safetensors"},
+        }
+        for name, meta in cases.items():
+            with self.subTest(name=name):
+                path = self._path(name)
+                self.assertEqual(self.api._ingest_scanned_file(path, meta), "found")
+                self.assertEqual(self._links(path), [])
+        self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
+
+    def test_created_prompts_carry_the_scan_marks(self):
+        path = self._path("pair.png")
+        meta = {"prompt": json.dumps(sampled_graph("cat", "blur"))}
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
+        for text in ("cat", "blur"):
+            prompt = self._prompt(text)
+            self.assertEqual(prompt["category"], "scanned")
+            self.assertEqual(prompt["tags"], ["auto-scanned"])
+            self.assertEqual(prompt["notes"], "Auto-scanned from pair.png")
+
+    def test_outcome_of_a_positive_reports_its_title(self):
+        """Regression guard (passes on the base): the detailed positive row."""
+        path = self._path("fox.png")
+        meta = {"prompt": json.dumps(flux_graph())}
+        first = self.api._ingest_scanned_file(path, meta, detailed=True)
+        self.assertEqual(first, {"status": "added", "title": "a red fox"})
+        again = self.api._ingest_scanned_file(path, meta, detailed=True)
+        self.assertEqual(again["status"], "unchanged")
+        other = self._path("fox2.png")
+        linked = self.api._ingest_scanned_file(other, meta, detailed=True)
+        self.assertEqual(linked["status"], "linked")
+
+    def test_outcome_of_a_pair_reports_its_counts(self):
+        """Regression guard (passes on the base): the detailed pair row."""
+        path = self._path("pair.png")
+        meta = {"prompt": json.dumps(sampled_graph("cat", "blur"))}
+        result = self.api._ingest_scanned_file(path, meta, detailed=True)
+        self.assertEqual(result["status"], "added")
+        self.assertEqual((result["pairs_added"], result["links_added"]), (1, 1))
+        self.assertEqual(result["title"], "cat")
+
+    def test_outcome_without_a_store_reports_the_unlinked_status(self):
+        """Regression guard (passes on the base): never-stored rows."""
+        path = self._path("notes.png")
+        meta = {"prompt": json.dumps({"40": {"class_type": "Textbox", "inputs": {}}})}
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "found")
+        self.assertIsNone(self.api._ingest_scanned_file(path, {}))
+
+    def test_review_lists_get_the_default_profiles_ambiguous_files(self):
+        """Regression guard (passes on the base): Needs review inputs."""
+        g = sampled_graph()
+        g["6"] = {"class_type": "CLIPTextEncode", "inputs": {"text": "dog"}}
+        g["12"] = {**g["3"], "inputs": {**g["3"]["inputs"], "positive": ["6", 0]}}
+        ambiguous, untraced, records = [], [], {}
+        path = self._path("two.png")
+        status = self.api._ingest_scanned_file(
+            path,
+            {"prompt": json.dumps(g)},
+            ambiguous=ambiguous,
+            untraced=untraced,
+            records=records,
+        )
+        # Linked by the legacy floor, still listed (queued only if unlinked).
+        self.assertEqual(status, "added")
+        self.assertEqual((ambiguous, untraced), ([path], []))
+        self.assertEqual(records["canonical"]["prompt"], g)
+
+    def test_an_ambiguous_file_with_a_junk_answer_is_still_queued(self):
+        from tests.test_file_prompt import _numeric_first_branches
+
+        path = self._path("numbers.png")
+        meta = {"prompt": json.dumps(_numeric_first_branches())}
+        with patch.object(
+            type(self.api), "_record_scanned_review", autospec=True
+        ) as record:
+            counts = self.api._ingest_scan_batch_sync([(path, meta)])
+        self.assertEqual(counts["ambiguous"], 1, counts)
+        record.assert_called_once()
+        self.assertEqual(self._links(path), [])
+
+    def test_a_conflict_row_says_the_link_was_kept(self):
+        from py.api.admin import _scan_outcome
+
+        _group, row, _line = _scan_outcome(self._path("c.png"), [], "conflict")
+        self.assertEqual(row["detail"], "Kept: already linked to a different prompt")
+
+    def test_a_library_move_reruns_the_scan_writes(self):
+        from py.api import admin
+
+        real = admin.ingest_resolved_pair
+        model = self.api.db.model
+
+        def moved_once(*args, **kwargs):
+            if not calls:
+                calls.append(1)
+                model._generation += 1
+                raise sqlite3.OperationalError("closed")
+            return real(*args, **kwargs)
+
+        calls = []
+        path = self._path("pair.png")
+        with patch.object(admin, "ingest_resolved_pair", side_effect=moved_once):
+            meta = {"prompt": json.dumps(sampled_graph("cat", "blur"))}
+            self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
+        self.assertEqual(calls, [1])
+
+    def test_an_off_node_behind_a_pass_through_stores_nothing(self):
+        """#90 regression guard for the scan (main stored nothing here)."""
+        from tests.test_save_to_library import run_graph
+
+        g = run_graph(save=False)
+        g["1"]["class_type"] = "PromptManagerText"
+        g["7"] = {"class_type": "easy showAnything", "inputs": {"anything": ["1", 0]}}
+        g["8"] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": ["7", 0], "clip": ["4", 1]},
+        }
+        g["3"]["inputs"]["positive"] = ["8", 0]
+        path = self._path("off.png")
+        meta = {"prompt": json.dumps(g)}
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "not_saved")
+        self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
+        self.assertEqual(self._links(path), [])

tests/test_recorded_saver_anchor.py

@@ -375,5 +375,89 @@ class TestRecordedTexts(unittest.TestCase):
                 self.assertEqual(evidence["resolution"]["sampler_id"], sampler)
 
 
+class TestInspectionResolution(unittest.TestCase):
+    """``inspection_resolution``: the inspector's resolve and recorded anchor,
+    shared with ingestion (#247), so the two cannot drift."""
+
+    def _result(self, graph, saver=None, *, as_text=True, name="IMG_GEN_00371_.png"):
+        from utils.prompt_inspector import inspection_resolution
+
+        prompt = json.dumps(graph) if as_text else graph
+        return inspection_resolution({"prompt": prompt}, name, saver)
+
+    def test_the_recorded_saver_names_its_branch_by_record(self):
+        result = self._result(shapes.shared_prefix(), _saver(2))
+        self.assertTrue(result.by_record)
+        self.assertFalse(result.malformed)
+        self.assertEqual(result.resolution["state"], "resolved")
+        self.assertEqual(_positive(result.resolution), shapes.BRANCH_TEXTS[1])
+        self.assertIs(result.anchored, result.resolution)
+
+    def test_without_a_saver_nothing_is_anchored(self):
+        result = self._result(shapes.shared_prefix())
+        self.assertEqual(result.resolution["state"], "ambiguous")
+        self.assertIsNone(result.anchored)
+        self.assertFalse(result.by_record)
+
+    def test_a_saver_that_names_nothing_keeps_the_open_result(self):
+        result = self._result(shapes.shared_prefix(), "missing")
+        self.assertEqual(result.resolution["state"], "ambiguous")
+        self.assertIsNotNone(result.anchored)
+        self.assertFalse(result.by_record)
+
+    def test_a_file_that_resolves_alone_is_never_anchored(self):
+        result = self._result(
+            shapes.branch_bench(), _saver(4), name="s45_branch_2_00001_.png"
+        )
+        self.assertEqual(_positive(result.resolution), shapes.BRANCH_TEXTS[1])
+        self.assertIsNone(result.anchored)
+        self.assertFalse(result.by_record)
+
+    def test_parsed_metadata_resolves_as_the_file_text_does(self):
+        # The monitor hands parsed dicts; the scan hands the file's strings.
+        parsed = self._result(shapes.shared_prefix(), _saver(3), as_text=False)
+        text = self._result(shapes.shared_prefix(), _saver(3))
+        self.assertEqual(parsed.resolution, text.resolution)
+
+    def test_malformed_metadata_is_reported_as_malformed(self):
+        from utils.prompt_inspector import inspection_resolution
+
+        result = inspection_resolution({"prompt": "{not json"}, None, None)
+        self.assertTrue(result.malformed)
+        self.assertEqual(result.resolution["state"], "unresolved")
+        self.assertEqual(result.resolution["reason_code"], "malformed")
+
+    def test_metadata_without_a_role_record_is_absent(self):
+        from utils.prompt_inspector import inspection_resolution
+
+        result = inspection_resolution({"workflow": "{}"}, None, "9")
+        self.assertFalse(result.malformed)
+        self.assertEqual(result.resolution["state"], "absent")
+
+    def test_without_the_name_anchor_a_file_name_names_no_branch(self):
+        from utils.prompt_inspector import inspection_resolution
+
+        graph = {"prompt": json.dumps(shapes.branch_bench())}
+        name = "s45_branch_2_00001_.png"
+        named = inspection_resolution(graph, name, None)
+        self.assertEqual(_positive(named.resolution), shapes.BRANCH_TEXTS[1])
+        blind = inspection_resolution(graph, name, None, name_anchor=False)
+        self.assertEqual(blind.resolution["state"], "ambiguous")
+        # The recorded saver still decides, and the name still contradicts it.
+        recorded = inspection_resolution(graph, name, _saver(2), name_anchor=False)
+        self.assertTrue(recorded.by_record)
+        shared = {"prompt": json.dumps(shapes.shared_prefix())}
+        bench = shapes.branch_bench()
+        for index in (1, 3, 4):
+            bench[_saver(index)]["inputs"]["filename_prefix"] = "bench/shared"
+        contradicted = inspection_resolution(
+            {"prompt": json.dumps(bench)}, name, _saver(4), name_anchor=False
+        )
+        self.assertFalse(contradicted.by_record)
+        self.assertTrue(
+            inspection_resolution(shared, name, _saver(4), name_anchor=False).by_record
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_run_attribution.py

@@ -17,6 +17,8 @@ from database.run_attribution import (
 from tests import test_execution_history_database as history_tests
 from tests.background_stops import close_handler
 from tests.test_api_admin import AdminAPITestCase
+from tests import saver_anchor_graphs as shapes
+from tests.test_pair_scan import sampled_graph
 from tests.test_execution_context_database import known, snapshot
 from tests.test_prompt_attribution import issue_graph, llm_text_graph, make_tracker
 from utils.execution_contexts import (
@@ -326,7 +328,7 @@ class TestRecordedOutputNode(RecordedRuns, unittest.TestCase):
             self.assertIsNone(recorded_output_lookup(broken)(self.image))
             with (
                 mock.patch(
-                    "database.run_attribution._recorded_runs",
+                    "database.run_attribution.recorded_runs",
                     side_effect=sqlite3.OperationalError("locked"),
                 ),
                 self.assertLogs("prompt_manager.run_attribution", "WARNING"),
@@ -451,6 +453,86 @@ class TestScanUsesHistory(RecordedRuns, AdminAPITestCase):
         self.assertIsNone(self.linked_prompt(path))
         self.assertEqual(db.list_prompt_pairs()["pairs"], [])
 
+    def scan_dataset(self, names, nodes):
+        """Two branches of the shared-prefix dataset graph, one run, then a scan."""
+        paths = [self.write(name) for name in names]
+        self.record_run(
+            self.api.db,
+            str(self.output_dir),
+            [("", path.name) for path in paths],
+            [],
+            nodes=nodes,
+        )
+        meta = {"prompt": json.dumps(shapes.shared_prefix())}
+        return paths, self.api._ingest_scan_batch_sync([(p, meta) for p in paths])
+
+    def test_the_recorded_saver_links_each_dataset_file_to_its_branch(self):
+        """#247: before, the legacy parser gave every branch the first prompt."""
+        paths, counts = self.scan_dataset(
+            ["IMG_GEN_00001_.png", "IMG_GEN_00002_.png"], ["13", "23"]
+        )
+        self.assertEqual(counts["added"], 2, counts)
+        for path, text in zip(paths, shapes.BRANCH_TEXTS):
+            prompt = self.api.db.get_prompt_by_hash(generate_prompt_hash(text))
+            self.assertEqual(self.linked_prompt(path), prompt["id"])
+            self.assertEqual(prompt["category"], "scanned")
+
+    def test_a_file_the_legacy_answer_linked_is_left_as_it_is(self):
+        """#247 Q5: a rescan never adds a second link to the branch prompt."""
+        db = self.api.db
+        legacy = shapes.BRANCH_TEXTS[0]
+        legacy_id = db.save_prompt(legacy, prompt_hash=generate_prompt_hash(legacy))
+        path = self.write("IMG_GEN_00002_.png")
+        db.link_image_to_prompt(legacy_id, str(path))
+        self.record_run(db, str(self.output_dir), [("", path.name)], [], nodes=["23"])
+        meta = {"prompt": json.dumps(shapes.shared_prefix())}
+        self.assertEqual(self.api._ingest_scanned_file(path, meta), "conflict")
+        self.assertEqual(self.linked_prompt(path), legacy_id)
+        self.assertIsNone(
+            db.get_prompt_by_hash(generate_prompt_hash(shapes.BRANCH_TEXTS[1]))
+        )
+
+    def test_a_recorded_run_never_adds_a_second_link(self):
+        db = self.api.db
+        db.save_prompt(text=LLM_TEXT, prompt_hash=generate_prompt_hash(LLM_TEXT))
+        other = db.save_prompt(text="a dog", prompt_hash=generate_prompt_hash("a dog"))
+        path = self.write("bare_00003_.png")
+        db.link_image_to_prompt(other, str(path))
+        self.record_run(db, str(self.output_dir), [("", path.name)], [LLM_TEXT])
+        self.assertEqual(self.api._ingest_scanned_file(path, {}), "conflict")
+        rows = db.model.get_connection().execute(
+            "SELECT prompt_id FROM generated_images WHERE image_path=?", (str(path),)
+        )
+        self.assertEqual([row[0] for row in rows], [other])
+
+    def test_a_single_file_asks_history_once(self):
+        db = self.api.db
+        db.save_prompt(text=LLM_TEXT, prompt_hash=generate_prompt_hash(LLM_TEXT))
+        path = self.write("llm_00002_.png")
+        self.record_run(db, str(self.output_dir), [("", path.name)], [LLM_TEXT])
+        meta = {"prompt": json.dumps(llm_text_graph())}
+        statements = []
+        conn = db.model.get_connection()
+        conn.set_trace_callback(statements.append)
+        try:
+            self.assertEqual(self.api._ingest_scanned_file(path, meta), "linked")
+        finally:
+            conn.set_trace_callback(None)
+        lookups = [s for s in statements if "FROM execution_run_outputs" in s]
+        self.assertEqual(len(lookups), 1)
+
+    def test_a_file_the_recorded_saver_links_is_not_queued(self):
+        """A default-ambiguous file linked at the inspector's step is no
+        Needs review case."""
+        with mock.patch.object(
+            type(self.api), "_record_scanned_review", autospec=True
+        ) as record:
+            _paths, counts = self.scan_dataset(
+                ["IMG_GEN_00003_.png", "IMG_GEN_00004_.png"], ["13", "23"]
+            )
+        self.assertEqual(counts["added"], 2, counts)
+        record.assert_not_called()
+
 
 class TestHostHelpers(unittest.TestCase):
     """The host lookups degrade to "no proof", never to an error."""
@@ -752,13 +834,17 @@ class TestScanBatchLookup(RecordedRuns, AdminAPITestCase):
     """M5: a scan batch asks history once per chunk of names, not per file."""
 
     def test_a_batch_its_metadata_links_never_asks_history(self):
-        """R3: the batch lookup is made only when a file needs it."""
+        """R3: the batch lookup is made only when a file needs it.
+
+        Since #247 every file the default profile leaves undecided asks (the
+        recorded saver may name its branch), so these files resolve as pairs.
+        """
         files = []
         for n in range(3):
             path = self.output_dir / f"typed_{n:05}_.png"
             path.write_bytes(b"png")
             files.append(path)
-        meta = {"prompt": json.dumps(issue_graph(text="typed mountains"))}
+        meta = {"prompt": json.dumps(sampled_graph("typed mountains"))}
         statements = []
         conn = self.api.db.model.get_connection()
         conn.set_trace_callback(statements.append)
@@ -802,6 +888,72 @@ class TestScanBatchLookup(RecordedRuns, AdminAPITestCase):
         }
         self.assertEqual(linked, {f.name for f in files[:4]})
 
+    def test_the_batch_saver_equals_the_files_recorded_output(self):
+        from py.api.admin import _BatchHistory
+
+        db = self.api.db
+        names = ["one_node.png", "two_nodes.png", "two_runs.png", "unrecorded.png"]
+        files = []
+        for name in names:
+            path = self.output_dir / name
+            path.write_bytes(b"png")
+            files.append(path)
+        root = str(self.output_dir)
+        self.record_run(db, root, [("", "one_node.png")], [], nodes=["13"])
+        self.record_run(
+            db, root, [("", "two_nodes.png")] * 2, [], item=1, nodes=["13", "23"]
+        )
+        self.record_run(db, root, [("", "two_runs.png")], [], item=2)
+        self.record_run(db, root, [("", "two_runs.png")], [], item=3)
+        history = _BatchHistory(self.api, files)
+        with mock.patch(
+            "database.run_attribution.host_output_directory", return_value=root
+        ):
+            for path in files:
+                with self.subTest(path=path.name):
+                    expected = recorded_output(db, str(path), root)
+                    found = history.recorded(str(path))
+                    if expected is None:
+                        self.assertIsNone(found)
+                    else:
+                        self.assertEqual(found.saver_id, expected.saver_id)
+        self.assertEqual(history.recorded(str(files[0])).saver_id, "13")
+        self.assertIsNone(history.recorded(str(files[1])).saver_id)
+
+    def test_a_failed_batch_lookup_is_logged_and_names_nothing(self):
+        from py.api.admin import _BatchHistory
+
+        path = self.output_dir / "x.png"
+        history = _BatchHistory(self.api, [path])
+        with (
+            mock.patch(
+                "py.api.admin.recorded_runs", side_effect=sqlite3.OperationalError("x")
+            ),
+            self.assertLogs("prompt_manager", "WARNING"),
+        ):
+            self.assertIsNone(history.recorded(str(path)))
+        self.assertIsNone(history.get(str(path)))
+
+    def test_a_failed_prompt_lookup_is_logged_and_names_nothing(self):
+        from py.api.admin import _BatchHistory
+
+        path = self.output_dir / "one.png"
+        path.write_bytes(b"png")
+        root = str(self.output_dir)
+        self.record_run(self.api.db, root, [("", path.name)], [LLM_TEXT])
+        history = _BatchHistory(self.api, [path])
+        with (
+            mock.patch(
+                "database.run_attribution.host_output_directory", return_value=root
+            ),
+            mock.patch(
+                "py.api.admin.named_prompt", side_effect=sqlite3.OperationalError("x")
+            ),
+            self.assertLogs("prompt_manager", "WARNING"),
+        ):
+            self.assertIsNone(history.get(str(path)))
+            self.assertEqual(history.recorded(str(path)).saver_id, "9")
+
 
 class TestLinkedOutputRoot(RecordedRuns, unittest.TestCase):
     """R2: ComfyUI's output folder reached through a symlink or junction."""

tests/test_save_to_library.py

@@ -0,0 +1,508 @@
+"""#90 item 3: a node's ``save_to_library`` switch.
+
+Off: PromptManager ignores the run completely (no prompt saved, no run count,
+no image link), while the node still returns its normal outputs. On (the
+default, and every workflow or image made before the switch existed): as before.
+"""
+
+import logging
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from tests.test_prompt_attribution import MonitorCase, NodeCase
+    from tests.test_execution_contexts import load_module
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_prompt_attribution import MonitorCase, NodeCase
+    from test_execution_contexts import load_module
+
+from prompt_manager import PromptManager, PromptManagerFocused, PromptManagerSplit
+from prompt_manager_pair import PromptManagerPair
+from prompt_manager_text import PromptManagerText
+from prompt_search_list import PromptSearchList
+from utils.hashing import generate_prompt_hash
+from utils.prompt_graph import (
+    intrinsic_pair_nodes,
+    positive_opt_outs,
+    run_prompt_components,
+    run_prompt_nodes,
+    save_setting,
+)
+from utils.run_keys import prompt_node_key
+from utils.usage_tracking import (
+    PendingFirstUse,
+    QueuedRuns,
+    handle_queued_prompt,
+    on_dequeue,
+    queued_pair_runs,
+    queued_prompt_runs,
+)
+
+POSITIVE = "a lighthouse at dusk"
+NEGATIVE = "blurry"
+NODE_CLASSES = (
+    PromptManager,
+    PromptManagerText,
+    PromptManagerFocused,
+    PromptManagerSplit,
+    PromptManagerPair,
+)
+
+
+def run_graph(save=None, kind="PromptManager", negative_save=None):
+    """A PromptManager positive feeding a KSampler; ``save`` None leaves it out."""
+    inputs = {"text": POSITIVE, "clip": ["4", 1]}
+    if save is not None:
+        inputs["save_to_library"] = save
+    negative_inputs = {"text": NEGATIVE, "clip": ["4", 1]}
+    if negative_save is not None:
+        negative_inputs["save_to_library"] = negative_save
+    return {
+        "1": {"class_type": kind, "inputs": inputs},
+        "2": {"class_type": "PromptManager", "inputs": negative_inputs},
+        "4": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
+        "3": {
+            "class_type": "KSampler",
+            "inputs": {
+                "model": ["4", 0],
+                "positive": ["1", 0],
+                "negative": ["2", 0],
+                "latent_image": ["6", 0],
+            },
+        },
+        "6": {"class_type": "EmptyLatentImage", "inputs": {"batch_size": 1}},
+    }
+
+
+def pair_graph(save=None):
+    inputs = {"text": POSITIVE, "negative_text": NEGATIVE, "clip": ["4", 1]}
+    if save is not None:
+        inputs["save_to_library"] = save
+    return {
+        "1": {"class_type": "PromptManagerPair", "inputs": inputs},
+        "4": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
+        "3": {
+            "class_type": "KSampler",
+            "inputs": {
+                "model": ["4", 0],
+                "positive": ["1", 0],
+                "negative": ["1", 1],
+                "latent_image": ["6", 0],
+            },
+        },
+        "6": {"class_type": "EmptyLatentImage", "inputs": {"batch_size": 1}},
+    }
+
+
+class TestInputDeclaration(unittest.TestCase):
+    def test_every_prompt_manager_node_declares_a_boolean_defaulting_on(self):
+        for cls in NODE_CLASSES:
+            optional = cls.INPUT_TYPES()["optional"]
+            kind, options = optional["save_to_library"]
+            self.assertEqual(kind, "BOOLEAN", cls.__name__)
+            self.assertIs(options["default"], True, cls.__name__)
+            # Declared last: no saved widget value of an older workflow shifts
+            self.assertEqual(list(optional)[-1], "save_to_library", cls.__name__)
+
+    def test_search_list_has_no_switch(self):
+        inputs = PromptSearchList.INPUT_TYPES()
+        self.assertNotIn("save_to_library", inputs.get("optional", {}))
+        self.assertNotIn("save_to_library", inputs.get("required", {}))
+
+
+class TestSaveSetting(unittest.TestCase):
+    def test_missing_input_is_on(self):
+        self.assertIs(save_setting(run_graph(), "1"), True)
+
+    def test_literals_follow_comfyui_boolean_coercion(self):
+        for value, expected in (
+            (True, True),
+            (False, False),
+            ("", False),
+            (0, False),
+            ({"__value__": False}, False),
+            ({"__value__": True}, True),
+        ):
+            with self.subTest(value=value):
+                self.assertIs(save_setting(run_graph(save=value), "1"), expected)
+
+    def test_link_to_primitive_boolean_is_read(self):
+        for value in (True, False):
+            graph = run_graph(save=["9", 0])
+            graph["9"] = {"class_type": "PrimitiveBoolean", "inputs": {"value": value}}
+            self.assertIs(save_setting(graph, "1"), value)
+
+    def test_link_to_an_unknown_node_is_not_knowable(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        self.assertIsNone(save_setting(graph, "1"))
+        self.assertIsNone(save_setting(run_graph(save=["missing", 0]), "1"))
+
+
+class TestGraphRoles(unittest.TestCase):
+    def test_opted_out_positive_is_not_a_run_prompt(self):
+        graph = run_graph(save=False)
+        self.assertEqual(run_prompt_nodes(graph), [])
+        self.assertEqual(run_prompt_components(graph), [])
+        self.assertEqual(positive_opt_outs(graph), ["1"])
+
+    def test_enabled_positive_is_unchanged(self):
+        for save in (None, True):
+            graph = run_graph(save=save)
+            self.assertEqual(run_prompt_nodes(graph), ["1"])
+            self.assertEqual(positive_opt_outs(graph), [])
+
+    def test_opted_out_negative_is_no_positive_opt_out(self):
+        graph = run_graph(negative_save=False)
+        self.assertEqual(run_prompt_nodes(graph), ["1"])
+        self.assertEqual(positive_opt_outs(graph), [])
+
+    def test_walk_stops_at_an_opted_out_node(self):
+        graph = run_graph(save=False)
+        graph["1"]["inputs"]["text"] = ["7", 0]
+        graph["7"] = {"class_type": "PromptManagerText", "inputs": {"text": "x"}}
+        self.assertEqual(run_prompt_nodes(graph), [])
+
+    def test_sampler_less_fallback_skips_opted_out_nodes(self):
+        graph = {
+            "1": {
+                "class_type": "PromptManagerText",
+                "inputs": {"text": "a", "save_to_library": False},
+            },
+            "2": {"class_type": "PromptManagerText", "inputs": {"text": "b"}},
+        }
+        self.assertEqual(run_prompt_nodes(graph), ["2"])
+        self.assertEqual(run_prompt_components(graph), [("2", "text")])
+        self.assertEqual(positive_opt_outs(graph), ["1"])
+
+    def test_opted_out_pair_is_not_an_intrinsic_pair(self):
+        self.assertEqual(intrinsic_pair_nodes(pair_graph()), ["1"])
+        self.assertEqual(intrinsic_pair_nodes(pair_graph(save=False)), [])
+
+
+class TestQueueHook(unittest.TestCase):
+    def test_opted_out_node_is_never_stashed(self):
+        self.assertEqual(queued_prompt_runs(run_graph(save=False)), [])
+        self.assertEqual(queued_pair_runs(pair_graph(save=False)), [])
+
+    def test_unknown_setting_is_left_to_the_node(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        self.assertEqual(queued_prompt_runs(graph), [])
+        pair = pair_graph(save=["9", 0])
+        pair["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        self.assertEqual(queued_pair_runs(pair), [])
+
+    def test_primitive_boolean_on_still_counts(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "PrimitiveBoolean", "inputs": {"value": True}}
+        self.assertEqual(
+            queued_prompt_runs(graph), [(generate_prompt_hash(POSITIVE), POSITIVE)]
+        )
+
+    def test_dequeue_counts_nothing_for_a_cached_opted_out_node(self):
+        """A cached node never executes: only the queued graph can say it is off."""
+        db = mock.Mock()
+        db.get_prompt_by_hash.return_value = {"id": 5}
+        queued, pending = QueuedRuns(), PendingFirstUse()
+        request = {"prompt": run_graph(save=False), "prompt_id": "job-1"}
+        handle_queued_prompt(request, queued, pair_queued=QueuedRuns())
+        on_dequeue("job-1", queued, lambda: db, pending, pair_queued=QueuedRuns())
+        db.record_prompt_use.assert_not_called()
+        self.assertEqual(pending.consume(generate_prompt_hash(POSITIVE)), 0)
+
+
+class TestNodesSaveNothing(NodeCase):
+    def run_node(self, cls, graph, save):
+        node = self.node(cls)
+        node.record_pair = mock.Mock()
+        kwargs = {"prompt": graph, "unique_id": "1", "save_to_library": save}
+        if cls is PromptManagerText:
+            return node, node.process_text(POSITIVE, **kwargs)
+        if cls is PromptManagerPair:
+            return node, node.encode_prompt(
+                self.clip(), POSITIVE, negative_text=NEGATIVE, **kwargs
+            )
+        return node, node.encode_prompt(self.clip(), POSITIVE, **kwargs)
+
+    def test_off_saves_counts_and_tracks_nothing_but_returns_outputs(self):
+        for cls in NODE_CLASSES:
+            with self.subTest(node=cls.__name__):
+                graph = (
+                    pair_graph(save=False)
+                    if cls is PromptManagerPair
+                    else run_graph(save=False, kind=cls.__name__)
+                )
+                node, result = self.run_node(cls, graph, False)
+                self.assertEqual(self.count_rows("prompts"), 0)
+                self.assertEqual(self.count_rows("prompt_pairs"), 0)
+                self.assertEqual(self.tracker.keyed_snapshot(), {})
+                self.assertEqual(self.tracker._prompt_queue, [])
+                node.comfyui_integration.register_prompt.assert_not_called()
+                if cls is PromptManagerText:
+                    self.assertEqual(result, (POSITIVE,))
+                elif cls is PromptManagerPair:
+                    self.assertEqual(result[0], ("cond", POSITIVE))
+                    self.assertEqual(result[1], ("cond", NEGATIVE))
+                    self.assertEqual(result[2:], (POSITIVE, NEGATIVE))
+                else:
+                    self.assertEqual(result, (("cond", POSITIVE), POSITIVE))
+
+    def test_off_without_graph_context_still_saves_nothing(self):
+        node = self.node(PromptManager)
+        node.encode_prompt(self.clip(), POSITIVE, save_to_library=False)
+        self.assertEqual(self.count_rows("prompts"), 0)
+        self.assertEqual(self.tracker._prompt_queue, [])
+
+    def test_off_leaves_an_existing_prompt_uncounted(self):
+        prompt_id = self.save(POSITIVE)
+        self.run_node(PromptManager, run_graph(save=False), False)
+        row = self.db.get_prompt_by_id(prompt_id)
+        self.assertFalse(row.get("run_count") or 0)
+
+    def test_on_is_unchanged(self):
+        self.run_node(PromptManager, run_graph(save=True), True)
+        self.assertIsNotNone(self.db.get_prompt_by_hash(generate_prompt_hash(POSITIVE)))
+        self.assertIn(
+            prompt_node_key(run_graph(save=True), "1"), self.tracker.keyed_snapshot()
+        )
+
+    def test_unknown_setting_on_at_run_time_counts_at_execution(self):
+        """The hook could not read a wired switch, so the node counts its run."""
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        prompt_id = self.save(POSITIVE)
+        self.run_node(PromptManager, graph, True)
+        self.assertEqual(self.db.get_prompt_by_id(prompt_id)["run_count"], 1)
+
+
+class TestWiredSwitchOffMarksItsRun(NodeCase):
+    """Review finding: a wired switch off at run time keeps the node a run prompt."""
+
+    def wired_graph(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        return graph
+
+    def test_an_earlier_entry_of_the_key_never_stands_in(self):
+        graph = self.wired_graph()
+        key = prompt_node_key(graph, "1")
+        node = self.node(PromptManager)
+        node.encode_prompt(
+            self.clip(), POSITIVE, prompt=graph, unique_id="1", save_to_library=True
+        )
+        node.encode_prompt(
+            self.clip(), POSITIVE, prompt=graph, unique_id="1", save_to_library=False
+        )
+        latest, _run, _count = self.tracker.keyed_snapshot()[key]
+        self.assertTrue(latest.get("placeholder"))
+
+    def test_literal_off_marks_nothing(self):
+        node = self.node(PromptManager)
+        node.encode_prompt(
+            self.clip(),
+            POSITIVE,
+            prompt=run_graph(save=False),
+            unique_id="1",
+            save_to_library=False,
+        )
+        self.assertEqual(self.tracker.keyed_snapshot(), {})
+
+
+class TestMonitorLeavesOptedOutFilesUnlinked(MonitorCase):
+    def test_existing_prompt_is_not_linked_from_metadata(self):
+        self.save(POSITIVE)
+        self.assertIsNone(self.process({"prompt": run_graph(save=False)}))
+
+    def test_find_prompt_reports_opted_out(self):
+        self.save(POSITIVE)
+        state, prompt = self.handler._find_prompt_from_metadata(
+            {"prompt": run_graph(save=False)}
+        )
+        self.assertEqual((state, prompt), ("opted_out", None))
+
+    def test_mixed_positive_settings_leave_the_file_unlinked(self):
+        self.save(POSITIVE)
+        graph = run_graph(save=True)
+        graph["8"] = {
+            "class_type": "PromptManager",
+            "inputs": {"text": "other", "save_to_library": False},
+        }
+        graph["5"] = {
+            "class_type": "KSampler",
+            "inputs": {
+                "model": ["4", 0],
+                "positive": ["8", 0],
+                "negative": ["2", 0],
+                "latent_image": ["6", 0],
+            },
+        }
+        self.assertIsNone(self.process({"prompt": graph}))
+
+    def test_workflow_only_metadata_with_switch_off_is_opted_out(self):
+        self.save(POSITIVE)
+        metadata = {
+            "text_encoder_nodes": [
+                {
+                    "type": "PromptManager",
+                    "inputs": [],
+                    "widgets_values": [POSITIVE, "", "", "", "", "", False],
+                }
+            ]
+        }
+        state, _prompt = self.handler._find_prompt_from_metadata(metadata)
+        self.assertEqual(state, "opted_out")
+
+    def test_workflow_only_ignores_muted_nodes_and_reads_subgraphs(self):
+        from utils.prompt_graph import metadata_opted_out
+
+        node = {
+            "type": "PromptManager",
+            "widgets_values": [POSITIVE, "", "", "", "", "", False],
+        }
+        self.assertFalse(
+            metadata_opted_out({"workflow": {"nodes": [{**node, "mode": 4}]}})
+        )
+        self.assertTrue(
+            metadata_opted_out(
+                {
+                    "workflow": {
+                        "nodes": [],
+                        "definitions": {"subgraphs": [{"nodes": [node]}]},
+                    }
+                }
+            )
+        )
+
+    def test_unknown_setting_is_left_to_the_run(self):
+        self.save(POSITIVE)
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        state, _prompt = self.handler._find_prompt_from_metadata({"prompt": graph})
+        self.assertEqual(state, "unknown")
+
+    def test_enabled_positive_still_links(self):
+        prompt_id = self.save(POSITIVE)
+        self.assertEqual(self.process({"prompt": run_graph(save=True)}), prompt_id)
+
+
+class TestLinkKeepsOptedOutNegativeOut(MonitorCase):
+    def setUp(self):
+        super().setUp()
+        # The real link, not the recording stub of MonitorCase
+        del self.handler.link_image_to_prompt
+
+    def test_opted_out_negative_is_never_stored_as_a_pair(self):
+        prompt_id = self.save(POSITIVE)
+        metadata = {"prompt": run_graph(save=True, negative_save=False)}
+        self.handler.link_image_to_prompt(self.image, {"id": prompt_id}, metadata)
+        self.assertIsNone(self.db.get_prompt_by_hash(generate_prompt_hash(NEGATIVE)))
+        self.assertEqual(self.count_pairs(), 0)
+        linked = (
+            self.db.model.get_connection()
+            .execute("SELECT prompt_id FROM generated_images")
+            .fetchall()
+        )
+        self.assertEqual([row[0] for row in linked], [prompt_id])
+
+    def test_opted_out_positive_links_nothing(self):
+        prompt_id = self.save(POSITIVE)
+        metadata = {"prompt": run_graph(save=False)}
+        self.handler.link_image_to_prompt(self.image, {"id": prompt_id}, metadata)
+        count = (
+            self.db.model.get_connection()
+            .execute("SELECT COUNT(*) FROM generated_images")
+            .fetchone()[0]
+        )
+        self.assertEqual(count, 0)
+
+    def count_pairs(self):
+        conn = self.db.model.get_connection()
+        return conn.execute("SELECT COUNT(*) FROM prompt_pairs").fetchone()[0]
+
+
+class TestScanSkipsOptedOutFiles(NodeCase):
+    def setUp(self):
+        super().setUp()
+        from py.api import admin
+
+        self.mixin = admin.AdminRoutesMixin.__new__(admin.AdminRoutesMixin)
+        self.mixin.db = self.db
+        self.mixin.logger = logging.getLogger("test.save_to_library.scan")
+        self.file = Path(self.tmp.name) / "out.png"
+
+    def test_opted_out_file_stores_nothing(self):
+        status = self.mixin._ingest_scanned_file(
+            self.file, {"prompt": run_graph(save=False)}
+        )
+        self.assertEqual(status, "not_saved")
+        self.assertEqual(self.count_rows("prompts"), 0)
+        self.assertEqual(self.count_rows("generated_images"), 0)
+
+    def test_opted_out_file_is_never_queued_for_review(self):
+        ambiguous = []
+        graph = run_graph(save=False)
+        graph["5"] = dict(graph["3"])  # a second sampler: ambiguous otherwise
+        graph["5"]["inputs"] = {**graph["3"]["inputs"], "positive": ["2", 0]}
+        status = self.mixin._ingest_scanned_file(
+            self.file, {"prompt": graph}, ambiguous=ambiguous
+        )
+        self.assertEqual(status, "not_saved")
+        self.assertEqual(ambiguous, [])
+
+    def test_opted_out_negative_stores_the_positive_only(self):
+        status = self.mixin._ingest_scanned_file(
+            self.file, {"prompt": run_graph(save=True, negative_save=False)}
+        )
+        self.assertEqual(status, "added")
+        self.assertIsNotNone(self.db.get_prompt_by_hash(generate_prompt_hash(POSITIVE)))
+        self.assertIsNone(self.db.get_prompt_by_hash(generate_prompt_hash(NEGATIVE)))
+        self.assertEqual(self.count_rows("prompt_pairs"), 0)
+
+    def test_wired_switch_is_left_to_the_live_run(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        status = self.mixin._ingest_scanned_file(self.file, {"prompt": graph})
+        self.assertEqual(status, "not_saved_wired")
+        self.assertEqual(self.count_rows("prompts"), 0)
+
+    def test_status_has_a_scan_detail(self):
+        from py.api import admin
+
+        group, row, _line = admin._scan_outcome(self.file, [], "not_saved")
+        self.assertEqual(group, "no_metadata")
+        self.assertIn("does not save", row["detail"])
+
+
+class TestHistoryRecordsNoText(unittest.TestCase):
+    def setUp(self):
+        self.m = load_module("execution_context_graph")
+
+    def test_opted_out_role_has_no_sources_and_no_text(self):
+        recipe = self.m.project_contexts(run_graph(save=False), None).recipes[0]
+        self.assertEqual(recipe.positive.sources, ())
+        self.assertIsNone(recipe.positive.inferred)
+        self.assertEqual(recipe.positive.reason, "not_saved")
+        # The enabled negative is recorded as before
+        self.assertEqual(recipe.negative.inferred.raw.value, NEGATIVE)
+
+    def test_unknown_setting_records_no_text_either(self):
+        graph = run_graph(save=["9", 0])
+        graph["9"] = {"class_type": "SomeCustomSwitch", "inputs": {}}
+        recipe = self.m.project_contexts(graph, None).recipes[0]
+        self.assertIsNone(recipe.positive.inferred)
+        self.assertEqual(recipe.positive.reason, "not_saved")
+
+    def test_enabled_role_is_unchanged(self):
+        recipe = self.m.project_contexts(run_graph(save=True), None).recipes[0]
+        self.assertEqual(recipe.positive.inferred.raw.value, POSITIVE)
+        self.assertEqual(len(recipe.positive.sources), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_scanned_prompts.py

@@ -0,0 +1,274 @@
+"""Storing a file's prompt from its text: one transaction, one prompt each (#247)."""
+
+import os
+import sqlite3
+import tempfile
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest import mock
+
+try:
+    from tests.open_handles import assert_closed
+    from tests import test_execution_history_database as history_tests
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed
+    import test_execution_history_database as history_tests
+
+from database.operations import PromptDatabase
+from database.pair_ingestion import (
+    RUN_MARKS,
+    SCAN_MARKS,
+    Marks,
+    ingest_resolved_pair,
+)
+from database.scanned_prompts import store_scanned_prompt, store_scanned_prompt_or_raise
+from utils.hashing import generate_prompt_hash
+
+META = {"prompt": {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "x"}}}}
+
+
+def resolved(positive="cat", negative="blur"):
+    return {
+        "state": "resolved",
+        "positive": {"text": positive, "provenance": {"source": "api"}},
+        "negative": {"text": negative, "provenance": {"source": "api"}},
+    }
+
+
+class StoreCase(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.db = PromptDatabase(str(Path(self.temp.name) / "test.db"))
+        self.path = str(Path(self.temp.name) / "gen.png")
+
+    def tearDown(self):
+        self.db.close_all()
+        assert_closed(self, self.temp.name)
+        self.temp.cleanup()
+
+    def prompts(self):
+        rows = self.db.model.get_connection().execute(
+            "SELECT id, text FROM prompts ORDER BY id"
+        )
+        return [tuple(row) for row in rows]
+
+    def links(self):
+        rows = self.db.model.get_connection().execute(
+            "SELECT prompt_id, image_path FROM generated_images ORDER BY id"
+        )
+        return [tuple(row) for row in rows]
+
+    def save(self, text, **fields):
+        return self.db.save_prompt(
+            text, prompt_hash=generate_prompt_hash(text), **fields
+        )
+
+
+class TestStoreScannedPrompt(StoreCase):
+    def test_a_new_text_is_created_with_the_scan_marks_and_linked(self):
+        status, prompt_id = store_scanned_prompt_or_raise(
+            self.db, self.path, META, " a cat ", SCAN_MARKS
+        )
+        self.assertEqual(status, "added")
+        prompt = self.db.get_prompt_by_id(prompt_id)
+        self.assertEqual(prompt["text"], "a cat")
+        self.assertEqual(prompt["category"], "scanned")
+        self.assertEqual(prompt["tags"], ["auto-scanned"])
+        self.assertEqual(prompt["notes"], "Auto-scanned from gen.png")
+        self.assertEqual(prompt["run_count"], 0)
+        self.assertEqual(self.links(), [(prompt_id, self.path)])
+
+    def test_a_run_created_prompt_names_the_workflow_run(self):
+        _status, prompt_id = store_scanned_prompt_or_raise(
+            self.db, self.path, META, "a cat", RUN_MARKS
+        )
+        prompt = self.db.get_prompt_by_id(prompt_id)
+        self.assertEqual(prompt["notes"], "Saved from a workflow run: gen.png")
+        self.assertEqual(prompt["tags"], ["auto-scanned"])
+
+    def test_a_stored_text_is_reused_by_hash_and_keeps_its_fields(self):
+        existing = self.save("A Cat", category="mine", notes="kept")
+        status, prompt_id = store_scanned_prompt_or_raise(
+            self.db, self.path, META, "a cat ", SCAN_MARKS
+        )
+        self.assertEqual((status, prompt_id), ("linked", existing))
+        prompt = self.db.get_prompt_by_id(existing)
+        self.assertEqual((prompt["category"], prompt["notes"]), ("mine", "kept"))
+        self.assertEqual(prompt["tags"], [])
+        self.assertEqual(len(self.prompts()), 1)
+
+    def test_a_repeat_is_unchanged(self):
+        first = store_scanned_prompt_or_raise(
+            self.db, self.path, META, "a cat", SCAN_MARKS
+        )
+        again = store_scanned_prompt_or_raise(
+            self.db, self.path, META, "a cat", SCAN_MARKS
+        )
+        self.assertEqual(again, ("unchanged", first[1]))
+        self.assertEqual(len(self.links()), 1)
+
+    def test_a_file_linked_to_another_prompt_is_a_conflict(self):
+        other = self.save("a dog")
+        # Linked under another spelling of the same file.
+        spelled = os.path.join(self.temp.name, ".", "gen.png")
+        self.db.link_image_to_prompt(other, spelled)
+        result = store_scanned_prompt_or_raise(
+            self.db, self.path, META, "a cat", SCAN_MARKS
+        )
+        self.assertEqual(result, ("conflict", None))
+        self.assertEqual(self.prompts(), [(other, "a dog")])
+        self.assertEqual(self.links(), [(other, spelled)])
+
+    def test_a_failed_link_raises_and_leaves_no_prompt_behind(self):
+        with mock.patch.object(
+            self.db, "_insert_image_link", side_effect=sqlite3.OperationalError("x")
+        ):
+            with self.assertRaises(sqlite3.OperationalError):
+                store_scanned_prompt_or_raise(
+                    self.db, self.path, META, "a cat", SCAN_MARKS
+                )
+            with self.assertLogs("prompt_manager", "ERROR"):
+                result = store_scanned_prompt(
+                    self.db, self.path, META, "a cat", SCAN_MARKS
+                )
+        self.assertEqual(result, ("found", None))
+        self.assertEqual(self.prompts(), [])
+        self.assertEqual(self.links(), [])
+
+    def test_two_concurrent_writers_store_one_prompt_and_one_link(self):
+        hold = history_tests.FirstWriterHold()
+        outcome = self.db._pair_component_outcome
+        started = threading.Barrier(2)
+
+        def held_open(*args, **kwargs):
+            hold()
+            return outcome(*args, **kwargs)
+
+        def write(_index):
+            started.wait(timeout=30)
+            try:
+                return store_scanned_prompt_or_raise(
+                    self.db, self.path, META, "a cat", SCAN_MARKS
+                )[0]
+            finally:
+                self.db.model.close()
+
+        with (
+            mock.patch.object(
+                self.db, "_pair_component_outcome", side_effect=held_open
+            ),
+            ThreadPoolExecutor(max_workers=2) as pool,
+        ):
+            statuses = sorted(pool.map(write, range(2)))
+        self.assertEqual(statuses, ["added", "unchanged"])
+        self.assertEqual(len(self.prompts()), 1)
+        self.assertEqual(len(self.links()), 1)
+
+    def test_a_library_move_reruns_the_store(self):
+        from database import scanned_prompts
+
+        real = scanned_prompts.store_scanned_prompt_or_raise
+        calls = []
+
+        def moved_once(*args, **kwargs):
+            if not calls:
+                calls.append(1)
+                self.db.model._generation += 1
+                raise sqlite3.OperationalError("closed")
+            return real(*args, **kwargs)
+
+        with mock.patch.object(
+            scanned_prompts, "store_scanned_prompt_or_raise", side_effect=moved_once
+        ):
+            status, _id = store_scanned_prompt(
+                self.db, self.path, META, "a cat", SCAN_MARKS
+            )
+        self.assertEqual((status, calls), ("added", [1]))
+
+    def _linked_folder(self):
+        """``(out_real, out_link)``: a symlink to the same output folder."""
+        from tests.test_api_duplicate_files import TestReview
+
+        if not TestReview._can_symlink(self.temp.name):  # pragma: no cover
+            self.skipTest("symlinks are not available")
+        real = Path(self.temp.name) / "out_real"
+        real.mkdir()
+        link = Path(self.temp.name) / "out_link"
+        os.symlink(str(real), str(link))
+        (real / "a.png").write_bytes(b"png")
+        return os.path.realpath(str(real)), str(link)
+
+    def test_a_file_linked_through_its_real_folder_is_found_through_a_link(self):
+        """The scan stores realpath spellings, the monitor the watcher's."""
+        real, link = self._linked_folder()
+        other = self.save("a dog")
+        self.db.link_image_to_prompt(other, os.path.join(real, "a.png"))
+        through_link = os.path.join(link, "a.png")
+        self.assertEqual(
+            store_scanned_prompt_or_raise(
+                self.db, through_link, META, "a cat", SCAN_MARKS
+            ),
+            ("conflict", None),
+        )
+        self.assertEqual(
+            store_scanned_prompt_or_raise(
+                self.db, through_link, META, "a dog", SCAN_MARKS
+            ),
+            ("unchanged", other),
+        )
+        pair = ingest_resolved_pair(self.db, through_link, META, resolved())
+        self.assertEqual(pair["status"], "conflict")
+        self.assertEqual(len(self.links()), 1)
+
+
+class TestPairMarks(StoreCase):
+    def test_marks_reach_only_created_components(self):
+        kept = self.save("cat", category="mine")
+        result = ingest_resolved_pair(
+            self.db, self.path, META, resolved(), marks=RUN_MARKS
+        )
+        self.assertEqual(result["status"], "added")
+        positive = self.db.get_prompt_by_id(kept)
+        self.assertEqual((positive["category"], positive["tags"]), ("mine", []))
+        negative_id = next(pid for pid, text in self.prompts() if text == "blur")
+        negative = self.db.get_prompt_by_id(negative_id)
+        self.assertEqual(negative["category"], "scanned")
+        self.assertEqual(negative["tags"], ["auto-scanned"])
+        self.assertEqual(negative["notes"], "Saved from a workflow run: gen.png")
+
+    def test_without_marks_created_components_stay_bare(self):
+        """Regression guard (passes on the base): the node path is unchanged."""
+        ingest_resolved_pair(self.db, self.path, META, resolved())
+        for prompt_id, _text in self.prompts():
+            prompt = self.db.get_prompt_by_id(prompt_id)
+            self.assertIsNone(prompt["category"])
+            self.assertEqual(prompt["tags"], [])
+
+    def test_a_blank_negative_creates_only_the_positive(self):
+        result = ingest_resolved_pair(
+            self.db, self.path, META, resolved(negative=""), marks=SCAN_MARKS
+        )
+        self.assertEqual(result["status"], "added")
+        self.assertEqual([text for _id, text in self.prompts()], ["cat"])
+
+    def test_marks_are_a_frozen_value(self):
+        self.assertEqual(
+            SCAN_MARKS, Marks("scanned", ("auto-scanned",), SCAN_MARKS.note)
+        )
+        with self.assertRaises(AttributeError):
+            SCAN_MARKS.category = "x"
+
+
+class TestPublicRunAttribution(unittest.TestCase):
+    def test_recorded_runs_and_named_prompt_are_public(self):
+        from database import run_attribution
+
+        self.assertTrue(callable(run_attribution.recorded_runs))
+        self.assertTrue(callable(run_attribution.named_prompt))
+        self.assertFalse(hasattr(run_attribution, "_recorded_runs"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/vendor/browser-server-fixture.test.mjs

@@ -2,12 +2,14 @@ import test from 'node:test';
 import assert from 'node:assert/strict';
 import { createServer } from 'node:net';
 import { spawn } from 'node:child_process';
-import { once } from 'node:events';
+import { EventEmitter, once } from 'node:events';
 import { readFileSync } from 'node:fs';
+import { isAbsolute } from 'node:path';
 import { fileURLToPath, pathToFileURL } from 'node:url';
 import { setTimeout as delay } from 'node:timers/promises';
 import { startBrowserServer, createServerSupervisor, attachServerStderr, forwardPrefixedLines, SLOW_RESET_PREFIX } from '../../tools/vendor/start-browser-server.mjs';
-import { workflowWorkers } from '../../tools/vendor/test-workers.mjs';
+import { reporters, workflowWorkers } from '../../tools/vendor/test-workers.mjs';
+import RunningTestsReporter from '../../tools/vendor/running-tests-reporter.mjs';
 
 const fakeServer = fileURLToPath(new URL('./fake-browser-server.mjs', import.meta.url));
 const fake = (...extra) => ({ command: process.execPath, args: port => [fakeServer, '--port', String(port), ...extra] });
@@ -255,3 +257,57 @@ test('worker count follows PM_TEST_WORKERS, then one in CI, then a quarter of th
     assert.throws(() => workflowWorkers({ env: { PM_TEST_WORKERS: 'many' }, cpus: 8 }), /PM_TEST_WORKERS must be a positive integer/);
     assert.throws(() => workflowWorkers({ env: { PM_TEST_WORKERS: '0' }, cpus: 8 }), /PM_TEST_WORKERS must be a positive integer/);
 });
+
+test('CI adds the running-tests reporter and a blob report under test-results/blob/, which the trace upload covers; local runs keep list', () => {
+    assert.equal(reporters({ env: {}, name: 'workflows' }), 'list');
+    const [list, running, blob] = reporters({ env: { CI: '1' }, name: 'workflows' });
+    assert.deepEqual(list, ['list']);
+    // Absolute: a config imported from another directory (workflow-artifacts.test.mjs) still finds it.
+    assert.ok(isAbsolute(running[0]) && running[0].endsWith('running-tests-reporter.mjs'), running[0]);
+    assert.deepEqual(blob, ['blob', { outputDir: './test-results/blob/workflows' }]);
+});
+
+function fakeHost() {
+    const host = new EventEmitter();
+    Object.assign(host, { pid: 4242, signals: [], cwd: () => '/repo/tools/vendor' });
+    host.kill = (pid, signal) => host.signals.push([pid, signal]);
+    return host;
+}
+
+const spec = (id, title, line) => ({ id, title, location: { file: '/repo/tests/browser/a.spec.mjs', line },
+    parent: { project: () => ({ name: 'firefox' }) } });
+
+test('the running-tests reporter names only the tests that never ended, relative to rootDir', () => {
+    let clock = 1_000;
+    const output = [];
+    const host = fakeHost();
+    const reporter = new RunningTestsReporter({ now: () => clock, write: text => output.push(text), host });
+    reporter.onBegin({ rootDir: '/repo/tests/browser' });
+    const done = spec('1', 'finishes', 10);
+    const stuck = spec('2', 'hangs on a fetch', 20);
+    reporter.onTestBegin(done, { retry: 0 });
+    reporter.onTestBegin(stuck, { retry: 0 });
+    clock += 500;
+    reporter.onTestEnd(done, { retry: 0 });
+    clock += 41_500;
+    reporter.onEnd({ status: 'interrupted' });
+    assert.equal(output.join(''), '\nStill running when the run ended (interrupted): 1\n'
+        + '  [firefox] › a.spec.mjs:20 › hangs on a fetch (running 42.0s)\n');
+    assert.equal(host.listenerCount('SIGTERM'), 0, 'the run is over: SIGTERM gets its default again');
+    output.length = 0;
+    reporter.onTestEnd(stuck, { retry: 0 });
+    reporter.onEnd({ status: 'passed' });
+    assert.deepEqual(output, [], 'a run where every test ended prints nothing');
+});
+
+test('on the first SIGTERM the reporter prints the running tests and hands over to Playwright\'s SIGINT', () => {
+    const output = [];
+    const host = fakeHost();
+    const reporter = new RunningTestsReporter({ now: () => 0, write: text => output.push(text), host });
+    reporter.onBegin({ rootDir: '/repo/tests/browser' });
+    reporter.onTestBegin(spec('2', 'hangs on a fetch', 20), { retry: 0 });
+    host.emit('SIGTERM');
+    assert.match(output.join(''), /Still running when the run ended \(SIGTERM\): 1\n {2}\[firefox\] › a\.spec\.mjs:20 › hangs on a fetch/);
+    assert.deepEqual(host.signals, [[4242, 'SIGINT']], 'Playwright\'s own interrupt ends the run and writes the blob');
+    assert.equal(host.listenerCount('SIGTERM'), 0, 'a second SIGTERM kills the process as usual');
+});

tests/vendor/ci-workflow.test.mjs

@@ -247,6 +247,26 @@ test('browser job timeouts keep a real margin over measured runs', () => {
     }
 });
 
+// A hang (#425) must end inside the test step, before the job limit, so the log and the always()
+// uploads survive: the step's timeout-minutes < the job's.
+const BROWSER_TEST_STEP = /npm run --prefix tools\/vendor test:(workflows|coverage)/;
+
+test('each browser test step stops itself before its job limit and its traces upload anyway', () => {
+    for (const id of browserJobIds) {
+        const body = allJobs.get(id);
+        const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]);
+        const jobSteps = steps(body);
+        const testAt = jobSteps.findIndex(step => BROWSER_TEST_STEP.test(step));
+        const testStep = jobSteps[testAt];
+        const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(testStep)?.[1]);
+        assert.ok(stepMinutes > 0, `${id}: the test step sets its own timeout-minutes`);
+        assert.ok(jobMinutes - stepMinutes >= 2, `${id}: the step limit leaves time for setup and the uploads`);
+        const upload = jobSteps.slice(testAt + 1).find(step => /^ {10}path: tools\/vendor\/test-results\/$/m.test(step));
+        assert.ok(upload, `${id} uploads tools/vendor/test-results/ (traces and blob report) after the tests`);
+        assert.match(upload, /^ {8}if: always\(\)$/m, `${id} uploads them when the step failed or timed out`);
+    }
+});
+
 // Every job is bounded (#350): a hang ends at a measured limit instead of GitHub's 6-hour default.
 const GATING_WORKFLOWS = ['test.yml', 'code-quality.yml', 'viewerjs-check.yml'];
 const BOUNDED_WORKFLOWS = [...GATING_WORKFLOWS, 'portability.yml'];

tests/vendor/comfyui-config.test.mjs

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

tests/vendor/frontend-coverage-cli.test.mjs

@@ -1,9 +1,11 @@
 import test from 'node:test';
 import assert from 'node:assert/strict';
-import { existsSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs';
+import { existsSync, mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { EventEmitter } from 'node:events';
+import { setTimeout as delay } from 'node:timers/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
-import { plan, runCoverage } from '../../tools/vendor/frontend-coverage.mjs';
+import { plan, runCoverage, runNode } from '../../tools/vendor/frontend-coverage.mjs';
 
 const steps = argv => plan(argv).map(({ step, ...options }) => (Object.keys(options).length ? { step, ...options } : step));
 
@@ -91,3 +93,46 @@ test('shard 1 runs the node units into the shared raw dir, then the viewer suite
     assert.ok(calls[1].args.includes(join(paths.tools, 'playwright.config.mjs')));
     assert.deepEqual(calls[2].args.slice(-2), ['--shard', '1/3']);
 });
+
+// #425: npm and a CI step timeout signal only frontend-coverage.mjs; its running child must get it.
+function hostWithSignals() {
+    return new EventEmitter();
+}
+
+test('runNode resolves on exit 0, rejects with the exit code otherwise and leaves no signal listener', async () => {
+    const host = hostWithSignals();
+    await runNode(['-e', 'process.exit(0)'], {}, { host });
+    await assert.rejects(runNode(['-e', 'process.exit(3)'], {}, { host }), /Command failed \(3\): -e process\.exit\(3\)/);
+    assert.equal(host.listenerCount('SIGTERM') + host.listenerCount('SIGINT'), 0);
+});
+
+test('runNode forwards SIGTERM to the running child, which can report before it exits', async t => {
+    const root = mkdtempSync(join(tmpdir(), 'pm-coverage-signal-'));
+    t.after(() => rmSync(root, { recursive: true, force: true }));
+    const ready = join(root, 'ready');
+    const seen = join(root, 'seen');
+    const child = `const fs = require('node:fs');
+process.on('SIGTERM', () => { fs.writeFileSync(${JSON.stringify(seen)}, 'SIGTERM'); process.exit(7); });
+fs.writeFileSync(${JSON.stringify(ready)}, ''); setInterval(() => {}, 1000);`;
+    const host = hostWithSignals();
+    const running = runNode(['-e', child], {}, { host });
+    while (!existsSync(ready)) await delay(20);
+    host.emit('SIGTERM');
+    await assert.rejects(running, /Command failed \(7\)/);
+    assert.equal(readFileSync(seen, 'utf8'), 'SIGTERM', 'the child handled the forwarded signal');
+    assert.equal(host.listenerCount('SIGTERM'), 0);
+});
+
+test('runNode forwards SIGINT and starts the child in its own process group', async () => {
+    const host = hostWithSignals();
+    const kills = [];
+    let options;
+    const fake = new EventEmitter();
+    fake.kill = signal => { kills.push(signal); queueMicrotask(() => fake.emit('exit', null, signal)); };
+    const running = runNode(['x.mjs'], {}, { host, spawnChild: (command, args, spawnOptions) => { options = spawnOptions; return fake; } });
+    assert.equal(options.detached, true, 'a group-wide signal reaches the child only through forwarding, once');
+    host.emit('SIGINT');
+    await assert.rejects(running, /Command failed \(null\): x\.mjs/);
+    assert.deepEqual(kills, ['SIGINT']);
+    assert.equal(host.listenerCount('SIGINT'), 0);
+});

tools/vendor/frontend-coverage-baseline.json

@@ -107,6 +107,12 @@
     "functions": 98,
     "statements": 98
   },
+  "save-to-library.js": {
+    "lines": 98,
+    "branches": 98,
+    "functions": 98,
+    "statements": 98
+  },
   "scan-progress.js": {
     "lines": 98,
     "branches": 98,

tools/vendor/frontend-coverage.mjs

@@ -1,7 +1,7 @@
 import { existsSync, readFileSync, readdirSync, mkdirSync, rmSync, writeFileSync } from 'node:fs';
 import { dirname, join, relative } from 'node:path';
 import { fileURLToPath, pathToFileURL } from 'node:url';
-import { spawnSync } from 'node:child_process';
+import { spawn } from 'node:child_process';
 import { repositoryRoot } from './vendor-viewerjs.mjs';
 
 import { discoverSources, checkCoverage } from '@comfyassets/toolkit/testing/coverage';
@@ -10,10 +10,27 @@ const metrics = ['lines', 'branches', 'functions', 'statements'];
 const SHARD = /^([1-9]\d*)\/([1-9]\d*)$/;
 const USAGE = 'usage: frontend-coverage.mjs [--update-baseline] | --shard I/N | --merge-only [--update-baseline]';
 
-function runNode(args, env = {}) {
-    const result = spawnSync(process.execPath, args, { cwd: repositoryRoot, stdio: 'inherit',
-        env: { ...process.env, ...env } });
-    if (result.status !== 0) throw new Error(`Command failed (${result.status}): ${args.join(' ')}`);
+const FORWARDED_SIGNALS = ['SIGTERM', 'SIGINT'];
+
+// Runs one step's node child and waits for it. npm and a CI step timeout signal only this process,
+// so SIGTERM and SIGINT are forwarded to the running child (#425): Playwright then reports the tests
+// still running and writes its blob. The child gets its own process group, so a signal sent to the
+// whole group (Ctrl-C, `timeout`) still reaches it exactly once, through this forwarding.
+export function runNode(args, env = {}, { spawnChild = spawn, host = process } = {}) {
+    return new Promise((resolve, reject) => {
+        const child = spawnChild(process.execPath, args, { cwd: repositoryRoot, stdio: 'inherit',
+            env: { ...process.env, ...env }, detached: true });
+        const forward = signal => () => child.kill(signal);
+        const handlers = FORWARDED_SIGNALS.map(signal => [signal, forward(signal)]);
+        for (const [signal, handler] of handlers) host.on(signal, handler);
+        const done = () => { for (const [signal, handler] of handlers) host.removeListener(signal, handler); };
+        child.once('error', error => { done(); reject(error); });
+        child.once('exit', code => {
+            done();
+            if (code === 0) resolve();
+            else reject(new Error(`Command failed (${code}): ${args.join(' ')}`));
+        });
+    });
 }
 
 function defaultPaths() {
@@ -112,17 +129,17 @@ export async function runCoverage({ argv = process.argv.slice(2), run = runNode,
         'node-units'() {
             const tests = readdirSync(join(root, 'tests/js')).filter(file => file.endsWith('.test.js'))
                 .map(file => `tests/js/${file}`);
-            run(['--test', ...tests], { NODE_V8_COVERAGE: raw });
+            return run(['--test', ...tests], { NODE_V8_COVERAGE: raw });
         },
         viewer() {
-            run([playwright, 'test', '--config', join(tools, 'playwright.config.mjs')]);
+            return run([playwright, 'test', '--config', join(tools, 'playwright.config.mjs')]);
         },
         workflows({ shard }) {
-            run([playwright, 'test', '--config', join(tools, 'playwright.workflows.config.mjs'),
+            return run([playwright, 'test', '--config', join(tools, 'playwright.workflows.config.mjs'),
                 '--project', 'chromium', '--project', 'touch', ...(shard ? ['--shard', shard] : [])], { PM_JS_COVERAGE: raw });
         },
         report() {
-            run([join(tools, 'node_modules/c8/bin/c8.js'), 'report', '--all', '--src', sourceRoot,
+            return run([join(tools, 'node_modules/c8/bin/c8.js'), 'report', '--all', '--src', sourceRoot,
                 '--include', 'web/js/**/*.js', '--include', 'web/js/**/*.mjs', '--temp-directory', raw, '--reports-dir', reportRoot,
                 '--reporter', 'text', '--reporter', 'json-summary', '--reporter', 'html']);
         },

tools/vendor/playwright.config.mjs

@@ -1,4 +1,5 @@
 import { defineConfig } from '@playwright/test';
+import { reporters } from './test-workers.mjs';
 
 export default defineConfig({
     testDir: './tests',
@@ -8,7 +9,7 @@ export default defineConfig({
     workers: 2,
     forbidOnly: Boolean(process.env.CI),
     retries: process.env.CI ? 1 : 0,
-    reporter: 'list',
+    reporter: reporters({ name: 'viewer' }),
     use: { browserName: 'chromium', headless: true, trace: 'retain-on-failure' },
     projects: [
         { name: 'desktop', use: { viewport: { width: 1280, height: 800 } } },

tools/vendor/playwright.workflows.config.mjs

@@ -1,5 +1,5 @@
 import { defineConfig, devices } from '@playwright/test';
-import { workflowWorkers } from './test-workers.mjs';
+import { reporters, workflowWorkers } from './test-workers.mjs';
 
 // No global webServer: tests/browser/fixtures.mjs starts one server per worker
 // on an OS-chosen port. Files run in parallel; tests inside a file keep their order.
@@ -11,7 +11,7 @@ export default defineConfig({
     fullyParallel: false,
     forbidOnly: Boolean(process.env.CI),
     retries: 0,
-    reporter: 'list',
+    reporter: reporters({ name: 'workflows' }),
     use: { trace: 'retain-on-failure', screenshot: 'only-on-failure' },
     projects: [
         { name: 'chromium', outputDir: './test-results/workflows/chromium', use: { browserName: 'chromium' } },

tools/vendor/running-tests-reporter.mjs

@@ -0,0 +1,58 @@
+import { writeSync } from 'node:fs';
+import { relative } from 'node:path';
+
+// Names the tests still running when a run is cut off (#425). A browser test step that reaches its
+// timeout-minutes gets SIGTERM, which ends Playwright at once: no reporter's onEnd runs, so neither
+// `list` nor the blob report says what was running. On the first SIGTERM this reporter prints every
+// started test that has not ended, then turns the signal into SIGINT, Playwright's own interrupt,
+// which stops the workers, ends the run as interrupted (non-zero exit) and writes the blob report.
+// A second SIGTERM gets the default behaviour: the process dies.
+export default class RunningTestsReporter {
+    constructor({ now = () => Date.now(), write = text => writeSync(1, text), host = process } = {}) {
+        this.now = now;
+        this.write = write;
+        this.host = host;
+        this.rootDir = host.cwd();
+        this.running = new Map();
+        this.onSigterm = () => {
+            this.host.removeListener('SIGTERM', this.onSigterm);
+            this.report('SIGTERM');
+            this.host.kill(this.host.pid, 'SIGINT');
+        };
+    }
+
+    printsToStdio() {
+        return false;
+    }
+
+    onBegin(config) {
+        this.rootDir = config.rootDir;
+        this.host.on('SIGTERM', this.onSigterm);
+    }
+
+    onTestBegin(test, result) {
+        this.running.set(`${test.id}#${result.retry}`, { test, startedAt: this.now() });
+    }
+
+    onTestEnd(test, result) {
+        this.running.delete(`${test.id}#${result.retry}`);
+    }
+
+    describe({ test, startedAt }) {
+        const project = test.parent?.project()?.name;
+        const where = `${relative(this.rootDir, test.location.file)}:${test.location.line}`;
+        const seconds = ((this.now() - startedAt) / 1000).toFixed(1);
+        return `${project ? `[${project}] › ` : ''}${where} › ${test.title} (running ${seconds}s)`;
+    }
+
+    report(reason) {
+        if (!this.running.size) return;
+        const lines = [...this.running.values()].map(entry => `  ${this.describe(entry)}`);
+        this.write(`\nStill running when the run ended (${reason}): ${this.running.size}\n${lines.join('\n')}\n`);
+    }
+
+    onEnd(result) {
+        this.host.removeListener('SIGTERM', this.onSigterm);
+        this.report(result.status);
+    }
+}

tools/vendor/test-workers.mjs

@@ -1,4 +1,8 @@
 import { availableParallelism } from 'node:os';
+import { fileURLToPath } from 'node:url';
+
+// Absolute, so a config that imports this one from another directory still finds it.
+const RUNNING_TESTS_REPORTER = fileURLToPath(new URL('./running-tests-reporter.mjs', import.meta.url));
 
 const MAX_LOCAL_WORKERS = 8;
 const CPUS_PER_WORKER = 4;
@@ -16,3 +20,10 @@ export function workflowWorkers({ env = process.env, cpus = availableParallelism
     if (env.CI) return 1;
     return Math.max(1, Math.min(MAX_LOCAL_WORKERS, Math.floor(cpus / CPUS_PER_WORKER)));
 }
+
+// Under CI, running-tests-reporter.mjs names the tests a step timeout's signal cut off (`list`
+// cannot: they never end), and a blob report beside the traces is uploaded with them.
+export function reporters({ env = process.env, name } = {}) {
+    if (!env.CI) return 'list';
+    return [['list'], [RUNNING_TESTS_REPORTER], ['blob', { outputDir: `./test-results/blob/${name}` }]];
+}

utils/execution_context_capture.py

@@ -11,6 +11,7 @@ import weakref
 
 from .execution_context_budget import ContextBudget, PayloadLease, SealedContext
 from .execution_context_graph import (
+    NOT_SAVED,
     GraphProjection,
     GraphRole,
     SourceRef,
@@ -325,6 +326,8 @@ class ContextCapture:
                     nonlocal ambiguous
                     if role.reason == "source-transformed":
                         return _unknown("unsupported")
+                    if role.reason == NOT_SAVED:
+                        return _unknown("unavailable")
                     if not role.sources:
                         return role.inferred or _unknown("unsupported")
                     values = [

utils/execution_context_graph.py

@@ -3,8 +3,13 @@
 import re
 from dataclasses import dataclass
 from .execution_contexts import PromptSnapshot, TextEvidence, text_size
+from .prompt_graph import save_setting
 from .prompt_pair_resolver import _DEFAULT_ADAPTERS
 
+# A role whose PromptManager source has save_to_library off, or wired from a
+# node the graph cannot read: no text is recorded for it (#90)
+NOT_SAVED = "not_saved"
+
 
 @dataclass(frozen=True)
 class SourceRef:
@@ -332,6 +337,11 @@ class _Projector:
             self.transformed = False
             # Only the capture-source-to-consumer path determines transforms.
             sources = self.discover_sources(value, "conditioning")
+            if any(
+                save_setting(self.graph, source.node_id) is not True
+                for source in sources
+            ):
+                return GraphRole((), None, NOT_SAVED)
             snapshot, _ = self.text(value, "conditioning")
             return GraphRole(
                 sources, snapshot, "source-transformed" if self.transformed else ""

utils/file_prompt.py

@@ -0,0 +1,634 @@
+"""The per-file prompt decision shared by the output scan and the monitor (#247).
+
+``decide_file_prompt`` is the ladder: which prompt one output file names,
+from its metadata alone plus, when the graph cannot decide, the save node
+recorded history says wrote it. It does no file or database I/O; the caller
+reads the metadata, the file size and history (through ``lookup``) and
+stores the ``Decision``.
+
+The default resolver profile decides first, as ingestion always has. Only
+for a file it leaves ambiguous or unresolved does the ladder ask the
+inspector's resolve (``inspection_resolution``: the inspection profile, then
+the recorded saver anchor; never the file-name saver anchor, whose files go
+to Needs review by the #390 ruling), then the #170 browser parser (the legacy floor).
+The policies differ in one place: for a file the inspector calls ambiguous,
+the scan keeps the legacy floor and the monitor stores nothing. Junk text
+(``junk_text``) is never stored, whichever step found it.
+"""
+
+import json
+import re
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Callable, Mapping, Optional, Union
+
+from .generation_params import (
+    PLACEHOLDERS,
+    GenerationParamsError,
+    extract_generation_params,
+    parse_page_json,
+)
+from .prompt_graph import (
+    PROMPT_MANAGER_TYPES,
+    _positive_roots,
+    is_negative_input,
+    metadata_opted_out,
+    resolve_text,
+    save_setting,
+    side_opted_out,
+)
+from .prompt_inspector import inspection_resolution
+from .prompt_pair_resolver import _numeric, resolve_prompt_pair, sampler_less_prompt
+from .review_capture import reviewable
+
+try:
+    from ..database.pair_ingestion import blank_positive
+except ImportError:
+    from database.pair_ingestion import blank_positive
+
+
+class Policy(Enum):
+    SCAN = "scan"
+    MONITOR = "monitor"
+
+
+PAIR = "pair"
+POSITIVE = "positive"
+HISTORY = "history"
+NOT_SAVED = "not_saved"
+NOT_SAVED_WIRED = "not_saved_wired"
+NEVER = "never"
+KINDS = (PAIR, POSITIVE, HISTORY, NOT_SAVED, NOT_SAVED_WIRED, NEVER)
+
+_UNDECIDED = ("ambiguous", "unresolved")
+
+
+@dataclass(frozen=True)
+class Decision:
+    """What to store for one file (see the design's "The ladder").
+
+    ``kind``: one of ``KINDS``. ``text``: the storage text of the positive.
+    ``canonical``: the parsed metadata stored with the link. ``resolution``:
+    the resolution that decided a ``pair``, else the default profile's
+    result. ``source``: which step named the text (log lines and the scan's
+    outcome detail). ``review``: ``ambiguous`` or ``untraced`` when the
+    default profile was ambiguous (the scan's Needs review lists).
+    ``try_recorded_run``: the scan tries the recorded run's stored prompt
+    when nothing else links. ``unlinked_status``: the scan status when
+    nothing links (None: no prompt metadata).
+    """
+
+    kind: str
+    text: Optional[str] = None
+    canonical: dict = field(default_factory=dict)
+    resolution: Optional[dict] = None
+    source: str = "none"
+    review: Optional[str] = None
+    try_recorded_run: bool = False
+    unlinked_status: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class _File:
+    """The ladder's inputs beside the metadata."""
+
+    metadata: Mapping
+    canonical: dict
+    resolution: dict
+    review: Optional[str]
+    file_name: Optional[str]
+    in_output: bool
+    lookup: Optional[Callable[[], Any]]
+    policy: Policy
+
+
+def decide_file_prompt(
+    metadata: Any,
+    *,
+    file_name: Optional[str],
+    file_size: Union[None, int, Callable[[], Optional[int]]],
+    in_output: bool,
+    lookup: Optional[Callable[[], Any]],
+    policy: Policy,
+) -> Decision:
+    """The ladder for one file's raw ``metadata``.
+
+    ``prompt``/``workflow`` may be JSON strings (the scan) or parsed values
+    (the monitor). ``lookup`` returns the file's ``RecordedOutput`` or None;
+    it is None when history must not be asked yet. It is called at most
+    once, and only for a graph the default profile leaves undecided.
+    ``file_size`` may be a callable: it is read only for a file with prompt
+    metadata whose ``file_info`` has no size.
+    """
+    if not metadata:
+        return Decision(NEVER, try_recorded_run=True)
+    canonical = _parsed(metadata)
+    if metadata_opted_out(canonical):
+        # Save to library was off for the run's prompt: nothing is stored,
+        # no recorded-run fallback, no Needs review item (#90)
+        return Decision(NOT_SAVED, canonical=canonical)
+    if metadata_opted_out(canonical, unreadable=True):
+        # The switch is wired from a node the file cannot show: only the
+        # live run (the node saw the value) can link it
+        return Decision(NOT_SAVED_WIRED, canonical=canonical)
+    switched = _switched_off_upstream(canonical.get("prompt"))
+    if switched is not None:
+        # The #90 walk follows only prompt-like inputs; any input name
+        # upstream of a positive can carry a switched-off node's text
+        return Decision(switched, canonical=canonical)
+    if not canonical.get("prompt") and not canonical.get("parameters"):
+        return Decision(NEVER, canonical=canonical, try_recorded_run=True)
+    _with_size(canonical, file_size)
+    resolution = resolve_prompt_pair(canonical)
+    switched = (
+        _side_switch(canonical.get("prompt"), resolution["positive"])
+        if resolution["state"] == "resolved"
+        else None
+    )
+    if switched is not None:
+        # The positive passes a node that is off, though the #90 graph walk
+        # did not reach it (steps 5 and 7)
+        return Decision(switched, canonical=canonical)
+    if resolution["state"] == "resolved" and side_opted_out(
+        canonical.get("prompt"), resolution["negative"]
+    ):
+        # Only the negative's node is off: the positive alone, never a pair
+        positive = resolution["positive"]
+        text = positive.get("storage_text", positive["text"]).strip()
+        return _gated(
+            Decision(
+                POSITIVE,
+                text=text,
+                canonical=canonical,
+                resolution=resolution,
+                source="opted-out-negative",
+            ),
+            None,
+        )
+    if blank_positive(resolution):
+        # Only prepend/append text: no stored prompt can own it (#342)
+        resolution = {**resolution, "state": "unresolved"}
+    found = _File(
+        metadata,
+        canonical,
+        resolution,
+        _review(resolution),
+        file_name,
+        in_output,
+        lookup,
+        policy,
+    )
+    return _gated(_decide(found), found)
+
+
+def _decide(found: _File) -> Decision:
+    state = found.resolution["state"]
+    graph = found.canonical.get("prompt")
+    if state == "resolved":
+        positive = found.resolution["positive"]
+        text = positive.get("storage_text", positive["text"])
+        return _decision(found, PAIR, text, "resolver")
+    if graph and state in _UNDECIDED:
+        return _undecided_graph(found)
+    if state == "ambiguous":
+        return _never(found, _unlinked(found))
+    if state == "unresolved" and not isinstance(found.canonical.get("parameters"), str):
+        return _never(found, "unresolved")
+    if graph:
+        return _sampler_less(found, graph)
+    text = _a1111_positive(found.metadata)
+    if text is None:
+        return _never(found, "found")
+    return _decision(found, POSITIVE, text, "a1111")
+
+
+def _undecided_graph(found: _File) -> Decision:
+    recorded = found.lookup() if found.lookup is not None else None
+    saver = recorded.saver_id if recorded is not None else None
+    inspected = inspection_resolution(
+        found.metadata, found.file_name, saver, name_anchor=False
+    )
+    text = _inspected_text(inspected.resolution)
+    switched = (
+        _side_switch(found.canonical["prompt"], inspected.resolution["positive"])
+        if text is not None
+        else None
+    )
+    if switched is not None:
+        # The inspector follows pass-throughs the #90 graph walk does not
+        return Decision(switched, canonical=found.canonical)
+    if text is not None:
+        source = "recorded-run" if inspected.by_record else "inspector"
+        return _decision(found, POSITIVE, text, source)
+    if found.policy is Policy.MONITOR:
+        if found.lookup is None and found.in_output:
+            return _decision(found, HISTORY, None, "none")
+        if not inspected.malformed and inspected.resolution["state"] == "ambiguous":
+            return _never(found, _unlinked(found))
+    text = graph_prompt_text(found.metadata)
+    switched = _off_text(found.canonical["prompt"], text) if text else None
+    if switched is not None:
+        # The legacy answer is a switched-off node's own text (#90)
+        return Decision(switched, canonical=found.canonical)
+    if text is not None and _joined_text(found.canonical["prompt"], text):
+        # The legacy parser joins prepend + text + append: the text a
+        # PromptManager node stores cannot be told from it (#343).
+        return _never(found, _unlinked(found), try_recorded_run=True)
+    if text is not None:
+        return _decision(found, POSITIVE, text, "legacy-parser")
+    return _never(found, _unlinked(found), try_recorded_run=True)
+
+
+def _sampler_less(found: _File, graph: Any) -> Decision:
+    # No node takes a positive or negative input (a guider or wrapper graph):
+    # the one wired text, never a guess (#252).
+    wired = sampler_less_prompt(graph)
+    if wired["text"] is not None:
+        return _decision(found, POSITIVE, wired["text"], "sampler-less")
+    if isinstance(found.canonical.get("parameters"), str):
+        # A metadata saver also wrote A1111 text: its positive prompt.
+        text = _a1111_positive(found.metadata)
+        if text is not None:
+            return _decision(found, POSITIVE, text, "a1111")
+    status = "unresolved" if wired["state"] == "ambiguous" else "found"
+    return _never(found, status, try_recorded_run=True)
+
+
+_UNDECIDED_SOURCES = ("inspector", "recorded-run", "legacy-parser")
+
+
+def _switch_kind(settings: Any) -> Optional[str]:
+    """``not_saved`` when a setting is off, ``not_saved_wired`` when one is
+    unknowable (wired from a node the graph cannot read), else None."""
+    settings = list(settings)
+    if False in settings:
+        return NOT_SAVED
+    if None in settings:
+        return NOT_SAVED_WIRED
+    return None
+
+
+def _prompt_manager_ids(graph: Any) -> list:
+    if not isinstance(graph, dict):
+        return []
+    return [
+        str(node_id)
+        for node_id, node in graph.items()
+        if isinstance(node, dict) and node.get("class_type") in PROMPT_MANAGER_TYPES
+    ]
+
+
+def _side_switch(graph: Any, side: Any) -> Optional[str]:
+    """The switch outcome of the PromptManager nodes on a side's path."""
+    if not isinstance(graph, dict) or not side_opted_out(graph, side):
+        return None
+    provenance = side.get("provenance")
+    path = provenance.get("path") if isinstance(provenance, dict) else None
+    path = path if isinstance(path, list) else []
+    nodes = {str(step.get("node_id")) for step in path if isinstance(step, dict)} & set(
+        _prompt_manager_ids(graph)
+    )
+    return _switch_kind(save_setting(graph, node_id) for node_id in nodes)
+
+
+# Inputs that never carry prompt text or positive conditioning.
+_NOT_TEXT_INPUTS = frozenset(
+    {
+        "model",
+        "clip",
+        "vae",
+        "latent_image",
+        "samples",
+        "pixels",
+        "image",
+        "images",
+        "mask",
+        "noise",
+        "sigmas",
+        "sampler",
+        "control_net",
+        "upscale_model",
+    }
+)
+_MAX_SWITCH_VISITS = 5000
+
+
+def _negative_name(key: Any) -> bool:
+    """#90's negative rule, plus any name containing "negative"
+    (``negative2``, ``NegativePrompt``)."""
+    return is_negative_input(key) or "negative" in str(key).lower()
+
+
+def _switched_off_upstream(graph: Any) -> Optional[str]:
+    """The switch outcome of every PromptManager node upstream of a positive.
+
+    Follows every input except negative and non-text ones (#90's walk follows
+    only prompt-like names). A graph without a recognisable sampler checks
+    every PromptManager node, as #90 does.
+    """
+    if not isinstance(graph, dict):
+        return None
+    roots = _positive_roots(graph)
+    if not roots:
+        ids = _prompt_manager_ids(graph)
+        return _switch_kind(save_setting(graph, node_id) for node_id in ids)
+    seen, found = set(), set()
+    stack = [str(link[0]) for link in roots]
+    while stack and len(seen) < _MAX_SWITCH_VISITS:
+        node_id = stack.pop()
+        if node_id in seen:
+            continue
+        seen.add(node_id)
+        node = graph.get(node_id)
+        if not isinstance(node, dict):
+            continue
+        if node.get("class_type") in PROMPT_MANAGER_TYPES:
+            found.add(node_id)
+        inputs = node.get("inputs")
+        for key, value in (inputs.items() if isinstance(inputs, dict) else ()):
+            if key in _NOT_TEXT_INPUTS or _negative_name(key):
+                continue
+            if isinstance(value, (list, tuple)) and len(value) == 2:
+                if isinstance(value[0], (str, int)) and isinstance(value[1], int):
+                    stack.append(str(value[0]))
+    return _switch_kind(save_setting(graph, node_id) for node_id in found)
+
+
+def _off_text(graph: Any, text: str) -> Optional[str]:
+    """The switch outcome when ``text`` is the text (resolved, else literal)
+    of a PromptManager-family node whose Save to library is not on."""
+    wanted = text.strip()
+    matched = []
+    for node_id in _prompt_manager_ids(graph):
+        node_text = resolve_text(graph, node_id)
+        if node_text is None:
+            inputs = graph[node_id].get("inputs")
+            literal = inputs.get("text") if isinstance(inputs, dict) else None
+            node_text = literal if isinstance(literal, str) else None
+        if node_text is not None and node_text.strip() == wanted:
+            matched.append(save_setting(graph, node_id))
+    return _switch_kind(matched)
+
+
+def _joined_text(graph: Any, text: str) -> bool:
+    """Whether ``text`` may hold a PromptManager node's prepend or append.
+
+    True when a PromptManager-family node has a prepend or append part and
+    ``text`` is not the literal ``text`` of one such node (the legacy
+    parser's answer when it does not follow a linked part).
+    """
+    own = _prompt_manager_parts(graph)
+    return own is not None and text.strip() not in own
+
+
+def _prompt_manager_parts(graph: Any) -> Optional[set]:
+    """The literal texts of PromptManager-family nodes with a prepend or
+    append part, or None when no such node exists."""
+    if not isinstance(graph, dict):
+        return None
+    found = None
+    for node in graph.values():
+        if not isinstance(node, dict):
+            continue
+        name, inputs = node.get("class_type"), node.get("inputs")
+        if not isinstance(name, str) or "promptmanager" not in name.lower():
+            continue
+        if not isinstance(inputs, dict) or not _has_part(inputs):
+            continue
+        found = found if found is not None else set()
+        if isinstance(inputs.get("text"), str):
+            found.add(inputs["text"].strip())
+    return found
+
+
+def _has_part(inputs: dict) -> bool:
+    for key in ("prepend_text", "append_text"):
+        part = inputs.get(key)
+        if part is not None and not (isinstance(part, str) and not part.strip()):
+            return True
+    return False
+
+
+def _gated(decision: Decision, found: Optional[_File]) -> Decision:
+    """Junk text is never stored (step 13).
+
+    A junk answer of the undecided-graph step keeps that step's unlinked
+    status, so an ambiguous file is still a Needs review case; any other
+    step reports ``found``.
+    """
+    if decision.kind not in (PAIR, POSITIVE):
+        return decision
+    junk = junk_text(decision.text)
+    if junk is None:
+        return decision
+    return Decision(
+        NEVER,
+        canonical=decision.canonical,
+        resolution=decision.resolution,
+        source=f"junk:{junk}",
+        review=decision.review,
+        try_recorded_run=True,
+        unlinked_status=(
+            _unlinked(found)
+            if found is not None and decision.source in _UNDECIDED_SOURCES
+            else "found"
+        ),
+    )
+
+
+def _decision(found: _File, kind: str, text: Optional[str], source: str) -> Decision:
+    return Decision(
+        kind,
+        text=text,
+        canonical=found.canonical,
+        resolution=found.resolution,
+        source=source,
+        review=found.review,
+    )
+
+
+def _never(found: _File, status: str, *, try_recorded_run: bool = False) -> Decision:
+    return Decision(
+        NEVER,
+        canonical=found.canonical,
+        resolution=found.resolution,
+        review=found.review,
+        try_recorded_run=try_recorded_run,
+        unlinked_status=status,
+    )
+
+
+def _review(resolution: dict) -> Optional[str]:
+    if resolution["state"] != "ambiguous":
+        return None
+    return "ambiguous" if reviewable(resolution) else "untraced"
+
+
+def _unlinked(found: _File) -> str:
+    """Ambiguous without a choice is reported unresolved."""
+    if found.review == "untraced":
+        return "unresolved"
+    return found.resolution["state"]
+
+
+def _inspected_text(resolution: dict) -> Optional[str]:
+    """The storage text of a positive the inspector shows, else None."""
+    if resolution.get("state") in _UNDECIDED:
+        return None
+    positive = resolution.get("positive")
+    if not isinstance(positive, dict):
+        return None
+    text = positive.get("storage_text", positive.get("text"))
+    return text if isinstance(text, str) and text.strip() else None
+
+
+def _parsed(metadata: Mapping) -> dict:
+    canonical = dict(metadata)
+    for key in ("prompt", "workflow"):
+        if isinstance(canonical.get(key), str):
+            try:
+                canonical[key] = json.loads(canonical[key])
+            except (ValueError, TypeError):
+                # Preserve malformed metadata for conservative resolver rejection.
+                pass
+    return canonical
+
+
+def _with_size(canonical: dict, file_size: Any) -> None:
+    """Record the file's size in ``file_info`` unless the metadata carries one.
+
+    Stored as ``generated_images.file_size``; the "new" scan scope skips a
+    known file only while its size is unchanged.
+    """
+    info = canonical.get("file_info")
+    info = dict(info) if isinstance(info, dict) else {}
+    if info.get("size") is None:
+        size = file_size() if callable(file_size) else file_size
+        if size is not None:
+            info["size"] = size
+    if info:
+        canonical["file_info"] = info
+
+
+def _a1111_positive(metadata: Mapping) -> Optional[str]:
+    text = parse_comfyui_prompt(metadata)["positive_prompt"]
+    return text.strip() if text and text.strip() else None
+
+
+_LORA_TAG = re.compile(r"<lora:[^>]*>")
+_LORA_SEPARATORS = " \t\r\n\f\v,;|"
+_MODEL_FILE = re.compile(r"\S*\.(?:safetensors|ckpt|pt|pth|bin|gguf|sft)", re.I)
+
+
+def junk_text(text: Any) -> Optional[str]:
+    """The #252 junk class of ``text``, or None when it may be a prompt."""
+    stripped = text.strip() if isinstance(text, str) else ""
+    if not stripped:
+        return "blank"
+    if _numeric(stripped):
+        return "number"
+    if not _LORA_TAG.sub("", stripped).strip(_LORA_SEPARATORS):
+        return "lora-only"
+    if _MODEL_FILE.fullmatch(stripped):
+        return "model-file"
+    return None
+
+
+def graph_prompt_text(metadata: Mapping) -> Optional[str]:
+    """The positive prompt the browser parser traces from the graphs, or None.
+
+    Records parse as the legacy pages parsed them (NaN reads as null).
+    """
+    records = {}
+    for key in ("prompt", "workflow"):
+        value = metadata.get(key)
+        records[key] = parse_page_json(value) if isinstance(value, str) else value
+    try:
+        positive = extract_generation_params(records)["positivePrompt"]
+    except GenerationParamsError:
+        return None
+    if not isinstance(positive, str) or positive == PLACEHOLDERS["positivePrompt"]:
+        return None
+    return positive.strip() or None
+
+
+_COMMON_FIELDS = (
+    "positive",
+    "negative",
+    "steps",
+    "cfg",
+    "sampler",
+    "scheduler",
+    "seed",
+)
+
+
+def _json_or_raw(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return value
+
+
+def parse_comfyui_prompt(metadata: Mapping) -> dict:
+    """Parse ComfyUI and A1111 prompt data from metadata.
+
+    Accepts the file's strings or parsed values: a parsed ``prompt`` or
+    ``workflow`` is kept as it is and a non-string ``parameters`` is ignored.
+    """
+    result = {
+        "prompt": None,
+        "workflow": None,
+        "parameters": {},
+        "positive_prompt": None,
+        "negative_prompt": None,
+    }
+    params = metadata.get("parameters")
+    if isinstance(params, str):
+        _parse_a1111(params, result)
+    if result["positive_prompt"] is None:
+        for key in ("prompt", "workflow"):
+            if key in metadata:
+                result[key] = _json_or_raw(metadata[key])
+        for name in _COMMON_FIELDS:
+            if name in metadata:
+                result["parameters"][name] = _json_or_raw(metadata[name])
+    return result
+
+
+_SETTINGS_FIELDS = ("Steps", "Sampler", "Schedule type", "CFG scale", "Seed")
+
+
+def _parse_a1111(params: str, result: dict) -> None:
+    """A1111/Forge ``parameters``: a possibly multi-line positive, then a
+    "Negative prompt:" section, then a trailing "Steps: ..." settings line.
+    Taking the first line as the positive would truncate a multi-line prompt,
+    so the text is split on the settings line instead.
+    """
+    body = params
+    settings = ""
+    found = re.search(r"\nSteps:\s", body)
+    if found:
+        settings = body[found.start() + 1 :].strip()
+        body = body[: found.start()]
+    negative = None
+    found = re.search(r"(?im)^\s*negative prompt:\s*", body)
+    if found:
+        negative = body[found.end() :].strip()
+        body = body[: found.start()]
+    elif body.lower().lstrip().startswith("negative prompt:"):
+        negative = body.lower().lstrip()[len("negative prompt:") :].strip()
+        body = ""
+    result["positive_prompt"] = body.strip() or None
+    if negative:
+        result["negative_prompt"] = negative
+    result["parameters"]["parameters"] = params
+    for name in _SETTINGS_FIELDS if settings else ():
+        found = re.search(
+            r"(?i)(?:^|,\s*)" + re.escape(name) + r":\s*([^,]+)", settings
+        )
+        if found:
+            key = name.lower().replace(" ", "_")
+            result["parameters"][key] = found.group(1).strip()

utils/image_monitor.py

@@ -37,7 +37,13 @@ from .hashing import generate_prompt_hash
 from .metadata_extractor import ComfyUIMetadataExtractor
 from .parallel import note_refusal, refused_at_shutdown
 from .logging_config import get_logger
-from .prompt_graph import resolve_text, run_prompt_nodes
+from .prompt_graph import (
+    metadata_opted_out,
+    resolve_text,
+    run_prompt_nodes,
+    save_setting,
+    side_opted_out,
+)
 from .prompt_pair_resolver import resolve_prompt_pair
 from .run_keys import image_run_keys
 from .usage_tracking import survive_switch
@@ -450,6 +456,12 @@ class ImageGenerationHandler(FileSystemEventHandler):
             # that produced this image, independent of node caching and of any
             # negative-prompt PromptManager nodes in the workflow.
             state, current_prompt = self._find_prompt_from_metadata(metadata)
+            if state == "opted_out":
+                self.logger.info(
+                    "Save to library is off for this run's prompt, skipping image: "
+                    f"{image_path}"
+                )
+                return
             if state == "foreign":
                 # Another workflow saved this image: it belongs to no prompt of
                 # ours, and must not consume a batch queue entry.
@@ -715,16 +727,27 @@ class ImageGenerationHandler(FileSystemEventHandler):
             ``("linked", prompt)`` with a context dict holding 'id' and 'text';
             ``("foreign", None)`` when the workflow has no positive PromptManager
             node, so the image belongs to no prompt of ours;
+            ``("opted_out", None)`` when a PromptManager node that would be the
+            run's prompt has its ``save_to_library`` switch off (#90): the image
+            is linked to no prompt, by no fallback;
             ``("unknown", None)`` when metadata can't decide (missing, a batch
             item whose text came from another node, or a lookup failure) and the
             caller's queue and tracker fallbacks apply.
         """
         if not metadata:
             return "unknown", None
+        if metadata_opted_out(metadata):
+            return "opted_out", None
+        graph = metadata.get("prompt")
+        if isinstance(graph, dict) and any(
+            save_setting(graph, node_id) is None for node_id in run_prompt_nodes(graph)
+        ):
+            # A switch wired from a node the graph cannot read: only the node's
+            # own run (it saw the real value) can link the image (#90)
+            return "unknown", None
 
         prompt_text = None
         resolution = resolve_prompt_pair(metadata)
-        graph = metadata.get("prompt")
         if resolution["state"] == "resolved" and isinstance(graph, dict):
             positive = resolution["positive"]
             path = positive["provenance"].get("path", [])
@@ -865,13 +888,23 @@ class ImageGenerationHandler(FileSystemEventHandler):
             metadata: Extracted metadata from the image file (workflow, parameters,
                 etc.)
         """
+        if metadata_opted_out(metadata):
+            # Reached from a run or history: the file's own graph says no (#90)
+            self.logger.info(
+                f"Save to library is off for {os.path.basename(image_path)}; "
+                "left unlinked"
+            )
+            return
         try:
             resolution = resolve_prompt_pair(metadata)
+            graph = metadata.get("prompt") if isinstance(metadata, dict) else None
+            # An opted-out negative is never stored: link the positive alone
+            pair_allowed = not side_opted_out(graph, resolution.get("negative"))
 
             def link():
                 # One write; run again on the moved library when a Move
                 # library switched while it waited (#249 S6b).
-                if resolution["state"] == "resolved":
+                if resolution["state"] == "resolved" and pair_allowed:
                     positive = resolution["positive"]
                     expected = self.db_manager.get_prompt_by_hash(
                         generate_prompt_hash(

utils/prompt_graph.py

@@ -21,6 +21,11 @@ PROMPT_MANAGER_TYPES = frozenset(
     }
 )
 
+# A node's switch (#90): False means PromptManager ignores that node's runs
+SAVE_INPUT = "save_to_library"
+# Core nodes whose output is a known boolean: (class_type) -> input holding it
+_BOOLEAN_SOURCES = {"PrimitiveBoolean": "value"}
+
 # Upper bound on nodes expanded per graph; queued and embedded graphs are untrusted
 MAX_VISITS = 500
 # Longest text resolve_text will build; StringConcatenate chains can double
@@ -36,9 +41,14 @@ _TEXT_KEY = re.compile(r"^(text|string|prompt)(_?[a-z0-9]+)?$", re.IGNORECASE)
 _NEGATIVE_KEY = re.compile(r"(^|_)neg(ative)?($|_)", re.IGNORECASE)
 
 
+def is_negative_input(key: Any) -> bool:
+    """Whether an input name carries a negative prompt (#90's one rule)."""
+    return isinstance(key, str) and bool(_NEGATIVE_KEY.search(key))
+
+
 def follows_input(key: Any) -> bool:
     """Whether an input name carries positive conditioning or prompt text."""
-    if not isinstance(key, str) or _NEGATIVE_KEY.search(key):
+    if not isinstance(key, str) or is_negative_input(key):
         return False
     return key == "positive" or bool(
         _CONDITIONING_KEY.match(key) or _TEXT_KEY.match(key)
@@ -59,6 +69,116 @@ def _inputs(node: Any) -> Dict[str, Any]:
     return inputs if isinstance(inputs, dict) else {}
 
 
+def _as_boolean(value: Any) -> bool:
+    """ComfyUI's coercion of a BOOLEAN input before the node runs."""
+    if isinstance(value, dict) and "__value__" in value:
+        value = value["__value__"]
+    return bool(value)
+
+
+def save_setting(graph: Any, node_id: Any) -> Optional[bool]:
+    """A PromptManager node's ``save_to_library`` as the node will receive it.
+
+    True when the input is missing (every workflow and image from before the
+    switch), the literal coerced like ComfyUI does, the value of a linked core
+    ``PrimitiveBoolean``, and None when the switch is wired from a node whose
+    output cannot be known from the graph.
+    """
+    node = graph.get(str(node_id)) if isinstance(graph, dict) else None
+    inputs = _inputs(node)
+    if SAVE_INPUT not in inputs:
+        return True
+    value = inputs[SAVE_INPUT]
+    if not _is_link(value):
+        return _as_boolean(value)
+    source = graph.get(str(value[0]))
+    if not isinstance(source, dict) or value[1] != 0:
+        return None
+    key = _BOOLEAN_SOURCES.get(str(source.get("class_type", "")))
+    source_inputs = _inputs(source)
+    if key is None or key not in source_inputs or _is_link(source_inputs[key]):
+        return None
+    return _as_boolean(source_inputs[key])
+
+
+def opted_out(graph: Any, node_id: Any) -> bool:
+    """Whether a PromptManager node's switch is off: PromptManager ignores it."""
+    return save_setting(graph, node_id) is False
+
+
+def side_opted_out(graph: Any, side: Any) -> bool:
+    """Whether a resolved side's path passes a PromptManager node that is off.
+
+    ``side`` is a resolver side (``positive`` or ``negative`` of a resolved
+    pair): its text must not be stored when any PromptManager node on its
+    provenance path has its switch off or wired from an unreadable node (#90).
+    """
+    if not isinstance(graph, dict) or not isinstance(side, dict):
+        return False
+    provenance = side.get("provenance")
+    path = provenance.get("path") if isinstance(provenance, dict) else None
+    for step in path if isinstance(path, list) else []:
+        node_id = step.get("node_id") if isinstance(step, dict) else None
+        node = graph.get(str(node_id)) if node_id is not None else None
+        if (
+            isinstance(node, dict)
+            and node.get("class_type") in PROMPT_MANAGER_TYPES
+            and save_setting(graph, node_id) is not True
+        ):
+            return True
+    return False
+
+
+def _workflow_node_opted_out(node: Any) -> bool:
+    """A saved-workflow PromptManager node whose widget values hold ``False``.
+
+    No other PromptManager widget holds a boolean, so the value is the switch.
+    """
+    if not isinstance(node, dict):
+        return False
+    if (node.get("type") or node.get("class_type")) not in PROMPT_MANAGER_TYPES:
+        return False
+    if node.get("mode") in (2, 4):
+        return False  # muted or bypassed: it did not run
+    values = node.get("widgets_values")
+    return isinstance(values, list) and any(value is False for value in values)
+
+
+def metadata_opted_out(metadata: Any, *, unreadable: bool = False) -> bool:
+    """Whether a file's metadata says its run's prompt does not save (#90).
+
+    The executed graph (``prompt``) decides when present: a node of
+    ``positive_opt_outs``, or with ``unreadable`` also a run prompt whose
+    switch is wired from a node the graph cannot read. Without one, the
+    saved workflow (``workflow`` nodes or the extractor's
+    ``text_encoder_nodes``) is opted out when a PromptManager node stores
+    ``False``.
+    """
+    if not isinstance(metadata, dict):
+        return False
+    graph = metadata.get("prompt")
+    if isinstance(graph, dict):
+        if positive_opt_outs(graph):
+            return True
+        return unreadable and any(
+            save_setting(graph, node_id) is None for node_id in run_prompt_nodes(graph)
+        )
+    workflow = metadata.get("workflow")
+    nodes = list(metadata.get("text_encoder_nodes") or [])
+    if isinstance(workflow, dict) and isinstance(workflow.get("nodes"), list):
+        nodes.extend(workflow["nodes"])
+        definitions = workflow.get("definitions")
+        subgraphs = (
+            definitions.get("subgraphs") if isinstance(definitions, dict) else None
+        )
+        for subgraph in subgraphs if isinstance(subgraphs, list) else []:
+            inner = subgraph.get("nodes") if isinstance(subgraph, dict) else None
+            nodes.extend(inner if isinstance(inner, list) else [])
+    # Without the executed graph the node's role is unknown: any opted-out
+    # PromptManager node opts the file out (never a guessed link).
+    return any(_workflow_node_opted_out(node) for node in nodes[:MAX_VISITS])
+
+
 def _positive_roots(graph: Dict[str, Any]) -> List[Any]:
     """Links into samplers' positive inputs (KSampler, CFGGuider, BasicGuider, ...).
 
@@ -100,11 +220,21 @@ def has_sampler(graph: Any) -> bool:
 
 
 def positive_prompt_components(graph: Any) -> List[Tuple[str, str]]:
-    """Node/input identities reaching actual sampler positive wires."""
+    """Node/input identities reaching actual sampler positive wires.
+
+    A node whose switch is off is not one, and nothing upstream of it is
+    followed (#90): see ``positive_opt_outs``.
+    """
+    return _walk_positive(graph)[0]
+
+
+def _walk_positive(graph: Any) -> Tuple[List[Tuple[str, str]], List[str]]:
+    """(components of enabled nodes, opted-out nodes) on sampler positive wires."""
     if not isinstance(graph, dict):
-        return []
+        return [], []
 
     found: List[Tuple[str, str]] = []
+    skipped: List[str] = []
     seen = set()
     stack = list(reversed(_positive_roots(graph)))
     while stack and len(seen) < MAX_VISITS:
@@ -117,6 +247,10 @@ def positive_prompt_components(graph: Any) -> List[Tuple[str, str]]:
         if not isinstance(node, dict):
             continue
         if node.get("class_type") in PROMPT_MANAGER_TYPES:
+            if opted_out(graph, node_id):
+                if node_id not in skipped:
+                    skipped.append(node_id)
+                continue
             field = "text"
             if node.get("class_type") == "PromptManagerPair":
                 field = {
@@ -132,7 +266,33 @@ def positive_prompt_components(graph: Any) -> List[Tuple[str, str]]:
                 found.append(identity)
             continue
         stack.extend(reversed(_upstream_links(_inputs(node))))
-    return found
+    return found, skipped
+
+
+def _sampler_less_nodes(graph: Dict[str, Any]) -> List[str]:
+    """Every PromptManager node of a graph without a recognisable sampler."""
+    return [
+        str(node_id)
+        for node_id, node in graph.items()
+        if isinstance(node, dict) and node.get("class_type") in PROMPT_MANAGER_TYPES
+    ]
+
+
+def positive_opt_outs(graph: Any) -> List[str]:
+    """PromptManager nodes with their switch off that would be a run's prompt.
+
+    Those on a sampler's positive wires, or, in a graph without a recognisable
+    sampler, any. A run with one of them links no file to any prompt (#90):
+    the file's graph cannot always say which sampler wrote it.
+    """
+    if not isinstance(graph, dict):
+        return []
+    found, skipped = _walk_positive(graph)
+    if found or skipped or has_sampler(graph):
+        return skipped
+    return [
+        node_id for node_id in _sampler_less_nodes(graph) if opted_out(graph, node_id)
+    ]
 
 
 def positive_prompt_nodes(graph: Any) -> List[str]:
@@ -148,13 +308,13 @@ def positive_prompt_nodes(graph: Any) -> List[str]:
 
 def run_prompt_components(graph: Any) -> List[Tuple[str, str]]:
     """Actual positive components, retaining standalone main-editor fallback."""
-    components = positive_prompt_components(graph)
-    if components or not isinstance(graph, dict) or has_sampler(graph):
+    components, skipped = _walk_positive(graph)
+    if components or skipped or not isinstance(graph, dict) or has_sampler(graph):
         return components
     return [
-        (str(node_id), "text")
-        for node_id, node in graph.items()
-        if isinstance(node, dict) and node.get("class_type") in PROMPT_MANAGER_TYPES
+        (node_id, "text")
+        for node_id in _sampler_less_nodes(graph)
+        if not opted_out(graph, node_id)
     ]
 
 
@@ -165,16 +325,16 @@ def run_prompt_nodes(graph: Any) -> List[str]:
     say), every PromptManager node counts, as before 3.2.4, so an exotic workflow
     never silently loses usage counting or image linking. When a sampler exists
     and no PromptManager node feeds its positive input, nothing counts: the only
-    PromptManager nodes are negative prompts.
+    PromptManager nodes are negative prompts. A node whose switch is off never
+    counts (#90).
     """
-    positive = positive_prompt_nodes(graph)
-    if positive or not isinstance(graph, dict) or has_sampler(graph):
-        return positive
-    return [
-        str(node_id)
-        for node_id, node in graph.items()
-        if isinstance(node, dict) and node.get("class_type") in PROMPT_MANAGER_TYPES
-    ]
+    return list(
+        dict.fromkeys(
+            node_id
+            for node_id, field in run_prompt_components(graph)
+            if field == "text"
+        )
+    )
 
 
 def _join_text_inputs(
@@ -270,7 +430,18 @@ def resolve_text(graph: Any, node_id: str) -> Optional[str]:
 
 
 def intrinsic_pair_nodes(graph: Any) -> List[str]:
-    """Paired nodes whose matching outputs feed the same sampler, or standalone."""
+    """Paired nodes whose matching outputs feed the same sampler, or standalone.
+
+    A pair whose switch is off is never one (#90).
+    """
+    return [
+        node_id
+        for node_id in _intrinsic_pair_nodes(graph)
+        if not opted_out(graph, node_id)
+    ]
+
+
+def _intrinsic_pair_nodes(graph: Any) -> List[str]:
     if not isinstance(graph, dict) or len(graph) > MAX_VISITS:
         return []
     if not has_sampler(graph):

utils/prompt_inspector.py

@@ -1,4 +1,10 @@
-"""Read prompt evidence without inference, indexing, or decoding image pixels."""
+"""Read prompt evidence without inference, indexing, or decoding image pixels.
+
+``inspection_resolution`` is the inspector's resolve (inspection profile, then
+the recorded saver anchor, #386). Ingestion (the output scan, through
+``utils.file_prompt``) calls it too, only for files the default profile
+cannot decide (#247), so the inspector and ingestion cannot drift.
+"""
 
 import hashlib
 import io
@@ -571,6 +577,102 @@ def _recorded_execution(resolved: dict, texts: Mapping) -> dict | None:
     return {"prompts": prompts, "resolution": resolution}
 
 
+_ABSENT_RESOLUTION = {
+    "state": "absent",
+    "sampler_id": None,
+    "reason": "No parsed prompt metadata",
+    "reason_code": None,
+    "node": None,
+}
+_MALFORMED_RESOLUTION = {
+    "state": "unresolved",
+    "sampler_id": None,
+    "reason": MALFORMED_EVIDENCE_REASON,
+    "reason_code": "malformed",
+    "node": None,
+}
+
+
+@dataclass(frozen=True)
+class InspectionResult:
+    """The inspector's resolution of one file's metadata (#386, #247).
+
+    ``resolution`` is the result to use. ``anchored`` is the result anchored
+    on the recorded saver, kept or not (None when no anchored resolve ran).
+    ``by_record`` says the anchored result was kept: history's saver chose
+    the samplers and the result names the positive. ``malformed`` marks
+    metadata the inspector cannot read as a graph.
+    """
+
+    resolution: dict
+    anchored: dict | None = None
+    by_record: bool = False
+    malformed: bool = False
+
+
+def inspection_resolution(
+    metadata: Mapping,
+    file_name: str | None,
+    recorded_saver_id: str | None,
+    *,
+    profile: ResolverProfile = INSPECTION_PROFILE,
+    name_anchor: bool = True,
+) -> InspectionResult:
+    """Resolve raw file metadata as the inspector does.
+
+    Keeps only the role records, parses a string ``prompt`` and attaches the
+    parsed ``workflow`` only when the graph may hold a PromptManager node.
+    Resolves with ``profile``; when that leaves the prompt open and
+    ``recorded_saver_id`` is given, resolves again anchored on that saver
+    and keeps the anchored result only when it names the positive.
+
+    ``name_anchor=False`` (ingestion, #247) resolves the unanchored pass
+    without ``file_name``, so the saver its file name names anchors nothing:
+    such a file goes to Needs review, never linked (#390 ruling). The
+    recorded-saver pass still gets ``file_name``, only to refuse a saver the
+    name contradicts.
+    """
+    workflow = metadata.get("workflow")
+    canonical = {
+        key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS
+    }
+    if not canonical:
+        return InspectionResult(dict(_ABSENT_RESOLUTION))
+    malformed = InspectionResult(dict(_MALFORMED_RESOLUTION), malformed=True)
+    if "prompt" not in canonical and "parameters" not in canonical:
+        return malformed
+    if isinstance(canonical.get("prompt"), str):
+        try:
+            canonical["prompt"] = json.loads(canonical["prompt"])
+        except (ValueError, RecursionError):
+            return malformed
+    workflow = (
+        _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None
+    )
+    if workflow is not None and "prompt" in canonical:
+        # Only to recognize overwritten PromptManager text (R1); never a role.
+        canonical["workflow"] = workflow
+    resolved = resolve_prompt_pair(
+        canonical, profile=profile, file_name=file_name if name_anchor else None
+    )
+    if resolved["state"] not in _UNDECIDED or recorded_saver_id is None:
+        return InspectionResult(resolved)
+    anchored = resolve_prompt_pair(
+        canonical,
+        profile=profile,
+        file_name=file_name,
+        recorded_saver_id=recorded_saver_id,
+    )
+    # Only a result whose samplers history's saver chose is about this file.
+    if (
+        anchored.get("recorded_anchor") is True
+        and anchored["state"] not in _UNDECIDED
+        and isinstance(anchored["positive"], dict)
+    ):
+        return InspectionResult(anchored, anchored, True)
+    return InspectionResult(resolved, anchored)
+
+
 def inspect_prompt_evidence(
     metadata: dict,
     saved: dict | None = None,
@@ -593,22 +695,13 @@ def inspect_prompt_evidence(
     ``execution``.
 
     Inspection routes resolve with ``INSPECTION_PROFILE``: bounded work, not
-    graph size. Ingestion (scan, live monitor, admin import) calls the resolver
-    directly with its default profile. ``resolution`` carries the closed
-    ``reason_code`` and, for a blocking node, ``node: {id, class_type}``.
+    graph size. The resolve itself is ``inspection_resolution``, which
+    ingestion also calls for files its default profile cannot decide (#247).
+    ``resolution`` carries the closed ``reason_code`` and, for a blocking
+    node, ``node: {id, class_type}``.
     """
-    workflow = metadata.get("workflow")
-    metadata = {
-        key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS
-    }
-    resolution = {
-        "state": "unresolved",
-        "sampler_id": None,
-        "reason": MALFORMED_EVIDENCE_REASON,
-        "reason_code": "malformed",
-        "node": None,
-    }
-    if not metadata:
+    roles = {key for key in metadata if key in _ROLE_METADATA_KEYS}
+    if not roles:
         result = {role: _side("absent") for role in ("positive", "negative")}
         if saved:
             context = saved.get("pair_context")
@@ -621,56 +714,30 @@ def inspect_prompt_evidence(
                         "verified against this file."
                     ),
                 }
-        return {
-            "prompts": result,
-            "resolution": {
-                "state": "absent",
-                "sampler_id": None,
-                "reason": "No parsed prompt metadata",
-                "reason_code": None,
-                "node": None,
-            },
-        }
-    canonical = dict(metadata)
-    malformed = {
-        "prompts": _explained(
-            {role: _side("unresolved") for role in ("positive", "negative")},
-            resolution,
-        ),
-        "resolution": resolution,
-    }
-    if "prompt" not in canonical and "parameters" not in canonical:
-        return malformed
-    if isinstance(canonical.get("prompt"), str):
-        try:
-            canonical["prompt"] = json.loads(canonical["prompt"])
-        except (ValueError, RecursionError):
-            return malformed
-    workflow = (
-        _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None
+        return {"prompts": result, "resolution": dict(_ABSENT_RESOLUTION)}
+    inspected = inspection_resolution(
+        metadata, file_name, recorded_saver_id, profile=profile
     )
-    if workflow is not None and "prompt" in canonical:
-        # Only to recognize overwritten PromptManager text (R1); never a role.
-        canonical["workflow"] = workflow
-    resolved = resolve_prompt_pair(canonical, profile=profile, file_name=file_name)
-    recorded = False
-    if resolved["state"] in _UNDECIDED and recorded_saver_id is not None:
-        anchored = resolve_prompt_pair(
-            canonical,
-            profile=profile,
-            file_name=file_name,
-            recorded_saver_id=recorded_saver_id,
-        )
-        # Only a result whose samplers history's saver chose is about this file.
-        if anchored.get("recorded_anchor") is True:
-            if anchored["state"] not in _UNDECIDED and isinstance(
-                anchored["positive"], dict
-            ):
-                resolved, recorded = anchored, True
-            elif recorded_texts:
-                executed = _recorded_execution(anchored, recorded_texts)
-                if executed is not None:
-                    return executed
+    if inspected.malformed:
+        resolution = inspected.resolution
+        return {
+            "prompts": _explained(
+                {role: _side("unresolved") for role in ("positive", "negative")},
+                resolution,
+            ),
+            "resolution": resolution,
+        }
+    resolved, recorded = inspected.resolution, inspected.by_record
+    anchored = inspected.anchored
+    if (
+        not recorded
+        and recorded_texts
+        and anchored is not None
+        and anchored.get("recorded_anchor") is True
+    ):
+        executed = _recorded_execution(anchored, recorded_texts)
+        if executed is not None:
+            return executed
     state = resolved["state"]
     result = {}
     for role in ("positive", "negative"):

utils/usage_tracking.py

@@ -30,7 +30,12 @@ from typing import Any, Callable, List, Optional, Tuple
 try:
     from .hashing import generate_prompt_hash
     from .logging_config import get_logger
-    from .prompt_graph import intrinsic_pair_nodes, resolve_text, run_prompt_components
+    from .prompt_graph import (
+        intrinsic_pair_nodes,
+        resolve_text,
+        run_prompt_components,
+        save_setting,
+    )
     from .prompt_pair_resolver import resolve_text_input
     from .pair_execution import clear_pair_execution
     from .queue_observers import subscribe_dequeue
@@ -41,6 +46,7 @@ except ImportError:
         intrinsic_pair_nodes,
         resolve_text,
         run_prompt_components,
+        save_setting,
     )
     from utils.prompt_pair_resolver import resolve_text_input
     from utils.pair_execution import clear_pair_execution
@@ -127,10 +133,14 @@ def queued_prompt_runs(graph: Any) -> List[Run]:
     """(hash, text) of each positive PromptManager node whose text is knowable.
 
     Each hash once: the same text in two positive nodes is one run of the prompt.
+    A node whose switch is off is never counted; one whose switch is wired from
+    a node the graph cannot read counts itself when it executes (#90).
     """
     runs: List[Run] = []
     seen = set()
     for node_id, field in run_prompt_components(graph):
+        if save_setting(graph, node_id) is not True:
+            continue
         node = graph.get(node_id, {}) if isinstance(graph, dict) else {}
         text = (
             resolve_text_input(graph, node_id, field)
@@ -156,6 +166,8 @@ def queued_pair_runs(graph: Any) -> List[Tuple[str, str, str]]:
     """Resolve raw paired recipes only when both outputs form the used pair."""
     found, seen = [], set()
     for node_id in intrinsic_pair_nodes(graph):
+        if save_setting(graph, node_id) is not True:
+            continue  # wired switch: the node counts itself (#90)
         positive = resolve_text_input(graph, node_id, "text")
         negative = resolve_text_input(graph, node_id, "negative_text")
         if positive is None or not positive.strip() or negative is None:

web/comfy/prompt_manager.js

@@ -6,6 +6,7 @@ const { installClassicControls } = await import(api.fileURL('/prompt_manager/js/
 const { installFocusedNode, installSplitNode, installPairNode } = await import(api.fileURL('/prompt_manager/js/focused-node.js'));
 const { installPromptBridge } = await import(api.fileURL('/prompt_manager/js/prompt-bridge-host.js'));
 const { collapseLinkedWidget, isLinked, linkNoteText, primitiveTextValue, storedWidgetValue } = await import(api.fileURL('/prompt_manager/js/text-links.js'));
+const { normalizeSaveWidget } = await import(api.fileURL('/prompt_manager/js/save-to-library.js'));
 const promptManagerTypes = new Set(['PromptManager', 'PromptManagerText', 'PromptManagerFocused', 'PromptManagerSplit', 'PromptManagerPair']);
 
 // ============================================================================
@@ -525,6 +526,8 @@ app.registerExtension({
           if (originalConfigure) {
             originalConfigure.call(this, data);
           }
+          // #90: an older workflow loads with Save to library on
+          normalizeSaveWidget(this);
           if (data._userHasResized) {
             this._userHasResized = data._userHasResized;
           }

web/js/focused-node.js

@@ -2,9 +2,13 @@ import {createScope, createButton, createField, mountTheme} from './shared/toolk
 import {createPromptLibrary} from './prompt-library.js';
 import {attachMetadataAutocomplete} from './metadata-autocomplete.js';
 import {exposeTextSockets, linkNoteText, preferTextInputs, textInputAtPoint} from './text-links.js';
+import {SAVE_WIDGET, normalizeSaveWidget, placeSaveWidgetLast, saveWidget, savesToLibrary, setSaveToLibrary} from './save-to-library.js';
 
 const names = ['text', 'category', 'tags', 'search_text', 'prepend_text', 'append_text'];
 let nextEditorId = 0;
+// The editor draws these widgets itself; the host keeps them as hidden, zero-height values.
+function hiddenWidgetSize() { return [0, -4]; }
+function hideNativeWidget(widget) { widget.hidden = true; widget.type = 'hidden'; widget.computeSize = hiddenWidgetSize; }
 
 function loadStyles() {
     const href = new URL('./prompt-editor.css', import.meta.url).href;
@@ -43,10 +47,15 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
         for (const name of pair ? [...names,'negative_text'] : names) {
             let widget = this.widgets?.find(item => item.name === name);
             if (!widget) widget = this.addWidget('text', name, '', () => {}, {});
-            widget.hidden = true; widget.type = 'hidden'; widget.computeSize = () => [0, -4];
+            hideNativeWidget(widget);
             widgets.set(name, widget);
         }
         exposeTextSockets(this, textInputs);
+        // #90: the switch follows the hidden fallbacks above (one widget order on every frontend) and is drawn in the
+        // toolbar, like the other hidden widgets are drawn by the editor.
+        placeSaveWidgetLast(this);
+        const nativeSave = saveWidget(this);
+        if (nativeSave) hideNativeWidget(nativeSave);
         const element = document.createElement('div'); element.className = `ca-tools pm-focused${split ? ' pm-split' : ''}${pair ? ' pm-pair' : ''}`;
         element.setAttribute('aria-label',pair ? 'Paired prompt editor' : split ? 'Split prompt editor' : 'Focused prompt editor');
         loadStyles(); mountTheme(element, app.ui?.settings, scope);
@@ -58,6 +67,11 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
         toolbar.append(web);
         const save = pair ? createButton('Save pair','ca-primary') : null;
         if (save) toolbar.append(save);
+        const saveSwitch = document.createElement('input'); saveSwitch.type = 'checkbox'; saveSwitch.dataset.saveToLibrary = '';
+        const saveLabel = document.createElement('label'); saveLabel.className = 'pm-editor-save';
+        saveLabel.title = 'Off: runs of this node are not saved, counted or linked to their images. The output is unchanged.';
+        saveLabel.append(saveSwitch, document.createTextNode('Save to library'));
+        if (nativeSave) toolbar.append(saveLabel);
         let saving = false;
         const workspace = document.createElement('div'); workspace.className = 'pm-editor-workspace';
         const editor = document.createElement('div'); editor.className = 'pm-editor-compose';
@@ -138,7 +152,11 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
         }
         const composed = disclosure('preview','Composed text preview');
         const preview = document.createElement('p'); preview.className = 'pm-editor-preview'; preview.dataset.composedPreview = ''; composed.append(preview);
-        const footer = document.createElement('p'); footer.className = 'pm-editor-status'; footer.setAttribute('role','status'); footer.textContent = pair ? 'The prompt pair is saved when this node runs, or use Save pair.' : 'Main prompt is saved when this node runs.';
+        const footer = document.createElement('p'); footer.className = 'pm-editor-status'; footer.setAttribute('role','status');
+        const SAVE_OFF = 'Save to library is off: runs of this node are not added to the library.';
+        const SAVE_LINKED = 'Save to library comes from a link: runs are added to the library only while it is on.';
+        function runNote(saved) { return isConnected(SAVE_WIDGET) ? SAVE_LINKED : savesToLibrary(thisNode) ? saved : SAVE_OFF; }
+        footer.textContent = runNote(pair ? 'The prompt pair is saved when this node runs, or use Save pair.' : 'Main prompt is saved when this node runs.');
         const saveConnectionNotice = pair ? document.createElement('p') : null;
         if (saveConnectionNotice) {
             saveConnectionNotice.className = 'pm-editor-status'; saveConnectionNotice.setAttribute('role','status');
@@ -166,6 +184,7 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
             mainField.hidden = Boolean(isConnected('text')); connected.hidden = !isConnected('text');
             if (negative) { negative.field.hidden = Boolean(isConnected('negative_text')); negative.note.hidden = !isConnected('negative_text'); }
             if (save) { save.disabled = saving || hasConnectedPairValues(); saveConnectionNotice.hidden = !hasConnectedPairValues(); }
+            saveSwitch.checked = savesToLibrary(thisNode); saveSwitch.disabled = isConnected(SAVE_WIDGET);
             controls.forEach((input,name) => {
                 input.value = value(name);
                 if (name !== 'text' && name !== 'negative_text') {
@@ -185,7 +204,7 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
                 write('category',prompt.category || '');
                 write('tags',Array.isArray(prompt.tags) ? prompt.tags.join(', ') : prompt.tags || '');
             }
-            render(); footer.setAttribute('role','status'); footer.textContent = pair ? 'Loaded pair. Saved when this node runs.' : 'Loaded prompt. Saved when this node runs.'; main.focus();
+            render(); footer.setAttribute('role','status'); footer.textContent = runNote(pair ? 'Loaded pair. Saved when this node runs.' : 'Loaded prompt. Saved when this node runs.'); main.focus();
         }});
         workspace.append(library.element); element.append(footer);
         function snapshot() {
@@ -197,7 +216,13 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
             library.element.hidden = !split && !state.libraryOpen; browse.textContent = state.libraryOpen ? 'Hide library' : 'Browse library'; browse.setAttribute('aria-expanded', String(Boolean(state.libraryOpen)));
             library.setQuery(value('search_text')); library.restoreFilters(state.libraryFilters); render();
         }
-        controls.forEach((input,name) => scope.listen(input,'input',() => {write(name,input.value);render();footer.setAttribute('role','status');footer.textContent=pair?'The pair is saved when this node runs.':'Main prompt is saved when this node runs.';}));
+        controls.forEach((input,name) => scope.listen(input,'input',() => {write(name,input.value);render();footer.setAttribute('role','status');footer.textContent=runNote(pair?'The pair is saved when this node runs.':'Main prompt is saved when this node runs.');}));
+        scope.listen(saveSwitch,'change',() => {
+            if (isConnected(SAVE_WIDGET)) return;
+            setSaveToLibrary(thisNode, saveSwitch.checked); render();
+            footer.setAttribute('role','status'); footer.textContent = runNote(pair ? 'The pair is saved when this node runs.' : 'Main prompt is saved when this node runs.');
+            thisNode.setDirtyCanvas(true,true); app.graph?.setDirtyCanvas(true,true);
+        });
         scope.listen(browse,'click',() => {
             library.element.hidden = !library.element.hidden;
             library.closeCompletions();
@@ -248,6 +273,8 @@ export function installFocusedNode(nodeType, app, api, parseQuery, {layout = 'fo
         const result = configure?.apply(this,args);
         // A saved workflow restores its inputs with their widget bindings; keep the text inputs as sockets.
         exposeTextSockets(this, textInputs);
+        // #90: an older workflow loads with Save to library on
+        normalizeSaveWidget(this);
         this._focusedView?.restore();
         return result;
     };

web/js/prompt-editor.css

@@ -2,6 +2,8 @@
 .pm-focused [hidden] { display: none !important; }
 .pm-editor-row { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; }
 .pm-editor-toolbar > strong { margin-inline-end: auto; font-weight: 500; }
+.pm-editor-save { display: inline-flex; align-items: center; gap: 6px; color: var(--ca-muted); cursor: pointer; }
+.pm-editor-save:has(input:checked) { color: inherit; }
 .pm-focused .ca-field-label { align-self: flex-start; margin-inline: 0; text-align: start; }
 .pm-editor-workspace, .pm-editor-compose { display: contents; }
 .pm-focused .pm-editor-main { display: flex; flex: 1; min-block-size: 180px; flex-direction: column; align-items: stretch; gap: 5px; text-align: start; }

web/js/save-to-library.js

@@ -0,0 +1,47 @@
+// #90: a PromptManager node's `save_to_library` switch. Off: PromptManager ignores the node's runs (nothing saved,
+// counted or linked); the node's output is unchanged.
+//
+// The input is declared last, so it takes the widget index an older workflow used for the editor's own
+// (non-serialized) DOM widget, where it stored `''` or nothing. ComfyUI coerces a BOOLEAN with `bool(value)`, so that
+// `''` would turn saving off: after a workflow loads, anything that is not a boolean is the default, on.
+
+export const SAVE_WIDGET = 'save_to_library';
+
+export function saveWidget(node) {
+    return node?.widgets?.find(widget => widget?.name === SAVE_WIDGET);
+}
+
+/** Whether the node saves its runs (a missing widget, as on an older backend, is on). */
+export function savesToLibrary(node) {
+    return saveWidget(node)?.value !== false;
+}
+
+/** Set a non-boolean switch value (from an older workflow) to on. Returns whether it changed. */
+export function normalizeSaveWidget(node) {
+    const widget = saveWidget(node);
+    if (!widget || typeof widget.value === 'boolean') return false;
+    widget.value = true;
+    const index = node.widgets.indexOf(widget);
+    if (Array.isArray(node.widgets_values) && index >= 0) node.widgets_values[index] = true;
+    return true;
+}
+
+/** Move the switch after every other widget the node has now, so its index is the same on every frontend. */
+export function placeSaveWidgetLast(node) {
+    const widget = saveWidget(node);
+    const widgets = node?.widgets;
+    if (!widget || widgets.at(-1) === widget) return false;
+    widgets.splice(widgets.indexOf(widget), 1);
+    widgets.push(widget);
+    return true;
+}
+
+/** Turn the switch on or off the way a native widget edit does. */
+export function setSaveToLibrary(node, value) {
+    const widget = saveWidget(node);
+    if (!widget) return;
+    widget.value = Boolean(value);
+    const index = node.widgets.indexOf(widget);
+    if (Array.isArray(node.widgets_values) && index >= 0) node.widgets_values[index] = widget.value;
+    widget.callback?.(widget.value, undefined, node);
+}