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

__init__.py

@@ -23,24 +23,20 @@ Features:
 """
 
 import atexit
-import re
 import threading
 import time
 from pathlib import Path
 
 
 def get_version():
-    """Parse version from pyproject.toml"""
+    """Parse version from pyproject.toml (``utils/version.py``)."""
+    from .utils.version import project_version
+
     try:
-        pyproject_path = Path(__file__).parent / "pyproject.toml"
-        if pyproject_path.exists():
-            content = pyproject_path.read_text()
-            match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
-            if match:
-                return match.group(1)
+        root = Path(__file__).parent
     except Exception:
-        pass
-    return "unknown"
+        return "unknown"
+    return project_version(root)
 
 
 from .prompt_manager import (  # noqa: E402

py/api/__init__.py

@@ -5,6 +5,8 @@ Split into domain-specific mixins for maintainability:
 - ImageRoutesMixin: gallery, thumbnails, image serving and linking
 - AdminRoutesMixin: duplicates, stats, settings, diagnostics, maintenance, backup, scan
 - LoggingRoutesMixin: log management endpoints
+- LogCleanupMixin: the reviewed rotated-log cleanup (a maintenance plan)
+- DiagnosticsReportMixin: the shareable diagnostic report, without secrets or paths
 - AutotagRoutesMixin: auto-tagging model management and tagging
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs
@@ -65,6 +67,8 @@ from .duplicate_files import DuplicateFilesMixin
 from .maintenance_plans import PlanStore
 from .jobs import JobRegistry
 from .logging_routes import LoggingRoutesMixin
+from .log_cleanup import LogCleanupMixin
+from .diagnostics_report import DiagnosticsReportMixin
 from .autotag_routes import AutotagRoutesMixin
 from .lora_integration import LoraIntegrationMixin
 from .lora_import_routes import LoraImportRoutesMixin
@@ -323,6 +327,8 @@ class PromptManagerAPI(
     WorkflowStorageRoutesMixin,
     DuplicateFilesMixin,
     LoggingRoutesMixin,
+    LogCleanupMixin,
+    DiagnosticsReportMixin,
     AutotagRoutesMixin,
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
@@ -678,6 +684,7 @@ class PromptManagerAPI(
         self._register_settings_routes(routes)
         self._register_job_routes(routes)
         self._register_system_routes(routes)
+        self._register_diagnostics_report_routes(routes)
         self._register_library_location_routes(routes)
         self._register_maintenance_routes(routes)
         self._register_backup_restore_routes(routes)

py/api/admin.py

@@ -489,9 +489,29 @@ def _diagnose_dependencies():
     }
 
 
-def _diagnose_output_dirs(configured_dirs):
+def legacy_checks(db_path, public_path=None):
+    """The four checks of ``GET /diagnostics`` (blocking).
+
+    Shared with the diagnostic report (``py/api/diagnostics_report.py``),
+    which passes its own ``public_path`` so a folder outside ComfyUI is never
+    named.
+    """
+    from ..config import GalleryConfig
+
+    return {
+        "database": _diagnose_database(db_path),
+        "dependencies": _diagnose_dependencies(),
+        "comfyui_output": _diagnose_output_dirs(
+            GalleryConfig.MONITORING_DIRECTORIES, public_path
+        ),
+        "image_monitor": _diagnose_image_monitor(public_path),
+    }
+
+
+def _diagnose_output_dirs(configured_dirs, public_path=None):
     """Configured gallery roots, ComfyUI's output dir, or relative fallbacks."""
     _public_path, _ = _public_helpers()
+    _public_path = public_path or _public_path
     candidates = list(configured_dirs)
     try:
         import folder_paths
@@ -525,9 +545,10 @@ def _diagnose_output_dirs(configured_dirs):
     }
 
 
-def _diagnose_image_monitor():
+def _diagnose_image_monitor(public_path=None):
     """Status of the running image monitor, with public directory names."""
     _public_path, _public_error = _public_helpers()
+    _public_path = public_path or _public_path
     try:
         monitor = _current_image_monitor()
         if monitor is None:
@@ -1055,16 +1076,7 @@ class AdminRoutesMixin:
     async def run_diagnostics(self, request):
         """Run comprehensive system diagnostics and health checks."""
         try:
-            from ..config import GalleryConfig
-
-            results = {
-                "database": _diagnose_database(self.db.model.db_path),
-                "dependencies": _diagnose_dependencies(),
-                "comfyui_output": _diagnose_output_dirs(
-                    GalleryConfig.MONITORING_DIRECTORIES
-                ),
-                "image_monitor": _diagnose_image_monitor(),
-            }
+            results = legacy_checks(self.db.model.db_path)
             return web.json_response({"success": True, "diagnostics": results})
 
         except Exception as e:

py/api/diagnostics_report.py

@@ -0,0 +1,289 @@
+"""The shareable diagnostic report (#329 System S12).
+
+``GET /prompt_manager/system/diagnostics`` answers a report built on the
+server from an explicit list of fields: the project and Viewer.js versions,
+Python and the platform, the four checks of the legacy ``GET /diagnostics``
+(one helper, ``admin.legacy_checks``), the database's public info, source
+counts, the log configuration and statistics, the last health summary and
+the configuration keys that only ``config.json`` holds. It never copies
+``PromptManagerConfig.get_config()``: no path leaf is listed, and the CivitAI
+key is reported only as ``has_civitai_api_key``.
+
+Every path in the report goes through ``report_path``: relative to the
+closest public anchor that is not a filesystem root, else ``"outside
+ComfyUI"``, never a basename that could be a user name. As a backstop,
+``scrub`` replaces any string that still looks like an absolute path, holds
+the saved key or the home folder (with or without its root), and counts each
+replacement in ``redacted`` (0 for a correct builder).
+
+The report opens the database, so it takes a registry lease: it is refused
+(409 ``busy``) while an ``exclusive`` job (a restore) runs, and a restore
+waits for it. It is built off the event loop.
+"""
+
+import datetime
+import json
+import ntpath
+import os
+import platform
+import posixpath
+import re
+from pathlib import Path, PurePosixPath, PureWindowsPath
+
+from aiohttp import web
+
+from . import admin as _admin
+from .job_routes import busy_response
+from .jobs import JobBusy
+from .system import health_summary
+
+try:
+    from ...utils.version import project_version
+except ImportError:  # loaded as the top-level ``py`` package (tests)
+    from utils.version import project_version
+
+DIAGNOSTICS_LEASE = "Diagnostics"
+_ROOT = Path(__file__).resolve().parents[2]
+VIEWERJS_PROVENANCE = str(_ROOT / "web" / "lib" / "viewerjs" / "provenance.json")
+OUTSIDE = "outside ComfyUI"
+REMOVED = "[removed]"
+# The CivitAI key is searched for only when it is long enough to be a key.
+MIN_KEY_LENGTH = 8
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+
+
+def anchored_path(canonical, anchors, home, pathmod=os.path):
+    """``canonical`` relative to the closest usable anchor, else ``outside ComfyUI``.
+
+    All three are already canonical (``normcase``) in ``pathmod``'s spelling.
+    A filesystem root is never usable, nor is a strict ancestor of ``home``
+    (``/home``, ``C:\\Users``), whose relative paths would start with a user
+    name; so a path inside the home folder needs an anchor at or below it.
+    """
+    flavor = PureWindowsPath if pathmod is ntpath else PurePosixPath
+    target = flavor(canonical)
+    home_path = flavor(home) if home else None
+    usable = []
+    for anchor in map(flavor, anchors):
+        if anchor.parent == anchor:
+            continue
+        if home_path is not None and home_path != anchor:
+            if home_path.is_relative_to(anchor):
+                continue
+        if target.is_relative_to(anchor):
+            usable.append(anchor)
+    if not usable:
+        return OUTSIDE
+    anchor = max(usable, key=lambda a: len(a.parts))
+    return target.relative_to(anchor).as_posix() or "."
+
+
+def _canonical(path):
+    return os.path.normcase(os.path.realpath(str(path)))
+
+
+def report_path(path):
+    """``path`` relative to a public anchor (POSIX style), else ``outside ComfyUI``."""
+    if not path:
+        return ""
+    from ..config import GalleryConfig
+
+    try:
+        canonical = _canonical(path)
+        anchors = [_canonical(anchor) for anchor in GalleryConfig.path_anchors()]
+        home = os.path.expanduser("~")
+        home = _canonical(home) if home and home != "~" else None
+    except (OSError, ValueError):
+        return OUTSIDE
+    return anchored_path(canonical, anchors, home)
+
+
+def _looks_absolute(text):
+    return (
+        posixpath.isabs(text)
+        or ntpath.isabs(text)
+        or _WINDOWS_ABSOLUTE.match(text) is not None
+        or text.startswith("\\\\")
+    )
+
+
+# The home folder is matched only as whole path parts, never inside a word.
+_PART = r"(?<![\w.-]){}(?![\w.-])"
+MIN_HOME_NAME = 3
+
+
+def _home_patterns():
+    """``(pattern, paths_only)`` for the home folder: in full, rootless, its name.
+
+    Accepted residual risk: only this process's own home folder is known, so
+    another user's home (an output folder under /home/<them> while ComfyUI
+    runs as a service account) can still be named by a report path.
+    """
+    home = os.path.expanduser("~")
+    if not home or home == "~" or len(home) < 4 or home in ("/", "\\"):
+        return ()
+    spellings = {home, home.replace("\\", "/")}
+    relative = os.path.splitdrive(home)[1].strip("\\/")
+    for spelling in (relative, relative.replace("\\", "/")):
+        if "/" in spelling or "\\" in spelling:
+            spellings.add(spelling)
+    patterns = [
+        (re.compile(_PART.format(re.escape(spelling)), re.IGNORECASE), False)
+        for spelling in spellings
+    ]
+    name = re.split(r"[\\/]", relative)[-1] if relative else ""
+    if len(name) >= MIN_HOME_NAME:
+        # The bare name only inside a path ("info/x"), not a word ("INFO").
+        patterns.append(
+            (re.compile(_PART.format(re.escape(name)), re.IGNORECASE), True)
+        )
+    return tuple(patterns)
+
+
+def scrub(report, key):
+    """``report`` with leaked paths and the key replaced, counted in ``redacted``."""
+    key = key if isinstance(key, str) and len(key) >= MIN_KEY_LENGTH else None
+    homes = _home_patterns()
+    count = 0
+
+    def clean(value):
+        nonlocal count
+        if isinstance(value, str):
+            if (
+                _looks_absolute(value)
+                or (key and key in value)
+                or any(
+                    pattern.search(value)
+                    for pattern, paths_only in homes
+                    if not paths_only or "/" in value or "\\" in value
+                )
+            ):
+                count += 1
+                return REMOVED
+            return value
+        if isinstance(value, dict):
+            return {name: clean(item) for name, item in value.items()}
+        if isinstance(value, (list, tuple)):
+            return [clean(item) for item in value]
+        return value
+
+    cleaned = clean(report)
+    cleaned["redacted"] = int(cleaned.get("redacted") or 0) + count
+    return cleaned
+
+
+def viewerjs_version(path=None):
+    """The pinned Viewer.js version from its provenance file, or ``unknown``."""
+    try:
+        with open(path or VIEWERJS_PROVENANCE, encoding="utf-8") as handle:
+            version = json.load(handle).get("version")
+    except (OSError, ValueError, AttributeError):
+        return "unknown"
+    return version if isinstance(version, str) and version else "unknown"
+
+
+def public_config():
+    """The configuration keys only ``config.json`` holds, from an explicit list."""
+    from ..config import GalleryConfig, IntegrationConfig, PromptManagerConfig
+
+    return {
+        "database": {
+            "enable_duplicate_detection": (
+                PromptManagerConfig.ENABLE_DUPLICATE_DETECTION
+            ),
+            "enable_auto_save": PromptManagerConfig.ENABLE_AUTO_SAVE,
+        },
+        "web_ui": {
+            "result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
+            "show_test_button": PromptManagerConfig.SHOW_TEST_BUTTON,
+            "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
+            "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
+        },
+        "performance": {
+            "max_search_results": PromptManagerConfig.MAX_SEARCH_RESULTS,
+            "enable_fuzzy_search": PromptManagerConfig.ENABLE_FUZZY_SEARCH,
+            "auto_backup_interval": PromptManagerConfig.AUTO_BACKUP_INTERVAL,
+            "worker_threads": PromptManagerConfig.WORKER_THREADS,
+        },
+        "gallery": {
+            "processing_delay": GalleryConfig.PROCESSING_DELAY,
+            "prompt_timeout": GalleryConfig.PROMPT_TIMEOUT,
+            "cleanup_interval": GalleryConfig.CLEANUP_INTERVAL,
+            "auto_cleanup_missing_files": GalleryConfig.AUTO_CLEANUP_MISSING_FILES,
+            "max_image_age_days": GalleryConfig.MAX_IMAGE_AGE_DAYS,
+            "max_concurrent_processing": GalleryConfig.MAX_CONCURRENT_PROCESSING,
+            "metadata_extraction_timeout": GalleryConfig.METADATA_EXTRACTION_TIMEOUT,
+        },
+        "integrations": {
+            "lora_manager_enabled": bool(IntegrationConfig.LORA_MANAGER_ENABLED),
+            "has_civitai_api_key": bool(IntegrationConfig.CIVITAI_API_KEY),
+        },
+    }
+
+
+def _source_counts(sources):
+    configured = sources.get("configured") or []
+    effective = sources.get("effective") or []
+    return {
+        "configured": len(configured),
+        "effective": len(effective),
+        "watching": sum(1 for row in effective if row.get("watch") == "watching"),
+        "unavailable": sum(1 for row in configured if row.get("valid") is False),
+    }
+
+
+class DiagnosticsReportMixin:
+    """Mixin providing ``GET /prompt_manager/system/diagnostics``."""
+
+    def _register_diagnostics_report_routes(self, routes):
+        @routes.get("/prompt_manager/system/diagnostics")
+        async def diagnostics_report_route(request):
+            return await self.get_diagnostics_report(request)
+
+    async def get_diagnostics_report(self, request):
+        """The report, built off the loop under a lease (409 ``busy`` in a restore)."""
+        from ..config import IntegrationConfig
+
+        try:
+            async with self._jobs.lease(DIAGNOSTICS_LEASE):
+                report = await self._run_in_executor(self.build_report)
+        except JobBusy as busy:
+            return busy_response(busy)
+        except Exception:
+            self.logger.exception("The diagnostic report could not be built")
+            return web.json_response(
+                {
+                    "success": False,
+                    "code": None,
+                    "error": "The diagnostic report could not be built.",
+                },
+                status=500,
+            )
+        report = scrub(report, IntegrationConfig.CIVITAI_API_KEY)
+        return web.json_response(report, headers={"Cache-Control": "no-store"})
+
+    def build_report(self):
+        """The report's fields (blocking: reads the database and the log folder)."""
+        db = self.db
+        manager = self._get_logger_manager()
+        stats = dict(manager.get_log_stats())
+        stats["log_directory"] = report_path(stats.get("log_directory"))
+        info = dict(db.model.get_database_info() or {})
+        if "database_path" in info:
+            info["database_path"] = report_path(info["database_path"])
+        sources, _monitor = self._source_status()
+        return {
+            "success": True,
+            "generated_at": datetime.datetime.now().replace(microsecond=0).isoformat(),
+            "version": project_version(_ROOT),
+            "python": platform.python_version(),
+            "platform": {"system": platform.system(), "release": platform.release()},
+            "viewerjs": viewerjs_version(VIEWERJS_PROVENANCE),
+            "checks": _admin.legacy_checks(db.model.db_path, report_path),
+            "database_info": info,
+            "sources": _source_counts(sources),
+            "logging": {"config": dict(manager.get_config()), "stats": stats},
+            "health": health_summary(self._current_health_report()),
+            "config": public_config(),
+            "redacted": 0,
+        }

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": {

py/api/log_cleanup.py

@@ -0,0 +1,207 @@
+"""Reviewed rotated-log cleanup as a maintenance plan (#329 System S12).
+
+``POST /prompt_manager/maintenance/preview {operation: "rotated_logs",
+options: {}}`` lists the rotated copies of the active log that Rule R allows
+(``utils/logging_config.py``: ``prompt_manager.log.<n>`` regular files that
+are not links, directly in the log folder) and writes nothing. The plan keeps
+each copy's name, size, ``mtime_ns`` and inode under its token; the response
+is the maintenance ``Plan`` shape with basenames only, a ``none`` backup
+policy (no library record changes) and no ``disk`` key.
+
+``POST /prompt_manager/maintenance/apply {token}`` starts a ``maintenance``
+job titled "Rotated log copies" with admission ``read``: it may run beside a
+scan and is refused only beside another maintenance job or an ``exclusive``
+one. The job deletes exactly the reviewed copies under the logger's files
+lock and the file handler's lock, or nothing, failing ``plan-changed``, when
+any of them is missing, is no longer a rotated copy or differs in size,
+``mtime_ns`` or inode. A copy that cannot be deleted (open in another program
+on Windows) is an ``errors`` row; the others are deleted.
+"""
+
+import datetime
+import json
+
+from aiohttp import web
+
+try:
+    from ...utils.logging_config import RotatedChanged, rotated_sort_key
+except ImportError:  # loaded as the top-level ``py`` package (tests)
+    from utils.logging_config import RotatedChanged, rotated_sort_key
+
+from .jobs import MAINTENANCE_JOB_KIND
+from .maintenance_plans import ScopeTooLarge
+
+ROTATED_LOGS = "rotated_logs"
+LOG_CLEANUP_TITLE = "Rotated log copies"
+LOGS_CHANGED = (
+    "The log files changed since the review; nothing was deleted. Review again."
+)
+LOG_CLEANUP_FAILED = "The log cleanup failed."
+TOO_MANY_LOG_FILES = "Too many log files to review at once."
+LOG_CLEANUP_MEDIA = (
+    "Deletes only the listed log files. No image, video or library record changes."
+)
+_GROUPS = ("changed", "protected", "errors")
+
+
+def size_label(size):
+    """``1000 bytes``, ``2.0 KB``, ``10.0 MB`` or ``1.2 GB`` (1 KB = 1024 bytes)."""
+    size = max(0, int(size or 0))
+    if size < 1024:
+        return f"{size} {'byte' if size == 1 else 'bytes'}"
+    for unit, scale in (("KB", 2**10), ("MB", 2**20)):
+        if size < scale * 1024:
+            return f"{size / scale:.1f} {unit}"
+    return f"{size / 2**30:.1f} GB"
+
+
+def time_label(mtime_ns):
+    """The local modification time as ``YYYY-MM-DD HH:MM``."""
+    try:
+        moment = datetime.datetime.fromtimestamp(mtime_ns / 1e9)
+    except (OSError, OverflowError, ValueError):  # a time before 1970 on Windows
+        return "unknown time"
+    return moment.strftime("%Y-%m-%d %H:%M")
+
+
+def _count(count, singular, plural):
+    return f"{count} {singular if count == 1 else plural}"
+
+
+def _protected_reasons(rows):
+    """The plan's Protected reasons: what the cleanup keeps, zero counts left out."""
+    active = sum(1 for row in rows if row["kind"] == "active")
+    links = sum(1 for row in rows if row["kind"] == "link")
+    other = sum(1 for row in rows if row["kind"] == "other")
+    reasons = []
+    if active:
+        reasons.append("Kept: the active log")
+    if links:
+        reasons.append(f"Kept: {_count(links, 'link', 'links')}")
+    if other:
+        reasons.append(f"Kept: {_count(other, 'other file', 'other files')}")
+    return active + links + other, reasons
+
+
+def _scope_label(files):
+    if not files:
+        return "No rotated log copies"
+    total = sum(row["size"] for row in files)
+    return (
+        f"{_count(len(files), 'rotated log copy', 'rotated log copies')}"
+        f" · {size_label(total)}"
+    )
+
+
+class LogCleanupMixin:
+    """The ``rotated_logs`` maintenance plan and its job.
+
+    The logger manager is reached only through ``_get_logger_manager``
+    (``logging_routes.py``), so the tests can point it at a temporary folder.
+    """
+
+    async def preview_log_cleanup(self):
+        """The ``rotated_logs`` plan (runs inside the preview's lease)."""
+        from . import maintenance
+
+        manager = self._get_logger_manager()
+        try:
+            rows = await self._run_in_executor(manager.scan_log_folder)
+        except Exception:
+            self.logger.exception("Rotated log review failed")
+            return maintenance._failure("Failed to review maintenance", 500)
+        rotated = sorted(
+            (row for row in rows if row["kind"] == "rotated"),
+            key=rotated_sort_key,
+        )
+        files = [
+            {
+                "name": row["name"],
+                "size": row["size"],
+                "mtime_ns": row["mtime_ns"],
+                "ino": row["ino"],
+            }
+            for row in rotated
+        ]
+        protected, reasons = _protected_reasons(rows)
+        store = self._maintenance_plans
+        try:
+            token = store.issue(
+                ROTATED_LOGS,
+                json.dumps(files, sort_keys=True),
+                {"files": files, "protected": protected},
+                (),
+            )
+        except ScopeTooLarge:
+            return maintenance._failure(TOO_MANY_LOG_FILES, 413, "too-large")
+        plan = store.peek(token, ROTATED_LOGS)
+        return web.json_response(
+            {
+                "success": True,
+                "token": token,
+                "operation": ROTATED_LOGS,
+                "expires_at": maintenance._iso(plan.expires_at),
+                "scope": {"count": len(files), "label": _scope_label(files)},
+                "protected": {"count": protected, "reasons": reasons},
+                "backup": {
+                    "policy": "none",
+                    "destination_label": None,
+                    "expected_bytes": None,
+                },
+                "media": {"deletes_files": True, "label": LOG_CLEANUP_MEDIA},
+                "details": [
+                    {
+                        "label": row["name"],
+                        "detail": (
+                            f"{size_label(row['size'])} · {time_label(row['mtime_ns'])}"
+                        ),
+                    }
+                    for row in files
+                ],
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+
+    def _start_log_cleanup(self, plan):
+        """Register the cleanup job (on the loop); raises JobBusy when refused."""
+        return self._jobs.start(
+            MAINTENANCE_JOB_KIND,
+            LOG_CLEANUP_TITLE,
+            admission="read",
+            cancellable=False,
+            scope={"operation": ROTATED_LOGS},
+            groups=_GROUPS,
+            runner=lambda job: self._run_log_cleanup(job, plan),
+        )
+
+    async def _run_log_cleanup(self, job, plan):
+        from . import maintenance
+
+        files = plan.payload["files"]
+        sizes = {row["name"]: row["size"] for row in files}
+        job.update(total=len(files), status_line="Deleting rotated log copies")
+        manager = self._get_logger_manager()
+        try:
+            result = await self._run_in_executor(manager.delete_rotated, files)
+        except RotatedChanged:
+            return maintenance._fail_job(job, "plan-changed", LOGS_CHANGED)
+        except Exception:
+            self.logger.exception("Rotated log cleanup failed")
+            return maintenance._fail_job(job, None, LOG_CLEANUP_FAILED)
+        for name in result["removed"]:
+            detail = f"Deleted · {size_label(sizes.get(name, 0))}"
+            job.record(
+                "changed", {"path": name, "detail": detail}, f"- {name} · {detail}"
+            )
+        for name, error in result["failed"]:
+            detail = self._public_error(error)
+            job.record(
+                "errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
+            )
+        job.record_counted("protected", plan.payload["protected"])
+        removed, failed = len(result["removed"]), len(result["failed"])
+        line = f"Deleted {_count(removed, 'rotated log copy', 'rotated log copies')}"
+        if failed:
+            line += f", {failed} could not be deleted"
+        job.log_line(line)
+        job.update(processed=len(files), progress=100, status_line=line)

py/api/logging_routes.py

@@ -1,6 +1,13 @@
-"""Logging API routes for PromptManager."""
+"""Logging API routes for PromptManager.
+
+Every route that names a log file goes through Rule R (#329,
+``utils/logging_config.py``): a download is served only for a regular,
+non-link file listed directly in the log folder, and Clear deletes only
+rotated copies. File work runs off the event loop.
+"""
 
 import os
+import stat
 
 from aiohttp import web
 
@@ -85,6 +92,41 @@ def _validate_log_config(data):
     return clean, None
 
 
+def _read_log_bytes(path):
+    """The whole file at ``path``, opened without following a link (blocking).
+
+    The file is checked again through the open handle: a regular file with
+    one name and, when ``path`` carries it, the inode it was listed with. A
+    name swapped for a link or another file after it was listed, or a hard
+    link to a file elsewhere, is refused as missing. The handle is closed
+    before this returns.
+    """
+    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    # A FIFO swapped in after the listing must not block the executor thread.
+    flags |= getattr(os, "O_NONBLOCK", 0)
+    try:
+        fd = os.open(str(path), flags)
+    except OSError as error:
+        if getattr(os, "O_NOFOLLOW", 0) and os.path.islink(str(path)):
+            raise FileNotFoundError("Log file not found") from error
+        raise
+    listed_ino = getattr(path, "listed_ino", 0)
+    with os.fdopen(fd, "rb") as handle:
+        opened = os.fstat(handle.fileno())
+        if (
+            not stat.S_ISREG(opened.st_mode)
+            or opened.st_nlink > 1  # a hard link to a file elsewhere
+            or (listed_ino and opened.st_ino and opened.st_ino != listed_ino)
+        ):
+            raise FileNotFoundError("Log file not found")
+        return handle.read()
+
+
+def _download(logger_manager, filename):
+    """``bytes`` of a downloadable log file; FileNotFoundError otherwise."""
+    return _read_log_bytes(logger_manager.log_file_entry(filename))
+
+
 class LoggingRoutesMixin:
     """Mixin providing logging-related API endpoints."""
 
@@ -195,16 +237,16 @@ class LoggingRoutesMixin:
                     {"success": False, "error": "Invalid filename"}, status=400
                 )
 
-            log_file_path = logger_manager.log_dir / filename
-
-            if not log_file_path.exists():
+            try:
+                file_content = await self._run_in_executor(
+                    _download, logger_manager, filename
+                )
+            except FileNotFoundError:
+                # Missing, a link, a folder or not listed: all read as absent.
                 return web.json_response(
                     {"success": False, "error": "Log file not found"}, status=404
                 )
 
-            with open(log_file_path, "rb") as f:
-                file_content = f.read()
-
             return web.Response(
                 body=file_content,
                 content_type="text/plain",
@@ -224,7 +266,7 @@ class LoggingRoutesMixin:
         """Truncate all log files."""
         try:
             logger_manager = self._get_logger_manager()
-            results = logger_manager.truncate_logs()
+            results = await self._run_in_executor(logger_manager.truncate_logs)
 
             return web.json_response(
                 {
@@ -266,7 +308,7 @@ class LoggingRoutesMixin:
                 return web.json_response({"success": False, "error": error}, status=400)
 
             logger_manager = self._get_logger_manager()
-            logger_manager.update_config(clean)
+            await self._run_in_executor(logger_manager.update_config, clean)
 
             return web.json_response(
                 {

py/api/maintenance.py

@@ -2,7 +2,9 @@
 
 ``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
 one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
-``missing_links``; ``database/maintenance.py``) off the event loop and
+``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
+``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
+``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
 answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
 label}, protected:{count, reasons}, backup:{policy, destination_label},
 media:{deletes_files, label}, details:[≤50 {label, detail}]}``. The preview
@@ -54,6 +56,7 @@ from aiohttp import web
 from .admin import _prune_safety_backups, _scan_location, _scan_roots
 from .job_routes import busy_response
 from .jobs import MAINTENANCE_JOB_KIND, JobBusy
+from .log_cleanup import ROTATED_LOGS
 from .maintenance_plans import PlanExpired, PlanUsed, ScopeTooLarge
 
 try:
@@ -83,7 +86,8 @@ MAINTENANCE_PREVIEW_LEASE = "Maintenance preview"
 DUPLICATE_PREVIEW_BODY_MAX = 1024 * 1024
 DUPLICATE_FILES = "duplicate_files"
 WORKFLOW_STORAGE = "workflow_storage"  # applied by py/api/workflow_storage.py
-PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES,)
+# rotated_logs (#329, py/api/log_cleanup.py) reviews log files, not records.
+PREVIEW_OPERATIONS = MAINTENANCE_OPERATIONS + (DUPLICATE_FILES, ROTATED_LOGS)
 # Room left over beyond what a step writes, so a disk is never filled to 0.
 DISK_MARGIN_BYTES = 16 * 2**20
 # The concept's Maintenance row titles, which name each operation's job.
@@ -391,6 +395,8 @@ class MaintenanceRoutesMixin:
             return await self.preview_duplicate_files(options)
         if not isinstance(options, dict) or options:
             return _invalid(f"{operation} takes no options")
+        if operation == ROTATED_LOGS:
+            return await self.preview_log_cleanup()
         db = self.db
         identity = self._database_identity(db)
         try:
@@ -492,6 +498,17 @@ class MaintenanceRoutesMixin:
         if refused is not None:
             return refused
         store = self._maintenance_plans
+        if plan.operation == ROTATED_LOGS:
+            # Log files only: no database identity, backup or disk check.
+            try:
+                job = self._start_log_cleanup(plan)
+            except JobBusy as busy:
+                return busy_response(busy)  # the token stays usable
+            # Same loop turn as the peek and the start: the runner has not run.
+            store.consume(token, plan.operation, job_id=job.id)
+            return web.json_response(
+                {"success": True, "job": job.summary()}, status=202
+            )
         db = self.db
         if plan.db_identity != self._database_identity(db):
             store.discard(token)

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_diagnostics_report.py

@@ -0,0 +1,409 @@
+"""The shareable diagnostic report (#329, py/api/diagnostics_report.py).
+
+``GET /prompt_manager/system/diagnostics`` builds the report on the server
+from an explicit list of fields: the project version, the four legacy checks,
+public paths only and no secret. A scrub replaces anything that still looks
+like an absolute path or holds the CivitAI key and counts it in ``redacted``;
+these tests require ``redacted == 0`` so a builder mistake fails instead of
+being hidden.
+"""
+
+import json
+import ntpath
+import os
+import posixpath
+import re
+import sqlite3
+import sys
+import unittest
+from unittest import mock
+from unittest.mock import MagicMock
+
+_mock_server = MagicMock()
+_mock_server.PromptServer.instance.routes = MagicMock()
+sys.modules.setdefault("server", _mock_server)
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from py.api import admin, diagnostics_report  # noqa: E402
+from py.api.diagnostics_report import OUTSIDE, report_path, scrub  # noqa: E402
+from py.config import (
+    GalleryConfig,
+    IntegrationConfig,
+    PromptManagerConfig,
+)  # noqa: E402
+
+try:
+    from tests.test_api_log_cleanup import LogCleanupTestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_log_cleanup import LogCleanupTestCase  # noqa: E402
+
+REPORT = "/prompt_manager/system/diagnostics"
+KEY = "civitai-secret-0123456789abcdef"
+SEEDED_PATHS = (
+    "/home/someone/private/LoraManager",
+    "/srv/someone/renders",
+    "C:\\Users\\someone\\Documents\\renders",
+    "D:/Users/someone/prompts.db",
+    "\\\\nas\\share\\someone\\renders",
+)
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+TOP_KEYS = {
+    "success",
+    "generated_at",
+    "version",
+    "python",
+    "platform",
+    "viewerjs",
+    "checks",
+    "database_info",
+    "sources",
+    "logging",
+    "health",
+    "config",
+    "redacted",
+}
+CONFIG = {
+    "database": {"enable_duplicate_detection", "enable_auto_save"},
+    "web_ui": {
+        "result_timeout",
+        "show_test_button",
+        "webui_display_mode",
+        "infinite_scroll",
+    },
+    "performance": {
+        "max_search_results",
+        "enable_fuzzy_search",
+        "auto_backup_interval",
+        "worker_threads",
+    },
+    "gallery": {
+        "processing_delay",
+        "prompt_timeout",
+        "cleanup_interval",
+        "auto_cleanup_missing_files",
+        "max_image_age_days",
+        "max_concurrent_processing",
+        "metadata_extraction_timeout",
+    },
+    "integrations": {"lora_manager_enabled", "has_civitai_api_key"},
+}
+
+
+def _strings(value):
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for key, item in value.items():
+            yield str(key)
+            yield from _strings(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            yield from _strings(item)
+
+
+class DiagnosticsReportTestCase(LogCleanupTestCase):
+    """The admin app with its logger manager on a temporary log folder."""
+
+    async def get_application(self):
+        app = await super().get_application()
+        saved = IntegrationConfig.get_config()
+        self.addCleanup(IntegrationConfig.update_config, saved)
+        return app
+
+    async def report(self, status=200):
+        response = await self.client.request("GET", REPORT)
+        text = await response.text()
+        self.assertEqual(response.status, status, text)
+        return json.loads(text), text
+
+    def seed_secrets(self):
+        IntegrationConfig.update_config(
+            {
+                "lora_manager": {
+                    "enabled": True,
+                    "path": SEEDED_PATHS[0],
+                    "civitai_api_key": KEY,
+                }
+            }
+        )
+        GalleryConfig.MONITORING_DIRECTORIES = list(SEEDED_PATHS[1:3]) + [
+            SEEDED_PATHS[4]
+        ]
+        PromptManagerConfig.update_config(
+            {"database": {"default_path": SEEDED_PATHS[3]}}
+        )
+
+    def assert_no_paths_or_key(self, report, text):
+        self.assertNotIn(KEY, text)
+        self.assertNotIn(self.tmpdir, text)
+        self.assertNotIn(self.tmpdir.replace("\\", "/"), text)
+        for seeded in SEEDED_PATHS:
+            self.assertNotIn(seeded, text)
+            self.assertNotIn(json.dumps(seeded)[1:-1], text)
+        self.assertNotIn("someone", text)
+        for value in _strings(report):
+            self.assertFalse(posixpath.isabs(value), value)
+            self.assertFalse(ntpath.isabs(value), value)
+            self.assertIsNone(_WINDOWS_ABSOLUTE.match(value), value)
+            self.assertFalse(value.startswith("\\\\"), value)
+
+
+class TestReport(DiagnosticsReportTestCase):
+    async def test_report_has_version_viewerjs_and_the_four_checks(self):
+        report, _text = await self.report()
+        self.assertEqual(set(report), TOP_KEYS)
+        self.assertTrue(report["success"])
+        with open(
+            os.path.join(os.path.dirname(os.path.dirname(__file__)), "pyproject.toml"),
+            encoding="utf-8",
+        ) as handle:
+            version = re.search(r'version\s*=\s*["\']([^"\']+)["\']', handle.read())
+        self.assertEqual(report["version"], version.group(1))
+        self.assertRegex(report["viewerjs"], r"^\d+\.\d+\.\d+$")
+        self.assertRegex(report["python"], r"^3\.\d+\.\d+")
+        self.assertEqual(set(report["platform"]), {"system", "release"})
+        self.assertEqual(
+            set(report["checks"]),
+            {"database", "dependencies", "comfyui_output", "image_monitor"},
+        )
+        self.assertEqual(report["checks"]["database"]["status"], "ok")
+        self.assertEqual(
+            set(report["sources"]),
+            {"configured", "effective", "watching", "unavailable"},
+        )
+        self.assertEqual(set(report["logging"]), {"config", "stats"})
+        self.assertIsNone(report["health"])
+        self.assertEqual(report["redacted"], 0)
+        self.assertRegex(report["generated_at"], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d$")
+
+    async def test_report_has_no_secret_or_absolute_path(self):
+        self.seed_secrets()
+        with mock.patch.object(
+            admin.sqlite3,
+            "connect",
+            side_effect=sqlite3.DatabaseError("file is not a database"),
+        ):
+            report, text = await self.report()
+        self.assertEqual(report["checks"]["database"]["status"], "error")
+        self.assertTrue(report["config"]["integrations"]["has_civitai_api_key"])
+        self.assertTrue(report["config"]["integrations"]["lora_manager_enabled"])
+        self.assertEqual(report["sources"]["configured"], 3)
+        self.assert_no_paths_or_key(report, text)
+        self.assertEqual(report["redacted"], 0)
+
+    async def test_report_leaves_get_config_unchanged(self):
+        self.seed_secrets()
+        before = json.dumps(PromptManagerConfig.get_config(), sort_keys=True)
+        directories = GalleryConfig.MONITORING_DIRECTORIES
+        await self.report()
+        self.assertEqual(
+            json.dumps(PromptManagerConfig.get_config(), sort_keys=True), before
+        )
+        self.assertIs(GalleryConfig.MONITORING_DIRECTORIES, directories)
+        self.assertEqual(IntegrationConfig.CIVITAI_API_KEY, KEY)
+
+    def test_outside_path_reads_outside_comfyui(self):
+        inside = os.path.join(str(self.comfy_dir), "user", "__promptmanager", "logs")
+        self.assertEqual(report_path(inside), "user/__promptmanager/logs")
+        self.assertEqual(
+            report_path(os.path.join(str(self.outside_dir), "vito")), "outside ComfyUI"
+        )
+        self.assertEqual(report_path(""), "")
+        self.assertEqual(report_path(None), "")
+
+    def test_a_filesystem_root_anchor_never_makes_a_path_public(self):
+        # --output-directory /renders makes "/" an anchor; a path outside
+        # ComfyUI must still read "outside ComfyUI", never "home/<user>/...".
+        root = os.path.abspath(os.sep)
+        anchors = [str(self.comfy_dir), root]
+        with mock.patch.object(GalleryConfig, "path_anchors", return_value=anchors):
+            self.assertEqual(
+                report_path(os.path.join(str(self.outside_dir), "x")), OUTSIDE
+            )
+            self.assertEqual(
+                report_path(os.path.join(str(self.comfy_dir), "output")), "output"
+            )
+
+    def test_scrub_removes_the_home_folder_without_its_root(self):
+        home = os.path.expanduser("~")
+        relative = os.path.splitdrive(home)[1].lstrip("\\/").replace("\\", "/")
+        cleaned = scrub({"a": f"{relative}/area/prompts.db", "redacted": 0}, None)
+        self.assertEqual(cleaned, {"a": "[removed]", "redacted": 1})
+
+    def test_the_home_folder_is_matched_as_a_whole_path_part(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/root"
+        ):
+            cleaned = scrub(
+                {
+                    "a": "output/rootfiles",
+                    "b": "output/root/x",
+                    "c": "see /root/x",
+                    "redacted": 0,
+                },
+                None,
+            )
+        self.assertEqual(cleaned["a"], "output/rootfiles")
+        # The home folder's name is a whole path part here: removed.
+        self.assertEqual(cleaned["b"], "[removed]")
+        self.assertEqual(cleaned["c"], "[removed]")
+        self.assertEqual(cleaned["redacted"], 2)
+
+    async def test_report_reads_only_the_temporary_log_folder(self):
+        self.write("prompt_manager.log.1", b"x" * 1234)
+        report, _text = await self.report()
+        stats = report["logging"]["stats"]
+        self.assertEqual(stats["log_files_count"], 2)
+        self.assertEqual(
+            stats["total_log_size"],
+            1234 + os.path.getsize(self.path("prompt_manager.log")),
+        )
+
+    def test_an_ancestor_of_home_is_never_an_anchor_on_linux(self):
+        anchored = diagnostics_report.anchored_path
+        home = "/home/someone"
+        anchors = ["/home", "/opt/ComfyUI"]
+        self.assertEqual(
+            anchored("/home/someone/renders/a.png", anchors, home, posixpath), OUTSIDE
+        )
+        self.assertEqual(
+            anchored("/home/other/renders", anchors, home, posixpath), OUTSIDE
+        )
+        self.assertEqual(
+            anchored("/opt/ComfyUI/output", anchors, home, posixpath), "output"
+        )
+
+    def test_an_ancestor_of_home_is_never_an_anchor_on_windows(self):
+        anchored = diagnostics_report.anchored_path
+        home = ntpath.normcase("C:\\Users\\someone")
+        anchors = [ntpath.normcase(a) for a in ("C:\\Users", "C:\\", "D:\\ComfyUI")]
+        path = ntpath.normcase("C:\\Users\\someone\\renders\\a.png")
+        self.assertEqual(anchored(path, anchors, home, ntpath), OUTSIDE)
+        inside = ntpath.normcase("D:\\ComfyUI\\output\\x")
+        self.assertEqual(anchored(inside, anchors, home, ntpath), "output/x")
+
+    def test_an_anchor_that_is_home_names_no_user(self):
+        anchored = diagnostics_report.anchored_path
+        home = "/home/someone"
+        self.assertEqual(
+            anchored("/home/someone/ComfyUI/output", [home], home, posixpath),
+            "ComfyUI/output",
+        )
+        self.assertEqual(
+            anchored(
+                "/home/someone/ComfyUI/output",
+                [home, "/home/someone/ComfyUI"],
+                home,
+                posixpath,
+            ),
+            "output",
+        )
+
+    def test_the_home_folder_name_is_removed_only_from_paths(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/home/info"
+        ):
+            cleaned = scrub({"level": "INFO", "p": "logs/info/x", "redacted": 0}, None)
+        self.assertEqual(cleaned, {"level": "INFO", "p": "[removed]", "redacted": 1})
+
+    def test_scrub_removes_the_home_folder_name(self):
+        with mock.patch.object(
+            diagnostics_report.os.path, "expanduser", return_value="/home/someone"
+        ):
+            cleaned = scrub(
+                {"a": "someone/renders", "b": "someones", "redacted": 0}, None
+            )
+        self.assertEqual(cleaned["a"], "[removed]")
+        self.assertEqual(cleaned["b"], "someones")
+
+    def test_scrub_removes_a_leaked_path_and_counts_it(self):
+        leaked = {
+            "a": "/home/someone/x",
+            "b": ["C:\\Users\\someone", "D:/data", "\\\\nas\\share", "ok"],
+            "c": {"d": f"token {KEY} here", "e": 3, "f": None, "g": "relative/path"},
+            "redacted": 0,
+        }
+        cleaned = scrub(leaked, KEY)
+        self.assertEqual(cleaned["redacted"], 5)
+        self.assertEqual(cleaned["a"], "[removed]")
+        self.assertEqual(cleaned["b"], ["[removed]"] * 3 + ["ok"])
+        self.assertEqual(
+            cleaned["c"], {"d": "[removed]", "e": 3, "f": None, "g": "relative/path"}
+        )
+        # A short key is not searched for (it would match ordinary words).
+        self.assertEqual(scrub({"x": "abc", "redacted": 0}, "abc")["redacted"], 0)
+
+    async def test_report_refused_while_exclusive_job_runs(self):
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ) as restore:
+            report, _text = await self.report(status=409)
+        self.assertEqual(report["code"], "busy")
+        self.assertEqual(
+            report["running"], {"id": restore.id, "title": "Database restore"}
+        )
+
+    async def test_report_lists_config_only_keys(self):
+        self.seed_secrets()
+        report, _text = await self.report()
+        self.assertEqual({k: set(v) for k, v in report["config"].items()}, CONFIG)
+        self.assertNotIn("default_path", json.dumps(report["config"]))
+        self.assertNotIn("directories", json.dumps(report["config"]))
+        self.assertNotIn(
+            "civitai_api_key",
+            json.dumps(report["config"]).replace("has_civitai_api_key", ""),
+        )
+
+    async def test_report_logging_has_public_directory(self):
+        report, _text = await self.report()
+        stats = report["logging"]["stats"]
+        self.assertFalse(posixpath.isabs(stats["log_directory"]))
+        self.assertIn("level", report["logging"]["config"])
+
+    async def test_report_survives_an_unreadable_viewerjs_provenance(self):
+        with mock.patch.object(
+            diagnostics_report,
+            "VIEWERJS_PROVENANCE",
+            os.path.join(self.tmpdir, "nope.json"),
+        ):
+            report, _text = await self.report()
+        self.assertEqual(report["viewerjs"], "unknown")
+
+    async def test_a_build_failure_is_coded_without_details(self):
+        with mock.patch.object(
+            self.api, "build_report", side_effect=RuntimeError(self.tmpdir)
+        ):
+            with self.assertLogs("prompt_manager", "ERROR"):
+                report, text = await self.report(status=500)
+        self.assertNotIn(self.tmpdir, text)
+        self.assertFalse(report["success"])
+
+
+class TestLegacyDiagnostics(DiagnosticsReportTestCase):
+    async def test_legacy_diagnostics_keeps_its_four_checks(self):
+        response = await self.client.request("GET", "/prompt_manager/diagnostics")
+        data = await response.json()
+        self.assertEqual(response.status, 200)
+        self.assertEqual(set(data), {"success", "diagnostics"})
+        self.assertEqual(
+            set(data["diagnostics"]),
+            {"database", "dependencies", "comfyui_output", "image_monitor"},
+        )
+        report, _text = await self.report()
+        self.assertEqual(report["checks"]["database"], data["diagnostics"]["database"])
+        self.assertEqual(
+            report["checks"]["dependencies"], data["diagnostics"]["dependencies"]
+        )
+
+    async def test_both_routes_build_their_checks_with_one_helper(self):
+        with mock.patch.object(
+            admin, "legacy_checks", wraps=admin.legacy_checks
+        ) as spy:
+            await self.client.request("GET", "/prompt_manager/diagnostics")
+            await self.report()
+        self.assertEqual(spy.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_api_log_cleanup.py

@@ -0,0 +1,419 @@
+"""Reviewed rotated-log cleanup as a maintenance plan (#329, py/api/log_cleanup.py).
+
+``POST /maintenance/preview {operation: "rotated_logs"}`` lists exactly the
+rotated copies Rule R allows and writes nothing; ``POST /maintenance/apply``
+deletes exactly those as a ``maintenance`` job (admission ``read``), or
+nothing, failing ``plan-changed``, when any of them changed after the review.
+
+Every test runs on a real ``PromptManagerLogger`` whose log folder is a
+temporary folder, never the real one.
+"""
+
+import asyncio
+import json
+import logging
+import ntpath
+import os
+import posixpath
+import re
+import sys
+import threading
+import unittest
+import unittest.mock as mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from py.api.maintenance_plans import PlanStore  # noqa: E402
+from utils import logging_config  # noqa: E402
+from utils.logging_config import PromptManagerLogger, get_logger_manager  # noqa: E402
+
+try:
+    from tests.open_handles import assert_closed  # noqa: E402
+    from tests.test_api_maintenance import MaintenanceAPITestCase  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed  # noqa: E402
+    from test_api_maintenance import MaintenanceAPITestCase  # noqa: E402
+
+WAIT = 10
+PREVIEW = "/prompt_manager/maintenance/preview"
+APPLY = "/prompt_manager/maintenance/apply"
+CHANGED = "The log files changed since the review; nothing was deleted. Review again."
+MEDIA = "Deletes only the listed log files. No image, video or library record changes."
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+
+
+def _strings(value):
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for item in value.values():
+            yield from _strings(item)
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            yield from _strings(item)
+
+
+class LogCleanupTestCase(MaintenanceAPITestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        package = os.path.join(self.tmpdir, "pkg")
+        os.makedirs(os.path.join(package, "utils"))
+        fake_file = os.path.join(package, "utils", "logging_config.py")
+        with mock.patch.object(logging_config, "__file__", fake_file):
+            self.manager = object.__new__(PromptManagerLogger)  # bypass singleton
+            self.manager.__init__()
+        self.addCleanup(self._restore_global_logging)
+        self.manager.update_config({"console_logging": False})
+        self._only_this_thread()
+        self.log_dir = str(self.manager.log_dir)
+        self.app_logger = logging.getLogger("prompt_manager.log_cleanup_test")
+        self.app_logger.info("active log line")
+        self.api._get_logger_manager = lambda: self.manager
+        self.outside = os.path.join(self.tmpdir, "link-targets")
+        os.makedirs(self.outside)
+        return app
+
+    def _only_this_thread(self):
+        """Only this test's records reach the temporary log (not other threads')."""
+        this_thread = threading.get_ident()
+        for handler in self.manager.logger.handlers:
+            handler.filters.clear()
+            handler.addFilter(lambda record: record.thread == this_thread)
+
+    def _restore_global_logging(self):
+        for handler in logging.getLogger("prompt_manager").handlers[:]:
+            handler.close()
+        assert_closed(self, self.log_dir)
+        get_logger_manager()._setup_loggers()
+
+    def path(self, name):
+        return os.path.join(self.log_dir, name)
+
+    def write(self, name, data=b"rotated\n"):
+        with open(self.path(name), "wb") as handle:
+            handle.write(data)
+
+    def symlink(self, name):
+        target = os.path.join(self.outside, "secret.txt")
+        with open(target, "wb") as handle:
+            handle.write(b"secret " * 100)
+        try:
+            os.symlink(target, self.path(name))
+        except (OSError, NotImplementedError):  # pragma: no cover - no privilege
+            return False
+        return True
+
+    def names(self):
+        return sorted(os.listdir(self.log_dir))
+
+    def snapshot(self):
+        """Every entry but the active log, with its bytes (None for a folder)."""
+        state = {}
+        for name in self.names():
+            if name == "prompt_manager.log":
+                continue
+            path = self.path(name)
+            if os.path.islink(path):
+                state[name] = ("link", os.readlink(path))
+            elif os.path.isdir(path):
+                state[name] = None
+            else:
+                with open(path, "rb") as handle:
+                    state[name] = handle.read()
+        return state
+
+    def seed(self):
+        """CUJ 5's folder: three rotated copies, a .bak, a folder and a link."""
+        self.write("prompt_manager.log.1", b"a" * 1000)
+        self.write("prompt_manager.log.2", b"b" * 2000)
+        self.write("prompt_manager.log.3", b"c" * 3000)
+        self.write("prompt_manager.log.bak", b"mine")
+        os.makedirs(self.path("prompt_manager.log.9"))
+        return self.symlink("prompt_manager.log.4")
+
+    async def preview(self, status=200, **body):
+        return await self.post(
+            PREVIEW, {"operation": "rotated_logs", "options": {}, **body}, status
+        )
+
+    async def apply(self, token, status=202):
+        return await self.post(APPLY, {"token": token}, status)
+
+    async def run_apply(self, plan):
+        body = await self.apply(plan["token"])
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        return body["job"], job
+
+
+class TestPreview(LogCleanupTestCase):
+    async def test_preview_lists_only_rotated_copies(self):
+        linked = self.seed()
+        plan = await self.preview()
+        self.assertEqual(
+            set(plan),
+            {
+                "success",
+                "token",
+                "operation",
+                "expires_at",
+                "scope",
+                "protected",
+                "backup",
+                "media",
+                "details",
+            },
+        )
+        self.assertEqual(plan["operation"], "rotated_logs")
+        self.assertEqual(
+            plan["scope"], {"count": 3, "label": "3 rotated log copies · 5.9 KB"}
+        )
+        self.assertEqual(
+            [row["label"] for row in plan["details"]],
+            ["prompt_manager.log.1", "prompt_manager.log.2", "prompt_manager.log.3"],
+        )
+        self.assertRegex(
+            plan["details"][2]["detail"], r"^2\.9 KB · \d{4}-\d\d-\d\d \d\d:\d\d$"
+        )
+        reasons = ["Kept: the active log"]
+        if linked:
+            reasons.append("Kept: 1 link")
+        reasons.append("Kept: 2 other files")
+        self.assertEqual(
+            plan["protected"], {"count": len(reasons) + 1, "reasons": reasons}
+        )
+        self.assertEqual(
+            plan["backup"],
+            {"policy": "none", "destination_label": None, "expected_bytes": None},
+        )
+        self.assertEqual(plan["media"], {"deletes_files": True, "label": MEDIA})
+        self.assertTrue(plan["expires_at"].endswith("+00:00"))
+        self.assertNotIn("mtime_ns", json.dumps(plan))
+
+    async def test_preview_writes_nothing(self):
+        self.seed()
+        before = self.snapshot()
+        await self.preview()
+        self.assertEqual(self.snapshot(), before)
+
+    async def test_preview_has_no_paths(self):
+        self.seed()
+        plan = await self.preview()
+        for text in _strings(plan):
+            self.assertFalse(posixpath.isabs(text), text)
+            self.assertFalse(ntpath.isabs(text), text)
+            self.assertIsNone(_WINDOWS_ABSOLUTE.match(text), text)
+
+    async def test_preview_rejects_non_empty_options(self):
+        for options in ({"all": True}, [], "x"):
+            with self.subTest(options=options):
+                reply = await self.preview(status=400, options=options)
+                self.assertEqual(reply["code"], "invalid-request")
+
+    async def test_preview_too_large_is_413(self):
+        self.seed()
+        self.api._maintenance_plans = PlanStore(max_bytes=10)
+        reply = await self.preview(status=413)
+        self.assertEqual(
+            reply,
+            {
+                "success": False,
+                "code": "too-large",
+                "error": "Too many log files to review at once.",
+            },
+        )
+
+    async def test_empty_folder_plan_has_scope_zero(self):
+        plan = await self.preview()
+        self.assertEqual(plan["scope"], {"count": 0, "label": "No rotated log copies"})
+        self.assertEqual(plan["details"], [])
+        self.assertEqual(
+            plan["protected"], {"count": 1, "reasons": ["Kept: the active log"]}
+        )
+
+    async def test_preview_refused_while_an_exclusive_job_runs(self):
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ):
+            reply = await self.preview(status=409)
+        self.assertEqual(reply["code"], "busy")
+
+
+class TestApply(LogCleanupTestCase):
+    async def test_apply_deletes_exactly_the_reviewed_files(self):
+        linked = self.seed()
+        plan = await self.preview()
+        summary, job = await self.run_apply(plan)
+        self.assertEqual(summary["kind"], "maintenance")
+        self.assertEqual(summary["title"], "Rotated log copies")
+        self.assertEqual(summary["admission"], "read")
+        self.assertFalse(summary["cancellable"])
+        self.assertEqual(summary["scope"], {"operation": "rotated_logs"})
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertEqual(
+            job.counters,
+            {"changed": 3, "protected": plan["protected"]["count"], "errors": 0},
+        )
+        expected = [
+            "prompt_manager.log",
+            "prompt_manager.log.9",
+            "prompt_manager.log.bak",
+        ]
+        if linked:
+            expected.append("prompt_manager.log.4")
+        self.assertEqual(self.names(), sorted(expected))
+        self.assertEqual(job.status_line, "Deleted 3 rotated log copies")
+        report = json.dumps(job.report())
+        self.assertNotIn(self.tmpdir, report)
+
+    async def test_apply_rows_name_each_file(self):
+        self.seed()
+        plan = await self.preview()
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(
+            job.outcomes["changed"],
+            [
+                {
+                    "path": "prompt_manager.log.1",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 1000 bytes",
+                },
+                {
+                    "path": "prompt_manager.log.2",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 2.0 KB",
+                },
+                {
+                    "path": "prompt_manager.log.3",
+                    "root": "",
+                    "root_index": None,
+                    "detail": "Deleted · 2.9 KB",
+                },
+            ],
+        )
+        log = "\n".join(job.log)
+        for n in (1, 2, 3):
+            self.assertIn(f"prompt_manager.log.{n}", log)
+
+    async def test_apply_after_rollover_fails_plan_changed_and_deletes_nothing(self):
+        self.seed()
+        plan = await self.preview()
+        handler = next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+        handler.acquire()
+        try:
+            handler.doRollover()  # every listed name now holds another file
+        finally:
+            handler.release()
+        before = self.snapshot()
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.scope["failure"], "plan-changed")
+        self.assertEqual(job.status_line, CHANGED)
+        self.assertEqual(job.counters["changed"], 0)
+        self.assertEqual(self.snapshot(), before)
+
+    async def test_apply_reports_a_file_in_use(self):
+        self.seed()
+        plan = await self.preview()
+        real_unlink = os.unlink
+        held = self.path("prompt_manager.log.2")
+
+        def unlink(path, *args, **kwargs):
+            if os.fspath(path) == held:
+                error = PermissionError(
+                    13,
+                    "The process cannot access the file because it is being "
+                    "used by another process",
+                    held,
+                )
+                error.winerror = 32
+                raise error
+            return real_unlink(path, *args, **kwargs)
+
+        with mock.patch.object(logging_config.os, "unlink", unlink):
+            _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.counters["changed"], 2)
+        self.assertEqual(job.counters["errors"], 1)
+        [row] = job.outcomes["errors"]
+        self.assertEqual(row["path"], "prompt_manager.log.2")
+        self.assertIn("being used by another process", row["detail"])
+        self.assertNotIn(self.tmpdir, json.dumps(job.report()))
+        self.assertIn("prompt_manager.log.2", self.names())
+        self.assertEqual(
+            job.status_line, "Deleted 2 rotated log copies, 1 could not be deleted"
+        )
+
+    async def test_token_is_single_use(self):
+        self.seed()
+        plan = await self.preview()
+        summary, _job = await self.run_apply(plan)
+        again = await self.apply(plan["token"], 409)
+        self.assertEqual(again["code"], "plan-used")
+        self.assertEqual(
+            again["job"], {"id": summary["id"], "title": "Rotated log copies"}
+        )
+
+    async def test_apply_runs_beside_a_scan(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold("scan", "Metadata scan", admission="records"):
+            _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertNotIn("prompt_manager.log.1", self.names())
+
+    async def test_apply_refused_while_restore_runs(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold(
+            "restore", "Database restore", admission="exclusive"
+        ) as restore:
+            busy = await self.apply(plan["token"], 409)
+        self.assertEqual(busy["code"], "busy")
+        self.assertEqual(
+            busy["running"], {"id": restore.id, "title": "Database restore"}
+        )
+        self.assertIn("prompt_manager.log.1", self.names())
+        # The refused apply kept the token.
+        _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "finished", job.log)
+
+    async def test_apply_refused_beside_another_maintenance_job(self):
+        self.seed()
+        plan = await self.preview()
+        async with self.api._jobs.hold(
+            "maintenance", "Optimize database", admission="records"
+        ):
+            busy = await self.apply(plan["token"], 409)
+        self.assertEqual(busy["code"], "busy")
+
+    async def test_active_log_survives_and_keeps_growing(self):
+        self.seed()
+        plan = await self.preview()
+        await self.run_apply(plan)
+        self.app_logger.info("written after the cleanup")
+        with open(self.path("prompt_manager.log"), "rb") as handle:
+            content = handle.read()
+        self.assertIn(b"active log line", content)
+        self.assertIn(b"written after the cleanup", content)
+
+    async def test_an_unexpected_failure_fails_the_job_without_details(self):
+        self.seed()
+        plan = await self.preview()
+        with mock.patch.object(
+            self.manager, "delete_rotated", side_effect=RuntimeError(self.tmpdir)
+        ):
+            with self.assertLogs("prompt_manager", "ERROR"):
+                _summary, job = await self.run_apply(plan)
+        self.assertEqual(job.state, "failed")
+        self.assertNotIn(self.tmpdir, json.dumps(job.report()))
+        self.assertEqual(job.status_line, "The log cleanup failed.")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_api_logging.py

@@ -5,9 +5,10 @@ import os
 import shutil
 import sys
 import tempfile
+import threading
 import unittest
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 _mock_server = MagicMock()
 _mock_server.PromptServer.instance.routes = MagicMock()
@@ -20,7 +21,9 @@ from aiohttp.test_utils import AioHTTPTestCase  # noqa: E402
 
 from database.operations import PromptDatabase  # noqa: E402
 from py.api import PromptManagerAPI  # noqa: E402
+from py.api import logging_routes  # noqa: E402
 from py.api.logging_routes import LOG_CONFIG_KEYS  # noqa: E402
+from utils import logging_config  # noqa: E402
 
 
 class FakeLoggerManager:
@@ -38,11 +41,16 @@ class FakeLoggerManager:
         }
         self.update_calls = []
         self.logs = []
+        self.threads = {}
 
     def update_config(self, new_config):
+        self.threads["update_config"] = threading.current_thread()
         self.update_calls.append(dict(new_config))
         self.config.update(new_config)
 
+    def log_file_entry(self, name):
+        return logging_config.log_file_entry(self.log_dir, name)
+
     def get_config(self):
         return dict(self.config)
 
@@ -54,23 +62,36 @@ class FakeLoggerManager:
 
     def get_log_files(self):
         files = []
-        for path in sorted(self.log_dir.glob("*.log*")):
+        for row in sorted(
+            logging_config.scan_log_folder(self.log_dir), key=lambda r: r["name"]
+        ):
+            if row["_dir"]:
+                continue
             files.append(
                 {
-                    "filename": path.name,
-                    "path": str(path),
-                    "size": path.stat().st_size,
+                    "filename": row["name"],
+                    "path": str(self.log_dir / row["name"]),
+                    "size": row["size"],
                     "modified": "2026-10-07T00:00:00",
-                    "is_main": path.name == "prompt_manager.log",
+                    "is_main": row["kind"] == "active",
+                    "kind": row["kind"],
                 }
             )
         return files
 
     def truncate_logs(self):
+        """Rule R: empty the active log, delete rotated copies, keep the rest."""
+        self.threads["truncate_logs"] = threading.current_thread()
         truncated = []
-        for path in self.log_dir.glob("prompt_manager.log*"):
-            path.write_text("")
-            truncated.append(path.name)
+        for row in logging_config.scan_log_folder(self.log_dir):
+            path = self.log_dir / row["name"]
+            if row["kind"] == "active":
+                path.write_text("")
+            elif row["kind"] == "rotated":
+                path.unlink()
+            else:
+                continue
+            truncated.append(row["name"])
         return {"truncated": truncated, "errors": []}
 
     def get_log_stats(self):
@@ -378,12 +399,129 @@ class TestLogFilesAndDownload(LoggingAPITestCase):
         self.assertEqual(resp.status, 400)
 
     async def test_download_read_failure_is_500(self):
-        (self.log_dir / "dir.log").mkdir()
-        resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.log")
+        # Rewritten (#329): a directory is now a 404; inject a read error.
+        (self.log_dir / "prompt_manager.log").write_bytes(b"x")
+        secret = os.path.join(self.tmpdir, "logs", "prompt_manager.log")
+
+        def fail(path):
+            raise PermissionError(13, "Permission denied", secret)
+
+        with patch.object(logging_routes, "_read_log_bytes", fail):
+            resp = await self.client.request(
+                "GET", "/prompt_manager/logs/download/prompt_manager.log"
+            )
         body = await resp.text()
         self.assertEqual(resp.status, 500)
         self.assertNotIn(self.tmpdir, body)
 
+    async def test_directory_download_is_404(self):
+        (self.log_dir / "dir.log").mkdir()
+        resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.log")
+        self.assertEqual(resp.status, 404)
+
+    async def test_download_refuses_a_symlink(self):
+        secret = Path(self.tmpdir) / "secret.txt"
+        secret.write_text("the key")
+        try:
+            os.symlink(secret, self.log_dir / "prompt_manager.log.3")
+        except (OSError, NotImplementedError):  # pragma: no cover
+            self.skipTest("this account cannot create symlinks")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.3"
+        )
+        body = await resp.text()
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("the key", body)
+
+    async def test_download_refuses_a_hard_link(self):
+        secret = Path(self.tmpdir) / "secret.txt"
+        secret.write_text("the key")
+        try:
+            os.link(secret, self.log_dir / "prompt_manager.log.3")
+        except (OSError, NotImplementedError):  # pragma: no cover
+            self.skipTest("this filesystem cannot create hard links")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.3"
+        )
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("the key", await resp.text())
+
+    async def test_download_refuses_a_file_swapped_after_it_was_listed(self):
+        (self.log_dir / "prompt_manager.log.2").write_text("listed")
+        other = Path(self.tmpdir) / "other.txt"
+        other.write_text("swapped in")
+        real_entry = self.manager.log_file_entry
+
+        def entry_then_swap(name):
+            path = real_entry(name)
+            os.replace(other, path)  # another file now has the listed name
+            return path
+
+        self.manager.log_file_entry = entry_then_swap
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.2"
+        )
+        self.assertEqual(resp.status, 404)
+        self.assertNotIn("swapped in", await resp.text())
+
+    @unittest.skipUnless(hasattr(os, "mkfifo"), "FIFOs are POSIX only")
+    async def test_download_refuses_a_fifo_without_blocking(self):
+        os.mkfifo(self.log_dir / "pipe.log")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/pipe.log"
+        )
+        self.assertEqual(resp.status, 404)
+
+    async def test_download_reads_off_the_loop(self):
+        (self.log_dir / "prompt_manager.log").write_bytes(b"line\n")
+        seen = []
+        real = logging_routes._read_log_bytes
+
+        def read(path):
+            seen.append(threading.current_thread())
+            return real(path)
+
+        with patch.object(logging_routes, "_read_log_bytes", read):
+            resp = await self.client.request(
+                "GET", "/prompt_manager/logs/download/prompt_manager.log"
+            )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(len(seen), 1)
+        self.assertIsNot(seen[0], threading.current_thread())
+
+    async def test_download_keeps_its_response_shape(self):
+        (self.log_dir / "prompt_manager.log.2").write_bytes(b"a\nb\n")
+        resp = await self.client.request(
+            "GET", "/prompt_manager/logs/download/prompt_manager.log.2"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.content_type, "text/plain")
+        self.assertEqual(
+            resp.headers["Content-Disposition"],
+            'attachment; filename="prompt_manager.log.2"',
+        )
+        self.assertEqual(resp.headers["Content-Length"], "4")
+        self.assertEqual(await resp.read(), b"a\nb\n")
+
+    async def test_log_files_rows_carry_their_kind(self):
+        for name in (
+            "prompt_manager.log",
+            "prompt_manager.log.1",
+            "prompt_manager.log.bak",
+        ):
+            (self.log_dir / name).write_text("x")
+        resp = await self.client.request("GET", "/prompt_manager/logs/files")
+        data = await resp.json()
+        kinds = {row["filename"]: row["kind"] for row in data["files"]}
+        self.assertEqual(
+            kinds,
+            {
+                "prompt_manager.log": "active",
+                "prompt_manager.log.1": "rotated",
+                "prompt_manager.log.bak": "other",
+            },
+        )
+
 
 class TestTruncateConfigAndStats(LoggingAPITestCase):
 
@@ -395,6 +533,40 @@ class TestTruncateConfigAndStats(LoggingAPITestCase):
         self.assertEqual(data["results"]["truncated"], ["prompt_manager.log"])
         self.assertEqual((self.log_dir / "prompt_manager.log").read_text(), "")
 
+    async def test_truncate_keeps_other_files(self):
+        for name in (
+            "prompt_manager.log",
+            "prompt_manager.log.1",
+            "prompt_manager.log.bak",
+        ):
+            (self.log_dir / name).write_text("data")
+        resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
+        data = await resp.json()
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(set(data), {"success", "message", "results"})
+        self.assertEqual(
+            sorted(data["results"]["truncated"]),
+            ["prompt_manager.log", "prompt_manager.log.1"],
+        )
+        self.assertEqual(
+            sorted(p.name for p in self.log_dir.iterdir()),
+            ["prompt_manager.log", "prompt_manager.log.bak"],
+        )
+
+    async def test_truncate_runs_off_the_loop(self):
+        resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
+        self.assertEqual(resp.status, 200)
+        self.assertIsNot(
+            self.manager.threads["truncate_logs"], threading.current_thread()
+        )
+
+    async def test_update_config_runs_off_the_loop(self):
+        resp = await self._post_config({"level": "DEBUG"})
+        self.assertEqual(resp.status, 200)
+        self.assertIsNot(
+            self.manager.threads["update_config"], threading.current_thread()
+        )
+
     async def test_truncate_failure_is_500(self):
         def boom():
             raise RuntimeError("locked")

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_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/test_utils_logging_config.py

@@ -1,8 +1,10 @@
 """PromptManagerLogger: memory buffer, file rotation, level changes, truncation."""
 
 import logging
+import ntpath
 import os
 import pathlib
+import stat
 import sys
 import tempfile
 import threading
@@ -11,14 +13,20 @@ import unittest.mock as mock
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils import logging_config
-from utils.logging_config import (
+from utils import logging_config  # noqa: E402
+from utils.logging_config import (  # noqa: E402
     MemoryBufferHandler,
     PromptManagerLogger,
+    RotatedChanged,
     get_logger,
     get_logger_manager,
 )
 
+try:
+    from tests.open_handles import assert_closed  # noqa: E402
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed  # noqa: E402
+
 
 class IsolatedLoggerTestCase(unittest.TestCase):
     """A logger manager whose log directory is a temporary folder."""
@@ -32,6 +40,7 @@ class IsolatedLoggerTestCase(unittest.TestCase):
         with mock.patch.object(logging_config, "__file__", fake_file):
             self.manager = object.__new__(PromptManagerLogger)  # bypass singleton
             self.manager.__init__()
+        self.allowed_thread_names = set()
         self.manager.update_config({"console_logging": False})
         # Drop the "Updated logging configuration" entry.
         self.manager._log_buffer.clear()
@@ -53,12 +62,21 @@ class IsolatedLoggerTestCase(unittest.TestCase):
 
     def _filter_to_this_thread(self):
         this_thread = threading.get_ident()
+        # A test may let named helper threads of its own through too.
+        allowed_names = self.allowed_thread_names
+
+        def allowed(record):
+            return record.thread == this_thread or record.threadName in allowed_names
+
         for handler in self.manager.logger.handlers:
-            handler.addFilter(lambda record: record.thread == this_thread)
+            handler.filters.clear()
+            handler.addFilter(allowed)
 
     def _restore_global_logging(self):
         for handler in logging.getLogger("prompt_manager").handlers[:]:
             handler.close()
+        # Windows refuses to delete an open log: prove every handle is closed.
+        assert_closed(self, self.tmp.name)
         get_logger_manager()._setup_loggers()
         self.tmp.cleanup()
 
@@ -175,15 +193,14 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_unreadable_files_are_skipped_in_the_listing(self):
         self.logger.info("something")
-        original_stat = pathlib.Path.stat
+        original_lstat = os.lstat
 
-        def stat_failing_for_logs(path, *args, **kwargs):
-            if path.suffix == ".log":
+        def lstat_failing_for_logs(path, *args, **kwargs):
+            if str(path).endswith(".log"):
                 raise OSError("denied")
-            return original_stat(path, *args, **kwargs)
+            return original_lstat(path, *args, **kwargs)
 
-        with mock.patch.object(pathlib.Path, "stat", autospec=True) as stat:
-            stat.side_effect = stat_failing_for_logs
+        with mock.patch.object(logging_config.os, "lstat", lstat_failing_for_logs):
             self.assertEqual(self.manager.get_log_files(), [])
 
     def test_truncate_does_not_roll_the_log_over_while_deleting_backups(self):
@@ -228,10 +245,522 @@ class TestFilesAndRotation(IsolatedLoggerTestCase):
 
     def test_truncate_reports_files_it_cannot_remove(self):
         self._rotate()
-        with mock.patch.object(pathlib.Path, "unlink", side_effect=OSError("busy")):
+
+        def busy(path, *args, **kwargs):
+            raise PermissionError(13, "busy", str(path))
+
+        with mock.patch.object(logging_config.os, "unlink", busy):
             result = self.manager.truncate_logs()
         self.assertEqual(len(result["errors"]), 2)
         self.assertEqual(result["truncated"], ["prompt_manager.log"])
+        for error in result["errors"]:
+            self.assertNotIn(self.tmp.name, error)
+            self.assertRegex(error, r"prompt_manager\.log\.[12]")
+
+
+class TestRotatedCopyRule(IsolatedLoggerTestCase):
+    """Rule R (#329): only prompt_manager.log.<n> regular files are deletable."""
+
+    def setUp(self):
+        super().setUp()
+        self.logger.info("active log line")  # the active log exists
+        self.outside = os.path.join(self.tmp.name, "outside")
+        os.makedirs(self.outside)
+
+    def _path(self, name):
+        return os.path.join(self.log_dir, name)
+
+    def _write(self, name, data=b"rotated\n"):
+        with open(self._path(name), "wb") as handle:
+            handle.write(data)
+        return self._path(name)
+
+    def _symlink(self, name, target_name="secret.txt", data=b"secret " * 1000):
+        """A symlink in the log folder to a file outside it, or None."""
+        target = os.path.join(self.outside, target_name)
+        with open(target, "wb") as handle:
+            handle.write(data)
+        try:
+            os.symlink(target, self._path(name))
+        except (OSError, NotImplementedError):  # pragma: no cover - no privilege
+            return None
+        return target
+
+    def _require_symlink(self, name, **kwargs):
+        target = self._symlink(name, **kwargs)
+        if target is None:  # pragma: no cover - Windows without the privilege
+            self.skipTest("this account cannot create symlinks")
+        return target
+
+    def _kinds(self):
+        return {row["name"]: row["kind"] for row in self.manager.scan_log_folder()}
+
+    def _rotated_rows(self):
+        return [r for r in self.manager.scan_log_folder() if r["kind"] == "rotated"]
+
+    def _names(self):
+        return sorted(os.listdir(self.log_dir))
+
+    def _file_handler(self):
+        return next(
+            h
+            for h in self.manager.logger.handlers
+            if isinstance(h, logging.FileHandler)
+        )
+
+    def test_scan_classes_rotated_active_link_and_other(self):
+        for name in ("prompt_manager.log.1", "prompt_manager.log.2"):
+            self._write(name)
+        self._write("prompt_manager.log.3", b"x" * 3000)
+        for name in (
+            "prompt_manager.log.0",
+            "prompt_manager.log.01",
+            "prompt_manager.log.bak",
+            "prompt_manager.log.1.gz",
+            "prompt_manager.log.-1",
+            "notes.txt",
+        ):
+            self._write(name)
+        os.makedirs(self._path("prompt_manager.log.5"))
+        self._require_symlink("prompt_manager.log.4")
+        kinds = self._kinds()
+        self.assertEqual(
+            kinds,
+            {
+                "prompt_manager.log": "active",
+                "prompt_manager.log.1": "rotated",
+                "prompt_manager.log.2": "rotated",
+                "prompt_manager.log.3": "rotated",
+                "prompt_manager.log.0": "other",
+                "prompt_manager.log.01": "other",
+                "prompt_manager.log.bak": "other",
+                "prompt_manager.log.1.gz": "other",
+                "prompt_manager.log.-1": "other",
+                "notes.txt": "other",
+                "prompt_manager.log.5": "other",
+                "prompt_manager.log.4": "link",
+            },
+        )
+        row = next(
+            r
+            for r in self.manager.scan_log_folder()
+            if r["name"] == "prompt_manager.log.3"
+        )
+        self.assertEqual(
+            set(row), {"name", "kind", "size", "mtime_ns", "ino", "modified"}
+        )
+        stat = os.lstat(self._path("prompt_manager.log.3"))
+        self.assertEqual(row["size"], 3000)
+        self.assertEqual(row["mtime_ns"], stat.st_mtime_ns)
+        self.assertEqual(row["ino"], stat.st_ino)
+
+    def test_scan_matches_names_case_insensitively_on_windows(self):
+        self._write("PROMPT_MANAGER.LOG.2")
+        if os.name == "nt":  # pragma: no cover - native case folding
+            self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
+            return
+        # POSIX names are case-sensitive: another file, never deleted.
+        self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "other")
+        with mock.patch.object(logging_config, "_name_key", ntpath.normcase):
+            self.assertEqual(self._kinds()["PROMPT_MANAGER.LOG.2"], "rotated")
+            result = self.manager.delete_rotated(self._rotated_rows())
+        self.assertEqual(result["removed"], ["PROMPT_MANAGER.LOG.2"])
+
+    def test_a_junction_is_classed_as_a_link_and_kept(self):
+        # Portable control: whatever os.path.isjunction reports is a link.
+        self._write("prompt_manager.log.4")
+        self._write("prompt_manager.log.1")
+        junction = self._path("prompt_manager.log.4")
+        real_isjunction = os.path.isjunction
+
+        def isjunction(path):
+            return os.fspath(path) == junction or real_isjunction(path)
+
+        with mock.patch.object(logging_config.os.path, "isjunction", isjunction):
+            self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
+            self.manager.delete_rotated(self._rotated_rows())
+            self.manager.truncate_logs()
+        self.assertIn("prompt_manager.log.4", self._names())
+        self.assertNotIn("prompt_manager.log.1", self._names())
+
+    @unittest.skipUnless(os.name == "nt", "junctions exist only on Windows")
+    def test_junction_is_never_deleted_on_windows(self):  # pragma: no cover
+        import _winapi
+
+        target = os.path.join(self.outside, "kept")
+        os.makedirs(target)
+        with open(os.path.join(target, "inside.txt"), "wb") as handle:
+            handle.write(b"keep me")
+        _winapi.CreateJunction(target, self._path("prompt_manager.log.4"))
+        self._write("prompt_manager.log.1")
+        self.assertEqual(self._kinds()["prompt_manager.log.4"], "link")
+        self.manager.delete_rotated(self._rotated_rows())
+        self.manager.truncate_logs()
+        self.assertIn("prompt_manager.log.4", self._names())
+        self.assertTrue(os.path.exists(os.path.join(target, "inside.txt")))
+        with self.assertRaises(FileNotFoundError):
+            self.manager.log_file_entry("prompt_manager.log.4")
+        os.rmdir(self._path("prompt_manager.log.4"))  # removes the junction only
+
+    def test_delete_rotated_removes_exactly_the_listed_files(self):
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}", b"r" * n)
+        self._write("prompt_manager.log.bak")
+        os.makedirs(self._path("prompt_manager.log.7"))
+        target = self._symlink("prompt_manager.log.4")
+        listed = [
+            r for r in self._rotated_rows() if r["name"] != "prompt_manager.log.3"
+        ]
+        result = self.manager.delete_rotated(listed)
+        self.assertEqual(
+            result,
+            {"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
+        )
+        expected = [
+            "prompt_manager.log",
+            "prompt_manager.log.3",
+            "prompt_manager.log.7",
+            "prompt_manager.log.bak",
+        ]
+        if target is not None:
+            expected.append("prompt_manager.log.4")
+            self.assertTrue(os.path.exists(target))
+        self.assertEqual(self._names(), sorted(expected))
+
+    def test_delete_rotated_refuses_all_when_one_changed(self):
+        def change_size(path):
+            with open(path, "ab") as handle:
+                handle.write(b"more")
+
+        def change_mtime(path):
+            stat = os.lstat(path)
+            os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
+
+        def change_inode(path):
+            stat = os.lstat(path)
+            with open(path, "rb") as handle:
+                data = handle.read()
+            other = path + ".new"
+            with open(other, "wb") as handle:
+                handle.write(data)
+            os.utime(other, ns=(stat.st_atime_ns, stat.st_mtime_ns))
+            os.replace(other, path)
+            if os.lstat(path).st_ino in (0, stat.st_ino):  # pragma: no cover
+                self.skipTest("this filesystem reports no inode numbers")
+
+        def remove(path):
+            os.unlink(path)
+
+        def replace_by_link(path):
+            os.unlink(path)
+            target = os.path.join(self.outside, "elsewhere.txt")
+            with open(target, "wb") as handle:
+                handle.write(b"rotated\n")
+            try:
+                os.symlink(target, path)
+            except (OSError, NotImplementedError):  # pragma: no cover
+                self.skipTest("this account cannot create symlinks")
+
+        changes = {
+            "size": change_size,
+            "mtime": change_mtime,
+            "inode": change_inode,
+            "missing": remove,
+            "link": replace_by_link,
+        }
+        for label, change in changes.items():
+            with self.subTest(change=label):
+                for n in (1, 2, 3):
+                    self._write(f"prompt_manager.log.{n}", b"rotated\n")
+                reviewed = self._rotated_rows()
+                change(self._path("prompt_manager.log.2"))
+                before = self._names()
+                with self.assertRaises(RotatedChanged):
+                    self.manager.delete_rotated(reviewed)
+                self.assertEqual(self._names(), before)
+                self.assertIn("prompt_manager.log.1", before)
+                self.assertIn("prompt_manager.log.3", before)
+                for name in self._names():
+                    if name.startswith("prompt_manager.log."):
+                        os.unlink(self._path(name))
+
+    def test_rollover_waits_while_cleanup_holds_the_handler_lock(self):
+        # The file is already over its limit: the next record rolls it over,
+        # renaming .1 to .2 and the active log to .1.
+        self.manager.update_config({"max_file_size": 64, "backup_count": 3})
+        self.logger.info("x" * 100)
+        self._write("prompt_manager.log.1", b"first copy\n")
+        self._write("prompt_manager.log.2", b"second copy\n")
+        reviewed = self._rotated_rows()
+        holding = threading.Event()
+        release = threading.Event()
+        logged = threading.Event()
+        real_unlink = os.unlink
+        self.allowed_thread_names.add("pm-test-rollover")
+
+        def log_from_another_thread():
+            self.logger.info("rolls the log over")
+            logged.set()
+
+        worker = threading.Thread(
+            target=log_from_another_thread, name="pm-test-rollover", daemon=True
+        )
+
+        def slow_unlink(path, *args, **kwargs):
+            if not holding.is_set():
+                worker.start()  # its record must wait for the cleanup's lock
+                holding.set()
+                if not release.wait(10):
+                    raise RuntimeError("the test never released the cleanup")
+            return real_unlink(path, *args, **kwargs)
+
+        result = {}
+
+        def cleanup():
+            result.update(self.manager.delete_rotated(reviewed))
+
+        with mock.patch.object(logging_config.os, "unlink", slow_unlink):
+            cleaner = threading.Thread(target=cleanup, daemon=True)
+            cleaner.start()
+            self.assertTrue(holding.wait(10), "the cleanup never started deleting")
+            # The record waits on the handler lock the cleanup holds.
+            self.assertFalse(logged.wait(0.3), "a record was written mid-cleanup")
+            release.set()
+            cleaner.join(10)
+            worker.join(10)
+        self.assertFalse(cleaner.is_alive())
+        self.assertTrue(logged.is_set())
+        self.assertEqual(
+            result,
+            {"removed": ["prompt_manager.log.1", "prompt_manager.log.2"], "failed": []},
+        )
+        # The rollover ran after the cleanup: only the old active log moved.
+        self.assertEqual(self._names(), ["prompt_manager.log", "prompt_manager.log.1"])
+        with open(self._path("prompt_manager.log.1"), "rb") as handle:
+            self.assertIn(b"x" * 100, handle.read())
+
+    def test_open_rotated_copy_is_reported_not_deleted(self):
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}")
+        real_unlink = os.unlink
+        held = self._path("prompt_manager.log.2")
+
+        def unlink(path, *args, **kwargs):
+            if os.fspath(path) == held:
+                error = PermissionError(
+                    13,
+                    "The process cannot access the file because it is being "
+                    "used by another process",
+                    held,
+                )
+                error.winerror = 32
+                raise error
+            return real_unlink(path, *args, **kwargs)
+
+        with mock.patch.object(logging_config.os, "unlink", unlink):
+            result = self.manager.delete_rotated(self._rotated_rows())
+        self.assertEqual(
+            result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
+        )
+        [(name, error)] = result["failed"]
+        self.assertEqual(name, "prompt_manager.log.2")
+        self.assertIsInstance(error, PermissionError)
+        self.assertIn("prompt_manager.log.2", self._names())
+
+    @unittest.skipUnless(os.name == "nt", "only Windows refuses to unlink an open file")
+    def test_open_rotated_copy_is_reported_not_deleted_on_windows(
+        self,
+    ):  # pragma: no cover
+        for n in (1, 2, 3):
+            self._write(f"prompt_manager.log.{n}")
+        reviewed = self._rotated_rows()
+        handle = open(self._path("prompt_manager.log.2"), "rb")
+        try:
+            result = self.manager.delete_rotated(reviewed)
+        finally:
+            handle.close()
+        self.assertEqual(
+            result["removed"], ["prompt_manager.log.1", "prompt_manager.log.3"]
+        )
+        [(name, error)] = result["failed"]
+        self.assertEqual(name, "prompt_manager.log.2")
+        self.assertIsInstance(error, PermissionError)
+        self.assertIn("prompt_manager.log.2", self._names())
+
+    def test_cleanup_never_logs_while_holding_the_handler_lock(self):
+        for n in (1, 2):
+            self._write(f"prompt_manager.log.{n}")
+        handler = self._file_handler()
+        depth = 0
+        calls = []
+        real_acquire, real_release = handler.acquire, handler.release
+
+        def acquire():
+            nonlocal depth
+            real_acquire()
+            depth += 1
+
+        def release():
+            nonlocal depth
+            depth -= 1
+            real_release()
+
+        real_call_handlers = logging.Logger.callHandlers
+
+        def call_handlers(logger_self, record):
+            calls.append((record.getMessage(), depth))
+            return real_call_handlers(logger_self, record)
+
+        with (
+            mock.patch.object(handler, "acquire", acquire),
+            mock.patch.object(handler, "release", release),
+            mock.patch.object(logging.Logger, "callHandlers", call_handlers),
+        ):
+            self.manager.delete_rotated(self._rotated_rows())
+            self.assertEqual(depth, 0)
+            self._write("prompt_manager.log.1")
+            self.manager.truncate_logs()
+            self.assertEqual(depth, 0)
+        self.assertTrue(calls, "the cleanup logged no summary")
+        self.assertEqual([c for c in calls if c[1]], [])
+
+    def test_truncate_keeps_other_files_and_links(self):
+        # Corrected (#329): the legacy glob deleted every prompt_manager.log*.
+        self._rotate()
+        self._write("prompt_manager.log.bak", b"mine")
+        self._write("prompt_manager.log.1.gz", b"zip")
+        target = self._symlink("prompt_manager.log.4")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result["errors"], [])
+        self.assertEqual(
+            sorted(result["truncated"]),
+            ["prompt_manager.log", "prompt_manager.log.1", "prompt_manager.log.2"],
+        )
+        names = self._names()
+        self.assertIn("prompt_manager.log.bak", names)
+        self.assertIn("prompt_manager.log.1.gz", names)
+        self.assertNotIn("prompt_manager.log.1", names)
+        if target is not None:
+            self.assertIn("prompt_manager.log.4", names)
+            self.assertTrue(os.path.islink(self._path("prompt_manager.log.4")))
+            with open(target, "rb") as handle:
+                self.assertEqual(handle.read(), b"secret " * 1000)
+
+    @unittest.skipIf(os.name == "nt", "Windows cannot rename a log that is open")
+    def test_truncate_never_empties_a_renamed_log_the_handler_still_writes(self):
+        # The user renamed the active log to keep it; the handler still has it
+        # open. Clear must empty the new prompt_manager.log, not the kept copy.
+        self.logger.info("worth keeping")
+        kept = os.path.join(self.log_dir, "saved_copy.txt")
+        os.rename(self._path("prompt_manager.log"), kept)
+        with open(kept, "rb") as handle:
+            before = handle.read()
+        self._write("prompt_manager.log", b"fresh active log\n")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result["errors"], [])
+        with open(kept, "rb") as handle:
+            self.assertTrue(handle.read().startswith(before))
+        with open(self._path("prompt_manager.log"), "rb") as handle:
+            self.assertEqual(handle.read(), b"")
+
+    def test_name_surrogate_reparse_points_are_links_and_placeholders_are_not(self):
+        regular = stat.S_IFREG | 0o644
+        path = self._write("prompt_manager.log.3")
+        for tag, is_link in (
+            (0xA000001D, True),  # WSL symlink
+            (0xA0000003, True),  # mount point (junction)
+            (0x9000001A, False),  # OneDrive cloud placeholder: a regular file
+            (0, False),
+        ):
+            with self.subTest(tag=hex(tag)):
+                st = mock.Mock(st_mode=regular, st_reparse_tag=tag)
+                self.assertIs(logging_config._is_link(path, st), is_link)
+
+    def test_truncate_with_file_logging_off_never_creates_the_active_log(self):
+        self.manager.update_config({"file_logging": False})
+        os.unlink(self._path("prompt_manager.log"))
+        self._write("prompt_manager.log.1")
+        result = self.manager.truncate_logs()
+        self.assertEqual(result, {"truncated": ["prompt_manager.log.1"], "errors": []})
+        self.assertEqual(self._names(), [])
+
+    def test_truncate_never_empties_a_linked_active_log(self):
+        self.manager.update_config({"file_logging": False})
+        os.unlink(self._path("prompt_manager.log"))
+        target = self._require_symlink("prompt_manager.log", data=b"keep")
+        self.manager.truncate_logs()
+        with open(target, "rb") as handle:
+            self.assertEqual(handle.read(), b"keep")
+
+    def test_log_files_mark_kind_and_lstat_links(self):
+        self._write("prompt_manager.log.1", b"r" * 10)
+        self._write("prompt_manager.log.bak", b"b")
+        target = self._require_symlink("prompt_manager.log.4")
+        rows = {row["filename"]: row for row in self.manager.get_log_files()}
+        self.assertEqual(rows["prompt_manager.log"]["kind"], "active")
+        self.assertTrue(rows["prompt_manager.log"]["is_main"])
+        self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
+        self.assertEqual(rows["prompt_manager.log.bak"]["kind"], "other")
+        link = rows["prompt_manager.log.4"]
+        self.assertEqual(link["kind"], "link")
+        self.assertEqual(
+            link["size"], os.lstat(self._path("prompt_manager.log.4")).st_size
+        )
+        self.assertNotEqual(link["size"], os.path.getsize(target))
+        for row in rows.values():
+            self.assertEqual(
+                set(row), {"filename", "path", "size", "modified", "is_main", "kind"}
+            )
+
+    def test_an_unreadable_timestamp_does_not_break_the_scan(self):
+        self._write("prompt_manager.log.1")
+        real = logging_config.datetime
+
+        class Clock:
+            @staticmethod
+            def fromtimestamp(value):
+                raise OSError(22, "Invalid argument")
+
+        with mock.patch.object(logging_config, "datetime", Clock):
+            rows = {r["name"]: r for r in self.manager.scan_log_folder()}
+        self.assertIs(logging_config.datetime, real)
+        self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated")
+        self.assertEqual(rows["prompt_manager.log.1"]["modified"], "")
+
+    def test_stats_exclude_links(self):
+        self._write("prompt_manager.log.1", b"r" * 10)
+        before = self.manager.get_log_stats()
+        self._require_symlink("prompt_manager.log.4")
+        after = self.manager.get_log_stats()
+        self.assertEqual(after["log_files_count"], before["log_files_count"])
+        self.assertEqual(after["total_log_size"], before["total_log_size"])
+
+    def test_log_file_entry_accepts_only_listed_regular_files(self):
+        self._write("prompt_manager.log.1")
+        self._write("prompt_manager.log.bak")
+        os.makedirs(self._path("folder.log"))
+        self._symlink("prompt_manager.log.4")
+        self.assertEqual(
+            os.path.normcase(str(self.manager.log_file_entry("prompt_manager.log.1"))),
+            os.path.normcase(self._path("prompt_manager.log.1")),
+        )
+        self.manager.log_file_entry("prompt_manager.log")
+        self.manager.log_file_entry("prompt_manager.log.bak")
+        for name in (
+            "prompt_manager.log.4",
+            "folder.log",
+            "missing.log",
+            "",
+            ".",
+            "..",
+            "../outside/secret.txt",
+            "..\\outside\\secret.txt",
+            "prompt_manager.log:stream",
+            "prompt_manager.log\x00",
+        ):
+            with self.subTest(name=name):
+                with self.assertRaises(FileNotFoundError):
+                    self.manager.log_file_entry(name)
 
 
 class TestConfiguration(IsolatedLoggerTestCase):

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/logging_config.py

@@ -10,9 +10,12 @@ This module provides centralized logging configuration with support for:
 """
 
 import collections
+import fnmatch
 import logging
 import logging.handlers
 import os
+import re
+import stat
 import threading
 from datetime import datetime
 from typing import Dict, Any, Optional, List
@@ -20,6 +23,160 @@ from pathlib import Path
 
 from . import library_location
 
+ACTIVE_LOG_NAME = "prompt_manager.log"
+# Rule R (#329): a rotated copy is prompt_manager.log.<n>, n >= 1, no leading 0.
+_ROTATED_NAME = re.compile(r"prompt_manager\.log\.([1-9][0-9]*)")
+# The legacy listing's pattern (GET /logs/files and /logs/stats).
+_LISTED_PATTERN = "*.log*"
+LOG_FILE_KINDS = ("active", "rotated", "link", "other")
+
+
+class RotatedChanged(Exception):
+    """A reviewed rotated copy changed since the review; nothing was deleted."""
+
+    def __init__(self, name):
+        super().__init__(f"{name} changed since the review")
+        self.name = name
+
+
+def _name_key(name: str) -> str:
+    """How a name compares: case-folded where the filesystem folds it (Windows)."""
+    return os.path.normcase(name)
+
+
+def _rotated_number(name: str) -> Optional[int]:
+    match = _ROTATED_NAME.fullmatch(_name_key(name))
+    return int(match.group(1)) if match else None
+
+
+# Windows reparse tags with the name-surrogate bit name another file or
+# folder (symlinks, junctions, WSL symlinks); cloud placeholders do not.
+_NAME_SURROGATE = 0x20000000
+
+
+def _is_link(path: str, st: os.stat_result) -> bool:
+    if getattr(st, "st_reparse_tag", 0) & _NAME_SURROGATE:
+        return True
+    return stat.S_ISLNK(st.st_mode) or os.path.isjunction(path)
+
+
+class LogFilePath(type(Path())):
+    """A listed log file's path, with the inode it had when it was listed."""
+
+    listed_ino = 0
+
+
+def _kind(name: str, path: str, st: os.stat_result) -> str:
+    """Rule R: ``active``, ``rotated``, ``link`` or ``other`` (never deleted)."""
+    if _is_link(path, st):
+        return "link"
+    if not stat.S_ISREG(st.st_mode):
+        return "other"
+    if _name_key(name) == _name_key(ACTIVE_LOG_NAME):
+        return "active"
+    if _rotated_number(name) is not None:
+        return "rotated"
+    return "other"
+
+
+def scan_log_folder(log_dir) -> List[Dict[str, Any]]:
+    """Every direct entry of ``log_dir`` with its Rule R kind (blocking).
+
+    Rows are ``{name, kind, size, mtime_ns, ino, modified}``, read with
+    ``lstat`` so a link is described, never followed. An entry that cannot
+    be read is left out. Never logs: the cleanup calls it while it holds the
+    file handler's lock.
+    """
+    rows = []
+    try:
+        names = [entry.name for entry in os.scandir(log_dir)]
+    except OSError:
+        return rows
+    for name in names:
+        path = os.path.join(log_dir, name)
+        try:
+            st = os.lstat(path)
+            kind = _kind(name, path, st)
+        except OSError:
+            continue
+        try:
+            modified = datetime.fromtimestamp(st.st_mtime).isoformat()
+        except (OSError, OverflowError, ValueError):  # a time before 1970 on Windows
+            modified = ""
+        rows.append(
+            {
+                "name": name,
+                "kind": kind,
+                "size": st.st_size,
+                "mtime_ns": st.st_mtime_ns,
+                "ino": st.st_ino,
+                "modified": modified,
+                "_dir": stat.S_ISDIR(st.st_mode) and kind != "link",
+                "_regular": stat.S_ISREG(st.st_mode) and kind != "link",
+            }
+        )
+    return rows
+
+
+def public_row(row: Dict[str, Any]) -> Dict[str, Any]:
+    """A scan row without its private keys."""
+    return {key: value for key, value in row.items() if not key.startswith("_")}
+
+
+def rotated_sort_key(row: Dict[str, Any]) -> int:
+    return _rotated_number(row["name"]) or 0
+
+
+def log_file_entry(log_dir, name) -> Path:
+    """The path of a downloadable log file named ``name`` (blocking).
+
+    Only a regular, non-link file listed directly in ``log_dir`` under that
+    exact name; a separator, ``..``, a stream suffix or any unlisted name
+    raises FileNotFoundError, as a missing file does.
+    """
+    if not isinstance(name, str) or not name or name in (".", ".."):
+        raise FileNotFoundError("Log file not found")
+    if any(ch in name for ch in ("/", "\\", "\x00", ":")):
+        raise FileNotFoundError("Log file not found")
+    for row in scan_log_folder(log_dir):
+        if row["name"] == name and row["kind"] in ("active", "rotated", "other"):
+            if not row["_regular"]:  # a folder, FIFO, socket or device
+                break
+            entry = LogFilePath(log_dir, row["name"])
+            entry.listed_ino = row["ino"]
+            return entry
+    raise FileNotFoundError("Log file not found")
+
+
+def _stream_is(stream, ino: int) -> bool:
+    """Whether ``stream`` is open on the file with inode ``ino`` (0: unknown)."""
+    try:
+        open_ino = os.fstat(stream.fileno()).st_ino
+    except (OSError, ValueError, AttributeError):
+        return False
+    return not (ino and open_ino and open_ino != ino)
+
+
+def _error_text(error: BaseException) -> str:
+    """An OSError's reason and file name, without the folder it is in."""
+    if isinstance(error, OSError):
+        parts = [error.strerror or type(error).__name__]
+        if error.filename:
+            parts.append(os.path.basename(str(error.filename)))
+        return ": ".join(parts)
+    return str(error) or type(error).__name__
+
+
+def _matches_review(row: Optional[Dict[str, Any]], reviewed: Dict[str, Any]) -> bool:
+    if row is None or row["kind"] != "rotated":
+        return False
+    if row["size"] != reviewed.get("size") or row["mtime_ns"] != reviewed.get(
+        "mtime_ns"
+    ):
+        return False
+    ino = reviewed.get("ino") or 0
+    return not (ino and row["ino"] and row["ino"] != ino)
+
 
 class _OwnedRotatingFileHandler(logging.handlers.RotatingFileHandler):
     """Keep a retired owned sink closed when an in-flight record resumes."""
@@ -87,6 +244,9 @@ class PromptManagerLogger:
         # Memory buffer for recent logs (for web viewer)
         self._log_buffer = collections.deque(maxlen=self.config["buffer_size"])
         self._buffer_lock = threading.Lock()
+        # Held while a handler is replaced or log files are deleted (#329):
+        # taken before any file handler's own lock, never the other way round.
+        self._files_lock = threading.Lock()
 
         # Initialize loggers
         self._setup_loggers()
@@ -238,6 +398,14 @@ class PromptManagerLogger:
         # Return most recent first
         return list(reversed(logs[-limit:]))
 
+    def scan_log_folder(self) -> List[Dict[str, Any]]:
+        """Every entry of the log folder with its Rule R kind (blocking)."""
+        return [public_row(row) for row in scan_log_folder(self.log_dir)]
+
+    def log_file_entry(self, name: str) -> Path:
+        """A downloadable (regular, non-link) log file; FileNotFoundError else."""
+        return log_file_entry(self.log_dir, name)
+
     def get_log_files(self) -> List[Dict[str, Any]]:
         """Get information about available log files.
 
@@ -245,26 +413,27 @@ class PromptManagerLogger:
             List of dictionaries containing file information:
             - filename: Name of the log file
             - path: Full path to the log file
-            - size: File size in bytes
+            - size: File size in bytes (a link's own size: links are not followed)
             - modified: ISO formatted modification timestamp
             - is_main: True if this is the main log file
+            - kind: ``active``, ``rotated``, ``link`` or ``other`` (Rule R)
         """
         log_files = []
-
-        for log_file in self.log_dir.glob("*.log*"):
-            try:
-                stat = log_file.stat()
-                log_files.append(
-                    {
-                        "filename": log_file.name,
-                        "path": str(log_file),
-                        "size": stat.st_size,
-                        "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
-                        "is_main": log_file.name == "prompt_manager.log",
-                    }
-                )
-            except Exception as e:
-                self.logger.error(f"Error getting info for log file {log_file}: {e}")
+        for row in scan_log_folder(self.log_dir):
+            if row["_dir"]:
+                continue
+            if not fnmatch.fnmatchcase(_name_key(row["name"]), _LISTED_PATTERN):
+                continue
+            log_files.append(
+                {
+                    "filename": row["name"],
+                    "path": str(self.log_dir / row["name"]),
+                    "size": row["size"],
+                    "modified": row["modified"],
+                    "is_main": row["kind"] == "active",
+                    "kind": row["kind"],
+                }
+            )
 
         # Sort by modification time, newest first
         log_files.sort(key=lambda x: x["modified"], reverse=True)
@@ -302,14 +471,44 @@ class PromptManagerLogger:
             self.logger.error(f"Error reading log file {filename}: {e}")
             raise
 
-    def _truncate_main_log(self, log_file: Path) -> None:
+    def _file_handlers(self) -> List[logging.Handler]:
+        return [h for h in self.logger.handlers if isinstance(h, logging.FileHandler)]
+
+    def _acquire_file_handlers(self) -> List[logging.Handler]:
+        """Take every file handler's lock: no rollover renames a file meanwhile.
+
+        The caller holds ``_files_lock`` (so no handler is replaced) and must
+        not log until it released them: the lock is re-entrant, so a record
+        logged here could roll the log over under the caller's feet.
+        """
+        held = []
+        try:
+            for handler in self._file_handlers():
+                handler.acquire()
+                held.append(handler)
+        except BaseException:
+            self._release_handlers(held)
+            raise
+        return held
+
+    @staticmethod
+    def _release_handlers(held) -> None:
+        for handler in reversed(held):
+            handler.release()
+
+    def _truncate_main_log(self, log_file: Path, expected_ino: int = 0) -> None:
         """Empty the main log through the handler that writes to it.
 
         Opening the file a second time with ``"w"`` leaves the handler's own
         stream at its old offset: later records land past the new end
         (zero-padded on POSIX) and ``tell()`` keeps reporting the old size, so
         the next record rolls the file over. Truncating the handler's stream
-        keeps position and size in step.
+        keeps position and size in step. Without a handler on it (file logging
+        off) the existing file is emptied in place, never through a link.
+
+        A handler whose open stream is not the listed file (``expected_ino``:
+        the user renamed the log, which the handler still writes to) is left
+        alone: the file under the active name is emptied instead.
         """
         handled = False
         target = os.path.normcase(os.path.realpath(str(log_file)))
@@ -320,6 +519,8 @@ class PromptManagerLogger:
                 continue
             if os.path.normcase(os.path.realpath(base)) != target:
                 continue
+            if not _stream_is(stream, expected_ino):
+                continue
             handler.acquire()
             try:
                 stream.flush()
@@ -329,14 +530,65 @@ class PromptManagerLogger:
             finally:
                 handler.release()
         if not handled:
-            with open(log_file, "w", encoding="utf-8", errors="replace") as f:
-                f.write("")
+            flags = os.O_WRONLY | os.O_TRUNC | getattr(os, "O_NOFOLLOW", 0)
+            os.close(os.open(str(log_file), flags))
+
+    def delete_rotated(self, expected: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Delete exactly the reviewed rotated copies, or none of them (blocking).
+
+        ``expected`` are scan rows (``name``, ``size``, ``mtime_ns``,
+        ``ino``). Under ``_files_lock`` and every file handler's lock the
+        folder is scanned again; when any expected file is missing, is no
+        longer a rotated copy (Rule R) or differs in size, ``mtime_ns`` or a
+        non-zero ``ino``, RotatedChanged is raised with nothing deleted.
+        Otherwise each is unlinked in ``<n>`` order.
+
+        Returns ``{"removed": [names], "failed": [(name, OSError)]}``. One
+        summary line is logged once both locks are released.
+        """
+        removed, failed = [], []
+        changed = None
+        with self._files_lock:
+            held = self._acquire_file_handlers()
+            try:
+                current = {
+                    _name_key(row["name"]): row for row in scan_log_folder(self.log_dir)
+                }
+                targets = []
+                for reviewed in sorted(expected, key=rotated_sort_key):
+                    row = current.get(_name_key(str(reviewed.get("name", ""))))
+                    if not _matches_review(row, reviewed):
+                        changed = str(reviewed.get("name", ""))
+                        break
+                    targets.append(row["name"])
+                if changed is None:
+                    for name in targets:
+                        try:
+                            os.unlink(os.path.join(self.log_dir, name))
+                            removed.append(name)
+                        except OSError as error:
+                            failed.append((name, error))
+            finally:
+                self._release_handlers(held)
+        if changed is not None:
+            self.logger.warning(
+                "Rotated log cleanup refused: %s changed since the review", changed
+            )
+            raise RotatedChanged(changed)
+        self.logger.info(
+            "Deleted %d rotated log copies%s",
+            len(removed),
+            f", {len(failed)} could not be deleted" if failed else "",
+        )
+        return {"removed": removed, "failed": failed}
 
     def truncate_logs(self) -> Dict[str, Any]:
-        """Truncate all log files.
+        """Empty the active log and delete its rotated copies (Rule R).
 
-        Clears the main log file and deletes rotated log files.
-        Also clears the memory buffer.
+        Every other file in the log folder, and every link, is kept. Also
+        clears the memory buffer. Runs under ``_files_lock`` and the file
+        handlers' locks, so no rollover recreates a just-deleted copy, and
+        logs only once they are released.
 
         Returns:
             Dictionary with:
@@ -345,28 +597,31 @@ class PromptManagerLogger:
         """
         results = {"truncated": [], "errors": []}
 
-        # Collect first and stay silent until every file has been handled:
-        # a log line written mid-way can roll the just-emptied main log over
-        # and recreate the backup that was deleted a moment earlier.
-        log_files = sorted(self.log_dir.glob("prompt_manager.log*"))
-        for log_file in log_files:
+        with self._files_lock:
+            held = self._acquire_file_handlers()
             try:
-                if log_file.name == "prompt_manager.log":
-                    self._truncate_main_log(log_file)
-                else:
-                    # For rotated files, delete them
-                    log_file.unlink()
-
-                results["truncated"].append(log_file.name)
-
-            except Exception as e:
-                results["errors"].append(
-                    f"Failed to truncate {log_file.name}: {str(e)}"
+                rows = scan_log_folder(self.log_dir)
+                active = [row for row in rows if row["kind"] == "active"]
+                rotated = sorted(
+                    (row for row in rows if row["kind"] == "rotated"),
+                    key=rotated_sort_key,
                 )
-
-        # Clear memory buffer
-        with self._buffer_lock:
-            self._log_buffer.clear()
+                for row in active + rotated:
+                    path = os.path.join(self.log_dir, row["name"])
+                    try:
+                        if row["kind"] == "active":
+                            self._truncate_main_log(Path(path), row["ino"])
+                        else:
+                            os.unlink(path)
+                        results["truncated"].append(row["name"])
+                    except OSError as error:
+                        results["errors"].append(
+                            f"Failed to truncate {row['name']}: {_error_text(error)}"
+                        )
+                with self._buffer_lock:
+                    self._log_buffer.clear()
+            finally:
+                self._release_handlers(held)
 
         if results["truncated"]:
             self.logger.info("Truncated log files: %s", ", ".join(results["truncated"]))
@@ -409,7 +664,9 @@ class PromptManagerLogger:
                 "backup_count",
             ]
         ):
-            self._setup_loggers()
+            # No cleanup may hold the old handler while it is replaced.
+            with self._files_lock:
+                self._setup_loggers()
 
         self.logger.info(f"Updated logging configuration: {new_config}")
 
@@ -440,7 +697,9 @@ class PromptManagerLogger:
                 level = entry["level"]
                 level_counts[level] = level_counts.get(level, 0) + 1
 
-        log_files = self.get_log_files()
+        # A link's target is not a log file of this folder: neither counted
+        # nor sized.
+        log_files = [f for f in self.get_log_files() if f["kind"] != "link"]
         total_log_size = sum(f["size"] for f in log_files)
 
         return {

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)

utils/version.py

@@ -0,0 +1,18 @@
+"""The project version, read from ``pyproject.toml`` (#329)."""
+
+import re
+from pathlib import Path
+
+
+def project_version(root):
+    """The version in ``root``'s ``pyproject.toml``; ``"unknown"`` on failure."""
+    try:
+        pyproject_path = Path(root) / "pyproject.toml"
+        if pyproject_path.exists():
+            content = pyproject_path.read_text()
+            match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
+            if match:
+                return match.group(1)
+    except Exception:  # nosec B110 - an unreadable file reads as unknown
+        pass
+    return "unknown"

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];