fix(video): scan videos written by current ComfyUI savers

The output scan and live monitor read container tags through
metadata_from_tags, which now delegates to the shared carrier selector, so
per-key prompt, workflow, parameters and comfyassets_prompt_pairs tags
(including Matroska upper-cased names) are ingested with the default
resolver profile exactly like the same graph in a PNG. Fold-duplicate tags
and null markers are skipped quietly. Pins the behavior with a mocked
ffprobe scan test and real MP4/MKV/WebM round trips when ffmpeg is present.

tests/test_api_admin.py

@@ -2506,6 +2506,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

tests/test_utils_video_metadata.py

@@ -33,25 +33,31 @@ 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, **tags):
+    """Write a tiny black video carrying one container tag per keyword.
+
+    MP4 keeps arbitrary keys only with ``use_metadata_tags``, the movflag the
+    core SaveVideo node sets; Matroska and WebM upper-case the key names.
+    """
+    command = [
+        "ffmpeg",
+        "-v",
+        "quiet",
+        "-f",
+        "lavfi",
+        "-i",
+        "color=c=black:s=16x16:d=0.2",
+    ]
+    if 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_video(path, comment=comment)
 
 
 class TestMetadataFromTags(unittest.TestCase):
@@ -273,6 +279,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(
@@ -303,6 +340,16 @@ 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_video_without_a_comment_is_empty(self):
         path = os.path.join(self.tmpdir, "plain.mp4")
         write_mp4(path, "")