fix(history): name the current file's prompt pair button apart from the frozen run's

The recorded run dialog nests the current-file inspector, whose footer repeated the "Send prompt pair" name. Name it from the prompt context, scope the real-host test to the frozen region, and add a dialog-wide duplicate-name check.

TESTING.md

@@ -227,7 +227,55 @@ workflow graph and its facts become limited). The serialized
 response stays under 8 MiB by omitting, in order, the workflow graph, then the prompt
 graph, then all raw records atomically. Retry does not raise these limits. Ingestion
 (scan, live monitor, admin import) and queue-time run counting keep the default
-profile, including its 500-node gate.
+profile, including its 500-node gate. Only `INSPECTION_PROFILE` reads the node
+semantics table `web/js/inspection-semantics.json` (at most 1 MiB and eight nesting
+levels, loaded on first use). With it, `SamplerCustomAdvanced` follows its `guider`
+link (through Reroutes) to `CFGGuider` (positive and negative) or `BasicGuider`
+(positive only; the negative is `absent`, "Guider has no negative input"). Any
+other guider, such as `DualCFGGuider`, is `unsupported-node`. Conditioning is
+traced through the table's pass-throughs (FluxGuidance, ReferenceLatent, the
+ConditioningSet* nodes, ControlNet apply nodes, InpaintModelConditioning and
+WanImageToVideo by output slot). `ConditioningZeroOut` is an explicit empty text
+with source `zeroed-conditioning`. `easy ifElse` with a literal boolean follows that
+branch only; with a linked boolean both branches are traced, equal texts resolve and
+different texts make the sampler `ambiguous` with one candidate per choice (at most
+32, else `work-limit`). ConditioningCombine/Concat/Average resolve only when every
+part has the same text, otherwise `unresolved` with `composite` naming the node.
+Results that depend on a forced choice are never memoized. A `FaceDetailer`
+(`detailer-stage`) is a sampler stage: the samplers reached through its `image`
+ancestry are compared with it. Equal texts resolve with the detailer as the sampler;
+different texts are `ambiguous` with `multistage-differs` and one candidate per
+stage; an unreadable stage makes the chain `unresolved`. Unanchored, a sampler
+whose output only feeds detailers (or previews) is not a candidate of its own.
+A sampler's latent ancestry is walked through every link input of non-sampler
+nodes within the same visit and depth budgets: with no earlier sampler it is a
+single stage whatever the origin (VAEEncode, EmptyLatentBatch, EmptyFlux2LatentImage,
+EmptyHunyuanLatentVideo, SetLatentNoiseMask, InpaintModelConditioning or
+WanImageToVideo slot 2). An earlier sampler (img2img from a decode, hires fix) stays
+`unresolved` "Multistage sampler ancestry is ambiguous" until stage texts are
+compared; an unlisted node with sampler-shaped inputs (positive, negative, guider,
+latent_image, a pipe, video embeds, custom-sampler parts, or a `model` link together
+with conditioning, neg_conditioning, samples or latent) is `unsupported-node`
+naming it, here and in a detailer's image ancestry; a latent that is not a link or
+reaches its own sampler is `malformed`; running out of budget is `work-limit` or
+`depth-limit`, never single stage. Each sampler's walk runs once per resolution;
+selector branch runs share it and other candidates are charged its visit cost.
+Text encoders: `single` encoders (the Qwen image encoders, `TextEncodeQwenImage21`
+slot 0 `prompt` and slot 1 `negative_prompt`, `KikoTextEncode`) read one string
+input unchanged. `parts-equal` encoders (CLIPTextEncodeSDXL, Flux, SD3) resolve when
+every part is identical or exactly one part is non-empty (exact `""`, whitespace is
+text); otherwise `composite`. String sources reproduce the pinned KJNodes code from
+literal settings only: `StringConstant` is its `string`; `StringConstantMultiline`
+with a literal true `strip_newlines` is `replace("\n", "").strip()`;
+`JoinStringMulti` keeps `string_1` (even empty), appends `string_2`..`string_N`
+(`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
+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.
 
 `tools/inspection_corpus.py --root DIR --out JSONL [--workers N] [--baseline JSONL]
 [--summary TXT] [--code TREE]` measures a media folder read-only against the #240

py/api/inspection_service.py

@@ -9,6 +9,11 @@ try:
         MetadataReadError,
         inspect_prompt_evidence,
         read_inspection_file,
+        unreadable_evidence,
+    )
+    from ...utils.prompt_pair_resolver import (
+        INSPECTION_PROFILE,
+        PREVIEW_INSPECTION_PROFILE,
     )
     from ...utils.video_metadata import is_video_path
 except ImportError:
@@ -16,6 +21,11 @@ except ImportError:
         MetadataReadError,
         inspect_prompt_evidence,
         read_inspection_file,
+        unreadable_evidence,
+    )
+    from utils.prompt_pair_resolver import (
+        INSPECTION_PROFILE,
+        PREVIEW_INSPECTION_PROFILE,
     )
     from utils.video_metadata import is_video_path
 
@@ -60,6 +70,28 @@ def _current_file_facts(path: Path) -> os.stat_result:
     return observed
 
 
+def _in_host_temp(path: Path) -> bool:
+    """Whether a file lies in ComfyUI's temp folder, where previews are written.
+
+    An unknown temp folder reads as no: the file is then inspected as saved
+    output.
+    """
+    try:
+        import folder_paths
+
+        temp = folder_paths.get_temp_directory()
+    except (ImportError, AttributeError, OSError):
+        return False
+    if type(temp) is not str or not temp:
+        return False
+    try:
+        folder = os.path.normcase(os.path.realpath(temp))
+        target = os.path.normcase(os.path.realpath(path))
+        return os.path.commonpath([folder, target]) == folder
+    except (OSError, ValueError):
+        return False
+
+
 def inspect_current_file(
     path: Path, *, source_label: str, saved: dict | None, library: dict
 ) -> dict:
@@ -73,24 +105,18 @@ def inspect_current_file(
     if any(getattr(before, key) != getattr(after, key) for key in fields):
         raise CurrentFileChanged
     if inspection is None:
-        evidence = {
-            "prompts": {
-                role: {"status": "unresolved", "text": None, "source": None}
-                for role in ("positive", "negative")
-            },
-            "resolution": {
-                "state": "unresolved",
-                "sampler_id": None,
-                "reason": "Media metadata could not be read",
-                "reason_code": None,
-                "node": None,
-            },
-        }
+        evidence = unreadable_evidence()
     else:
         role_saved = (
             None if saved and saved.get("association_status") == "ambiguous" else saved
         )
-        evidence = inspect_prompt_evidence(inspection.metadata, role_saved)
+        # A preview may be the output of a stage that reaches only previews.
+        profile = (
+            PREVIEW_INSPECTION_PROFILE if _in_host_temp(path) else INSPECTION_PROFILE
+        )
+        evidence = inspect_prompt_evidence(
+            inspection.metadata, role_saved, profile=profile
+        )
     return {
         "file": {
             "observation": "current-file",

tests/browser/full-inspector.spec.mjs

@@ -26,7 +26,10 @@ test('inspection facts survive actual API to browser transport', async ({ page,
     const complete = result.complete;
     expect(Object.values(complete.generation).map(fact => fact.value)).toEqual(['9007199254740993', '28', '6.5', 'dpmpp_2m', 'karras', 'fixture-model.safetensors']);
     expect(Object.values(complete.generation).every(fact => fact.source === 'api-graph' && fact.evidence === 'configuration')).toBe(true);
-    expect(Object.values(result.ambiguous.generation).every(fact => fact.status === 'ambiguous' && fact.value === null)).toBe(true);
+    // Ambiguous branches merge per field: the differing seed is ambiguous, the shared fields known.
+    expect(Object.values(result.ambiguous.generation).map(fact => [fact.status, fact.value])).toEqual([
+        ['ambiguous', null], ['known', '28'], ['known', '6.5'], ['known', 'dpmpp_2m'], ['known', 'karras'], ['known', 'fixture-model.safetensors']]);
+    expect(Object.values(result.ambiguous.generation).every(fact => fact.sampler_id === null)).toBe(true);
     expect(result['workflow-only'].generation.seed).toMatchObject({ status: 'known', value: '9007199254740993', source: 'workflow', evidence: 'configuration' });
     expect(result['workflow-only'].prompts.positive.source).toBe('saved-association');
     expect(Object.values(result['control-heavy'].generation).every(fact => fact.status === 'unresolved' && fact.value === null && fact.reason === 'projection-limited')).toBe(true);

tests/browser/prompt-inspector.spec.mjs

@@ -232,6 +232,17 @@ test('unverified saved associations explain uncertainty and cannot be sent',asyn
     await expect(details(page).getByRole('button',{name:/^Send /})).toHaveCount(0);
 });
 
+test('unresolved roles name the blocking node as text and offer no prompt actions',async({page})=>{
+    const line='Prompt uses a node the inspector does not support (<img src=x onerror="window.injected=1">).';
+    await page.route('**/images/inspect/uncertain.png?root=0*',route=>route.fulfill({json:{success:true,root_scope:new URL(route.request().url()).searchParams.get('root_scope'),image:{root_index:0,path:'uncertain.png',filename:'uncertain.png'},prompts:{
+        positive:{status:'unresolved',text:null,source:null,message:line},negative:{status:'unresolved',text:null,source:null,message:line},
+    }}}));
+    await page.goto(workspace);await select(page,'0:uncertain.png');
+    for (const name of ['Positive','Negative']) await expect(role(details(page),name)).toContainText(line);
+    await expect(details(page).locator('img[onerror]')).toHaveCount(0);
+    expect(await page.evaluate(()=>window.injected)).toBeUndefined();
+    await expect(details(page).getByRole('button',{name:/^(Send|Copy) /})).toHaveCount(0);
+});
 
 test('prompt role headings contain compact real actions and honest source evidence',async({page})=>{
     await page.goto(workspace);await select(page,'0:shared/same.png');

tests/browser/workspace-history.spec.mjs

@@ -458,6 +458,46 @@ test('frozen Effective pair uses the shared acknowledged sender and Escape close
     } finally {await host.evaluate(()=>window.historyHost.dispose());await host.close();}
 });
 
+test('the inline current file names its prompts apart from the frozen run fields',async({page})=>{
+    // The current file resolves its own prompts, as a real host's saved PNG does, so the dialog holds two prompt texts per role.
+    await page.route(url=>url.pathname.endsWith('/inspection'),async route=>{const response=await route.fetch();const body=await response.json();
+        body.prompts={positive:{status:'resolved',text:'current file positive',source:'graph-inferred'},negative:{status:'resolved',text:'current file negative',source:'graph-inferred'}};
+        await route.fulfill({response,json:body});});
+    await open(page,page.fixture.detail_id);await marker(page,page.fixture.runs.shared).click();await expect(dialog(page).locator('[data-receipt]')).toHaveCount(5);
+    await dialog(page).getByRole('button',{name:'Raw',exact:true}).click();
+    const frozen=dialog(page).getByRole('region',{name:'Frozen run prompt texts',exact:true});
+    const before=await exact(frozen.getByRole('region',{name:'Positive prompt',exact:true}).locator('pre'));
+    await dialog(page).locator('[data-receipt]').first().getByRole('button',{name:'View current file',exact:true}).click();
+    await expect(page.getByRole('dialog',{name:'Image preview',exact:true})).toBeVisible();await page.keyboard.press('Escape');await expect(dialog(page)).toBeVisible();
+    const current=dialog(page).getByRole('region',{name:'Current reported file',exact:true});await expect(current).toBeVisible();
+    await expect(current.getByRole('region',{name:'Current file positive prompt',exact:true}).locator('pre')).toHaveText('current file positive');
+    await expect(current.getByRole('region',{name:'Current file negative prompt',exact:true}).locator('pre')).toHaveText('current file negative');
+    await expect(current.getByRole('button',{name:'Copy current file positive prompt',exact:true})).toBeVisible();
+    for(const role of ['Positive','Negative'])await expect(dialog(page).getByRole('region',{name:`${role} prompt`,exact:true})).toHaveCount(1);
+    await expect(current.getByRole('region',{name:/^(Positive|Negative) prompt$/})).toHaveCount(0);
+    expect(await exact(dialog(page).getByRole('region',{name:'Positive prompt',exact:true}).locator('pre'))).toBe(before);
+});
+
+test('the recorded run dialog with a resolved current file has no duplicate button or region names',async({page})=>{
+    await page.route(url=>url.pathname.endsWith('/inspection'),async route=>{const response=await route.fetch();const body=await response.json();
+        body.prompts={positive:{status:'resolved',text:'current file positive',source:'graph-inferred'},negative:{status:'resolved',text:'current file negative',source:'graph-inferred'}};
+        await route.fulfill({response,json:body});});
+    await open(page,page.fixture.detail_id);await marker(page,page.fixture.runs.shared).click();await expect(dialog(page).locator('[data-receipt]')).toHaveCount(5);
+    await dialog(page).getByRole('button',{name:'Raw',exact:true}).click();
+    await dialog(page).locator('[data-receipt]').first().getByRole('button',{name:'View current file',exact:true}).click();
+    await expect(page.getByRole('dialog',{name:'Image preview',exact:true})).toBeVisible();await page.keyboard.press('Escape');
+    await expect(dialog(page).getByRole('region',{name:'Current reported file',exact:true})).toBeVisible();
+    await expect(dialog(page).getByRole('region',{name:'Current file positive prompt',exact:true})).toBeVisible();
+    // Receipt rows repeat per-row actions by design; every other control and region in the dialog must be uniquely named.
+    const snapshot=await dialog(page).ariaSnapshot();const counts=new Map();
+    for(const line of snapshot.split('\n')){const match=/^\s*- (button|region|link) "((?:[^"\\]|\\.)*)"/.exec(line);if(!match)continue;
+        const key=`${match[1]} ${match[2]}`;counts.set(key,(counts.get(key)||0)+1);}
+    expect(counts.size).toBeGreaterThan(10);
+    const receiptNames=await dialog(page).locator('[data-receipt]').first().evaluate(row=>[...row.querySelectorAll('button,a,[role="region"]')].map(node=>node.getAttribute('aria-label')||node.textContent.trim()));
+    const repeated=[...counts].filter(([key,n])=>n>1&&!receiptNames.some(name=>key.endsWith(` ${name}`)));
+    expect(repeated).toEqual([]);
+});
+
 test('pagehide tears down dialogs and persisted reentry revalidates before enabling recorded actions',async({page})=>{
     await open(page,page.fixture.detail_id);await marker(page,page.fixture.runs.shared).click();await expect(dialog(page).locator('[data-receipt]')).toHaveCount(5);
     await dialog(page).locator('[data-receipt]').first().getByRole('button',{name:'View current file',exact:true}).click();await expect(page.getByRole('dialog',{name:'Image preview',exact:true})).toBeVisible();

tests/browser/workspace_inspector_fixture.py

@@ -89,7 +89,9 @@ def seed_workspace_inspector(
         ],
         "links": [],
     }
-    ambiguous = {**complete, "5": complete["3"]}
+    # Two branches that differ only in seed: facts merge per field (spec section 10).
+    branch = {**complete["3"], "inputs": {**complete["3"]["inputs"], "seed": 7}}
+    ambiguous = {**complete, "5": branch}
     examples = (
         ("complete.png", {"prompt": complete, "workflow": workflow}),
         ("empty.png", {}),

tests/comfyui/execution-history.spec.mjs

@@ -243,10 +243,10 @@ test.describe('public chronological history with the actual acknowledged host br
                 const frozen = attempts[0].context;
                 for (const mode of ['Raw', 'Effective']) {
                     await run.getByRole('button', { name: mode, exact: true }).click();
-                    for (const [role, value] of [['Positive', frozen.positive[mode.toLowerCase()]], ['Negative', frozen.negative[mode.toLowerCase()]]]) expect(await run.getByRole('region', { name: `${role} prompt`, exact: true }).locator('pre').evaluate(node => node.textContent)).toBe(value);
+                    for (const [role, value] of [['Positive', frozen.positive[mode.toLowerCase()]], ['Negative', frozen.negative[mode.toLowerCase()]]]) expect(await run.getByRole('region', { name: 'Frozen run prompt texts', exact: true }).getByRole('region', { name: `${role} prompt`, exact: true }).locator('pre').evaluate(node => node.textContent)).toBe(value);
                     await page.evaluate(async id => {const { app } = await import('/scripts/app.js');const node = app.graph.getNodeById(id);node.widgets.find(w => w.name === 'text').value = 'Before send';window.chronologyEffects = { queue: 0, load: 0 };window.chronologyOriginals = {};for (const [method, key] of [['queuePrompt', 'queue'], ['loadGraphData', 'load']]) {const original = app[method];window.chronologyOriginals[method] = original;app[method] = function(...args) {window.chronologyEffects[key]++;return original.apply(this, args);};}}, destination);
                     try {
-                        await run.getByRole('button', { name: 'Send prompt pair', exact: true }).click();const sender = workspace.getByRole('dialog', { name: 'Send to ComfyUI', exact: true });
+                        await run.getByRole('region', { name: 'Frozen run prompt texts', exact: true }).getByRole('button', { name: 'Send prompt pair', exact: true }).click();const sender = workspace.getByRole('dialog', { name: 'Send to ComfyUI', exact: true });
                         const hosts = sender.getByRole('combobox', { name: 'ComfyUI window', exact: true });await expect(hosts.locator('option')).toHaveCount(2);await hosts.selectOption({ index: 1 });
                         await sender.getByRole('combobox', { name: 'Destination node', exact: true }).selectOption({ index: 1 });await sender.getByRole('button', { name: 'Preview changes', exact: true }).click();
                         await sender.getByRole('button', { name: 'Apply to ComfyUI', exact: true }).click();await expect(sender).toContainText('Sent to ComfyUI.');

tests/fixtures/README.md

@@ -15,7 +15,7 @@ Browser tests use the committed file; FFmpeg is not required to run them.
 
 # Inspection parity fixtures
 
-`inspection_parity/` holds small synthetic PNGs (`f01b.png` ... `f17.png`, each
+`inspection_parity/` holds small synthetic PNGs (`f01b.png` ... `f24.png`, each
 under 32 KiB) and `vectors.json`, used by the #240 inspection parity tests. Every
 graph, prompt text, model name and filename prefix in them is original synthetic
 content written for this repository. Nothing comes from user output. They are

tests/fixtures/inspection_parity/vectors.json

@@ -6,9 +6,9 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "No supported sampler pair found (saw FaceDetailer, SamplerCustomAdvanced)",
-      "reason_code": "no-sampler",
+      "state": "ambiguous",
+      "reason": "Multiple sampler branches",
+      "reason_code": null,
       "selection": null,
       "positive": null,
       "negative": null,
@@ -19,24 +19,24 @@
           "value": null
         },
         "steps": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "20"
         },
         "cfg": {
-          "status": "ambiguous",
+          "status": "unresolved",
           "value": null
         },
         "sampler": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "euler"
         },
         "scheduler": {
           "status": "ambiguous",
           "value": null
         },
         "model": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "flux-2-klein-9b-fp8.safetensors"
         }
       }
     }
@@ -48,12 +48,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "No supported sampler pair found (saw FaceDetailer, SamplerCustomAdvanced)",
-      "reason_code": "no-sampler",
+      "state": "absent",
+      "reason": "Guider has no negative input",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur, Scene 0: unique shot description soft winter light,",
+      "negative": "<absent>",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -61,8 +61,8 @@
           "value": "1000"
         },
         "steps": {
-          "status": "unresolved",
-          "value": null
+          "status": "known",
+          "value": "20"
         },
         "cfg": {
           "status": "not-recorded",
@@ -73,8 +73,8 @@
           "value": "euler"
         },
         "scheduler": {
-          "status": "unresolved",
-          "value": null
+          "status": "known",
+          "value": "Flux2Scheduler"
         },
         "model": {
           "status": "known",
@@ -258,12 +258,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported node semantics or output slot",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -300,12 +300,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported node semantics or output slot",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -342,12 +342,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported node semantics or output slot",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -385,8 +385,8 @@
     "expect": {
       "metadata_status": "available",
       "state": "unresolved",
-      "reason": "Unsupported node semantics or output slot",
-      "reason_code": "unsupported-node",
+      "reason": "Combined conditioning has different texts",
+      "reason_code": "composite",
       "selection": null,
       "positive": null,
       "negative": null,
@@ -426,12 +426,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported node semantics or output slot",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -510,12 +510,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "No supported sampler pair found (saw SamplerCustomAdvanced, CFGGuider)",
-      "reason_code": "no-sampler",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -603,28 +603,28 @@
       "candidate_count": 0,
       "facts": {
         "seed": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "42"
         },
         "steps": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "25"
         },
         "cfg": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "6.5"
         },
         "sampler": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "euler"
         },
         "scheduler": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "normal"
         },
         "model": {
-          "status": "ambiguous",
-          "value": null
+          "status": "known",
+          "value": "sdxl_base.safetensors"
         }
       }
     }
@@ -636,12 +636,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported latent ancestry",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -678,12 +678,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "Unsupported latent ancestry",
-      "reason_code": "unsupported-node",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -888,12 +888,12 @@
     "anchor": null,
     "expect": {
       "metadata_status": "available",
-      "state": "unresolved",
-      "reason": "No supported sampler pair found (saw SamplerCustomAdvanced)",
-      "reason_code": "no-sampler",
+      "state": "absent",
+      "reason": "Guider has no negative input",
+      "reason_code": null,
       "selection": null,
-      "positive": null,
-      "negative": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "<absent>",
       "candidate_count": 0,
       "facts": {
         "seed": {
@@ -923,6 +923,258 @@
       }
     }
   },
+  {
+    "name": "f19-inspection",
+    "fixture": "f19",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "ambiguous",
+      "reason": "Sampler stages have different prompts",
+      "reason_code": "multistage-differs",
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "ambiguous",
+          "value": null
+        },
+        "steps": {
+          "status": "ambiguous",
+          "value": null
+        },
+        "cfg": {
+          "status": "ambiguous",
+          "value": null
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
+  {
+    "name": "f20-inspection",
+    "fixture": "f20",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
+      "selection": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "known",
+          "value": "42"
+        },
+        "steps": {
+          "status": "known",
+          "value": "25"
+        },
+        "cfg": {
+          "status": "known",
+          "value": "6.5"
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
+  {
+    "name": "f21-inspection",
+    "fixture": "f21",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Text is decided at run time",
+      "reason_code": "runtime-value",
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "known",
+          "value": "42"
+        },
+        "steps": {
+          "status": "known",
+          "value": "25"
+        },
+        "cfg": {
+          "status": "known",
+          "value": "6.5"
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
+  {
+    "name": "f22-inspection",
+    "fixture": "f22",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Text is decided at run time",
+      "reason_code": "runtime-value",
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "known",
+          "value": "42"
+        },
+        "steps": {
+          "status": "known",
+          "value": "25"
+        },
+        "cfg": {
+          "status": "known",
+          "value": "6.5"
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
+  {
+    "name": "f23-inspection",
+    "fixture": "f23",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
+      "selection": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "",
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "known",
+          "value": "42"
+        },
+        "steps": {
+          "status": "known",
+          "value": "25"
+        },
+        "cfg": {
+          "status": "known",
+          "value": "6.5"
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
+  {
+    "name": "f24-inspection",
+    "fixture": "f24",
+    "profile": "inspection",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
+      "selection": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
+      "candidate_count": 0,
+      "facts": {
+        "seed": {
+          "status": "known",
+          "value": "42"
+        },
+        "steps": {
+          "status": "known",
+          "value": "25"
+        },
+        "cfg": {
+          "status": "known",
+          "value": "6.5"
+        },
+        "sampler": {
+          "status": "known",
+          "value": "euler"
+        },
+        "scheduler": {
+          "status": "known",
+          "value": "normal"
+        },
+        "model": {
+          "status": "known",
+          "value": "sdxl_base.safetensors"
+        }
+      }
+    }
+  },
   {
     "name": "f01-default",
     "fixture": "f01",
@@ -1296,5 +1548,107 @@
       "candidate_count": 0,
       "facts": null
     }
+  },
+  {
+    "name": "f19-default",
+    "fixture": "f19",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "resolved",
+      "reason": "",
+      "reason_code": null,
+      "selection": null,
+      "positive": "a red fox in snow, detailed fur",
+      "negative": "blurry, lowres",
+      "candidate_count": 0,
+      "facts": null
+    }
+  },
+  {
+    "name": "f20-default",
+    "fixture": "f20",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Unsupported node semantics or output slot",
+      "reason_code": null,
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": null
+    }
+  },
+  {
+    "name": "f21-default",
+    "fixture": "f21",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Unsupported node semantics or output slot",
+      "reason_code": null,
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": null
+    }
+  },
+  {
+    "name": "f22-default",
+    "fixture": "f22",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Unsupported node semantics or output slot",
+      "reason_code": null,
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": null
+    }
+  },
+  {
+    "name": "f23-default",
+    "fixture": "f23",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Unsupported node semantics or output slot",
+      "reason_code": null,
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": null
+    }
+  },
+  {
+    "name": "f24-default",
+    "fixture": "f24",
+    "profile": "default",
+    "anchor": null,
+    "expect": {
+      "metadata_status": "available",
+      "state": "unresolved",
+      "reason": "Unsupported node semantics or output slot",
+      "reason_code": null,
+      "selection": null,
+      "positive": null,
+      "negative": null,
+      "candidate_count": 0,
+      "facts": null
+    }
   }
 ]

tests/inspection_parity.py

@@ -2,7 +2,7 @@
 
 Every graph here is original synthetic content (CC0): neutral texts, generic model
 file names and invented filename prefixes. Nothing is copied from user outputs.
-The shapes reproduce the parity audit's fixture numbering (``f01``..``f17``).
+The shapes reproduce the parity audit's fixture numbering (``f01``..``f24``).
 
 ``build(name)`` returns ``{"prompt", "workflow", "extra_chunks"}``. ``write_png``
 writes the two records the way core ``SaveImage`` does (Pillow ``PngInfo.add_text``,
@@ -66,6 +66,12 @@ COMMITTED = (
     "f15",
     "f16",
     "f17",
+    "f19",
+    "f20",
+    "f21",
+    "f22",
+    "f23",
+    "f24",
 )
 # Built in a temporary directory at test time (too large to commit).
 TEST_TIME = ("f01", "f04", "f05")
@@ -100,6 +106,10 @@ def _enc(text):
     return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["4", 1]}}
 
 
+def _linked_enc(link):
+    return {"class_type": "CLIPTextEncode", "inputs": {"text": link, "clip": ["4", 1]}}
+
+
 def _ksampler_graph(positive, negative, sampler="KSampler", latent=None, extra=None):
     graph = {
         "4": {
@@ -691,6 +701,105 @@ def _f17():
     return _basic(graph, _ui(graph))
 
 
+DETAILER_POS = "a red fox in snow, sharp detailed face"
+
+
+def _f19():
+    """KSampler -> VAEDecode -> FaceDetailer -> SaveImage; the detailer re-prompts."""
+    graph = _ksampler_graph(_enc(POS), _enc(NEG))
+    graph["10"] = _enc(DETAILER_POS)
+    graph["50"] = {
+        "class_type": "FaceDetailer",
+        "inputs": {
+            "image": ["8", 0],
+            "model": ["4", 0],
+            "clip": ["4", 1],
+            "vae": ["4", 2],
+            "positive": ["10", 0],
+            "negative": ["7", 0],
+            "seed": 7,
+            "steps": 12,
+            "cfg": 4.5,
+            "sampler_name": "euler",
+            "scheduler": "normal",
+            "denoise": 0.4,
+        },
+    }
+    graph["9"]["inputs"]["images"] = ["50", 0]
+    return _basic(graph)
+
+
+def _f20():
+    """``easy ifElse`` with a literal boolean picks the positive branch."""
+    select = {
+        "class_type": "easy ifElse",
+        "inputs": {"boolean": True, "on_true": ["10", 0], "on_false": ["11", 0]},
+    }
+    extra = {"10": _enc(POS), "11": _enc("a grey wolf in fog")}
+    return _basic(_ksampler_graph(select, _enc(NEG), extra=extra))
+
+
+def _f21():
+    """``JoinStringMulti`` with ``return_list: true`` feeds the positive encoder."""
+    join = {
+        "class_type": "JoinStringMulti",
+        "inputs": {
+            "inputcount": 2,
+            "delimiter": ", ",
+            "return_list": True,
+            "string_1": ["11", 0],
+            "string_2": ["12", 0],
+        },
+    }
+    extra = {
+        "10": join,
+        "11": {"class_type": "PrimitiveString", "inputs": {"value": "a red fox"}},
+        "12": {"class_type": "PrimitiveString", "inputs": {"value": "in snow"}},
+    }
+    positive = _linked_enc(["10", 0])
+    return _basic(_ksampler_graph(positive, _enc(NEG), extra=extra))
+
+
+def _f22():
+    """``PromptSearchList`` (database search at run time) feeds the positive."""
+    search = {
+        "class_type": "PromptSearchList",
+        "inputs": {
+            "text": "fox",
+            "category": "",
+            "tags": "",
+            "min_rating": 0,
+            "limit": 50,
+        },
+    }
+    positive = _linked_enc(["10", 0])
+    return _basic(_ksampler_graph(positive, _enc(NEG), extra={"10": search}))
+
+
+def _f24():
+    """``TextEncodeQwenImage21``: slot 0 is the positive, slot 1 the negative."""
+    qwen = {
+        "class_type": "TextEncodeQwenImage21",
+        "inputs": {
+            "clip": ["4", 1],
+            "prompt": POS,
+            "negative_prompt": NEG,
+            "vae": ["4", 2],
+            "resolution": 1024,
+        },
+    }
+    graph = _ksampler_graph(qwen, None)
+    del graph["7"]
+    graph["3"]["inputs"]["negative"] = ["6", 1]
+    return _basic(graph)
+
+
+def _f23():
+    """The negative is ``ConditioningZeroOut`` of the positive encoder."""
+    zero = {"class_type": "ConditioningZeroOut", "inputs": {"conditioning": ["6", 0]}}
+    return _basic(_ksampler_graph(_enc(POS), zero))
+
+
 _BUILDERS = {
     "f01": _f01,
     "f01b": _f01b,
@@ -714,6 +823,12 @@ _BUILDERS = {
     "f15": _f15,
     "f16": _f16,
     "f17": _f17,
+    "f19": _f19,
+    "f20": _f20,
+    "f21": _f21,
+    "f22": _f22,
+    "f23": _f23,
+    "f24": _f24,
 }
 
 

tests/js/comfy-metadata.test.js

@@ -164,6 +164,232 @@ test("facts require model output ancestry with supported loaders and passthrough
     assert.equal(factsFor(numericModel).model.status, "unresolved", "model names require string evidence");
 });
 
+// D19: a FaceDetailer re-samples the KSampler's decoded image with its own settings.
+function detailerFactGraph(detailer = {}) {
+    const graph = factGraph({ seed: "42", steps: 25, cfg: "6.5", sampler_name: "euler", scheduler: "normal" });
+    graph.decode = { class_type: "VAEDecode", inputs: { samples: ["s", 0], vae: ["model", 2] } };
+    graph.d = { class_type: "FaceDetailer", inputs: { image: ["decode", 0], model: ["model", 0], seed: "7", steps: 12, cfg: "6.5", sampler_name: "euler", scheduler: "normal", denoise: "0.4", ...detailer } };
+    return graph;
+}
+const statusValues = facts => Object.fromEntries(factFields.map(field => [field, [facts[field].status, facts[field].value]]));
+
+test("facts use resolution.sampler_id for detailer-stage nodes", () => {
+    // Resolved (the stages' prompts agree): the final stage the server named supplies every fact.
+    const resolved = factsFor(detailerFactGraph(), { resolution: { state: "resolved", sampler_id: "d" } });
+    assert.deepEqual(statusValues(resolved), {
+        seed: ["known", "7"], steps: ["known", "12"], cfg: ["known", "6.5"],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    });
+    assert.ok(factFields.every(field => resolved[field].sampler_id === "d" && resolved[field].evidence === "configuration"));
+    // The upstream KSampler is also accepted when it is the one resolved.
+    assert.equal(factsFor(detailerFactGraph(), { resolution: { state: "resolved", sampler_id: "s" } }).seed.value, "42");
+    // Ambiguous stages: merged per field, never the detailer's or the upstream value alone.
+    const ambiguous = factsFor(detailerFactGraph(), { resolution: { state: "ambiguous", sampler_id: "d", reason_code: "multistage-differs" } });
+    assert.deepEqual(statusValues(ambiguous), {
+        seed: ["ambiguous", null], steps: ["ambiguous", null], cfg: ["known", "6.5"],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    });
+    assert.ok(factFields.every(field => ambiguous[field].sampler_id === null), "a merged fact names no single sampler");
+    // A field one stage cannot read and the others agree on stays unresolved; not-recorded is not agreement.
+    const unreadable = factsFor(detailerFactGraph({ cfg: ["missing", 0] }), { resolution: { state: "ambiguous", sampler_id: "d" } });
+    assert.deepEqual([unreadable.cfg.status, unreadable.cfg.value], ["unresolved", null]);
+    const unrecorded = detailerFactGraph(); delete unrecorded.d.inputs.scheduler;
+    assert.equal(factsFor(unrecorded, { resolution: { state: "ambiguous", sampler_id: "d" } }).scheduler.status, "unresolved");
+    // Two detailer stages over one KSampler: three stages merge.
+    const chained = detailerFactGraph({ seed: "42" });
+    chained.d2 = { class_type: "FaceDetailer", inputs: { image: ["d", 0], model: ["model", 0], seed: "42", steps: 25, cfg: "3", sampler_name: "euler", scheduler: "normal" } };
+    assert.deepEqual(statusValues(factsFor(chained, { resolution: { state: "ambiguous", sampler_id: "d2" } })), {
+        seed: ["known", "42"], steps: ["ambiguous", null], cfg: ["ambiguous", null],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    });
+    // Without a resolution, two samplers stay ambiguous (today's rules, now over the samplers table).
+    assert.ok(Object.values(factsFor(detailerFactGraph())).every(fact => fact.status === "ambiguous"));
+    // A sampler_id the graph or the samplers table does not hold never falls back to a guess.
+    for (const sampler_id of ["missing", "decode", "model", "__proto__"]) {
+        for (const state of ["resolved", "ambiguous"]) {
+            const facts = factsFor(detailerFactGraph(), { resolution: { state, sampler_id } });
+            assert.ok(Object.values(facts).every(fact => fact.status === "unresolved" && fact.value === null), `${state} ${sampler_id}`);
+        }
+    }
+    // Stage ancestry that cannot be walked is unresolved, not a partial merge.
+    const cyclic = detailerFactGraph(); cyclic.decode.inputs.samples = ["d", 0];
+    const unlinked = detailerFactGraph({ image: "not a link" });
+    for (const graph of [cyclic, unlinked]) {
+        assert.ok(Object.values(factsFor(graph, { resolution: { state: "ambiguous", sampler_id: "d" } })).every(fact => fact.status === "unresolved"));
+    }
+    const wide = detailerFactGraph();
+    let previous = ["s", 0];
+    for (let index = 0; index < 600; index++) { wide[`hop${index}`] = { class_type: "ImageScale", inputs: { image: previous } }; previous = [`hop${index}`, 0]; }
+    wide.decode.inputs.samples = ["s", 0]; wide.d.inputs.image = previous;
+    assert.ok(Object.values(factsFor(wide, { resolution: { state: "ambiguous", sampler_id: "d" } })).every(fact => fact.status === "unresolved"), "the stage walk is bounded by MAX_VISITS");
+});
+
+test("ambiguous results without a sampler_id merge facts across every listed stage", () => {
+    const ambiguous = { resolution: { state: "ambiguous", sampler_id: null, reason: "Multiple sampler branches" } };
+    // f01b's shape: an upstream sampler and a detailer whose settings partly differ.
+    assert.deepEqual(statusValues(factsFor(detailerFactGraph(), ambiguous)), {
+        seed: ["ambiguous", null], steps: ["ambiguous", null], cfg: ["known", "6.5"],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    });
+    assert.ok(Object.values(factsFor(detailerFactGraph(), ambiguous)).every(fact => fact.sampler_id === null));
+    // Equal independent branches agree on every field once the server says ambiguous;
+    // without a resolution they stay ambiguous (today's rules, reversed in PR3).
+    const equal = factGraph(); equal.other = structuredClone(equal.s);
+    assert.ok(Object.values(factsFor(equal, ambiguous)).every(fact => fact.status === "known"));
+    assert.ok(Object.values(factsFor(equal)).every(fact => fact.status === "ambiguous"));
+    // A walk that fails keeps every field unresolved; so does a graph with no listed sampler.
+    const cyclic = detailerFactGraph(); cyclic.decode.inputs.samples = ["d", 0];
+    const broken = detailerFactGraph(); broken.d.inputs.image = ["gone", 0];
+    for (const graph of [cyclic, broken, { model: factGraph().model }]) {
+        assert.ok(Object.values(factsFor(graph, ambiguous)).every(fact => fact.status === "unresolved" && fact.value === null));
+    }
+    const many = factGraph();
+    for (let index = 0; index < 40; index++) many[`s${index}`] = structuredClone(many.s);
+    assert.ok(Object.values(factsFor(many, ambiguous)).every(fact => fact.status === "unresolved"), "more stages than the candidate cap");
+});
+
+test("stage walks fail on unlisted sampler-shaped nodes like the resolver", () => {
+    const shaped = [
+        { positive: ["model", 0] }, { latent_image: ["model", 0] }, { guider: ["model", 0] }, { negative: ["model", 0] },
+        { pipe: ["model", 0] }, { basic_pipe: ["model", 0] }, { detailer_pipe: ["model", 0] }, { sampler: ["model", 0] },
+        { sigmas: ["model", 0] }, { noise: ["model", 0] }, { text_embeds: ["model", 0] }, { hyvid_embeds: ["model", 0] },
+        { model: ["model", 0], conditioning: ["model", 0] }, { model: ["model", 0], neg_conditioning: ["model", 0] },
+        { model: ["model", 0], samples: ["model", 0] }, { model: ["model", 0], latent: ["model", 0] },
+    ];
+    for (const inputs of shaped) {
+        const graph = detailerFactGraph();
+        graph.efficient = { class_type: "MysterySampler", inputs: { ...inputs, image: ["decode", 0] } };
+        graph.d.inputs.image = ["efficient", 0];
+        for (const sampler_id of ["d", null]) {
+            const facts = factsFor(graph, { resolution: { state: "ambiguous", sampler_id } });
+            assert.ok(Object.values(facts).every(fact => fact.status === "unresolved"), `${JSON.stringify(inputs)} ${sampler_id}`);
+        }
+    }
+    // Model patches (model alone, or with clip) and listed table classes are walked through.
+    for (const [class_type, inputs] of [["ModelPatch", { model: ["model", 0] }], ["LoraPatch", { model: ["model", 0], clip: ["model", 1] }], ["ImageScale", { samples: ["model", 0] }], ["CFGGuider", { model: ["model", 0], positive: ["model", 0] }]]) {
+        const graph = detailerFactGraph();
+        graph.hop = { class_type, inputs: { ...inputs, image: ["decode", 0] } };
+        graph.d.inputs.image = ["hop", 0];
+        assert.equal(factsFor(graph, { resolution: { state: "ambiguous", sampler_id: "d" } }).seed.status, "ambiguous", class_type);
+    }
+});
+
+test("unresolved and absent results over several stages merge facts per field", () => {
+    // Review I-1: the run's settings are facts even when its prompt is not, e.g. an
+    // upstream prompt behind an unsupported node (LoRAfromText) leaves the server unresolved.
+    const merged = {
+        seed: ["ambiguous", null], steps: ["ambiguous", null], cfg: ["known", "6.5"],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    };
+    for (const resolution of [
+        { state: "unresolved", sampler_id: "d", reason_code: "unsupported-node" },
+        { state: "unresolved", sampler_id: null, reason_code: "malformed" },
+        { state: "absent", sampler_id: "d" },
+    ]) {
+        const facts = factsFor(detailerFactGraph(), { resolution });
+        assert.deepEqual(statusValues(facts), merged, JSON.stringify(resolution));
+        assert.ok(factFields.every(field => facts[field].sampler_id === null));
+    }
+    // A stage missing a field leaves it unresolved; a failed walk leaves everything unresolved.
+    const unrecorded = detailerFactGraph(); delete unrecorded.d.inputs.scheduler;
+    assert.equal(factsFor(unrecorded, { resolution: { state: "unresolved", sampler_id: "d" } }).scheduler.status, "unresolved");
+    const cyclic = detailerFactGraph(); cyclic.decode.inputs.samples = ["d", 0];
+    assert.ok(Object.values(factsFor(cyclic, { resolution: { state: "unresolved", sampler_id: "d" } })).every(fact => fact.status === "unresolved"));
+    // One listed sampler keeps today's rule: its own facts, attributed to it.
+    const single = factsFor(factGraph(), { resolution: { state: "unresolved", sampler_id: "s" } });
+    assert.ok(factFields.every(field => single[field].status === "known" && single[field].sampler_id === "s"));
+});
+
+test("a named sampler limits absent and unresolved facts to its own stages", () => {
+    // Codex P2: the server dropped a preview-only detailer and named the saved sampler, so
+    // the detailer's settings never leak into the saved file's facts.
+    const graph = detailerFactGraph();
+    graph.save = { class_type: "SaveImage", inputs: { images: ["decode", 0] } };
+    graph.preview = { class_type: "PreviewImage", inputs: { images: ["d", 0] } };
+    const own = {
+        seed: ["known", "42"], steps: ["known", "25"], cfg: ["known", "6.5"],
+        sampler: ["known", "euler"], scheduler: ["known", "normal"], model: ["known", "model.safetensors"],
+    };
+    for (const state of ["absent", "unresolved"]) assert.deepEqual(statusValues(factsFor(graph, { resolution: { state, sampler_id: "s" } })), own, state);
+    // Without a named sampler every listed sampler still merges.
+    assert.equal(factsFor(graph, { resolution: { state: "absent", sampler_id: null } }).seed.status, "ambiguous");
+});
+
+test("stage walks follow only the resolver's links", () => {
+    // Review M-2: `["eff", -1]` is not a link to the server, so it is not walked.
+    for (const slot of [-1, 0.5, "0"]) {
+        const graph = detailerFactGraph();
+        graph.eff = { class_type: "MysterySampler", inputs: { positive: ["model", 0], model: ["model", 0] } };
+        graph.decode.inputs.extra = ["eff", slot];
+        assert.equal(factsFor(graph, { resolution: { state: "ambiguous", sampler_id: "d" } }).seed.status, "ambiguous", String(slot));
+        const shaped = detailerFactGraph();
+        shaped.eff = { class_type: "Unlisted", inputs: { positive: ["model", slot], image: ["decode", 0] } };
+        shaped.d.inputs.image = ["eff", 0];
+        assert.equal(factsFor(shaped, { resolution: { state: "ambiguous", sampler_id: "d" } }).seed.status, "ambiguous", `shaped ${slot}`);
+    }
+    const linked = detailerFactGraph();
+    linked.eff = { class_type: "MysterySampler", inputs: { positive: ["model", 0], model: ["model", 0] } };
+    linked.decode.inputs.extra = ["eff", 0];
+    assert.equal(factsFor(linked, { resolution: { state: "ambiguous", sampler_id: "d" } }).seed.status, "unresolved");
+});
+
+test("Flux2Scheduler provides steps and scheduler is its node type", () => {
+    const graph = customFactGraph();
+    graph.sigmas = { class_type: "Flux2Scheduler", inputs: { steps: 4, width: 1024, height: 1024 } };
+    const facts = factsFor(graph);
+    assert.deepEqual([facts.steps.status, facts.steps.value], ["known", "4"]);
+    assert.deepEqual([facts.scheduler.status, facts.scheduler.value, facts.scheduler.evidence], ["known", "Flux2Scheduler", "configuration"]);
+    for (const slot of [1, -1]) {
+        const linked = customFactGraph(); linked.sigmas = graph.sigmas; linked.s.inputs.sigmas = ["sigmas", slot];
+        assert.equal(factsFor(linked).scheduler.status, "unresolved", `output${slot}`);
+    }
+    // Its own inputs cannot replace the node type, and steps keep exact-integer rules.
+    const named = customFactGraph(); named.sigmas = { class_type: "Flux2Scheduler", inputs: { steps: 0, scheduler: "karras" } };
+    assert.equal(factsFor(named).scheduler.value, "Flux2Scheduler");
+    assert.equal(factsFor(named).steps.status, "unresolved");
+});
+
+test("ModelSampling passthroughs reach UNETLoader", () => {
+    const graph = customFactGraph();
+    graph.g.inputs.model = ["flux", 0];
+    graph.flux = { class_type: "ModelSamplingFlux", inputs: { model: ["aura", 0], max_shift: "1.15", base_shift: "0.5" } };
+    graph.aura = { class_type: "ModelSamplingAuraFlow", inputs: { model: ["sd3", 0], shift: "1.73" } };
+    graph.sd3 = { class_type: "ModelSamplingSD3", inputs: { model: ["lora", 0], shift: "3" } };
+    graph.lora = { class_type: "LoraLoaderModelOnly", inputs: { model: ["m", 0], lora_name: "style", strength_model: "1" } };
+    assert.deepEqual([factsFor(graph).model.status, factsFor(graph).model.value], ["known", "flux.safetensors"]);
+    for (const node of ["flux", "aura", "sd3"]) {
+        const wrongSlot = structuredClone(graph);
+        const parent = { flux: "g", aura: "flux", sd3: "aura" }[node];
+        wrongSlot[parent].inputs.model = [node, 1];
+        assert.equal(factsFor(wrongSlot).model.status, "unresolved", `${node} output1`);
+        const unlisted = structuredClone(graph); unlisted[node].class_type = "ModelSamplingUnknown";
+        assert.equal(factsFor(unlisted).model.status, "unresolved", `${node} unlisted`);
+    }
+});
+
+test("vectors.json pins full facts for every inspection fixture", () => {
+    const fs = require("node:fs");
+    const path = require("node:path");
+    const vectors = JSON.parse(fs.readFileSync(path.join(__dirname, "..", "fixtures", "inspection_parity", "vectors.json"), "utf8"));
+    const inspection = vectors.filter(vector => vector.profile === "inspection");
+    assert.ok(inspection.length > 0);
+    const statuses = new Set(["known", "ambiguous", "unresolved", "not-recorded"]);
+    for (const vector of inspection) {
+        const facts = vector.expect.facts;
+        assert.deepEqual(Object.keys(facts), factFields, vector.name);
+        for (const field of factFields) {
+            assert.ok(statuses.has(facts[field].status), `${vector.name} ${field}`);
+            assert.equal(facts[field].value === null, facts[field].status !== "known", `${vector.name} ${field} value`);
+        }
+    }
+    // f19: KSampler seed 42 / steps 25 / cfg 6.5 then FaceDetailer seed 7 / steps 12 / cfg 4.5, different prompts.
+    const f19 = inspection.find(vector => vector.fixture === "f19");
+    assert.equal(f19.expect.state, "ambiguous");
+    assert.deepEqual(Object.fromEntries(factFields.map(field => [field, f19.expect.facts[field].status])), {
+        seed: "ambiguous", steps: "ambiguous", cfg: "ambiguous", sampler: "known", scheduler: "known", model: "known",
+    });
+});
+
 test("wide workflow lookups retain selected core sampler facts", () => {
     const nodes = [{ id: "text", type: "PrimitiveString", widgets_values_named: { value: "unrelated text" } }], links = [];
     let linkId = 1;

tests/js/inspection-legacy-ports.test.js

@@ -0,0 +1,119 @@
+// Run with: node --test tests/js/
+// Legacy parser rules ported to the inspection facts (#240, Task 2.7b). Synthetic texts.
+const test = require("node:test");
+const assert = require("node:assert/strict");
+const { extractGenerationFacts } = require("../../web/js/comfy-metadata.js");
+
+const fields = ["seed", "steps", "cfg", "sampler", "scheduler", "model"];
+const FOOTER = "Steps: 20, Sampler: Euler, CFG scale: 7, Seed: 5";
+
+function parameterFacts(parameters) {
+    const facts = extractGenerationFacts({
+        prompt_graph: { status: "absent", value: null },
+        workflow_graph: { status: "absent", value: null },
+        parameters,
+        recordFormats: { parameters: "original-text" },
+        metadataStatus: "available",
+    });
+    return Object.fromEntries(fields.map(field => [field, [facts[field].status, facts[field].value]]));
+}
+
+const FOUR_KNOWN = {
+    seed: ["known", "5"], steps: ["known", "20"], cfg: ["known", "7"], sampler: ["known", "Euler"],
+    scheduler: ["not-recorded", null], model: ["not-recorded", null],
+};
+const ALL_UNRESOLVED = Object.fromEntries(fields.map(field => [field, ["unresolved", null]]));
+
+test("R4: text starting with the negative delimiter has no positive, as with a leading newline", () => {
+    for (const text of [`Negative prompt: bad\n${FOOTER}`, `\nNegative prompt: bad\n${FOOTER}`, `Negative prompt: a\nNegative prompt: b\n${FOOTER}`]) {
+        assert.deepEqual(parameterFacts(text), ALL_UNRESOLVED, text);
+    }
+    assert.deepEqual(parameterFacts(`a cat\nNegative prompt: bad\n${FOOTER}`), FOUR_KNOWN);
+});
+
+test("R5: template records after the footer and a bare Steps footer", () => {
+    const base = `a cat\nNegative prompt: bad\n${FOOTER}`;
+    for (const trailer of ["\nTemplate: a {cat|dog}", "\nTemplate: a {cat|dog}\nNegative Template: bad", "\nNegative Template: bad\n"]) {
+        assert.deepEqual(parameterFacts(base + trailer), FOUR_KNOWN, trailer);
+    }
+    for (const trailer of ["\nTemplate: a\nTemplate: b", "\nNegative Template: bad\nTemplate: a", "\nTemplate: a\nmore", "\ntrailing"]) {
+        assert.deepEqual(parameterFacts(base + trailer), ALL_UNRESOLVED, trailer);
+    }
+    const bare = parameterFacts("a cat\nNegative prompt: bad\nSteps: 20\nTemplate: a {cat|dog}");
+    assert.deepEqual(bare.steps, ["known", "20"]);
+    assert.ok(["seed", "cfg", "sampler", "scheduler", "model"].every(field => bare[field][0] === "not-recorded"));
+    // A full footer wins over a bare `Steps: N` line in the prompt; two bare ones are not unique.
+    assert.deepEqual(parameterFacts(`a cat\nSteps: 3\nNegative prompt: bad\n${FOOTER}`), FOUR_KNOWN);
+    assert.deepEqual(parameterFacts("a cat\nSteps: 3\nNegative prompt: bad\nSteps: 20"), ALL_UNRESOLVED);
+});
+
+test("R10: commas inside quoted settings values do not split them", () => {
+    assert.deepEqual(parameterFacts(`a cat\nNegative prompt: bad\n${FOOTER}, ADetailer prompt: "face, smile"`), FOUR_KNOWN);
+    assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Note: "say \\"hi, there\\"", Model: base`), { ...FOUR_KNOWN, model: ["known", "base"] });
+    // A quoted value of a fact is its JSON string.
+    assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Model: "base, v2"`).model, ["known", "base, v2"]);
+    // Unquoted JSON (comfy-image-saver hashes) keeps the historical split, field by field.
+    assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Hashes: {"model": "abc", "lora:x": "def"}, Model: base`).model, ["known", "base"]);
+    // An unbalanced quote is not a quoted value: the historical split, as A1111 reads it.
+    assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Note: "open, Model: base`).model, ["known", "base"]);
+});
+
+function graphFacts(graph, resolution) {
+    const facts = extractGenerationFacts({
+        prompt_graph: { status: "parsed", value: graph },
+        workflow_graph: { status: "absent", value: null },
+        metadataStatus: "available",
+        resolution,
+    });
+    return Object.fromEntries(fields.map(field => [field, [facts[field].status, facts[field].value]]));
+}
+
+function modelGraph(between) {
+    return {
+        ckpt: { class_type: "CheckpointLoaderSimple", inputs: { ckpt_name: "base.safetensors" } },
+        hop: { class_type: between, inputs: { model: ["ckpt", 0], clip: ["ckpt", 1] } },
+        s: { class_type: "KSampler", inputs: { seed: 5, steps: 20, cfg: 7, sampler_name: "euler", scheduler: "normal", model: ["hop", 0] } },
+    };
+}
+
+test("R9: model pass-throughs verified in their sources lead to the loader", () => {
+    for (const name of ["Lora Loader (LoraManager)", "Power Lora Loader (rgthree)", "ModelPatchTorchSettings", "CFGNorm", "PerturbedAttentionGuidance"]) {
+        assert.deepEqual(graphFacts(modelGraph(name)).model, ["known", "base.safetensors"], name);
+        // Only output 0 is the model.
+        const other = modelGraph(name); other.s.inputs.model = ["hop", 1];
+        assert.deepEqual(graphFacts(other).model, ["unresolved", null], name);
+    }
+    assert.deepEqual(graphFacts(modelGraph("ModelMergeSDXL")).model, ["unresolved", null]);
+});
+
+test("R6: upscale and SEGS detailers supply stage facts like FaceDetailer", () => {
+    for (const [class_type, image] of [["UltimateSDUpscale", "image"], ["UltimateSDUpscaleNoUpscale", "upscaled_image"], ["DetailerForEach", "image"]]) {
+        const graph = modelGraph("CFGNorm");
+        graph.decode = { class_type: "VAEDecode", inputs: { samples: ["s", 0], vae: ["ckpt", 2] } };
+        graph.d = { class_type, inputs: { [image]: ["decode", 0], model: ["hop", 0], seed: 7, steps: 12, cfg: 7, sampler_name: "euler", scheduler: "normal" } };
+        assert.deepEqual(graphFacts(graph, { state: "resolved", sampler_id: "d" }).seed, ["known", "7"], class_type);
+        const merged = graphFacts(graph, { state: "ambiguous", sampler_id: "d" });
+        assert.deepEqual([merged.seed[0], merged.steps[0], merged.cfg], ["ambiguous", "ambiguous", ["known", "7"]], class_type);
+    }
+});
+
+test("R6: upstream-verified samplers supply facts unless an override input is set", () => {
+    for (const [class_type, overrides] of [["ClownsharKSampler_Beta", ["options", "sigmas"]], ["KSampler (Efficient)", ["script"]], ["ttN KSampler_v2", ["adv_xyPlot"]]]) {
+        const graph = modelGraph("CR Apply LoRA Stack");
+        graph.s.class_type = class_type;
+        assert.deepEqual(graphFacts(graph), {
+            seed: ["known", "5"], steps: ["known", "20"], cfg: ["known", "7"], sampler: ["known", "euler"],
+            scheduler: ["known", "normal"], model: ["known", "base.safetensors"],
+        }, class_type);
+        for (const name of overrides) {
+            const overridden = structuredClone(graph);
+            overridden.s.inputs[name] = ["script", 0];
+            overridden.script = { class_type: "SomeScript", inputs: {} };
+            assert.deepEqual(graphFacts(overridden), ALL_UNRESOLVED, `${class_type} ${name}`);
+            for (const unset of [null, ""]) {
+                overridden.s.inputs[name] = unset;
+                assert.equal(graphFacts(overridden).seed[0], "known", `${class_type} ${name} ${unset}`);
+            }
+        }
+    }
+});

tests/js/inspection-semantics-sync.test.js

@@ -0,0 +1,79 @@
+// Run with: node --test tests/js/
+// comfy-metadata.js stays a classic script for the legacy pages, so its facts
+// tables are inline literals. This test keeps them equal to the one semantics
+// table (web/js/inspection-semantics.json) that Python also loads.
+const test = require("node:test");
+const assert = require("node:assert/strict");
+const fs = require("node:fs");
+const path = require("node:path");
+const metadata = require("../../web/js/comfy-metadata.js");
+
+const tablePath = path.join(__dirname, "..", "..", "web", "js", "inspection-semantics.json");
+const semantics = JSON.parse(fs.readFileSync(tablePath, "utf8"));
+
+function plain(value) {
+    if (value instanceof Set) return [...value];
+    if (Array.isArray(value)) return value.map(plain);
+    if (value && typeof value === "object") return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, plain(item)]));
+    return value;
+}
+
+test("facts tables equal semantics.facts", () => {
+    assert.equal(typeof metadata._factsTables, "object", "comfy-metadata exposes its facts tables for this test");
+    assert.deepStrictEqual(plain(metadata._factsTables), semantics.facts);
+});
+
+test("facts guider providers are guiders in the semantics table", () => {
+    const providers = Object.entries(semantics.facts.configuration_outputs).filter(([, adapter]) => adapter.kind === "guider");
+    assert.ok(providers.length > 0);
+    for (const [name] of providers) assert.ok(Object.hasOwn(semantics.guiders, name), name);
+});
+
+test("the facts tables are not part of the legacy page contract", () => {
+    assert.deepStrictEqual(Object.keys(metadata).filter(key => !key.startsWith("_")).sort(), ["PLACEHOLDERS", "extractGenerationFacts", "extractGenerationParams"]);
+    assert.ok(Object.isFrozen(metadata._factsTables));
+});
+
+// Task 2.1 checked only `facts`; the facts code also reads the samplers table
+// (which classes are samplers, their facts shape and a detailer's image input).
+test("sampler facts table equals semantics.samplers", () => {
+    assert.equal(typeof metadata._samplerTable, "object", "comfy-metadata exposes its sampler table for this test");
+    // Task 2.7b: plus `fact_overrides`; `prompt_overrides` is read only by the resolver.
+    const expected = Object.fromEntries(Object.entries(semantics.samplers).map(([name, entry]) => [name, {
+        facts: entry.facts,
+        ...(entry.image_stage === undefined ? {} : { image_stage: entry.image_stage }),
+        ...(entry.fact_overrides === undefined ? {} : { fact_overrides: entry.fact_overrides }),
+    }]));
+    assert.deepStrictEqual(plain(metadata._samplerTable), expected);
+    assert.ok(Object.isFrozen(metadata._samplerTable));
+});
+
+test("detailer-stage samplers, and only they, name an image stage", () => {
+    for (const [name, entry] of Object.entries(semantics.samplers)) {
+        assert.equal(Object.hasOwn(metadata._samplerTable[name], "image_stage"), entry.strategy === "detailer-stage", name);
+    }
+});
+
+// The stage walk treats every table class as known (the resolver's `all_classes`).
+test("stage walk table classes equal every class in the semantics table", () => {
+    const classes = new Set();
+    for (const category of ["samplers", "guiders", "selectors", "composites", "savers"]) Object.keys(semantics[category]).forEach(name => classes.add(name));
+    for (const category of ["conditioning", "encoders", "strings"]) semantics[category].forEach(entry => classes.add(entry.class));
+    for (const category of ["runtime", "non_file_outputs"]) semantics[category].forEach(name => classes.add(name));
+    assert.deepStrictEqual([...metadata._tableClasses].sort(), [...classes].sort());
+});
+
+// Review I-2: one source for the stage-walk input lists (Python reads them there too).
+test("stage walk inputs equal semantics.stage_inputs", () => {
+    assert.equal(typeof semantics.stage_inputs, "object");
+    assert.equal(typeof metadata._stageInputs, "object", "comfy-metadata exposes its stage inputs for this test");
+    assert.deepStrictEqual(plain(metadata._stageInputs), semantics.stage_inputs);
+    assert.ok(Object.isFrozen(metadata._stageInputs));
+});
+
+// Review M-3: JS treats every samplers entry as a candidate; the server only these strategies.
+test("every samplers entry is a candidate strategy", () => {
+    for (const [name, entry] of Object.entries(semantics.samplers)) {
+        assert.ok(["ksampler", "custom-advanced", "detailer-stage"].includes(entry.strategy), name);
+    }
+});

tests/js/legacy-metadata-characterization.test.js

@@ -3,7 +3,10 @@
 // (web/metadata.html, web/js/gallery.js) show today on the synthetic inspection
 // parity fixtures, so the #240 workspace work cannot change them by accident.
 // Some answers are known to be wrong (f01b shows the preview-only detailer text
-// as both roles; f16 shows the last of two conflicting prompt chunks). They are
+// as both roles; f16 shows the last of two conflicting prompt chunks; f19 shows the
+// upstream sampler's prompt and settings, not the detailer's; f23 shows the positive
+// text as the zeroed negative; f21 joins a list output; f22 shows PromptSearchList's
+// search text as the prompt). They are
 // recorded as-is and intentionally unchanged (spec section 11).
 const test = require("node:test");
 const assert = require("node:assert/strict");
@@ -56,6 +59,12 @@ const EXPECTED = {
     f15: { positivePrompt: "a red fox in snow, detailed fur", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
     f16: { positivePrompt: "a grey wolf in fog", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
     f17: { positivePrompt: "No prompt found", negativePrompt: "No negative prompt found", checkpoint: "flux1-dev.safetensors", steps: "Unknown", cfgScale: "Unknown", sampler: "euler", seed: "Unknown" },
+    f19: { positivePrompt: "a red fox in snow, detailed fur", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
+    f20: { positivePrompt: "No prompt found", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
+    f21: { positivePrompt: "a red fox, in snow", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
+    f22: { positivePrompt: "fox", negativePrompt: "blurry, lowres", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
+    f23: { positivePrompt: "a red fox in snow, detailed fur", negativePrompt: "a red fox in snow, detailed fur", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
+    f24: { positivePrompt: "No prompt found", negativePrompt: "No negative prompt found", checkpoint: "sdxl_base.safetensors", steps: 25, cfgScale: 6.5, sampler: "euler", seed: 42 },
 };
 
 test("legacy extractGenerationParams characterization covers every committed fixture", () => {

tests/js/prompt-blocks.test.js

@@ -70,3 +70,76 @@ test('frozen fields disclose their exact source and keep null, empty and whitesp
         container.click(f.walk(container,node=>node.dataset.sendRole==='pair')[0]);assert.equal(sent.length,2);
     }finally{scope.dispose();f.restore();}
 });
+
+test('a zeroed conditioning negative shows as an explicit empty negative with its own source',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div'),sent=[];
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:roles=>sent.push(roles)},prompts:{positive:{status:'resolved',text:'positive',source:'graph-inferred'},negative:{status:'resolved',text:'',source:'zeroed-conditioning'}}});
+        assert.equal(f.walk(container,node=>node.textContent==='Zeroed conditioning (no negative text)').length,1);
+        assert.equal(f.walk(container,node=>node.textContent==='Empty negative prompt').length,1);
+        container.click(f.walk(container,node=>node.dataset.sendRole==='pair')[0]);
+        assert.deepEqual(sent,[{positive:'positive',negative:''}]);
+    }finally{scope.dispose();f.restore();}
+});
+
+const unresolved=message=>({status:'unresolved',text:null,source:null,...(message===undefined?{}:{message})});
+const statusLines=(f,container)=>f.walk(container,node=>node.tagName==='p'&&node.className==='muted').map(node=>node.textContent);
+test('unsupported node status names the class',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    const line='Prompt uses a node the inspector does not support (DualCFGGuider).';
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
+        assert.deepEqual(statusLines(f,container),[line,line]);
+        assert.equal(f.walk(container,node=>node.dataset.sendRole!==undefined||node.dataset.copyRole!==undefined).length,0,'no action implies a prompt exists');
+    }finally{scope.dispose();f.restore();}
+});
+test('runtime and composite statuses',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    const runtime='Prompt is chosen at run time (PromptSearchList).',composite='Prompt combines several texts (ConditioningCombine).';
+    try{
+        const view=module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(runtime),negative:unresolved(runtime)}});
+        assert.deepEqual(statusLines(f,container),[runtime,runtime]);
+        view.update({positive:unresolved(composite),negative:{status:'resolved',text:'kept',source:'execution'}});
+        assert.deepEqual(statusLines(f,container),[composite]);
+        assert.equal(f.walk(container,node=>node.tagName==='pre'&&node.textContent==='kept').length,1);
+    }finally{scope.dispose();f.restore();}
+});
+test('limit status',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    const line='Prompt exceeds inspection limits.';
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
+        assert.deepEqual(statusLines(f,container),[line,line]);
+    }finally{scope.dispose();f.restore();}
+});
+test('roles without a usable line keep the status copy, and a candidate is never shown as a prompt',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(''),negative:{status:'ambiguous',text:'candidate text',source:null}}});
+        assert.deepEqual(statusLines(f,container),['Prompt could not be resolved.','Multiple possible prompts. Source could not be determined.']);
+        assert.equal(f.walk(container,node=>node.textContent==='candidate text').length,0);
+    }finally{scope.dispose();f.restore();}
+});
+test('class names in status lines render as text, never markup',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    const line='Prompt uses a node the inspector does not support (<img src=x onerror=alert(1)>).';
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
+        assert.deepEqual(statusLines(f,container),[line,line]);
+        assert.equal(f.walk(container,node=>node.tagName==='img').length,0);
+        assert.equal(f.walk(container,node=>Object.hasOwn(node,'innerHTML')).length,0);
+    }finally{scope.dispose();f.restore();}
+});
+
+test('a context label names nested prompt regions and actions apart from the frozen blocks while the visible heading stays',()=>{
+    const f=dom(),scope=createScope(),container=new f.Element('div');
+    try{
+        module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},contextLabel:'Current file',prompts:{positive:record('p'),negative:record('n')}});
+        const labels=f.walk(container,node=>node.attributes['aria-label']).map(node=>node.attributes['aria-label']);
+        for(const name of ['Current file positive prompt','Current file negative prompt','Copy current file positive prompt','Send current file positive prompt','Copy current file negative prompt','Send current file negative prompt']) assert.ok(labels.includes(name),name);
+        assert.equal(labels.filter(name=>/^(Positive|Negative) prompt$/.test(name)).length,0);
+        assert.deepEqual(f.walk(container,node=>node.tagName==='h3').map(node=>node.textContent),['Positive prompt','Negative prompt']);
+        const plain=new f.Element('div');module.mountPromptBlocks(plain,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:record('p')}});
+        assert.ok(f.walk(plain,node=>node.attributes['aria-label']==='Positive prompt').length);
+    }finally{scope.dispose();f.restore();}
+});

tests/js/workspace-inspection-state.test.js

@@ -225,6 +225,48 @@ test('record accepts closed reason codes and rejects contradictory ones', () =>
     }
 });
 
+test('resolution validator accepts closed reason codes and a bounded node record, rejects others', () => {
+    const toRecord = contract('toInspectionRecord');
+    for (const reason_code of ['unsupported-node', 'runtime-value', 'composite']) {
+        for (const node of [{ id: '7', class_type: 'N'.repeat(200) }, { id: '7'.repeat(200), class_type: null }, null]) {
+            const data = envelope(); Object.assign(data.resolution, { state: 'unresolved', reason_code, node });
+            assert.doesNotThrow(() => toRecord(data, image(), extractGenerationFacts), JSON.stringify({ reason_code, node }));
+        }
+    }
+    for (const change of [
+        { reason_code: 'unsupported-node', node: { id: '7', class_type: 'N'.repeat(201) } },
+        { reason_code: 'unsupported-node', node: { id: '7'.repeat(201), class_type: 'X' } },
+        { reason_code: 'unsupported-node', node: { id: '', class_type: 'X' } },
+        { reason_code: 'unsupported-node', node: { id: '7', class_type: 'X', extra: true } },
+        { reason_code: 'unsupported-node', node: [] },
+        { reason_code: 'malformed', node: { id: '7', class_type: 'X' } },
+        { reason_code: 'work-limit', node: { id: '7', class_type: 'X' } },
+        { reason_code: 'no-sampler', node: { id: '7', class_type: 'X' } },
+        { reason_code: 'Unsupported-Node', node: null },
+        { reason_code: '', node: null },
+    ]) {
+        const data = envelope(); Object.assign(data.resolution, { state: 'unresolved', ...change });
+        assert.throws(() => toRecord(data, image(), extractGenerationFacts), TypeError, JSON.stringify(change));
+    }
+});
+
+test('role outcome lines are bounded text on non-resolved roles only', () => {
+    const toRecord = contract('toInspectionRecord');
+    const line = 'Prompt uses a node the inspector does not support (<img src=x onerror=alert(1)>).';
+    const data = envelope(); data.prompts.negative = { status: 'unresolved', text: null, source: null, message: line };
+    assert.equal(toRecord(data, image(), extractGenerationFacts).prompts.negative.message, line);
+    for (const [role, record] of [
+        ['positive', { status: 'resolved', text: 'exact', source: 'execution', message: 'Hidden reason.' }],
+        ['negative', { status: 'unresolved', text: null, source: null, message: '' }],
+        ['negative', { status: 'unresolved', text: null, source: null, message: 7 }],
+        ['negative', { status: 'unresolved', text: null, source: null, message: null }],
+        ['negative', { status: 'unresolved', text: null, source: null, message: 'x'.repeat(513) }],
+    ]) {
+        const bad = envelope(); bad.prompts[role] = record;
+        assert.throws(() => toRecord(bad, image(), extractGenerationFacts), TypeError, JSON.stringify(record));
+    }
+});
+
 test('record rejects root path media and invalid versioned fields with retryable TypeError', () => {
     const toRecord = contract('toInspectionRecord');
     for (const mutate of [

tests/test_api_prompt_inspector.py

@@ -244,10 +244,10 @@ class TestPromptInspector(ImageAPITestCase):
 
     async def test_corrupt_malformed_and_no_metadata_do_not_invent_text(self):
         path = self.output_dir / "image.png"
-        for content, expected in (
-            (b"broken", "unresolved"),
-            (None, "absent"),
-            ("bad-json", "unresolved"),
+        for content, expected, message in (
+            (b"broken", "unresolved", "Metadata in this file could not be read."),
+            (None, "absent", None),
+            ("bad-json", "unresolved", "Prompt metadata in this file is malformed."),
         ):
             if isinstance(content, bytes):
                 path.write_bytes(content)
@@ -256,6 +256,7 @@ class TestPromptInspector(ImageAPITestCase):
             _, result = await self.inspect()
             self.assertEqual(result["prompts"]["positive"]["status"], expected)
             self.assertIsNone(result["prompts"]["positive"]["text"])
+            self.assertEqual(result["prompts"]["negative"].get("message"), message)
 
     async def test_compressed_international_and_post_pixel_metadata(self):
         import struct
@@ -773,8 +774,11 @@ class TestPromptInspector(ImageAPITestCase):
         self.assertEqual(result["metadata"]["prompt_graph"]["status"], "parsed")
 
     async def test_inspect_route_reports_no_sampler_instead_of_absent(self):
-        # f17: a BasicGuider-only prompt graph is a record, never "not recorded".
-        self.parity_fixture("f17")
+        # f17 with a sampler class the inspector does not know: a prompt graph
+        # is a record, never "not recorded".
+        built = parity.build("f17")
+        built["prompt"]["13"]["class_type"] = "FutureSampler"
+        parity.write_png(self.output_dir / "f17.png", built)
         response, result = await self.inspect("f17.png")
         self.assertEqual(response.status, 200)
         self.assertEqual(
@@ -785,17 +789,32 @@ class TestPromptInspector(ImageAPITestCase):
         self.assertEqual(result["resolution"]["reason_code"], "no-sampler")
         self.assertEqual(
             result["resolution"]["reason"],
-            "No supported sampler pair found (saw SamplerCustomAdvanced)",
+            "No supported sampler pair found (saw FutureSampler)",
+        )
+        self.assertEqual(
+            result["prompts"]["positive"]["message"], "No supported sampler was found."
         )
 
     async def test_inspect_route_exposes_reason_code_and_blocking_node(self):
-        self.parity_fixture("f06a")
+        # f06a's encoder is supported since Task 2.4; an unknown encoder is not.
+        built = parity.build("f06a")
+        built["prompt"]["6"]["class_type"] = "FutureTextEncode"
+        parity.write_png(self.output_dir / "f06a.png", built)
         _, result = await self.inspect("f06a.png")
         resolution = result["resolution"]
         self.assertEqual(resolution["state"], "unresolved")
         self.assertEqual(resolution["reason_code"], "unsupported-node")
-        self.assertEqual(set(resolution["node"]), {"id", "class_type"})
-        self.assertIsInstance(resolution["node"]["class_type"], str)
+        self.assertEqual(
+            resolution["node"], {"id": "6", "class_type": "FutureTextEncode"}
+        )
+        self.assertEqual(
+            {role: side.get("message") for role, side in result["prompts"].items()},
+            dict.fromkeys(
+                ("positive", "negative"),
+                "Prompt uses a node the inspector does not support "
+                "(FutureTextEncode).",
+            ),
+        )
         png(self.output_dir / "image.png", prompt="{broken")
         _, result = await self.inspect()
         self.assertEqual(result["resolution"]["reason_code"], "malformed")

tests/test_history_current.py

@@ -449,6 +449,46 @@ class CurrentLocationTests(unittest.TestCase):
                 )
         self.assertEqual(reads, [], "Unsupported media must not reach the parser")
 
+    def test_preview_files_in_the_host_temp_folder_keep_preview_only_stages(self):
+        # Saved raw image plus a FaceDetailer that only previews (Codex P2).
+        import json
+
+        from PIL import Image
+        from PIL.PngImagePlugin import PngInfo
+        from tests import inspection_parity as parity
+
+        service = self.module("py.api.inspection_service")
+        graph = parity.build("f19")["prompt"]
+        graph["9"]["class_type"] = "PreviewImage"
+        graph["14"] = {"class_type": "SaveImage", "inputs": {"images": ["8", 0]}}
+        info = PngInfo()
+        info.add_text("prompt", json.dumps(graph))
+        observed = {}
+        for name, folder in (("output", self.output), ("temp", self.temp)):
+            path = folder / "nested" / "same.png"
+            path.parent.mkdir()
+            with Image.new("RGB", (4, 4)) as image:
+                image.save(path, pnginfo=info)
+            observed[name] = service.inspect_current_file(
+                path, source_label=name, saved=None, library={"status": "none"}
+            )
+        saved = observed["output"]["prompts"]
+        self.assertEqual(saved["positive"]["status"], "resolved")
+        self.assertEqual(saved["positive"]["text"], parity.POS)
+        self.assertEqual(observed["output"]["resolution"]["sampler_id"], "3")
+        preview = observed["temp"]["prompts"]
+        self.assertEqual(preview["positive"]["status"], "ambiguous")
+        self.assertIsNone(preview["positive"]["text"])
+        # An unknown host temp folder treats the file as saved output.
+        self.host_temp = None
+        again = service.inspect_current_file(
+            self.output / "nested" / "same.png",
+            source_label="output",
+            saved=None,
+            library={"status": "none"},
+        )
+        self.assertEqual(again["prompts"]["positive"]["status"], "resolved")
+
 
 class HistoryImportTests(unittest.TestCase):
     def test_api_import_does_not_require_a_running_host_server(self):

tests/test_inspection_legacy_ports.py

@@ -0,0 +1,943 @@
+"""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 io
+import json
+import tempfile
+import unittest
+import unittest.mock
+import zlib
+from pathlib import Path
+
+from PIL import Image
+
+from utils import prompt_pair_resolver as resolver
+from utils.prompt_inspector import (
+    DUPLICATE_ROLE_BYTES,
+    MetadataReadError,
+    inspect_prompt_evidence,
+    read_inspection_file,
+)
+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.
+        orphans = {
+            "nodes": [{"id": 5, "type": "PromptManager", "widgets_values": ["a"]}]
+        }
+        for workflow in (None, "nodes", {"nodes": "x"}, {"nodes": [1, None]}, orphans):
+            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_overwritten_nodes_inside_subgraphs_are_refused(self):
+        # Review M-1: subgraph nodes run as "<instance>:<inner>" API IDs.
+        def nested(depth: int = 1) -> dict:
+            metadata = _overwritten()
+            prompt, workflow = metadata["prompt"], metadata["workflow"]
+            prefix = ":".join(["50", "60", "70"][:depth]) + ":"
+            for node_id in ("17", "113", "119"):
+                prompt[prefix + node_id] = prompt.pop(node_id)
+            prompt["3"]["inputs"]["positive"] = [prefix + "17", 0]
+            inner = [n for n in workflow["nodes"] if n["type"] == "PromptManager"]
+            outer = [n for n in workflow["nodes"] if n["type"] != "PromptManager"]
+            subgraphs = []
+            for level in range(depth, 0, -1):
+                nodes = (
+                    inner
+                    if level == depth
+                    else [{"id": [50, 60, 70][level], "type": f"sub-{level + 1}"}]
+                )
+                subgraphs.append({"id": f"sub-{level}", "nodes": nodes})
+            metadata["workflow"] = {
+                "nodes": outer + [{"id": 50, "type": "sub-1", "widgets_values": []}],
+                "links": [],
+                "definitions": {"subgraphs": subgraphs},
+            }
+            return metadata
+
+        for depth in (1, 2, 3):
+            with self.subTest(depth=depth):
+                for anchor in (None, "9"):
+                    result = _inspect(nested(depth), output_node_id=anchor)
+                    self.assertEqual(
+                        result["reason"], resolver.OVERWRITTEN_PROMPT_REASON, anchor
+                    )
+        # A subgraph definition with no instance runs nothing.
+        metadata = nested()
+        metadata["workflow"]["nodes"] = metadata["workflow"]["nodes"][:-1]
+        self.assertEqual(_inspect(metadata)["state"], "resolved")
+        # Malformed definitions never raise.
+        for definitions in ("x", {"subgraphs": "x"}, {"subgraphs": [1, {"id": 3}]}):
+            metadata = nested()
+            metadata["workflow"]["definitions"] = definitions
+            self.assertEqual(_inspect(metadata)["state"], "resolved", definitions)
+        # A self-referencing subgraph is bounded.
+        metadata = nested()
+        metadata["workflow"]["definitions"]["subgraphs"].append(
+            {"id": "loop", "nodes": [{"id": 1, "type": "loop"}]}
+        )
+        metadata["workflow"]["nodes"].append({"id": 99, "type": "loop"})
+        self.assertEqual(
+            _inspect(metadata)["reason"], resolver.OVERWRITTEN_PROMPT_REASON
+        )
+
+    def test_object_widget_values_name_the_text(self):
+        metadata = _overwritten()
+        for node in metadata["workflow"]["nodes"]:
+            node["widgets_values"] = {"text": node["widgets_values"][0]}
+        self.assertEqual(
+            _inspect(metadata)["reason"], resolver.OVERWRITTEN_PROMPT_REASON
+        )
+
+    def test_the_workflow_is_parsed_only_for_prompt_manager_graphs(self):
+        # Review M-4: no PromptManager-like node, no workflow parse.
+
+        from utils import prompt_inspector
+
+        prompt = parity.build("f02")["prompt"]
+        with unittest.mock.patch.object(
+            prompt_inspector,
+            "_parsed_workflow",
+            wraps=prompt_inspector._parsed_workflow,
+        ) as parsed:
+            inspect_prompt_evidence(
+                {"prompt": json.dumps(prompt), "workflow": json.dumps({"nodes": []})}
+            )
+            self.assertEqual(parsed.call_count, 0)
+            metadata = _overwritten()
+            inspect_prompt_evidence(
+                {
+                    "prompt": json.dumps(metadata["prompt"]),
+                    "workflow": json.dumps(metadata["workflow"]),
+                }
+            )
+            self.assertEqual(parsed.call_count, 1)
+
+    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"),
+            # Review M-5: an explicit null part is skipped, as ``if prepend_text:``.
+            ({"text": "a cat", "prepend_text": None, "append_text": None}, "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")
+
+
+FOOTER = "Steps: 20, Sampler: Euler, CFG scale: 7, Seed: 5"
+
+
+def _parameters(text: str, *, legacy: bool = False) -> dict:
+    profile = resolver.DEFAULT_PROFILE if legacy else resolver.INSPECTION_PROFILE
+    return resolve_prompt_pair({"parameters": text}, profile=profile)
+
+
+class TestA1111Text(unittest.TestCase):
+    """R4 and R5 (#165, py/api/__init__.py:738-750) for inspection."""
+
+    def test_text_starting_with_the_negative_delimiter_has_no_positive(self):
+        for text in (
+            f"Negative prompt: bad\n{FOOTER}",
+            f"\nNegative prompt: bad\n{FOOTER}",
+        ):
+            with self.subTest(text=text):
+                result = _parameters(text)
+                self.assertEqual(result["state"], "unresolved")
+                self.assertEqual(result["reason"], resolver.EMPTY_POSITIVE_REASON)
+                self.assertIsNone(result["positive"])
+        # Two delimiters, one at the start, are still two.
+        result = _parameters(f"Negative prompt: a\nNegative prompt: b\n{FOOTER}")
+        self.assertEqual(result["state"], "ambiguous")
+
+    def test_dynamic_prompts_template_lines_follow_the_footer(self):
+        base = f"a cat\nNegative prompt: bad\n{FOOTER}"
+        for trailer in (
+            "\nTemplate: a {cat|dog}",
+            "\nTemplate: a {cat|dog}\nNegative Template: bad",
+            "\nNegative Template: bad",
+            "\nTemplate: a {cat|dog}\nNegative Template: bad\n",
+        ):
+            with self.subTest(trailer=trailer):
+                result = _parameters(base + trailer)
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(_texts(result), ("a cat", "bad"))
+        for trailer in (
+            "\nTemplate: a\nTemplate: b",
+            "\nNegative Template: bad\nTemplate: a",
+            "\nTemplate: a\nmore template text",
+            "\ntrailing",
+        ):
+            with self.subTest(trailer=trailer):
+                result = _parameters(base + trailer)
+                self.assertEqual(result["reason_code"], "malformed")
+
+    def test_footer_with_only_steps(self):
+        result = _parameters("a cat\nNegative prompt: bad\nSteps: 20")
+        self.assertEqual(_texts(result), ("a cat", "bad"))
+        result = _parameters("a cat\nSteps: 20\nTemplate: a {cat|dog}")
+        self.assertEqual((result["state"], _texts(result)[0]), ("absent", "a cat"))
+        # A full footer wins over a bare ``Steps: N`` line in the prompt text.
+        result = _parameters(f"a cat\nSteps: 3\nNegative prompt: bad\n{FOOTER}")
+        self.assertEqual(_texts(result), ("a cat\nSteps: 3", "bad"))
+        result = _parameters("a cat\nSteps: 3\nNegative prompt: bad\nSteps: 20")
+        self.assertEqual(result["reason_code"], "malformed")
+
+    def test_default_profile_keeps_the_historical_split(self):
+        result = _parameters(f"Negative prompt: bad\n{FOOTER}", legacy=True)
+        self.assertEqual(_texts(result), ("Negative prompt: bad", None))
+        for text in (
+            f"a cat\nNegative prompt: bad\n{FOOTER}\nTemplate: a {{cat|dog}}",
+            "a cat\nNegative prompt: bad\nSteps: 20",
+        ):
+            with self.subTest(text=text):
+                result = _parameters(text, legacy=True)
+                self.assertEqual(
+                    result["reason"], "A1111 settings footer is not unique"
+                )
+
+
+def _chunk(keyword: str, text: str, kind: bytes = b"tEXt") -> bytes:
+    body = text.encode("latin-1")
+    if kind == b"zTXt":
+        body = b"\0" + zlib.compress(body)
+    elif kind == b"iTXt":
+        body = b"\0\0\0\0" + text.encode("utf-8")
+    return parity._raw_chunk(kind, keyword.encode("latin-1") + b"\0" + body)
+
+
+def _batch(index: int, positive: str = "a cat", seed: int = 5) -> str:
+    return (
+        f"{positive}\nNegative prompt: bad\nSteps: 20, Sampler: Euler, "
+        f"CFG scale: 7, Seed: {seed}, Batch index: {index}, Batch size: 12"
+    )
+
+
+class TestBatchIndexDuplicates(unittest.TestCase):
+    """R3: an accumulating batch saver repeats ``parameters`` per batch item."""
+
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.directory = Path(temporary.name)
+
+    def png(self, chunks, name="batch.png"):
+        buffer = io.BytesIO()
+        with Image.new("RGB", (4, 4), (40, 80, 120)) as image:
+            image.save(buffer, format="PNG")
+        data = buffer.getvalue()
+        iend = data.rindex(b"IEND") - 4
+        path = self.directory / name
+        path.write_bytes(data[:iend] + b"".join(chunks) + data[iend:])
+        return path
+
+    def test_copies_differing_only_in_batch_index_read_as_the_last(self):
+        cases = {
+            "same length": [0, 1, 2],
+            "growing length": [8, 9, 10, 11],
+            "any order": [3, 0],
+            "repeat of an earlier copy": [0, 1, 0],
+        }
+        for label, indexes in cases.items():
+            with self.subTest(label):
+                path = self.png([_chunk("parameters", _batch(i)) for i in indexes])
+                read = read_inspection_file(path)
+                self.assertEqual(read.metadata["parameters"], _batch(indexes[-1]))
+                self.assertEqual(read.record_formats["parameters"], "original-text")
+                evidence = inspect_prompt_evidence(read.metadata)
+                self.assertEqual(evidence["resolution"]["state"], "resolved")
+                self.assertEqual(evidence["prompts"]["positive"]["text"], "a cat")
+        # Encodings may differ; the decoded copies are compared.
+        path = self.png(
+            [
+                _chunk("parameters", _batch(0)),
+                _chunk("parameters", _batch(1), b"zTXt"),
+                _chunk("parameters", _batch(2), b"iTXt"),
+            ]
+        )
+        self.assertEqual(read_inspection_file(path).metadata["parameters"], _batch(2))
+
+    def test_any_other_difference_is_still_a_conflict(self):
+        cases = {
+            "positive": [_batch(0), _batch(1, positive="a dog")],
+            "seed": [_batch(0), _batch(1, seed=6)],
+            "no batch index": [_batch(0), _batch(0).replace(", Batch index: 0", "")],
+            "two batch indexes": [_batch(0), _batch(1) + ", Batch index: 2"],
+            "prose": [_batch(0), _batch(1).replace("a cat", "a cat Batch index: 1")],
+            # Review I-1: the field is matched only in the settings footer.
+            "prompt field": [
+                _batch(0).replace("a cat", "a cat, Batch index: 0"),
+                _batch(0).replace("a cat", "a cat, Batch index: 1"),
+            ],
+            "prompt only": [
+                "a cat, Batch index: 0\nNegative prompt: bad\nSteps: 20, Seed: 5",
+                "a cat, Batch index: 1\nNegative prompt: bad\nSteps: 20, Seed: 5",
+            ],
+            "negative field": [
+                _batch(0).replace("bad", "bad, Batch index: 0"),
+                _batch(1).replace("bad", "bad, Batch index: 1"),
+            ],
+            "no footer": [
+                "a cat\nBatch index: 0, Batch size: 2",
+                "a cat\nBatch index: 1, Batch size: 2",
+            ],
+        }
+        for label, texts in cases.items():
+            with self.subTest(label):
+                path = self.png([_chunk("parameters", text) for text in texts])
+                with self.assertRaisesRegex(
+                    MetadataReadError, "Duplicate prompt metadata"
+                ):
+                    read_inspection_file(path)
+        # The rule is only for ``parameters``: prompt graphs stay byte-identical.
+        prompt = json.dumps(parity.build("f02")["prompt"])
+        path = self.png([_chunk("prompt", prompt), _chunk("prompt", prompt + " ")])
+        with self.assertRaisesRegex(MetadataReadError, "Duplicate prompt metadata"):
+            read_inspection_file(path)
+
+    def test_decoded_copies_are_bounded(self):
+        # Small compressed copies must not decode unbounded text.
+        text = _batch(0) + "," + " " * (DUPLICATE_ROLE_BYTES // 8)
+        chunks = [
+            _chunk("parameters", text.replace("index: 0", f"index: {i}"), b"zTXt")
+            for i in range(10)
+        ]
+        path = self.png(chunks)
+        with self.assertRaisesRegex(MetadataReadError, "exceeds inspection limits"):
+            read_inspection_file(path)
+        # The kept copy still fits the role budget when a later one is longer.
+        from utils.prompt_inspector import ROLE_METADATA_BYTES
+
+        first = _batch(9)
+        first += "," + " " * (ROLE_METADATA_BYTES - len(first) - 1)
+        later = first.replace("index: 9", "index: 10")
+        for copy in (later, later + " " * ROLE_METADATA_BYTES):
+            with self.subTest(size=len(copy)):
+                path = self.png(
+                    [
+                        _chunk("parameters", first, b"zTXt"),
+                        _chunk("parameters", copy, b"zTXt"),
+                    ]
+                )
+                with self.assertRaisesRegex(
+                    MetadataReadError, "exceeds inspection limits"
+                ):
+                    read_inspection_file(path)
+
+
+def _f02(**changes) -> dict:
+    """``f02`` (KSampler 3, encoders 6 and 7) with nodes added or replaced."""
+    prompt = parity.build("f02")["prompt"]
+    prompt.update(changes)
+    return prompt
+
+
+def _stage(class_type: str, image_input: str, positive=("6", 0), **inputs) -> dict:
+    """``f19``'s shape with ``class_type`` as the detailer stage (node 50)."""
+    prompt = parity.build("f19")["prompt"]
+    if "upscale_model" in inputs:
+        prompt["12"] = {"class_type": "UpscaleModelLoader", "inputs": {}}
+    if "segs" in inputs:
+        prompt["13"] = {"class_type": "SegsDetector", "inputs": {"image": ["8", 0]}}
+    prompt["50"] = {
+        "class_type": class_type,
+        "inputs": {
+            image_input: ["8", 0],
+            "model": ["4", 0],
+            "vae": ["4", 2],
+            "positive": list(positive),
+            "negative": ["7", 0],
+            "seed": 7,
+            "steps": 12,
+            "cfg": 4.5,
+            "sampler_name": "euler",
+            "scheduler": "normal",
+            "denoise": 0.4,
+            **inputs,
+        },
+    }
+    return prompt
+
+
+DETAILERS = (
+    ("UltimateSDUpscale", "image", {"upscale_model": ["12", 0], "upscale_by": 2}),
+    ("UltimateSDUpscaleNoUpscale", "upscaled_image", {}),
+    ("DetailerForEach", "image", {"segs": ["13", 0], "wildcard": ""}),
+)
+
+
+class TestInstalledNodeSemantics(unittest.TestCase):
+    """R6 entries verified against the installed node sources (see the report)."""
+
+    def test_upscale_and_segs_detailers_are_stages(self):
+        for class_type, image, extra in DETAILERS:
+            with self.subTest(node=class_type):
+                same = _inspect({"prompt": _stage(class_type, image, **extra)})
+                self.assertEqual(same["state"], "resolved")
+                self.assertEqual(same["sampler_id"], "50")
+                self.assertEqual(_texts(same), (parity.POS, parity.NEG))
+                other = _inspect(
+                    {"prompt": _stage(class_type, image, ("10", 0), **extra)}
+                )
+                self.assertEqual(other["state"], "ambiguous")
+                self.assertEqual(other["reason_code"], "multistage-differs")
+                self.assertEqual(
+                    sorted(item["sampler_id"] for item in other["candidates"]),
+                    ["3", "50"],
+                )
+
+    def test_a_detailer_wildcard_replaces_its_prompt(self):
+        # Impact Pack core.py: a non-empty wildcard is encoded instead of positive.
+        for class_type in ("DetailerForEach", "FaceDetailer"):
+            with self.subTest(node=class_type):
+                result = _inspect(
+                    {"prompt": _stage(class_type, "image", wildcard="a smiling face")}
+                )
+                self.assertEqual(result["state"], "unresolved")
+                self.assertEqual(result["reason_code"], "runtime-value")
+                self.assertEqual(result["node"], {"id": "50", "class_type": class_type})
+                linked = _stage(class_type, "image", wildcard=["14", 0])
+                linked["14"] = {"class_type": "WildcardSource", "inputs": {}}
+                result = _inspect({"prompt": linked})
+                self.assertEqual(result["reason_code"], "unsupported-node")
+                self.assertEqual(
+                    result["node"], {"id": "14", "class_type": "WildcardSource"}
+                )
+                for empty in ("", None):
+                    result = _inspect(
+                        {"prompt": _stage(class_type, "image", wildcard=empty)}
+                    )
+                    self.assertEqual(result["state"], "resolved", empty)
+
+    def test_conditioning_passthroughs_keep_each_role(self):
+        pair = {"positive": ["6", 0], "negative": ["7", 0]}
+        cases = {
+            "ControlNetInpaintingAliMamaApply": (
+                {
+                    **pair,
+                    "control_net": ["11", 0],
+                    "vae": ["4", 2],
+                    "image": ["12", 0],
+                    "mask": ["12", 1],
+                    "strength": 1.0,
+                },
+                0,
+                1,
+            ),
+            "LTXVConditioning": ({**pair, "frame_rate": 25.0}, 0, 1),
+            "FluxKontextMultiReferenceLatentMethod": (
+                {"conditioning": ["6", 0], "reference_latents_method": "offset"},
+                0,
+                None,
+            ),
+        }
+        for class_type, (inputs, positive, negative) in cases.items():
+            with self.subTest(node=class_type):
+                prompt = _f02(**{"20": {"class_type": class_type, "inputs": inputs}})
+                prompt["3"]["inputs"]["positive"] = ["20", positive]
+                if negative is not None:
+                    prompt["3"]["inputs"]["negative"] = ["20", negative]
+                result = _inspect({"prompt": prompt})
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(_texts(result), (parity.POS, parity.NEG))
+                self.assertEqual(
+                    result["positive"]["provenance"]["path"][0]["node_id"], "20"
+                )
+                if negative is None:
+                    prompt["3"]["inputs"]["negative"] = ["20", 1]
+                    self.assertEqual(
+                        _inspect({"prompt": prompt})["reason_code"], "unsupported-node"
+                    )
+
+    def test_text_nodes(self):
+        prompt = _f02(
+            **{
+                "6": {
+                    "class_type": "Text to Conditioning",
+                    "inputs": {"clip": ["4", 1], "text": ["21", 0]},
+                },
+                "21": {"class_type": "DisplayText", "inputs": {"text": ["22", 0]}},
+                "22": {"class_type": "TextInput", "inputs": {"text": parity.POS}},
+            }
+        )
+        result = _inspect({"prompt": prompt})
+        self.assertEqual(_texts(result), (parity.POS, parity.NEG))
+        self.assertEqual(
+            [step["node_id"] for step in result["positive"]["provenance"]["path"]],
+            ["6", "21", "22"],
+        )
+        # Text to Conditioning has one output; TextInput reads only ``text``.
+        prompt["3"]["inputs"]["positive"] = ["6", 1]
+        self.assertEqual(
+            _inspect({"prompt": prompt})["reason_code"], "unsupported-node"
+        )
+
+    def test_overwritten_text_behind_a_new_passthrough_is_still_refused(self):
+        # R1 before R6: the pass-through must not let overwritten text resolve.
+        metadata = _overwritten()
+        metadata["prompt"]["20"] = {
+            "class_type": "FluxKontextMultiReferenceLatentMethod",
+            "inputs": {"conditioning": ["17", 0], "reference_latents_method": "offset"},
+        }
+        metadata["prompt"]["3"]["inputs"]["positive"] = ["20", 0]
+        result = _inspect(metadata)
+        self.assertEqual(result["reason"], resolver.OVERWRITTEN_PROMPT_REASON)
+
+    def test_default_profile_keeps_them_unsupported(self):
+        for class_type, image, extra in DETAILERS:
+            with self.subTest(node=class_type):
+                result = resolve_prompt_pair(
+                    {"prompt": _stage(class_type, image, **extra)}
+                )
+                # The historical rule: the one KSampler, whatever reads its image.
+                self.assertEqual(result["sampler_id"], "3")
+        prompt = _f02(
+            **{"6": {"class_type": "Text to Conditioning", "inputs": {"text": "a"}}}
+        )
+        self.assertEqual(
+            resolve_prompt_pair({"prompt": prompt})["reason"],
+            "Unsupported node semantics or output slot",
+        )
+
+
+def _sampled(class_type: str, latent: str = "latent_image", **inputs) -> dict:
+    """``f02`` with its KSampler (node 3) replaced by ``class_type``."""
+    prompt = _f02()
+    sampler = dict(prompt["3"]["inputs"], **inputs)
+    sampler[latent] = sampler.pop("latent_image")
+    prompt["3"] = {"class_type": class_type, "inputs": sampler}
+    return prompt
+
+
+SAMPLERS = (
+    ("ClownsharKSampler_Beta", "latent_image", "options"),
+    ("KSampler (Efficient)", "latent_image", "script"),
+    ("ttN KSampler_v2", "latent", "adv_xyPlot"),
+)
+
+
+class TestUpstreamVerifiedNodeSemantics(unittest.TestCase):
+    """R6 entries for nodes not installed here, read from their pinned sources.
+
+    Each source is the commit the maintainer's saved workflows name in the
+    node's ``cnr_id``/``ver`` properties (the report lists repository, commit,
+    file and line).
+    """
+
+    def test_ksampler_shaped_samplers_resolve(self):
+        for class_type, latent, _ in SAMPLERS:
+            with self.subTest(node=class_type):
+                result = _inspect({"prompt": _sampled(class_type, latent)})
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(result["sampler_id"], "3")
+                self.assertEqual(_texts(result), (parity.POS, parity.NEG))
+
+    def test_inputs_that_replace_the_prompt_are_not_followed(self):
+        # RES4LYF options may carry a guider; Efficiency scripts and the ttN
+        # XY plot re-run the sampler with other settings.
+        for class_type, latent, override in SAMPLERS:
+            with self.subTest(node=class_type):
+                prompt = _sampled(class_type, latent, **{override: ["30", 0]})
+                prompt["30"] = {"class_type": "SomeScript", "inputs": {}}
+                result = _inspect({"prompt": prompt})
+                self.assertEqual(result["reason_code"], "unsupported-node")
+                self.assertEqual(
+                    result["node"], {"id": "30", "class_type": "SomeScript"}
+                )
+
+    def test_chained_samplers_are_stages(self):
+        prompt = _sampled("ClownsharKSampler_Beta")
+        prompt["31"] = dict(prompt["3"], inputs=dict(prompt["3"]["inputs"]))
+        prompt["31"]["inputs"]["latent_image"] = ["3", 0]
+        prompt["8"]["inputs"]["samples"] = ["31", 0]
+        # Unanchored, two listed samplers are two branches, as for KSampler.
+        self.assertEqual(_inspect({"prompt": prompt})["state"], "ambiguous")
+        result = _inspect({"prompt": prompt}, output_node_id="9")
+        self.assertEqual(result["reason"], resolver.MULTISTAGE_UNATTRIBUTED_REASON)
+
+    def test_seed_variance_enhancer_passes_its_conditioning(self):
+        prompt = _f02(
+            **{
+                "20": {
+                    "class_type": "SeedVarianceEnhancer",
+                    "inputs": {"conditioning": ["6", 0], "seed": 1, "strength": 20.0},
+                }
+            }
+        )
+        prompt["3"]["inputs"]["positive"] = ["20", 0]
+        self.assertEqual(_texts(_inspect({"prompt": prompt})), (parity.POS, parity.NEG))
+
+    def test_overwritten_text_behind_seed_variance_is_still_refused(self):
+        # R1 before R6: the recon's 24 files differ from this only in names.
+        metadata = _overwritten()
+        metadata["prompt"]["20"] = {
+            "class_type": "SeedVarianceEnhancer",
+            "inputs": {"conditioning": ["17", 0], "seed": 1},
+        }
+        metadata["prompt"]["3"]["inputs"]["positive"] = ["20", 0]
+        result = _inspect(metadata)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertEqual(result["reason"], resolver.OVERWRITTEN_PROMPT_REASON)
+
+    def test_instantid_advanced_outputs(self):
+        prompt = _f02(
+            **{
+                "20": {
+                    "class_type": "ApplyInstantIDAdvanced",
+                    "inputs": {
+                        "model": ["4", 0],
+                        "positive": ["6", 0],
+                        "negative": ["7", 0],
+                        "ip_weight": 0.8,
+                    },
+                }
+            }
+        )
+        prompt["3"]["inputs"].update(positive=["20", 1], negative=["20", 2])
+        self.assertEqual(_texts(_inspect({"prompt": prompt})), (parity.POS, parity.NEG))
+        prompt["3"]["inputs"]["positive"] = ["20", 0]  # output 0 is the MODEL
+        self.assertEqual(
+            _inspect({"prompt": prompt})["reason_code"], "unsupported-node"
+        )
+
+    def test_crystools_conditioning_switch(self):
+        def switch(condition) -> dict:
+            prompt = _f02(
+                **{
+                    "11": _pm_encoder("a dog"),
+                    "20": {
+                        "class_type": "Switch conditioning [Crystools]",
+                        "inputs": {
+                            "on_true": ["6", 0],
+                            "on_false": ["11", 0],
+                            "boolean": condition,
+                        },
+                    },
+                    "21": {"class_type": "PrimitiveBoolean", "inputs": {"value": True}},
+                }
+            )
+            prompt["3"]["inputs"]["positive"] = ["20", 0]
+            return _inspect({"prompt": prompt})
+
+        self.assertEqual(_texts(switch(True))[0], parity.POS)
+        self.assertEqual(_texts(switch(False))[0], "a dog")
+        result = switch(["21", 0])
+        self.assertEqual(result["state"], "ambiguous")
+        self.assertEqual(
+            sorted(_texts(item)[0] for item in result["candidates"]),
+            sorted([parity.POS, "a dog"]),
+        )
+
+    def test_smz_encoder_reads_text_unless_sdxl_may_apply(self):
+        def smz(**inputs) -> dict:
+            node = {
+                "class_type": "smZ CLIPTextEncode",
+                "inputs": {
+                    "clip": ["4", 1],
+                    "text": "a cat",
+                    "text_g": "a cat",
+                    "text_l": "a cat",
+                    "with_SDXL": False,
+                    "parser": "comfy",
+                    **inputs,
+                },
+            }
+            return _inspect({"prompt": _f02(**{"6": node})})
+
+        self.assertEqual(_texts(smz(text_g="", text_l="x")), ("a cat", parity.NEG))
+        # with_SDXL: the connected CLIP decides at run time which inputs are read.
+        self.assertEqual(_texts(smz(with_SDXL=True)), ("a cat", parity.NEG))
+        for parts in ({"text_g": "x"}, {"text_l": ""}, {"text": "x"}):
+            with self.subTest(parts=parts):
+                result = smz(with_SDXL=True, **parts)
+                self.assertEqual(result["reason_code"], "runtime-value")
+                self.assertEqual(
+                    result["node"], {"id": "6", "class_type": "smZ CLIPTextEncode"}
+                )
+        linked = smz(with_SDXL=["21", 0])
+        self.assertEqual(linked["reason_code"], "unsupported-node")
+        for broken in ({"with_SDXL": "yes"}, {"with_SDXL": None}):
+            with self.subTest(flag=broken):
+                self.assertEqual(smz(**broken)["reason_code"], "malformed")
+
+    def test_string_sources(self):
+        cases = (
+            ("ttN text", "text"),
+            ("CR Text", "text"),
+            ("CR Prompt Text", "prompt"),
+            ("TextBoxMira", "text"),
+        )
+        for class_type, name in cases:
+            with self.subTest(node=class_type):
+                prompt = _f02(
+                    **{"10": {"class_type": class_type, "inputs": {name: parity.POS}}}
+                )
+                prompt["6"]["inputs"]["text"] = ["10", 0]
+                self.assertEqual(
+                    _texts(_inspect({"prompt": prompt})), (parity.POS, parity.NEG)
+                )
+                # Output 1 of the Comfyroll nodes is their help URL.
+                prompt["6"]["inputs"]["text"] = ["10", 1]
+                self.assertEqual(
+                    _inspect({"prompt": prompt})["reason_code"], "unsupported-node"
+                )
+
+    def test_default_profile_keeps_them_unsupported(self):
+        for class_type, latent, _ in SAMPLERS:
+            with self.subTest(node=class_type):
+                result = resolve_prompt_pair({"prompt": _sampled(class_type, latent)})
+                self.assertEqual(result["reason"], "No supported sampler pair found")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_inspection_outcomes.py

@@ -0,0 +1,486 @@
+"""Non-resolved prompt outcomes carry one short, truthful line per role."""
+
+import ast
+import json
+import unittest
+import unittest.mock
+from pathlib import Path
+from dataclasses import replace
+
+from utils import prompt_pair_resolver
+from utils.inspection_outcomes import (
+    _UNCODED_LINES,
+    CLASS_NAME_LENGTH,
+    REASON_CODES,
+    outcome_message,
+    project_resolution,
+)
+from utils.inspection_response import encode_inspection_response
+from utils.prompt_inspector import inspect_prompt_evidence
+from utils.prompt_pair_resolver import (
+    EMPTY_POSITIVE_REASON,
+    INSPECTION_PROFILE,
+    MULTISTAGE_UNATTRIBUTED_REASON,
+    SEMANTICS_UNAVAILABLE_REASON,
+)
+
+LIMITS = "Prompt exceeds inspection limits."
+
+
+def _encode(text: str) -> dict:
+    return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["9", 0]}}
+
+
+def _latent() -> dict:
+    return {
+        "class_type": "EmptyLatentImage",
+        "inputs": {"width": 8, "height": 8, "batch_size": 1},
+    }
+
+
+def _ksampler(positive: str = "7") -> dict:
+    return {
+        "class_type": "KSampler",
+        "inputs": {
+            "positive": [positive, 0],
+            "negative": ["6", 0],
+            "latent_image": ["4", 0],
+            "model": ["9", 0],
+            "seed": 1,
+            "steps": 2,
+            "cfg": 1.0,
+            "sampler_name": "euler",
+            "scheduler": "normal",
+            "denoise": 1.0,
+        },
+    }
+
+
+def _dual_cfg() -> dict:
+    return {
+        "1": _encode("positive"),
+        "6": _encode("negative"),
+        "2": {
+            "class_type": "DualCFGGuider",
+            "inputs": {
+                "cond1": ["1", 0],
+                "cond2": ["1", 0],
+                "negative": ["6", 0],
+                "model": ["9", 0],
+            },
+        },
+        "3": {
+            "class_type": "SamplerCustomAdvanced",
+            "inputs": {
+                "guider": ["2", 0],
+                "noise": ["9", 0],
+                "sampler": ["9", 0],
+                "sigmas": ["9", 0],
+                "latent_image": ["4", 0],
+            },
+        },
+        "4": _latent(),
+    }
+
+
+def _runtime() -> dict:
+    return {
+        "1": {"class_type": "PromptSearchList", "inputs": {"search": "cat"}},
+        "7": {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": ["1", 0], "clip": ["9", 0]},
+        },
+        "6": _encode("negative"),
+        "3": _ksampler(),
+        "4": _latent(),
+    }
+
+
+def _composite() -> dict:
+    graph = _runtime()
+    graph["7"] = {
+        "class_type": "ConditioningCombine",
+        "inputs": {"conditioning_1": ["6", 0], "conditioning_2": ["8", 0]},
+    }
+    graph["8"] = _encode("other")
+    return graph
+
+
+def _messages(evidence: dict) -> dict:
+    return {role: side.get("message") for role, side in evidence["prompts"].items()}
+
+
+def _statuses(evidence: dict) -> dict:
+    return {role: side["status"] for role, side in evidence["prompts"].items()}
+
+
+class TestOutcomeMessages(unittest.TestCase):
+    def assert_unresolved_with(self, evidence: dict, message: str) -> None:
+        self.assertEqual(
+            _statuses(evidence), {"positive": "unresolved", "negative": "unresolved"}
+        )
+        self.assertEqual(
+            _messages(evidence), {"positive": message, "negative": message}
+        )
+
+    def test_unsupported_node_status_names_the_class(self):
+        evidence = inspect_prompt_evidence({"prompt": _dual_cfg()})
+        self.assertEqual(evidence["resolution"]["reason_code"], "unsupported-node")
+        self.assert_unresolved_with(
+            evidence,
+            "Prompt uses a node the inspector does not support (DualCFGGuider).",
+        )
+
+    def test_runtime_and_composite_statuses(self):
+        self.assert_unresolved_with(
+            inspect_prompt_evidence({"prompt": _runtime()}),
+            "Prompt is chosen at run time (PromptSearchList).",
+        )
+        self.assert_unresolved_with(
+            inspect_prompt_evidence({"prompt": _composite()}),
+            "Prompt combines several texts (ConditioningCombine).",
+        )
+
+    def test_limit_status(self):
+        tight = replace(INSPECTION_PROFILE, per_candidate_visits=1, total_visits=1)
+        evidence = inspect_prompt_evidence({"prompt": _runtime()}, profile=tight)
+        self.assertEqual(evidence["resolution"]["reason_code"], "work-limit")
+        self.assert_unresolved_with(evidence, LIMITS)
+        for code in ("work-limit", "depth-limit", "text-limit", "graph-ceiling"):
+            self.assertEqual(
+                outcome_message({"state": "unresolved", "reason_code": code}), LIMITS
+            )
+
+    def test_every_closed_code_and_known_uncoded_reason_has_a_line(self):
+        for code in REASON_CODES:
+            message = outcome_message({"state": "unresolved", "reason_code": code})
+            self.assertIsInstance(message, str, code)
+            self.assertTrue(message.endswith("."), code)
+            self.assertNotIn("could not be resolved", message)
+        for reason, expected in (
+            (EMPTY_POSITIVE_REASON, "Positive prompt is empty."),
+            (
+                SEMANTICS_UNAVAILABLE_REASON,
+                "The inspector's node table is unavailable.",
+            ),
+            (
+                MULTISTAGE_UNATTRIBUTED_REASON,
+                "Several sampler stages were found; their prompts were not compared.",
+            ),
+            (
+                "Media metadata could not be read",
+                "Metadata in this file could not be read.",
+            ),
+        ):
+            resolution = {"state": "unresolved", "reason": reason, "reason_code": None}
+            self.assertEqual(outcome_message(resolution), expected)
+
+    def test_named_codes_without_a_class_omit_the_name(self):
+        for code, line in (
+            ("unsupported-node", "Prompt uses a node the inspector does not support."),
+            ("runtime-value", "Prompt is chosen at run time."),
+            ("composite", "Prompt combines several texts."),
+        ):
+            for node in (None, {"id": "7", "class_type": None}):
+                resolution = {"state": "unresolved", "reason_code": code, "node": node}
+                self.assertEqual(outcome_message(resolution), line)
+
+    def test_class_name_is_bounded(self):
+        resolution = {
+            "state": "unresolved",
+            "reason_code": "unsupported-node",
+            "node": {"id": "7", "class_type": "N" * 500},
+        }
+        self.assertEqual(
+            outcome_message(resolution),
+            "Prompt uses a node the inspector does not support "
+            f"({'N' * CLASS_NAME_LENGTH}).",
+        )
+        self.assertEqual(CLASS_NAME_LENGTH, 200)
+
+    def test_resolved_ambiguous_absent_and_unknown_outcomes_carry_no_line(self):
+        for resolution in (
+            {"state": "resolved", "reason_code": None},
+            {"state": "ambiguous", "reason_code": "multistage-differs"},
+            {"state": "ambiguous", "reason_code": None, "reason": "Multiple"},
+            {"state": "absent", "reason_code": None, "reason": "No parsed"},
+            {"state": "unresolved", "reason_code": "guess"},
+            {"state": "unresolved", "reason_code": None, "reason": "Anything else"},
+            None,
+        ):
+            self.assertIsNone(outcome_message(resolution), resolution)
+
+    def test_inspector_lines_for_no_sampler_malformed_and_empty_positive(self):
+        evidence = inspect_prompt_evidence({"prompt": {"1": _encode("x")}})
+        self.assert_unresolved_with(evidence, "No supported sampler was found.")
+        evidence = inspect_prompt_evidence({"prompt": "not-json"})
+        self.assert_unresolved_with(
+            evidence, "Prompt metadata in this file is malformed."
+        )
+        graph = _runtime()
+        graph["7"] = _encode("   ")
+        evidence = inspect_prompt_evidence({"prompt": graph})
+        self.assertEqual(evidence["resolution"]["reason"], EMPTY_POSITIVE_REASON)
+        self.assert_unresolved_with(evidence, "Positive prompt is empty.")
+
+    def test_a_resolved_pair_with_an_empty_positive_explains_only_that_role(self):
+        side = {"text": " ", "provenance": {"text_source": "graph-inferred"}}
+        negative = {"text": "kept", "provenance": {"text_source": "graph-inferred"}}
+        resolved = {
+            "state": "resolved",
+            "positive": side,
+            "negative": negative,
+            "sampler_id": "3",
+            "reason": "",
+            "reason_code": None,
+            "node": None,
+        }
+        with unittest.mock.patch(
+            "utils.prompt_inspector.resolve_prompt_pair", return_value=resolved
+        ):
+            evidence = inspect_prompt_evidence({"prompt": {}})
+        self.assertEqual(
+            evidence["prompts"]["positive"],
+            {
+                "status": "unresolved",
+                "text": None,
+                "source": None,
+                "message": "Positive prompt is empty.",
+            },
+        )
+        self.assertEqual(evidence["prompts"]["negative"]["text"], "kept")
+        self.assertNotIn("message", evidence["prompts"]["negative"])
+
+    def test_candidates_and_absence_keep_their_status_lines(self):
+        graph = _runtime()
+        graph["7"] = _encode("positive")
+        graph["5"] = _ksampler()
+        evidence = inspect_prompt_evidence({"prompt": graph})
+        self.assertEqual(evidence["resolution"]["state"], "ambiguous")
+        self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
+        evidence = inspect_prompt_evidence({})
+        self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
+        graph = _runtime()
+        graph["7"] = _encode("positive")
+        evidence = inspect_prompt_evidence({"prompt": graph})
+        self.assertEqual(evidence["resolution"]["state"], "resolved")
+        self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
+
+
+class TestResolutionProjection(unittest.TestCase):
+    def test_closed_codes_and_named_nodes_pass_unchanged(self):
+        for code in REASON_CODES:
+            resolution = {
+                "state": "unresolved",
+                "sampler_id": "3",
+                "reason": "r",
+                "reason_code": code,
+                "node": None,
+            }
+            self.assertEqual(project_resolution(resolution), resolution)
+        named = {
+            "state": "unresolved",
+            "sampler_id": None,
+            "reason": "r",
+            "reason_code": "composite",
+            "node": {"id": "7", "class_type": None},
+        }
+        self.assertEqual(project_resolution(named), named)
+
+    def test_unknown_codes_and_unbounded_nodes_fail_closed(self):
+        base = {"state": "unresolved", "sampler_id": None, "reason": "r"}
+        for code, node in (
+            ("guess", None),
+            (1, None),
+            ("malformed", {"id": "7", "class_type": "X"}),
+            (None, {"id": "7", "class_type": "X"}),
+            ("unsupported-node", {"id": 7, "class_type": "X"}),
+            ("unsupported-node", {"id": "", "class_type": "X"}),
+            ("unsupported-node", {"id": "7"}),
+            ("unsupported-node", {"id": "7", "class_type": 7}),
+            ("unsupported-node", {"id": "7", "class_type": "X", "extra": 1}),
+            ("unsupported-node", "X"),
+        ):
+            projected = project_resolution({**base, "reason_code": code, "node": node})
+            self.assertIsNone(projected["node"], (code, node))
+            if code not in REASON_CODES:
+                self.assertIsNone(projected["reason_code"], code)
+        for state in ("resolved", "absent"):
+            projected = project_resolution(
+                {**base, "state": state, "reason_code": "work-limit", "node": None}
+            )
+            self.assertIsNone(projected["reason_code"])
+        long_name = project_resolution(
+            {
+                **base,
+                "reason_code": "runtime-value",
+                "node": {"id": "7" * 300, "class_type": "N" * 300},
+            }
+        )["node"]
+        self.assertEqual(
+            long_name, {"id": "7" * CLASS_NAME_LENGTH, "class_type": "N" * 200}
+        )
+        unnamed = project_resolution(
+            {
+                **base,
+                "reason_code": "composite",
+                "node": {"id": "7", "class_type": ""},
+            }
+        )["node"]
+        self.assertEqual(unnamed, {"id": "7", "class_type": None})
+
+    def test_wire_carries_role_lines_and_projected_resolution(self):
+        evidence = inspect_prompt_evidence({"prompt": _dual_cfg()})
+        resolution = dict(evidence["resolution"], reason_code="guess")
+        body = json.loads(
+            encode_inspection_response(
+                image={"root_index": 0, "path": "a.png", "filename": "a.png"},
+                root_scope="a" * 64,
+                file={
+                    "observation": "current-file",
+                    "source_label": "output",
+                    "media_type": "image",
+                    "extension": ".png",
+                    "size": 1,
+                    "modified_time": 1,
+                },
+                prompts=evidence["prompts"],
+                resolution=resolution,
+                inspection=None,
+                library={"status": "none", "tags": []},
+            )
+        )
+        self.assertEqual(body["prompts"], evidence["prompts"])
+        self.assertIsNone(body["resolution"]["reason_code"])
+        self.assertIsNone(body["resolution"]["node"])
+
+
+# Uncoded unresolved reasons only the default (ingestion) profile emits; the
+# inspector never shows them, so they need no line.
+_DEFAULT_PROFILE_ONLY = frozenset(
+    {"Cyclic or excessive latent ancestry", "Resolution limit reached"}
+)
+
+
+def _uncoded_unresolved_reasons() -> set:
+    """Every literal reason the resolver can return as unresolved without a code.
+
+    Reads the source: ``_Unresolved(reason)`` without a code, and
+    ``_result("unresolved", ..., reason=...)`` without ``reason_code``. A reason
+    is a string literal or a module constant naming one.
+    """
+    source = Path(prompt_pair_resolver.__file__).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+
+    def literal(node):
+        if isinstance(node, ast.Constant) and isinstance(node.value, str):
+            return node.value
+        if isinstance(node, ast.Name):
+            value = getattr(prompt_pair_resolver, node.id, None)
+            return value if isinstance(value, str) else None
+        return None
+
+    found = set()
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name):
+            continue
+        keywords = {item.arg: item.value for item in node.keywords}
+        if node.func.id == "_Unresolved":
+            if len(node.args) < 2 and "code" not in keywords and node.args:
+                reason = literal(node.args[0])
+                if reason is None:
+                    raise AssertionError(f"line {node.lineno}: unreadable reason")
+                found.add(reason)
+        elif node.func.id == "_result" and node.args:
+            state = literal(node.args[0])
+            coded = "reason_code" in keywords or len(node.args) > 6
+            reason = keywords.get(
+                "reason", node.args[4] if len(node.args) > 4 else None
+            )
+            if state == "unresolved" and not coded and reason is not None:
+                text = literal(reason)
+                if text is None:
+                    raise AssertionError(f"line {node.lineno}: unreadable reason")
+                found.add(text)
+    return found
+
+
+class TestEveryUncodedOutcomeHasALine(unittest.TestCase):
+    """Task 2.7 review M1: a new uncoded unresolved reason fails here."""
+
+    def test_resolver_reasons_without_a_code_have_a_line(self):
+        reasons = _uncoded_unresolved_reasons()
+        self.assertIn(EMPTY_POSITIVE_REASON, reasons, "the scan sees the source")
+        missing = reasons - set(_UNCODED_LINES) - _DEFAULT_PROFILE_ONLY
+        self.assertEqual(missing, set(), "add a code or a line in inspection_outcomes")
+        # The allowance names only reasons the source still emits.
+        self.assertLessEqual(_DEFAULT_PROFILE_ONLY, reasons)
+
+    def test_default_profile_only_reasons_are_coded_in_inspection(self):
+        # The limit outcome is uncoded only on the default profile.
+        source = {"prompt": {"1": {"class_type": "KSampler", "inputs": {}}}}
+        for limited in ({"max_depth": 0}, {"max_text_length": 0}):
+            with self.subTest(**limited):
+                default = prompt_pair_resolver.resolve_prompt_pair(source, **limited)
+                inspected = prompt_pair_resolver.resolve_prompt_pair(
+                    source, profile=INSPECTION_PROFILE, **limited
+                )
+                self.assertEqual(default["reason"], "Resolution limit reached")
+                self.assertIsNone(default["reason_code"])
+                self.assertEqual(inspected["reason"], "Resolution limit reached")
+                self.assertIsNotNone(inspected["reason_code"])
+
+
+class TestRoleMessageBound(unittest.TestCase):
+    """Task 2.7 review M4: the encoder bounds role lines as the client does."""
+
+    def encode(self, prompts: dict) -> dict:
+        return json.loads(
+            encode_inspection_response(
+                image={"root_index": 0, "path": "a.png", "filename": "a.png"},
+                root_scope="a" * 64,
+                file={
+                    "observation": "current-file",
+                    "source_label": "output",
+                    "media_type": "image",
+                    "extension": ".png",
+                    "size": 1,
+                    "modified_time": 1,
+                },
+                prompts=prompts,
+                resolution={"state": "unresolved", "reason": "x"},
+                inspection=None,
+                library={"status": "none", "tags": []},
+            )
+        )["prompts"]
+
+    def test_role_messages_are_bounded_text_on_unresolved_roles(self):
+        side = {"status": "unresolved", "text": None, "source": None}
+        prompts = self.encode(
+            {
+                "positive": {**side, "message": "x" * 600},
+                "negative": {**side, "message": "\U0001f600" * 300},
+            }
+        )
+        self.assertEqual(prompts["positive"]["message"], "x" * 512)
+        # The client counts UTF-16 code units.
+        self.assertLessEqual(
+            len(prompts["negative"]["message"].encode("utf-16-le")), 1024
+        )
+        self.assertEqual(prompts["negative"]["message"], "\U0001f600" * 256)
+        for message in ("", None, 7):
+            with self.subTest(message=message):
+                prompts = self.encode(
+                    {"positive": {**side, "message": message}, "negative": side}
+                )
+                self.assertNotIn("message", prompts["positive"])
+        resolved = {"status": "resolved", "text": "a", "source": "graph-inferred"}
+        prompts = self.encode(
+            {"positive": {**resolved, "message": "line"}, "negative": side}
+        )
+        self.assertNotIn("message", prompts["positive"])
+        self.assertEqual(prompts["negative"], side)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_inspection_semantics.py

@@ -0,0 +1,838 @@
+"""The declarative inspection node semantics table and its loader (#240, spec 7).
+
+``web/js/inspection-semantics.json`` is the single table. Python validates it at
+import and fails loudly; the browser facts tables are checked against its
+``facts`` section by ``tests/js/inspection-semantics-sync.test.js``.
+"""
+
+import copy
+import json
+import tempfile
+import unittest
+import unittest.mock
+from pathlib import Path
+from types import MappingProxyType
+
+from utils import inspection_semantics as semantics
+from utils.inspection_semantics import SemanticsError, load_semantics
+
+REPO = Path(__file__).resolve().parents[1]
+TABLE = REPO / "web" / "js" / "inspection-semantics.json"
+
+
+def _minimal():
+    """A small valid table; mutation tests break exactly one thing in a copy."""
+    return {
+        "schema": 1,
+        "samplers": {
+            "KSampler": {
+                "strategy": "ksampler",
+                "conditioning": {"positive": "positive", "negative": "negative"},
+                "latent": "latent_image",
+                "facts": "ksampler",
+            }
+        },
+        "guiders": {"BasicGuider": {"strategy": "basic", "positive": "conditioning"}},
+        "conditioning": [
+            {
+                "class": "FluxGuidance",
+                "slot": 0,
+                "strategy": "passthrough",
+                "input": "conditioning",
+            }
+        ],
+        "selectors": {},
+        "composites": {},
+        "encoders": [
+            {
+                "class": "CLIPTextEncode",
+                "slot": 0,
+                "strategy": "single",
+                "inputs": ["text"],
+            }
+        ],
+        "strings": [],
+        "runtime": [],
+        "savers": {},
+        "non_file_outputs": [],
+        "stage_inputs": {"shaped": ["positive"], "with_model": ["samples"]},
+        "facts": {
+            "configuration_outputs": {},
+            "model_outputs": {},
+            "model_passthroughs": [],
+            "scalar_outputs": {},
+        },
+    }
+
+
+class _TableCase(unittest.TestCase):
+    def load_text(self, text):
+        # Windows: the file handle is closed before the directory is removed.
+        with tempfile.TemporaryDirectory() as root:
+            path = Path(root) / "semantics.json"
+            path.write_text(text, encoding="utf-8")
+            return load_semantics(path)
+
+    def load_doc(self, doc):
+        return self.load_text(json.dumps(doc))
+
+    def assertRejected(self, doc, fragment):
+        with self.assertRaises(SemanticsError) as caught:
+            self.load_doc(doc)
+        self.assertIn(fragment, str(caught.exception))
+
+
+class SpecEntriesTest(unittest.TestCase):
+    def test_semantics_file_validates_and_lists_spec_entries(self):
+        table = load_semantics()
+
+        self.assertEqual(semantics.SEMANTICS_PATH, TABLE)
+        self.assertEqual(
+            semantics.SEMANTICS_PATH,
+            Path(semantics.__file__).resolve().parents[1]
+            / "web"
+            / "js"
+            / "inspection-semantics.json",
+        )
+
+        pair = {"positive": "positive", "negative": "negative"}
+        for name in ("KSampler", "KSamplerAdvanced"):
+            self.assertEqual(
+                table.samplers[name],
+                {
+                    "strategy": "ksampler",
+                    "conditioning": pair,
+                    "latent": "latent_image",
+                    "facts": "ksampler",
+                },
+            )
+        self.assertEqual(
+            table.samplers["SamplerCustom"],
+            {
+                "strategy": "ksampler",
+                "conditioning": pair,
+                "latent": "latent_image",
+                "facts": "custom",
+            },
+        )
+        self.assertEqual(
+            table.samplers["SamplerCustomAdvanced"],
+            {
+                "strategy": "custom-advanced",
+                "conditioning": {"guider": "guider"},
+                "latent": "latent_image",
+                "facts": "custom-advanced",
+            },
+        )
+        # D19: FaceDetailer is a stage over its upstream image, ksampler-shaped facts.
+        # A non-empty wildcard replaces the detail prompt (Impact core.py:267).
+        self.assertEqual(
+            table.samplers["FaceDetailer"],
+            {
+                "strategy": "detailer-stage",
+                "conditioning": pair,
+                "image_stage": "image",
+                "facts": "ksampler",
+                "prompt_overrides": ("wildcard",),
+            },
+        )
+
+        self.assertEqual(
+            table.guiders["CFGGuider"],
+            {"strategy": "cfg", "positive": "positive", "negative": "negative"},
+        )
+        self.assertEqual(
+            table.guiders["BasicGuider"],
+            {
+                "strategy": "basic",
+                "positive": "conditioning",
+                "absent_reason": "Guider has no negative input",
+            },
+        )
+        # D17: DualCFGGuider has two positives and stays an unsupported node.
+        self.assertNotIn("DualCFGGuider", table.guiders)
+        self.assertNotIn("DualCFGGuider", table.all_classes)
+
+        single_input = (
+            "FluxGuidance",
+            "ReferenceLatent",
+            "ConditioningSetArea",
+            "ConditioningSetAreaPercentage",
+            "ConditioningSetAreaStrength",
+            "ConditioningSetMask",
+            "ConditioningSetTimestepRange",
+            "ControlNetApply",
+        )
+        for name in single_input:
+            self.assertEqual(
+                table.conditioning[(name, 0)],
+                {"strategy": "passthrough", "input": "conditioning"},
+            )
+            self.assertNotIn((name, 1), table.conditioning)
+        for name in (
+            "ControlNetApplyAdvanced",
+            "ControlNetApplySD3",
+            "InpaintModelConditioning",
+            "WanImageToVideo",
+        ):
+            self.assertEqual(
+                table.conditioning[(name, 0)],
+                {"strategy": "passthrough", "input": "positive"},
+            )
+            self.assertEqual(
+                table.conditioning[(name, 1)],
+                {"strategy": "passthrough", "input": "negative"},
+            )
+            # Slot 2 (a latent) is not conditioning; spec 7.4 handles origins.
+            self.assertNotIn((name, 2), table.conditioning)
+        self.assertEqual(
+            table.conditioning[("ConditioningZeroOut", 0)],
+            {"strategy": "zeroed", "input": "conditioning"},
+        )
+
+        self.assertEqual(
+            table.selectors["easy ifElse"],
+            {
+                "strategy": "boolean-select",
+                "condition": "boolean",
+                "branches": {"true": "on_true", "false": "on_false"},
+            },
+        )
+
+        self.assertEqual(
+            table.composites["ConditioningCombine"],
+            {"strategy": "combine", "inputs": ("conditioning_1", "conditioning_2")},
+        )
+        for name in ("ConditioningConcat", "ConditioningAverage"):
+            self.assertEqual(
+                table.composites[name],
+                {
+                    "strategy": "combine",
+                    "inputs": ("conditioning_to", "conditioning_from"),
+                },
+            )
+
+        encoders = {
+            ("CLIPTextEncode", 0): ("single", ("text",)),
+            ("CLIPTextEncodeSDXL", 0): ("parts-equal", ("text_g", "text_l")),
+            ("CLIPTextEncodeFlux", 0): ("parts-equal", ("clip_l", "t5xxl")),
+            ("CLIPTextEncodeSD3", 0): ("parts-equal", ("clip_l", "clip_g", "t5xxl")),
+            ("TextEncodeQwenImageEdit", 0): ("single", ("prompt",)),
+            ("TextEncodeQwenImageEditPlus", 0): ("single", ("prompt",)),
+            ("TextEncodeQwenImage21", 0): ("single", ("prompt",)),
+            ("TextEncodeQwenImage21", 1): ("single", ("negative_prompt",)),
+            # D11: a plain text encoder, no composition.
+            ("KikoTextEncode", 0): ("single", ("text",)),
+        }
+        for key, (strategy, inputs) in encoders.items():
+            self.assertEqual(
+                table.encoders[key], {"strategy": strategy, "inputs": inputs}, key
+            )
+        self.assertNotIn(("TextEncodeQwenImage21", 2), table.encoders)
+
+        self.assertEqual(
+            table.strings[("StringConstant", 0)],
+            {"strategy": "literal", "inputs": ("string",)},
+        )
+        self.assertEqual(
+            table.strings[("StringConstantMultiline", 0)],
+            {
+                "strategy": "strip-newlines",
+                "inputs": ("string",),
+                "params": {"flag": "strip_newlines"},
+            },
+        )
+        self.assertEqual(
+            table.strings[("JoinStringMulti", 0)],
+            {
+                "strategy": "join-multi",
+                "inputs": ("string_1",),
+                "params": {
+                    "count": "inputcount",
+                    "delimiter": "delimiter",
+                    "list_flag": "return_list",
+                    "input_prefix": "string_",
+                },
+            },
+        )
+
+        # D18: decided at run time.
+        self.assertEqual(table.runtime, frozenset({"PromptSearchList"}))
+
+        comfy = "{name}_{counter:05}_.{ext}"
+        for name, image in (
+            ("SaveImage", "images"),
+            ("SaveAnimatedWEBP", "images"),
+            ("SaveAnimatedPNG", "images"),
+            ("SaveVideo", "video"),
+            ("SaveWEBM", "images"),
+        ):
+            self.assertEqual(
+                table.savers[name],
+                {
+                    "strategy": "comfy-counter",
+                    "prefix_input": "filename_prefix",
+                    "image_input": image,
+                    "pattern": comfy,
+                },
+            )
+        self.assertEqual(
+            table.savers["VHS_VideoCombine"],
+            {
+                "strategy": "vhs-counter",
+                "prefix_input": "filename_prefix",
+                "image_input": "images",
+                "pattern": "{name}_{counter:05}.{ext}",
+            },
+        )
+        # Until its naming is verified in pinned source it stays terminal-unknown.
+        self.assertNotIn("KikoSaveImage", table.all_classes)
+        self.assertEqual(
+            table.non_file_outputs,
+            frozenset(
+                {
+                    "PreviewImage",
+                    "PreviewAny",
+                    "SaveImageWebsocket",
+                    "Image Comparer (rgthree)",
+                }
+            ),
+        )
+
+        # Not listed this slice: they stay unsupported-node with the class named.
+        for name in (
+            "ImpactWildcardProcessor",
+            "RegexReplace",
+            "LoRAfromText",
+            "XYZPrompt",
+        ):
+            self.assertNotIn(name, table.all_classes)
+        # Existing code adapters stay in code, unchanged.
+        for name in ("PromptManager", "PromptManagerPair", "PromptManagerText"):
+            self.assertNotIn(name, table.all_classes)
+
+    def test_legacy_parser_ports_are_listed_with_their_source_inputs(self):
+        # Task 2.7b (R6): input names and slots read from each node's source;
+        # the report lists the file and line for every one.
+        table = load_semantics()
+        pair = {"positive": "positive", "negative": "negative"}
+        for name, image in (
+            ("DetailerForEach", "image"),
+            ("UltimateSDUpscale", "image"),
+            ("UltimateSDUpscaleNoUpscale", "upscaled_image"),
+        ):
+            entry = dict(table.samplers[name])
+            self.assertEqual(
+                entry.pop("prompt_overrides", None) is not None,
+                name == "DetailerForEach",
+                name,
+            )
+            self.assertEqual(
+                entry,
+                {
+                    "strategy": "detailer-stage",
+                    "conditioning": pair,
+                    "image_stage": image,
+                    "facts": "ksampler",
+                },
+                name,
+            )
+        for (name, slot), source in {
+            ("ControlNetInpaintingAliMamaApply", 0): "positive",
+            ("ControlNetInpaintingAliMamaApply", 1): "negative",
+            ("LTXVConditioning", 0): "positive",
+            ("LTXVConditioning", 1): "negative",
+            ("FluxKontextMultiReferenceLatentMethod", 0): "conditioning",
+        }.items():
+            self.assertEqual(
+                table.conditioning[(name, slot)],
+                {"strategy": "passthrough", "input": source},
+            )
+        self.assertNotIn(
+            ("FluxKontextMultiReferenceLatentMethod", 1), table.conditioning
+        )
+        self.assertEqual(
+            table.encoders[("Text to Conditioning", 0)],
+            {"strategy": "single", "inputs": ("text",)},
+        )
+        for name in ("TextInput", "DisplayText"):
+            self.assertEqual(
+                table.strings[(name, 0)], {"strategy": "literal", "inputs": ("text",)}
+            )
+
+    def test_stage_inputs_are_the_one_source_for_both_stage_walks(self):
+        # Review I-2: Python's resolver and comfy-metadata.js read these lists here.
+        stage_inputs = load_semantics().stage_inputs
+        self.assertEqual(
+            stage_inputs["shaped"],
+            (
+                "latent_image",
+                "guider",
+                "positive",
+                "negative",
+                "pipe",
+                "basic_pipe",
+                "detailer_pipe",
+                "sampler",
+                "sigmas",
+                "noise",
+                "text_embeds",
+                "hyvid_embeds",
+            ),
+        )
+        self.assertEqual(
+            stage_inputs["with_model"],
+            ("conditioning", "neg_conditioning", "samples", "latent"),
+        )
+        # The default profile's own sampler-shape names stay a subset of the table's.
+        from utils.prompt_pair_resolver import _SAMPLER_LIKE_INPUTS
+
+        self.assertLessEqual(set(_SAMPLER_LIKE_INPUTS), set(stage_inputs["shaped"]))
+
+    def test_facts_section_mirrors_the_browser_tables(self):
+        facts = load_semantics().facts
+        self.assertEqual(
+            facts["configuration_outputs"],
+            {
+                "RandomNoise": {"kind": "noise", "fields": ("noise_seed",)},
+                "KSamplerSelect": {"kind": "sampler", "fields": ("sampler_name",)},
+                "BasicScheduler": {"kind": "sigmas", "fields": ("steps", "scheduler")},
+                # D8: Flux2Scheduler has no scheduler input; its type names it.
+                "Flux2Scheduler": {
+                    "kind": "sigmas",
+                    "fields": ("steps",),
+                    "type_fields": ("scheduler",),
+                },
+                "CFGGuider": {"kind": "guider", "fields": ("cfg", "model")},
+                "BasicGuider": {"kind": "guider", "fields": ("model",)},
+            },
+        )
+        self.assertEqual(
+            facts["model_outputs"],
+            {
+                "CheckpointLoader": "ckpt_name",
+                "CheckpointLoaderSimple": "ckpt_name",
+                "unCLIPCheckpointLoader": "ckpt_name",
+                "UNETLoader": "unet_name",
+            },
+        )
+        self.assertEqual(
+            facts["model_passthroughs"],
+            (
+                "LoraLoader",
+                "LoraLoaderModelOnly",
+                "ModelSamplingFlux",
+                "ModelSamplingAuraFlow",
+                "ModelSamplingSD3",
+                # Task 2.7b (R9): model in, model out, verified in their sources.
+                "Lora Loader (LoraManager)",
+                "Power Lora Loader (rgthree)",
+                "ModelPatchTorchSettings",
+                "CFGNorm",
+                "PerturbedAttentionGuidance",
+                "CR Apply LoRA Stack",
+            ),
+        )
+        self.assertEqual(
+            facts["scalar_outputs"],
+            {
+                "PrimitiveInt": "number",
+                "PrimitiveFloat": "number",
+                "PrimitiveString": "string",
+                "PrimitiveStringMultiline": "string",
+            },
+        )
+
+    def test_table_is_loaded_lazily_and_once(self):
+        # Importing the module reads nothing; the inspection path asks for it.
+        self.assertFalse(hasattr(semantics, "SEMANTICS"))
+        semantics.get_semantics.cache_clear()
+        first = semantics.get_semantics()
+        self.assertIs(first, semantics.get_semantics())
+        self.assertEqual(first, load_semantics())
+
+    def test_loaded_table_is_frozen(self):
+        table = load_semantics()
+        self.assertIsInstance(table.samplers, MappingProxyType)
+        self.assertIsInstance(table.samplers["KSampler"], MappingProxyType)
+        with self.assertRaises(TypeError):
+            table.samplers["Other"] = {}
+        with self.assertRaises(TypeError):
+            table.samplers["KSampler"]["latent"] = "x"
+        with self.assertRaises(AttributeError):
+            table.samplers = {}
+        self.assertIsInstance(table.composites["ConditioningCombine"]["inputs"], tuple)
+
+
+class ValidationTest(_TableCase):
+    def test_minimal_table_loads(self):
+        table = self.load_doc(_minimal())
+        self.assertEqual(table.encoders[("CLIPTextEncode", 0)]["inputs"], ("text",))
+        self.assertEqual(table.runtime, frozenset())
+
+    def test_unknown_strategy_rejected(self):
+        for category, mutate in (
+            ("samplers", lambda d: d["samplers"]["KSampler"].update(strategy="dual")),
+            # A known strategy of another category is still unknown here.
+            ("conditioning", lambda d: d["conditioning"][0].update(strategy="single")),
+            ("encoders", lambda d: d["encoders"][0].update(strategy="guess")),
+            ("guiders", lambda d: d["guiders"]["BasicGuider"].update(strategy=None)),
+        ):
+            with self.subTest(category=category):
+                doc = _minimal()
+                mutate(doc)
+                self.assertRejected(doc, "unknown strategy")
+
+    def test_strategies_are_the_closed_spec_set(self):
+        self.assertEqual(
+            dict(semantics.STRATEGIES),
+            {
+                "samplers": frozenset(
+                    {"ksampler", "custom-advanced", "detailer-stage"}
+                ),
+                "guiders": frozenset({"cfg", "basic"}),
+                "conditioning": frozenset({"passthrough", "zeroed"}),
+                "selectors": frozenset({"boolean-select"}),
+                "composites": frozenset({"combine"}),
+                "encoders": frozenset(
+                    {"single", "parts-equal", "pm-compose", "flag-parts"}
+                ),
+                "strings": frozenset(
+                    {"literal", "strip-newlines", "join-multi", "concat"}
+                ),
+                "savers": frozenset({"comfy-counter", "vhs-counter"}),
+            },
+        )
+        with self.assertRaises(TypeError):
+            semantics.STRATEGIES["samplers"] = frozenset()
+
+    def test_duplicate_output_key_rejected(self):
+        doc = _minimal()
+        doc["encoders"].append(copy.deepcopy(doc["encoders"][0]))
+        self.assertRejected(doc, "duplicate")
+
+        doc = _minimal()
+        doc["conditioning"].append(
+            {
+                "class": "CLIPTextEncode",
+                "slot": 0,
+                "strategy": "passthrough",
+                "input": "conditioning",
+            }
+        )
+        self.assertRejected(doc, "duplicate")
+
+        doc = _minimal()
+        doc["runtime"] = ["KSampler"]
+        self.assertRejected(doc, "duplicate")
+
+        doc = _minimal()
+        doc["non_file_outputs"] = ["PreviewImage", "PreviewImage"]
+        self.assertRejected(doc, "duplicate")
+
+        # JSON objects with a repeated key would silently keep the last one.
+        text = json.dumps(_minimal())
+        entry = json.dumps(_minimal()["samplers"]["KSampler"])
+        text = text.replace(
+            '"samplers": {', '"samplers": {"KSampler": ' + entry + ", ", 1
+        )
+        with self.assertRaises(SemanticsError) as caught:
+            self.load_text(text)
+        self.assertIn("duplicate", str(caught.exception))
+
+    def test_slot_must_be_nonnegative_int(self):
+        for slot in (-1, 1.5, True, "0", None, 2**31):
+            with self.subTest(slot=slot):
+                doc = _minimal()
+                doc["encoders"][0]["slot"] = slot
+                self.assertRejected(doc, "slot")
+        doc = _minimal()
+        del doc["conditioning"][0]["slot"]
+        self.assertRejected(doc, "slot")
+
+    def test_missing_input_name_rejected(self):
+        cases = (
+            lambda d: d["samplers"]["KSampler"]["conditioning"].pop("negative"),
+            lambda d: d["samplers"]["KSampler"].pop("latent"),
+            lambda d: d["guiders"]["BasicGuider"].update(positive=""),
+            lambda d: d["conditioning"][0].pop("input"),
+            lambda d: d["encoders"][0].update(inputs=[]),
+            lambda d: d["encoders"][0].update(inputs=["text", "text"]),
+            lambda d: d["encoders"][0].update(inputs=[3]),
+            lambda d: d["encoders"][0].update(inputs="text"),
+        )
+        for index, mutate in enumerate(cases):
+            with self.subTest(case=index):
+                doc = _minimal()
+                mutate(doc)
+                self.assertRejected(doc, "input")
+
+    def test_entry_shape_is_strict(self):
+        cases = (
+            (lambda d: d.update(extra={}), "category"),
+            (lambda d: d.pop("guiders"), "category"),
+            (lambda d: d.update(schema=2), "schema"),
+            (lambda d: d["samplers"]["KSampler"].update(note="x"), "field"),
+            (lambda d: d["samplers"]["KSampler"].update(facts="flux"), "facts"),
+            (lambda d: d["guiders"]["BasicGuider"].update(negative="n"), "field"),
+            (lambda d: d["encoders"][0].update({"class": ""}), "class"),
+            (lambda d: d["facts"].update(extra={}), "facts"),
+            (
+                lambda d: d["facts"]["scalar_outputs"].update(PrimitiveInt="int"),
+                "facts",
+            ),
+            (lambda d: d.pop("stage_inputs"), "category"),
+            (lambda d: d["stage_inputs"].update(extra=["x"]), "stage_inputs"),
+            (lambda d: d["stage_inputs"].pop("with_model"), "stage_inputs"),
+            (lambda d: d["stage_inputs"].update(shaped=[]), "input name"),
+            (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["facts"]["configuration_outputs"].update(
+                    X={"kind": "sigmas", "fields": ["steps"], "type_fields": []}
+                ),
+                "input name",
+            ),
+            (
+                lambda d: d["facts"]["configuration_outputs"].update(
+                    X={"kind": "sigmas", "fields": ["steps"], "type_fields": ["steps"]}
+                ),
+                "type_fields",
+            ),
+            (
+                lambda d: d["facts"]["configuration_outputs"].update(
+                    X={"kind": "sigmas", "fields": ["steps"], "names": ["x"]}
+                ),
+                "field",
+            ),
+        )
+        for index, (mutate, fragment) in enumerate(cases):
+            with self.subTest(case=index):
+                doc = _minimal()
+                mutate(doc)
+                self.assertRejected(doc, fragment)
+
+    def test_strategy_specific_requirements(self):
+        doc = _minimal()
+        doc["samplers"]["X"] = {
+            "strategy": "custom-advanced",
+            "conditioning": {"positive": "positive", "negative": "negative"},
+            "latent": "latent_image",
+            "facts": "custom-advanced",
+        }
+        self.assertRejected(doc, "input")
+
+        doc = _minimal()
+        doc["samplers"]["D"] = {
+            "strategy": "detailer-stage",
+            "conditioning": {"positive": "positive", "negative": "negative"},
+            "facts": "ksampler",
+        }
+        self.assertRejected(doc, "input")
+
+        doc = _minimal()
+        doc["guiders"]["G"] = {"strategy": "cfg", "positive": "positive"}
+        self.assertRejected(doc, "input")
+
+        doc = _minimal()
+        doc["encoders"][0].update(strategy="parts-equal", inputs=["text_g"])
+        self.assertRejected(doc, "input")
+
+        doc = _minimal()
+        doc["savers"]["S"] = {
+            "strategy": "comfy-counter",
+            "prefix_input": "filename_prefix",
+            "image_input": "images",
+            "pattern": "{name}_{counter:05}.{ext}",
+        }
+        self.assertRejected(doc, "pattern")
+
+        doc = _minimal()
+        doc["selectors"]["easy ifElse"] = {
+            "strategy": "boolean-select",
+            "condition": "boolean",
+            "branches": {"true": "on_true"},
+        }
+        self.assertRejected(doc, "input")
+
+        doc = _minimal()
+        doc["strings"].append(
+            {
+                "class": "JoinStringMulti",
+                "slot": 0,
+                "strategy": "join-multi",
+                "inputs": ["string_1"],
+                "params": {"count": "inputcount"},
+            }
+        )
+        self.assertRejected(doc, "params")
+
+    def test_unreadable_table_fails_loudly(self):
+        with self.assertRaises(SemanticsError):
+            self.load_text("{not json")
+        with self.assertRaises(SemanticsError):
+            self.load_text("[]")
+        with self.assertRaises(SemanticsError):
+            self.load_text(
+                json.dumps(_minimal()).replace('"schema": 1', '"schema": NaN')
+            )
+        with tempfile.TemporaryDirectory() as root:
+            with self.assertRaises(SemanticsError):
+                load_semantics(Path(root) / "missing.json")
+
+    def test_oversized_table_is_rejected_before_parsing(self):
+        text = json.dumps(_minimal()) + " " * semantics.MAX_TABLE_BYTES
+        with self.assertRaises(SemanticsError) as caught:
+            self.load_text(text)
+        self.assertIn("exceeds", str(caught.exception))
+
+    def test_deeply_nested_table_is_a_semantics_error(self):
+        # Past the nesting bound, and far past the interpreter's recursion limit.
+        for depth in (semantics.MAX_NESTING + 1, 100_000):
+            with self.subTest(depth=depth):
+                nested = "[" * depth + "]" * depth
+                text = json.dumps(_minimal()).replace(
+                    '"selectors": {}', '"selectors": ' + nested
+                )
+                with self.assertRaises(SemanticsError) as caught:
+                    self.load_text(text)
+                self.assertIn("nest", str(caught.exception))
+        self.assertGreater(semantics.MAX_NESTING, _depth(json.loads(TABLE.read_text())))
+        # Whichever interpreter limit the parser hits, it is a SemanticsError.
+        with unittest.mock.patch.object(
+            semantics.json, "loads", side_effect=RecursionError
+        ):
+            with self.assertRaises(SemanticsError) as caught:
+                self.load_doc(_minimal())
+        self.assertIn("nests deeper", str(caught.exception))
+
+
+def _depth(value, level=1):
+    children = value.values() if isinstance(value, dict) else value
+    if not isinstance(value, (dict, list)):
+        return level - 1
+    return max([level] + [_depth(item, level + 1) for item in children])
+
+
+class SamplerOptionsTest(_TableCase):
+    """Task 2.7b: optional sampler lists are input names, or the table fails."""
+
+    def test_upstream_verified_ports_are_listed_with_their_source_inputs(self):
+        # Not installed here: read from the pinned upstream commits the saved
+        # workflows name (the report lists repository, commit, file and line).
+        table = load_semantics()
+        pair = {"positive": "positive", "negative": "negative"}
+        for name, latent, prompt_overrides, fact_overrides in (
+            (
+                "ClownsharKSampler_Beta",
+                "latent_image",
+                ("options",),
+                ("options", "sigmas"),
+            ),
+            ("KSampler (Efficient)", "latent_image", ("script",), ("script",)),
+            ("ttN KSampler_v2", "latent", ("adv_xyPlot",), ("adv_xyPlot",)),
+        ):
+            self.assertEqual(
+                table.samplers[name],
+                {
+                    "strategy": "ksampler",
+                    "conditioning": pair,
+                    "latent": latent,
+                    "facts": "ksampler",
+                    "prompt_overrides": prompt_overrides,
+                    "fact_overrides": fact_overrides,
+                },
+                name,
+            )
+        for (name, slot), source in {
+            ("SeedVarianceEnhancer", 0): "conditioning",
+            ("ApplyInstantIDAdvanced", 1): "positive",
+            ("ApplyInstantIDAdvanced", 2): "negative",
+        }.items():
+            self.assertEqual(
+                table.conditioning[(name, slot)],
+                {"strategy": "passthrough", "input": source},
+            )
+        self.assertNotIn(("ApplyInstantIDAdvanced", 0), table.conditioning)
+        self.assertEqual(
+            table.selectors["Switch conditioning [Crystools]"],
+            table.selectors["easy ifElse"],
+        )
+        self.assertEqual(
+            table.encoders[("smZ CLIPTextEncode", 0)],
+            {
+                "strategy": "flag-parts",
+                "inputs": ("text", "text_g", "text_l"),
+                "params": {"flag": "with_SDXL"},
+            },
+        )
+        for name, source in (
+            ("ttN text", "text"),
+            ("CR Text", "text"),
+            ("CR Prompt Text", "prompt"),
+            ("TextBoxMira", "text"),
+        ):
+            self.assertEqual(
+                table.strings[(name, 0)], {"strategy": "literal", "inputs": (source,)}
+            )
+            self.assertNotIn((name, 1), table.strings)
+
+    def test_flag_parts_needs_a_flag_and_two_inputs(self):
+        doc = _minimal()
+        entry = {
+            "class": "smZ CLIPTextEncode",
+            "slot": 0,
+            "strategy": "flag-parts",
+            "inputs": ["text", "text_g"],
+            "params": {"flag": "with_SDXL"},
+        }
+        doc["encoders"].append(entry)
+        self.assertEqual(
+            self.load_doc(doc).encoders[("smZ CLIPTextEncode", 0)]["params"],
+            {"flag": "with_SDXL"},
+        )
+        for change in (
+            {"params": {}},
+            {"params": {"flag": "x", "other": "y"}},
+            {"params": {"flag": ""}},
+            {"params": None},
+            {"inputs": ["text"]},
+            {"compose": True},
+        ):
+            with self.subTest(change=change):
+                broken = _minimal()
+                broken["encoders"].append({**entry, **change})
+                with self.assertRaises(SemanticsError):
+                    self.load_doc(broken)
+        broken = _minimal()
+        broken["encoders"].append(
+            {key: item for key, item in entry.items() if key != "params"}
+        )
+        with self.assertRaises(SemanticsError):
+            self.load_doc(broken)
+
+    def test_prompt_overrides_are_input_names(self):
+        doc = _minimal()
+        doc["samplers"]["KSampler"]["prompt_overrides"] = ["wildcard"]
+        sampler = self.load_doc(doc).samplers["KSampler"]
+        self.assertEqual(sampler["prompt_overrides"], ("wildcard",))
+        for value in ([], "wildcard", [""], ["a", "a"], [1]):
+            with self.subTest(value=value):
+                doc["samplers"]["KSampler"]["prompt_overrides"] = value
+                with self.assertRaises(SemanticsError):
+                    self.load_doc(doc)
+        doc["samplers"]["KSampler"].pop("prompt_overrides")
+        doc["samplers"]["KSampler"]["fact_overrides"] = ["script", "sigmas"]
+        sampler = self.load_doc(doc).samplers["KSampler"]
+        self.assertEqual(sampler["fact_overrides"], ("script", "sigmas"))
+        doc["samplers"]["KSampler"]["fact_overrides"] = ["script", "script"]
+        with self.assertRaises(SemanticsError):
+            self.load_doc(doc)
+        doc["samplers"]["KSampler"].pop("fact_overrides")
+        doc["samplers"]["KSampler"]["overrides"] = ["wildcard"]
+        self.assertRejected(doc, "unknown field")
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/vendor/frontend-coverage-baseline.json

@@ -19,7 +19,7 @@
   },
   "comfy-metadata.js": {
     "lines": 97,
-    "branches": 91,
+    "branches": 92,
     "functions": 98,
     "statements": 97
   },

utils/inspection_outcomes.py

@@ -0,0 +1,111 @@
+"""User-facing lines for non-resolved prompt outcomes, from the closed vocabulary.
+
+The inspector shows one line per unresolved role in place of the generic
+"Prompt could not be resolved." Ambiguous and absent roles keep their status
+copy, so a candidate is never presented as a confirmed prompt. The wire
+projection accepts exactly the closed ``reason_code`` vocabulary and names a
+blocking node only beside a code that names one.
+"""
+
+from typing import Any, Optional
+
+from .prompt_pair_resolver import (
+    EMPTY_POSITIVE_REASON,
+    MULTISTAGE_UNATTRIBUTED_REASON,
+    SEMANTICS_UNAVAILABLE_REASON,
+)
+
+REASON_CODES = frozenset(
+    {
+        "work-limit",
+        "depth-limit",
+        "text-limit",
+        "graph-ceiling",
+        "unsupported-node",
+        "runtime-value",
+        "composite",
+        "no-sampler",
+        "malformed",
+        "multistage-differs",
+    }
+)
+LIMIT_CODES = frozenset({"work-limit", "depth-limit", "text-limit", "graph-ceiling"})
+# Codes whose outcome names the blocking node: ``node: {id, class_type}``.
+NAMED_CODES = frozenset({"unsupported-node", "runtime-value", "composite"})
+# Bound for a node ID or class name on the wire and in a line.
+CLASS_NAME_LENGTH = 200
+# Bound for a role's line on the wire, in UTF-16 code units (the client's count).
+MESSAGE_LENGTH = 512
+UNREADABLE_METADATA_REASON = "Media metadata could not be read"
+_NODE_KEYS = frozenset({"id", "class_type"})
+
+_NAMED_LINES = {
+    "unsupported-node": "Prompt uses a node the inspector does not support",
+    "runtime-value": "Prompt is chosen at run time",
+    "composite": "Prompt combines several texts",
+}
+_CODED_LINES = {
+    **dict.fromkeys(LIMIT_CODES, "Prompt exceeds inspection limits."),
+    "no-sampler": "No supported sampler was found.",
+    "malformed": "Prompt metadata in this file is malformed.",
+    "multistage-differs": "Sampler stages have different prompts.",
+}
+# Uncoded outcomes, keyed by the resolver's exact reason.
+_UNCODED_LINES = {
+    EMPTY_POSITIVE_REASON: "Positive prompt is empty.",
+    MULTISTAGE_UNATTRIBUTED_REASON: (
+        "Several sampler stages were found; their prompts were not compared."
+    ),
+    SEMANTICS_UNAVAILABLE_REASON: "The inspector's node table is unavailable.",
+    UNREADABLE_METADATA_REASON: "Metadata in this file could not be read.",
+}
+
+
+def _text(value: Any) -> Optional[str]:
+    return value if isinstance(value, str) else None
+
+
+def _class_name(node: Any) -> Optional[str]:
+    if not isinstance(node, dict):
+        return None
+    name = node.get("class_type")
+    return name[:CLASS_NAME_LENGTH] if isinstance(name, str) and name else None
+
+
+def outcome_message(resolution: Any) -> Optional[str]:
+    """One line for an unresolved outcome, or ``None`` to keep the status copy."""
+    if not isinstance(resolution, dict) or resolution.get("state") != "unresolved":
+        return None
+    code = _text(resolution.get("reason_code"))
+    if code in _NAMED_LINES:
+        name = _class_name(resolution.get("node"))
+        return f"{_NAMED_LINES[code]} ({name})." if name else f"{_NAMED_LINES[code]}."
+    if code is None:
+        return _UNCODED_LINES.get(_text(resolution.get("reason")))
+    return _CODED_LINES.get(code)
+
+
+def _node(code: Optional[str], node: Any) -> Optional[dict]:
+    if code not in NAMED_CODES or not isinstance(node, dict) or set(node) != _NODE_KEYS:
+        return None
+    node_id, name = node["id"], node["class_type"]
+    if not isinstance(node_id, str) or not node_id:
+        return None
+    if name is not None and not isinstance(name, str):
+        return None
+    return {
+        "id": node_id[:CLASS_NAME_LENGTH],
+        "class_type": name[:CLASS_NAME_LENGTH] if name else None,
+    }
+
+
+def project_resolution(resolution: dict) -> dict:
+    """Fail closed: an unknown or contradictory code and its node are dropped."""
+    code = _text(resolution.get("reason_code"))
+    if code not in REASON_CODES or resolution.get("state") in ("resolved", "absent"):
+        code = None
+    return {
+        **resolution,
+        "reason_code": code,
+        "node": _node(code, resolution.get("node")),
+    }

utils/inspection_response.py

@@ -3,7 +3,9 @@
 import json
 import math
 from decimal import Decimal, InvalidOperation
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal, Optional
+
+from .inspection_outcomes import MESSAGE_LENGTH, project_resolution
 
 if TYPE_CHECKING:
     from .prompt_inspector import InspectionRead
@@ -312,6 +314,29 @@ def _library_projection(library: dict) -> dict:
     return result
 
 
+def _bounded_message(value: Any) -> Optional[str]:
+    """A role line as the client accepts it: 1 to 512 UTF-16 code units."""
+    if not isinstance(value, str) or not value:
+        return None
+    value = value[:MESSAGE_LENGTH]
+    while len(value.encode("utf-16-le", "surrogatepass")) > 2 * MESSAGE_LENGTH:
+        value = value[:-1]
+    return value
+
+
+def _bounded_messages(prompts: dict) -> dict:
+    """Keep a role ``message`` only as bounded text on a non-resolved role."""
+    bounded = {}
+    for role, side in prompts.items():
+        if isinstance(side, dict) and "message" in side:
+            side = {key: item for key, item in side.items() if key != "message"}
+            message = _bounded_message(prompts[role]["message"])
+            if message is not None and side.get("status") != "resolved":
+                side["message"] = message
+        bounded[role] = side
+    return bounded
+
+
 def _encode_inspection(
     *,
     identity: dict,
@@ -328,8 +353,8 @@ def _encode_inspection(
         "inspection_version": 1,
         **identity,
         "file": _file_projection(file, inspection),
-        "prompts": prompts,
-        "resolution": resolution,
+        "prompts": _bounded_messages(prompts),
+        "resolution": project_resolution(resolution),
         "metadata": metadata,
         "library": _library_projection(library),
     }

utils/inspection_semantics.py

@@ -0,0 +1,492 @@
+"""Load and validate the declarative inspection node semantics table (#240).
+
+The table lives once in ``web/js/inspection-semantics.json`` so a browser-local
+resolver can load the same data. It maps node classes (or ``(class, output
+slot)`` pairs) to a named strategy from the closed ``STRATEGIES`` set plus the
+exact input names that strategy reads. Code implements each strategy; adding a
+node that reuses one is a data-only change.
+
+Only the inspection resolver profile consumes this table, through
+``get_semantics()``, which reads and validates it on first use and caches only a
+valid result. Importing this module reads nothing, and the default (ingestion)
+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 category;
+classes that are not listed stay ``unsupported-node``.
+"""
+
+import json
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from types import MappingProxyType
+from typing import Any, Callable, Dict, FrozenSet, Mapping, Optional, Tuple, Union
+
+SEMANTICS_PATH = (
+    Path(__file__).resolve().parents[1] / "web" / "js" / "inspection-semantics.json"
+)
+SCHEMA_VERSION = 1
+# The real table is a few KiB and four levels deep; these only reject nonsense.
+MAX_TABLE_BYTES = 1024 * 1024
+MAX_NESTING = 8
+_NAME_LENGTH = 128
+# Comfy output slots are small; this only rejects nonsense, never a real node.
+_MAX_SLOT = 255
+
+STRATEGIES: Mapping[str, FrozenSet[str]] = MappingProxyType(
+    {
+        "samplers": frozenset({"ksampler", "custom-advanced", "detailer-stage"}),
+        "guiders": frozenset({"cfg", "basic"}),
+        "conditioning": frozenset({"passthrough", "zeroed"}),
+        "selectors": frozenset({"boolean-select"}),
+        "composites": frozenset({"combine"}),
+        "encoders": frozenset({"single", "parts-equal", "pm-compose", "flag-parts"}),
+        "strings": frozenset({"literal", "strip-newlines", "join-multi", "concat"}),
+        "savers": frozenset({"comfy-counter", "vhs-counter"}),
+    }
+)
+# Shapes of the facts each sampler class provides (consumed by comfy-metadata.js).
+FACT_SHAPES = frozenset({"ksampler", "custom", "custom-advanced"})
+SAVER_PATTERNS = MappingProxyType(
+    {
+        "comfy-counter": "{name}_{counter:05}_.{ext}",
+        "vhs-counter": "{name}_{counter:05}.{ext}",
+    }
+)
+_SAMPLER_OPTIONS = frozenset({"prompt_overrides", "fact_overrides"})
+_CONFIGURATION_KINDS = frozenset({"noise", "sampler", "sigmas", "guider"})
+_SCALAR_KINDS = frozenset({"number", "string"})
+_TYPE_FIELDS = frozenset({"type_fields"})
+_CLASS_CATEGORIES = ("samplers", "guiders", "selectors", "composites", "savers")
+_SLOT_CATEGORIES = ("conditioning", "encoders", "strings")
+_SET_CATEGORIES = ("runtime", "non_file_outputs")
+_FACTS_TABLES = (
+    "configuration_outputs",
+    "model_outputs",
+    "model_passthroughs",
+    "scalar_outputs",
+)
+_CATEGORIES = frozenset(
+    ("schema",)
+    + _CLASS_CATEGORIES
+    + _SLOT_CATEGORIES
+    + _SET_CATEGORIES
+    + ("stage_inputs", "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
+# without a ``positive`` link), and those that do so beside a linked ``model``
+# (``with_model``: XlabsSampler's ``conditioning``, video samplers' ``samples``).
+# ``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")
+_STRING_PARAMS = MappingProxyType(
+    {
+        "literal": frozenset(),
+        "concat": frozenset(),
+        "strip-newlines": frozenset({"flag"}),
+        "join-multi": frozenset({"count", "delimiter", "list_flag", "input_prefix"}),
+    }
+)
+
+Frozen = Union[Mapping[str, Any], Tuple[Any, ...], str, int, None]
+
+
+class SemanticsError(ValueError):
+    """The semantics table is unreadable or violates its schema."""
+
+
+@dataclass(frozen=True)
+class Semantics:
+    """Validated, deeply frozen semantics. Slot categories key by ``(class, slot)``."""
+
+    samplers: Mapping[str, Mapping[str, Any]]
+    guiders: Mapping[str, Mapping[str, Any]]
+    conditioning: Mapping[Tuple[str, int], Mapping[str, Any]]
+    selectors: Mapping[str, Mapping[str, Any]]
+    composites: Mapping[str, Mapping[str, Any]]
+    encoders: Mapping[Tuple[str, int], Mapping[str, Any]]
+    strings: Mapping[Tuple[str, int], Mapping[str, Any]]
+    runtime: FrozenSet[str]
+    savers: Mapping[str, Mapping[str, Any]]
+    non_file_outputs: FrozenSet[str]
+    stage_inputs: Mapping[str, Tuple[str, ...]]
+    facts: Mapping[str, Any]
+    all_classes: FrozenSet[str]
+
+
+def _freeze(value: Any) -> Frozen:
+    if isinstance(value, dict):
+        return MappingProxyType({key: _freeze(item) for key, item in value.items()})
+    if isinstance(value, list):
+        return tuple(_freeze(item) for item in value)
+    return value
+
+
+def _reject_duplicate_keys(pairs: list) -> dict:
+    result: Dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise SemanticsError(f"duplicate key {key!r}")
+        result[key] = value
+    return result
+
+
+def _reject_constant(name: str) -> None:
+    raise SemanticsError(f"non-finite number {name} is not allowed")
+
+
+def _name(value: Any, where: str, what: str = "input") -> str:
+    if not isinstance(value, str) or not value or len(value) > _NAME_LENGTH:
+        raise SemanticsError(f"{where}: missing or invalid {what} name {value!r}")
+    return value
+
+
+def _names(value: Any, where: str, minimum: int = 1) -> None:
+    if not isinstance(value, list) or len(value) < minimum:
+        raise SemanticsError(f"{where}: needs at least {minimum} input name(s)")
+    for item in value:
+        _name(item, where)
+    if len(set(value)) != len(value):
+        raise SemanticsError(f"{where}: repeated input name")
+
+
+def _fields(
+    entry: Any, where: str, required: FrozenSet[str], optional: FrozenSet[str]
+) -> None:
+    if not isinstance(entry, dict):
+        raise SemanticsError(f"{where}: entry must be an object")
+    missing = sorted(required - entry.keys())
+    if missing:
+        raise SemanticsError(f"{where}: missing input field(s) {missing}")
+    unknown = sorted(entry.keys() - required - optional)
+    if unknown:
+        raise SemanticsError(f"{where}: unknown field(s) {unknown}")
+
+
+def _strategy(entry: Any, category: str, where: str) -> str:
+    strategy = entry.get("strategy") if isinstance(entry, dict) else None
+    if not isinstance(strategy, str) or strategy not in STRATEGIES[category]:
+        raise SemanticsError(f"{where}: unknown strategy {strategy!r}")
+    return strategy
+
+
+def _check_sampler(entry: dict, strategy: str, where: str) -> None:
+    stage = "image_stage" if strategy == "detailer-stage" else "latent"
+    _fields(
+        entry,
+        where,
+        frozenset({"strategy", "conditioning", stage, "facts"}),
+        _SAMPLER_OPTIONS,
+    )
+    _name(entry[stage], where)
+    # Inputs that, when set, replace the prompt the sampler's conditioning
+    # inputs carry (a detailer's wildcard); the prompt is then not read.
+    # ``fact_overrides`` (read by comfy-metadata.js): inputs that, when set,
+    # replace the settings the facts would read (a script or an XY plot).
+    for option in sorted(_SAMPLER_OPTIONS & entry.keys()):
+        _names(entry[option], f"{where}.{option}")
+    keys = {"guider"} if strategy == "custom-advanced" else {"positive", "negative"}
+    conditioning = entry["conditioning"]
+    _fields(conditioning, where + ".conditioning", frozenset(keys), frozenset())
+    for key in keys:
+        _name(conditioning[key], where + ".conditioning")
+    if entry["facts"] not in FACT_SHAPES:
+        raise SemanticsError(f"{where}: unknown facts shape {entry['facts']!r}")
+
+
+def _check_guider(entry: dict, strategy: str, where: str) -> None:
+    if strategy == "cfg":
+        _fields(
+            entry, where, frozenset({"strategy", "positive", "negative"}), frozenset()
+        )
+        _name(entry["negative"], where)
+    else:
+        # Negative absent by design; its reason is fixed text, not an input.
+        _fields(
+            entry,
+            where,
+            frozenset({"strategy", "positive"}),
+            frozenset({"absent_reason"}),
+        )
+        if "absent_reason" in entry:
+            _name(entry["absent_reason"], where, "absent reason")
+    _name(entry["positive"], where)
+
+
+def _check_selector(entry: dict, strategy: str, where: str) -> None:
+    _fields(entry, where, frozenset({"strategy", "condition", "branches"}), frozenset())
+    _name(entry["condition"], where)
+    branches = entry["branches"]
+    _fields(branches, where + ".branches", frozenset({"true", "false"}), frozenset())
+    _name(branches["true"], where)
+    _name(branches["false"], where)
+
+
+def _check_composite(entry: dict, strategy: str, where: str) -> None:
+    _fields(entry, where, frozenset({"strategy", "inputs"}), frozenset())
+    _names(entry["inputs"], where, minimum=2)
+
+
+def _check_saver(entry: dict, strategy: str, where: str) -> None:
+    _fields(
+        entry,
+        where,
+        frozenset({"strategy", "prefix_input", "image_input", "pattern"}),
+        frozenset(),
+    )
+    _name(entry["prefix_input"], where)
+    _name(entry["image_input"], where)
+    if entry["pattern"] != SAVER_PATTERNS[strategy]:
+        raise SemanticsError(f"{where}: pattern does not match {strategy}")
+
+
+def _check_conditioning(entry: dict, strategy: str, where: str) -> None:
+    _fields(entry, where, frozenset({"strategy", "input"}), frozenset())
+    _name(entry["input"], where)
+
+
+def _check_encoder(entry: dict, strategy: str, where: str) -> None:
+    if strategy == "flag-parts":
+        # A literal boolean ``flag`` false reads the first input; true needs
+        # every input equal (smZ CLIPTextEncode's with_SDXL).
+        _fields(entry, where, frozenset({"strategy", "inputs", "params"}), frozenset())
+        params = entry["params"]
+        if not isinstance(params, dict) or set(params) != {"flag"}:
+            raise SemanticsError(f"{where}: params must be exactly ['flag']")
+        _name(params["flag"], where + ".params")
+        _names(entry["inputs"], where, minimum=2)
+        return
+    _fields(entry, where, frozenset({"strategy", "inputs"}), frozenset({"compose"}))
+    _names(entry["inputs"], where, minimum=2 if strategy == "parts-equal" else 1)
+    if "compose" in entry and not isinstance(entry["compose"], bool):
+        raise SemanticsError(f"{where}: compose must be a boolean")
+
+
+def _check_string(entry: dict, strategy: str, where: str) -> None:
+    required = _STRING_PARAMS[strategy]
+    if required:
+        _fields(entry, where, frozenset({"strategy", "inputs", "params"}), frozenset())
+        params = entry["params"]
+        if not isinstance(params, dict) or set(params) != required:
+            raise SemanticsError(f"{where}: params must be exactly {sorted(required)}")
+        for value in params.values():
+            _name(value, where + ".params")
+    else:
+        _fields(entry, where, frozenset({"strategy", "inputs"}), frozenset())
+    _names(entry["inputs"], where)
+
+
+_CHECKS: Mapping[str, Callable[[dict, str, str], None]] = MappingProxyType(
+    {
+        "samplers": _check_sampler,
+        "guiders": _check_guider,
+        "selectors": _check_selector,
+        "composites": _check_composite,
+        "savers": _check_saver,
+        "conditioning": _check_conditioning,
+        "encoders": _check_encoder,
+        "strings": _check_string,
+    }
+)
+
+
+class _Owners:
+    """Each class belongs to one category; each (class, slot) is listed once."""
+
+    def __init__(self) -> None:
+        self._category: Dict[str, str] = {}
+        self._outputs: set = set()
+
+    def claim(self, name: str, category: str, slot: Optional[int] = None) -> None:
+        owner = self._category.setdefault(name, category)
+        if owner != category:
+            raise SemanticsError(f"duplicate class {name!r} in {owner} and {category}")
+        key = (name, slot)
+        if slot is not None and key in self._outputs:
+            raise SemanticsError(f"{category}: duplicate output key {key!r}")
+        self._outputs.add(key)
+
+    def classes(self) -> FrozenSet[str]:
+        return frozenset(self._category)
+
+
+def _class_category(doc: dict, category: str, owners: _Owners) -> Mapping:
+    table = doc[category]
+    if not isinstance(table, dict):
+        raise SemanticsError(f"category {category} must be an object")
+    result = {}
+    for name, entry in table.items():
+        where = f"{category}[{name!r}]"
+        _name(name, where, "class")
+        strategy = _strategy(entry, category, where)
+        _CHECKS[category](entry, strategy, where)
+        owners.claim(name, category)
+        result[name] = _freeze(entry)
+    return MappingProxyType(result)
+
+
+def _slot(entry: dict, where: str) -> int:
+    slot = entry.get("slot")
+    # bool is an int subclass; a JSON true is not an output slot.
+    if type(slot) is not int or not 0 <= slot <= _MAX_SLOT:
+        raise SemanticsError(f"{where}: slot must be a non-negative int, not {slot!r}")
+    return slot
+
+
+def _slot_category(doc: dict, category: str, owners: _Owners) -> Mapping:
+    rows = doc[category]
+    if not isinstance(rows, list):
+        raise SemanticsError(f"category {category} must be a list")
+    result = {}
+    for index, row in enumerate(rows):
+        where = f"{category}[{index}]"
+        if not isinstance(row, dict):
+            raise SemanticsError(f"{where}: entry must be an object")
+        name = _name(row.get("class"), where, "class")
+        slot = _slot(row, where)
+        entry = {
+            key: value for key, value in row.items() if key not in ("class", "slot")
+        }
+        strategy = _strategy(entry, category, where)
+        _CHECKS[category](entry, strategy, where)
+        owners.claim(name, category, slot)
+        result[(name, slot)] = _freeze(entry)
+    return MappingProxyType(result)
+
+
+def _set_category(doc: dict, category: str, owners: _Owners) -> FrozenSet[str]:
+    rows = doc[category]
+    if not isinstance(rows, list):
+        raise SemanticsError(f"category {category} must be a list")
+    seen: set = set()
+    for name in rows:
+        _name(name, category, "class")
+        if name in seen:
+            raise SemanticsError(f"{category}: duplicate class {name!r}")
+        seen.add(name)
+        owners.claim(name, category)
+    return frozenset(seen)
+
+
+def _string_map(table: Any, where: str, allowed: Optional[FrozenSet[str]]) -> None:
+    if not isinstance(table, dict):
+        raise SemanticsError(f"{where} must be an object")
+    for name, value in table.items():
+        _name(name, where, "class")
+        _name(value, where)
+        if allowed is not None and value not in allowed:
+            raise SemanticsError(f"{where}[{name!r}]: unknown kind {value!r}")
+
+
+def _stage_inputs(doc: dict) -> Mapping[str, Tuple[str, ...]]:
+    table = doc["stage_inputs"]
+    if not isinstance(table, dict) or set(table) != set(_STAGE_INPUT_LISTS):
+        raise SemanticsError(
+            f"stage_inputs must have exactly {list(_STAGE_INPUT_LISTS)}"
+        )
+    for key in _STAGE_INPUT_LISTS:
+        _names(table[key], f"stage_inputs.{key}")
+        if "model" in table[key]:
+            raise SemanticsError(f"stage_inputs.{key}: model is a patch input")
+    if set(table["shaped"]) & set(table["with_model"]):
+        raise SemanticsError("stage_inputs: an input is listed in both lists")
+    return _freeze(table)
+
+
+def _facts(doc: dict) -> Mapping[str, Any]:
+    facts = doc["facts"]
+    if not isinstance(facts, dict) or set(facts) != set(_FACTS_TABLES):
+        raise SemanticsError(f"facts must have exactly {list(_FACTS_TABLES)}")
+    configuration = facts["configuration_outputs"]
+    if not isinstance(configuration, dict):
+        raise SemanticsError("facts.configuration_outputs must be an object")
+    for name, adapter in configuration.items():
+        where = f"facts.configuration_outputs[{name!r}]"
+        _name(name, where, "class")
+        _fields(adapter, where, frozenset({"kind", "fields"}), _TYPE_FIELDS)
+        if adapter["kind"] not in _CONFIGURATION_KINDS:
+            raise SemanticsError(f"{where}: unknown kind {adapter['kind']!r}")
+        _names(adapter["fields"], where)
+        if "type_fields" in adapter:
+            # A fact whose value is the provider's node type (D8: Flux2Scheduler).
+            _names(adapter["type_fields"], where)
+            if set(adapter["type_fields"]) & set(adapter["fields"]):
+                raise SemanticsError(f"{where}: type_fields repeat an input field")
+    _string_map(facts["model_outputs"], "facts.model_outputs", None)
+    _string_map(facts["scalar_outputs"], "facts.scalar_outputs", _SCALAR_KINDS)
+    passthroughs = facts["model_passthroughs"]
+    _names(passthroughs, "facts.model_passthroughs", minimum=0)
+    return _freeze(facts)
+
+
+def _check_nesting(doc: Any) -> None:
+    """Reject containers nested deeper than ``MAX_NESTING`` (iterative walk)."""
+    stack = [(doc, 1)]
+    while stack:
+        value, level = stack.pop()
+        if not isinstance(value, (dict, list)):
+            continue
+        if level > MAX_NESTING:
+            raise SemanticsError(f"semantics table nests deeper than {MAX_NESTING}")
+        items = value.values() if isinstance(value, dict) else value
+        stack.extend((item, level + 1) for item in items)
+
+
+def _validate(doc: Any) -> Semantics:
+    if not isinstance(doc, dict):
+        raise SemanticsError("semantics table must be an object")
+    if set(doc) != _CATEGORIES:
+        missing = sorted(_CATEGORIES - doc.keys())
+        unknown = sorted(doc.keys() - _CATEGORIES)
+        raise SemanticsError(f"category mismatch: missing {missing}, unknown {unknown}")
+    if type(doc["schema"]) is not int or doc["schema"] != SCHEMA_VERSION:
+        raise SemanticsError(f"unsupported schema {doc['schema']!r}")
+    owners = _Owners()
+    tables: Dict[str, Any] = {}
+    for category in _CLASS_CATEGORIES:
+        tables[category] = _class_category(doc, category, owners)
+    for category in _SLOT_CATEGORIES:
+        tables[category] = _slot_category(doc, category, owners)
+    for category in _SET_CATEGORIES:
+        tables[category] = _set_category(doc, category, owners)
+    return Semantics(
+        stage_inputs=_stage_inputs(doc),
+        facts=_facts(doc),
+        all_classes=owners.classes(),
+        **tables,
+    )
+
+
+def load_semantics(path: Optional[Union[str, Path]] = None) -> Semantics:
+    """Read and validate a semantics table; ``SemanticsError`` on any problem."""
+    source = Path(path) if path is not None else SEMANTICS_PATH
+    try:
+        with source.open("rb") as stream:
+            data = stream.read(MAX_TABLE_BYTES + 1)
+        if len(data) > MAX_TABLE_BYTES:
+            raise SemanticsError(
+                f"semantics table {source.name} exceeds {MAX_TABLE_BYTES} bytes"
+            )
+        doc = json.loads(
+            data.decode("utf-8"),
+            object_pairs_hook=_reject_duplicate_keys,
+            parse_constant=_reject_constant,
+        )
+    except SemanticsError:
+        raise
+    except RecursionError:
+        raise SemanticsError(
+            f"semantics table {source.name} nests deeper than {MAX_NESTING}"
+        ) from None
+    except (OSError, UnicodeDecodeError, ValueError) as error:
+        raise SemanticsError(f"cannot read semantics table {source.name}: {error}")
+    _check_nesting(doc)
+    return _validate(doc)
+
+
+@lru_cache(maxsize=1)
+def get_semantics() -> Semantics:
+    """The repository table, loaded on first use; failures are not cached."""
+    return load_semantics()

utils/prompt_inspector.py

@@ -3,6 +3,7 @@
 import hashlib
 import io
 import json
+import re
 import struct
 import subprocess  # nosec B404
 import threading
@@ -13,8 +14,11 @@ from typing import Any
 
 from PIL import Image
 
+from .inspection_outcomes import UNREADABLE_METADATA_REASON, outcome_message
 from .pair_execution import EXECUTION_KEY
 from .prompt_pair_resolver import (
+    _FOOTER,
+    EMPTY_POSITIVE_REASON,
     INSPECTION_PROFILE,
     ResolverProfile,
     resolve_prompt_pair,
@@ -39,6 +43,23 @@ MAX_PNG_CHUNKS = 10_000
 _ROLE_METADATA_KEYS = frozenset({"prompt", "parameters", EXECUTION_KEY})
 _METADATA_KEYS = _ROLE_METADATA_KEYS | {"workflow"}
 _TEXT_CHUNKS = (b"tEXt", b"zTXt", b"iTXt")
+# An accumulating batch saver writes one A1111 ``parameters`` record per batch
+# item, identical except this field (R3); Pillow keeps the last one.
+# Matched only inside the record's one ``Steps: N, ...`` settings footer line.
+_BATCH_INDEX = re.compile(r"(?:^|(?<=, ))Batch index: \d+(?=, |$)")
+
+
+def _batch_normalized(text: str) -> str | None:
+    """``text`` with its footer's single batch index blanked, else ``None``."""
+    footers = list(_FOOTER.finditer(text))
+    if len(footers) != 1:
+        return None
+    line, count = _BATCH_INDEX.subn("Batch index: #", footers[0].group())
+    if count != 1:
+        return None
+    return text[: footers[0].start()] + line + text[footers[0].end() :]
+
+
 _SIGNATURE = b"\x89PNG\r\n\x1a\n"
 
 
@@ -121,13 +142,14 @@ class _PngRecords:
         self.digests: dict[str, tuple[int, bytes]] = {}
         self.role_encoded = self.role_decoded = 0
         self.duplicate_workflow = self.duplicate_role = 0
+        self.duplicate_decoded = 0
 
     def skip(self, key: str, length: int) -> bool:
         """Decide from the keyword and length alone whether to seek past a chunk."""
         if key not in _METADATA_KEYS or key in self.limited:
             return True  # never retained, or a duplicate of a limited workflow
         if key in self.digests:
-            if self.digests[key][0] != length:
+            if self.digests[key][0] != length and key != "parameters":
                 raise MetadataReadError("Duplicate prompt metadata")
             if key == "workflow":
                 if self.duplicate_workflow + length > DUPLICATE_WORKFLOW_BYTES:
@@ -155,7 +177,11 @@ class _PngRecords:
         digest = hashlib.sha256(kind + data).digest()
         if key in self.digests:
             if self.digests[key][1] != digest:
-                raise MetadataReadError("Duplicate prompt metadata")
+                if key != "parameters":
+                    raise MetadataReadError("Duplicate prompt metadata")
+                self.batch_copy(kind, data)
+                # The kept copy is the last one; later repeats compare with it.
+                self.digests[key] = (len(data), digest)
             return
         self.digests[key] = (len(data), digest)
         workflow = key == "workflow"
@@ -176,6 +202,27 @@ class _PngRecords:
                 raise MetadataReadError("Metadata exceeds inspection limits")
         self.found[key] = value
 
+    def batch_copy(self, kind: bytes, data: bytes) -> None:
+        """Keep a later ``parameters`` copy that differs only in its batch index.
+
+        Each copy must carry exactly one ``Batch index: N`` settings field; any
+        other difference is still a conflict. Decoding every copy draws from the
+        duplicate role budget, so small compressed copies stay bounded.
+        """
+        _, value = _png_text(kind, data, ROLE_METADATA_BYTES)
+        size = ROLE_METADATA_BYTES + 1 if value is None else len(value.encode())
+        self.duplicate_decoded += size
+        if value is None or self.duplicate_decoded > DUPLICATE_ROLE_BYTES:
+            raise MetadataReadError("Metadata exceeds inspection limits")
+        kept = self.found["parameters"]
+        normalized = _batch_normalized(kept)
+        if normalized is None or normalized != _batch_normalized(value):
+            raise MetadataReadError("Duplicate prompt metadata")
+        self.role_decoded += max(0, size - len(kept.encode()))
+        if self.role_decoded > ROLE_METADATA_BYTES:
+            raise MetadataReadError("Metadata exceeds inspection limits")
+        self.found["parameters"] = value
+
 
 def _png_metadata(stream: Any, size: int) -> tuple[dict, frozenset, int, int]:
     """CRC-check retained text and structural chunks, seeking past everything else.
@@ -400,6 +447,70 @@ def _side(status: str, text: str | None = None, source: str | None = None) -> di
     return {"status": status, "text": text, "source": source}
 
 
+_EMPTY_POSITIVE = outcome_message(
+    {"state": "unresolved", "reason": EMPTY_POSITIVE_REASON}
+)
+
+
+def _explained(prompts: dict, resolution: dict) -> dict:
+    """Each unresolved role carries the outcome's line; other roles are kept."""
+    message = outcome_message(resolution)
+    if message is None:
+        return prompts
+    return {
+        role: (
+            {**side, "message": message}
+            if side["status"] == "unresolved" and "message" not in side
+            else side
+        )
+        for role, side in prompts.items()
+    }
+
+
+def unreadable_evidence() -> dict:
+    """Both roles when the file's metadata could not be read at all."""
+    resolution = {
+        "state": "unresolved",
+        "sampler_id": None,
+        "reason": UNREADABLE_METADATA_REASON,
+        "reason_code": None,
+        "node": None,
+    }
+    prompts = {role: _side("unresolved") for role in ("positive", "negative")}
+    return {"prompts": _explained(prompts, resolution), "resolution": resolution}
+
+
+def _may_hold_prompt_manager(metadata: dict) -> bool:
+    """Whether the API graph can show the R1 signature, so the workflow is needed.
+
+    PromptManager classes, or a ``CLIPTextEncode`` carrying PromptManager's own
+    ``search_text`` input (an older save rewrote the class).
+    """
+    graph = metadata.get("prompt")
+    if not isinstance(graph, dict):
+        return False
+    for node in graph.values():
+        if not isinstance(node, dict):
+            continue
+        name = node.get("class_type")
+        inputs = node.get("inputs")
+        if isinstance(name, str) and "promptmanager" in name.lower():
+            return True
+        if isinstance(inputs, dict) and "search_text" in inputs:
+            return True
+    return False
+
+
+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,
@@ -413,6 +524,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
     }
@@ -447,21 +559,26 @@ def inspect_prompt_evidence(
             },
         }
     canonical = dict(metadata)
+    malformed = {
+        "prompts": _explained(
+            {role: _side("unresolved") for role in ("positive", "negative")},
+            resolution,
+        ),
+        "resolution": resolution,
+    }
     if "prompt" not in canonical and "parameters" not in canonical:
-        return {
-            "prompts": {role: _side("unresolved") for role in ("positive", "negative")},
-            "resolution": resolution,
-        }
+        return malformed
     if isinstance(canonical.get("prompt"), str):
         try:
             canonical["prompt"] = json.loads(canonical["prompt"])
         except (ValueError, RecursionError):
-            return {
-                "prompts": {
-                    role: _side("unresolved") for role in ("positive", "negative")
-                },
-                "resolution": resolution,
-            }
+            return malformed
+    workflow = (
+        _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None
+    )
+    if workflow is not None and "prompt" in canonical:
+        # Only to recognize overwritten PromptManager text (R1); never a role.
+        canonical["workflow"] = workflow
     resolved = resolve_prompt_pair(canonical, profile=profile)
     state = resolved["state"]
     result = {}
@@ -475,18 +592,19 @@ def inspect_prompt_evidence(
             result[role] = _side(
                 "resolved", side["text"], side["provenance"]["text_source"]
             )
+        elif state == "resolved" and isinstance(side, dict):
+            # A resolved pair withholds only an empty positive.
+            result[role] = {**_side("unresolved"), "message": _EMPTY_POSITIVE}
         else:
             result[role] = _side(state if state != "resolved" else "unresolved")
-    return {
-        "prompts": result,
-        "resolution": {
-            "state": state,
-            "sampler_id": resolved.get("sampler_id"),
-            "reason": resolved.get("reason", ""),
-            "reason_code": resolved.get("reason_code"),
-            "node": resolved.get("node"),
-        },
+    resolution = {
+        "state": state,
+        "sampler_id": resolved.get("sampler_id"),
+        "reason": resolved.get("reason", ""),
+        "reason_code": resolved.get("reason_code"),
+        "node": resolved.get("node"),
     }
+    return {"prompts": _explained(result, resolution), "resolution": resolution}
 
 
 def inspect_prompt_roles(metadata: dict, saved: dict | None = None) -> dict:

web/js/comfy-metadata.js

@@ -403,8 +403,53 @@
     }
 
     const FACT_FIELDS = ["seed", "steps", "cfg", "sampler", "scheduler", "model"];
-    const GENERATION_SAMPLERS = new Set(["KSampler", "KSamplerAdvanced", "SamplerCustom", "SamplerCustomAdvanced"]);
     const WORKFLOW_PASSTHROUGH = Symbol("inspection workflow passthrough");
+    // These tables must equal inspection-semantics.json: SAMPLERS its `samplers` section
+    // (facts shape, and a detailer stage's image input), the other four its `facts`
+    // section (tests/js/inspection-semantics-sync.test.js). They stay inline literals
+    // because the legacy pages load this file as a classic script.
+    const SAMPLERS = Object.freeze({
+        KSampler: { facts: "ksampler" },
+        KSamplerAdvanced: { facts: "ksampler" },
+        SamplerCustom: { facts: "custom" },
+        SamplerCustomAdvanced: { facts: "custom-advanced" },
+        FaceDetailer: { facts: "ksampler", image_stage: "image" },
+        DetailerForEach: { facts: "ksampler", image_stage: "image" },
+        UltimateSDUpscale: { facts: "ksampler", image_stage: "image" },
+        UltimateSDUpscaleNoUpscale: { facts: "ksampler", image_stage: "upscaled_image" },
+        // `fact_overrides`: inputs that, when set, replace the settings read here.
+        ClownsharKSampler_Beta: { facts: "ksampler", fact_overrides: ["options", "sigmas"] },
+        "KSampler (Efficient)": { facts: "ksampler", fact_overrides: ["script"] },
+        "ttN KSampler_v2": { facts: "ksampler", fact_overrides: ["adv_xyPlot"] },
+    });
+    // Stages merged for one result: the resolver's candidate cap. Together with
+    // MAX_VISITS per candidate walk these are deliberately tighter than the server's
+    // budgets (2,000 visits, depth 64), so the JS walk always fails closed first.
+    const MAX_STAGES = 32;
+    // Every class the semantics table lists (the resolver's `all_classes`): a stage walk
+    // passes these, and treats any other node shaped like a sampler as an unknown stage.
+    const TABLE_CLASSES = new Set([
+        "KSampler", "KSamplerAdvanced", "SamplerCustom", "SamplerCustomAdvanced", "FaceDetailer",
+        "CFGGuider", "BasicGuider", "easy ifElse", "ConditioningCombine", "ConditioningConcat",
+        "ConditioningAverage", "SaveImage", "SaveAnimatedWEBP", "SaveAnimatedPNG", "SaveVideo",
+        "SaveWEBM", "VHS_VideoCombine", "FluxGuidance", "ReferenceLatent", "ConditioningSetArea",
+        "ConditioningSetAreaPercentage", "ConditioningSetAreaStrength", "ConditioningSetMask", "ConditioningSetTimestepRange", "ControlNetApply",
+        "ControlNetApplyAdvanced", "ControlNetApplySD3", "InpaintModelConditioning", "WanImageToVideo", "ConditioningZeroOut",
+        "CLIPTextEncode", "CLIPTextEncodeSDXL", "CLIPTextEncodeFlux", "CLIPTextEncodeSD3", "TextEncodeQwenImageEdit",
+        "TextEncodeQwenImageEditPlus", "TextEncodeQwenImage21", "KikoTextEncode", "StringConstant", "StringConstantMultiline",
+        "JoinStringMulti", "PromptSearchList", "PreviewImage", "PreviewAny", "SaveImageWebsocket",
+        "Image Comparer (rgthree)", "DetailerForEach", "UltimateSDUpscale", "UltimateSDUpscaleNoUpscale",
+        "ControlNetInpaintingAliMamaApply", "LTXVConditioning", "FluxKontextMultiReferenceLatentMethod", "Text to Conditioning",
+        "TextInput", "DisplayText", "ClownsharKSampler_Beta", "KSampler (Efficient)", "ttN KSampler_v2",
+        "SeedVarianceEnhancer", "ApplyInstantIDAdvanced", "Switch conditioning [Crystools]", "smZ CLIPTextEncode", "ttN text",
+        "CR Text", "CR Prompt Text", "TextBoxMira",
+    ]);
+    // The table's `stage_inputs` (the resolver reads the same lists): inputs that make an
+    // unlisted node a possible stage, and those that do so beside a linked `model`.
+    const STAGE_INPUTS = Object.freeze({
+        shaped: ["latent_image", "guider", "positive", "negative", "pipe", "basic_pipe", "detailer_pipe", "sampler", "sigmas", "noise", "text_embeds", "hyvid_embeds"],
+        with_model: ["conditioning", "neg_conditioning", "samples", "latent"],
+    });
     // Fixed output0 contracts from the pinned core node schemas, not input-name inference.
     const SCALAR_OUTPUTS = Object.freeze({
         PrimitiveInt: "number", PrimitiveFloat: "number",
@@ -414,11 +459,17 @@
         CheckpointLoader: "ckpt_name", CheckpointLoaderSimple: "ckpt_name",
         unCLIPCheckpointLoader: "ckpt_name", UNETLoader: "unet_name",
     });
-    const MODEL_PASSTHROUGHS = new Set(["LoraLoader", "LoraLoaderModelOnly"]);
+    const MODEL_PASSTHROUGHS = new Set([
+        "LoraLoader", "LoraLoaderModelOnly", "ModelSamplingFlux", "ModelSamplingAuraFlow", "ModelSamplingSD3",
+        "Lora Loader (LoraManager)", "Power Lora Loader (rgthree)", "ModelPatchTorchSettings", "CFGNorm", "PerturbedAttentionGuidance",
+        "CR Apply LoRA Stack",
+    ]);
     const CONFIGURATION_OUTPUTS = Object.freeze({
         RandomNoise: { kind: "noise", fields: ["noise_seed"] },
         KSamplerSelect: { kind: "sampler", fields: ["sampler_name"] },
         BasicScheduler: { kind: "sigmas", fields: ["steps", "scheduler"] },
+        // D8: no scheduler input; the node type is the scheduler configuration.
+        Flux2Scheduler: { kind: "sigmas", fields: ["steps"], type_fields: ["scheduler"] },
         CFGGuider: { kind: "guider", fields: ["cfg", "model"] },
         BasicGuider: { kind: "guider", fields: ["model"] },
     });
@@ -463,24 +514,85 @@
         const seen = new Set();
         const [, node, adapter] = configurationOutput(ctx, link, kind, seen);
         if (!node) return fact("unresolved", source, samplerId);
+        if (keys.some(name => adapter.type_fields?.includes(name))) {
+            const value = exactFactValue(node.class_type, field);
+            return fact(value === null ? "unresolved" : "known", source, samplerId, value);
+        }
         const key = keys.find(name => adapter.fields.includes(name) && Object.hasOwn(node.inputs, name));
         return key === undefined ? fact("not-recorded", source, samplerId) : scalarFact(ctx, node.inputs, key, field, source, samplerId, seen);
     }
 
-    function graphFacts(graph, source, resolution) {
-        if (!object(graph)) return factSet("unresolved", source);
-        // Graph size is not a bound: each lookup is bounded by its MAX_VISITS seen-set,
-        // and the server's node ceiling arrives as a `limited` projection.
-        const entries = Object.entries(graph);
-        if (entries.some(([, node]) => !object(node) || !object(node.inputs) || typeof node.class_type !== "string")) return factSet("unresolved", source);
-        const candidates = entries.filter(([, node]) => GENERATION_SAMPLERS.has(node.class_type));
-        const selected = candidates.find(([id]) => typeof resolution?.sampler_id === "string" && id === resolution.sampler_id);
-        if (!selected && candidates.length !== 1) return factSet(candidates.length > 1 ? "ambiguous" : "unresolved", source);
-        const [samplerId, sampler] = selected || candidates[0];
-        const ctx = createContext(graph, null, true);
+    const samplerEntry = (graph, id) => (Object.hasOwn(graph, id) && Object.hasOwn(SAMPLERS, graph[id].class_type) ? SAMPLERS[graph[id].class_type] : undefined);
+
+    // The resolver's `_link`: an integer or string id and a non-negative integer slot.
+    // Stage walks use it so they cross exactly the server's edges; the legacy `isLink` stays.
+    const stageLink = value => Array.isArray(value) && value.length === 2 &&
+        (typeof value[0] === "string" || Number.isInteger(value[0])) && Number.isInteger(value[1]) && value[1] >= 0;
+    const linked = (inputs, key) => Object.hasOwn(inputs, key) && stageLink(inputs[key]);
+    /** An unlisted node that may itself be a sampler stage: the walk cannot judge it. */
+    const stageShaped = ({ class_type, inputs }) => !TABLE_CLASSES.has(class_type) &&
+        (STAGE_INPUTS.shaped.some(key => linked(inputs, key)) || (linked(inputs, "model") && STAGE_INPUTS.with_model.some(key => linked(inputs, key))));
+
+    /** Samplers reached from a detailer's image input, breadth-first, stopping at listed samplers. */
+    function upstreamStages(graph, origin, value, budget) {
+        if (!stageLink(value)) return null;
+        const found = [], queue = [String(value[0])], queued = new Set(queue);
+        for (let head = 0; head < queue.length; head++) {
+            const id = queue[head];
+            if (id === origin || ++budget.visits > MAX_VISITS || !Object.hasOwn(graph, id)) return null;
+            if (samplerEntry(graph, id)) { found.push(id); continue; }
+            if (stageShaped(graph[id])) return null;
+            for (const input of Object.values(graph[id].inputs)) {
+                const next = stageLink(input) ? String(input[0]) : null;
+                if (next !== null && !queued.has(next)) { queued.add(next); queue.push(next); }
+            }
+        }
+        return found;
+    }
+
+    /**
+     * Collect a final stage and every stage in its image ancestry into `stages`, as the
+     * resolver judges a detailer chain. `later` holds the detailers already being judged
+     * below this one, so a cyclic chain fails instead of recursing. False when the chain
+     * cannot be walked within the bounds.
+     */
+    // `stages` is shared on purpose across one result's candidates: re-adding a stage is
+    // harmless, and MAX_STAGES caps the union, not each candidate.
+    function collectStages(graph, id, later, stages, budget) {
+        stages.add(id);
+        if (stages.size > MAX_STAGES || later.size >= MAX_STAGES) return false;
+        const entry = samplerEntry(graph, id);
+        if (entry.image_stage === undefined) return true;
+        const inputs = graph[id].inputs;
+        const upstream = upstreamStages(graph, id, Object.hasOwn(inputs, entry.image_stage) ? inputs[entry.image_stage] : undefined, budget);
+        if (upstream === null || upstream.some(stage => later.has(stage))) return false;
+        const below = new Set(later).add(id);
+        return upstream.every(stage => collectStages(graph, stage, below, stages, budget));
+    }
+
+    /** Spec section 10 merge: equal known values stay known, differing ones are ambiguous. */
+    function mergeFacts(sets, source) {
+        if (sets.length === 1) return sets[0];
+        return Object.fromEntries(FACT_FIELDS.map(field => {
+            const facts = sets.map(set => set[field]);
+            if (new Set(facts.filter(item => item.status === "known").map(item => item.value)).size > 1) return [field, fact("ambiguous", source)];
+            const [first] = facts;
+            const agreed = facts.every(item => item.status === first.status && item.value === first.value);
+            // A merged fact names no single sampler: it is every stage's value.
+            return [field, agreed ? fact(first.status, source, null, first.value, first.reason) : fact("unresolved", source)];
+        }));
+    }
+
+    function samplerFacts(ctx, samplerId, source) {
+        const sampler = ctx.graph[samplerId];
         const inputs = sampler.inputs;
-        const advanced = sampler.class_type === "SamplerCustomAdvanced";
-        const custom = advanced || sampler.class_type === "SamplerCustom";
+        const { facts: shape, fact_overrides: overrides = [] } = SAMPLERS[sampler.class_type];
+        // A set override (a script, an XY plot, custom sigmas) replaces the settings below.
+        if (overrides.some(name => Object.hasOwn(inputs, name) && inputs[name] !== null && inputs[name] !== "")) {
+            return Object.fromEntries(FACT_FIELDS.map(name => [name, fact("unresolved", source, samplerId)]));
+        }
+        const advanced = shape === "custom-advanced";
+        const custom = advanced || shape === "custom";
         let settings = inputs;
         let settingsAdapter;
         const settingsPath = new Set();
@@ -510,6 +622,45 @@
         return facts;
     }
 
+    /**
+     * Facts follow the server's resolution. Its `sampler_id` drives them alone when
+     * resolved. Any other result naming a sampler (an ambiguous multistage chain, or an
+     * absent or unresolved prompt of the branch the server chose, e.g. after dropping a
+     * preview-only stage) merges per field across that sampler's stages, independent of
+     * any selection (D5). Ambiguous without a named sampler (no candidates on the wire
+     * until PR3), and unresolved or absent over several listed samplers, merge the same
+     * way across every listed sampler and its stages. Otherwise (no resolution, or one
+     * listed sampler) exactly one listed sampler supplies them; several leave every field ambiguous.
+     */
+    function graphFacts(graph, source, resolution) {
+        if (!object(graph)) return factSet("unresolved", source);
+        // Graph size is not a bound: each lookup is bounded by its MAX_VISITS seen-set,
+        // and the server's node ceiling arrives as a `limited` projection.
+        const entries = Object.entries(graph);
+        if (entries.some(([, node]) => !object(node) || !object(node.inputs) || typeof node.class_type !== "string")) return factSet("unresolved", source);
+        const ctx = createContext(graph, null, true);
+        const state = object(resolution) ? resolution.state : undefined;
+        const named = typeof resolution?.sampler_id === "string" ? resolution.sampler_id : null;
+        if (["resolved", "ambiguous", "absent", "unresolved"].includes(state) && named !== null) {
+            if (!samplerEntry(graph, named)) return factSet("unresolved", source);
+            if (state === "resolved") return samplerFacts(ctx, named, source);
+            const stages = new Set();
+            if (!collectStages(graph, named, new Set(), stages, { visits: 0 })) return factSet("unresolved", source);
+            return mergeFacts([...stages].map(id => samplerFacts(ctx, id, source)), source);
+        }
+        const candidates = entries.filter(([id]) => samplerEntry(graph, id));
+        // Review I-1: an unresolved or absent prompt does not make the run's settings
+        // unknown, so several listed samplers merge the same way.
+        if (state === "ambiguous" || ((state === "unresolved" || state === "absent") && candidates.length > 1)) {
+            // The server's branches: every listed sampler, a detailer judged with its stages.
+            const stages = new Set();
+            if (!candidates.length || !candidates.every(([id]) => collectStages(graph, id, new Set(), stages, { visits: 0 }))) return factSet("unresolved", source);
+            return mergeFacts([...stages].map(id => samplerFacts(ctx, id, source)), source);
+        }
+        if (candidates.length !== 1) return factSet(candidates.length > 1 ? "ambiguous" : "unresolved", source);
+        return samplerFacts(ctx, candidates[0][0], source);
+    }
+
     function validFactWorkflow(workflow) {
         if (!object(workflow) || !Array.isArray(workflow.nodes) || (workflow.links !== undefined && !Array.isArray(workflow.links))) return false;
         const ids = new Set(), links = new Set(), setters = new Set();
@@ -531,18 +682,51 @@
         return true;
     }
 
+    // The inspection resolver's `_footer` (Python): R5 also reads a bare `Steps: N`
+    // footer, and sd-dynamic-prompts' one-line template records after it.
+    const FOOTER_TRAILERS = ["Template:", "Negative Template:"];
+    function trailerLines(text) {
+        const lines = text.trim().split(/\r?\n|\r/).filter((line, index, all) => all.length > 1 || line);
+        const order = lines.map(line => FOOTER_TRAILERS.findIndex(name => line.startsWith(name)));
+        return !order.includes(-1) && order.every((value, index) => index === 0 || value > order[index - 1]);
+    }
+    function settingsFooter(parameters) {
+        let matches = [...parameters.matchAll(/^Steps:\s*\d+\s*,[^\n]*$/gm)];
+        if (!matches.length) matches = [...parameters.matchAll(/^Steps:\s*\d+\s*$/gm)];
+        if (matches.length !== 1) return null;
+        return trailerLines(parameters.slice(matches[0].index + matches[0][0].length)) ? matches[0] : null;
+    }
+    /** R10: split settings on commas outside A1111's quoted values; unbalanced quotes split as before. */
+    function settingsParts(line) {
+        const parts = [];
+        let start = 0, quoted = false;
+        for (let index = 0; index < line.length; index++) {
+            if (quoted && line[index] === "\\") index++;
+            else if (line[index] === '"') quoted = !quoted;
+            else if (line[index] === "," && !quoted) { parts.push(line.slice(start, index)); start = index + 1; }
+        }
+        if (quoted) return line.split(",");
+        parts.push(line.slice(start));
+        return parts;
+    }
+    function settingValue(value) {
+        if (!/^"(?:\\.|[^\\"])*"$/.test(value)) return value;
+        try { return JSON.parse(value); } catch { return value; }
+    }
+
     function parameterFacts(parameters) {
         const source = "embedded-parameters";
-        const matches = [...parameters.matchAll(/^Steps:\s*\d+\s*,[^\n]*$/gm)];
-        if (matches.length !== 1 || parameters.slice(matches[0].index + matches[0][0].length).trim()) return factSet("unresolved", source);
-        const body = parameters.slice(0, matches[0].index).replace(/[\r\n]+$/, "");
+        const footer = settingsFooter(parameters);
+        if (!footer) return factSet("unresolved", source);
+        // R4: like the resolver, text starting with the delimiter has an empty positive.
+        const body = "\n" + parameters.slice(0, footer.index).replace(/[\r\n]+$/, "");
         if (!body.split("\nNegative prompt:")[0].trim() || body.split("\nNegative prompt:").length > 2) return factSet("unresolved", source);
         const settings = new Map();
-        for (const part of matches[0][0].split(",")) {
+        for (const part of settingsParts(footer[0])) {
             const match = part.match(/^\s*([^:]+):\s*(.*?)\s*$/);
             if (!match) return factSet("unresolved", source);
             const [, key, value] = match;
-            settings.set(key, settings.has(key) ? null : value);
+            settings.set(key, settings.has(key) ? null : settingValue(value));
         }
         const keys = { seed: "Seed", steps: "Steps", cfg: "CFG scale", sampler: "Sampler", scheduler: "Schedule type", model: "Model" };
         return Object.fromEntries(FACT_FIELDS.map(field => {
@@ -575,6 +759,18 @@
     }
 
     const api = { extractGenerationParams, extractGenerationFacts, PLACEHOLDERS };
+    // For the semantics sync test only; not part of the legacy page contract.
+    Object.defineProperty(api, "_factsTables", {
+        value: Object.freeze({
+            configuration_outputs: CONFIGURATION_OUTPUTS,
+            model_outputs: MODEL_OUTPUTS,
+            model_passthroughs: MODEL_PASSTHROUGHS,
+            scalar_outputs: SCALAR_OUTPUTS,
+        }),
+    });
+    Object.defineProperty(api, "_samplerTable", { value: SAMPLERS });
+    Object.defineProperty(api, "_tableClasses", { value: TABLE_CLASSES });
+    Object.defineProperty(api, "_stageInputs", { value: STAGE_INPUTS });
     if (typeof module !== "undefined" && module.exports) module.exports = api;
     else root.ComfyMetadata = api;
 })(typeof window !== "undefined" ? window : globalThis);

web/js/inspection-semantics.json

@@ -0,0 +1,270 @@
+{
+  "schema": 1,
+  "samplers": {
+    "KSampler": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent_image",
+      "facts": "ksampler"
+    },
+    "KSamplerAdvanced": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent_image",
+      "facts": "ksampler"
+    },
+    "SamplerCustom": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent_image",
+      "facts": "custom"
+    },
+    "SamplerCustomAdvanced": {
+      "strategy": "custom-advanced",
+      "conditioning": { "guider": "guider" },
+      "latent": "latent_image",
+      "facts": "custom-advanced"
+    },
+    "FaceDetailer": {
+      "strategy": "detailer-stage",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "image_stage": "image",
+      "facts": "ksampler",
+      "prompt_overrides": ["wildcard"]
+    },
+    "DetailerForEach": {
+      "strategy": "detailer-stage",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "image_stage": "image",
+      "facts": "ksampler",
+      "prompt_overrides": ["wildcard"]
+    },
+    "UltimateSDUpscale": {
+      "strategy": "detailer-stage",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "image_stage": "image",
+      "facts": "ksampler"
+    },
+    "UltimateSDUpscaleNoUpscale": {
+      "strategy": "detailer-stage",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "image_stage": "upscaled_image",
+      "facts": "ksampler"
+    },
+    "ClownsharKSampler_Beta": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent_image",
+      "facts": "ksampler",
+      "prompt_overrides": ["options"],
+      "fact_overrides": ["options", "sigmas"]
+    },
+    "KSampler (Efficient)": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent_image",
+      "facts": "ksampler",
+      "prompt_overrides": ["script"],
+      "fact_overrides": ["script"]
+    },
+    "ttN KSampler_v2": {
+      "strategy": "ksampler",
+      "conditioning": { "positive": "positive", "negative": "negative" },
+      "latent": "latent",
+      "facts": "ksampler",
+      "prompt_overrides": ["adv_xyPlot"],
+      "fact_overrides": ["adv_xyPlot"]
+    }
+  },
+  "guiders": {
+    "CFGGuider": { "strategy": "cfg", "positive": "positive", "negative": "negative" },
+    "BasicGuider": {
+      "strategy": "basic",
+      "positive": "conditioning",
+      "absent_reason": "Guider has no negative input"
+    }
+  },
+  "conditioning": [
+    { "class": "FluxGuidance", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ReferenceLatent", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ConditioningSetArea", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ConditioningSetAreaPercentage", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ConditioningSetAreaStrength", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ConditioningSetMask", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ConditioningSetTimestepRange", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ControlNetApply", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ControlNetApplyAdvanced", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "ControlNetApplyAdvanced", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "ControlNetApplySD3", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "ControlNetApplySD3", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "InpaintModelConditioning", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "InpaintModelConditioning", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "WanImageToVideo", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "WanImageToVideo", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "ControlNetInpaintingAliMamaApply", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "ControlNetInpaintingAliMamaApply", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "LTXVConditioning", "slot": 0, "strategy": "passthrough", "input": "positive" },
+    { "class": "LTXVConditioning", "slot": 1, "strategy": "passthrough", "input": "negative" },
+    { "class": "FluxKontextMultiReferenceLatentMethod", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "SeedVarianceEnhancer", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
+    { "class": "ApplyInstantIDAdvanced", "slot": 1, "strategy": "passthrough", "input": "positive" },
+    { "class": "ApplyInstantIDAdvanced", "slot": 2, "strategy": "passthrough", "input": "negative" },
+    { "class": "ConditioningZeroOut", "slot": 0, "strategy": "zeroed", "input": "conditioning" }
+  ],
+  "selectors": {
+    "easy ifElse": {
+      "strategy": "boolean-select",
+      "condition": "boolean",
+      "branches": { "true": "on_true", "false": "on_false" }
+    },
+    "Switch conditioning [Crystools]": {
+      "strategy": "boolean-select",
+      "condition": "boolean",
+      "branches": { "true": "on_true", "false": "on_false" }
+    }
+  },
+  "composites": {
+    "ConditioningCombine": { "strategy": "combine", "inputs": ["conditioning_1", "conditioning_2"] },
+    "ConditioningConcat": { "strategy": "combine", "inputs": ["conditioning_to", "conditioning_from"] },
+    "ConditioningAverage": { "strategy": "combine", "inputs": ["conditioning_to", "conditioning_from"] }
+  },
+  "encoders": [
+    { "class": "CLIPTextEncode", "slot": 0, "strategy": "single", "inputs": ["text"] },
+    { "class": "CLIPTextEncodeSDXL", "slot": 0, "strategy": "parts-equal", "inputs": ["text_g", "text_l"] },
+    { "class": "CLIPTextEncodeFlux", "slot": 0, "strategy": "parts-equal", "inputs": ["clip_l", "t5xxl"] },
+    { "class": "CLIPTextEncodeSD3", "slot": 0, "strategy": "parts-equal", "inputs": ["clip_l", "clip_g", "t5xxl"] },
+    { "class": "TextEncodeQwenImageEdit", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
+    { "class": "TextEncodeQwenImageEditPlus", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
+    { "class": "TextEncodeQwenImage21", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
+    { "class": "TextEncodeQwenImage21", "slot": 1, "strategy": "single", "inputs": ["negative_prompt"] },
+    { "class": "KikoTextEncode", "slot": 0, "strategy": "single", "inputs": ["text"] },
+    { "class": "Text to Conditioning", "slot": 0, "strategy": "single", "inputs": ["text"] },
+    {
+      "class": "smZ CLIPTextEncode",
+      "slot": 0,
+      "strategy": "flag-parts",
+      "inputs": ["text", "text_g", "text_l"],
+      "params": { "flag": "with_SDXL" }
+    }
+  ],
+  "strings": [
+    { "class": "StringConstant", "slot": 0, "strategy": "literal", "inputs": ["string"] },
+    { "class": "TextInput", "slot": 0, "strategy": "literal", "inputs": ["text"] },
+    { "class": "DisplayText", "slot": 0, "strategy": "literal", "inputs": ["text"] },
+    { "class": "ttN text", "slot": 0, "strategy": "literal", "inputs": ["text"] },
+    { "class": "CR Text", "slot": 0, "strategy": "literal", "inputs": ["text"] },
+    { "class": "CR Prompt Text", "slot": 0, "strategy": "literal", "inputs": ["prompt"] },
+    { "class": "TextBoxMira", "slot": 0, "strategy": "literal", "inputs": ["text"] },
+    {
+      "class": "StringConstantMultiline",
+      "slot": 0,
+      "strategy": "strip-newlines",
+      "inputs": ["string"],
+      "params": { "flag": "strip_newlines" }
+    },
+    {
+      "class": "JoinStringMulti",
+      "slot": 0,
+      "strategy": "join-multi",
+      "inputs": ["string_1"],
+      "params": {
+        "count": "inputcount",
+        "delimiter": "delimiter",
+        "list_flag": "return_list",
+        "input_prefix": "string_"
+      }
+    }
+  ],
+  "runtime": ["PromptSearchList"],
+  "savers": {
+    "SaveImage": {
+      "strategy": "comfy-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "images",
+      "pattern": "{name}_{counter:05}_.{ext}"
+    },
+    "SaveAnimatedWEBP": {
+      "strategy": "comfy-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "images",
+      "pattern": "{name}_{counter:05}_.{ext}"
+    },
+    "SaveAnimatedPNG": {
+      "strategy": "comfy-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "images",
+      "pattern": "{name}_{counter:05}_.{ext}"
+    },
+    "SaveVideo": {
+      "strategy": "comfy-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "video",
+      "pattern": "{name}_{counter:05}_.{ext}"
+    },
+    "SaveWEBM": {
+      "strategy": "comfy-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "images",
+      "pattern": "{name}_{counter:05}_.{ext}"
+    },
+    "VHS_VideoCombine": {
+      "strategy": "vhs-counter",
+      "prefix_input": "filename_prefix",
+      "image_input": "images",
+      "pattern": "{name}_{counter:05}.{ext}"
+    }
+  },
+  "non_file_outputs": ["PreviewImage", "PreviewAny", "SaveImageWebsocket", "Image Comparer (rgthree)"],
+  "stage_inputs": {
+    "shaped": [
+      "latent_image",
+      "guider",
+      "positive",
+      "negative",
+      "pipe",
+      "basic_pipe",
+      "detailer_pipe",
+      "sampler",
+      "sigmas",
+      "noise",
+      "text_embeds",
+      "hyvid_embeds"
+    ],
+    "with_model": ["conditioning", "neg_conditioning", "samples", "latent"]
+  },
+  "facts": {
+    "configuration_outputs": {
+      "RandomNoise": { "kind": "noise", "fields": ["noise_seed"] },
+      "KSamplerSelect": { "kind": "sampler", "fields": ["sampler_name"] },
+      "BasicScheduler": { "kind": "sigmas", "fields": ["steps", "scheduler"] },
+      "Flux2Scheduler": { "kind": "sigmas", "fields": ["steps"], "type_fields": ["scheduler"] },
+      "CFGGuider": { "kind": "guider", "fields": ["cfg", "model"] },
+      "BasicGuider": { "kind": "guider", "fields": ["model"] }
+    },
+    "model_outputs": {
+      "CheckpointLoader": "ckpt_name",
+      "CheckpointLoaderSimple": "ckpt_name",
+      "unCLIPCheckpointLoader": "ckpt_name",
+      "UNETLoader": "unet_name"
+    },
+    "model_passthroughs": [
+      "LoraLoader",
+      "LoraLoaderModelOnly",
+      "ModelSamplingFlux",
+      "ModelSamplingAuraFlow",
+      "ModelSamplingSD3",
+      "Lora Loader (LoraManager)",
+      "Power Lora Loader (rgthree)",
+      "ModelPatchTorchSettings",
+      "CFGNorm",
+      "PerturbedAttentionGuidance",
+      "CR Apply LoRA Stack"
+    ],
+    "scalar_outputs": {
+      "PrimitiveInt": "number",
+      "PrimitiveFloat": "number",
+      "PrimitiveString": "string",
+      "PrimitiveStringMultiline": "string"
+    }
+  }
+}

web/js/workspace/file-inspector.mjs

@@ -85,7 +85,7 @@ function formatBytes(value) {
     return value === null || value === undefined ? null : value >= 1048576 ? `${(value / 1048576).toFixed(1)} MB` : `${Math.max(1, Math.round(value / 1024))} KB`;
 }
 
-export function mountFileInspector(container, {sender, parentScope, inline = false, onPreview, onRetry, onRefresh, onInteraction = () => {}}) {
+export function mountFileInspector(container, {sender, parentScope, inline = false, promptContext = '', onPreview, onRetry, onRefresh, onInteraction = () => {}}) {
     if (parentScope.signal.aborted) return {setState() {}, dispose() {}};
     const scope = childScope(parentScope);
     let contentScope;
@@ -145,7 +145,7 @@ export function mountFileInspector(container, {sender, parentScope, inline = fal
         }
         if (chips.childNodes.length) inner.append(chips);
         const prompts = el('div'); inner.append(prompts);
-        const blocks = mountPromptBlocks(prompts, {sender, parentScope:contentScope, includePairAction:false});
+        const blocks = mountPromptBlocks(prompts, {sender, parentScope:contentScope, includePairAction:false, contextLabel:promptContext});
         blocks.setState({prompts:record.prompts}); contentScope.cleanup(blocks.dispose);
         if (library?.status === 'linked') {
             const vocabulary = el('div', undefined, 'file-vocabulary');
@@ -226,7 +226,7 @@ export function mountFileInspector(container, {sender, parentScope, inline = fal
         inner.append(raw);
         const footer = el('footer', undefined, 'file-inspector-footer'), roles = promptRoles(record.prompts);
         if (Object.hasOwn(roles, 'positive') && Object.hasOwn(roles, 'negative')) {
-            const pair = createSendButton('Send prompt pair'); pair.dataset.sendRole = 'pair';
+            const pair = createSendButton(promptContext ? `Send ${promptContext.toLowerCase()} prompt pair` : 'Send prompt pair'); pair.dataset.sendRole = 'pair';
             contentScope.listen(pair, 'click', () => sender.open(roles, pair)); footer.append(pair);
         }
         if (record.historyTarget) {

web/js/workspace/inspection-state.mjs

@@ -9,6 +9,10 @@ const FACT_FIELDS = ['seed', 'steps', 'cfg', 'sampler', 'scheduler', 'model'];
 const RECORD_FORMATS = ['original-text', 'reserialized-json', 'invalid', 'limited'];
 const TEXTLESS_FORMATS = ['invalid', 'limited'];
 const REASON_CODES = ['work-limit', 'depth-limit', 'text-limit', 'graph-ceiling', 'unsupported-node', 'runtime-value', 'composite', 'no-sampler', 'malformed', 'multistage-differs'];
+// Only these codes name the blocking node; node IDs and class names are bounded.
+const NAMED_CODES = ['unsupported-node', 'runtime-value', 'composite'];
+const NAME_LENGTH = 200, MESSAGE_LENGTH = 512;
+const boundedText = (value, limit) => typeof value === 'string' && value.length > 0 && value.length <= limit;
 const object = value => value !== null && typeof value === 'object' && !Array.isArray(value);
 const validScope = value => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value) && value.length === 64;
 
@@ -77,7 +81,8 @@ function promptsFor(value) {
         if (record === undefined) continue;
         if (!object(record) || typeof record.status !== 'string' ||
             (record.text !== undefined && record.text !== null && typeof record.text !== 'string') ||
-            (record.status === 'resolved' && !Object.hasOwn(confirmed, role))) throw new TypeError('Invalid inspection role');
+            (record.status === 'resolved' && !Object.hasOwn(confirmed, role)) ||
+            (record.message !== undefined && (record.status === 'resolved' || !boundedText(record.message, MESSAGE_LENGTH)))) throw new TypeError('Invalid inspection role');
     }
     return value;
 }
@@ -141,7 +146,8 @@ function resolutionFor(value) {
     const code = value.reason_code ?? null, node = value.node ?? null;
     // Codes explain a non-resolved outcome; a blocking node is named only with its code.
     if ((code !== null && (!REASON_CODES.includes(code) || ['resolved', 'absent'].includes(value.state))) ||
-        (node !== null && (code === null || !object(node) || typeof node.id !== 'string' || (node.class_type !== null && typeof node.class_type !== 'string')))) throw new TypeError('Invalid inspection resolution reason');
+        (node !== null && (!NAMED_CODES.includes(code) || !object(node) || Object.keys(node).sort().join() !== 'class_type,id' ||
+            !boundedText(node.id, NAME_LENGTH) || (node.class_type !== null && !boundedText(node.class_type, NAME_LENGTH))))) throw new TypeError('Invalid inspection resolution reason');
     return value;
 }
 

web/js/workspace/prompt-blocks.mjs

@@ -11,6 +11,7 @@ const sources = {
     execution: 'Recorded at execution',
     'graph-inferred': 'Inferred from the saved workflow',
     'embedded-parameters': 'Embedded generation parameters',
+    'zeroed-conditioning': 'Zeroed conditioning (no negative text)',
     'saved-association': 'Saved library association',
 };
 const statuses = {
@@ -40,7 +41,11 @@ function childScope(parentScope) {
     return scope;
 }
 
-export function mountPromptBlocks(container, {prompts, sender, parentScope, roles:visibleRoles = ['positive', 'negative'], onRetry, includePairAction = true}) {
+export function mountPromptBlocks(container, {prompts, sender, parentScope, roles:visibleRoles = ['positive', 'negative'], onRetry, includePairAction = true, contextLabel = ''}) {
+    // A nested set of blocks (the current file inside a recorded run) names its regions and actions with a
+    // context so assistive technology never meets two identically named prompt regions in one dialog.
+    const named = role => contextLabel ? `${contextLabel} ${role} prompt` : `${role} prompt`;
+    const action = role => contextLabel ? `${contextLabel[0].toLowerCase()}${contextLabel.slice(1)} ${role} prompt` : `${role} prompt`;
     if (parentScope.signal.aborted) return {update() {}, setState() {}, dispose() {}};
     const scope = childScope(parentScope);
     container.classList.add('prompt-inspector');
@@ -86,7 +91,7 @@ export function mountPromptBlocks(container, {prompts, sender, parentScope, role
             const title = `${role[0].toUpperCase() + role.slice(1)} prompt`;
             const record = state.prompts?.[role];
             const block = el('section', undefined, `prompt-block prompt-${role}`);
-            block.setAttribute('aria-label', title);
+            block.setAttribute('aria-label', contextLabel ? named(role) : title);
             const heading = el('div', undefined, 'prompt-block-heading');
             heading.append(el('h3', title));
             block.append(heading);
@@ -100,12 +105,12 @@ export function mountPromptBlocks(container, {prompts, sender, parentScope, role
                     block.append(text);
                 }
                 const actions = el('div', undefined, 'prompt-actions');
-                const send = createSendButton(`Send ${role} prompt`);
+                const send = createSendButton(`Send ${action(role)}`);
                 send.dataset.sendRole = role;
                 send.disabled = !actionsEnabled;
                 const copy = createButton('', 'ca-icon-button');
-                copy.setAttribute('aria-label', `Copy ${role} prompt`);
-                copy.title = `Copy ${role} prompt`;
+                copy.setAttribute('aria-label', `Copy ${action(role)}`);
+                copy.title = `Copy ${action(role)}`;
                 copy.append(createWorkspaceIcon('copy'));
                 copy.dataset.copyRole = role;
                 copy.disabled = !actionsEnabled;
@@ -113,7 +118,7 @@ export function mountPromptBlocks(container, {prompts, sender, parentScope, role
                 heading.append(actions);
                 const status = el('p', '', 'prompt-copy-status muted'); status.setAttribute('role', 'status');
                 block.append(status);
-            } else block.append(el('p', typeof record?.message === 'string' ? record.message : statuses[record?.status] || statuses.unresolved, 'muted'));
+            } else block.append(el('p', typeof record?.message === 'string' && record.message ? record.message : statuses[record?.status] || statuses.unresolved, 'muted'));
             container.append(block);
         }
         if (includePairAction && Object.hasOwn(roles, 'positive') && Object.hasOwn(roles, 'negative')) {

web/js/workspace/recorded-run.mjs

@@ -68,7 +68,7 @@ export function mountRecordedRun(container, {api, controller, sender, parentScop
         const receiptNodes = new Map();
         const evidenceValues = new Map();
         const blocks = mountPromptBlocks(prompts, {sender, parentScope:local});
-        const inspector = mountFileInspector(inspectorContainer, {sender, parentScope:local, inline:true, onRetry:() => inspection.retry(), onPreview:() => showViewer(), onRefresh:refreshCurrent});
+        const inspector = mountFileInspector(inspectorContainer, {sender, parentScope:local, inline:true, promptContext:'Current file', onRetry:() => inspection.retry(), onPreview:() => showViewer(), onRefresh:refreshCurrent});
         function selectedReceipt() {
             const selected = latest.selection.receipt;
             return selected && latest.outputs.items.find(item => occurrenceKey(item.occurrence) === occurrenceKey(selected));