diff --git a/TESTING.md b/TESTING.md
index 76031db..936139c 100644
--- a/TESTING.md
+++ b/TESTING.md
@@ -332,7 +332,30 @@ becomes record format `limited` (no raw text, graph `limited`, roles and facts
 unaffected); other text chunks are skipped unread; byte-identical duplicates are
 accepted within a bounded read budget, while differing duplicates (and any duplicate
 role record over that budget) still fail. Other read bounds are unchanged: 8 MiB
-non-PNG header, 10,000 PNG chunks and a three-second ffprobe. Resolution under
+non-PNG header, 10,000 PNG chunks and a three-second ffprobe (per probe, at most two
+probes). Video containers are read through one shared carrier selector
+(`utils/video_metadata.select_video_carrier`, used by the inspector and by the admin
+scan): carriers are, in order, the per-key tags that current savers write (`prompt`,
+`workflow`, `parameters`, `comfyassets_prompt_pairs`, each `original-text`; Matroska
+upper-cased names fold on our side), then the old VideoHelperSuite `comment` JSON
+envelope, then `description` (object values are `reserialized-json`; envelope keys are
+matched exactly, as the legacy readers did, never folded). The first carrier holding a
+role key wins, otherwise the first holding a workflow, and carriers never merge. Two
+per-key tags folding to one key are a duplicate record: byte-identical ones are one
+record, differing ones fail the read. ffprobe is asked for exactly those six tags.
+Its JSON output escapes tag text (at most doubling it), so the stdout bounds are sized
+so that only the decoded per-record budgets above can fail or limit a record, and a
+video reads exactly as the identical PNG: the first probe allows twice the role and
+workflow budgets plus 64 KiB of envelope (12 MiB + 64 KiB); past that the inspector
+probes once more without the `workflow` tag within twice the role budget plus 64 KiB
+(4 MiB + 64 KiB) and the per-key workflow becomes `limited`, while an envelope over
+that bound still fails the read because it cannot be split. The admin scan reads the
+same carrier with the default resolver profile, so a per-key video links like the same
+graph in a PNG; the live monitor processes images only (unchanged, see the follow-up
+below) (`tests/test_utils_video_metadata.py`, `tests/test_api_admin.py`
+`TestScanReadsVideos`, `tests/test_api_prompt_inspector.py` `test_video_*`,
+`tests/test_inspection_response.py` `TestVideoProbeBoundsMatchPngBudgets`; the
+real-container round trips skip without ffmpeg). Resolution under
 `INSPECTION_PROFILE` bounds work, not graph size: 2,000 visits per candidate chain,
 20,000 in total, 64 link/JSON-container levels, 100,000 resolved characters, and a
 two 20,000-node ceilings: one over the top-level entries of the API prompt, and one
@@ -1656,6 +1679,12 @@ responsive production Viewer keeps its toolbar in the media panel and prompt det
 mobile toolbar collision. These are explicit remaining visual differences, not
 a whole-application or full-gallery/Viewer pixel-parity claim.
 
+Follow-up: monitor video support. The live image monitor (`utils/image_monitor.py`,
+`GalleryConfig.SUPPORTED_EXTENSIONS`) processes images only; videos written by
+ComfyUI savers reach the Library through the admin scan, which reads the shared video
+carrier. Accepting `VIDEO_EXTENSIONS` in the monitor and routing them through
+`read_video_metadata` is a separate, issue-linked change.
+
 Preserve raw logs, failed traces, paired PNGs, geometry, runtime versions and
 source hashes before later managed commands clear their output directories.
 Frontend baseline updates are additive through `coverage:update` only after
diff --git a/docs/prompt-inspector.md b/docs/prompt-inspector.md
index 539fd39..81366b7 100644
--- a/docs/prompt-inspector.md
+++ b/docs/prompt-inspector.md
@@ -29,6 +29,12 @@ Each role has `status` (`resolved`, `absent`, `unresolved`, or `ambiguous`), `te
 (string only when resolved, otherwise null), and `source`. Inspection does not
 index files, alter saved prompts, run scans, or decode image pixels. PNG text and
 video probe output are bounded; unsupported or malformed evidence remains
-unresolved rather than becoming sendable placeholder text.
+unresolved rather than becoming sendable placeholder text. Videos are read from
+one container carrier: the per-key tags written by core SaveVideo/SaveWEBM and
+current VideoHelperSuite (`prompt`, `workflow`, `parameters`,
+`comfyassets_prompt_pairs`), else the old VideoHelperSuite `comment` envelope,
+else `description`. The admin scan reads the same carrier, so Images and
+ingestion agree on a video's prompts (see TESTING.md for the order and budgets;
+the live monitor processes images only).
 
 Python route changes require restarting ComfyUI. Browser assets require a refresh.
diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py
index 363a9b1..bddcc39 100644
--- a/tests/test_api_admin.py
+++ b/tests/test_api_admin.py
@@ -2476,7 +2476,8 @@ def _write_mp4(path, comment):
 
 
 class TestScanReadsVideos(AdminAPITestCase):
-    """Videos saved by ComfyUI carry the graph in their comment tag."""
+    """Videos carry the graph in per-key container tags (current savers) or in
+    the old VideoHelperSuite comment envelope; the scan reads one carrier."""
 
     def setUp(self):
         super().setUp()
@@ -2506,6 +2507,56 @@ class TestScanReadsVideos(AdminAPITestCase):
         images = self.api.db.get_prompt_images(prompts[0]["id"])
         self.assertTrue(images[0]["image_path"].endswith("clip.mp4"))
 
+    async def test_scan_links_per_key_tag_video_with_default_profile(self):
+        """A current saver's per-key tags scan exactly like the same graph as PNG."""
+        from utils import video_metadata
+
+        _write_png(self.output_dir / "same.png", prompt_text="a cat")
+        graph = {
+            "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}},
+            "2": {"class_type": "CLIPTextEncode", "inputs": {"text": "bad"}},
+            "3": {
+                "class_type": "KSampler",
+                "inputs": {"positive": ["1", 0], "negative": ["2", 0]},
+            },
+        }
+        (self.output_dir / "clip.mkv").write_bytes(b"\x1aE\xdf\xa3 fixture")
+        probed = {
+            "format": {
+                "tags": {
+                    "PROMPT": json.dumps(graph),
+                    "WORKFLOW": '{"nodes":[]}',
+                    "COMMENT": json.dumps({"prompt": "must not win"}),
+                }
+            }
+        }
+        GalleryConfig.MONITORING_DIRECTORIES = [self._root(self.output_dir)]
+
+        with (
+            patch.object(video_metadata.shutil, "which", return_value="/bin/ffprobe"),
+            patch.object(
+                video_metadata.subprocess,
+                "run",
+                return_value=MagicMock(returncode=0, stdout=json.dumps(probed)),
+            ) as run,
+            self.assertNoLogs(level="ERROR"),
+        ):
+            body = await (
+                await self.client.request("POST", "/prompt_manager/scan")
+            ).text()
+
+        self.assertEqual(run.call_args.args[0][-1], str(self.output_dir / "clip.mkv"))
+        self.assertIn('"found": 2', body)
+        prompts = self.api.db.get_recent_prompts(limit=10)["prompts"]
+        self.assertEqual(sorted(p["text"] for p in prompts), ["a cat", "bad"])
+        positive = next(p for p in prompts if p["text"] == "a cat")
+        images = self.api.db.get_prompt_images(positive["id"])
+        self.assertEqual(
+            sorted(os.path.basename(image["image_path"]) for image in images),
+            ["clip.mkv", "same.png"],
+        )
+        self.assertNotIn("must not win", json.dumps(prompts))
+
     async def test_scan_without_ffprobe_skips_videos_quietly(self):
         from unittest.mock import patch
         from utils import video_metadata
diff --git a/tests/test_api_prompt_inspector.py b/tests/test_api_prompt_inspector.py
index c4702a9..fcd9261 100644
--- a/tests/test_api_prompt_inspector.py
+++ b/tests/test_api_prompt_inspector.py
@@ -1,7 +1,11 @@
 """Physical-file prompt inspection is bounded, conservative and read-only."""
 
+import io
 import json
-from unittest.mock import patch
+import shutil
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
 
 from PIL import Image, PngImagePlugin
 
@@ -15,6 +19,9 @@ except ImportError:
 from utils.pair_execution import EXECUTION_KEY, record_pair_execution
 from database.pair_ingestion import ingest_resolved_pair
 from utils.prompt_pair_resolver import resolve_prompt_pair
+from utils import video_metadata
+
+HAVE_FFMPEG = bool(shutil.which("ffmpeg") and shutil.which("ffprobe"))
 
 
 def graph(positive="portrait", negative="blurry"):
@@ -28,6 +35,54 @@ def graph(positive="portrait", negative="blurry"):
     }
 
 
+def write_video(path, metadata_tags=True, **tags):
+    """Tiny black clip carrying one container tag per keyword (needs ffmpeg).
+
+    MP4 keeps arbitrary keys only with ``use_metadata_tags`` (mdta atoms), the
+    movflag the core SaveVideo node sets; without it ffmpeg writes the legacy
+    ilst atoms that old VideoHelperSuite files carry. Matroska and WebM
+    upper-case the key names.
+    """
+    command = [
+        "ffmpeg",
+        "-v",
+        "quiet",
+        "-f",
+        "lavfi",
+        "-i",
+        "color=c=black:s=16x16:d=0.2",
+    ]
+    if metadata_tags and path.suffix.lower() in (".mp4", ".mov", ".m4v"):
+        command += ["-movflags", "use_metadata_tags"]
+    for key, value in tags.items():
+        command += ["-metadata", f"{key}={value}"]
+    subprocess.run(command + ["-y", str(path)], check=True, timeout=60)
+    return path
+
+
+class ProbeOutput(io.BytesIO):
+    """ffprobe stdout that records the bounded read size asked of it."""
+
+    requested_size = None
+
+    def read(self, size=-1):
+        self.requested_size = size
+        return super().read(size)
+
+
+def probe_process(payload, poll=0, returncode=0):
+    """A fake ffprobe whose stdout is ``payload`` bytes."""
+    process = MagicMock()
+    process.stdout = ProbeOutput(payload)
+    process.poll.return_value = poll
+    process.returncode = returncode
+    return process
+
+
+def probe_payload(tags):
+    return json.dumps({"format": {"tags": tags}}).encode()
+
+
 def png(path, **metadata):
     info = PngImagePlugin.PngInfo()
     for key, value in metadata.items():
@@ -312,7 +367,7 @@ class TestPromptInspector(ImageAPITestCase):
     async def test_corrupt_truncated_duplicate_and_expanding_metadata_fail_closed(self):
         import struct
         import zlib
-        from utils.prompt_inspector import MAX_METADATA_BYTES
+        from utils.prompt_inspector import VIDEO_REPROBE_BYTES
 
         path = self.output_dir / "image.png"
         base = png(path, prompt=graph()).read_bytes()
@@ -325,8 +380,8 @@ class TestPromptInspector(ImageAPITestCase):
         cases.append(bytes(broken))
         for kind, payload in (
             (b"tEXt", b"prompt\0{}"),
-            (b"zTXt", b"prompt\0\0" + zlib.compress(b"x" * (MAX_METADATA_BYTES + 1))),
-            (b"tEXt", b"prompt\0" + b"x" * (MAX_METADATA_BYTES + 1)),
+            (b"zTXt", b"prompt\0\0" + zlib.compress(b"x" * (VIDEO_REPROBE_BYTES + 1))),
+            (b"tEXt", b"prompt\0" + b"x" * (VIDEO_REPROBE_BYTES + 1)),
             (b"tEXt", b"bad-key-without-separator"),
             (b"zTXt", b"prompt\0\1bad-compression"),
             (b"iTXt", b"prompt\0\2\0\0\0text"),
@@ -385,9 +440,7 @@ class TestPromptInspector(ImageAPITestCase):
         self.assertEqual(result["prompts"]["positive"]["status"], "absent")
 
     async def test_video_inspection_uses_bounded_container_evidence(self):
-        import io
-        from unittest.mock import MagicMock
-        from utils.prompt_inspector import MAX_METADATA_BYTES
+        from utils.prompt_inspector import VIDEO_PROBE_BYTES, VIDEO_REPROBE_BYTES
 
         (self.output_dir / "clip.mp4").write_bytes(b"video fixture")
         metadata = {"prompt": graph("Video portrait", "")}
@@ -412,14 +465,25 @@ class TestPromptInspector(ImageAPITestCase):
         process.wait.assert_called()
         self.assertNotIn("shell", spawn.call_args.kwargs)
         self.assertIn(str(self.output_dir / "clip.mp4"), spawn.call_args.args[0])
-        process.stdout = io.BytesIO(b"x" * (MAX_METADATA_BYTES + 1))
+        # An envelope over the probe bound cannot be split: the role-only
+        # re-probe overflows too and the read fails instead of guessing.
+        processes = [
+            probe_process(b"x" * (VIDEO_PROBE_BYTES + 1)),
+            probe_process(b"x" * (VIDEO_REPROBE_BYTES + 1)),
+        ]
         with (
             patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
-            patch("utils.prompt_inspector.subprocess.Popen", return_value=process),
+            patch(
+                "utils.prompt_inspector.subprocess.Popen", side_effect=processes
+            ) as spawn,
         ):
             _, result = await self.inspect("clip.mp4")
+        self.assertEqual(spawn.call_count, 2)
         self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
-        self.assertTrue(process.stdout.closed)
+        self.assertEqual(result["metadata"]["status"], "unreadable")
+        for process in processes:
+            self.assertTrue(process.stdout.closed)
+            process.wait.assert_called()
 
     async def test_video_missing_probe_and_timeout_preserve_uncertainty(self):
         import io
@@ -895,11 +959,9 @@ class TestPromptInspector(ImageAPITestCase):
         self.assertNotIn("\ufffd".encode(), body)
         self.assertLessEqual(len(body), 8 * 1024 * 1024)
 
-    async def video_inspect(self, comment=None, description=None):
-        import io
+    async def video_inspect(self, comment=None, description=None, **per_key):
         import threading
-        from unittest.mock import MagicMock
-        from utils.prompt_inspector import MAX_METADATA_BYTES
+        from utils.prompt_inspector import VIDEO_PROBE_BYTES
 
         path = self.output_dir / "clip.mp4"
         path.write_bytes(b"fixture")
@@ -908,17 +970,8 @@ class TestPromptInspector(ImageAPITestCase):
             for key, value in (("comment", comment), ("description", description))
             if value is not None
         }
-        payload = json.dumps({"format": {"tags": tags}}).encode()
-
-        class ProbeOutput(io.BytesIO):
-            def read(self, size=-1):
-                self.requested_size = size
-                return super().read(size)
-
-        process = MagicMock()
-        process.stdout = ProbeOutput(payload)
-        process.poll.return_value = 0
-        process.returncode = 0
+        tags.update(per_key)
+        process = probe_process(probe_payload(tags))
         timers = []
         timer_type = threading.Timer
 
@@ -937,11 +990,156 @@ class TestPromptInspector(ImageAPITestCase):
             response, result = await self.inspect("clip.mp4")
         self.assertEqual(spawn.call_count, 1)
         self.assertTrue(process.stdout.closed)
-        self.assertEqual(process.stdout.requested_size, MAX_METADATA_BYTES + 1)
+        self.assertEqual(process.stdout.requested_size, VIDEO_PROBE_BYTES + 1)
         self.assertEqual(len(timers), 1)
         self.assertTrue(timers[0].finished.is_set())
+        self.assertEqual(
+            spawn.call_args.args[0][spawn.call_args.args[0].index("-show_entries") + 1],
+            "format_tags=prompt,workflow,parameters,"
+            "comfyassets_prompt_pairs,comment,description",
+        )
         return response, result
 
+    async def test_video_separate_prompt_workflow_tags_are_read(self):
+        source = graph("per-key portrait", "per-key negative")
+        workflow = '{"nodes":[{"id":"per-key"}],"cfg":6.5000000000000000001}'
+        for tags in (
+            {"prompt": json.dumps(source), "workflow": workflow},
+            {"PROMPT": json.dumps(source), "WORKFLOW": workflow},
+        ):
+            _, result = await self.video_inspect(
+                json.dumps({"prompt": graph("comment must not win")}), **tags
+            )
+            self.assertEqual(
+                result["prompts"]["positive"], self.side("per-key portrait")
+            )
+            self.assertEqual(
+                result["prompts"]["negative"], self.side("per-key negative")
+            )
+            self.assertEqual(
+                result["metadata"]["record_formats"],
+                {"prompt": "original-text", "workflow": "original-text"},
+            )
+            self.assertEqual(result["metadata"]["raw"]["workflow"], workflow)
+            self.assertEqual(
+                result["metadata"]["workflow_graph"]["value"]["cfg"],
+                "6.5000000000000000001",
+            )
+            self.assertNotIn("must not win", json.dumps(result))
+        _, result = await self.video_inspect(
+            **{"PROMPT": json.dumps(source), "prompt": json.dumps(graph("other"))}
+        )
+        self.assertEqual(result["prompts"]["positive"]["status"], "unresolved")
+        self.assertEqual(result["metadata"]["status"], "unreadable")
+
+    @unittest.skipUnless(HAVE_FFMPEG, "ffmpeg/ffprobe not installed")
+    async def test_video_per_key_and_comment_tags_read_from_real_containers(self):
+        video_metadata.ffprobe_path.cache_clear()
+        self.addCleanup(video_metadata.ffprobe_path.cache_clear)
+        source = json.dumps(graph("real portrait", ""))
+        workflow = '{"nodes":[{"id":"real"}]}'
+        for name in ("per_key.mp4", "per_key.mkv", "per_key.webm"):
+            write_video(self.output_dir / name, prompt=source, workflow=workflow)
+            response, result = await self.inspect(name)
+            self.assertEqual(response.status, 200, name)
+            self.assertEqual(
+                result["prompts"]["positive"], self.side("real portrait"), name
+            )
+            self.assertEqual(result["metadata"]["raw"]["workflow"], workflow, name)
+            self.assertEqual(
+                result["metadata"]["record_formats"],
+                {"prompt": "original-text", "workflow": "original-text"},
+                name,
+            )
+        # Old VideoHelperSuite envelope in the legacy ilst comment atom, unchanged.
+        write_video(
+            self.output_dir / "vhs.mp4",
+            metadata_tags=False,
+            comment=json.dumps({"prompt": json.loads(source), "workflow": {"v": 1}}),
+        )
+        _, result = await self.inspect("vhs.mp4")
+        self.assertEqual(result["prompts"]["positive"], self.side("real portrait"))
+        self.assertEqual(
+            result["metadata"]["record_formats"],
+            {"prompt": "reserialized-json", "workflow": "reserialized-json"},
+        )
+        self.assertEqual(result["metadata"]["workflow_graph"]["value"], {"v": 1})
+
+    async def test_video_oversize_workflow_reprobes_role_only_and_marks_limited(self):
+        from utils.prompt_inspector import VIDEO_PROBE_BYTES, VIDEO_REPROBE_BYTES
+
+        path = self.output_dir / "clip.mp4"
+        path.write_bytes(b"fixture")
+        source = json.dumps(graph("bounded portrait", ""))
+        processes = [
+            probe_process(b"x" * (VIDEO_PROBE_BYTES + 1)),
+            probe_process(probe_payload({"PROMPT": source})),
+        ]
+        with (
+            patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
+            patch(
+                "utils.prompt_inspector.subprocess.Popen", side_effect=processes
+            ) as spawn,
+        ):
+            response, result = await self.inspect("clip.mp4")
+        self.assertEqual(response.status, 200)
+        self.assertEqual(spawn.call_count, 2)
+        first, second = (call.args[0] for call in spawn.call_args_list)
+        self.assertEqual(
+            first[first.index("-show_entries") + 1],
+            "format_tags=prompt,workflow,parameters,"
+            "comfyassets_prompt_pairs,comment,description",
+        )
+        self.assertEqual(
+            second[second.index("-show_entries") + 1],
+            "format_tags=prompt,parameters,comfyassets_prompt_pairs,"
+            "comment,description",
+        )
+        self.assertEqual(processes[0].stdout.requested_size, VIDEO_PROBE_BYTES + 1)
+        self.assertEqual(processes[1].stdout.requested_size, VIDEO_REPROBE_BYTES + 1)
+        for process in processes:
+            self.assertTrue(process.stdout.closed)
+            process.wait.assert_called()
+        self.assertEqual(result["prompts"]["positive"], self.side("bounded portrait"))
+        self.assertEqual(result["metadata"]["status"], "available")
+        self.assertEqual(
+            result["metadata"]["record_formats"],
+            {"prompt": "original-text", "workflow": "limited"},
+        )
+        self.assertNotIn("workflow", result["metadata"]["raw"])
+        self.assertEqual(result["metadata"]["workflow_graph"]["status"], "limited")
+        # The dropped per-key workflow belonged to an unselected carrier when
+        # an envelope wins the re-probe: that envelope is whole, not limited.
+        envelope = json.dumps({"prompt": json.loads(source), "workflow": {"v": 2}})
+        processes = [
+            probe_process(b"x" * (VIDEO_PROBE_BYTES + 1)),
+            probe_process(probe_payload({"comment": envelope})),
+        ]
+        with (
+            patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
+            patch("utils.prompt_inspector.subprocess.Popen", side_effect=processes),
+        ):
+            _, result = await self.inspect("clip.mp4")
+        self.assertEqual(result["prompts"]["positive"], self.side("bounded portrait"))
+        self.assertEqual(result["metadata"]["workflow_graph"]["value"], {"v": 2})
+        self.assertEqual(
+            result["metadata"]["record_formats"]["workflow"], "reserialized-json"
+        )
+        # A workflow-only per-key video over budget is limited, not absent.
+        processes = [
+            probe_process(b"x" * (VIDEO_PROBE_BYTES + 1)),
+            probe_process(probe_payload({})),
+        ]
+        with (
+            patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
+            patch("utils.prompt_inspector.subprocess.Popen", side_effect=processes),
+        ):
+            _, result = await self.inspect("clip.mp4")
+        self.assertEqual(result["metadata"]["status"], "available")
+        self.assertEqual(result["metadata"]["record_formats"], {"workflow": "limited"})
+        self.assertEqual(result["metadata"]["workflow_graph"]["status"], "limited")
+        self.assertEqual(result["prompts"]["positive"]["status"], "absent")
+
     async def test_video_role_description_wins_over_workflow_only_comment(self):
         evidence = {}
         source = graph("chosen", "")
diff --git a/tests/test_inspection_response.py b/tests/test_inspection_response.py
index 37de619..936caa6 100644
--- a/tests/test_inspection_response.py
+++ b/tests/test_inspection_response.py
@@ -823,6 +823,88 @@ def _png_with_chunks(path, chunks):
     return path
 
 
+class TestVideoProbeBoundsMatchPngBudgets(unittest.TestCase):
+    """ffprobe escapes tag text; the stdout bounds must never gate a record
+    that the identical PNG reads within its decoded budget (I3)."""
+
+    @staticmethod
+    def _dense(size):
+        # Quote-dense JSON text: every other byte escapes, so ffprobe's JSON
+        # output is close to twice the decoded size.
+        unit = '{"k":"v",' + "".join('"' for _ in range(6)) + "}"
+        return (unit * (size // len(unit) + 1))[:size]
+
+    def _video_read(self, path, tags, bound):
+        payload = json.dumps({"format": {"tags": tags}}).encode()
+        self.assertGreater(len(payload), bound)
+        process = MagicMock()
+        process.stdout = io.BytesIO(payload)
+        process.poll.return_value = 0
+        process.returncode = 0
+        with (
+            patch("utils.prompt_inspector.ffprobe_path", return_value="/tool/ffprobe"),
+            patch(
+                "utils.prompt_inspector.subprocess.Popen", return_value=process
+            ) as spawn,
+        ):
+            read = read_inspection_file(path)
+        self.assertEqual(spawn.call_count, 1)
+        self.assertTrue(process.stdout.closed)
+        return read
+
+    def test_near_budget_records_read_the_same_as_png(self):
+        from utils.prompt_inspector import (
+            ROLE_METADATA_BYTES,
+            VIDEO_PROBE_BYTES,
+            VIDEO_REPROBE_BYTES,
+            WORKFLOW_METADATA_BYTES,
+        )
+
+        prompt = self._dense(ROLE_METADATA_BYTES - 1024)
+        workflow = self._dense(WORKFLOW_METADATA_BYTES - 1024)
+        with tempfile.TemporaryDirectory() as directory:
+            png = _png_with_chunks(
+                Path(directory) / "same.png",
+                [
+                    _text_chunk(b"tEXt", "prompt", prompt.encode("latin-1")),
+                    _text_chunk(b"tEXt", "workflow", workflow.encode("latin-1")),
+                ],
+            )
+            expected = read_inspection_file(png)
+            self.assertEqual(
+                expected.record_formats,
+                {"prompt": "original-text", "workflow": "original-text"},
+            )
+            video = Path(directory) / "same.mp4"
+            video.write_bytes(b"fixture")
+            # Escaped output is over the old 6 MiB + 64 KiB stdout bound.
+            read = self._video_read(
+                video,
+                {"prompt": prompt, "workflow": workflow},
+                ROLE_METADATA_BYTES + WORKFLOW_METADATA_BYTES + 64 * 1024,
+            )
+            self.assertEqual(read.metadata, expected.metadata)
+            self.assertEqual(read.record_formats, expected.record_formats)
+            self.assertEqual(read.limited_records, frozenset())
+            self.assertGreaterEqual(
+                VIDEO_PROBE_BYTES,
+                2 * (ROLE_METADATA_BYTES + WORKFLOW_METADATA_BYTES) + 64 * 1024,
+            )
+            self.assertGreaterEqual(
+                VIDEO_REPROBE_BYTES, 2 * ROLE_METADATA_BYTES + 64 * 1024
+            )
+            # A role record near its budget alone escapes past 2 MiB and must
+            # still read as the PNG does.
+            png = _png_with_chunks(
+                Path(directory) / "role.png",
+                [_text_chunk(b"tEXt", "prompt", prompt.encode("latin-1"))],
+            )
+            expected = read_inspection_file(png)
+            read = self._video_read(video, {"prompt": prompt}, ROLE_METADATA_BYTES)
+            self.assertEqual(read.metadata, expected.metadata)
+            self.assertEqual(read.record_formats, {"prompt": "original-text"})
+
+
 class TestPerRecordBudgets(unittest.TestCase):
     """Role records keep the 2 MiB budget; a workflow has its own 4 MiB budget."""
 
diff --git a/tests/test_utils_video_metadata.py b/tests/test_utils_video_metadata.py
index 1d56267..1cacefa 100644
--- a/tests/test_utils_video_metadata.py
+++ b/tests/test_utils_video_metadata.py
@@ -1,8 +1,11 @@
 """Tests for utils/video_metadata.py: prompt metadata from video containers.
 
-ComfyUI's video savers store ``{"prompt": ..., "workflow": ...}`` as JSON in
-the container's ``comment`` tag. ffprobe reads it; when ffprobe is missing
-videos are skipped without noise.
+Current ComfyUI savers (core SaveVideo/SaveWEBM, VideoHelperSuite) write one
+container tag per key (``prompt``, ``workflow``, ``parameters``,
+``comfyassets_prompt_pairs``); older VideoHelperSuite wrote one ``comment``
+JSON envelope, with ``description`` as a fallback. One selector picks a single
+carrier for the inspector and the scan. ffprobe reads the tags; when ffprobe
+is missing videos are skipped without noise.
 """
 
 import json
@@ -19,34 +22,44 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from utils import video_metadata  # noqa: E402
 from utils.video_metadata import (  # noqa: E402
     VIDEO_EXTENSIONS,
+    DuplicateVideoTagError,
     is_video_path,
     metadata_from_tags,
     read_video_metadata,
+    select_video_carrier,
 )
 
 HAVE_FFMPEG = bool(shutil.which("ffmpeg") and shutil.which("ffprobe"))
 GRAPH = json.dumps({"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}})
 
 
+def write_video(path, metadata_tags=True, **tags):
+    """Write a tiny black video carrying one container tag per keyword.
+
+    MP4 keeps arbitrary keys only with ``use_metadata_tags`` (mdta atoms), the
+    movflag the core SaveVideo node sets; without it ffmpeg writes the legacy
+    ilst atoms (``comment`` becomes ``©cmt``) that old VideoHelperSuite files
+    carry. Matroska and WebM upper-case the key names.
+    """
+    command = [
+        "ffmpeg",
+        "-v",
+        "quiet",
+        "-f",
+        "lavfi",
+        "-i",
+        "color=c=black:s=16x16:d=0.2",
+    ]
+    if metadata_tags and str(path).lower().endswith((".mp4", ".mov", ".m4v")):
+        command += ["-movflags", "use_metadata_tags"]
+    for key, value in tags.items():
+        command += ["-metadata", f"{key}={value}"]
+    subprocess.run(command + ["-y", str(path)], check=True, timeout=60)
+
+
 def write_mp4(path, comment):
-    """Write a tiny black video carrying ``comment`` in its container tags."""
-    subprocess.run(
-        [
-            "ffmpeg",
-            "-v",
-            "quiet",
-            "-f",
-            "lavfi",
-            "-i",
-            "color=c=black:s=16x16:d=0.2",
-            "-metadata",
-            f"comment={comment}",
-            "-y",
-            str(path),
-        ],
-        check=True,
-        timeout=60,
-    )
+    """Write a tiny black MP4 with ``comment`` in its legacy ilst ``©cmt`` atom."""
+    write_video(path, metadata_tags=False, comment=comment)
 
 
 class TestMetadataFromTags(unittest.TestCase):
@@ -69,6 +82,158 @@ class TestMetadataFromTags(unittest.TestCase):
         self.assertEqual(metadata_from_tags(tags), {"prompt": GRAPH})
 
 
+class TestSelectVideoCarrier(unittest.TestCase):
+    """Spec 2026-10-19 resolution parity section 8: one carrier, never merged."""
+
+    WORKFLOW = json.dumps({"nodes": [{"id": "per-key"}]})
+
+    def test_per_key_tags_win_over_comment(self):
+        tags = {
+            "prompt": GRAPH,
+            "workflow": self.WORKFLOW,
+            "comment": json.dumps({"prompt": json.dumps({"9": {}})}),
+        }
+        metadata, formats, carrier = select_video_carrier(tags)
+        self.assertEqual(metadata, {"prompt": GRAPH, "workflow": self.WORKFLOW})
+        self.assertEqual(
+            formats, {"prompt": "original-text", "workflow": "original-text"}
+        )
+        self.assertIsNone(carrier)
+
+    def test_matroska_uppercase_keys_fold(self):
+        tags = {"PROMPT": GRAPH, "Workflow": self.WORKFLOW, "ENCODER": "Lavf"}
+        metadata, formats, carrier = select_video_carrier(tags)
+        self.assertEqual(metadata, {"prompt": GRAPH, "workflow": self.WORKFLOW})
+        self.assertEqual(
+            formats, {"prompt": "original-text", "workflow": "original-text"}
+        )
+        self.assertIsNone(carrier)
+        # Envelope keys are read exactly as legacy did: no fold, so a Matroska
+        # upper-cased COMMENT is not an envelope.
+        self.assertIsNone(
+            select_video_carrier({"COMMENT": json.dumps({"prompt": GRAPH})})
+        )
+
+    def test_folded_duplicate_keys_fail(self):
+        for tags in (
+            {"PROMPT": GRAPH, "prompt": json.dumps({"9": {}})},
+            {"Workflow": self.WORKFLOW, "WORKFLOW": "{}", "prompt": GRAPH},
+        ):
+            with self.assertRaises(DuplicateVideoTagError):
+                select_video_carrier(tags)
+        # Byte-identical duplicates are one record, as for PNG text chunks.
+        metadata, _, _ = select_video_carrier({"PROMPT": GRAPH, "prompt": GRAPH})
+        self.assertEqual(metadata, {"prompt": GRAPH})
+        # Envelope keys never fold, so a foreign COMMENT note is not a duplicate.
+        envelope = json.dumps({"prompt": GRAPH})
+        metadata, _, carrier = select_video_carrier(
+            {"comment": envelope, "COMMENT": "a free-text note"}
+        )
+        self.assertEqual(metadata, {"prompt": GRAPH})
+        self.assertEqual(carrier, envelope)
+
+    def test_comment_envelope_still_read(self):
+        envelope = json.dumps({"prompt": GRAPH, "workflow": {"nodes": []}})
+        metadata, formats, carrier = select_video_carrier(
+            {"comment": envelope, "encoder": "Lavf"}
+        )
+        self.assertEqual(metadata["prompt"], GRAPH)
+        self.assertEqual(json.loads(metadata["workflow"]), {"nodes": []})
+        self.assertEqual(
+            formats, {"prompt": "original-text", "workflow": "reserialized-json"}
+        )
+        self.assertEqual(carrier, envelope)
+
+    def test_description_fallback(self):
+        envelope = json.dumps({"prompt": GRAPH})
+        tags = {"comment": "not json", "description": envelope}
+        self.assertEqual(
+            select_video_carrier(tags),
+            ({"prompt": GRAPH}, {"prompt": "original-text"}, envelope),
+        )
+        self.assertIsNone(select_video_carrier({"comment": "note", "encoder": "x"}))
+        self.assertIsNone(select_video_carrier({"comment": "[1]"}))
+        self.assertIsNone(select_video_carrier({}))
+        self.assertIsNone(select_video_carrier(None))
+        self.assertIsNone(select_video_carrier({"comment": 7}))
+
+    def test_role_carrier_beats_workflow_only_carrier(self):
+        envelope = json.dumps({"prompt": GRAPH, "workflow": {"nodes": ["c"]}})
+        metadata, formats, carrier = select_video_carrier(
+            {"workflow": self.WORKFLOW, "comment": envelope}
+        )
+        self.assertEqual(metadata["prompt"], GRAPH)
+        self.assertEqual(json.loads(metadata["workflow"]), {"nodes": ["c"]})
+        self.assertEqual(carrier, envelope)
+        for role in ("parameters", "comfyassets_prompt_pairs"):
+            metadata, _, carrier = select_video_carrier(
+                {"comment": json.dumps({"workflow": {}}), role: "x"}
+            )
+            self.assertEqual(metadata, {role: "x"})
+            self.assertIsNone(carrier)
+        # Without a role anywhere, the first workflow-only carrier wins.
+        metadata, _, carrier = select_video_carrier(
+            {"workflow": self.WORKFLOW, "comment": json.dumps({"workflow": {}})}
+        )
+        self.assertEqual(metadata, {"workflow": self.WORKFLOW})
+        self.assertIsNone(carrier)
+
+    def test_carriers_never_merge(self):
+        metadata, _, carrier = select_video_carrier(
+            {
+                "prompt": GRAPH,
+                "comment": json.dumps({"workflow": {"nodes": ["comment"]}}),
+                "description": json.dumps({"parameters": "Steps: 1"}),
+            }
+        )
+        self.assertEqual(metadata, {"prompt": GRAPH})
+        self.assertIsNone(carrier)
+        metadata, _, carrier = select_video_carrier(
+            {
+                "workflow": self.WORKFLOW,
+                "comment": json.dumps({"prompt": GRAPH}),
+                "description": json.dumps({"parameters": "Steps: 1"}),
+            }
+        )
+        self.assertEqual(metadata, {"prompt": GRAPH})
+        self.assertEqual(carrier, json.dumps({"prompt": GRAPH}))
+
+    def test_per_key_values_are_original_text_and_envelope_objects_reserialized(
+        self,
+    ):
+        exact = '{"3":{"inputs":{"cfg":6.5000000000000000001}}}'
+        metadata, formats, _ = select_video_carrier({"prompt": exact})
+        self.assertEqual(metadata, {"prompt": exact})
+        self.assertEqual(formats, {"prompt": "original-text"})
+        envelope = json.dumps(
+            {"prompt": {"3": {}}, "workflow": {"nodes": []}, "parameters": "Steps: 1"}
+        )
+        metadata, formats, carrier = select_video_carrier({"comment": envelope})
+        self.assertEqual(json.loads(metadata["prompt"]), {"3": {}})
+        self.assertEqual(json.loads(metadata["workflow"]), {"nodes": []})
+        self.assertEqual(metadata["parameters"], "Steps: 1")
+        self.assertEqual(
+            formats,
+            {
+                "prompt": "reserialized-json",
+                "workflow": "reserialized-json",
+                "parameters": "original-text",
+            },
+        )
+        self.assertEqual(carrier, envelope)
+
+    def test_null_role_markers_and_invalid_workflow_keep_their_shape(self):
+        # The inspector refuses a null role marker rather than falling back.
+        metadata, formats, _ = select_video_carrier(
+            {"comment": json.dumps({"prompt": None, "workflow": [1]})}
+        )
+        self.assertEqual(metadata, {"prompt": None})
+        self.assertEqual(formats, {"workflow": "invalid"})
+        metadata, formats, _ = select_video_carrier({"prompt": 7, "workflow": 1})
+        self.assertEqual(metadata, {"prompt": None})
+        self.assertEqual(formats, {"workflow": "invalid"})
+
+
 class TestIsVideoPath(unittest.TestCase):
 
     def test_extensions(self):
@@ -125,6 +290,37 @@ class TestReadVideoMetadataWithoutFfprobe(unittest.TestCase):
             ):
                 self.assertEqual(read_video_metadata("clip.mp4"), {})
 
+    def test_read_video_metadata_returns_per_key_prompt(self):
+        payload = {
+            "format": {
+                "tags": {
+                    "PROMPT": GRAPH,
+                    "WORKFLOW": '{"nodes":[]}',
+                    "COMFYASSETS_PROMPT_PAIRS": '{"version":1}',
+                    "COMMENT": json.dumps({"prompt": "must not win"}),
+                    "ENCODER": "Lavf",
+                }
+            }
+        }
+        with mock.patch.object(
+            video_metadata.shutil, "which", return_value="/bin/ffprobe"
+        ):
+            with self._run_returning(stdout=json.dumps(payload)):
+                self.assertEqual(
+                    read_video_metadata("clip.mkv"),
+                    {
+                        "prompt": GRAPH,
+                        "workflow": '{"nodes":[]}',
+                        "comfyassets_prompt_pairs": '{"version":1}',
+                    },
+                )
+            duplicates = {"format": {"tags": {"PROMPT": GRAPH, "prompt": "{}"}}}
+            with self._run_returning(stdout=json.dumps(duplicates)):
+                self.assertEqual(read_video_metadata("clip.mkv"), {})
+            markers = {"format": {"tags": {"comment": json.dumps({"prompt": None})}}}
+            with self._run_returning(stdout=json.dumps(markers)):
+                self.assertEqual(read_video_metadata("clip.mp4"), {})
+
     def test_ffprobe_output_is_parsed(self):
         payload = {"format": {"tags": {"comment": json.dumps({"prompt": GRAPH})}}}
         with mock.patch.object(
@@ -155,6 +351,39 @@ class TestReadVideoMetadataWithRealFfprobe(unittest.TestCase):
         self.assertEqual(meta["prompt"], GRAPH)
         self.assertEqual(json.loads(meta["workflow"]), {"v": 1})
 
+    def test_per_key_tags_round_trip_through_real_containers(self):
+        for name in ("per_key.mp4", "per_key.mkv", "per_key.webm"):
+            path = os.path.join(self.tmpdir, name)
+            write_video(path, prompt=GRAPH, workflow='{"nodes":[]}')
+            self.assertEqual(
+                read_video_metadata(path),
+                {"prompt": GRAPH, "workflow": '{"nodes":[]}'},
+                name,
+            )
+
+    def test_legacy_ilst_comment_atom_is_read(self):
+        path = os.path.join(self.tmpdir, "vhs.mp4")
+        write_mp4(path, json.dumps({"prompt": GRAPH}))
+        probe = subprocess.run(
+            [
+                shutil.which("ffprobe"),
+                "-v",
+                "quiet",
+                "-show_entries",
+                "format_tags",
+                "-of",
+                "json",
+                path,
+            ],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=60,
+        )
+        tags = json.loads(probe.stdout)["format"]["tags"]
+        self.assertIn("comment", tags)
+        self.assertEqual(read_video_metadata(path), {"prompt": GRAPH})
+
     def test_video_without_a_comment_is_empty(self):
         path = os.path.join(self.tmpdir, "plain.mp4")
         write_mp4(path, "")
diff --git a/utils/prompt_inspector.py b/utils/prompt_inspector.py
index 62b9515..9c1bb08 100644
--- a/utils/prompt_inspector.py
+++ b/utils/prompt_inspector.py
@@ -23,7 +23,15 @@ from .prompt_pair_resolver import (
     ResolverProfile,
     resolve_prompt_pair,
 )
-from .video_metadata import ffprobe_path, is_video_path
+from .video_metadata import (
+    ENVELOPE_TAG_KEYS,
+    ROLE_TAG_KEYS,
+    VIDEO_TAG_KEYS,
+    DuplicateVideoTagError,
+    ffprobe_path,
+    is_video_path,
+    select_video_carrier,
+)
 
 # Role records (prompt, parameters, the pair execution record) share one
 # budget, encoded and decoded: roles cannot be trusted from a partial record,
@@ -36,7 +44,18 @@ WORKFLOW_METADATA_BYTES = 4 * 1024 * 1024
 # repeated chunks cost bounded I/O instead of file size.
 DUPLICATE_WORKFLOW_BYTES = 2 * WORKFLOW_METADATA_BYTES
 DUPLICATE_ROLE_BYTES = 2 * ROLE_METADATA_BYTES
-MAX_METADATA_BYTES = ROLE_METADATA_BYTES  # video stdout bound, unchanged here
+# ffprobe stdout bounds. ffprobe JSON-escapes tag text (at most doubling it),
+# and the decoded per-record budgets below are the only gate that may make a
+# record fail or ``limited``: a video must read exactly as the identical PNG.
+# So the first probe allows twice the role and workflow budgets plus 64 KiB of
+# JSON envelope; past that only the per-key ``workflow`` tag can be dropped,
+# and the re-probe asks for role and envelope tags within twice the role
+# budget (an envelope cannot be split, so an over-budget one fails the read).
+MAX_METADATA_BYTES = ROLE_METADATA_BYTES
+VIDEO_PROBE_BYTES = 2 * (ROLE_METADATA_BYTES + WORKFLOW_METADATA_BYTES) + 64 * 1024
+VIDEO_REPROBE_BYTES = 2 * ROLE_METADATA_BYTES + 64 * 1024
+VIDEO_PROBE_SECONDS = 3
+_VIDEO_ROLE_PROBE_KEYS = ROLE_TAG_KEYS + ENVELOPE_TAG_KEYS
 MAX_KEYWORD_BYTES = 80  # a 1-79 byte PNG keyword plus its NUL separator
 MAX_HEADER_BYTES = 8 * 1024 * 1024
 MAX_PNG_CHUNKS = 10_000
@@ -86,6 +105,7 @@ class _VideoInspection:
     metadata: dict[str, str | None]
     record_formats: dict[str, str]
     carrier_text: str | None
+    limited: frozenset[str] = dataclass_field(default_factory=frozenset)
 
 
 def _inflate(data: bytes, limit: int) -> bytes | None:
@@ -274,16 +294,16 @@ def _png_metadata(stream: Any, size: int) -> tuple[dict, frozenset, int, int]:
     raise MetadataReadError("PNG chunk count exceeds inspection limits")
 
 
-def _video_metadata(path: Path) -> _VideoInspection:
-    probe = ffprobe_path()
-    if not probe:
-        raise MetadataReadError("Video metadata reader is unavailable")
+def _probe_video_tags(
+    probe: str, path: Path, keys: tuple[str, ...], limit: int
+) -> bytes | None:
+    """ffprobe's JSON for the container tags ``keys``; ``None`` past ``limit``."""
     command = [
         probe,
         "-v",
         "quiet",
         "-show_entries",
-        "format_tags=comment,description",
+        "format_tags=" + ",".join(keys),
         "-of",
         "json",
         str(path),
@@ -303,13 +323,13 @@ def _video_metadata(path: Path) -> _VideoInspection:
                 # Timeout and final cleanup can race with the process exiting.
                 pass
 
-    timer = threading.Timer(3, stop)
+    timer = threading.Timer(VIDEO_PROBE_SECONDS, stop)
     timer.daemon = True
     timer.start()
     try:
-        raw = process.stdout.read(MAX_METADATA_BYTES + 1)
-        if len(raw) > MAX_METADATA_BYTES:
-            raise MetadataReadError("Video metadata exceeds inspection limits")
+        raw = process.stdout.read(limit + 1)
+        if len(raw) > limit:
+            return None
         process.wait()
         if process.returncode:
             raise MetadataReadError("Video metadata could not be read")
@@ -318,49 +338,45 @@ def _video_metadata(path: Path) -> _VideoInspection:
         stop()
         process.wait()
         process.stdout.close()
+    return raw
+
+
+def _video_metadata(path: Path) -> _VideoInspection:
+    probe = ffprobe_path()
+    if not probe:
+        raise MetadataReadError("Video metadata reader is unavailable")
+    limited = frozenset()
+    raw = _probe_video_tags(probe, path, VIDEO_TAG_KEYS, VIDEO_PROBE_BYTES)
+    if raw is None:
+        raw = _probe_video_tags(
+            probe, path, _VIDEO_ROLE_PROBE_KEYS, VIDEO_REPROBE_BYTES
+        )
+        if raw is None:
+            raise MetadataReadError("Video metadata exceeds inspection limits")
+        # Only the per-key workflow tag was left out, so it is the record over
+        # budget: limited when the per-key carrier (or no carrier) is selected.
+        limited = frozenset({"workflow"})
     probed = json.loads(raw)
     if not isinstance(probed, dict) or not isinstance(probed.get("format"), dict):
         raise MetadataReadError("Malformed video metadata")
     tags = probed["format"].get("tags", {})
     if not isinstance(tags, dict):
         raise MetadataReadError("Malformed video metadata")
-    # The scan normalizer drops null fields and unknown keys. Inspection must
-    # retain explicit prompt/execution markers so malformed evidence cannot
-    # silently fall back to an unrelated saved recipe or another comment.
-    candidates = []
-    for key in ("comment", "description"):
-        raw = tags.get(key)
-        if not isinstance(raw, str):
-            continue
-        try:
-            envelope = json.loads(raw)
-        except ValueError:
-            continue
-        if not isinstance(envelope, dict):
-            continue
-        candidates.append((raw, envelope))
-    selected = next(
-        (item for item in candidates if _ROLE_METADATA_KEYS.intersection(item[1])),
-        None,
-    )
+    # The shared selector keeps explicit null role markers so malformed
+    # evidence cannot silently fall back to an unrelated saved recipe or
+    # another carrier; ``read_inspection_file`` refuses them.
+    try:
+        selected = select_video_carrier(tags)
+    except DuplicateVideoTagError:
+        raise MetadataReadError("Duplicate prompt metadata") from None
     if selected is None:
-        selected = next((item for item in candidates if "workflow" in item[1]), None)
-    if selected is not None:
-        raw, envelope = selected
-        metadata, formats = {}, {}
-        for field in _METADATA_KEYS.intersection(envelope):
-            value = envelope[field]
-            if field == "workflow" and not isinstance(value, (str, dict)):
-                formats[field] = "invalid"
-                continue
-            metadata[field] = (
-                value if value is None or isinstance(value, str) else json.dumps(value)
-            )
-            formats[field] = (
-                "original-text" if isinstance(value, str) else "reserialized-json"
-            )
-        return _VideoInspection(metadata, formats, raw)
-    return _VideoInspection({}, {}, None)
+        return _VideoInspection({}, {}, None, limited)
+    metadata, formats, carrier = selected
+    if carrier is not None:
+        # An envelope carrier is whole; the dropped per-key workflow belonged
+        # to the carrier that lost.
+        limited = frozenset()
+    return _VideoInspection(metadata, formats, carrier, limited)
 
 
 def read_inspection_file(path: Path) -> InspectionRead:
@@ -370,10 +386,11 @@ def read_inspection_file(path: Path) -> InspectionRead:
         format_name = width = height = None
         if is_video_path(path):
             video = _video_metadata(path)
-            metadata, formats, carrier = (
+            metadata, formats, carrier, limited = (
                 video.metadata,
                 video.record_formats,
                 video.carrier_text,
+                video.limited,
             )
         else:
             with path.open("rb") as stream:
diff --git a/utils/video_metadata.py b/utils/video_metadata.py
index a7884dc..382560d 100644
--- a/utils/video_metadata.py
+++ b/utils/video_metadata.py
@@ -1,14 +1,18 @@
 """Prompt metadata from video containers.
 
-ComfyUI's video savers (core SaveVideo, VideoHelperSuite, ...) write the
-queued graph into the container's ``comment`` tag as JSON:
-``{"prompt": "<graph json>", "workflow": {...}}``. ffprobe reads it on every
-platform; when ffprobe is not installed, videos are skipped with a single
-notice instead of one error per file.
+Current ComfyUI video savers (core SaveVideo/SaveWEBM, VideoHelperSuite since
+``c67ff3e``) write one container tag per key, each a JSON string:
+``prompt``, ``workflow``, ``parameters``, ``comfyassets_prompt_pairs``. Older
+VideoHelperSuite wrote one ``comment`` tag holding a JSON envelope
+``{"prompt": "<graph json>", "workflow": {...}}``; ``description`` is read as a
+fallback envelope. ffprobe reads the tags on every platform; when ffprobe is
+not installed, videos are skipped with a single notice instead of one error
+per file.
 
+``select_video_carrier`` picks exactly one carrier for both the inspector and
+the scan, so Images and ingestion read the same prompts from the same file.
 The result uses the same shape as PNG text chunks (string values keyed
-``prompt`` / ``workflow`` / ``parameters``) so the existing prompt parser
-handles both.
+``prompt`` / ``workflow`` / ...) so the existing prompt parser handles both.
 """
 
 import functools
@@ -20,10 +24,15 @@ import shutil
 # Fixed argv, absolute binary, no shell.
 import subprocess  # nosec B404
 
+from .pair_execution import EXECUTION_KEY
+
 VIDEO_EXTENSIONS = (".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", ".wmv")
 FFPROBE_TIMEOUT = 20  # seconds; a stuck probe must not hang a worker thread
-_COMMENT_TAGS = ("comment", "description")
-_METADATA_KEYS = ("prompt", "workflow", "parameters")
+# Carrier order: per-key tags, then the ``comment`` envelope, then ``description``.
+ROLE_TAG_KEYS = ("prompt", "parameters", EXECUTION_KEY)
+METADATA_TAG_KEYS = ("prompt", "workflow", "parameters", EXECUTION_KEY)
+ENVELOPE_TAG_KEYS = ("comment", "description")
+VIDEO_TAG_KEYS = METADATA_TAG_KEYS + ENVELOPE_TAG_KEYS
 _NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)  # no console flash on Windows
 
 logger = logging.getLogger("prompt_manager.video")
@@ -46,32 +55,122 @@ def reset_missing_notice():
     _missing_notice_shown = False
 
 
-def metadata_from_tags(tags) -> dict:
-    """Normalise container tags into PNG-text-chunk shaped metadata.
+class DuplicateVideoTagError(ValueError):
+    """Two container tags fold to the same metadata key with different text."""
 
-    Returns ``{}`` when no tag carries a JSON object with a known key.
+
+def _folded_tags(tags: dict) -> dict:
+    """Known tags keyed by their carrier name.
+
+    Per-key prompt carriers fold ASCII case: Matroska muxers upper-case tag
+    names and ffprobe reports them as written. Two tags folding to one key are
+    a duplicate record: byte-identical duplicates are one record (as for PNG
+    text chunks), differing ones fail the read. Envelope keys (``comment``,
+    ``description``) are read exactly as the legacy readers did, never folded.
     """
-    if not isinstance(tags, dict):
-        return {}
-    for tag in _COMMENT_TAGS:
-        raw = tags.get(tag)
+    folded = {}
+    for key, value in tags.items():
+        if not isinstance(key, str):
+            continue
+        if key in ENVELOPE_TAG_KEYS:
+            folded[key] = value
+            continue
+        name = key.lower()
+        if name not in METADATA_TAG_KEYS:
+            continue
+        if name in folded and folded[name] != value:
+            raise DuplicateVideoTagError(f"Duplicate video metadata tag: {name}")
+        folded[name] = value
+    return folded
+
+
+def _record(value, key):
+    """One carrier field as (metadata text or None, record format or None).
+
+    A null marker (or a non-text role value) has no record format: it is kept
+    in the metadata as ``None`` so the inspector refuses it instead of letting
+    fallback evidence win, while scans drop it.
+    """
+    if key == "workflow" and not isinstance(value, (str, dict)):
+        return None, "invalid"
+    if isinstance(value, str):
+        return value, "original-text"
+    if isinstance(value, dict):
+        return json.dumps(value), "reserialized-json"
+    return None, None
+
+
+def _carriers(folded: dict):
+    """Candidate carriers in order: (fields, raw envelope text or None)."""
+    per_key = {key: folded[key] for key in METADATA_TAG_KEYS if key in folded}
+    if per_key:
+        yield per_key, None
+    for tag in ENVELOPE_TAG_KEYS:
+        raw = folded.get(tag)
         if not isinstance(raw, str):
             continue
         try:
-            data = json.loads(raw)
+            envelope = json.loads(raw)
         except ValueError:
             continue
-        if not isinstance(data, dict):
+        if isinstance(envelope, dict):
+            yield envelope, raw
+
+
+def select_video_carrier(tags):
+    """Choose the one container carrier of prompt metadata.
+
+    Returns ``(metadata, record_formats, carrier_text)`` or ``None`` when no
+    carrier holds a known key. ``metadata`` maps known keys to their text, or
+    to ``None`` for an explicit null marker that the inspector must refuse;
+    ``record_formats`` is ``original-text``, ``reserialized-json`` (an envelope
+    object) or ``invalid`` (a workflow that is neither text nor an object), and
+    ``carrier_text`` is the exact envelope for reserialized values (``None``
+    for per-key tags). The first carrier holding a role key wins; otherwise the
+    first holding ``workflow``. Carriers are never merged.
+    Raises :class:`DuplicateVideoTagError` for differing fold-duplicate tags.
+    """
+    if not isinstance(tags, dict):
+        return None
+    candidates = list(_carriers(_folded_tags(tags)))
+    selected = next(
+        (item for item in candidates if any(key in item[0] for key in ROLE_TAG_KEYS)),
+        None,
+    )
+    if selected is None:
+        selected = next((item for item in candidates if "workflow" in item[0]), None)
+    if selected is None:
+        return None
+    fields, carrier_text = selected
+    metadata, formats = {}, {}
+    for key in METADATA_TAG_KEYS:
+        if key not in fields:
             continue
-        found = {}
-        for key in _METADATA_KEYS:
-            value = data.get(key)
-            if value is None:
-                continue
-            found[key] = value if isinstance(value, str) else json.dumps(value)
-        if found:
-            return found
-    return {}
+        text, record_format = _record(fields[key], key)
+        if record_format == "invalid":
+            formats[key] = record_format
+            continue
+        metadata[key] = text
+        if record_format is not None:
+            formats[key] = record_format
+    return metadata, formats, carrier_text
+
+
+def metadata_from_tags(tags) -> dict:
+    """Normalise container tags into PNG-text-chunk shaped metadata for scans.
+
+    Returns ``{}`` when no carrier holds a known key, and when fold-duplicate
+    tags make the file's evidence untrustworthy.
+    """
+    try:
+        selected = select_video_carrier(tags)
+    except DuplicateVideoTagError as e:
+        logger.debug(f"Video tags skipped: {e}")
+        return {}
+    if selected is None:
+        return {}
+    metadata, _, _ = selected
+    return {key: value for key, value in metadata.items() if isinstance(value, str)}
 
 
 def read_video_metadata(path, timeout=FFPROBE_TIMEOUT) -> dict: