feat(inspect): port legacy parser rules the new resolver missed (PromptManager)

R1: before 3.2.3 a save patch overwrote every PromptManager node's saved
text with the last-run node's prompt. The legacy viewer recognizes the
signature (comfy-metadata.js createContext: two or more PromptManager
workflow nodes with one API text and different widget values). Inspection
now refuses those nodes as malformed instead of resolving to another
node's prompt; the inspector hands the parsed workflow to the resolver for
this check only. Lands before the new pass-throughs (R6) that would
otherwise make these graphs resolve to the wrong text.

R7: this repository's node classes up to 07a5af3 (PromptManagerV2,
Positive, Negative, Detailer, V2Text, PositiveText, NegativeText) resolve
with their own outputs and their exact strip and ", " join.

The default (ingestion) profile ignores both.

tests/test_inspection_legacy_ports.py

@@ -0,0 +1,278 @@
+"""Legacy parser rules ported into the inspection resolver (#240, Task 2.7b).
+
+Every graph here is synthetic (neutral texts, generic names). Each class names
+the rule from the parser reconciliation (R1, R3, ...) it ports. The default
+(ingestion) profile must stay exactly as it was, so each rule also checks that
+``DEFAULT_PROFILE`` ignores it.
+"""
+
+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 inspection_parity as parity
+except ImportError:  # discovered with ``-s tests``
+    import inspection_parity as parity
+
+
+def _inspect(metadata: dict, **options) -> dict:
+    return resolve_prompt_pair(metadata, profile=resolver.INSPECTION_PROFILE, **options)
+
+
+def _texts(result: dict) -> tuple:
+    return tuple(
+        result[role]["text"] if isinstance(result[role], dict) else None
+        for role in ("positive", "negative")
+    )
+
+
+def _pm_encoder(text: str) -> dict:
+    """A PromptManager node whose API class was rewritten to CLIPTextEncode."""
+    return {
+        "class_type": "CLIPTextEncode",
+        "inputs": {
+            "text": text,
+            "category": "",
+            "tags": "",
+            "search_text": "",
+            "clip": ["4", 1],
+        },
+    }
+
+
+def _overwritten(widgets=("a cat", "a dog", "a bird"), api_text="a bird") -> dict:
+    """Pre-3.2.3 shape: three PromptManager nodes, every API ``text`` overwritten.
+
+    KSampler 3 reads node 17 (widget A) for its positive; the saved API text of
+    17, 113 and 119 is the last-run node's prompt (C) for all three.
+    """
+    prompt = parity.build("f02")["prompt"]
+    prompt["3"]["inputs"]["positive"] = ["17", 0]
+    del prompt["6"]
+    for node_id in ("17", "113", "119"):
+        prompt[node_id] = _pm_encoder(api_text)
+    workflow = {
+        "nodes": [
+            {"id": int(node_id), "type": "PromptManager", "widgets_values": [widget]}
+            for node_id, widget in zip(("17", "113", "119"), widgets)
+        ]
+        + [{"id": 7, "type": "CLIPTextEncode", "widgets_values": [parity.NEG]}],
+        "links": [],
+    }
+    return {"prompt": prompt, "workflow": workflow}
+
+
+class TestOverwrittenPromptManagerText(unittest.TestCase):
+    """R1 (comfy-metadata.js:52-86): never resolve to the overwritten text."""
+
+    def test_overwritten_text_is_refused_not_resolved(self):
+        result = _inspect(_overwritten())
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason_code"], "malformed")
+        self.assertEqual(result["reason"], resolver.OVERWRITTEN_PROMPT_REASON)
+        self.assertIsNone(result["positive"])
+
+    def test_the_inspector_passes_the_workflow_to_the_resolver(self):
+        metadata = _overwritten()
+        evidence = inspect_prompt_evidence(
+            {
+                "prompt": json.dumps(metadata["prompt"]),
+                "workflow": json.dumps(metadata["workflow"]),
+            }
+        )
+        self.assertEqual(evidence["resolution"]["reason_code"], "malformed")
+        self.assertEqual(evidence["prompts"]["positive"]["status"], "unresolved")
+        self.assertNotEqual(evidence["prompts"]["positive"]["text"], "a bird")
+        self.assertEqual(
+            evidence["prompts"]["positive"]["message"],
+            "Prompt metadata in this file is malformed.",
+        )
+
+    def test_untouched_shapes_still_resolve(self):
+        cases = {
+            # One PromptManager node, or several whose widgets agree.
+            "single": ("a bird",),
+            "agreeing widgets": ("a bird", "a bird", "a bird"),
+        }
+        for label, widgets in cases.items():
+            with self.subTest(label):
+                metadata = _overwritten()
+                metadata["workflow"]["nodes"] = [
+                    node
+                    for node in metadata["workflow"]["nodes"]
+                    if node["type"] != "PromptManager"
+                ] + [
+                    {"id": node_id, "type": "PromptManager", "widgets_values": [text]}
+                    for node_id, text in zip((17, 113, 119), widgets)
+                ]
+                self.assertEqual(_inspect(metadata)["state"], "resolved")
+        # Distinct API texts are what ran, whatever the widgets say.
+        metadata = _overwritten()
+        metadata["prompt"]["113"]["inputs"]["text"] = "a dog"
+        metadata["prompt"]["119"]["inputs"]["text"] = "a cat"
+        metadata["prompt"]["17"]["inputs"]["text"] = "a fish"
+        self.assertEqual(_texts(_inspect(metadata)), ("a fish", parity.NEG))
+        # A PromptManagerPair widget is its positive and negative: not the signature.
+        metadata = _overwritten()
+        for node in metadata["workflow"]["nodes"]:
+            node["type"] = node["type"].replace("PromptManager", "PromptManagerPair")
+        self.assertEqual(_inspect(metadata)["state"], "resolved")
+        # No workflow, or a malformed one, cannot show the signature.
+        for workflow in (None, "nodes", {"nodes": "x"}, {"nodes": [1, None]}):
+            with self.subTest(workflow=workflow):
+                metadata = _overwritten()
+                metadata["workflow"] = workflow
+                self.assertEqual(_inspect(metadata)["state"], "resolved")
+
+    def test_an_untouched_branch_beside_overwritten_nodes_resolves(self):
+        # Only the overwritten nodes are refused: the negative is not one.
+        metadata = _overwritten()
+        metadata["prompt"]["3"]["inputs"]["positive"] = ["7", 0]
+        self.assertEqual(_texts(_inspect(metadata)), (parity.NEG, parity.NEG))
+
+    def test_default_profile_ignores_the_workflow(self):
+        result = resolve_prompt_pair(_overwritten())
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(_texts(result), ("a bird", parity.NEG))
+
+
+def _historical(class_type: str, slot: int, inputs: dict, kind: str) -> dict:
+    """``f02`` whose positive reads ``class_type`` output ``slot`` (node 10).
+
+    A conditioning output feeds the sampler directly; a string output feeds a
+    CLIPTextEncode (node 6).
+    """
+    prompt = parity.build("f02")["prompt"]
+    prompt["10"] = {"class_type": class_type, "inputs": {"clip": ["4", 1], **inputs}}
+    if kind == "conditioning":
+        prompt["3"]["inputs"]["positive"] = ["10", slot]
+    else:
+        prompt["6"]["inputs"]["text"] = ["10", slot]
+    return {"prompt": prompt}
+
+
+class TestHistoricalPromptManagerNodes(unittest.TestCase):
+    """R7: this repository's node classes up to 07a5af3, with their own join."""
+
+    COMPOSED = {"text": " a cat ", "prepend_text": " trig ", "append_text": "sky "}
+
+    def test_v2_outputs_and_comma_join(self):
+        inputs = {**self.COMPOSED, "negative_text": " bad hands "}
+        for slot, kind, expected in (
+            (0, "conditioning", "trig, a cat, sky"),
+            (1, "conditioning", "bad hands"),
+            (2, "string", "trig, a cat, sky"),
+            (3, "string", "bad hands"),
+        ):
+            with self.subTest(slot=slot):
+                result = _inspect(_historical("PromptManagerV2", slot, inputs, kind))
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(result["positive"]["text"], expected)
+
+    def test_comma_rule_matches_the_historical_code(self):
+        cases = (
+            ({"text": "a cat", "prepend_text": "trig,"}, "trig,a cat"),
+            ({"text": "a cat,", "append_text": "sky"}, "a cat,sky"),
+            ({"text": "", "prepend_text": "trig", "append_text": "sky"}, "trig, sky"),
+            ({"text": "  ", "prepend_text": "  ", "append_text": "sky"}, "sky"),
+            ({"text": "a cat"}, "a cat"),
+        )
+        for inputs, expected in cases:
+            with self.subTest(inputs=inputs):
+                result = _inspect(_historical("PromptManagerV2", 2, inputs, "string"))
+                self.assertEqual(result["positive"]["text"], expected)
+
+    def test_single_role_classes(self):
+        cases = (
+            ("PromptManagerPositive", 0, "conditioning", "trig, a cat, sky"),
+            ("PromptManagerPositive", 1, "string", "trig, a cat, sky"),
+            ("PromptManagerDetailer", 0, "conditioning", "trig, a cat, sky"),
+            ("PromptManagerDetailer", 1, "string", "trig, a cat, sky"),
+            ("PromptManagerV2Text", 0, "string", "trig, a cat, sky"),
+            ("PromptManagerPositiveText", 0, "string", "trig, a cat, sky"),
+            ("PromptManagerNegativeText", 0, "string", "trig, a cat, sky"),
+            # The negative-only encoder never composes.
+            ("PromptManagerNegative", 0, "conditioning", "a cat"),
+            ("PromptManagerNegative", 1, "string", "a cat"),
+        )
+        for class_type, slot, kind, expected in cases:
+            with self.subTest(node=class_type, slot=slot):
+                result = _inspect(_historical(class_type, slot, self.COMPOSED, kind))
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(result["positive"]["text"], expected)
+        result = _inspect(
+            _historical(
+                "PromptManagerV2Text",
+                1,
+                {"text": "a cat", "negative_text": " bad "},
+                "string",
+            )
+        )
+        self.assertEqual(result["positive"]["text"], "bad")
+
+    def test_optional_negative_defaults_to_empty(self):
+        result = _inspect(
+            _historical("PromptManagerV2", 0, {"text": "a cat"}, "conditioning")
+        )
+        self.assertEqual(result["positive"]["text"], "a cat")
+        prompt = _historical("PromptManagerV2", 0, {"text": "a cat"}, "conditioning")
+        prompt["prompt"]["3"]["inputs"]["negative"] = ["10", 1]
+        self.assertEqual(_texts(_inspect(prompt)), ("a cat", ""))
+
+    def test_unlisted_outputs_and_kinds_are_not_guessed(self):
+        # PromptManagerDetailer output 2 is an INT prompt id.
+        result = _inspect(
+            _historical("PromptManagerDetailer", 2, self.COMPOSED, "string")
+        )
+        self.assertEqual(result["reason_code"], "unsupported-node")
+        # A STRING output read as conditioning is a malformed link.
+        result = _inspect(
+            _historical("PromptManagerV2", 2, self.COMPOSED, "conditioning")
+        )
+        self.assertEqual(result["reason_code"], "malformed")
+
+    def test_linked_parts_are_traced(self):
+        source = _historical(
+            "PromptManagerV2",
+            0,
+            {"text": ["11", 0], "prepend_text": ["12", 0]},
+            "conditioning",
+        )
+        source["prompt"]["11"] = {
+            "class_type": "PrimitiveString",
+            "inputs": {"value": "a cat"},
+        }
+        source["prompt"]["12"] = {
+            "class_type": "PrimitiveString",
+            "inputs": {"value": "trig"},
+        }
+        result = _inspect(source)
+        self.assertEqual(result["positive"]["text"], "trig, a cat")
+        self.assertEqual(
+            [step["node_id"] for step in result["positive"]["provenance"]["path"]],
+            ["10", "12", "11"],
+        )
+
+    def test_overwritten_historical_nodes_are_still_refused(self):
+        metadata = _overwritten()
+        for node_id in ("17", "113", "119"):
+            metadata["prompt"][node_id]["class_type"] = "PromptManagerV2"
+        for node in metadata["workflow"]["nodes"]:
+            if node["type"] == "PromptManager":
+                node["type"] = "PromptManagerV2"
+        self.assertEqual(_inspect(metadata)["reason_code"], "malformed")
+
+    def test_default_profile_keeps_them_unsupported(self):
+        result = resolve_prompt_pair(
+            _historical("PromptManagerV2", 0, self.COMPOSED, "conditioning")
+        )
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason"], "Unsupported node semantics or output slot")
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/prompt_inspector.py

@@ -435,6 +435,16 @@ def unreadable_evidence() -> dict:
     return {"prompts": _explained(prompts, resolution), "resolution": resolution}
 
 
+def _parsed_workflow(value: Any) -> dict | None:
+    """The UI workflow as an object, or ``None`` (absent, invalid or not JSON)."""
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except (ValueError, RecursionError):
+            return None
+    return value if isinstance(value, dict) else None
+
+
 def inspect_prompt_evidence(
     metadata: dict,
     saved: dict | None = None,
@@ -448,6 +458,7 @@ def inspect_prompt_evidence(
     directly with its default profile. ``resolution`` carries the closed
     ``reason_code`` and, for a blocking node, ``node: {id, class_type}``.
     """
+    workflow = metadata.get("workflow")
     metadata = {
         key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS
     }
@@ -496,6 +507,10 @@ def inspect_prompt_evidence(
             canonical["prompt"] = json.loads(canonical["prompt"])
         except (ValueError, RecursionError):
             return malformed
+    workflow = _parsed_workflow(workflow)
+    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)
     state = resolved["state"]
     result = {}

utils/prompt_pair_resolver.py

@@ -53,6 +53,12 @@ class OutputAdapter:
     kind: str
     compose: bool = False
     storage_input: bool = False
+    # Historical PromptManager classes (inspection only): ``comma`` strips the
+    # text and joins prepend/append with ", " exactly as their code did;
+    # ``strip`` only strips. ``default`` stands in for a missing optional input.
+    comma: bool = False
+    strip: bool = False
+    default: Optional[str] = None
 
 
 _SAMPLERS = frozenset({"KSampler", "KSamplerAdvanced", "SamplerCustom"})
@@ -76,6 +82,25 @@ for _name in ("PrimitiveString", "PrimitiveStringMultiline"):
     _DEFAULT_ADAPTERS[(_name, 0)] = OutputAdapter("value", "string")
 for _name in ("Text Multiline", "String Literal", "ShowText|pysssss"):
     _DEFAULT_ADAPTERS[(_name, 0)] = OutputAdapter("text", "string")
+# This repository's node classes up to 07a5af3 (``prompt_manager_*.py``), still in
+# saved workflows. Inspection only: the default profile keeps them unsupported.
+_COMMA = OutputAdapter("text", "conditioning", comma=True)
+_NEGATIVE = OutputAdapter("negative_text", "conditioning", strip=True, default="")
+_HISTORICAL_ADAPTERS = {
+    ("PromptManagerV2", 0): _COMMA,
+    ("PromptManagerV2", 1): _NEGATIVE,
+    ("PromptManagerV2", 2): replace(_COMMA, kind="string"),
+    ("PromptManagerV2", 3): replace(_NEGATIVE, kind="string"),
+    ("PromptManagerV2Text", 0): replace(_COMMA, kind="string"),
+    ("PromptManagerV2Text", 1): replace(_NEGATIVE, kind="string"),
+    ("PromptManagerNegative", 0): OutputAdapter("text", "conditioning", strip=True),
+    ("PromptManagerNegative", 1): OutputAdapter("text", "string", strip=True),
+    ("PromptManagerPositiveText", 0): replace(_COMMA, kind="string"),
+    ("PromptManagerNegativeText", 0): replace(_COMMA, kind="string"),
+}
+for _name in ("PromptManagerPositive", "PromptManagerDetailer"):
+    _HISTORICAL_ADAPTERS[(_name, 0)] = _COMMA
+    _HISTORICAL_ADAPTERS[(_name, 1)] = replace(_COMMA, kind="string")
 
 
 @dataclass(frozen=True)
@@ -155,6 +180,9 @@ _RUNTIME_REASON = "Text is decided at run time"
 EMPTY_POSITIVE_REASON = "Positive prompt is empty"
 MULTISTAGE_UNATTRIBUTED_REASON = "Multistage sampler ancestry is ambiguous"
 _INCOMPATIBLE_SLOT = "Output slot has an incompatible type"
+# Before 3.2.3 a save patch overwrote every PromptManager node's saved ``text``
+# with the last-run node's prompt (comfy-metadata.js ``createContext``).
+OVERWRITTEN_PROMPT_REASON = "Saved PromptManager text was overwritten by another node"
 # Trace outcomes that leave an input unverifiable against execution evidence.
 _UNVERIFIABLE_CODES = frozenset({"unsupported-node", "runtime-value"})
 # JoinStringMulti's own ``inputcount`` range; anything else never executed.
@@ -296,12 +324,16 @@ class _GraphResolver:
         budget: Optional[_Budget] = None,
         profile: ResolverProfile = DEFAULT_PROFILE,
         semantics: Optional[Any] = None,
+        overwritten: frozenset = frozenset(),
     ) -> None:
         # ``max_visits`` is the per-candidate allowance; ``profile`` supplies only
         # the path selection and whole-resolution bounds. ``resolve_text_input``
         # relies on the legacy default profile here.
         self.graph = graph
-        self.adapters = {**_DEFAULT_ADAPTERS, **adapters}
+        historical = _HISTORICAL_ADAPTERS if semantics is not None else {}
+        self.adapters = {**_DEFAULT_ADAPTERS, **historical, **adapters}
+        # Nodes whose saved text is not their own (inspection; R1).
+        self.overwritten = overwritten
         self.per_candidate_visits = max_visits
         self.budget = budget if budget is not None else _Budget(max_visits)
         self.budget.reset(max_visits)
@@ -433,6 +465,8 @@ class _GraphResolver:
             self.budget.spend()
             name, inputs = self.node(node_id)
             self.scan(len(inputs))
+            if node_id in self.overwritten:
+                raise _Unresolved(OVERWRITTEN_PROMPT_REASON, "malformed")
             self.active.add(key)
             if name == "Reroute" and slot == 0:
                 keys = [key for key in ("value", "input", "in") if key in inputs]
@@ -463,6 +497,8 @@ class _GraphResolver:
                     storage_text = evidence[adapter.input_name]
                     path = []
                     self.text_sources[(node_id, slot)] = "execution"
+                elif adapter.comma or adapter.strip:
+                    text, path, storage_text = self.historical(adapter, inputs, depth)
                 else:
                     text, path, storage_text = self.text(
                         inputs.get(adapter.input_name), "string", depth + 1
@@ -562,6 +598,40 @@ class _GraphResolver:
             self.tainted = outer_tainted or self.tainted
             self.active.discard(key)
 
+    def historical(
+        self, adapter: OutputAdapter, inputs: dict, depth: int
+    ) -> Tuple[str, list, str]:
+        """``prompt_manager_*.py`` at 07a5af3: strip, then join with ", ".
+
+        Prepend and append are stripped and skipped when empty; the separator is
+        left out when either side is empty or already ends with a comma. The
+        stripped main text is the recipe.
+        """
+        text, path, _ = self.text(
+            inputs.get(adapter.input_name, adapter.default), "string", depth + 1
+        )
+        self.scan(len(text))
+        text = text.strip()
+        storage_text = text
+        if adapter.comma:
+            before, before_path, _ = self.text(
+                inputs.get("prepend_text", ""), "string", depth + 1
+            )
+            after, after_path, _ = self.text(
+                inputs.get("append_text", ""), "string", depth + 1
+            )
+            self.scan(len(before) + len(after) + len(before_path) + len(after_path))
+            before, after = before.strip(), after.strip()
+            if before:
+                joint = ", " if text and not before.endswith(",") else ""
+                text = f"{before}{joint}{text}"
+            if after:
+                joint = ", " if text and not text.endswith(",") else ""
+                text = f"{text}{joint}{after}"
+            text = self.built(self.bounded(text))
+            path = before_path + path + after_path
+        return text, path, storage_text
+
     def follows(self, name: str, slot: int, kind: str) -> bool:
         """Whether the semantics table traces this output (inspection only)."""
         semantics = self.semantics
@@ -1506,6 +1576,9 @@ def _resolve(
             reason="Graph node IDs are not unique",
             reason_code="malformed",
         )
+    overwritten = frozenset()
+    if semantics is not None and "prompt" in metadata:
+        overwritten = _overwritten_prompt_nodes(metadata.get("workflow"), normalized)
     resolver = _GraphResolver(
         normalized,
         adapters or {},
@@ -1516,6 +1589,7 @@ def _resolve(
         budget=_Budget(profile.total_visits),
         profile=profile,
         semantics=semantics,
+        overwritten=overwritten,
     )
     if output_node_id is not None:
         try:
@@ -1545,6 +1619,54 @@ def _resolve(
     return resolver.pair(sampler_ids[0])
 
 
+def _overwritten_prompt_nodes(workflow: Any, graph: dict) -> frozenset:
+    """API node IDs whose saved PromptManager text is another node's (R1).
+
+    The pre-3.2.3 signature (comfy-metadata.js ``createContext``): two or more
+    workflow nodes of a PromptManager type other than ``PromptManagerPair``
+    share one API ``text`` while their first widget values differ. Their API
+    class may have been rewritten to ``CLIPTextEncode``, so nodes are matched by
+    ID. Unlike the legacy viewer, the widget text is never substituted: it was
+    not recorded as executed, so these nodes are refused instead.
+    """
+    nodes = workflow.get("nodes") if isinstance(workflow, dict) else None
+    if not isinstance(nodes, list):
+        return frozenset()
+    widgets: Dict[str, str] = {}
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        kind, values, node_id = (
+            node.get("type"),
+            node.get("widgets_values"),
+            node.get("id"),
+        )
+        if (
+            isinstance(kind, str)
+            and kind != "PromptManagerPair"
+            and "promptmanager" in kind.lower()
+            and isinstance(values, list)
+            and values
+            and isinstance(values[0], str)
+            and isinstance(node_id, (str, int))
+            and not isinstance(node_id, bool)
+        ):
+            widgets[str(node_id)] = values[0]
+    by_text: Dict[str, list] = {}
+    for node_id, widget in widgets.items():
+        node = graph.get(node_id)
+        inputs = node.get("inputs") if isinstance(node, dict) else None
+        text = inputs.get("text") if isinstance(inputs, dict) else None
+        if isinstance(text, str):
+            by_text.setdefault(text, []).append(node_id)
+    return frozenset(
+        node_id
+        for ids in by_text.values()
+        if len(ids) > 1 and len({widgets[node_id] for node_id in ids}) > 1
+        for node_id in ids
+    )
+
+
 def _no_sampler(graph: dict, profile: ResolverProfile) -> Dict[str, Any]:
     reason = "No supported sampler pair found"
     if profile.legacy: