From 301d81acbc52469c35d78464e8455d82c6484223 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sun, 18 Oct 2026 03:39:01 -0700 Subject: [PATCH] fix(inspector): distinguish ambiguous saved associations --- database/operations.py | 10 +++- py/api/images.py | 14 ++++- tests/test_api_prompt_inspector.py | 84 +++++++++++++++++++++++++++++- tests/test_database.py | 64 +++++++++++++++++++++++ 4 files changed, 167 insertions(+), 5 deletions(-) diff --git a/database/operations.py b/database/operations.py index 3df8f5f..55c2614 100644 --- a/database/operations.py +++ b/database/operations.py @@ -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]) diff --git a/py/api/images.py b/py/api/images.py index 5eae351..b920ec3 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -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( diff --git a/tests/test_api_prompt_inspector.py b/tests/test_api_prompt_inspector.py index 3c08875..a297b5a 100644 --- a/tests/test_api_prompt_inspector.py +++ b/tests/test_api_prompt_inspector.py @@ -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 + ) diff --git a/tests/test_database.py b/tests/test_database.py index a73fcee..648faf4 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -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):