test(inspector): cover every plain-first-stage outcome on both resolvers (#381)

The frontend coverage gate found local-resolver-stages.mjs branches at 92.44%
(< 93). Each way an earlier stage stops being a plain first stage (no latent,
literal, missing or cyclic latent, a sampler-shaped node, a detailer) is now a
resolver test and an anchor-first-stage-* parity vector.

TESTING.md

@@ -745,11 +745,55 @@ with a literal true `strip_newlines` is `replace("\n", "").strip()`;
 (`N` = `inputcount`) after the literal delimiter unless missing or exactly `""`, and
 ignores later inputs. A linked setting is `unsupported-node`; a list output
 (`return_list` true or linked) and `PromptSearchList` are `runtime-value` ("Text is
-decided at run time"), never one list item. Unverified text nodes (LoRAfromText,
-RegexReplace, XYZPrompt, ImpactWildcardProcessor) stay `unsupported-node`. Every
+decided at run time"), never one list item. Every
 intermediate text is length-checked and charged to the scanned-character budget. If
 the table cannot be loaded, graph inspection is `unresolved` ("Inspection node
 semantics are unavailable") and ingestion is unaffected, because it never imports the table.
+Unknown nodes pass through (#381, maintainer-approved 2026-10-27; no per-node
+inventory): an output whose class the table, its `facts`, `lineage` and the code's
+adapters never name is read through, not refused. A `text` input is its text (the
+legacy encoder rule, so RegexReplace or LoRAfromText show their input text), except
+that a string output of a node with a seed input (`seed`, `noise_seed`,
+`sampling_mode.seed`) and no linked `conditioning`/`positive`/`negative` input is
+generated at run time (a language model such as TextGenerate, a wildcard):
+`runtime-value` naming it. Else, at output slot 0 or 1 of a node with a `positive` or
+`negative` input, only `positive` (slot 1, `negative`) is followed, nothing when it
+is not linked; else a linked `conditioning` alone; else every linked input except
+the table's `pass_through.skip_inputs` (core type inputs: model, clip, vae, image,
+latent, ...), each traced with kind `any`. A followed input with no text adds
+nothing; limits and selector choices propagate; another failure (`runtime-value`,
+`composite`, `malformed`) blocks, unless a PromptManager node was reached: this
+repository's own nodes are authoritative, and then only their texts count. One
+distinct text resolves; several make the sampler `ambiguous` "Several prompts reach
+the sampler" with one candidate per distinct outcome, in input order (at most 32,
+else `work-limit`). A PromptManagerPair's execution record is what ran: while its
+inputs are checked against it, a walk through an unknown node that fails, diverges
+or passes text through is unverifiable and never overrides the record. No text
+at all is `unsupported-node` naming the outermost unknown node. An unlisted output
+of a listed class (a sampler, a loader, a lineage node) stays `unsupported-node`, so
+the walk never reads another stage's prompt. `INSPECTION_RESOLVER_VERSION` is 2 for
+this rule; ingestion keeps the default profile (`RESOLVER_VERSION` 1). The legacy
+parity gate (`tests/test_legacy_parity_gate.py`) records, per corpus (the
+generation-params vectors, the inspection parity fixtures, the local resolver golden
+and selection cases, the unknown-node shapes in `tests/unknown_node_graphs.py`),
+every graph whose inspection positive equals the legacy parser's; a graph may join
+that list (`python3 -m tests.test_legacy_parity_gate --write`), never leave it. A graph
+may read differently from legacy only when `LEGACY_WRONG` there states why (legacy's
+first-sampler pick on a branch the file name identifies).
+Saver anchoring (#381): an inspected file's name (`path.name` on the server, the chosen
+file's name locally) anchors an unanchored inspection on the one node whose
+`filename_prefix` (literal, or traced as a string) has a last path part that makes the
+file's stem `<prefix>_<digits>` with an optional trailing `_`. That node's linked inputs
+are walked like a detailer's image ancestry, through any image node, and the listed
+samplers found are the candidates. No match, several matches, a walk that fails or no
+sampler found means no anchor: nothing is guessed and the file stays "several possible
+prompts". Ingestion ignores file names. Shared-conditioning stages: in a sampler's
+latent walk, an earlier sampler reading the very same conditioning links (positive and
+negative, or a guider; compared by node ID text and slot) is the same prompt, not an
+unattributed stage, when it is a plain first stage: not a detailer, no prompt override
+set, and no listed sampler in its own latent ancestry (else it stays unattributed) (`tests/test_resolver_saver_anchor.py`, shapes in
+`tests/saver_anchor_graphs.py` from the maintainer's `gpu_unlock_bench`, dataset and
+ANIMA runs; browser parity through the `anchor-*` cases, which carry `__file_name`).
 
 `tools/inspection_corpus.py --root DIR --out JSONL [--workers N] [--baseline JSONL]
 [--summary TXT] [--code TREE]` measures a media folder read-only against the #240

py/api/inspection_service.py

@@ -129,8 +129,9 @@ def inspect_current_file(
         profile = (
             PREVIEW_INSPECTION_PROFILE if _in_host_temp(path) else INSPECTION_PROFILE
         )
+        # The file name anchors on the saver that wrote it (#381).
         evidence = inspect_prompt_evidence(
-            inspection.metadata, role_saved, profile=profile
+            inspection.metadata, role_saved, profile=profile, file_name=path.name
         )
     return {
         "file": {

tests/browser/fast_reset.py

@@ -1,283 +0,0 @@
-"""Run a browser-test fixture seeder off the live database, then publish once (#384).
-
-The seeders make hundreds of small commits. On a slow CI disk the live
-database's commits queue behind WAL checkpoint syncs and a reset took over
-30 s. ``ScratchSeeder.run`` instead:
-
-1. copies the live database into a scratch database (SQLite backup API) whose
-   connections never sync (a crash loses only the copy);
-2. runs the unchanged seeder against an API proxy whose ``db`` is the scratch
-   database; everything else (files, thumbnails, caches, command state) is
-   the live API's, written exactly as before;
-3. replaces the rows of every table the seed changed in the live database, in
-   one transaction on the live connection. Tables the seed left alone are not
-   touched, so state another fixture owns stays as it was.
-
-A commit by any other live connection while the seed ran fails the reset
-instead of being overwritten (``PRAGMA data_version``).
-"""
-
-from __future__ import annotations
-
-import hashlib
-import os
-import shutil
-import sqlite3
-import tempfile
-import weakref
-from pathlib import Path
-from typing import Any, Callable, TypeVar
-
-T = TypeVar("T")
-
-# Set to 1 to run seeders on the live database as before #384 (the equivalence
-# proof, tests/browser/reset_equivalence.py, compares both ways).
-SEED_ON_LIVE_ENV = "PM_TEST_SEED_ON_LIVE"
-SCRATCH_SCHEMA = "reset_seed"
-# API calls a seed makes that must see the published rows (see _SeedApi).
-DEFERRED_CALLS = frozenset({"invalidate_gallery_cache"})
-# Linux's shared-memory tmpfs (present in the Playwright image and on runners).
-MEMORY_ROOT = Path("/dev/shm")
-# The largest fixtures' scratch copy is about 8 MB (database plus a WAL that
-# autocheckpoints near 4 MB). Docker's default /dev/shm is 64 MB, shared with
-# the browsers and the other workers; with less free than this, the scratch
-# copy goes on disk instead.
-MEMORY_FREE_BYTES = 32 * 1024 * 1024
-# Scratch folders are named <prefix><owner pid>-<random>.
-SCRATCH_PREFIX = "pm-reset-"
-
-
-def _quote(name: str) -> str:
-    return '"' + name.replace('"', '""') + '"'
-
-
-def _tables(conn: sqlite3.Connection, schema: str = "main") -> list[str]:
-    rows = conn.execute(
-        f"SELECT name, sql FROM {schema}.sqlite_master WHERE type = 'table' "
-        "ORDER BY name"
-    ).fetchall()
-    for name, sql in rows:
-        if sql and sql.lstrip().upper().startswith("CREATE VIRTUAL"):
-            raise NotImplementedError(f"virtual table {name} cannot be copied")
-    return [name for name, _sql in rows]
-
-
-def _columns(conn: sqlite3.Connection, table: str, schema: str = "main") -> list[str]:
-    """Writable columns (hidden and generated ones excluded), plus ``rowid``
-    when the table has a rowid that no INTEGER PRIMARY KEY column carries."""
-    info = conn.execute(f"PRAGMA {schema}.table_xinfo({_quote(table)})").fetchall()
-    columns = [row[1] for row in info if row[6] == 0]
-    keys = [row for row in info if row[5]]
-    sql = conn.execute(
-        f"SELECT sql FROM {schema}.sqlite_master WHERE type = 'table' AND name = ?",
-        (table,),
-    ).fetchone()[0]
-    without_rowid = sql is not None and "WITHOUT ROWID" in sql.upper()
-    integer_key = len(keys) == 1 and str(keys[0][2]).upper() == "INTEGER"
-    if not without_rowid and not integer_key and table != "sqlite_sequence":
-        columns.insert(0, "rowid")
-    return columns
-
-
-def table_digests(conn: sqlite3.Connection, schema: str = "main") -> dict[str, str]:
-    """A digest of every table's rows (rowids included), by table name."""
-    result = {}
-    for table in _tables(conn, schema):
-        selected = ", ".join(
-            _quote(c) if c != "rowid" else "rowid"
-            for c in _columns(conn, table, schema)
-        )
-        digest = hashlib.sha256()
-        rows = conn.execute(f"SELECT {selected} FROM {schema}.{_quote(table)}")
-        for row in sorted(repr(tuple(row)) for row in rows):
-            digest.update(row.encode("utf-8", "surrogatepass"))
-            digest.update(b"\n")
-        result[table] = digest.hexdigest()
-    return result
-
-
-def _process_alive(pid: int) -> bool:
-    try:
-        os.kill(pid, 0)
-    except ProcessLookupError:
-        return False
-    except OSError:
-        return True  # it exists but belongs to someone else
-    return True
-
-
-def remove_orphaned_scratch(root: Path) -> None:
-    """Remove scratch folders of servers that died without cleaning up.
-
-    A server killed outright (SIGKILL, a crash) never runs ``close``, and
-    tmpfs keeps its folder in memory until reboot. The folder name carries
-    the owner's process id; a folder whose process is gone is removed.
-    """
-    if os.name != "posix":
-        return  # os.kill(pid, 0) would terminate the process on Windows
-    for folder in root.glob(f"{SCRATCH_PREFIX}*-*"):
-        pid = folder.name[len(SCRATCH_PREFIX) :].split("-", 1)[0]
-        if pid.isdigit() and not _process_alive(int(pid)):
-            shutil.rmtree(folder, ignore_errors=True)
-
-
-class _SeedApi:
-    """The live API, except that ``db`` is the scratch database.
-
-    Cache drops the seed asks for (``DEFERRED_CALLS``) are recorded and made
-    after the publish: dropped earlier, a request in between would cache the
-    live database's old rows again.
-    """
-
-    def __init__(self, api: Any, db: Any) -> None:
-        object.__setattr__(self, "_api", api)
-        object.__setattr__(self, "db", db)
-        object.__setattr__(self, "deferred", [])
-
-    def __getattr__(self, name: str) -> Any:
-        if name in DEFERRED_CALLS:
-
-            def record(*args: Any, **kwargs: Any) -> None:
-                self.deferred.append((name, args, kwargs))
-
-            return record
-        return getattr(self._api, name)
-
-    def __setattr__(self, name: str, value: Any) -> None:
-        setattr(self._api, name, value)
-
-
-class ScratchSeeder:
-    """Seed into a scratch copy and publish the changed tables in one commit."""
-
-    def __init__(
-        self,
-        api: Any,
-        directory: Path,
-        *,
-        on_live: bool = False,
-        memory_root: Path | None = MEMORY_ROOT,
-    ) -> None:
-        self._api = api
-        self._fallback = Path(directory)
-        self._memory_root = memory_root
-        self._directory: Path | None = None
-        self._on_live = on_live
-        self._scratch = None
-        self._unsynced: "weakref.WeakSet[sqlite3.Connection]" = weakref.WeakSet()
-
-    def run(self, seed: Callable[[Any], T]) -> T:
-        """Run ``seed(api)`` (blocking; call it in an executor thread)."""
-        if self._on_live:
-            return seed(self._api)
-        live = self._api.db.model.get_connection()
-        if live.in_transaction:
-            raise RuntimeError("the live connection has an open transaction")
-        version = live.execute("PRAGMA data_version").fetchone()[0]
-        scratch = self._scratch_database()
-        target = self._scratch_connection(scratch)
-        with target.using():
-            live.backup(target)
-        before = table_digests(target)
-        target_api = _SeedApi(self._api, scratch)
-        result = seed(target_api)
-        target = self._scratch_connection(scratch)
-        if target.in_transaction:
-            raise RuntimeError("the seed left a scratch transaction open")
-        after = table_digests(target)
-        changed = [table for table in after if after[table] != before.get(table)]
-        if set(before) != set(after):
-            raise RuntimeError("the seed changed the database schema")
-        if live.execute("PRAGMA data_version").fetchone()[0] != version:
-            raise RuntimeError("the live database changed while the seed ran")
-        self._publish(live, scratch.model.db_path, changed, version)
-        for name, args, kwargs in target_api.deferred:
-            getattr(self._api, name)(*args, **kwargs)
-        return result
-
-    def _scratch_database(self):
-        if self._scratch is None:
-            from database.operations import PromptDatabase
-
-            self._directory = self._scratch_directory()
-            self._scratch = PromptDatabase(str(self._directory / "seed.db"))
-        return self._scratch
-
-    def _scratch_connection(self, scratch) -> sqlite3.Connection:
-        """The calling thread's scratch connection, set never to sync."""
-        conn = scratch.model.get_connection()
-        if conn not in self._unsynced:
-            conn.execute("PRAGMA synchronous = OFF")
-            self._unsynced.add(conn)
-        return conn
-
-    @staticmethod
-    def _publish(
-        live: sqlite3.Connection, path: str, tables: list[str], version: int
-    ) -> None:
-        """Replace ``tables`` with the scratch rows in one live transaction.
-
-        ``version`` is the live ``data_version`` read before the seed. It is
-        checked again under the write lock: a writer whose transaction was
-        open during the seed commits before ``BEGIN IMMEDIATE`` returns, and
-        its rows must fail the reset instead of being overwritten.
-        """
-        if not tables:
-            return
-        foreign_keys = live.execute("PRAGMA foreign_keys").fetchone()[0]
-        # Rows are copied as they are; no cascade may reach a table left alone.
-        live.execute("PRAGMA foreign_keys = OFF")
-        try:
-            live.execute(f"ATTACH DATABASE ? AS {SCRATCH_SCHEMA}", (path,))
-            try:
-                live.execute("BEGIN IMMEDIATE")
-                try:
-                    if live.execute("PRAGMA data_version").fetchone()[0] != version:
-                        raise RuntimeError(
-                            "the live database changed while the seed ran"
-                        )
-                    for table in tables:
-                        columns = ", ".join(
-                            _quote(c) if c != "rowid" else "rowid"
-                            for c in _columns(live, table)
-                        )
-                        name = _quote(table)
-                        live.execute(f"DELETE FROM main.{name}")
-                        live.execute(
-                            f"INSERT INTO main.{name} ({columns}) "
-                            f"SELECT {columns} FROM {SCRATCH_SCHEMA}.{name}"
-                        )
-                    live.commit()
-                except BaseException:
-                    live.rollback()
-                    raise
-            finally:
-                live.execute(f"DETACH DATABASE {SCRATCH_SCHEMA}")
-        finally:
-            live.execute(f"PRAGMA foreign_keys = {int(foreign_keys)}")
-
-    def close(self) -> None:
-        """Close every scratch connection, then remove the scratch files."""
-        if self._on_live:
-            return
-        if self._scratch is not None:
-            self._scratch.close_all()
-            self._scratch = None
-        if self._directory is not None:
-            shutil.rmtree(self._directory, ignore_errors=True)
-            self._directory = None
-
-    def _scratch_directory(self) -> Path:
-        """A folder in memory (tmpfs) where there is one, else ``directory``.
-
-        Even unsynced, scratch writes on the runner's disk wait whenever the
-        kernel is writing back the pages they rewrite.
-        """
-        root = self._memory_root
-        if root is not None and root.is_dir() and os.access(root, os.W_OK):
-            remove_orphaned_scratch(root)
-            if shutil.disk_usage(root).free >= MEMORY_FREE_BYTES:
-                prefix = f"{SCRATCH_PREFIX}{os.getpid()}-"
-                return Path(tempfile.mkdtemp(prefix=prefix, dir=root))
-        self._fallback.mkdir(parents=True, exist_ok=True)
-        return self._fallback

tests/browser/folder_trash.py

@@ -1,78 +0,0 @@
-"""Empty a browser-test media folder without waiting for the disk (#384).
-
-A reset used to ``shutil.rmtree`` the output and input folders the last
-fixture wrote. Unlinking a file whose pages are still being written back
-waits for that write, so on a slow CI disk a reset stalled in ``rmtree``.
-``replace_folder`` renames the old folder into ``<parent>/reset-trash`` (one
-metadata change) and makes a new empty one; a background thread deletes the
-trash, and ``close_trash`` stops that thread and removes what is left before
-the server's temp root goes.
-"""
-
-from __future__ import annotations
-
-import queue
-import shutil
-import threading
-import uuid
-from pathlib import Path
-
-TRASH_NAME = "reset-trash"
-
-
-class _Trash:
-    def __init__(self, directory: Path) -> None:
-        self.directory = directory
-        self._queue: "queue.SimpleQueue[Path | None]" = queue.SimpleQueue()
-        self._thread = threading.Thread(
-            target=self._work, name="pm-reset-trash", daemon=True
-        )
-        self._thread.start()
-
-    def _work(self) -> None:
-        while (folder := self._queue.get()) is not None:
-            shutil.rmtree(folder, ignore_errors=True)
-
-    def discard(self, folder: Path) -> None:
-        self.directory.mkdir(exist_ok=True)
-        moved = self.directory / f"{folder.name}-{uuid.uuid4().hex}"
-        folder.rename(moved)
-        self._queue.put(moved)
-
-    def close(self) -> None:
-        self._queue.put(None)
-        self._thread.join()
-        shutil.rmtree(self.directory, ignore_errors=True)
-
-
-_trashes: dict[Path, _Trash] = {}
-# Trash folders whose server has cleaned up: a reset still running then
-# deletes in place instead of starting a thread nobody would stop.
-_closed: set[Path] = set()
-_lock = threading.Lock()
-
-
-def replace_folder(folder: Path) -> None:
-    """Make ``folder`` an empty folder; its old content is deleted later."""
-    folder = Path(folder)
-    if folder.exists():
-        directory = folder.parent / TRASH_NAME
-        with _lock:
-            trash = _trashes.get(directory)
-            if trash is None and directory not in _closed:
-                trash = _trashes[directory] = _Trash(directory)
-            if trash is not None:
-                trash.discard(folder)
-        if folder.exists():
-            shutil.rmtree(folder)
-    folder.mkdir()
-
-
-def close_trash(parent: Path) -> None:
-    """Stop deleting under ``parent``'s trash and remove it (blocking)."""
-    directory = Path(parent) / TRASH_NAME
-    with _lock:
-        _closed.add(directory)
-        trash = _trashes.pop(directory, None)
-    if trash is not None:
-        trash.close()

tests/browser/reset_equivalence.py

@@ -1,279 +0,0 @@
-"""Prove the scratch-seeded resets leave the state the old resets left (#384).
-
-Run by hand on Linux (the browser server's platform):
-
-    python -m tests.browser.reset_equivalence
-
-It starts the browser server three times: twice seeding on the live database
-as before #384 (``PM_TEST_SEED_ON_LIVE=1``) and once with the scratch seeder.
-Each server gets the same sequence: every reset kind of ``/__test__/reset``
-and ``/__test__/concept/reset``, with other fixtures' resets and API writes
-between them. After every reset it reads every table's rows (by rowid), the
-output and input file trees (size, SHA-256, mtime) and the response body.
-
-Values that differ between the two old runs are wall-clock values (created_at,
-an unpinned concept "now", file mtimes); they are masked. Everything else must
-be equal between the old runs and the new one. Exit status 0 means equal.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-import sqlite3
-import subprocess
-import sys
-import tempfile
-import time
-import urllib.request
-from pathlib import Path
-
-REPO_ROOT = Path(__file__).resolve().parents[2]
-READY_TIMEOUT_S = 60
-CLOCK = "<clock>"
-
-RESETS = [
-    ("/__test__/reset", {}),
-    ("/__test__/reset", {"extra_images": 2}),
-    ("/__test__/reset", {"extra_images": 108}),
-    ("/__test__/reset", {"include_video": True}),
-    ("/__test__/reset", {"prompt_inspector": True}),
-    ("/__test__/reset", {"native_library": True}),
-    ("/__test__/reset", {"native_library": True, "extra_prompts": 125}),
-    ("/__test__/reset", {"native_library": True, "extra_variants": 41}),
-    ("/__test__/reset", {"workspace_visual": True}),
-    ("/__test__/reset", {"full_inspector": True}),
-    (
-        "/__test__/reset",
-        {"full_inspector": True, "include_video": True, "extra_images": 2},
-    ),
-    ("/__test__/reset", {"chronological_history": True}),
-    ("/__test__/reset", {"chronological_history": True, "history_runs": 85}),
-    ("/__test__/concept/reset", {}),
-    ("/__test__/concept/reset", {"extra_images": 100}),
-    ("/__test__/concept/reset", {"seedless_images": 2, "short_seedless": True}),
-    ("/__test__/concept/reset", {"now": 1_780_000_000}),
-]
-# Writes by other fixtures and the API between resets; the resets must keep or
-# replace them exactly as before.
-DIRTY = [
-    ("/__test__/tagging/reset", {}),
-    ("/__test__/system/reset", None),
-    (
-        "/__test__/saved-views/raw",
-        {"page": "images", "name": "dirty", "state": "{}"},
-    ),
-]
-
-
-class Server:
-    def __init__(self, on_live: bool) -> None:
-        self.tmp = tempfile.TemporaryDirectory(prefix="pm-equivalence-")
-        env = {**os.environ, "TMPDIR": self.tmp.name}
-        env.update(TEMP=self.tmp.name, TMP=self.tmp.name)
-        env.pop("PM_TEST_SEED_ON_LIVE", None)
-        if on_live:
-            env["PM_TEST_SEED_ON_LIVE"] = "1"
-        self.process = subprocess.Popen(
-            [
-                sys.executable,
-                "-m",
-                "tests.browser.server",
-                "--port",
-                "0",
-                "--exit-on-stdin-eof",
-            ],
-            cwd=REPO_ROOT,
-            env=env,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.DEVNULL,
-        )
-        line = self.process.stdout.readline().decode()
-        if not line.startswith("PM_TEST_SERVER_PORT="):
-            raise RuntimeError(f"no port line: {line!r}")
-        self.base = f"http://127.0.0.1:{int(line.split('=', 1)[1])}"
-        deadline = time.monotonic() + READY_TIMEOUT_S
-        while True:
-            try:
-                self.post_json("GET", "/prompt_manager/test", None)
-                break
-            except OSError:
-                if time.monotonic() > deadline or self.process.poll() is not None:
-                    raise
-                time.sleep(0.1)
-        (self.root,) = Path(self.tmp.name).glob("pm-browser-*")
-
-    def post_json(self, method: str, path: str, data) -> dict:
-        body = None if data is None else json.dumps(data).encode()
-        request = urllib.request.Request(
-            self.base + path,
-            data=body if method == "POST" else None,
-            method=method,
-            headers={"Content-Type": "application/json"} if body else {},
-        )
-        with urllib.request.urlopen(request, timeout=120) as response:  # nosec B310
-            text = response.read().decode()
-        try:
-            return json.loads(text)
-        except ValueError:
-            return {"text": text[:200]}
-
-    def stop(self) -> None:
-        self.process.stdin.close()
-        self.process.wait(timeout=READY_TIMEOUT_S)
-        self.process.stdout.close()
-        self.tmp.cleanup()
-
-
-def read_tables(path: Path) -> dict:
-    conn = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)
-    try:
-        tables = {}
-        names = [
-            row[0]
-            for row in conn.execute(
-                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
-            )
-        ]
-        for name in names:
-            quoted = '"' + name.replace('"', '""') + '"'
-            columns = [row[1] for row in conn.execute(f"PRAGMA table_info({quoted})")]
-            try:
-                cursor = conn.execute(f"SELECT rowid, * FROM {quoted} ORDER BY rowid")
-                columns = ["rowid", *columns]
-            except sqlite3.OperationalError:
-                cursor = conn.execute(f"SELECT * FROM {quoted} ORDER BY 1")
-            rows = [
-                [value.hex() if isinstance(value, bytes) else value for value in row]
-                for row in cursor
-            ]
-            tables[name] = {"columns": columns, "rows": rows}
-        return tables
-    finally:
-        conn.close()
-
-
-def read_files(root: Path) -> dict:
-    files = {}
-    for top in ("output", "input"):
-        for path in sorted((root / top).rglob("*")):
-            if path.is_file():
-                stat = path.stat()
-                files[path.relative_to(root).as_posix()] = [
-                    stat.st_size,
-                    hashlib.sha256(path.read_bytes()).hexdigest(),
-                    stat.st_mtime_ns,
-                ]
-    return files
-
-
-def flatten(value, prefix="") -> dict:
-    if isinstance(value, dict):
-        out = {}
-        for key, item in value.items():
-            out.update(flatten(item, f"{prefix}/{key}"))
-        return out
-    if isinstance(value, list):
-        out = {f"{prefix}#len": len(value)}
-        for index, item in enumerate(value):
-            out.update(flatten(item, f"{prefix}[{index}]"))
-        return out
-    return {prefix: value}
-
-
-def snapshot(server: Server, response: dict) -> dict:
-    flat = {}
-    for table, data in read_tables(server.root / "browser.db").items():
-        flat[f"table {table}#rows"] = len(data["rows"])
-        for index, row in enumerate(data["rows"]):
-            for column, value in zip(data["columns"], row):
-                flat[f"table {table}[{index}].{column}"] = value
-    for name, (size, digest, mtime) in read_files(server.root).items():
-        flat[f"file {name}"] = [size, digest]
-        flat[f"file {name}#mtime"] = mtime
-    flat["files#count"] = sum(
-        1 for key in flat if key.startswith("file ") and "#" not in key
-    )
-    for key, value in flatten(response).items():
-        flat[f"response{key}"] = value
-    # Each server has its own temp root and a random storage epoch; name them
-    # instead of masking every column that carries them.
-    epoch = flat.get("table execution_history_metadata[0].storage_epoch")
-    names = [(str(server.root), "<root>")]
-    if isinstance(epoch, str) and epoch:
-        names.append((epoch, "<epoch>"))
-    for key, value in flat.items():
-        if isinstance(value, str):
-            for text, name in names:
-                value = value.replace(text, name)
-            flat[key] = value
-    return flat
-
-
-def run(on_live: bool) -> list[dict]:
-    server = Server(on_live)
-    try:
-        states = []
-        for index, (path, data) in enumerate(RESETS * 2):
-            dirty_path, dirty_data = DIRTY[index % len(DIRTY)]
-            server.post_json("POST", dirty_path, dirty_data)
-            started = time.monotonic()
-            response = server.post_json("POST", path, data)
-            elapsed = time.monotonic() - started
-            print(f"  {'old' if on_live else 'new'} {path} {data}: {elapsed:.2f}s")
-            states.append(snapshot(server, response))
-        return states
-    finally:
-        server.stop()
-
-
-def column_key(key: str) -> str:
-    """``table t[3].created_at`` -> ``table t.created_at`` (mask whole columns)."""
-    if key.startswith("table ") and "[" in key:
-        head, _, tail = key.partition("[")
-        return head + tail.partition("]")[2]
-    return key
-
-
-def main() -> int:
-    old_a, old_b, new = run(True), run(True), run(False)
-    failures = 0
-    masked_all = set()
-    for step, (path, data) in enumerate(RESETS * 2):
-        a, b, n = old_a[step], old_b[step], new[step]
-        masked = {
-            column_key(key)
-            for key in set(a) | set(b)
-            if a.get(key, CLOCK) != b.get(key, CLOCK)
-        }
-        masked_all |= masked
-        keys = set(a) | set(n)
-        diffs = [
-            key
-            for key in sorted(keys)
-            if column_key(key) not in masked
-            and a.get(key, "<missing>") != n.get(key, "<missing>")
-        ]
-        tables = len(
-            {k.split("[")[0].split("#")[0] for k in a if k.startswith("table ")}
-        )
-        status = "EQUAL" if not diffs else f"DIFFERENT ({len(diffs)} values)"
-        print(
-            f"{status}: {path} {json.dumps(data)}: {tables} tables, "
-            f"{a['files#count']} files, {len(keys)} values, "
-            f"{len(masked)} masked columns"
-        )
-        for key in diffs[:20]:
-            old, now = a.get(key, "<missing>"), n.get(key, "<missing>")
-            print(f"    {key}: old {old!r} new {now!r}")
-        failures += bool(diffs)
-    print("masked (differ between the two old runs):")
-    for key in sorted(masked_all):
-        print(f"    {key}")
-    return 1 if failures else 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

tests/browser/server.py

@@ -19,10 +19,8 @@ from aiohttp.web_runner import GracefulExit
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
-from tests.browser.folder_trash import close_trash, replace_folder
 from tests.browser.request_hold import register_request_holds
 from tests.browser import slow_reset_watch
-from tests.browser.fast_reset import SEED_ON_LIVE_ENV, ScratchSeeder
 from tests.browser.workspace_activity_fixture import register_activity_commands
 from tests.browser.workspace_system_fixture import register_system_commands
 from tests.browser.workspace_integrations_fixture import (
@@ -227,10 +225,6 @@ def create_app(root: Path) -> web.Application:
     register_integrations_commands(app, routes, api, root)
     request_holds = register_request_holds(app, routes)
     slow_reset_watch.install(app, api)
-    # Resets seed a scratch copy and publish it in one commit (#384).
-    seeder = ScratchSeeder(
-        api, root / "reset-scratch", on_live=os.environ.get(SEED_ON_LIVE_ENV) == "1"
-    )
 
     async def canonical_asset(request: web.Request) -> web.Response:
         filename, content_type = CANONICAL_ASSETS[request.path]
@@ -244,7 +238,6 @@ def create_app(root: Path) -> web.Application:
         routes.get(path)(canonical_asset)
 
     def seed(
-        target,
         extra_images: int = 0,
         include_video: bool = False,
         prompt_inspector: bool = False,
@@ -252,7 +245,7 @@ def create_app(root: Path) -> web.Application:
         extra_prompts: int = 0,
         extra_variants: int = 0,
     ) -> None:
-        with target.db.model.get_connection() as conn:
+        with api.db.model.get_connection() as conn:
             for table in (
                 "saved_views",
                 "image_favorites",
@@ -271,7 +264,8 @@ def create_app(root: Path) -> web.Application:
             ):
                 conn.execute(f"DELETE FROM {table}")
         for name in ("output", "input"):
-            replace_folder(root / name)
+            shutil.rmtree(root / name)
+            (root / name).mkdir()
         for index, (directory, text, tag) in enumerate(
             [
                 ("output", "Cat portrait", "cats"),
@@ -314,30 +308,28 @@ def create_app(root: Path) -> web.Application:
                     image.save(image_path, pnginfo=info)
             stamp = time.time() - index * 10
             os.utime(image_path, (stamp, stamp))
-            pid = target.db.save_prompt(
+            pid = api.db.save_prompt(
                 text, tags=[tag], prompt_hash=generate_prompt_hash(text)
             )
-            target.db.link_image_to_prompt(pid, str(image_path))
+            api.db.link_image_to_prompt(pid, str(image_path))
             if index == 0:
                 video_prompt_id = pid
                 for number in range(extra_images):
                     extra = root / directory / "bulk" / f"{number:04}.png"
                     extra.parent.mkdir(exist_ok=True)
                     shutil.copyfile(image_path, extra)
-                    target.db.link_image_to_prompt(pid, str(extra))
+                    api.db.link_image_to_prompt(pid, str(extra))
         if native_library:
-            positive = target.db.get_prompt_by_hash(
-                generate_prompt_hash("Cat portrait")
-            )
+            positive = api.db.get_prompt_by_hash(generate_prompt_hash("Cat portrait"))
             positive_id = positive["id"]
-            target.db.update_prompt_metadata(
+            api.db.update_prompt_metadata(
                 positive_id, category="Animals", tags=["cats", "studio"]
             )
             variants = []
             for index, negative in enumerate(
                 ["blurry, watermark", "low quality", "low contrast", ""]
             ):
-                pair_id = target.db.save_prompt_pair(
+                pair_id = api.db.save_prompt_pair(
                     "Cat portrait",
                     negative,
                     title=f"Portrait variant {index + 1}",
@@ -345,30 +337,28 @@ def create_app(root: Path) -> web.Application:
                     tags=["studio"] if index == 0 else ["outdoors"],
                 )
                 variants.append(pair_id)
-            with target.db.model.get_connection() as conn:
+            with api.db.model.get_connection() as conn:
                 images = conn.execute(
                     "SELECT id FROM generated_images WHERE prompt_id=? ORDER BY id",
                     (positive_id,),
                 ).fetchall()
             for index, image in enumerate(images[:2]):
-                target.db.link_image_to_pair(
+                api.db.link_image_to_pair(
                     variants[index], image["id"], {"source": "fixture", "anchor": index}
                 )
-            target.db.link_image_to_pair(
+            api.db.link_image_to_pair(
                 variants[0], images[0]["id"], {"source": "fixture second evidence"}
             )
-            target.db.record_prompt_pair_use(variants[0], times=3)
+            api.db.record_prompt_pair_use(variants[0], times=3)
             for number in range(extra_variants):
-                target.db.save_prompt_pair(
-                    "Cat portrait", f"Negative extra {number:04}"
-                )
+                api.db.save_prompt_pair("Cat portrait", f"Negative extra {number:04}")
             legacy = root / "output" / "legacy-unassigned.png"
             shutil.copyfile(root / "output" / "shared" / "same.png", legacy)
-            target.db.link_image_to_prompt(positive_id, str(legacy))
-            target.db.save_prompt("Dog landscape", category="Animals", tags=["dogs"])
-            target.db.save_prompt_pair("Dual role portrait", "Cat portrait")
+            api.db.link_image_to_prompt(positive_id, str(legacy))
+            api.db.save_prompt("Dog landscape", category="Animals", tags=["dogs"])
+            api.db.save_prompt_pair("Dual role portrait", "Cat portrait")
             for number in range(extra_prompts):
-                target.db.save_prompt(
+                api.db.save_prompt(
                     f"Library extra {number:04}", category="Bulk", tags=["bulk"]
                 )
         if prompt_inspector:
@@ -389,8 +379,8 @@ def create_app(root: Path) -> web.Application:
         from py.api.thumbnail_settings import ThumbnailSettings
 
         # 150 px previews in the default format (#317 thumbnail settings).
-        target._generate_gallery_thumbnails_sync(
-            target._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
+        api._generate_gallery_thumbnails_sync(
+            api._get_all_output_dirs(), ThumbnailSettings("webp", 80, 150)
         )
         if include_video:
             # Seed after thumbnail generation to exercise a real unthumbnailed video.
@@ -398,8 +388,8 @@ def create_app(root: Path) -> web.Application:
             shutil.copyfile(
                 Path(__file__).resolve().parents[1] / "fixtures" / "tiny.webm", video
             )
-            target.db.link_image_to_prompt(video_prompt_id, str(video))
-        target.invalidate_gallery_cache()
+            api.db.link_image_to_prompt(video_prompt_id, str(video))
+        api.invalidate_gallery_cache()
 
     history_fixture = None
 
@@ -444,10 +434,7 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if chronological_history:
             manifest = await api._run_in_executor(
-                seeder.run,
-                lambda target: seed_workspace_history(
-                    target, root, history_runs=history_runs
-                ),
+                lambda: seed_workspace_history(api, root, history_runs=history_runs)
             )
             history_fixture = manifest
             return web.json_response({"success": True, **manifest})
@@ -505,31 +492,25 @@ def create_app(root: Path) -> web.Application:
             raise web.HTTPBadRequest()
         if workspace_visual:
             history_fixture = None
-            await api._run_in_executor(
-                seeder.run, lambda target: seed_workspace_visual(target, root)
-            )
+            await api._run_in_executor(seed_workspace_visual, api, root)
             return web.json_response({"success": True})
         if full_inspector:
             history_fixture = None
             await api._run_in_executor(
-                seeder.run,
-                lambda target: seed_workspace_inspector(
-                    target, root, extra_images=extra, include_video=include_video
-                ),
+                lambda: seed_workspace_inspector(
+                    api, root, extra_images=extra, include_video=include_video
+                )
             )
             return web.json_response({"success": True})
         history_fixture = None
         await api._run_in_executor(
-            seeder.run,
-            lambda target: seed(
-                target,
-                extra,
-                include_video,
-                prompt_inspector,
-                native_library,
-                extra_prompts,
-                extra_variants,
-            ),
+            seed,
+            extra,
+            include_video,
+            prompt_inspector,
+            native_library,
+            extra_prompts,
+            extra_variants,
         )
         return web.json_response({"success": True})
 
@@ -569,15 +550,14 @@ def create_app(root: Path) -> web.Application:
         reveal_stub.reset()
         history_fixture = None
         manifest = await api._run_in_executor(
-            seeder.run,
-            lambda target: seed_workspace_concept(
-                target,
+            lambda: seed_workspace_concept(
+                api,
                 root,
                 extra_images=data.get("extra_images", 0),
                 seedless_images=data.get("seedless_images", 0),
                 short_seedless=data.get("short_seedless", False),
                 now=data.get("now"),
-            ),
+            )
         )
         return web.json_response({"success": True, **manifest})
 
@@ -645,12 +625,9 @@ def create_app(root: Path) -> web.Application:
     async def cleanup(application: web.Application) -> None:
         # A scan of an ambiguous fixture file captures it for Needs review.
         api.close_annotation_captures()
-        seeder.close()
         api.db.close_all()
-        # Old media folders still waiting for deletion go before the temp root.
-        close_trash(root)
 
-    seed(api)
+    seed()
     # Open Needs review change streams end first, as on the real host.
     app.on_shutdown.append(api.close_review_events)
     app.on_cleanup.append(cleanup)

tests/browser/workspace_concept_fixture.py

@@ -16,6 +16,7 @@ import json
 import math
 import os
 from pathlib import Path
+import shutil
 import time
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
@@ -33,7 +34,6 @@ from utils.execution_contexts import (
 from utils.execution_history_events import make_finish, make_start
 from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots
 from utils.pair_execution import EXECUTION_KEY
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -300,7 +300,8 @@ def _clear(api: PromptManagerAPI, root: Path) -> None:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
 
 
 def _record_run(

tests/browser/workspace_library_editor_fixture.py

@@ -3,11 +3,11 @@
 from __future__ import annotations
 
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from aiohttp import web
 from PIL import Image
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -51,7 +51,8 @@ def seed_library_editor(api: PromptManagerAPI, root: Path) -> dict:
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     epoch = db.get_execution_epoch()
     receipt = db.create_library_recipe(
         EDITOR_POSITIVE,

tests/browser/workspace_review_fixture.py

@@ -52,7 +52,6 @@ from typing import TYPE_CHECKING
 from aiohttp import web
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -226,7 +225,8 @@ def seed_review(
             conn.execute(f"DELETE FROM {table}")
     api.close_annotation_captures()
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     shutil.rmtree(root / "temp" / "review", ignore_errors=True)
     output = root / "output"
     files = []

tests/browser/workspace_tagging_fixture.py

@@ -19,7 +19,6 @@ from typing import TYPE_CHECKING
 from aiohttp import web
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -99,7 +98,8 @@ def seed_tagging(api: PromptManagerAPI, root: Path, extra_images: int = 0) -> di
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     output = root / "output"
     _png(output / FILES[0][0], (196, 140, 92))
     _png(output / FILES[1][0], (150, 110, 80))

tests/browser/workspace_visual_fixture.py

@@ -5,13 +5,13 @@ from __future__ import annotations
 import json
 import os
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from PIL import Image, ImageDraw
 from PIL.PngImagePlugin import PngInfo
 
 from utils.hashing import generate_prompt_hash
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -129,7 +129,8 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None:
         ):
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     for parent_index, (positive, negatives) in enumerate(zip(_POSITIVES, _NEGATIVES)):
         category = ("Architecture", "Landscape", "Abstract")[parent_index % 3]
         tags = ["geometric", ("warm", "cool")[parent_index % 2]]

tests/browser/workspace_vocabulary_fixture.py

@@ -7,10 +7,10 @@ in concept order, so tag creation order and category first use match the mock.
 from __future__ import annotations
 
 from pathlib import Path
+import shutil
 from typing import TYPE_CHECKING
 
 from aiohttp import web
-from tests.browser.folder_trash import replace_folder
 
 if TYPE_CHECKING:
     from py.api import PromptManagerAPI
@@ -120,7 +120,8 @@ def seed_vocabulary(
         for table in _TABLES:
             conn.execute(f"DELETE FROM {table}")
     for name in ("output", "input"):
-        replace_folder(root / name)
+        shutil.rmtree(root / name)
+        (root / name).mkdir()
     epoch = db.get_execution_epoch()
     recipes = []
     for recipe in CONCEPT_RECIPES:

tests/fixtures/local-inspect/legacy-agreement.json

@@ -0,0 +1,107 @@
+{
+ "generation_params": [
+  "case:basic",
+  "case:shared-prompt-pipeline",
+  "tests/fixtures/inspection_parity/f02.png",
+  "tests/fixtures/inspection_parity/f03.png",
+  "tests/fixtures/inspection_parity/f06c.png",
+  "tests/fixtures/inspection_parity/f06e.png",
+  "tests/fixtures/inspection_parity/f07a.png",
+  "tests/fixtures/inspection_parity/f07b.png",
+  "tests/fixtures/inspection_parity/f08.png",
+  "tests/fixtures/inspection_parity/f09.png",
+  "tests/fixtures/inspection_parity/f10.png",
+  "tests/fixtures/inspection_parity/f11.png",
+  "tests/fixtures/inspection_parity/f15.png",
+  "tests/fixtures/inspection_parity/f16.png",
+  "tests/fixtures/inspection_parity/f23.png"
+ ],
+ "inspection_parity": [
+  "f02",
+  "f03",
+  "f06c",
+  "f06e",
+  "f07a",
+  "f07b",
+  "f08",
+  "f09",
+  "f10",
+  "f11",
+  "f14",
+  "f15",
+  "f16",
+  "f23"
+ ],
+ "local_resolver": [
+  "anchor-first-stage-no-latent",
+  "anchor-same-links",
+  "anchor-same-links-unanchored",
+  "built-chars",
+  "class-type-missing",
+  "class-type-number",
+  "combine-equal",
+  "controlnet-slot1",
+  "custom-cfg",
+  "duplicate-key",
+  "encode",
+  "int-id",
+  "latent-encode",
+  "overwritten-bad-workflow",
+  "pair-execution",
+  "pair-execution-bad-json",
+  "pair-execution-mismatch",
+  "pair-execution-negative-changed",
+  "pair-execution-other-node",
+  "pair-execution-truncated",
+  "pair-execution-two-records",
+  "pair-execution-version-float",
+  "pair-execution-version-true",
+  "pair-graph",
+  "pass-through-unknown-role-inputs",
+  "passthrough",
+  "prompt-and-parameters",
+  "prompt-manager",
+  "select-detailer-chain-same",
+  "select-detailer-no-sampler-upstream",
+  "select-detailer-output-walk-exhausted",
+  "select-detailer-same-prompt",
+  "select-edge-hires-fix-denoised-slot",
+  "select-edge-hires-fix-lineage-cycle",
+  "select-edge-hires-fix-two-refiners",
+  "select-flux-concat",
+  "select-flux-reroute-input",
+  "select-flux-three-stage",
+  "select-hires-fix",
+  "select-hires-fix-both-saved",
+  "select-hires-fix-reroute",
+  "select-hires-fix-unlisted",
+  "select-multistage-latent",
+  "select-preview-dropped",
+  "select-preview-through-bad-node",
+  "select-runtime-flux-other-full-prompt",
+  "select-runtime-flux-trigger",
+  "select-runtime-search-list-append",
+  "select-runtime-through-chained",
+  "select-runtime-through-concat",
+  "select-runtime-through-text-encode",
+  "select-runtime-traced-prepend",
+  "select-runtime-trigger-append-single",
+  "select-runtime-trigger-pair",
+  "select-runtime-trigger-prepend",
+  "select-three-identical-one-advanced",
+  "select-two-identical",
+  "select-unconsumed-stage-kept",
+  "several-samplers",
+  "smz-false",
+  "text-at-limit",
+  "unknown-encoder",
+  "workflow-type-dict",
+  "workflow-type-in-subgraph",
+  "workflow-type-list",
+  "workflow-type-without-id",
+  "zeroed"
+ ],
+ "unknown_nodes": [
+  "unknown-role-inputs"
+ ]
+}

tests/js/support/local-inspect-oracle.mjs

@@ -40,7 +40,7 @@ const semantics = compileSemantics(JSON.parse(fs.readFileSync(path.join(here, '.
 const { extractGenerationFacts } = require(path.join(here, '..', '..', '..', 'web', 'js', 'comfy-metadata.js'));
 
 /** The local evidence for one set of record texts. */
-const resolved = records => ({ evidence: resolveLocalEvidence(records, semantics) });
+const resolved = (records, fileName = null) => ({ evidence: resolveLocalEvidence(records, semantics, fileName) });
 
 /** The local wire's evidence and six facts for one read (no legacy fill, as in the vectors' own harness). */
 function localPayload(read, file, classification) {
@@ -59,14 +59,14 @@ const output = {};
 for (const entry of JSON.parse(Buffer.concat(chunks).toString('utf8'))) {
     const { name } = entry;
     if (entry.mode === 'resolve') {
-        if (entry.records !== undefined) { output[name] = resolved(entry.records); continue; }
+        if (entry.records !== undefined) { output[name] = resolved(entry.records, entry.fileName ?? null); continue; }
         const blob = await fs.openAsBlob(entry.path);
         const { classification, result } = await readLocalFile(blob, { pngMetadata });
         const read = classification === null ? null : toInspectionRead(classification, result).read;
         if (classification === null) { output[name] = { unreadable: true }; continue; }
         // The evidence is the local wire's own (buildLocalPayload), and it equals resolveLocalEvidence of the records.
         const local = localPayload(result, { name: path.basename(entry.path), size: blob.size }, classification);
-        const direct = read === null ? unreadableEvidence() : resolved(read.records).evidence;
+        const direct = read === null ? unreadableEvidence() : resolved(read.records, path.basename(entry.path)).evidence;
         if (JSON.stringify(direct) !== JSON.stringify(local.evidence)) throw new Error(`${name}: the wire's roles are not the resolver's`);
         output[name] = read === null ? { unreadable: true, ...local } : local;
         continue;

tests/js/workspace-local-resolver.test.js

@@ -155,7 +155,9 @@ test('committed server answers equal the local evidence', () => {
         value !== null && typeof value === 'object' ? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])])) : value;
     let compared = 0;
     for (const {name, records, evidence} of vectors) {
-        assert.deepEqual(canonical(resolver.resolveLocalEvidence(records, semantics)), canonical(evidence), name);
+        // A case may name the inspected file (`__file_name`, #381 saver anchoring); it is no record.
+        const {__file_name: fileName = null, ...roles} = records;
+        assert.deepEqual(canonical(resolver.resolveLocalEvidence(roles, semantics, fileName)), canonical(evidence), name);
         compared++;
     }
     assert.ok(compared > 280, `compared ${compared}`);

tests/local_resolver_cases.py

@@ -26,6 +26,8 @@ VECTORS_PATH = (
     / "local-inspect"
     / "resolver-vectors.json"
 )
+# A case may name the inspected file (#381 saver anchoring); it is no record.
+FILE_NAME = "__file_name"
 # Committed record sets stay small; the large bound cases are built at test time.
 MAX_VECTOR_CHARS = 20_000
 COMMITTED_FUZZ = (0, 60)
@@ -906,6 +908,60 @@ def golden_cases():
             },
         ),
     )
+    # #381: unknown nodes between the prompt and the sampler pass through.
+    try:
+        from tests import unknown_node_graphs
+    except ImportError:  # pragma: no cover - discover -s tests puts tests/ first
+        import unknown_node_graphs
+    for name, graph in unknown_node_graphs.shapes().items():
+        add(f"pass-through-{name}", graph)
+    for name, records in {
+        **unknown_node_graphs.recorded_shapes(),
+        **unknown_node_graphs.review_shapes(),
+    }.items():
+        cases[f"pass-through-{name}"] = records
+    try:
+        from tests import saver_anchor_graphs as anchor
+    except ImportError:  # pragma: no cover - discover -s tests puts tests/ first
+        import saver_anchor_graphs as anchor
+    for name, graph, file_name in [
+        ("branch-3", anchor.branch_bench(), "s45_branch_3_00002_.png"),
+        ("branch-unanchored", anchor.branch_bench(), None),
+        ("branch-no-match", anchor.branch_bench(), "photo_00001_.png"),
+        ("branch-linked", anchor.linked_prefix_bench(), "s45_branch_2_00007_.png"),
+        (
+            "branch-unknown-saver",
+            anchor.branch_bench(saver="VendorSaveImage"),
+            "s45_branch_4_00001.webp",
+        ),
+        ("shared-prefix", anchor.shared_prefix(), "IMG_GEN_00371_.png"),
+        ("same-links", anchor.two_stage_same_conditioning(), "final_00003_.png"),
+        ("same-links-unanchored", anchor.two_stage_same_conditioning(), None),
+        (
+            "other-links",
+            anchor.two_stage_same_conditioning(["91", 0]),
+            "final_00003_.png",
+        ),
+        ("hidden-earlier", anchor.hidden_earlier_prompt(), "out_00001_.png"),
+        (
+            "hidden-override",
+            anchor.hidden_earlier_prompt(override=True),
+            "out_00001_.png",
+        ),
+    ] + [
+        (f"first-stage-{kind}", anchor.first_stage_variant(kind), "final_00003_.png")
+        for kind in (
+            "no-latent",
+            "literal-latent",
+            "missing-latent",
+            "cyclic-latent",
+            "stage-shaped",
+            "detailer",
+        )
+    ]:
+        cases[f"anchor-{name}"] = {"prompt": _dump(graph)}
+        if file_name is not None:
+            cases[f"anchor-{name}"][FILE_NAME] = file_name
     cases["parameters-only"] = {
         "parameters": "a fox\nNegative prompt: blur\nSteps: 20, Sampler: euler"
     }
@@ -1059,7 +1115,8 @@ def boundary_cases():
                 ["7", 0],
                 {
                     "\U0001f600"
-                    * 70: {"class_type": long_name, "inputs": {"text": "x"}},
+                    # No text and no links: still named, not passed through (#381).
+                    * 70: {"class_type": long_name, "inputs": {"seed": 1}},
                     "7": _encode("b"),
                 },
             )
@@ -1387,7 +1444,10 @@ def committed_vectors():
             "name": name,
             "records": records,
             "evidence": inspect_prompt_evidence(
-                records, None, profile=INSPECTION_PROFILE
+                {key: value for key, value in records.items() if key != FILE_NAME},
+                None,
+                profile=INSPECTION_PROFILE,
+                file_name=records.get(FILE_NAME),
             ),
         }
         for name, records in cases.items()

tests/saver_anchor_graphs.py

@@ -0,0 +1,216 @@
+"""Synthetic graphs for saver anchoring and shared-conditioning stages (#381).
+
+``branch_bench`` reproduces the maintainer's ``gpu_unlock_bench`` run: four
+KSampler branches with different prompts, each saved by its own SaveImage whose
+literal prefix names the branch (``bench/run/s45_branch_N``). ``shared_prefix``
+is the dataset run's shape: every branch's saver uses one prefix, so the file
+name cannot name the branch. ``two_stage_same_conditioning`` reproduces the
+ANIMA run: a hires stage whose latent passes an unlisted node and reads the
+same positive and negative links as the first stage. Texts are neutral.
+"""
+
+BRANCH_TEXTS = (
+    "a knit beanie on a wooden table",
+    "an old fishing boat at dawn",
+    "a red vintage car in the rain",
+    "a cozy library with tall shelves",
+)
+NEGATIVE = "blurry, lowres"
+
+
+def _encode(text):
+    return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["1", 1]}}
+
+
+def _ksampler(positive, negative, latent, denoise=1.0):
+    return {
+        "class_type": "KSampler",
+        "inputs": {
+            "seed": 1,
+            "steps": 20,
+            "cfg": 7.0,
+            "sampler_name": "euler",
+            "scheduler": "normal",
+            "denoise": denoise,
+            "model": ["1", 0],
+            "positive": positive,
+            "negative": negative,
+            "latent_image": latent,
+        },
+    }
+
+
+def _base():
+    return {
+        "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
+        "2": {"class_type": "EmptyLatentImage", "inputs": {"width": 64}},
+        "3": _encode(NEGATIVE),
+    }
+
+
+def branch_bench(prefixes=None, saver="SaveImage"):
+    """Four branches; ``prefixes`` replaces each saver's ``filename_prefix``."""
+    prefixes = prefixes or [
+        f"bench/run/s45_branch_{index}" for index in range(1, len(BRANCH_TEXTS) + 1)
+    ]
+    graph = _base()
+    for index, text in enumerate(BRANCH_TEXTS, start=1):
+        base = index * 10
+        graph[str(base)] = _encode(text)
+        graph[str(base + 1)] = _ksampler([str(base), 0], ["3", 0], ["2", 0])
+        graph[str(base + 2)] = {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": [str(base + 1), 0], "vae": ["1", 2]},
+        }
+        graph[str(base + 3)] = {
+            "class_type": saver,
+            "inputs": {
+                "filename_prefix": prefixes[index - 1],
+                "images": [str(base + 2), 0],
+            },
+        }
+    return graph
+
+
+def linked_prefix_bench():
+    """The branch run with each prefix coming from a PrimitiveString."""
+    graph = branch_bench(prefixes=[["90", 0], ["91", 0], ["92", 0], ["93", 0]])
+    for index in range(4):
+        graph[str(90 + index)] = {
+            "class_type": "PrimitiveString",
+            "inputs": {"value": f"bench/run/s45_branch_{index + 1}"},
+        }
+    return graph
+
+
+def shared_prefix():
+    """Every branch saved with one prefix: the file name names no branch."""
+    return branch_bench(prefixes=["dataset/IMG_GEN"] * len(BRANCH_TEXTS))
+
+
+def two_stage_same_conditioning(second_positive=None):
+    """Base and hires stage on the same links; the hires latent passes an
+    unlisted node (``VendorGpuCleaner``), so the stages' lineage is unknown."""
+    graph = _base()
+    graph["90"] = _encode(BRANCH_TEXTS[0])
+    graph["63"] = _ksampler(["90", 0], ["3", 0], ["2", 0])
+    graph["30"] = {
+        "class_type": "VAEDecode",
+        "inputs": {"samples": ["63", 0], "vae": ["1", 2]},
+    }
+    graph["98"] = {
+        "class_type": "SaveImage",
+        "inputs": {"filename_prefix": "draft", "images": ["30", 0]},
+    }
+    graph["110"] = {"class_type": "VendorGpuCleaner", "inputs": {"anything": ["63", 0]}}
+    graph["108"] = {
+        "class_type": "LatentUpscaleBy",
+        "inputs": {
+            "upscale_method": "nearest-exact",
+            "scale_by": 1.2,
+            "samples": ["110", 0],
+        },
+    }
+    graph["114"] = _ksampler(second_positive or ["90", 0], ["3", 0], ["108", 0], 0.5)
+    graph["113"] = {
+        "class_type": "VAEDecode",
+        "inputs": {"samples": ["114", 0], "vae": ["1", 2]},
+    }
+    graph["115"] = {
+        "class_type": "VendorGpuCleaner",
+        "inputs": {"anything": ["113", 0]},
+    }
+    graph["107"] = {
+        "class_type": "SaveImage",
+        "inputs": {"filename_prefix": "final", "images": ["115", 0]},
+    }
+    if second_positive is not None:
+        graph["91"] = _encode(BRANCH_TEXTS[1])
+    return graph
+
+
+def hidden_earlier_prompt(override=False):
+    """Review shape: the skipped same-links stage continues a different prompt.
+
+    10 (cat) -> 11 (dog) -> unlisted LatentBlend -> 13 (dog, same links as 11)
+    -> VAEDecode -> SaveImage ``out``. With ``override``, 11 is a
+    ``KSampler (Efficient)`` with a linked ``script``.
+    """
+    graph = _base()
+    graph["5"] = _encode("cat X")
+    graph["6"] = _encode("dog Y")
+    graph["10"] = _ksampler(["5", 0], ["3", 0], ["2", 0])
+    graph["11"] = _ksampler(["6", 0], ["3", 0], ["10", 0])
+    if override:
+        graph["11"] = _ksampler(["6", 0], ["3", 0], ["2", 0])
+        graph["11"]["class_type"] = "KSampler (Efficient)"
+        graph["11"]["inputs"]["script"] = ["40", 0]
+        graph["40"] = {"class_type": "XY Plot", "inputs": {"grid_spacing": 0}}
+    graph["12"] = {
+        "class_type": "LatentBlend",
+        "inputs": {"samples1": ["11", 0], "samples2": ["2", 0], "blend_factor": 0.5},
+    }
+    graph["13"] = _ksampler(["6", 0], ["3", 0], ["12", 0], 0.5)
+    graph["14"] = {
+        "class_type": "VAEDecode",
+        "inputs": {"samples": ["13", 0], "vae": ["1", 2]},
+    }
+    graph["15"] = {
+        "class_type": "SaveImage",
+        "inputs": {"filename_prefix": "out", "images": ["14", 0]},
+    }
+    return graph
+
+
+def deep_prefixes(chain):
+    """Review shape: two branches whose savers both trace the prefix ``dup``.
+
+    Saver A's prefix runs through ``chain`` DisplayText nodes; saver B's
+    through one more, so a short visit allowance can finish only one trace.
+    """
+    graph = branch_bench(prefixes=[["p0", 0], ["q", 0], "x_1", "x_2"])
+    graph["pz"] = {"class_type": "StringConstant", "inputs": {"string": "dup"}}
+    previous = "pz"
+    for index in range(chain - 1, -1, -1):
+        graph[f"p{index}"] = {
+            "class_type": "DisplayText",
+            "inputs": {"text": [previous, 0]},
+        }
+        previous = f"p{index}"
+    graph["q"] = {"class_type": "DisplayText", "inputs": {"text": ["p0", 0]}}
+    return graph
+
+
+def first_stage_variant(kind):
+    """``two_stage_same_conditioning`` whose first stage (63) is not plain.
+
+    ``no-latent`` drops its latent input (still plain); ``literal-latent``,
+    ``missing-latent``, ``cyclic-latent``, ``stage-shaped`` and ``detailer``
+    each make it something other than a plain first stage, so the later
+    stage stays unattributed.
+    """
+    graph = two_stage_same_conditioning()
+    inputs = graph["63"]["inputs"]
+    if kind == "no-latent":
+        del inputs["latent_image"]
+    elif kind == "literal-latent":
+        inputs["latent_image"] = "latent"
+    elif kind == "missing-latent":
+        inputs["latent_image"] = ["999", 0]
+    elif kind == "cyclic-latent":
+        graph["70"] = {"class_type": "VendorLatentOp", "inputs": {"samples": ["63", 0]}}
+        inputs["latent_image"] = ["70", 0]
+    elif kind == "stage-shaped":
+        graph["70"] = {
+            "class_type": "VendorSampler",
+            "inputs": {
+                "model": ["1", 0],
+                "positive": ["90", 0],
+                "latent_image": ["2", 0],
+            },
+        }
+        inputs["latent_image"] = ["70", 0]
+    elif kind == "detailer":
+        graph["63"]["class_type"] = "FaceDetailer"
+        inputs["image"] = ["30", 0]
+    return graph

tests/test_api_prompt_inspector.py

@@ -190,7 +190,8 @@ class TestPromptInspector(ImageAPITestCase):
                     "inputs": {"positive": ["5", 0], "negative": ["2", 0]},
                 }
             else:
-                source["1"]["class_type"] = "UnknownEncoder"
+                # No text and no links: nothing passes through it (#381).
+                source["1"] = {"class_type": "UnknownEncoder", "inputs": {"seed": 1}}
             png(
                 path,
                 prompt=source,
@@ -215,7 +216,7 @@ class TestPromptInspector(ImageAPITestCase):
             result["prompts"]["negative"], self.side("", "embedded-parameters")
         )
         source = graph()
-        source["2"]["class_type"] = "UnknownEncoder"
+        source["2"] = {"class_type": "UnknownEncoder", "inputs": {"seed": 1}}
         png(path, prompt=source)
         _, result = await self.inspect()
         self.assertEqual(result["prompts"]["positive"], self.side("portrait"))

tests/test_browser_fast_reset.py

@@ -1,335 +0,0 @@
-"""Browser-test resets seed a scratch copy and publish it in one commit (#384)."""
-
-import hashlib
-import os
-import sqlite3
-import subprocess
-import sys
-import tempfile
-import threading
-import time
-import types
-import unittest
-from pathlib import Path
-from unittest import mock
-
-try:
-    from tests.open_handles import assert_closed
-except ImportError:  # discovered with tests/ as the top-level directory
-    from open_handles import assert_closed
-
-from database.operations import PromptDatabase
-from tests.browser import fast_reset
-from tests.browser.fast_reset import ScratchSeeder, table_digests
-
-
-def digests(db: PromptDatabase) -> dict:
-    return table_digests(db.model.get_connection())
-
-
-def add_prompt(db: PromptDatabase, text: str) -> int:
-    with db.model.get_connection() as conn:
-        return conn.execute(
-            "INSERT INTO prompts (text, hash, created_at, updated_at) "
-            "VALUES (?, ?, '2026-01-01 00:00:00', '2026-01-01 00:00:00')",
-            (text, hashlib.sha256(text.encode()).hexdigest()),
-        ).lastrowid
-
-
-def add_image(db: PromptDatabase, prompt_id: int, name: str) -> int:
-    with db.model.get_connection() as conn:
-        return conn.execute(
-            "INSERT INTO generated_images (prompt_id, image_path, filename, "
-            "generation_time) VALUES (?, ?, ?, '2026-01-01 00:00:00')",
-            (prompt_id, f"/out/{name}", name),
-        ).lastrowid
-
-
-def reseed(api) -> str:
-    """A seeder in the old style: clears its tables, then many small commits."""
-    db = api.db
-    with db.model.get_connection() as conn:
-        for table in ("generated_images", "prompt_tags", "tags", "prompts"):
-            conn.execute(f"DELETE FROM {table}")
-    for index in range(30):
-        prompt_id = add_prompt(db, f"seeded {index}")
-        add_image(db, prompt_id, f"seeded-{index}.png")
-        with db.model.get_connection() as conn:
-            tag_id = conn.execute(
-                "INSERT INTO tags (name) VALUES (?)", (f"tag{index}",)
-            ).lastrowid
-            conn.execute(
-                "INSERT INTO prompt_tags (prompt_id, tag_id) VALUES (?, ?)",
-                (prompt_id, tag_id),
-            )
-    return "manifest"
-
-
-class TestScratchSeeder(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.addCleanup(assert_closed, self, self.tmp.name)
-        self.live = self.database("live.db")
-        self.api = types.SimpleNamespace(db=self.live, marker="live api")
-        self.seeder = ScratchSeeder(self.api, self.root / "scratch", memory_root=None)
-        self.addCleanup(self.seeder.close)
-        # Rows from "earlier tests": AUTOINCREMENT positions have moved on.
-        for index in range(5):
-            add_image(self.live, add_prompt(self.live, f"old {index}"), f"o{index}")
-        with self.live.model.get_connection() as conn:
-            conn.execute("DELETE FROM prompts WHERE text = 'old 4'")
-
-    def database(self, name: str) -> PromptDatabase:
-        db = PromptDatabase(str(self.root / name))
-        self.addCleanup(db.close_all)
-        return db
-
-    def oracle(self, seed):
-        """Run ``seed`` the old way, on a copy of the live database."""
-        path = self.root / "oracle.db"
-        target = sqlite3.connect(path)
-        try:
-            self.live.model.get_connection().backup(target)
-        finally:
-            target.close()
-        db = self.database("oracle.db")
-        seed(types.SimpleNamespace(db=db))
-        return digests(db)
-
-    def test_publishes_exactly_what_the_seed_writes_on_live(self):
-        expected = self.oracle(reseed)
-        self.assertEqual(self.seeder.run(reseed), "manifest")
-        self.assertEqual(digests(self.live), expected)
-        with self.live.model.get_connection() as conn:
-            first = conn.execute("SELECT MIN(id) FROM prompts").fetchone()[0]
-        self.assertEqual(first, 6, "AUTOINCREMENT continues as on live")
-
-    def test_runs_twice_from_the_state_the_first_run_left(self):
-        self.seeder.run(reseed)
-        expected = self.oracle(reseed)
-        self.seeder.run(reseed)
-        self.assertEqual(digests(self.live), expected)
-
-    def test_tables_the_seed_did_not_change_keep_their_rows(self):
-        before = digests(self.live)
-
-        def rename(api):
-            with api.db.model.get_connection() as conn:
-                conn.execute("UPDATE prompts SET text = 'renamed' WHERE id = 1")
-
-        self.seeder.run(rename)
-        after = digests(self.live)
-        changed = {table for table in after if after[table] != before[table]}
-        self.assertEqual(changed, {"prompts"})
-        with self.live.model.get_connection() as conn:
-            images = conn.execute("SELECT COUNT(*) FROM generated_images").fetchone()
-        # Replacing prompts' rows never cascades into generated_images.
-        self.assertEqual(images[0], 4)
-
-    def test_the_seed_sees_the_scratch_database_and_the_live_api(self):
-        seen = {}
-
-        def seed(api):
-            seen["db"] = api.db
-            seen["marker"] = api.marker
-            api.note = "set by the seed"
-
-        self.seeder.run(seed)
-        self.assertIsNot(seen["db"], self.live)
-        self.assertNotEqual(seen["db"].model.db_path, self.live.model.db_path)
-        self.assertEqual(seen["marker"], "live api")
-        self.assertEqual(self.api.note, "set by the seed")
-        self.assertIs(self.api.db, self.live)
-
-    def test_cache_drops_wait_until_the_rows_are_published(self):
-        seen = []
-
-        def invalidate_gallery_cache(retire=False):
-            with self.live.model.get_connection() as conn:
-                count = conn.execute("SELECT COUNT(*) FROM tags").fetchone()[0]
-            seen.append((count, retire))
-
-        self.api.invalidate_gallery_cache = invalidate_gallery_cache
-
-        def seed(api):
-            reseed(api)
-            api.invalidate_gallery_cache(retire=True)
-            self.assertEqual(seen, [], "not dropped while live has the old rows")
-
-        self.seeder.run(seed)
-        self.assertEqual(seen, [(30, True)])
-
-    def test_a_failing_seed_drops_no_cache(self):
-        seen = []
-        self.api.invalidate_gallery_cache = lambda: seen.append(True)
-
-        def broken(api):
-            api.invalidate_gallery_cache()
-            raise ValueError("seed failed")
-
-        with self.assertRaises(ValueError):
-            self.seeder.run(broken)
-        self.assertEqual(seen, [])
-
-    def test_a_failing_seed_leaves_live_untouched(self):
-        before = digests(self.live)
-
-        def broken(api):
-            reseed(api)
-            raise ValueError("seed failed")
-
-        with self.assertRaisesRegex(ValueError, "seed failed"):
-            self.seeder.run(broken)
-        self.assertEqual(digests(self.live), before)
-        self.assertIs(self.api.db, self.live)
-
-    def test_a_concurrent_live_commit_fails_the_reset(self):
-        other = self.database("live.db")
-
-        def racing(api):
-            reseed(api)
-            add_prompt(other, "written to live meanwhile")
-
-        with self.assertRaisesRegex(RuntimeError, "changed while"):
-            self.seeder.run(racing)
-        with self.live.model.get_connection() as conn:
-            texts = [row[0] for row in conn.execute("SELECT text FROM prompts")]
-        self.assertIn("written to live meanwhile", texts)
-        self.assertNotIn("seeded 0", texts)
-
-    def test_a_write_open_during_the_seed_and_committed_later_fails_it(self):
-        other = self.database("live.db")
-        began, finished = threading.Event(), []
-
-        def write():
-            with other.model.get_connection() as conn:
-                conn.execute("BEGIN IMMEDIATE")
-                conn.execute(
-                    "INSERT INTO prompts (text, hash) VALUES ('concurrent', 'c')"
-                )
-                began.set()
-                time.sleep(0.5)  # the publish waits on this lock meanwhile
-            other.model.close()
-            finished.append(True)
-
-        writer = threading.Thread(target=write)
-
-        def racing(api):
-            reseed(api)
-            writer.start()
-            self.assertTrue(began.wait(10))
-
-        with self.assertRaisesRegex(RuntimeError, "changed while"):
-            self.seeder.run(racing)
-        writer.join()
-        self.assertEqual(finished, [True])
-        with self.live.model.get_connection() as conn:
-            texts = [row[0] for row in conn.execute("SELECT text FROM prompts")]
-        self.assertIn("concurrent", texts)
-        self.assertNotIn("seeded 0", texts)
-        self.assert_live_connection_restored()
-
-    def test_a_failed_publish_leaves_live_and_its_connection_as_they_were(self):
-        before = digests(self.live)
-        columns = fast_reset._columns
-
-        def broken_columns(conn, table, *schema):
-            # The publish asks for the live table's columns without a schema.
-            return columns(conn, table, *schema) if schema else ["no_such_column"]
-
-        with mock.patch.object(fast_reset, "_columns", broken_columns):
-            with self.assertRaises(sqlite3.OperationalError):
-                self.seeder.run(reseed)
-        self.assertEqual(digests(self.live), before)
-        self.assert_live_connection_restored()
-
-    def assert_live_connection_restored(self):
-        conn = self.live.model.get_connection()
-        self.assertFalse(conn.in_transaction)
-        self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 1)
-        names = [row[1] for row in conn.execute("PRAGMA database_list")]
-        self.assertEqual(names, ["main"])
-
-    def test_the_live_connection_is_left_as_it_was(self):
-        self.seeder.run(reseed)
-        conn = self.live.model.get_connection()
-        self.assertFalse(conn.in_transaction)
-        self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone()[0], 1)
-        names = [row[1] for row in conn.execute("PRAGMA database_list")]
-        self.assertEqual(names, ["main"])
-        self.assertEqual(conn.execute("PRAGMA foreign_key_check").fetchall(), [])
-
-    def test_on_live_mode_runs_the_seed_on_the_live_api(self):
-        seeder = ScratchSeeder(
-            self.api, self.root / "unused", on_live=True, memory_root=self.root
-        )
-        self.addCleanup(seeder.close)
-        expected = self.oracle(reseed)
-        seen = []
-        self.assertEqual(
-            seeder.run(lambda api: seen.append(api) or reseed(api)), "manifest"
-        )
-        self.assertIs(seen[0], self.api)
-        self.assertEqual(digests(self.live), expected)
-        self.assertFalse((self.root / "unused").exists())
-
-    def test_the_scratch_lives_under_the_memory_root_when_there_is_one(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        expected = self.oracle(reseed)
-        seeder.run(reseed)
-        self.assertEqual(digests(self.live), expected)
-        self.assertEqual(len(list(memory.glob("pm-reset-*/seed.db"))), 1)
-        self.assertFalse((self.root / "fallback").exists())
-        seeder.close()
-        self.assertEqual(list(memory.iterdir()), [])
-
-    def test_a_full_memory_root_puts_the_scratch_on_disk(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        full = mock.patch.object(
-            fast_reset.shutil,
-            "disk_usage",
-            return_value=types.SimpleNamespace(free=fast_reset.MEMORY_FREE_BYTES - 1),
-        )
-        with full:
-            seeder.run(reseed)
-        self.assertEqual(list(memory.iterdir()), [])
-        self.assertTrue((self.root / "fallback" / "seed.db").exists())
-
-    @unittest.skipUnless(os.name == "posix", "process probe is POSIX only")
-    def test_scratch_folders_of_dead_servers_are_removed(self):
-        memory = self.root / "memory"
-        memory.mkdir()
-        dead = subprocess.Popen([sys.executable, "-c", "pass"])
-        dead.wait()
-        orphan = memory / f"pm-reset-{dead.pid}-abc"
-        alive = memory / f"pm-reset-{os.getpid()}-def"
-        other = memory / "pm-reset-notapid"
-        for folder in (orphan, alive, other):
-            folder.mkdir()
-            (folder / "seed.db").write_bytes(b"x")
-        seeder = ScratchSeeder(self.api, self.root / "fallback", memory_root=memory)
-        self.addCleanup(seeder.close)
-        seeder.run(reseed)
-        self.assertFalse(orphan.exists())
-        self.assertTrue(alive.exists())
-        self.assertTrue(other.exists())
-        own = [p for p in memory.glob(f"pm-reset-{os.getpid()}-*") if p != alive]
-        self.assertEqual(len(own), 1)
-
-    def test_close_releases_the_scratch_files(self):
-        self.seeder.run(reseed)
-        self.seeder.close()
-        assert_closed(self, self.root / "scratch")
-        self.seeder.close()  # idempotent
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_browser_folder_trash.py

@@ -1,67 +0,0 @@
-"""Browser-test media folders are emptied by a rename, deleted later (#384)."""
-
-import tempfile
-import threading
-import unittest
-from pathlib import Path
-
-from tests.browser.folder_trash import TRASH_NAME, close_trash, replace_folder
-
-
-class TestFolderTrash(unittest.TestCase):
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmp.cleanup)
-        self.root = Path(self.tmp.name)
-        self.addCleanup(close_trash, self.root)
-
-    def trash_threads(self):
-        return [t for t in threading.enumerate() if t.name == "pm-reset-trash"]
-
-    def test_replaces_the_folder_with_an_empty_one(self):
-        folder = self.root / "output"
-        (folder / "nested").mkdir(parents=True)
-        (folder / "nested" / "image.png").write_bytes(b"png")
-        replace_folder(folder)
-        self.assertTrue(folder.is_dir())
-        self.assertEqual(list(folder.iterdir()), [])
-        close_trash(self.root)
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        self.assertEqual(self.trash_threads(), [])
-
-    def test_a_missing_folder_is_created_without_a_trash(self):
-        replace_folder(self.root / "input")
-        self.assertTrue((self.root / "input").is_dir())
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        self.assertEqual(self.trash_threads(), [])
-
-    def test_repeated_replacements_share_one_thread_and_all_are_removed(self):
-        folder = self.root / "output"
-        for number in range(5):
-            folder.mkdir(exist_ok=True)
-            (folder / f"{number}.png").write_bytes(b"png")
-            replace_folder(folder)
-        self.assertEqual(len(self.trash_threads()), 1)
-        close_trash(self.root)
-        self.assertEqual(list(folder.iterdir()), [])
-        self.assertFalse((self.root / TRASH_NAME).exists())
-        close_trash(self.root)  # idempotent
-
-
-class TestFolderTrashAfterClose(unittest.TestCase):
-    def test_a_replacement_after_close_deletes_in_place_without_a_thread(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            close_trash(root)
-            folder = root / "output"
-            folder.mkdir()
-            (folder / "late.png").write_bytes(b"png")
-            replace_folder(folder)
-            self.assertEqual(list(folder.iterdir()), [])
-            self.assertFalse((root / TRASH_NAME).exists())
-            names = [t.name for t in threading.enumerate()]
-            self.assertNotIn("pm-reset-trash", names)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_browser_slow_reset_watch.py

@@ -221,12 +221,12 @@ class TestSlowResetOnTheServer(unittest.TestCase):
             self.assertRegex(
                 text, r"kernel: asyncio_\d+ \(0x[0-9a-f]{16}\) state \w, waiting in "
             )
-        # The plain reset seeds a scratch copy (#384), then waits to publish it.
-        self.assertRegex(text, r"line \d+ in _publish\n")
+        # The plain reset's seed() waits on its first DELETE.
+        self.assertRegex(text, r"line \d+ in seed\n")
         self.assertIn("POST /__test__/reset answered 200 after", text)
         self.assertRegex(
             text,
-            r"executor call ScratchSeeder.run: queued \d+\.\d+s, ran \d+\.\d+s",
+            r"executor call create_app.<locals>.seed: queued \d+\.\d+s, ran \d+\.\d+s",
         )
 
 

tests/test_inspection_semantics.py

@@ -59,6 +59,7 @@ def _minimal():
             {"class": "VAEEncode", "slot": 0, "strategy": "continue", "input": "pixels"}
         ],
         "stage_inputs": {"shaped": ["positive"], "with_model": ["samples"]},
+        "pass_through": {"skip_inputs": ["clip"]},
         "facts": {
             "configuration_outputs": {},
             "model_outputs": {},
@@ -690,6 +691,18 @@ class ValidationTest(_TableCase):
             (lambda d: d["stage_inputs"].update(shaped="pipe"), "input name"),
             (lambda d: d["stage_inputs"].update(with_model=["positive"]), "both"),
             (lambda d: d["stage_inputs"].update(shaped=["model"]), "model"),
+            (lambda d: d.pop("pass_through"), "category"),
+            (lambda d: d["pass_through"].update(extra=["x"]), "pass_through"),
+            (lambda d: d["pass_through"].update(skip_inputs="clip"), "input name"),
+            (lambda d: d["pass_through"].update(skip_inputs=[]), "input name"),
+            (
+                lambda d: d["pass_through"].update(skip_inputs=["clip", "clip"]),
+                "repeated",
+            ),
+            (
+                lambda d: d["pass_through"].update(skip_inputs=["conditioning"]),
+                "carries text",
+            ),
             (
                 lambda d: d["facts"]["configuration_outputs"].update(
                     X={"kind": "sigmas", "fields": ["steps"], "type_fields": []}

tests/test_legacy_parity_gate.py

@@ -0,0 +1,187 @@
+"""Legacy parity gate for the inspection resolver (#381, the PR #170 floor).
+
+The legacy parser (``utils.generation_params``, the port of the browser's
+``extractGenerationParams``) is the floor: a graph whose positive prompt it
+found, and that the inspection resolver resolved to the same text, must keep
+resolving to that text. ``legacy-agreement.json`` records those graphs per
+corpus: the generation-params vectors, the inspection parity fixtures, the
+local resolver golden and selection cases, and the unknown-node shapes of
+#381. A graph may join the list, never leave it; regenerate with::
+
+    python3 -m tests.test_legacy_parity_gate --write
+"""
+
+import argparse
+import json
+import unittest
+from pathlib import Path
+
+from utils import prompt_pair_resolver as resolver
+from utils.generation_params import (
+    GenerationParamsError,
+    extract_generation_params,
+    parse_js_json,
+)
+
+try:
+    from tests import inspection_parity as parity
+    from tests import local_resolver_cases as cases
+    from tests import local_resolver_selection_cases as selection
+    from tests import unknown_node_graphs as unknown
+except ImportError:  # discovered with ``-s tests``
+    import inspection_parity as parity
+    import local_resolver_cases as cases
+    import local_resolver_selection_cases as selection
+    import unknown_node_graphs as unknown
+
+HERE = Path(__file__).resolve().parent
+AGREEMENT_PATH = HERE / "fixtures" / "local-inspect" / "legacy-agreement.json"
+VECTORS_PATH = HERE / "fixtures" / "generation_params" / "vectors.json"
+NOT_FOUND = "No prompt found"
+
+
+def _parsed(value):
+    return parse_js_json(value) if isinstance(value, str) else value
+
+
+def corpora() -> dict:
+    """Corpus name to ``{case name: (prompt graph, workflow or None[, file])}``."""
+    vectors = json.loads(VECTORS_PATH.read_text(encoding="utf-8"))
+    named = {**cases.golden_cases(), **selection.selection_cases()}
+    return {
+        "generation_params": {
+            item["name"]: (_parsed(item["prompt"]), _parsed(item.get("workflow")))
+            for item in vectors
+        },
+        "inspection_parity": {
+            name: (parity.build(name)["prompt"], parity.build(name)["workflow"])
+            for name in parity.COMMITTED
+        },
+        "local_resolver": {
+            name: (
+                _parsed(records["prompt"]),
+                _parsed(records.get("workflow")),
+                records.get(cases.FILE_NAME),
+            )
+            for name, records in named.items()
+            if "prompt" in records
+        },
+        "unknown_nodes": {
+            name: (graph, None) for name, graph in unknown.shapes().items()
+        },
+    }
+
+
+def legacy_positive(graph, workflow):
+    """The positive prompt the legacy parser found, or ``None``."""
+    try:
+        found = extract_generation_params({"prompt": graph, "workflow": workflow})
+    except GenerationParamsError:
+        return None
+    text = found["positivePrompt"]
+    return None if text == NOT_FOUND else text
+
+
+def inspected(graph, workflow=None, file_name=None) -> dict:
+    """The inspection resolver's answer, anchored by ``file_name`` when given."""
+    metadata = {"prompt": graph}
+    if workflow is not None:
+        metadata["workflow"] = workflow
+    return resolver.resolve_prompt_pair(
+        metadata, profile=resolver.INSPECTION_PROFILE, file_name=file_name
+    )
+
+
+# Graphs where the legacy parser's positive is wrong and the inspector's
+# differs on purpose: (corpus, case) -> (the inspector's positive, why).
+LEGACY_WRONG = {
+    ("local_resolver", "anchor-branch-3"): (
+        "a red vintage car in the rain",
+        "legacy shows the first sampler's prompt for every branch; the file "
+        "name s45_branch_3_00002_ names branch 3's saver (gpu_unlock_bench)",
+    ),
+    ("local_resolver", "anchor-branch-linked"): (
+        "an old fishing boat at dawn",
+        "legacy shows branch 1; the traced prefix names branch 2's saver",
+    ),
+    ("local_resolver", "anchor-branch-unknown-saver"): (
+        "a cozy library with tall shelves",
+        "legacy shows branch 1; the file name names branch 4's saver",
+    ),
+}
+
+
+def agreeing(graphs: dict) -> list:
+    """Names whose inspection positive equals the legacy parser's, sorted."""
+    names = []
+    for name, case in graphs.items():
+        graph, workflow = case[0], case[1]
+        if not isinstance(graph, dict):
+            continue
+        expected = legacy_positive(graph, workflow)
+        if expected is None:
+            continue
+        result = inspected(*case)
+        positive = result["positive"]
+        if (
+            result["state"] == "resolved"
+            and isinstance(positive, dict)
+            and positive["text"] == expected
+        ):
+            names.append(name)
+    return sorted(names)
+
+
+class LegacyParityGateTest(unittest.TestCase):
+    def test_every_legacy_difference_is_stated(self):
+        """A graph may read differently from legacy only with a stated reason."""
+        all_graphs = corpora()
+        for (corpus, name), (expected, why) in LEGACY_WRONG.items():
+            with self.subTest(case=name, why=why):
+                case = all_graphs[corpus][name]
+                legacy = legacy_positive(case[0], case[1])
+                self.assertIsNotNone(legacy)
+                result = inspected(*case)
+                positive = result["positive"]
+                text = positive["text"] if isinstance(positive, dict) else None
+                if result["state"] != "resolved":
+                    text = None
+                self.assertEqual(text, expected)
+                self.assertNotEqual(text, legacy)
+
+    def test_no_graph_leaves_the_legacy_agreement(self):
+        recorded = json.loads(AGREEMENT_PATH.read_text(encoding="utf-8"))
+        current = {name: agreeing(graphs) for name, graphs in corpora().items()}
+        self.assertEqual(set(recorded), set(current))
+        for corpus, names in recorded.items():
+            with self.subTest(corpus=corpus):
+                lost = sorted(set(names) - set(current[corpus]))
+                self.assertEqual(lost, [], "these graphs no longer match legacy")
+
+    def test_every_unknown_node_shape_legacy_reads_still_agrees(self):
+        graphs = corpora()["unknown_nodes"]
+        legacy = {name for name, pair in graphs.items() if legacy_positive(*pair)}
+        self.assertLessEqual(legacy, set(agreeing(graphs)))
+
+    def test_the_reported_graph_reads_better_than_legacy(self):
+        # Legacy found no positive here (cf0cef3 showed the positive text as
+        # the negative); the floor is met, and the inspector now reads it.
+        graph = unknown.any_switch_graph()
+        self.assertIsNone(legacy_positive(graph, None))
+        result = resolver.resolve_prompt_pair(
+            {"prompt": graph}, profile=resolver.INSPECTION_PROFILE
+        )
+        self.assertEqual(result["positive"]["text"], unknown.POSITIVE)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--write", action="store_true", help="record the agreement")
+    if parser.parse_args().write:
+        current = {name: agreeing(graphs) for name, graphs in corpora().items()}
+        text = json.dumps(current, indent=1, sort_keys=True, ensure_ascii=False)
+        AGREEMENT_PATH.write_text(text + "\n", encoding="utf-8", newline="\n")
+
+
+if __name__ == "__main__":
+    main()

tests/test_local_inspect_parity.py

@@ -1118,8 +1118,17 @@ class LocalProjectionParityTests(unittest.TestCase):
         self.assertTrue({"parsed", "malformed", "limited"} <= seen)
 
 
-def _server_evidence(records):
-    return inspect_prompt_evidence(records, None, profile=INSPECTION_PROFILE)
+def _server_evidence(records, file_name=None):
+    return inspect_prompt_evidence(
+        records, None, profile=INSPECTION_PROFILE, file_name=file_name
+    )
+
+
+FILE_NAME = resolver_cases.FILE_NAME
+
+
+def _records(case):
+    return {key: value for key, value in case.items() if key != FILE_NAME}
 
 
 class LocalResolverParityTests(unittest.TestCase):
@@ -1143,7 +1152,12 @@ class LocalResolverParityTests(unittest.TestCase):
         observed = _run_oracle(
             self,
             [
-                {"name": name, "mode": "resolve", "records": records}
+                {
+                    "name": name,
+                    "mode": "resolve",
+                    "records": _records(records),
+                    "fileName": records.get(FILE_NAME),
+                }
                 for name, records in cases.items()
             ],
         )
@@ -1151,7 +1165,7 @@ class LocalResolverParityTests(unittest.TestCase):
         for name, records in cases.items():
             with self.subTest(case=name):
                 local = observed[name]
-                server = _server_evidence(records)
+                server = _server_evidence(_records(records), records.get(FILE_NAME))
                 answers[name] = server
                 self.assertEqual(
                     _canonical_bytes(local["evidence"]), _canonical_bytes(server)
@@ -1190,7 +1204,7 @@ class LocalResolverParityTests(unittest.TestCase):
                 # The single-chain subset (Task 8); the rest is Task 9's test below.
                 if not resolver_cases.single_chain(records):
                     continue
-                server = _server_evidence(records)
+                server = _server_evidence(records, Path(path).name)
                 self.assertEqual(
                     _canonical_bytes(local["evidence"]), _canonical_bytes(server)
                 )
@@ -1237,7 +1251,8 @@ class LocalResolverParityTests(unittest.TestCase):
                 local, expect = observed[name], vector["expect"]
                 try:
                     server = _server_evidence(
-                        read_inspection_file(files[name]).metadata
+                        read_inspection_file(files[name]).metadata,
+                        Path(files[name]).name,
                     )
                 except MetadataReadError:
                     server = unreadable_evidence()

tests/test_prompt_pair_resolver.py

@@ -1019,7 +1019,8 @@ class TestResolverProfiles(unittest.TestCase):
 
     def test_unsupported_node_names_the_blocking_node(self):
         source = graph()
-        source["1"]["class_type"] = "UnknownEncoder"
+        # No text and no links: nothing passes through it (#381).
+        source["1"] = {"class_type": "UnknownEncoder", "inputs": {"seed": 1}}
         result = resolve_prompt_pair(source, profile=resolver.INSPECTION_PROFILE)
         self.assertEqual(result["reason_code"], "unsupported-node")
         self.assertEqual(result["node"], {"id": "1", "class_type": "UnknownEncoder"})
@@ -1161,14 +1162,9 @@ class TestGuiderSamplers(unittest.TestCase):
 
     def test_unsupported_node_names_first_blocking_node(self):
         source = _guided()
-        source["30"] = {
-            "class_type": "FutureConditioner",
-            "inputs": {"conditioning": ["6", 0]},
-        }
-        source["31"] = {
-            "class_type": "OtherConditioner",
-            "inputs": {"conditioning": ["7", 0]},
-        }
+        # No links: nothing passes through them (#381).
+        source["30"] = {"class_type": "FutureConditioner", "inputs": {"seed": 1}}
+        source["31"] = {"class_type": "OtherConditioner", "inputs": {"seed": 2}}
         source["21"]["inputs"]["positive"] = ["30", 0]
         source["21"]["inputs"]["negative"] = ["31", 0]
         result = _inspect(source)
@@ -1548,7 +1544,8 @@ class TestConditioningSemantics(unittest.TestCase):
     def test_diverging_branch_that_cannot_be_traced_is_never_guessed(self):
         extra = {
             "10": _encoded(parity.POS),
-            "11": {"class_type": "FutureEncoder", "inputs": {"text": WOLF}},
+            # No text and no links: nothing passes through it (#381).
+            "11": {"class_type": "FutureEncoder", "inputs": {"strength": 1.0}},
             "12": _image_exists(),
         }
         source = _with_positive(_if_else(["12", 1], ["10", 0], ["11", 0]), extra)
@@ -2048,12 +2045,17 @@ class TestEncoderAndStringSemantics(unittest.TestCase):
                 result = _inspect(_with_positive(node))
                 self.assertEqual(result["reason_code"], "malformed")
 
-    def test_unverified_text_nodes_stay_unsupported_and_named(self):
-        # TextBoxMira was verified in its source in Task 2.7b and is listed.
+    def test_unknown_text_nodes_pass_their_text_through(self):
+        # #381: an unlisted node's ``text`` input is its text (the legacy rule);
+        # without one, and with nothing linked, it is still named.
         for name in ("LoRAfromText", "RegexReplace", "XYZPrompt"):
             with self.subTest(node=name):
                 node = {"class_type": name, "inputs": {"text": parity.POS}}
                 result = _inspect(_string_positive(node))
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(result["positive"]["text"], parity.POS)
+                node = {"class_type": name, "inputs": {"pattern": parity.POS}}
+                result = _inspect(_string_positive(node))
                 self.assertEqual(result["reason_code"], "unsupported-node")
                 self.assertEqual(result["node"], {"id": "10", "class_type": name})
 
@@ -2313,7 +2315,8 @@ class TestDetailerStages(unittest.TestCase):
 
     def test_unresolved_stage_blocks_the_chain(self):
         source = _detailed(same_prompt=True)
-        source["30"] = {"class_type": "FutureConditioner", "inputs": {"c": ["6", 0]}}
+        # No links: nothing passes through it (#381).
+        source["30"] = {"class_type": "FutureConditioner", "inputs": {"c": 1.0}}
         source["3"]["inputs"]["positive"] = ["30", 0]
         source["50"]["inputs"]["positive"] = ["10", 0]
         result = _inspect(source)
@@ -2662,11 +2665,13 @@ class TestLatentAncestry(unittest.TestCase):
             },
             {
                 "50": {"class_type": "LoadImage", "inputs": {"image": "a.png"}},
+                # Its own encoder (#381: the same links would be one prompt).
+                "73": parity._enc(parity.POS),
                 "70": {
                     "class_type": "KSampler",
                     "inputs": {
                         "model": ["4", 0],
-                        "positive": ["6", 0],
+                        "positive": ["73", 0],
                         "negative": ["7", 0],
                         "latent_image": ["72", 0],
                     },
@@ -2969,7 +2974,8 @@ class TestLatentAncestry(unittest.TestCase):
         source["60"]["class_type"] = "UltimateSDUpscale"
         result = _inspect(source, output_node_id="9")
         self.assertEqual((result["state"], result["sampler_id"]), ("resolved", "3"))
-        # Reached only through an unlisted image node, it stays unattributed.
+        # Reached only through an unlisted image node, it stays unattributed:
+        # a detailer is never a plain first stage, even on the same links (#381).
         source["61"] = {"class_type": "MysteryFilter", "inputs": {"image": ["60", 0]}}
         source["5"]["inputs"]["pixels"] = ["61", 0]
         result = _inspect(source, output_node_id="9")

tests/test_resolver_generic_pass_through.py

@@ -0,0 +1,391 @@
+"""Unknown nodes between the prompt and the sampler pass through (#381).
+
+The inspection resolver no longer needs a node class on an allow-list to read
+through it. An output the semantics table does not describe passes through: a
+``text`` input is the text, a ``positive``/``negative`` input routes by output
+slot, a ``conditioning`` input is followed alone, otherwise every linked input
+except core type inputs (``pass_through.skip_inputs``) is followed. One text,
+or several equal texts, resolve; PromptManager nodes are authoritative; several
+different texts are one candidate each. Ingestion (the default profile) is
+unchanged. Every graph here is synthetic, with neutral texts.
+"""
+
+import json
+import unittest
+
+from utils import prompt_pair_resolver as resolver
+from utils.prompt_inspector import inspect_prompt_evidence
+from utils.prompt_pair_resolver import resolve_prompt_pair
+
+try:
+    from tests import unknown_node_graphs as graphs
+except ImportError:  # discovered with ``-s tests``
+    import unknown_node_graphs as graphs
+
+POSITIVE, OTHER, NEGATIVE = graphs.POSITIVE, graphs.OTHER, graphs.NEGATIVE
+any_switch_graph = graphs.any_switch_graph
+_encode, _pair, _with = graphs.encode, graphs.pair, graphs.with_nodes
+
+
+def _inspect(graph, **options):
+    return resolve_prompt_pair(
+        {"prompt": graph}, profile=resolver.INSPECTION_PROFILE, **options
+    )
+
+
+def _texts(result):
+    return tuple(
+        result[role]["text"] if isinstance(result[role], dict) else None
+        for role in ("positive", "negative")
+    )
+
+
+class TestTheReportedGraph(unittest.TestCase):
+    def test_prompt_manager_pair_through_any_switch_resolves(self):
+        result = _inspect(any_switch_graph())
+        self.assertEqual(result["state"], "resolved", result["reason"])
+        self.assertEqual(_texts(result), (POSITIVE, ""))
+        self.assertEqual(result["sampler_id"], "17")
+        self.assertIsNone(result["reason_code"])
+        path = [step["node_id"] for step in result["positive"]["provenance"]["path"]]
+        self.assertEqual(path, ["15", "10"])
+
+    def test_the_inspector_shows_the_prompt(self):
+        evidence = inspect_prompt_evidence({"prompt": json.dumps(any_switch_graph())})
+        self.assertEqual(evidence["resolution"]["state"], "resolved")
+        self.assertEqual(evidence["prompts"]["positive"]["status"], "resolved")
+        self.assertEqual(evidence["prompts"]["positive"]["text"], POSITIVE)
+
+    def test_ingestion_is_unchanged(self):
+        result = resolve_prompt_pair({"prompt": any_switch_graph()})
+        self.assertEqual(result["state"], "unresolved")
+        # Its first failure is the custom latent node, as before this change.
+        self.assertEqual(result["reason"], "Unsupported latent ancestry")
+
+    def test_the_inspection_version_moves(self):
+        self.assertEqual(resolver.INSPECTION_RESOLVER_VERSION, 2)
+        self.assertEqual(resolver.RESOLVER_VERSION, 1)
+
+
+class TestSourcesThroughUnknownNodes(unittest.TestCase):
+    def test_several_inputs_with_one_text_resolve(self):
+        graph = any_switch_graph(any_01=["10", 0], any_02=["10", 0])
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, ""))
+
+    def test_equal_texts_from_different_nodes_resolve(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["21", 0]),
+            **{"20": _encode(OTHER), "21": _encode(OTHER)},
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(_texts(result)[0], OTHER)
+
+    def test_different_texts_are_candidates_not_a_refusal(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["21", 0]),
+            **{"20": _encode(POSITIVE), "21": _encode(OTHER)},
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "ambiguous")
+        self.assertEqual(result["reason"], resolver.SEVERAL_SOURCES_REASON)
+        self.assertEqual(
+            [_texts(item) for item in result["candidates"]],
+            [(POSITIVE, ""), (OTHER, "")],
+        )
+        for item in result["candidates"]:
+            self.assertEqual(item["state"], "resolved")
+            self.assertEqual(item["sampler_id"], "17")
+
+    def test_prompt_manager_is_authoritative(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["10", 0]),
+            **{"20": _encode(OTHER)},
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(_texts(result), (POSITIVE, ""))
+
+    def test_prompt_manager_wins_over_an_unreadable_branch(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["10", 0]),
+            **{
+                "20": {
+                    "class_type": "CLIPTextEncode",
+                    "inputs": {"text": ["21", 0], "clip": ["7", 0]},
+                },
+                "21": {"class_type": "PromptSearchList", "inputs": {"search": "x"}},
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, ""))
+
+    def test_an_unreadable_branch_blocks_without_prompt_manager(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["22", 0]),
+            **{
+                "20": {
+                    "class_type": "CLIPTextEncode",
+                    "inputs": {"text": ["21", 0], "clip": ["7", 0]},
+                },
+                "21": {"class_type": "PromptSearchList", "inputs": {"search": "x"}},
+                "22": _encode(OTHER),
+            },
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "runtime-value")
+
+    def test_chained_unknown_nodes(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "Router Node (vendor)",
+                    "inputs": {"in": ["10", 0]},
+                }
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, ""))
+
+    def test_no_text_names_the_unknown_node(self):
+        graph = any_switch_graph(any_01=["7", 0])
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "unsupported-node")
+        self.assertEqual(
+            result["node"], {"id": "15", "class_type": "Any Switch (rgthree)"}
+        )
+
+    def test_another_sampler_is_never_entered(self):
+        graph = any_switch_graph(any_01=["17", 0], any_02=["10", 0])
+        graph["30"] = {
+            "class_type": "KSampler",
+            "inputs": {**graph["17"]["inputs"], "positive": ["31", 0]},
+        }
+        graph["31"] = _encode(OTHER)
+        graph["15"]["inputs"]["any_01"] = ["30", 0]
+        result = _inspect(graph)
+        found = {item["sampler_id"]: _texts(item) for item in result["candidates"]}
+        self.assertEqual(found["17"], (POSITIVE, ""))
+
+
+class TestLegacyShapedRules(unittest.TestCase):
+    def test_a_text_input_is_the_text(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorTextEncode",
+                    "inputs": {"text": OTHER, "clip": ["9", 1]},
+                }
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph))[0], OTHER)
+
+    def test_role_inputs_route_by_output_slot(self):
+        graph = any_switch_graph()
+        graph["20"] = {
+            "class_type": "VendorControlApply",
+            "inputs": {
+                "positive": ["21", 0],
+                "negative": ["22", 0],
+                "control_net": ["8", 0],
+                "image": ["8", 0],
+            },
+        }
+        graph["21"] = _encode(POSITIVE)
+        graph["22"] = _encode(NEGATIVE)
+        graph["17"]["inputs"]["positive"] = ["20", 0]
+        graph["17"]["inputs"]["negative"] = ["20", 1]
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, NEGATIVE))
+
+    def test_a_conditioning_input_is_followed_alone(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorStyleApply",
+                    "inputs": {"conditioning": ["10", 0], "style": ["21", 0]},
+                },
+                "21": _encode(OTHER),
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, ""))
+
+    def test_core_type_inputs_are_skipped(self):
+        # The vendor encoder's ``clip`` reaches a loader with its own ``text``:
+        # that text is a lora tag, never the prompt.
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorPromptEncode",
+                    "inputs": {"prompt": ["21", 0], "clip": ["22", 1]},
+                },
+                "21": {"class_type": "PrimitiveString", "inputs": {"value": OTHER}},
+                "22": {
+                    "class_type": "VendorLoraLoader",
+                    "inputs": {"text": "<lora:x:1>", "clip": ["7", 0]},
+                },
+            },
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(_texts(result)[0], OTHER)
+
+
+class TestBounds(unittest.TestCase):
+    def test_a_long_unknown_chain_reaches_the_depth_limit(self):
+        graph = any_switch_graph(any_01=["100", 0])
+        for index in range(100, 100 + resolver.INSPECTION_PROFILE.max_depth + 2):
+            graph[str(index)] = {
+                "class_type": "Router Node (vendor)",
+                "inputs": {"in": [str(index + 1), 0]},
+            }
+        graph[str(index + 1)] = _pair()
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "depth-limit")
+
+    def test_many_different_sources_hit_the_candidate_cap(self):
+        inputs = {}
+        graph = any_switch_graph()
+        for index in range(resolver._MAX_BRANCH_CANDIDATES + 1):
+            graph[str(200 + index)] = _encode(f"{OTHER} {index}")
+            inputs[f"any_{index:02}"] = [str(200 + index), 0]
+        graph["15"]["inputs"] = inputs
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "work-limit")
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestGeneratedAndTransformedText(unittest.TestCase):
+    def test_a_seeded_text_node_is_generated_at_run_time(self):
+        result = _inspect(graphs.generated_text_graph())
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "runtime-value")
+        self.assertEqual(
+            result["node"], {"id": "30", "class_type": "VendorTextGenerate"}
+        )
+
+    def test_the_recorded_text_is_authoritative(self):
+        # The record is what ran: a generated or transformed text is never
+        # checked against the generator's or transform's inputs.
+        for name, records in graphs.recorded_shapes().items():
+            with self.subTest(shape=name):
+                metadata = {
+                    "prompt": json.loads(records["prompt"]),
+                    "comfyassets_prompt_pairs": records["comfyassets_prompt_pairs"],
+                }
+                result = resolve_prompt_pair(
+                    metadata, profile=resolver.INSPECTION_PROFILE
+                )
+                self.assertEqual(result["state"], "resolved", result["reason"])
+                self.assertEqual(_texts(result), (graphs.GENERATED, ""))
+                self.assertEqual(
+                    result["positive"]["provenance"]["text_source"], "execution"
+                )
+
+    def test_a_seeded_conditioning_node_still_passes_through(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorConditioningNoise",
+                    "inputs": {"conditioning": ["10", 0], "seed": 3},
+                }
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph)), (POSITIVE, ""))
+
+
+class TestReviewFindings(unittest.TestCase):
+    """Shapes from the independent review of the first #381 commit."""
+
+    def _recorded(self, text_source: dict) -> dict:
+        graph = any_switch_graph()
+        graph["10"]["inputs"]["text"] = ["30", 0]
+        graph.update(text_source)
+        metadata = {
+            "prompt": graph,
+            "comfyassets_prompt_pairs": graphs.execution("10", graphs.GENERATED),
+        }
+        return resolve_prompt_pair(metadata, profile=resolver.INSPECTION_PROFILE)
+
+    def test_a_failing_walk_never_overrides_the_recorded_text(self):
+        sources = {
+            "missing": {"30": {"class_type": "XLLM", "inputs": {"src": ["99", 0]}}},
+            "composite": {
+                "30": {"class_type": "XLLM", "inputs": {"src": ["31", 0]}},
+                "31": {
+                    "class_type": "ConditioningCombine",
+                    "inputs": {
+                        "conditioning_1": ["32", 0],
+                        "conditioning_2": ["33", 0],
+                    },
+                },
+                "32": _encode(POSITIVE),
+                "33": _encode(OTHER),
+            },
+            "several": {
+                "30": {
+                    "class_type": "XLLM",
+                    "inputs": {"a": ["31", 0], "b": ["32", 0]},
+                },
+                "31": {"class_type": "PrimitiveString", "inputs": {"value": POSITIVE}},
+                "32": {"class_type": "PrimitiveString", "inputs": {"value": OTHER}},
+            },
+            "list-text": {"30": {"class_type": "XLLM", "inputs": {"text": [1, 2, 3]}}},
+        }
+        for name, source in sources.items():
+            with self.subTest(source=name):
+                result = self._recorded(source)
+                self.assertEqual(result["state"], "resolved", result["reason"])
+                self.assertEqual(_texts(result)[0], graphs.GENERATED)
+
+    def test_the_seed_rule_reads_only_string_outputs(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {"class_type": "XNoise", "inputs": {"conditioning": ["21", 0]}},
+                "21": {
+                    "class_type": "XSeedEncode",
+                    "inputs": {"clip": ["7", 0], "text": OTHER, "seed": 3},
+                },
+            },
+        )
+        self.assertEqual(_texts(_inspect(graph))[0], OTHER)
+
+    def test_an_unlinked_role_input_adds_nothing(self):
+        graph = any_switch_graph()
+        graph["20"] = {
+            "class_type": "XApply",
+            "inputs": {"positive": ["21", 0], "negative_opt": ""},
+        }
+        graph["21"] = _encode(POSITIVE)
+        graph["17"]["inputs"]["positive"] = ["20", 0]
+        graph["17"]["inputs"]["negative"] = ["20", 1]
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "unsupported-node")
+        self.assertEqual(result["node"], {"id": "20", "class_type": "XApply"})
+
+    def test_one_text_offered_twice_is_one_candidate(self):
+        graph = _with(
+            any_switch_graph(any_01=["20", 0], any_02=["23", 0]),
+            **{
+                "20": {"class_type": "XB", "inputs": {"p": ["21", 0], "q": ["22", 0]}},
+                "21": _encode(POSITIVE),
+                "22": _encode(OTHER),
+                "23": _encode(POSITIVE),
+            },
+        )
+        result = _inspect(graph)
+        self.assertEqual(result["state"], "ambiguous")
+        self.assertEqual(
+            [_texts(item)[0] for item in result["candidates"]], [POSITIVE, OTHER]
+        )

tests/test_resolver_pass_through.py

@@ -72,16 +72,24 @@ class TestHiresFixChain(unittest.TestCase):
     def test_an_unlisted_image_node_breaks_the_lineage(self):
         # The walk still finds the earlier sampler, but not through listed
         # pass-through nodes: the stages are never compared.
-        result = _inspect(shapes.hires_fix(filter_class="MysteryImageEffect"))
+        graph = shapes.hires_fix(
+            second_positive=shapes.POSITIVE, filter_class="MysteryImageEffect"
+        )
+        result = _inspect(graph)
         self.assertEqual(result["state"], "ambiguous")
         later = next(
             item for item in result["candidates"] if item["sampler_id"] == "20"
         )
         self.assertEqual(later["reason"], resolver.MULTISTAGE_UNATTRIBUTED_REASON)
+        # Stages on the very same conditioning links are one prompt (#381).
+        result = _inspect(shapes.hires_fix(filter_class="MysteryImageEffect"))
+        self.assertEqual(result["state"], "resolved")
 
     def test_a_second_earlier_sampler_in_the_ancestry_stays_unattributed(self):
         graph = shapes.hires_fix()
-        graph["30"] = shapes.ksampler(["2", 0], ["3", 0], ["4", 0])
+        # Its own encoder: the same links would be one prompt (#381).
+        graph["33"] = shapes.encode(shapes.POSITIVE)
+        graph["30"] = shapes.ksampler(["33", 0], ["3", 0], ["4", 0])
         graph["31"] = {"class_type": "VAEDecode", "inputs": {"samples": ["30", 0]}}
         graph["32"] = {
             "class_type": "ImageBlend",

tests/test_resolver_saver_anchor.py

@@ -0,0 +1,151 @@
+"""Saver anchoring by file name and shared-conditioning stages (#381).
+
+An inspected file's name anchors the walk on the one saver whose
+``filename_prefix`` (literal or traced) makes ``<prefix>_<digits>[_]``; the
+samplers in that saver's image ancestry are the candidates. When no saver or
+more than one matches, nothing is guessed: the file stays "several possible
+prompts". A later stage reading the same positive and negative links as the
+earlier stage has the same prompt whatever its latent ancestry. Shapes come
+from the maintainer's runs (``tests/saver_anchor_graphs.py``).
+"""
+
+import json
+import unittest
+
+from utils import prompt_pair_resolver as resolver
+from utils.prompt_inspector import inspect_prompt_evidence
+
+try:
+    from tests import saver_anchor_graphs as shapes
+except ImportError:  # discovered with ``-s tests``
+    import saver_anchor_graphs as shapes
+
+
+def _inspect(graph, file_name=None):
+    return resolver.resolve_prompt_pair(
+        {"prompt": graph}, profile=resolver.INSPECTION_PROFILE, file_name=file_name
+    )
+
+
+def _positive(result):
+    side = result["positive"]
+    return side["text"] if isinstance(side, dict) else None
+
+
+class TestSaverAnchor(unittest.TestCase):
+    def test_the_file_name_picks_its_branch(self):
+        for index, text in enumerate(shapes.BRANCH_TEXTS, start=1):
+            for name in (
+                f"s45_branch_{index}_00002_.png",
+                f"s45_branch_{index}_00002.webp",
+            ):
+                with self.subTest(file=name):
+                    result = _inspect(shapes.branch_bench(), name)
+                    self.assertEqual(result["state"], "resolved")
+                    self.assertEqual(_positive(result), text)
+                    self.assertEqual(result["sampler_id"], str(index * 10 + 1))
+
+    def test_unanchored_branches_stay_several(self):
+        self.assertEqual(_inspect(shapes.branch_bench())["state"], "ambiguous")
+
+    def test_a_traced_prefix_anchors(self):
+        result = _inspect(shapes.linked_prefix_bench(), "s45_branch_3_00007_.png")
+        self.assertEqual(_positive(result), shapes.BRANCH_TEXTS[2])
+
+    def test_an_unknown_saver_class_anchors_too(self):
+        result = _inspect(
+            shapes.branch_bench(saver="VendorSaveImage"), "s45_branch_2_00001_.png"
+        )
+        self.assertEqual(_positive(result), shapes.BRANCH_TEXTS[1])
+
+    def test_no_guessing_when_the_name_matches_none_or_several(self):
+        cases = {
+            "no saver matches": (shapes.branch_bench(), "photo_00001_.png"),
+            "not a counter": (shapes.branch_bench(), "s45_branch_3_final.png"),
+            "prefix of another branch": (
+                shapes.branch_bench(),
+                "s45_branch_00001_.png",
+            ),
+            "shared prefix": (shapes.shared_prefix(), "IMG_GEN_00371_.png"),
+        }
+        for label, (graph, name) in cases.items():
+            with self.subTest(case=label):
+                result = _inspect(graph, name)
+                self.assertEqual(result["state"], "ambiguous")
+                self.assertEqual(len(result["candidates"]), len(shapes.BRANCH_TEXTS))
+
+    def test_the_inspector_passes_the_file_name(self):
+        records = {"prompt": json.dumps(shapes.branch_bench())}
+        evidence = inspect_prompt_evidence(records, file_name="s45_branch_4_00009_.png")
+        self.assertEqual(
+            evidence["prompts"]["positive"]["text"], shapes.BRANCH_TEXTS[3]
+        )
+        self.assertEqual(
+            inspect_prompt_evidence(records)["resolution"]["state"], "ambiguous"
+        )
+
+    def test_ingestion_ignores_the_file_name(self):
+        result = resolver.resolve_prompt_pair(
+            {"prompt": shapes.branch_bench()}, file_name="s45_branch_1_00001_.png"
+        )
+        self.assertEqual(result["state"], "ambiguous")
+
+
+class TestSharedConditioningStages(unittest.TestCase):
+    def test_a_later_stage_on_the_same_links_resolves(self):
+        for name in (None, "final_00003_.png"):
+            with self.subTest(file=name):
+                result = _inspect(shapes.two_stage_same_conditioning(), name)
+                self.assertEqual(result["state"], "resolved", result["reason"])
+                self.assertEqual(_positive(result), shapes.BRANCH_TEXTS[0])
+
+    def test_different_links_stay_unattributed(self):
+        graph = shapes.two_stage_same_conditioning(second_positive=["91", 0])
+        result = _inspect(graph, "final_00003_.png")
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason"], resolver.MULTISTAGE_UNATTRIBUTED_REASON)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestReviewFindings(unittest.TestCase):
+    """Shapes from the independent review of the saver-anchor commit."""
+
+    def test_a_skipped_stage_never_hides_its_own_earlier_prompt(self):
+        result = _inspect(shapes.hidden_earlier_prompt(), "out_00001_.png")
+        self.assertNotEqual(result["state"], "resolved")
+        self.assertEqual(result["reason"], resolver.MULTISTAGE_UNATTRIBUTED_REASON)
+
+    def test_a_skipped_stage_never_hides_its_override(self):
+        result = _inspect(shapes.hidden_earlier_prompt(override=True), "out_00001_.png")
+        self.assertNotEqual(result["state"], "resolved")
+
+    def test_an_unfinished_prefix_scan_anchors_nothing(self):
+        graph = shapes.deep_prefixes(9)
+        for visits in range(10, 30):
+            with self.subTest(visits=visits):
+                result = resolver.resolve_prompt_pair(
+                    {"prompt": graph},
+                    profile=resolver.INSPECTION_PROFILE,
+                    max_visits=visits,
+                    file_name="dup_00001_.png",
+                )
+                self.assertNotEqual(result["state"], "resolved")
+
+
+class TestPlainFirstStage(unittest.TestCase):
+    def test_only_a_plain_first_stage_is_skipped(self):
+        expected = {
+            "no-latent": "resolved",
+            "literal-latent": "unresolved",
+            "missing-latent": "unresolved",
+            "cyclic-latent": "unresolved",
+            "stage-shaped": "unresolved",
+            "detailer": "unresolved",
+        }
+        for kind, state in expected.items():
+            with self.subTest(kind=kind):
+                result = _inspect(shapes.first_stage_variant(kind), "final_00003_.png")
+                self.assertEqual(result["state"], state, result["reason"])

tests/test_review_inspection.py

@@ -864,7 +864,11 @@ class TestInspectionResolverVersion(InspectionTestCase):
             }
 
         with (
-            patch.object(review_capture, "INSPECTION_RESOLVER_VERSION", 2),
+            patch.object(
+                review_capture,
+                "INSPECTION_RESOLVER_VERSION",
+                review_capture.INSPECTION_RESOLVER_VERSION + 1,
+            ),
             patch.object(review_capture, "resolve_inspection", side_effect=newer),
         ):
             self.assertNotEqual(_version(), version)

tests/unknown_node_graphs.py

@@ -0,0 +1,321 @@
+"""Synthetic graphs with unknown nodes between the prompt and the sampler (#381).
+
+``any_switch_graph`` reproduces the shape of the maintainer's krea2 run
+(PromptManagerPair -> Any Switch (rgthree) -> KSampler, a LoraManager loader
+with its own lora ``text``, custom latent, filename and saver nodes) with
+neutral texts. ``shapes()`` names one graph per pass-through rule; the resolver
+tests and the browser-port parity vectors share them. No user file is involved.
+"""
+
+import json
+
+POSITIVE = "a lighthouse on a cliff, golden hour"
+OTHER = "a red bicycle in the rain"
+NEGATIVE = "blurry, lowres"
+
+
+def encode(text):
+    return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["7", 0]}}
+
+
+def pair(text=POSITIVE, negative=""):
+    return {
+        "class_type": "PromptManagerPair",
+        "inputs": {
+            "text": text,
+            "negative_text": negative,
+            "category": "",
+            "tags": "",
+            "search_text": "",
+            "prepend_text": "",
+            "append_text": "",
+            "clip": ["9", 1],
+        },
+    }
+
+
+def any_switch_graph(**switch_inputs):
+    """The shape of #381: PromptManagerPair -> Any Switch (rgthree) -> KSampler.
+
+    Classes and links reproduce the maintainer's krea2 run (a LoraManager
+    loader with its own lora ``text``, a project-path filename node, a custom
+    latent and saver); the texts are neutral.
+    """
+    inputs = switch_inputs or {"any_02": ["10", 0]}
+    return {
+        "3": {"class_type": "SeedHistory", "inputs": {"seed": 1538595059}},
+        "5": {
+            "class_type": "KikoProjectFilePath",
+            "inputs": {"project_name": "demo", "filename": "final"},
+        },
+        "6": {"class_type": "UNETLoader", "inputs": {"unet_name": "model.safetensors"}},
+        "7": {"class_type": "CLIPLoader", "inputs": {"clip_name": "clip.safetensors"}},
+        "8": {"class_type": "VAELoader", "inputs": {"vae_name": "vae.safetensors"}},
+        "9": {
+            "class_type": "Lora Loader (LoraManager)",
+            "inputs": {
+                "text": "<lora:style-demo:1>",
+                "loras": [{"name": "style-demo", "strength": 1, "active": True}],
+                "model": ["6", 0],
+                "clip": ["7", 0],
+            },
+        },
+        "10": pair(),
+        "15": {"class_type": "Any Switch (rgthree)", "inputs": inputs},
+        "16": {
+            "class_type": "EmptyLatentBatch",
+            "inputs": {"width": 1024, "height": 1536, "batch_size": 1},
+        },
+        "17": {
+            "class_type": "KSampler",
+            "inputs": {
+                "seed": ["3", 0],
+                "steps": 8,
+                "cfg": 1.0,
+                "sampler_name": "er_sde",
+                "scheduler": "beta",
+                "denoise": 1.0,
+                "model": ["9", 0],
+                "positive": ["15", 0],
+                "negative": ["10", 1],
+                "latent_image": ["16", 0],
+            },
+        },
+        "18": {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": ["17", 0], "vae": ["8", 0]},
+        },
+        "19": {
+            "class_type": "KikoSaveImage",
+            "inputs": {"filename_prefix": ["5", 0], "images": ["18", 0]},
+        },
+    }
+
+
+def with_nodes(graph, **nodes):
+    return {**graph, **nodes}
+
+
+GENERATED = "a generated caption of a lighthouse"
+
+
+def generated_text_graph():
+    """PromptManagerPair whose ``text`` is a language model's output.
+
+    The generator (``sampling_mode.seed``, ``prompt``, ``system_prompt``)
+    writes new text at run time: its inputs are not the prompt.
+    """
+    graph = any_switch_graph()
+    graph["10"]["inputs"]["text"] = ["30", 0]
+    graph["30"] = {
+        "class_type": "VendorTextGenerate",
+        "inputs": {
+            "prompt": ["31", 0],
+            "system_prompt": ["32", 0],
+            "sampling_mode.seed": 0,
+            "clip": ["7", 0],
+        },
+    }
+    graph["31"] = {"class_type": "PrimitiveStringMultiline", "inputs": {"value": OTHER}}
+    graph["32"] = {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "You"}}
+    return graph
+
+
+def transformed_text_graph():
+    """PromptManagerPair whose ``text`` passes an unknown string transform."""
+    graph = any_switch_graph()
+    graph["10"]["inputs"]["text"] = ["30", 0]
+    graph["30"] = {
+        "class_type": "VendorRegexReplace",
+        "inputs": {"text": OTHER, "pattern": "rain", "replace": "snow"},
+    }
+    return graph
+
+
+def execution(node_id, text):
+    """A ``comfyassets_prompt_pairs`` record of what the pair node ran."""
+    record = {
+        "text": text,
+        "negative_text": "",
+        "positive_text": text,
+        "negative_effective_text": "",
+        "prepend_text": "",
+        "append_text": "",
+    }
+    return json.dumps({"version": 1, "nodes": {node_id: {"records": [record]}}})
+
+
+def recorded_shapes():
+    """Record sets whose PromptManager node recorded the text it ran."""
+    return {
+        "generated-recorded": {
+            "prompt": json.dumps(generated_text_graph()),
+            "comfyassets_prompt_pairs": execution("10", GENERATED),
+        },
+        "transformed-recorded": {
+            "prompt": json.dumps(transformed_text_graph()),
+            "comfyassets_prompt_pairs": execution("10", GENERATED),
+        },
+    }
+
+
+def _sampler_inputs(graph, positive, negative):
+    graph["17"]["inputs"]["positive"] = positive
+    graph["17"]["inputs"]["negative"] = negative
+    return graph
+
+
+def shapes():
+    """One named graph per rule (resolved, candidates, PromptManager, refusals)."""
+    vendor_encode = {
+        "class_type": "VendorPromptEncode",
+        "inputs": {"prompt": ["21", 0], "clip": ["22", 1]},
+    }
+    control = {
+        "class_type": "VendorControlApply",
+        "inputs": {
+            "positive": ["21", 0],
+            "negative": ["22", 0],
+            "control_net": ["8", 0],
+            "image": ["8", 0],
+        },
+    }
+    return {
+        "any-switch-pair": any_switch_graph(),
+        "any-switch-same-pair": any_switch_graph(any_01=["10", 0], any_02=["10", 0]),
+        "any-switch-equal-texts": with_nodes(
+            any_switch_graph(any_01=["20", 0], any_02=["21", 0]),
+            **{"20": encode(OTHER), "21": encode(OTHER)},
+        ),
+        "any-switch-candidates": with_nodes(
+            any_switch_graph(any_01=["20", 0], any_02=["21", 0]),
+            **{"20": encode(POSITIVE), "21": encode(OTHER)},
+        ),
+        "any-switch-prompt-manager-wins": with_nodes(
+            any_switch_graph(any_01=["20", 0], any_02=["10", 0]),
+            **{"20": encode(OTHER)},
+        ),
+        "any-switch-runtime-blocks": with_nodes(
+            any_switch_graph(any_01=["20", 0], any_02=["22", 0]),
+            **{
+                "20": {
+                    "class_type": "CLIPTextEncode",
+                    "inputs": {"text": ["21", 0], "clip": ["7", 0]},
+                },
+                "21": {"class_type": "PromptSearchList", "inputs": {"search": "x"}},
+                "22": encode(OTHER),
+            },
+        ),
+        "any-switch-no-text": any_switch_graph(any_01=["7", 0]),
+        "unknown-chain": with_nodes(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "Router Node (vendor)",
+                    "inputs": {"in": ["10", 0]},
+                }
+            },
+        ),
+        "unknown-text-input": with_nodes(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorTextEncode",
+                    "inputs": {"text": OTHER, "clip": ["9", 1]},
+                }
+            },
+        ),
+        "unknown-role-inputs": _sampler_inputs(
+            with_nodes(
+                any_switch_graph(),
+                **{"20": control, "21": encode(POSITIVE), "22": encode(NEGATIVE)},
+            ),
+            ["20", 0],
+            ["20", 1],
+        ),
+        "unknown-conditioning-input": with_nodes(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": {
+                    "class_type": "VendorStyleApply",
+                    "inputs": {"conditioning": ["10", 0], "style": ["21", 0]},
+                },
+                "21": encode(OTHER),
+            },
+        ),
+        "generated-text-runtime": generated_text_graph(),
+        "transformed-text": transformed_text_graph(),
+        "unknown-skips-clip": with_nodes(
+            any_switch_graph(any_01=["20", 0]),
+            **{
+                "20": vendor_encode,
+                "21": {"class_type": "PrimitiveString", "inputs": {"value": OTHER}},
+                "22": {
+                    "class_type": "VendorLoraLoader",
+                    "inputs": {"text": "<lora:x:1>", "clip": ["7", 0]},
+                },
+            },
+        ),
+    }
+
+
+def review_shapes():
+    """Record sets for the review findings on the first #381 commit."""
+
+    def recorded(source):
+        graph = any_switch_graph()
+        graph["10"]["inputs"]["text"] = ["30", 0]
+        graph.update(source)
+        return {
+            "prompt": json.dumps(graph),
+            "comfyassets_prompt_pairs": execution("10", GENERATED),
+        }
+
+    seeded = with_nodes(
+        any_switch_graph(any_01=["20", 0]),
+        **{
+            "20": {"class_type": "XNoise", "inputs": {"conditioning": ["21", 0]}},
+            "21": {
+                "class_type": "XSeedEncode",
+                "inputs": {"clip": ["7", 0], "text": OTHER, "seed": 3},
+            },
+        },
+    )
+    unlinked = any_switch_graph()
+    unlinked["20"] = {
+        "class_type": "XApply",
+        "inputs": {"positive": ["21", 0], "negative_opt": ""},
+    }
+    unlinked["21"] = encode(POSITIVE)
+    unlinked["17"]["inputs"]["positive"] = ["20", 0]
+    unlinked["17"]["inputs"]["negative"] = ["20", 1]
+    twice = with_nodes(
+        any_switch_graph(any_01=["20", 0], any_02=["23", 0]),
+        **{
+            "20": {"class_type": "XB", "inputs": {"p": ["21", 0], "q": ["22", 0]}},
+            "21": encode(POSITIVE),
+            "22": encode(OTHER),
+            "23": encode(POSITIVE),
+        },
+    )
+    return {
+        "recorded-missing": recorded(
+            {"30": {"class_type": "XLLM", "inputs": {"src": ["99", 0]}}}
+        ),
+        "recorded-several": recorded(
+            {
+                "30": {
+                    "class_type": "XLLM",
+                    "inputs": {"a": ["31", 0], "b": ["32", 0]},
+                },
+                "31": {"class_type": "PrimitiveString", "inputs": {"value": POSITIVE}},
+                "32": {"class_type": "PrimitiveString", "inputs": {"value": OTHER}},
+            }
+        ),
+        "recorded-list-text": recorded(
+            {"30": {"class_type": "XLLM", "inputs": {"text": [1, 2, 3]}}}
+        ),
+        "seeded-encoder-behind-noise": {"prompt": json.dumps(seeded)},
+        "unlinked-role-input": {"prompt": json.dumps(unlinked)},
+        "one-text-twice": {"prompt": json.dumps(twice)},
+    }

utils/inspection_semantics.py

@@ -13,7 +13,7 @@ profile never imports it, so a missing or bad table can never break ingestion.
 An invalid table raises ``SemanticsError`` so a bad edit fails loudly instead of
 silently changing what the inspector resolves. The file is bounded in size and
 nesting before it is validated. Every class appears in exactly one prompt
-category; classes that are not listed stay ``unsupported-node``. ``lineage``
+category; an unlisted output of a listed class stays ``unsupported-node``. ``lineage``
 (image and latent pass-through outputs for multi-stage chains) describes no
 prompt, so its classes are not claimed by a category.
 
@@ -23,9 +23,13 @@ keys, no whitespace): any change of content, never of layout, moves it.
 folds into the version of the items it records from the inspection profile
 (``utils.review_capture``): the SHA-256 of ``class_projection``, the table
 restricted to what it says about ``classes`` (the node classes of one graph)
-plus the class-independent ``schema`` and ``stage_inputs``. The resolver looks
-the table up only by the class of a graph node, so an edit about other
-classes cannot change how that graph resolves and leaves the version alone.
+plus the class-independent ``schema``, ``stage_inputs`` and ``pass_through``.
+The resolver looks the table up only by the class of a graph node, so an edit
+about other classes cannot change how that graph resolves and leaves the
+version alone. A class the table does not list at all passes text through
+(#381); that rule lives in code, so ``INSPECTION_RESOLVER_VERSION`` moves for it.
+``pass_through.skip_inputs`` names the core type inputs (model, clip, image,
+...) such a node's text is never read through.
 """
 
 import hashlib
@@ -86,7 +90,7 @@ _CATEGORIES = frozenset(
     + _CLASS_CATEGORIES
     + _SLOT_CATEGORIES
     + _SET_CATEGORIES
-    + ("lineage", "stage_inputs", "facts")
+    + ("lineage", "stage_inputs", "pass_through", "facts")
 )
 # Input names that make an unlisted node a possible sampler stage in an ancestry
 # walk (``shaped``: pipes, video embeds and custom-sampler parts carry a prompt
@@ -95,6 +99,10 @@ _CATEGORIES = frozenset(
 # ``model`` alone (ModelSampling*) or with ``clip`` (LoraLoader) is a model patch.
 # The resolver and comfy-metadata.js both read these lists.
 _STAGE_INPUT_LISTS = ("shaped", "with_model")
+# Inputs of core ComfyUI types (model, clip, image, ...) that an unknown node's
+# output is never read through when the resolver passes text through it (#381):
+# a loader's lora tag reached through ``clip`` is not the prompt.
+_PASS_THROUGH_KEYS = ("skip_inputs",)
 _STRING_PARAMS = MappingProxyType(
     {
         "literal": frozenset(),
@@ -127,6 +135,7 @@ class Semantics:
     non_file_outputs: FrozenSet[str]
     lineage: Mapping[Tuple[str, int], Mapping[str, Any]]
     stage_inputs: Mapping[str, Tuple[str, ...]]
+    pass_through: Mapping[str, Tuple[str, ...]]
     facts: Mapping[str, Any]
     all_classes: FrozenSet[str]
     # SHA-256 of the canonical table JSON (set by ``load_semantics``).
@@ -447,6 +456,19 @@ def _stage_inputs(doc: dict) -> Mapping[str, Tuple[str, ...]]:
     return _freeze(table)
 
 
+def _pass_through(doc: dict) -> Mapping[str, Tuple[str, ...]]:
+    table = doc["pass_through"]
+    if not isinstance(table, dict) or set(table) != set(_PASS_THROUGH_KEYS):
+        raise SemanticsError(
+            f"pass_through must have exactly {list(_PASS_THROUGH_KEYS)}"
+        )
+    _names(table["skip_inputs"], "pass_through.skip_inputs")
+    for name in ("text", "positive", "negative", "conditioning"):
+        if name in table["skip_inputs"]:
+            raise SemanticsError(f"pass_through.skip_inputs: {name} carries text")
+    return _freeze(table)
+
+
 def _facts(doc: dict) -> Mapping[str, Any]:
     facts = doc["facts"]
     if not isinstance(facts, dict) or set(facts) != set(_FACTS_TABLES):
@@ -506,6 +528,7 @@ def _validate(doc: Any) -> Semantics:
     return Semantics(
         lineage=_lineage(doc),
         stage_inputs=_stage_inputs(doc),
+        pass_through=_pass_through(doc),
         facts=_facts(doc),
         all_classes=owners.classes(),
         **tables,
@@ -542,7 +565,7 @@ def load_semantics(path: Optional[Union[str, Path]] = None) -> Semantics:
 
 
 # Table parts that hold no class names: they apply to every graph.
-_CLASS_INDEPENDENT = frozenset({"schema", "stage_inputs"})
+_CLASS_INDEPENDENT = frozenset({"schema", "stage_inputs", "pass_through"})
 
 
 def _canonical_json(doc: Any) -> str:

utils/prompt_inspector.py

@@ -534,6 +534,7 @@ def inspect_prompt_evidence(
     saved: dict | None = None,
     *,
     profile: ResolverProfile = INSPECTION_PROFILE,
+    file_name: str | None = None,
 ) -> dict:
     """Keep uncertainty visible; a saved association never replaces file evidence.
 
@@ -597,7 +598,7 @@ def inspect_prompt_evidence(
     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)
+    resolved = resolve_prompt_pair(canonical, profile=profile, file_name=file_name)
     state = resolved["state"]
     result = {}
     for role in ("positive", "negative"):

utils/prompt_pair_resolver.py

@@ -6,7 +6,9 @@ contains effective ``text``, deduplicated-recipe ``storage_text`` and
 ``provenance``; ComfyUI provenance records the exact output
 slot and complete followed link path. States distinguish a resolved pair (an
 explicit empty negative is valid), absent data, unsupported/unresolved data and
-ambiguous sampler selection. Unknown conditioning semantics are never flattened.
+ambiguous sampler selection. Listed composites are never flattened; in the
+inspection profile, nodes the semantics table does not name pass their text
+through (#381, ``_GraphResolver.passed``).
 
 Future nodes can register declarative ``OutputAdapter`` entries by
 ``(class_type, output_slot)``. This does not prescribe a future paired-node schema.
@@ -48,7 +50,9 @@ RESOLVER_VERSION = 1
 # review items recorded from that profile (with the profile's budgets and
 # the semantics table). Bump it when a change would make the inspection
 # profile resolve the same metadata otherwise, even if ingestion is unchanged.
-INSPECTION_RESOLVER_VERSION = 1
+# 2: unknown nodes pass through generically (#381); unknown classes are not in
+# the table's class projection, so only this number can move those items.
+INSPECTION_RESOLVER_VERSION = 2
 
 
 @dataclass(frozen=True)
@@ -113,6 +117,15 @@ _HISTORICAL_ADAPTERS = {
 for _name in ("PromptManagerPositive", "PromptManagerDetailer"):
     _HISTORICAL_ADAPTERS[(_name, 0)] = _COMMA
     _HISTORICAL_ADAPTERS[(_name, 1)] = replace(_COMMA, kind="string")
+# This repository's own node classes, current and historical: their text is the
+# prompt whatever unknown nodes sit between them and the sampler (#381).
+_OWN_CLASSES = frozenset(
+    name
+    for name, _ in list(_DEFAULT_ADAPTERS) + list(_HISTORICAL_ADAPTERS)
+    if name.startswith("PromptManager")
+)
+# Nodes with these exact names are traced by code, not by the table.
+_CODED_CLASSES = frozenset({"Reroute", "StringConcatenate", "Text Concatenate"})
 
 
 @dataclass(frozen=True)
@@ -197,6 +210,17 @@ _ZEROED_SOURCE = "zeroed-conditioning"
 _MAX_BRANCH_CANDIDATES = 32
 # Statically undecidable text (a search, a list output): never pick an item.
 _RUNTIME_REASON = "Text is decided at run time"
+# Different texts reach one sampler through unknown nodes (#381): one candidate
+# per text, never a refusal.
+SEVERAL_SOURCES_REASON = "Several prompts reach the sampler"
+SELECTOR_BRANCHES_REASON = "Selector branches have different prompts"
+# Failures that bound the work: never skipped by a pass-through walk.
+_LIMIT_CODES = frozenset({"work-limit", "depth-limit", "text-limit"})
+_PASSED_UNVERIFIABLE = "Text passed through an unknown node is not verifiable"
+# Inputs that make an unknown node a conditioning transform, not a generator.
+_CONDITIONING_INPUTS = ("conditioning", "positive", "negative")
+# The first item of a pass-through divergence's 3-tuple choice key.
+_PASS_THROUGH = "pass-through"
 # Uncoded outcomes the inspector explains by their exact reason.
 EMPTY_POSITIVE_REASON = "Positive prompt is empty"
 # Several sampler branches, every one resolved to the same pair (#249).
@@ -241,17 +265,23 @@ class _Unresolved(Exception):
 
 
 class _Diverging(Exception):
-    """A runtime selector whose branches trace to different texts.
+    """A runtime choice whose options trace to different texts.
 
     Raised from inside tracing; ``_GraphResolver.branches`` resolves the sampler
-    again once per choice, so each candidate is internally consistent.
+    again once per choice, so each candidate is internally consistent. A
+    selector's choices are ``(True, False)``; an unknown node passing several
+    texts through (#381) has one choice per distinct text, ``0..n-1``.
     """
 
-    def __init__(self, key: Tuple[str, int]) -> None:
+    def __init__(self, key: tuple, choices: tuple, sources: bool = False) -> None:
         super().__init__(key)
-        # The condition output (node ID, slot): one runtime value, so every
-        # selector it drives is forced together.
+        # A selector's condition output (node ID, slot): one runtime value, so
+        # every selector it drives is forced together. A pass-through key is
+        # ``(_PASS_THROUGH, node ID, slot)``.
         self.key = key
+        self.choices = choices
+        # Raised by an unknown node passing several texts (the outcome's reason).
+        self.sources = sources
 
 
 def _node_ref(node_id: str, class_type: Any) -> dict:
@@ -447,6 +477,8 @@ class _GraphResolver:
         self.latent_charged: set = set()
         # ``semantics`` (inspection) names the samplers and guiders it follows.
         self.semantics = semantics
+        # Nesting depth of execution-record checks (``passed``).
+        self.verifying = 0
         if semantics is None:
             self.sampler_classes = self.stage_classes = _SAMPLERS
         else:
@@ -457,6 +489,19 @@ class _GraphResolver:
             )
             # Any listed sampler upstream in latent ancestry is an earlier stage.
             self.stage_classes = frozenset(semantics.samplers)
+            # Classes the resolver knows something about: an unlisted output of
+            # one stays ``unsupported-node``; any other class passes through.
+            facts = semantics.facts
+            self.described = (
+                semantics.all_classes
+                | {name for name, _ in semantics.lineage}
+                | {name for name, _ in self.adapters}
+                | _CODED_CLASSES
+                | frozenset(facts["configuration_outputs"])
+                | frozenset(facts["model_outputs"])
+                | frozenset(facts["model_passthroughs"])
+                | frozenset(facts["scalar_outputs"])
+            )
 
     def node(self, node_id: str) -> Tuple[str, dict]:
         if self.legacy:
@@ -557,7 +602,9 @@ class _GraphResolver:
                     self.inherit_source(node_id, slot, [path])
             elif (name, slot) in self.adapters:
                 adapter = self.adapters[(name, slot)]
-                if not isinstance(adapter, OutputAdapter) or adapter.kind != kind:
+                if not isinstance(adapter, OutputAdapter) or (
+                    kind != "any" and adapter.kind != kind
+                ):
                     raise _Unresolved(_INCOMPATIBLE_SLOT, "malformed")
                 evidence = (
                     self.execution_record(node_id, inputs)
@@ -613,7 +660,7 @@ class _GraphResolver:
             elif (
                 name in ("StringConcatenate", "Text Concatenate")
                 and slot == 0
-                and kind == "string"
+                and kind in ("string", "any")
             ):
                 if name == "StringConcatenate":
                     keys = ["string_a", "string_b"]
@@ -652,6 +699,10 @@ class _GraphResolver:
                 text, path, storage_text = self.semantic(
                     node_id, name, slot, kind, inputs, depth
                 )
+            elif self.semantics is not None and name not in self.described:
+                text, path, storage_text = self.passed(
+                    node_id, slot, kind, inputs, depth
+                )
             else:
                 raise self.unsupported(
                     "Unsupported node semantics or output slot", node_id
@@ -747,7 +798,7 @@ class _GraphResolver:
             return True
         if slot == 0 and name in semantics.selectors:
             return True
-        return kind == "conditioning" and (
+        return kind in ("conditioning", "any") and (
             (name, slot) in semantics.conditioning
             or (slot == 0 and name in semantics.composites)
         )
@@ -769,11 +820,13 @@ class _GraphResolver:
         ):
             listed = table.get((name, slot))
             if listed is not None:
-                if kind != expected:
+                if kind not in (expected, "any"):
                     raise _Unresolved(_INCOMPATIBLE_SLOT, "malformed")
                 return build(node_id, listed, inputs, depth)
         entry = (
-            semantics.conditioning.get((name, slot)) if kind == "conditioning" else None
+            semantics.conditioning.get((name, slot))
+            if kind in ("conditioning", "any")
+            else None
         )
         if entry is not None and entry["strategy"] == "zeroed":
             # The upstream text is discarded, so it is not traced.
@@ -825,9 +878,120 @@ class _GraphResolver:
             for label in ("true", "false")
         ]
         if parts[0][0] != parts[1][0]:
-            raise _Diverging((str(condition[0]), condition[1]))
+            raise _Diverging((str(condition[0]), condition[1]), (True, False))
         return self.merged(node_id, 0, parts)
 
+    def passed(
+        self, node_id: str, slot: int, kind: str, inputs: dict, depth: int
+    ) -> Tuple[str, list, str]:
+        """``read_through``, except while checking an execution record.
+
+        A pair node's recorded text is what ran. While its inputs are checked
+        against the record (``execution_record``), any outcome of a walk through
+        an unknown node but running out of visits only means "cannot be read
+        statically" (``unsupported-node``, unverifiable), as before #381.
+        """
+        if not self.verifying:
+            return self.read_through(node_id, slot, kind, inputs, depth)
+        try:
+            return self.read_through(node_id, slot, kind, inputs, depth)
+        except _Diverging:
+            raise self.unsupported(_PASSED_UNVERIFIABLE, node_id) from None
+        except _Unresolved as error:
+            if error.code == "work-limit":
+                raise
+            raise self.unsupported(_PASSED_UNVERIFIABLE, node_id) from None
+
+    def read_through(
+        self, node_id: str, slot: int, kind: str, inputs: dict, depth: int
+    ) -> Tuple[str, list, str]:
+        """An output no rule describes passes its text through (#381).
+
+        A string output (kind ``string``) of a node with a seed input
+        (``seed``, ``noise_seed``, ``sampling_mode.seed``) and no linked
+        conditioning input is generated at run time (a language model, a
+        wildcard or random line): ``runtime-value``, never its inputs. With a
+        linked conditioning input it is a conditioning transform (noise). A
+        ``text`` input is the text (the legacy encoder rule). Otherwise the
+        followed inputs (``pass_inputs``) are traced with kind ``any``: an
+        unknown node may turn a string into conditioning. A followed input that
+        finds no text (``unsupported-node``) adds nothing; limits and runtime
+        choices propagate; any other failure blocks, unless a PromptManager
+        node was reached: this repository's own nodes are authoritative, so
+        only their texts count then. One distinct text is the result; several
+        are a runtime choice with one candidate per text, in input order.
+        """
+        if (
+            kind == "string"
+            and any(name.endswith("seed") for name in inputs)
+            and not any(_link(inputs.get(name)) for name in _CONDITIONING_INPUTS)
+        ):
+            raise self.named(_RUNTIME_REASON, "runtime-value", node_id)
+        if "text" in inputs:
+            traced = self.text(inputs["text"], "string", depth + 1)
+            self.inherit_source(node_id, slot, [traced[1]])
+            return traced
+        found: list = []
+        blocking: Optional[_Unresolved] = None
+        for value in self.pass_inputs(slot, inputs):
+            try:
+                found.append(self.text(value, "any", depth + 1))
+            except _Unresolved as error:
+                if error.code in _LIMIT_CODES:
+                    raise
+                if error.code != "unsupported-node" and blocking is None:
+                    blocking = error
+        owned = [traced for traced in found if self.owned(traced[1])]
+        if owned:
+            found = owned
+        elif blocking is not None:
+            raise blocking
+        if not found:
+            raise self.unsupported("Unsupported node semantics or output slot", node_id)
+        texts = list(dict.fromkeys(traced[0] for traced in found))
+        if len(texts) == 1:
+            return self.merged(node_id, slot, found)
+        if len(texts) > _MAX_BRANCH_CANDIDATES:
+            raise _Unresolved("Too many selector branches", "work-limit")
+        key = (_PASS_THROUGH, node_id, slot)
+        if key not in self.choices:
+            raise _Diverging(key, tuple(range(len(texts))), sources=True)
+        self.tainted = True
+        chosen = texts[self.choices[key]]
+        return self.merged(node_id, slot, [t for t in found if t[0] == chosen])
+
+    def pass_inputs(self, slot: int, inputs: dict) -> list:
+        """The linked inputs an unknown node's output is read through.
+
+        Output slot 0 of a node with a ``positive`` or ``negative`` input reads
+        only ``positive`` (slot 1, ``negative``), nothing when that one is not
+        linked: the paired-conditioning convention, which the legacy parser
+        followed by role. A linked ``conditioning`` is read alone (the
+        legacy order). Otherwise every linked input but the core type inputs
+        (``pass_through.skip_inputs``: model, clip, image, ...) is read.
+        """
+        if slot in (0, 1) and ("positive" in inputs or "negative" in inputs):
+            value = inputs.get(("positive", "negative")[slot])
+            return [value] if _link(value) else []
+        if _link(inputs.get("conditioning")):
+            return [inputs["conditioning"]]
+        skipped = self.semantics.pass_through["skip_inputs"]
+        return [
+            value
+            for name, value in inputs.items()
+            if _link(value) and name not in skipped
+        ]
+
+    def through_unknown(self, path: list) -> bool:
+        """Whether a followed path passes through a node no rule describes."""
+        self.scan(len(path))
+        return any(self.node(step["node_id"])[0] not in self.described for step in path)
+
+    def owned(self, path: list) -> bool:
+        """Whether a followed path passes through one of this repository's nodes."""
+        self.scan(len(path))
+        return any(self.node(step["node_id"])[0] in _OWN_CLASSES for step in path)
+
     def merged(self, node_id: str, slot: int, parts: list) -> Tuple[str, list, str]:
         """One text reached through several followed paths (deduplicated)."""
         text, merged, storage_text = self.joined_paths(parts, parts)
@@ -1021,8 +1185,9 @@ class _GraphResolver:
         if record is None:
             return None
         for name in ("text", "negative_text", "prepend_text", "append_text"):
+            self.verifying += 1
             try:
-                expected = self.text(inputs.get(name, ""), "string")[0]
+                expected, followed, _ = self.text(inputs.get(name, ""), "string")
             except _Unresolved as error:
                 if self.unverifiable(error):
                     value = inputs.get(name)
@@ -1043,6 +1208,11 @@ class _GraphResolver:
                             raise
                     continue
                 raise
+            finally:
+                self.verifying -= 1
+            if self.semantics is not None and self.through_unknown(followed):
+                # A text passed through an unknown node may be transformed there.
+                continue
             if name in ("prepend_text", "append_text"):
                 expected = expected.strip()
             if expected != record[name]:
@@ -1362,6 +1532,121 @@ class _GraphResolver:
             _link(inputs.get(key)) for key in stage_inputs["with_model"]
         )
 
+    def same_conditioning(self, first: str, second: str) -> bool:
+        """Whether two listed samplers read the very same conditioning links.
+
+        Compared by the sampler table's conditioning inputs (positive and
+        negative, or a custom sampler's guider): each a link, equal by node ID
+        text and slot. Then their prompt is one, whatever their stage order.
+        """
+        keys = []
+        for node_id in (first, second):
+            name, inputs = self.node(node_id)
+            conditioning = self.semantics.samplers[name]["conditioning"]
+            values = []
+            for role in sorted(conditioning):
+                value = inputs.get(conditioning[role])
+                if not _link(value):
+                    return False
+                values.append((role, str(value[0]), str(value[1])))
+            keys.append(tuple(values))
+        return keys[0] == keys[1]
+
+    def alone(self, node_id: str) -> bool:
+        """Whether a sampler's prompt is its own links' text alone.
+
+        True when it sets no prompt override and its latent ancestry reaches no
+        listed sampler (a plain first stage). Any failure but running out of
+        visits is ``False``; the walk never recurses into another stage.
+        """
+        name, inputs = self.node(node_id)
+        entry = self.semantics.samplers[name]
+        if entry["strategy"] == _DETAILER:
+            return False
+        if any(
+            inputs.get(override) not in (None, "")
+            for override in entry.get("prompt_overrides", ())
+        ):
+            return False
+        value = inputs.get(entry.get("latent")) if entry.get("latent") else None
+        if value is None:
+            return True
+        if not _link(value):
+            return False
+        queue, queued = self.walk_from(value)
+        try:
+            while queue:
+                current, depth = queue.popleft()
+                if current == node_id or depth >= self.max_depth:
+                    return False
+                self.budget.spend()
+                name, inputs = self.node(current)
+                if name in self.stage_classes or self.stage_shaped(name, inputs):
+                    return False
+                self.enqueue(queue, queued, inputs, depth + 1)
+        except _Unresolved as error:
+            if error.code == "work-limit":
+                raise
+            return False
+        return True
+
+    def saver_stages(self, file_name: Any) -> Optional[list]:
+        """The candidate samplers of the one saver that wrote ``file_name``.
+
+        A saver is any node with a ``filename_prefix`` (literal, or traced as a
+        string) whose last path part makes the file's stem
+        ``<prefix>_<digits>`` with an optional trailing ``_`` (ComfyUI's and
+        VideoHelperSuite's counters). Exactly one saver must match; its
+        linked inputs are walked like a detailer's image ancestry. No match,
+        several, an unreadable walk or no candidate sampler is ``None``: the
+        file is then judged unanchored, never guessed (#381).
+        """
+        stem = _file_stem(file_name)
+        if stem is None:
+            return None
+        self.budget.reset(self.per_candidate_visits)
+        try:
+            matches = [
+                node_id
+                for node_id, node in self.graph.items()
+                if self.writes(node, stem)
+            ]
+            if len(matches) != 1:
+                return None
+            _, inputs = self.node(matches[0])
+            values = [
+                value
+                for name, value in inputs.items()
+                if name != "filename_prefix" and _link(value)
+            ]
+            found = self.stages_from(matches[0], values)
+        except (_Unresolved, _Diverging):
+            return None
+        candidates = [
+            node_id
+            for node_id in found
+            if self.graph[node_id]["class_type"] in self.sampler_classes
+        ]
+        return candidates or None
+
+    def writes(self, node: Any, stem: str) -> bool:
+        """Whether a node's ``filename_prefix`` names the file ``stem``."""
+        if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict):
+            return False
+        prefix = node["inputs"].get("filename_prefix")
+        if _link(prefix):
+            try:
+                prefix = self.text(prefix, "string")[0]
+            except _Diverging:
+                return False
+            except _Unresolved as error:
+                if error.code == "work-limit":
+                    raise  # an unfinished scan anchors nothing (never a guess)
+                return False
+        if not isinstance(prefix, str):
+            return False
+        return _counter_of(prefix.replace("\\", "/").split("/")[-1], stem)
+
     def upstream_stages(self, origin: str, value: Any) -> list:
         """Samplers reached from a detailer's image input, in breadth-first order.
 
@@ -1370,8 +1655,18 @@ class _GraphResolver:
         """
         if not _link(value):
             raise _Unresolved("Expected an image link", "malformed")
+        return self.stages_from(origin, [value])
+
+    def stages_from(self, origin: str, values: list) -> list:
+        """``upstream_stages`` from several start links, queued in order once."""
         found: list = []
-        queue, queued = self.walk_from(value)
+        queue: deque = deque()
+        queued: set = set()
+        for value in values:
+            start = str(value[0])
+            if start not in queued:
+                queued.add(start)
+                queue.append((start, 0))
         while queue:
             node_id, depth = queue.popleft()
             if node_id == origin:
@@ -1563,17 +1858,20 @@ class _GraphResolver:
         """
         pending: list = [{}]
         results: list = []
+        several = False
         try:
             while pending:
                 self.choices = pending.pop(0)
                 try:
                     results.append(self.stage(sampler_id, frozenset()))
                 except _Diverging as diverging:
-                    if len(results) + len(pending) + 2 > _MAX_BRANCH_CANDIDATES:
+                    count = len(diverging.choices)
+                    if len(results) + len(pending) + count > _MAX_BRANCH_CANDIDATES:
                         raise _Unresolved("Too many selector branches", "work-limit")
+                    several = several or diverging.sources
                     pending.extend(
                         {**self.choices, diverging.key: choice}
-                        for choice in (True, False)
+                        for choice in diverging.choices
                     )
         except _Unresolved as error:
             return _result(
@@ -1587,12 +1885,18 @@ class _GraphResolver:
             self.choices = {}
         if all(item["state"] == "unresolved" for item in results):
             return results[0]
+        if several:
+            # Chained unknown nodes can offer one text twice: list it once.
+            unique: dict = {}
+            for item in results:
+                unique.setdefault(_outcome_key(item), item)
+            results = list(unique.values())
         if len({_outcome_key(item) for item in results}) == 1:
             return results[0]
         return _result(
             "ambiguous",
             sampler_id=sampler_id,
-            reason="Selector branches have different prompts",
+            reason=SEVERAL_SOURCES_REASON if several else SELECTOR_BRANCHES_REASON,
             candidates=results,
         )
 
@@ -1750,6 +2054,10 @@ class _GraphResolver:
                 if not traced:
                     lineage, traced = self.lineage_predecessor(value), True
                 if node_id != lineage:
+                    if self.same_conditioning(origin, node_id) and self.alone(node_id):
+                        # Same positive and negative links: the same prompt
+                        # whichever stage this is (#381).
+                        continue
                     # Not the one continued stage: no stage's prompt is chosen.
                     raise _Unresolved(MULTISTAGE_UNATTRIBUTED_REASON)
                 earlier = node_id
@@ -1883,6 +2191,26 @@ def _parameters(value: Any, max_text_length: int, inspection: bool = False) -> d
     return _result("resolved", positive_side, negative_side)
 
 
+def _file_stem(file_name: Any) -> Optional[str]:
+    """A file name's last path part without its extension, or ``None``."""
+    if not isinstance(file_name, str) or not file_name:
+        return None
+    base = file_name.replace("\\", "/").split("/")[-1]
+    dot = base.rfind(".")
+    stem = base[:dot] if dot > 0 else base
+    return stem or None
+
+
+def _counter_of(prefix: str, stem: str) -> bool:
+    """Whether ``stem`` is ``prefix`` + ``_`` + ASCII digits [+ ``_``]."""
+    if not prefix or not stem.startswith(prefix + "_"):
+        return False
+    rest = stem[len(prefix) + 1 :]
+    if rest.endswith("_"):
+        rest = rest[:-1]
+    return rest != "" and all("0" <= char <= "9" for char in rest)
+
+
 def _profile_with(
     profile: ResolverProfile,
     max_visits: Optional[int],
@@ -1927,6 +2255,7 @@ def resolve_prompt_pair(
     max_visits: Optional[int] = None,
     max_depth: Optional[int] = None,
     max_text_length: Optional[int] = None,
+    file_name: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Resolve parsed metadata; a present API graph is authoritative over text.
 
@@ -1937,9 +2266,11 @@ def resolve_prompt_pair(
     ``sampler_id`` (the inspection profile first drops stages that only reach
     previews).
     ``profile`` selects the bounds; explicit ``max_*`` keywords override it.
+    ``file_name`` (inspection only, unanchored) anchors on the one saver that
+    wrote that file (``_GraphResolver.saver_stages``) when exactly one matches.
     """
     profile = _profile_with(profile, max_visits, max_depth, max_text_length)
-    result = _resolve(metadata, output_node_id, adapters, profile)
+    result = _resolve(metadata, output_node_id, adapters, profile, file_name)
     return _legacy_result(result) if profile.legacy else result
 
 
@@ -1992,6 +2323,7 @@ def _resolve(
     output_node_id: Optional[str],
     adapters: Optional[Mapping[Tuple[str, int], OutputAdapter]],
     profile: ResolverProfile,
+    file_name: Optional[str] = None,
 ) -> Dict[str, Any]:
     if not isinstance(metadata, dict) or not metadata:
         return _result("absent", reason="No parsed prompt metadata")
@@ -2047,14 +2379,19 @@ def _resolve(
                 "unresolved", reason=str(error), reason_code=error.code, node=error.node
             )
     else:
-        sampler_ids = [
+        anchored = (
+            resolver.saver_stages(file_name)
+            if semantics is not None and file_name is not None
+            else None
+        )
+        sampler_ids = anchored or [
             node_id
             for node_id, node in normalized.items()
             if isinstance(node, dict)
             and isinstance(node.get("class_type"), str)
             and node["class_type"] in resolver.sampler_classes
         ]
-        if semantics is not None:
+        if semantics is not None and anchored is None:
             sampler_ids = resolver.final_stages(sampler_ids)
     if not sampler_ids:
         return _no_sampler(normalized, profile)

web/js/inspection-semantics.json

@@ -246,6 +246,29 @@
     ],
     "with_model": ["conditioning", "neg_conditioning", "samples", "latent"]
   },
+  "pass_through": {
+    "skip_inputs": [
+      "model",
+      "clip",
+      "vae",
+      "clip_vision",
+      "clip_vision_output",
+      "control_net",
+      "style_model",
+      "upscale_model",
+      "image",
+      "images",
+      "pixels",
+      "mask",
+      "latent",
+      "latent_image",
+      "samples",
+      "noise",
+      "sampler",
+      "sigmas",
+      "guider"
+    ]
+  },
   "facts": {
     "configuration_outputs": {
       "RandomNoise": { "kind": "noise", "fields": ["noise_seed"] },

web/js/workspace/local-inspection.mjs

@@ -106,9 +106,9 @@ export function toInspectionRead(classification, result) {
 const dimension = value => Number.isSafeInteger(value) && value > 0 ? value : null;
 
 /** The roles for one local read (`{prompts, resolution}`): the bounded parse and resolution step of buildLocalPayload. */
-export function localEvidence(read, classification, semantics = null) {
+export function localEvidence(read, classification, semantics = null, fileName = null) {
     const inspection = toInspectionRead(classificationOf(classification), read);
-    return inspection.read === null ? unreadableEvidence() : resolveLocalEvidence(inspection.read.records, semantics);
+    return inspection.read === null ? unreadableEvidence() : resolveLocalEvidence(inspection.read.records, semantics, fileName);
 }
 
 /**
@@ -117,7 +117,7 @@ export function localEvidence(read, classification, semantics = null) {
  * "unavailable" as the server does), so the legacy fill in projectInspectionPayload applies exactly where it does
  * for Images.
  */
-export function buildLocalPayload({ read, file, classification, semantics = null, evidence = localEvidence(read, classification, semantics) }) {
+export function buildLocalPayload({ read, file, classification, semantics = null, evidence = localEvidence(read, classification, semantics, file?.name ?? null) }) {
     const classified = classificationOf(classification);
     const inspection = toInspectionRead(classified, read);
     const filename = sanitizeLocalFilename(file?.name);
@@ -211,7 +211,7 @@ export const localFileAdapter = Object.freeze({
         }
         const semantics = await untilAborted(loadInspectionSemantics(), signal);
         await yieldControl(signal);
-        const evidence = localEvidence(result, secret.classification, semantics);
+        const evidence = localEvidence(result, secret.classification, semantics, secret.filename);
         await yieldControl(signal);
         const file = { name: secret.filename, size: secret.file.size, lastModified: secret.file.lastModified };
         const payload = buildLocalPayload({ read: result, file, classification: secret.classification, semantics, evidence });

web/js/workspace/local-resolver-stages.mjs

@@ -129,7 +129,16 @@ function compositeOf(resolver, value) {
 /** `upstream_stages`: samplers reached from a detailer's image input, breadth first; each node draws a visit. */
 function upstreamStages(resolver, origin, value) {
     if (!isLink(value)) throw new Unresolved('Expected an image link', 'malformed');
-    const found = [], start = String(value[0]), queue = [[start, 0]], queued = new Set([start]);
+    return stagesFrom(resolver, origin, [value]);
+}
+
+/** `stages_from`: `upstream_stages` from several start links, queued in order once. */
+export function stagesFrom(resolver, origin, values) {
+    const found = [], queue = [], queued = new Set();
+    for (const value of values) {
+        const start = String(value[0]);
+        if (!queued.has(start)) { queued.add(start); queue.push([start, 0]); }
+    }
     for (let head = 0; head < queue.length; head++) {
         const [nodeId, depth] = queue[head];
         if (nodeId === origin) throw new Unresolved('Cyclic sampler stage ancestry', 'malformed');
@@ -296,6 +305,49 @@ function lineagePredecessor(resolver, value) {
 }
 
 /** `latent_stages`: the one earlier sampler a latent continues, null for a single stage; else unattributed. */
+/** `same_conditioning`: whether two listed samplers read the very same conditioning links. */
+function sameConditioning(resolver, first, second) {
+    const keys = [];
+    for (const nodeId of [first, second]) {
+        const [name, inputs] = resolver.node(nodeId);
+        const conditioning = resolver.semantics.samplers.get(name).conditioning;
+        const values = [];
+        for (const role of Object.keys(conditioning).sort()) {
+            const value = pyGet(inputs, conditioning[role]);
+            if (!isLink(value)) return false;
+            values.push([role, String(value[0]), String(value[1])]);
+        }
+        keys.push(JSON.stringify(values));
+    }
+    return keys[0] === keys[1];
+}
+
+/** `alone`: no prompt override set and no listed sampler in its latent ancestry (a plain first stage). */
+function alone(resolver, nodeId) {
+    let [name, inputs] = resolver.node(nodeId);
+    const entry = resolver.semantics.samplers.get(name);
+    if (entry.strategy === DETAILER) return false;
+    if ((entry.prompt_overrides ?? []).some(override => { const value = pyGet(inputs, override); return value !== null && value !== undefined && value !== ''; })) return false;
+    const value = entry.latent ? pyGet(inputs, entry.latent) : null;
+    if (value === null || value === undefined) return true;
+    if (!isLink(value)) return false;
+    const start = String(value[0]), queue = [[start, 0]], queued = new Set([start]);
+    try {
+        for (let head = 0; head < queue.length; head++) {
+            const [current, depth] = queue[head];
+            if (current === nodeId || depth >= resolver.maxDepth) return false;
+            resolver.budget.spend();
+            [name, inputs] = resolver.node(current);
+            if (resolver.stageClasses.has(name) || resolver.stageShaped(name, inputs)) return false;
+            resolver.enqueue(queue, queued, inputs, depth + 1);
+        }
+    } catch (error) {
+        if (!(error instanceof Unresolved) || error.code === 'work-limit') throw error;
+        return false;
+    }
+    return true;
+}
+
 export function latentStages(resolver, origin, value) {
     if (!isLink(value)) throw new Unresolved('Unknown latent ancestry', 'malformed');
     const start = String(value[0]), queue = [[start, 0]], queued = new Set([start]);
@@ -308,7 +360,11 @@ export function latentStages(resolver, origin, value) {
         const [name, inputs] = resolver.node(nodeId);
         if (resolver.stageClasses.has(name)) {
             if (!traced) { lineage = lineagePredecessor(resolver, value); traced = true; }
-            if (nodeId !== lineage) throw new Unresolved(MULTISTAGE_UNATTRIBUTED_REASON);
+            if (nodeId !== lineage) {
+                // Same positive and negative links: the same prompt whichever stage this is (#381).
+                if (sameConditioning(resolver, origin, nodeId) && alone(resolver, nodeId)) continue;
+                throw new Unresolved(MULTISTAGE_UNATTRIBUTED_REASON);
+            }
             earlier = nodeId;
             continue;
         }

web/js/workspace/local-resolver-support.mjs

@@ -29,6 +29,15 @@ export const ZEROED_SOURCE = 'zeroed-conditioning';
 export const MAX_BRANCH_CANDIDATES = 32;
 export const RUNTIME_REASON = 'Text is decided at run time';
 export const EMPTY_POSITIVE_REASON = 'Positive prompt is empty';
+// #381: different texts reach one sampler through unknown nodes (one candidate each).
+export const SEVERAL_SOURCES_REASON = 'Several prompts reach the sampler';
+export const SELECTOR_BRANCHES_REASON = 'Selector branches have different prompts';
+// Failures that bound the work: never skipped by a pass-through walk (`_LIMIT_CODES`).
+export const PASS_LIMIT_CODES = new Set(['work-limit', 'depth-limit', 'text-limit']);
+export const PASS_THROUGH = 'pass-through';
+export const PASSED_UNVERIFIABLE = 'Text passed through an unknown node is not verifiable';
+// Classes traced by code, not by the table (`_CODED_CLASSES`).
+export const CODED_CLASSES = ['Reroute', 'StringConcatenate', 'Text Concatenate'];
 export const IDENTICAL_BRANCHES_REASON = 'Every sampler branch uses the same prompt pair';
 export const MULTISTAGE_UNATTRIBUTED_REASON = 'Multistage sampler ancestry is ambiguous';
 export const ADDED_CONDITIONING_REASON = 'A later stage adds conditioning to the base prompt';
@@ -96,6 +105,8 @@ const historical = [
 ];
 for (const name of ['PromptManagerPositive', 'PromptManagerDetailer']) historical.push([name, 0, 'text', 'conditioning', COMMA], [name, 1, 'text', 'string', COMMA]);
 for (const [name, slot, input, kind, options] of historical) ADAPTERS.set(slotKey(name, slot), adapter(input, kind, options));
+// This repository's own classes (`_OWN_CLASSES`): authoritative through unknown nodes (#381).
+export const OWN_CLASSES = new Set([...ADAPTERS.keys()].map(key => key.split('\u0000')[0]).filter(name => name.startsWith('PromptManager')));
 // `known_slots` in execution_record: every adapter slot per class.
 export const ADAPTER_SLOTS = new Map();
 for (const key of ADAPTERS.keys()) {
@@ -112,9 +123,13 @@ export class Unresolved extends Error {
     }
 }
 export class Diverging extends Error {
-    constructor(key) {
+    /** A selector's choices are [true, false]; an unknown node's are 0..n-1, one per distinct text (#381). */
+    constructor(key, choices, sources = false) {
         super('diverging');
         this.key = key;
+        this.choices = choices;
+        // Raised by an unknown node passing several texts (the outcome's reason).
+        this.sources = sources;
     }
 }
 
@@ -247,9 +262,22 @@ export function compileSemantics(doc) {
     }
     const stageInputs = doc.stage_inputs;
     if (!plain(stageInputs) || !Array.isArray(stageInputs.shaped) || !Array.isArray(stageInputs.with_model)) fail('stage_inputs');
+    // #381: core type inputs an unknown node's text is never read through.
+    const passThrough = doc.pass_through;
+    if (!plain(passThrough) || Object.keys(passThrough).length !== 1 || !Array.isArray(passThrough.skip_inputs) ||
+        !passThrough.skip_inputs.length || passThrough.skip_inputs.some(name => typeof name !== 'string' || !name) ||
+        new Set(passThrough.skip_inputs).size !== passThrough.skip_inputs.length ||
+        ['text', 'positive', 'negative', 'conditioning'].some(name => passThrough.skip_inputs.includes(name))) fail('pass_through');
+    // Classes the facts tables describe (loaders, model patches, configuration and scalar outputs).
+    const facts = doc.facts;
+    if (!plain(facts) || !plain(facts.configuration_outputs) || !plain(facts.model_outputs) || !Array.isArray(facts.model_passthroughs) ||
+        !plain(facts.scalar_outputs)) fail('facts');
+    const factClasses = new Set([...Object.keys(facts.configuration_outputs), ...Object.keys(facts.model_outputs), ...facts.model_passthroughs,
+        ...Object.keys(facts.scalar_outputs)]);
     return Object.freeze({
         samplers, guiders, selectors, composites, savers, conditioning, encoders, strings, runtime, nonFileOutputs, lineage,
         stageInputs: { shaped: [...stageInputs.shaped], withModel: [...stageInputs.with_model] }, allClasses: new Set(owners.keys()),
+        skipInputs: new Set(passThrough.skip_inputs), factClasses,
     });
 }
 

web/js/workspace/local-resolver.mjs

@@ -23,7 +23,14 @@ import {
     MAX_JOIN_INPUTS,
     MAX_SUBGRAPH_DEPTH,
     MAX_WORKFLOW_NODES,
+    CODED_CLASSES,
     NO_SAMPLER_NAMES,
+    OWN_CLASSES,
+    PASS_LIMIT_CODES,
+    PASS_THROUGH,
+    SELECTOR_BRANCHES_REASON,
+    SEVERAL_SOURCES_REASON,
+    PASSED_UNVERIFIABLE,
     OVERWRITTEN_PROMPT_REASON,
     PROFILE,
     ROLE_METADATA_KEYS,
@@ -45,7 +52,7 @@ import {
     sideOf,
     slotKey,
 } from './local-resolver-support.mjs';
-import { finalStages, latentStages, outcomeKey, promptPart, sharedPair, stage } from './local-resolver-stages.mjs';
+import { finalStages, latentStages, outcomeKey, promptPart, sharedPair, stage, stagesFrom } from './local-resolver-stages.mjs';
 import { parameters } from './local-parameters.mjs';
 
 export { SemanticsError, compileSemantics, readPairExecution, outcomeMessage, isLink } from './local-resolver-support.mjs';
@@ -76,8 +83,13 @@ class GraphResolver {
         this.latentWalks = new Map();
         this.latentCharged = new Set();
         this.semantics = semantics;
+        // Nesting depth of execution-record checks (`passed`).
+        this.verifying = 0;
         this.samplerClasses = new Set([...semantics.samplers].filter(([, entry]) => CANDIDATE_STRATEGIES.has(entry.strategy)).map(([name]) => name));
         this.stageClasses = new Set(semantics.samplers.keys());
+        // `described`: an unlisted output of one of these stays `unsupported-node`; any other class passes through (#381).
+        this.described = new Set([...semantics.allClasses, ...[...semantics.lineage.keys()].map(key => key.split('\u0000')[0]),
+            ...[...ADAPTERS.keys()].map(key => key.split('\u0000')[0]), ...CODED_CLASSES, ...semantics.factClasses]);
     }
 
     node(nodeId) {
@@ -156,7 +168,7 @@ class GraphResolver {
                 [text, path, storageText] = this.text(inputs.get(keys[0]), kind, depth + 1);
                 this.inheritSource(nodeId, slot, [path]);
             } else if (adapted) {
-                if (adapted.kind !== kind) throw new Unresolved(INCOMPATIBLE_SLOT, 'malformed');
+                if (kind !== 'any' && adapted.kind !== kind) throw new Unresolved(INCOMPATIBLE_SLOT, 'malformed');
                 const evidence = name === 'PromptManagerPair' ? this.executionRecord(nodeId, inputs) : null;
                 if (evidence !== null) {
                     text = evidence[adapted.inputName === 'text' ? 'positive_text' : 'negative_effective_text'];
@@ -178,11 +190,13 @@ class GraphResolver {
                         if (omitted.length) this.omitting.add(slotKey(nodeId, slot));
                     }
                 }
-            } else if ((name === 'StringConcatenate' || name === 'Text Concatenate') && slot === 0 && kind === 'string') {
+            } else if ((name === 'StringConcatenate' || name === 'Text Concatenate') && slot === 0 && (kind === 'string' || kind === 'any')) {
                 [text, path] = this.concatenate(name, inputs, depth);
                 storageText = text;
             } else if (this.follows(name, slot, kind)) {
                 [text, path, storageText] = this.semantic(nodeId, name, slot, kind, inputs, depth);
+            } else if (!this.described.has(name)) {
+                [text, path, storageText] = this.passed(nodeId, slot, kind, inputs, depth);
             } else {
                 throw this.unsupported('Unsupported node semantics or output slot', nodeId);
             }
@@ -258,7 +272,7 @@ class GraphResolver {
         if (semantics.runtime.has(name)) return true;
         if (semantics.encoders.has(key) || semantics.strings.has(key)) return true;
         if (slot === 0 && semantics.selectors.has(name)) return true;
-        return kind === 'conditioning' && (semantics.conditioning.has(key) || (slot === 0 && semantics.composites.has(name)));
+        return (kind === 'conditioning' || kind === 'any') && (semantics.conditioning.has(key) || (slot === 0 && semantics.composites.has(name)));
     }
 
     semantic(nodeId, name, slot, kind, inputs, depth) {
@@ -267,11 +281,11 @@ class GraphResolver {
         for (const [table, expected, build] of [[semantics.encoders, 'conditioning', 'encoded'], [semantics.strings, 'string', 'stringed']]) {
             const listed = table.get(key);
             if (listed !== undefined) {
-                if (kind !== expected) throw new Unresolved(INCOMPATIBLE_SLOT, 'malformed');
+                if (kind !== expected && kind !== 'any') throw new Unresolved(INCOMPATIBLE_SLOT, 'malformed');
                 return this[build](nodeId, listed, inputs, depth);
             }
         }
-        const entry = kind === 'conditioning' ? semantics.conditioning.get(key) : undefined;
+        const entry = kind === 'conditioning' || kind === 'any' ? semantics.conditioning.get(key) : undefined;
         if (entry !== undefined && entry.strategy === 'zeroed') {
             this.textSources.set(slotKey(nodeId, slot), ZEROED_SOURCE);
             return ['', [], ''];
@@ -302,10 +316,114 @@ class GraphResolver {
             return traced;
         }
         const parts = ['true', 'false'].map(label => this.text(pyGet(inputs, branches[label]), kind, depth + 1));
-        if (parts[0][0] !== parts[1][0]) throw new Diverging(slotKey(String(condition[0]), condition[1]));
+        if (parts[0][0] !== parts[1][0]) throw new Diverging(slotKey(String(condition[0]), condition[1]), [true, false]);
         return this.merged(nodeId, 0, parts);
     }
 
+    /** `passed`: `readThrough`, except that while an execution record is checked any outcome but `work-limit` is unverifiable. */
+    passed(nodeId, slot, kind, inputs, depth) {
+        if (!this.verifying) return this.readThrough(nodeId, slot, kind, inputs, depth);
+        try {
+            return this.readThrough(nodeId, slot, kind, inputs, depth);
+        } catch (error) {
+            if (error instanceof Unresolved && error.code === 'work-limit') throw error;
+            if (error instanceof Unresolved || error instanceof Diverging) throw this.unsupported(PASSED_UNVERIFIABLE, nodeId);
+            throw error;
+        }
+    }
+
+    /** `read_through` (#381): an output no rule describes passes its text through; PromptManager nodes are authoritative. */
+    readThrough(nodeId, slot, kind, inputs, depth) {
+        // A string output of a node with a seed input is generated at run time (a language model, a wildcard).
+        // With a linked conditioning input it is a conditioning transform (noise) instead.
+        if (kind === 'string' && [...inputs.keys()].some(name => name.endsWith('seed')) &&
+            !['conditioning', 'positive', 'negative'].some(name => isLink(pyGet(inputs, name)))) throw this.named(RUNTIME_REASON, 'runtime-value', nodeId);
+        if (inputs.has('text')) {
+            const traced = this.text(inputs.get('text'), 'string', depth + 1);
+            this.inheritSource(nodeId, slot, [traced[1]]);
+            return traced;
+        }
+        let found = [], blocking = null;
+        for (const value of this.passInputs(slot, inputs)) {
+            try {
+                found.push(this.text(value, 'any', depth + 1));
+            } catch (error) {
+                if (!(error instanceof Unresolved) || PASS_LIMIT_CODES.has(error.code)) throw error;
+                if (error.code !== 'unsupported-node' && blocking === null) blocking = error;
+            }
+        }
+        const owned = found.filter(traced => this.owned(traced[1]));
+        if (owned.length) found = owned;
+        else if (blocking !== null) throw blocking;
+        if (!found.length) throw this.unsupported('Unsupported node semantics or output slot', nodeId);
+        const texts = [...new Set(found.map(traced => traced[0]))];
+        if (texts.length === 1) return this.merged(nodeId, slot, found);
+        if (texts.length > MAX_BRANCH_CANDIDATES) throw new Unresolved('Too many selector branches', 'work-limit');
+        // JSON escapes U+0000, so this never equals a selector's raw `slotKey` (Python: a 3-tuple vs a 2-tuple).
+        const key = JSON.stringify([PASS_THROUGH, nodeId, slot]);
+        if (!this.choices.has(key)) throw new Diverging(key, texts.map((_, index) => index), true);
+        this.tainted = true;
+        const chosen = texts[this.choices.get(key)];
+        return this.merged(nodeId, slot, found.filter(traced => traced[0] === chosen));
+    }
+
+    /** `pass_inputs`: the role input by output slot, else `conditioning`, else every linked non-type input. */
+    passInputs(slot, inputs) {
+        if ((slot === 0 || slot === 1) && (inputs.has('positive') || inputs.has('negative'))) {
+            const value = pyGet(inputs, slot === 0 ? 'positive' : 'negative');
+            return isLink(value) ? [value] : [];
+        }
+        if (isLink(pyGet(inputs, 'conditioning'))) return [inputs.get('conditioning')];
+        return [...inputs].filter(([name, value]) => isLink(value) && !this.semantics.skipInputs.has(name)).map(([, value]) => value);
+    }
+
+    /** `saver_stages` (#381): the candidate samplers of the one saver whose prefix names the file, or null (never guessed). */
+    saverStages(fileName) {
+        const stem = fileStem(fileName);
+        if (stem === null) return null;
+        this.budget.reset(this.perCandidateVisits);
+        let found;
+        try {
+            const matches = [...this.graph].filter(([, node]) => this.writes(node, stem)).map(([id]) => id);
+            if (matches.length !== 1) return null;
+            const [, inputs] = this.node(matches[0]);
+            const values = [...inputs].filter(([name, value]) => name !== 'filename_prefix' && isLink(value)).map(([, value]) => value);
+            found = stagesFrom(this, matches[0], values);
+        } catch (error) {
+            if (error instanceof Unresolved || error instanceof Diverging) return null;
+            throw error;
+        }
+        const candidates = found.filter(id => this.samplerClasses.has(this.graph.get(id).get('class_type')));
+        return candidates.length ? candidates : null;
+    }
+
+    /** `writes`: whether a node's `filename_prefix` (literal or traced) names the file `stem`. */
+    writes(node, stem) {
+        if (!isDict(node) || !isDict(pyGet(node, 'inputs'))) return false;
+        let prefix = pyGet(node.get('inputs'), 'filename_prefix');
+        if (isLink(prefix)) {
+            try { prefix = this.text(prefix, 'string')[0]; } catch (error) {
+                // An unfinished scan anchors nothing (never a guess): work-limit ends the match scan.
+                if (error instanceof Diverging || (error instanceof Unresolved && error.code !== 'work-limit')) return false;
+                throw error;
+            }
+        }
+        if (typeof prefix !== 'string') return false;
+        return counterOf(prefix.replaceAll('\\', '/').split('/').at(-1), stem);
+    }
+
+    /** `through_unknown`: whether a followed path passes through a node no rule describes. */
+    throughUnknown(path) {
+        this.scan(path.length);
+        return path.some(step => !this.described.has(this.node(step.node_id)[0]));
+    }
+
+    /** `owned`: whether a followed path passes through one of this repository's nodes. */
+    owned(path) {
+        this.scan(path.length);
+        return path.some(step => OWN_CLASSES.has(this.node(step.node_id)[0]));
+    }
+
     merged(nodeId, slot, parts) {
         const traced = this.joinedPaths(parts, parts);
         this.inheritSource(nodeId, slot, parts.map(part => part[1]));
@@ -421,9 +539,10 @@ class GraphResolver {
         const record = readPairExecution(this.metadata, nodeId);
         if (record === null) return null;
         for (const name of ['text', 'negative_text', 'prepend_text', 'append_text']) {
-            let expected;
+            let expected, followed;
+            this.verifying += 1;
             try {
-                expected = this.text(pyGet(inputs, name, ''), 'string')[0];
+                [expected, followed] = this.text(pyGet(inputs, name, ''), 'string');
             } catch (error) {
                 if (error instanceof Unresolved && this.unverifiable(error)) {
                     const value = pyGet(inputs, name);
@@ -436,7 +555,11 @@ class GraphResolver {
                     continue;
                 }
                 throw error;
+            } finally {
+                this.verifying -= 1;
             }
+            // A text passed through an unknown node may be transformed there.
+            if (this.throughUnknown(followed)) continue;
             if (name === 'prepend_text' || name === 'append_text') expected = pyStrip(expected);
             if (expected !== record[name]) throw new Unresolved('Execution evidence does not match recorded node inputs', 'malformed');
         }
@@ -491,7 +614,8 @@ class GraphResolver {
     }
 
     branches(samplerId) {
-        const pending = [new Map()], results = [];
+        const pending = [new Map()];
+        let results = [], several = false;
         try {
             while (pending.length) {
                 this.choices = pending.shift();
@@ -499,8 +623,9 @@ class GraphResolver {
                     results.push(stage(this, samplerId, new Set()));
                 } catch (error) {
                     if (!(error instanceof Diverging)) throw error;
-                    if (results.length + pending.length + 2 > MAX_BRANCH_CANDIDATES) throw new Unresolved('Too many selector branches', 'work-limit');
-                    for (const choice of [true, false]) pending.push(new Map([...this.choices, [error.key, choice]]));
+                    if (results.length + pending.length + error.choices.length > MAX_BRANCH_CANDIDATES) throw new Unresolved('Too many selector branches', 'work-limit');
+                    several = several || error.sources;
+                    for (const choice of error.choices) pending.push(new Map([...this.choices, [error.key, choice]]));
                 }
             }
         } catch (error) {
@@ -510,8 +635,14 @@ class GraphResolver {
             this.choices = new Map();
         }
         if (results.every(item => item.state === 'unresolved')) return results[0];
+        if (several) {
+            // Chained unknown nodes can offer one text twice: list it once.
+            const unique = new Map();
+            for (const item of results) if (!unique.has(outcomeKey(item))) unique.set(outcomeKey(item), item);
+            results = [...unique.values()];
+        }
         if (new Set(results.map(outcomeKey)).size === 1) return results[0];
-        return result('ambiguous', { samplerId, reason: 'Selector branches have different prompts', candidates: results });
+        return result('ambiguous', { samplerId, reason: several ? SEVERAL_SOURCES_REASON : SELECTOR_BRANCHES_REASON, candidates: results });
     }
 
     sides(samplerId) {
@@ -673,8 +804,25 @@ function overwrittenPromptNodes(workflow, graph) {
     return found;
 }
 
-/** `_resolve` for an unanchored inspection (output_node_id None). */
-function resolve(metadata, semantics) {
+/** `_file_stem`: a file name's last path part without its extension, or null. */
+function fileStem(fileName) {
+    if (typeof fileName !== 'string' || !fileName) return null;
+    const base = fileName.replaceAll('\\', '/').split('/').at(-1);
+    const dot = base.lastIndexOf('.');
+    const stem = dot > 0 ? base.slice(0, dot) : base;
+    return stem || null;
+}
+
+/** `_counter_of`: whether `stem` is `prefix` + `_` + ASCII digits [+ `_`]. */
+function counterOf(prefix, stem) {
+    if (!prefix || !stem.startsWith(`${prefix}_`)) return false;
+    let rest = stem.slice(prefix.length + 1);
+    if (rest.endsWith('_')) rest = rest.slice(0, -1);
+    return rest !== '' && /^[0-9]+$/.test(rest);
+}
+
+/** `_resolve` for an unanchored inspection (output_node_id None); `fileName` anchors on its saver (#381). */
+function resolve(metadata, semantics, fileName = null) {
     if (!isDict(metadata) || !metadata.size) return result('absent', { reason: 'No parsed prompt metadata' });
     let graph;
     if (metadata.has('prompt')) graph = metadata.get('prompt');
@@ -688,7 +836,8 @@ function resolve(metadata, semantics) {
         overwritten = overwrittenPromptNodes(pyGet(metadata, 'workflow'), graph);
     }
     const resolver = new GraphResolver(graph, metadata.has('prompt') ? metadata : null, semantics, overwritten);
-    const samplerIds = finalStages(resolver, [...graph].filter(([, node]) => isDict(node) && typeof pyGet(node, 'class_type') === 'string' &&
+    const anchored = fileName === null ? null : resolver.saverStages(fileName);
+    const samplerIds = anchored ?? finalStages(resolver, [...graph].filter(([, node]) => isDict(node) && typeof pyGet(node, 'class_type') === 'string' &&
         resolver.samplerClasses.has(node.get('class_type'))).map(([id]) => id));
     if (!samplerIds.length) return noSampler(graph);
     if (samplerIds.length > 1) {
@@ -734,7 +883,7 @@ export function unreadableEvidence() {
  * local read found (`{prompt, workflow, parameters, comfyassets_prompt_pairs}` texts). `semantics` is
  * `compileSemantics(...)`'s table, or null when it could not be loaded.
  */
-export function resolveLocalEvidence(records, semantics) {
+export function resolveLocalEvidence(records, semantics, fileName = null) {
     const source = records !== null && typeof records === 'object' ? records : {};
     const metadata = new Map(Object.entries(source).filter(([key]) => ROLE_METADATA_KEYS.has(key)));
     const malformedResolution = { state: 'unresolved', sampler_id: null, reason: MALFORMED_EVIDENCE_REASON, reason_code: 'malformed', node: null };
@@ -754,7 +903,7 @@ export function resolveLocalEvidence(records, semantics) {
     }
     const workflow = mayHoldPromptManager(metadata) ? parsedWorkflow(source.workflow ?? null) : null;
     if (workflow !== null && metadata.has('prompt')) metadata.set('workflow', workflow);
-    const resolved = resolve(metadata, semantics);
+    const resolved = resolve(metadata, semantics, fileName);
     const state = resolved.state, prompts = {};
     for (const role of ['positive', 'negative']) {
         const side = resolved[role];