From a77ea50227274886946747426ac467e4640fe680 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 22 Oct 2026 04:11:57 -0700 Subject: [PATCH] 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 | 241 ++++++++++++++++++++++++++--- utils/prompt_inspector.py | 107 +++++++------ 2 files changed, 277 insertions(+), 71 deletions(-) diff --git a/tests/test_api_prompt_inspector.py b/tests/test_api_prompt_inspector.py index c4702a9f..3a84f42e 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,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", "") diff --git a/utils/prompt_inspector.py b/utils/prompt_inspector.py index 62b95157..d544751c 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,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: