nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(inspector): distinguish ambiguous saved associations

Browse Source
This commit is contained in:
Vito Sansevero
2026-10-18 03:39:01 -07:00
parent 8ccc183b16
commit 301d81acbc
4 changed files with 167 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/operations.py
+9 -1
View File
@@ -2014,12 +2014,18 @@ class PromptDatabase(
return count
def get_image_prompt_info(
self, image_path: str, allow_filename_fallback: bool = True
self,
image_path: str,
allow_filename_fallback: bool = True,
*,
report_ambiguity: bool = False,
) -> Optional[Dict[str, Any]]:
"""Look up prompt + image metadata for a given image path.
Returns a dict with prompt_id, text, category, tags, rating, notes,
workflow_data, prompt_metadata, generation_time, parameters — or None.
Opting into report_ambiguity returns only an association_status marker
for multiple matches; default callers still receive None.
"""
with self.model.get_connection() as conn:
cursor = conn.execute(
@@ -2045,6 +2051,8 @@ class PromptDatabase(
]
matches = exact or rows
if len(matches) != 1:
if report_ambiguity and matches:
return {"association_status": "ambiguous"}
return None
row = matches[0]
associations = self.get_image_prompt_pairs(row[9])
py/api/images.py
+12 -2
View File
@@ -1746,7 +1746,9 @@ class ImageRoutesMixin:
return None
saved, unavailable = None, False
try:
saved = self.db.get_image_prompt_info(str(path), False)
saved = self.db.get_image_prompt_info(
str(path), False, report_ambiguity=True
)
except Exception:
unavailable = True
self.logger.warning("Saved inspection display could not be read")
@@ -1763,7 +1765,12 @@ class ImageRoutesMixin:
},
}
else:
evidence = inspect_prompt_evidence(inspection.metadata, saved)
role_saved = (
None
if saved and saved.get("association_status") == "ambiguous"
else saved
)
evidence = inspect_prompt_evidence(inspection.metadata, role_saved)
return encode_inspection_response(
image={
"root_index": index,
@@ -1827,6 +1834,9 @@ class ImageRoutesMixin:
}
if not isinstance(saved, dict) or not saved:
return result
if saved.get("association_status") == "ambiguous":
result["status"] = "ambiguous"
return result
pair = saved.get("pair")
if isinstance(pair, dict):
result.update(
tests/test_api_prompt_inspector.py
+82 -2
View File
@@ -369,7 +369,9 @@ class TestPromptInspector(ImageAPITestCase):
wraps=self.api.db.get_image_prompt_info,
) as lookup:
_, result = await self.inspect(root="1")
lookup.assert_called_once_with(str(other / "image.png"), False)
lookup.assert_called_once_with(
str(other / "image.png"), False, report_ambiguity=True
)
self.assertEqual(result["prompts"]["positive"]["status"], "absent")
async def test_video_inspection_uses_bounded_container_evidence(self):
@@ -707,7 +709,7 @@ class TestPromptInspector(ImageAPITestCase):
self.assertEqual(result["resolution"]["sampler_id"], "3")
self.assertEqual(result["library"]["status"], "linked")
self.assertEqual(reads, [path])
lookup.assert_called_once_with(str(path), False)
lookup.assert_called_once_with(str(path), False, report_ambiguity=True)
self.assertLessEqual(len(await response.read()), 8 * 1024 * 1024)
self.assertNotIn(str(self.comfy_dir), json.dumps(result))
@@ -1182,3 +1184,81 @@ class TestPromptInspector(ImageAPITestCase):
(await response.json())["root_scope"], descriptor["root_scope"]
)
roots.assert_called_once_with()
async def test_exact_saved_association_ambiguity_preserves_physical_file_evidence(
self,
):
for readable in (False, True):
with self.subTest(readable=readable):
filename = "readable.png" if readable else "empty.png"
metadata = (
{"prompt": json.dumps(graph("Physical portrait", ""))}
if readable
else {}
)
path = png(self.output_dir / filename, **metadata)
before = path.read_bytes()
results, lookups = [], []
for count, expected in ((0, "none"), (1, "linked"), (2, "ambiguous")):
if count:
positive = self._save_prompt(
f"Saved owner {readable} {count}",
category=f"Category {count}",
tags=[f"owner-{count}"],
)
self.api.db.link_image_to_prompt(positive, str(path))
with patch.object(
self.api.db,
"get_image_prompt_info",
wraps=self.api.db.get_image_prompt_info,
) as lookup:
response, result = await self.inspect(filename)
self.assertEqual(response.status, 200)
self.assertEqual(result["library"]["status"], expected)
self.assertEqual(lookup.call_count, 1)
lookups.append(lookup)
results.append(result)
self.assertEqual(result["metadata"]["raw"], metadata)
self.assertEqual(path.read_bytes(), before)
if count == 1:
self.assertEqual(result["library"]["positive_id"], positive)
self.assertEqual(
result["library"]["metadata_scope"], "positive"
)
self.assertEqual(result["library"]["category"], "Category 1")
self.assertEqual(result["library"]["tags"], ["owner-1"])
if not readable:
self.assertEqual(
result["prompts"]["positive"]["source"],
"saved-association",
)
if readable or count != 1:
self.assertEqual(result["prompts"], results[0]["prompts"])
self.assertEqual(
results[2]["library"],
{
"status": "ambiguous",
"scope": "saved-library",
"positive_id": None,
"pair_id": None,
"title": None,
"category": None,
"tags": [],
"metadata_scope": None,
},
)
for field in (
"image",
"file",
"metadata",
"prompts",
"resolution",
"root_scope",
):
self.assertEqual(results[2][field], results[0][field])
self.assertNotIn("Saved owner", json.dumps(results[2]))
self.assertNotIn("association_status", json.dumps(results[2]))
for lookup in lookups:
lookup.assert_called_once_with(
str(path), False, report_ambiguity=True
)
tests/test_database.py
+64
View File
@@ -960,6 +960,70 @@ class TestImageQueries(DatabaseTestCase):
self.assertIsNone(info["workflow_data"])
self.assertEqual(info["prompt_metadata"], {"k": 1})
def test_image_prompt_info_reports_ambiguity_only_when_requested(self):
stored = os.path.join("out", "exact.png")
self.assertIsNone(self.db.get_image_prompt_info(stored, False))
self.assertIsNone(
self.db.get_image_prompt_info(stored, False, report_ambiguity=True)
)
first = self._save("First owner", category="First", tags=["first"])
self.db.link_image_to_prompt(first, stored, self.METADATA)
original = self.db.get_image_prompt_info(stored, False)
self.assertEqual(original["prompt_id"], first)
self.assertEqual(original["category"], "First")
self.assertEqual(original["tags"], ["first"])
self.assertEqual(
self.db.get_image_prompt_info(stored, False, report_ambiguity=True),
original,
)
second = self._save("Second owner", category="Second", tags=["second"])
self.db.link_image_to_prompt(second, stored, self.METADATA)
self.assertIsNone(self.db.get_image_prompt_info(stored))
self.assertIsNone(self.db.get_image_prompt_info(stored, False))
statements = []
with self.db.model.get_connection() as conn:
conn.set_trace_callback(statements.append)
try:
result = self.db.get_image_prompt_info(
stored, False, report_ambiguity=True
)
finally:
conn.set_trace_callback(None)
self.assertEqual(result, {"association_status": "ambiguous"})
self.assertEqual(
len(
[sql for sql in statements if sql.lstrip().upper().startswith("SELECT")]
),
1,
)
self.assertIsNone(
self.db.get_image_prompt_info("missing.png", False, report_ambiguity=True)
)
def test_image_prompt_info_ambiguity_opt_in_keeps_exact_path_precedence(self):
first = self._save("Exact owner")
second = self._save("Same basename")
exact = os.path.join("out", "one", "same.png")
other = os.path.join("out", "two", "same.png")
self.db.link_image_to_prompt(first, exact)
self.db.link_image_to_prompt(second, other)
original = self.db.get_image_prompt_info(exact)
self.assertEqual(original["prompt_id"], first)
self.assertEqual(
self.db.get_image_prompt_info(exact, report_ambiguity=True), original
)
absent = os.path.join("elsewhere", "same.png")
self.assertIsNone(self.db.get_image_prompt_info(absent))
self.assertIsNone(
self.db.get_image_prompt_info(absent, False, report_ambiguity=True)
)
self.assertEqual(
self.db.get_image_prompt_info(absent, report_ambiguity=True),
{"association_status": "ambiguous"},
)
class TestMaintenanceOperations(DatabaseTestCase):
def test_statistics_count_everything(self):