fix(inspect): read per-key video prompt tags

The inspector asked ffprobe only for comment and description, so videos
written by core SaveVideo/SaveWEBM and current VideoHelperSuite (one tag per
key) were inspected as having no metadata. It now asks for
prompt,workflow,parameters,comfyassets_prompt_pairs,comment,description and
hands the tags to the shared carrier selector. ffprobe stdout is bounded at
the role and workflow budgets plus 64 KiB of envelope; past that it probes
once more without the workflow tag within the role budget and marks the
per-key workflow limited, while an over-budget envelope still fails the
read. Differing fold-duplicate tags fail like duplicate PNG keys.

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,52 @@ def graph(positive="portrait", negative="blurry"):
     }
 
 
+def write_video(path, **tags):
+    """Tiny black clip carrying one container tag per keyword (needs ffmpeg).
+
+    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 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():
@@ -385,9 +438,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 MAX_METADATA_BYTES, VIDEO_PROBE_BYTES
 
         (self.output_dir / "clip.mp4").write_bytes(b"video fixture")
         metadata = {"prompt": graph("Video portrait", "")}
@@ -412,14 +463,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" * (MAX_METADATA_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 +957,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 +968,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 +988,155 @@ 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, unchanged.
+        write_video(
+            self.output_dir / "vhs.mp4",
+            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 MAX_METADATA_BYTES, VIDEO_PROBE_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, MAX_METADATA_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", "")

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,14 @@ 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: every carrier tag fits the role and workflow budgets
+# plus 64 KiB of JSON envelope; past that only the per-key ``workflow`` tag can
+# be dropped, so the re-probe asks for role and envelope tags within 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 = ROLE_METADATA_BYTES + WORKFLOW_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 +101,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 +290,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 +319,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 +334,43 @@ 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, MAX_METADATA_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 +380,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: