feat(api): expose reusable pairs and atomic metadata enrichment (#193)

database/pair_ingestion.py

@@ -0,0 +1,129 @@
+"""Atomic metadata enrichment for a single known generation context."""
+
+import json
+import os
+from typing import Any, Dict
+
+from .models import normalize_image_path
+from .operations import _coerce_file_info
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+
+def ingest_resolved_pair(
+    db: Any, image_path: str, metadata: Dict[str, Any], resolution: Dict[str, Any]
+) -> Dict[str, Any]:
+    state = resolution.get("state", "unresolved")
+    if state != "resolved":
+        return {"status": state, "pairs_added": 0, "links_added": 0}
+    positive, negative = resolution["positive"], resolution["negative"]
+    positive_text = positive.get("storage_text", positive["text"])
+    negative_text = negative.get("storage_text", negative["text"])
+    normalized = normalize_image_path(image_path)
+    provenance = {"resolver_version": 1, "positive": positive, "negative": negative}
+    encoded = json.dumps(
+        provenance, sort_keys=True, separators=(",", ":"), allow_nan=False
+    )
+    with db._pair_transaction() as conn:
+        images = conn.execute(
+            "SELECT gi.*,p.hash AS positive_hash FROM generated_images gi "
+            "JOIN prompts p ON p.id=gi.prompt_id "
+            "WHERE gi.file_path=? OR gi.image_path=?",
+            (normalized, image_path),
+        ).fetchall()
+        if len(images) > 1 or (
+            images and images[0]["positive_hash"] != generate_prompt_hash(positive_text)
+        ):
+            return {"status": "conflict", "pairs_added": 0, "links_added": 0}
+        existing = images[0] if images else None
+        if existing:
+            prior = conn.execute(
+                "SELECT n.hash AS negative_hash FROM image_prompt_pairs ip "
+                "JOIN prompt_pairs pp ON pp.id=ip.pair_id "
+                "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id WHERE ip.image_id=?",
+                (existing["id"],),
+            ).fetchall()
+            expected = (
+                generate_prompt_hash(negative_text) if negative_text.strip() else None
+            )
+            if any(row["negative_hash"] != expected for row in prior):
+                return {"status": "conflict", "pairs_added": 0, "links_added": 0}
+        before = conn.execute("SELECT COUNT(*) FROM prompt_pairs").fetchone()[0]
+        pair_id = db._save_pair(conn, positive_text, negative_text)
+        pair = conn.execute(
+            "SELECT * FROM prompt_pairs WHERE id=?", (pair_id,)
+        ).fetchone()
+        pairs_added = (
+            conn.execute("SELECT COUNT(*) FROM prompt_pairs").fetchone()[0] - before
+        )
+        file_size, width, height, image_format = _coerce_file_info(metadata)
+        snapshots = {
+            column: json.dumps(metadata[key], sort_keys=True, allow_nan=False)
+            for column, key in (
+                ("workflow_data", "workflow"),
+                ("prompt_metadata", "prompt"),
+                ("parameters", "parameters"),
+            )
+            if key in metadata
+        }
+        info = {
+            "file_size": file_size,
+            "width": width,
+            "height": height,
+            "format": image_format,
+        }
+        snapshots.update(
+            {key: value for key, value in info.items() if value is not None}
+        )
+        changed = False
+        if existing:
+            image_id = existing["id"]
+            updates = {
+                key: value for key, value in snapshots.items() if existing[key] != value
+            }
+            if updates:
+                conn.execute(
+                    "UPDATE generated_images SET "
+                    + ",".join(f"{key}=?" for key in updates)
+                    + " WHERE id=?",
+                    [*updates.values(), image_id],
+                )
+                changed = True
+        else:
+            columns = ["prompt_id", "image_path", "filename", "file_path", *snapshots]
+            values = [
+                pair["positive_prompt_id"],
+                image_path,
+                os.path.basename(image_path),
+                normalized,
+                *snapshots.values(),
+            ]
+            image_id = conn.execute(
+                "INSERT INTO generated_images("
+                + ",".join(columns)
+                + ") VALUES("
+                + ",".join("?" for _ in columns)
+                + ")",
+                values,
+            ).lastrowid
+        linked = conn.execute(
+            "SELECT 1 FROM image_prompt_pairs WHERE pair_id=? "
+            "AND image_id=? AND provenance=?",
+            (pair_id, image_id, encoded),
+        ).fetchone()
+        db._link_image_to_pair(conn, pair_id, image_id, provenance)
+        status = (
+            "added"
+            if existing is None
+            else "enriched" if changed or not linked else "unchanged"
+        )
+        return {
+            "status": status,
+            "pairs_added": pairs_added,
+            "links_added": int(not linked),
+            "pair_id": pair_id,
+            "image_id": image_id,
+        }

docs/reusable-prompt-pairs.md

@@ -0,0 +1,57 @@
+# Reusable positive and negative prompts
+
+Tracking: [#193](https://github.com/ComfyAssets/ComfyUI_PromptManager/issues/193).
+This work builds on the Split node in PR #192. Keep the feature unmerged until
+maintainer live testing is complete.
+
+## Storage contract
+
+Keep existing prompt IDs, text hashes, metadata, image links and node schemas.
+A component can serve either role. Ordered combinations refer to shared prompt
+rows; an explicitly empty negative uses a null reference. Missing or ambiguous
+metadata does not mean an empty negative. Pair metadata belongs to the
+combination; saving another combination does not overwrite component metadata.
+Existing trim/lowercase hash normalization remains unchanged.
+
+Saved components are immutable while referenced. To change text, create or reuse
+a component and save a new combination. Archive combinations that have generation
+history. Pruning and legacy editing/deletion must respect these dependencies.
+Historical workflow snapshots remain intact, including their original text.
+
+## Migration and recovery
+
+Create pair tables after existing legacy rebuild migrations, in an atomic,
+repeatable migration. Never scan media during schema initialization. Test fresh,
+legacy and current databases, repeated startup, injected failures and retries.
+Preserve old backup compatibility. Migrate and validate a disposable backup copy
+before replacing live database pages. Reject broken pair schemas, foreign keys,
+triggers and views. Use SQLite backup APIs so committed WAL data is included.
+
+## Metadata and scanning
+
+Resolve positive and negative from the same sampler/output context. Track both
+node ID and output slot; bound graph traversal and text construction. Unknown
+conditioning operations and conflicting sampler candidates remain unresolved.
+Never guess that the first encoder or most recent negative belongs to an image.
+
+Import high-confidence pairs atomically with their image association and source
+provenance. Repeated scans must not duplicate components, pairs or image links,
+regenerate thumbnails, or increment generation run counts. Preserve existing
+conflicting/manual associations for review. Report new, enriched, unchanged,
+ambiguous, unresolved and failed results separately.
+
+## Live node
+
+Use the shared toolkit and Split foundation. Positive and negative editors have
+equal size. Only positive receives prepend, append and automatic LoRA triggers.
+Keep upstream connections authoritative. Library selection previews both fields
+before an explicit load. Saving text changes creates another reusable combination.
+Verify Classic and Nodes 2.0 startup, switching both directions, reload, caching,
+linked inputs, popup behavior and lifecycle cleanup before maintainer testing.
+
+## Delivery gate
+
+Keep all existing coverage floors and test suites. Record regression failures
+before implementing each behavior. Tests use disposable state and the managed
+CPU host; never mutate the maintainer's running ComfyUI or personal database.
+Python changes require a ComfyUI restart. Live approval is required before merge.

py/api/__init__.py

@@ -22,6 +22,7 @@ from aiohttp import web
 from PIL import Image
 
 from .prompts import PromptRoutesMixin
+from .pairs import PairRoutesMixin
 from .images import ImageRoutesMixin
 from .admin import AdminRoutesMixin
 from .logging_routes import LoggingRoutesMixin
@@ -240,6 +241,7 @@ async def _gzip_middleware(request, handler):
 
 
 class PromptManagerAPI(
+    PairRoutesMixin,
     PromptRoutesMixin,
     ImageRoutesMixin,
     AdminRoutesMixin,
@@ -391,6 +393,7 @@ class PromptManagerAPI(
 
         # ── Register domain-specific routes from mixins ───────────────
 
+        self._register_pair_routes(routes)
         self._register_prompt_routes(routes)
         self._register_image_routes(routes)
         self._register_admin_routes(routes)

py/api/pairs.py

@@ -0,0 +1,202 @@
+"""Reusable ordered prompt combinations, separate from legacy prompt CRUD."""
+
+from functools import wraps
+
+from aiohttp import web
+
+from .prompts import parse_page_params
+
+try:
+    from ...database.pairs import PairDependencyError
+    from ...utils.validators import (
+        validate_category,
+        validate_prompt_text,
+        validate_rating,
+        validate_tags,
+    )
+except ImportError:
+    from database.pairs import PairDependencyError
+    from utils.validators import (
+        validate_category,
+        validate_prompt_text,
+        validate_rating,
+        validate_tags,
+    )
+
+
+def pair_response(handler):
+    @wraps(handler)
+    async def wrapped(self, request):
+        try:
+            return await handler(self, request)
+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
+        except (ValueError, TypeError) as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=400)
+        except Exception:
+            self.logger.exception("Prompt pair operation failed")
+            return web.json_response(
+                {"success": False, "error": "Prompt pair operation failed"}, status=500
+            )
+
+    return wrapped
+
+
+def metadata_fields(data: dict) -> dict:
+    fields = {}
+    for name in ("title", "category", "tags", "rating", "notes"):
+        if name not in data:
+            continue
+        value = data[name]
+        if name == "tags":
+            if not isinstance(value, list):
+                raise ValueError("Tags must be a list of strings")
+            validate_tags(value)
+        elif name == "rating":
+            if isinstance(value, bool):
+                raise ValueError("Rating must be an integer")
+            validate_rating(value)
+        else:
+            if value is not None and not isinstance(value, str):
+                raise ValueError(f"{name} must be a string")
+            if name == "category":
+                validate_category(value)
+            elif value is not None and len(value) > 10000:
+                raise ValueError(f"{name} is too long")
+        fields[name] = value
+    return fields
+
+
+async def object_body(request) -> dict:
+    data = await request.json()
+    if not isinstance(data, dict):
+        raise ValueError("Expected a JSON object")
+    return data
+
+
+def missing_pair():
+    return web.json_response({"success": False, "error": "Pair not found"}, status=404)
+
+
+class PairRoutesMixin:
+    def _register_pair_routes(self, routes) -> None:
+        routes.get("/prompt_manager/pairs/export")(self.export_prompt_pairs)
+        routes.get("/prompt_manager/pairs")(self.list_prompt_pairs)
+        routes.post("/prompt_manager/pairs")(self.save_prompt_pair)
+        routes.get("/prompt_manager/pairs/{pair_id}")(self.get_prompt_pair)
+        routes.patch("/prompt_manager/pairs/{pair_id}")(self.update_prompt_pair)
+        routes.delete("/prompt_manager/pairs/{pair_id}")(self.delete_prompt_pair)
+
+    @pair_response
+    async def save_prompt_pair(self, request):
+        data = await object_body(request)
+        positive = data.get("positive_text")
+        negative = data.get("negative_text")
+        validate_prompt_text(positive)
+        if not isinstance(negative, str):
+            raise ValueError(
+                "negative_text must be explicit; use an empty string for none"
+            )
+        if negative.strip():
+            validate_prompt_text(negative)
+        fields = metadata_fields(data)
+        pair_id = await self._run_in_executor(
+            self.db.save_prompt_pair, positive, negative, **fields
+        )
+        pair = await self._run_in_executor(self.db.get_prompt_pair, pair_id)
+        return web.json_response({"success": True, "pair": pair})
+
+    @pair_response
+    async def get_prompt_pair(self, request):
+        pair = await self._run_in_executor(
+            self.db.get_prompt_pair, int(request.match_info["pair_id"])
+        )
+        return (
+            web.json_response({"success": True, "pair": pair})
+            if pair
+            else missing_pair()
+        )
+
+    @pair_response
+    async def list_prompt_pairs(self, request):
+        limit, offset = parse_page_params(request.query)
+        archived = request.query.get("include_archived", "false")
+        if archived not in ("true", "false"):
+            raise ValueError("include_archived must be true or false")
+        result = await self._run_in_executor(
+            self.db.list_prompt_pairs,
+            text=request.query.get("text"),
+            category=request.query.get("category"),
+            tags=request.query.getall("tag", []),
+            include_archived=archived == "true",
+            limit=limit,
+            offset=offset,
+        )
+        return web.json_response({"success": True, **result})
+
+    @pair_response
+    async def update_prompt_pair(self, request):
+        pair_id = int(request.match_info["pair_id"])
+        data = await object_body(request)
+        allowed = {"title", "category", "tags", "rating", "notes", "archived"}
+        if not data or set(data) - allowed:
+            raise ValueError(
+                "Update metadata or archive; save a new pair to change text"
+            )
+        fields = metadata_fields(data)
+        if "archived" in data:
+            if not isinstance(data["archived"], bool) or fields:
+                raise ValueError("Archive must be a separate boolean update")
+            updated = await self._run_in_executor(
+                self.db.archive_prompt_pair, pair_id, data["archived"]
+            )
+        else:
+            updated = await self._run_in_executor(
+                self.db.update_prompt_pair_metadata, pair_id, **fields
+            )
+        if not updated:
+            return missing_pair()
+        pair = await self._run_in_executor(self.db.get_prompt_pair, pair_id)
+        return web.json_response({"success": True, "pair": pair})
+
+    @pair_response
+    async def delete_prompt_pair(self, request):
+        deleted = await self._run_in_executor(
+            self.db.delete_prompt_pair, int(request.match_info["pair_id"])
+        )
+        return web.json_response({"success": True}) if deleted else missing_pair()
+
+    @pair_response
+    async def export_prompt_pairs(self, request):
+        pairs = []
+        offset = 0
+        while True:
+            result = await self._run_in_executor(
+                self.db.list_prompt_pairs,
+                include_archived=True,
+                limit=500,
+                offset=offset,
+            )
+            for pair in result["pairs"]:
+                pairs.append(
+                    {
+                        key: pair[key]
+                        for key in (
+                            "positive_text",
+                            "negative_text",
+                            "title",
+                            "category",
+                            "tags",
+                            "rating",
+                            "notes",
+                            "archived",
+                        )
+                    }
+                )
+            offset += len(result["pairs"])
+            if not result["has_more"] or not result["pairs"]:
+                break
+        return web.json_response(
+            {"schema_version": 1, "kind": "prompt-manager-pairs", "pairs": pairs},
+            headers={"Content-Disposition": 'attachment; filename="prompt-pairs.json"'},
+        )

py/api/prompts.py

@@ -6,6 +6,11 @@ import os
 import sqlite3
 
 from aiohttp import web
+
+try:
+    from ...database.pairs import PairDependencyError
+except ImportError:
+    from database.pairs import PairDependencyError
 from urllib.parse import quote
 
 # Upper bounds on page sizes and offsets for every list endpoint. The offset
@@ -793,6 +798,8 @@ class PromptRoutesMixin:
                     status=404,
                 )
 
+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
         except ValueError:
             return web.json_response(
                 {"success": False, "error": "Invalid prompt ID"}, status=400
@@ -840,6 +847,8 @@ class PromptRoutesMixin:
                     {"success": False, "error": "Prompt not found"}, status=404
                 )
 
+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
         except sqlite3.IntegrityError:
             return web.json_response(
                 {"success": False, "error": "Another prompt already has this text"},
@@ -1081,6 +1090,8 @@ class PromptRoutesMixin:
                 }
             )
 
+        except PairDependencyError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=409)
         except Exception as e:
             self.logger.error(f"Bulk delete error: {e}")
             return web.json_response(

tests/test_api_pairs.py

@@ -0,0 +1,115 @@
+"""Pair routes preserve shared text and expose portable combinations."""
+
+from tests.test_api_prompts import PromptAPITestCase
+
+
+class TestPairAPI(PromptAPITestCase):
+    async def _create(self, **overrides):
+        payload = {"positive_text": "cat", "negative_text": "blur"}
+        payload.update(overrides)
+        return await self.client.post("/prompt_manager/pairs", json=payload)
+
+    async def test_create_reuse_read_and_search(self):
+        response = await self._create(tags=["studio"], title="Portrait")
+        self.assertEqual(response.status, 200)
+        pair = (await response.json())["pair"]
+        response = await self._create(negative_text="noise")
+        second = (await response.json())["pair"]
+        self.assertEqual(pair["positive_prompt_id"], second["positive_prompt_id"])
+        self.assertNotEqual(pair["id"], second["id"])
+        response = await self.client.get("/prompt_manager/pairs?text=blur")
+        data = await response.json()
+        self.assertEqual(data["total"], 1)
+        self.assertEqual(data["pairs"][0]["tags"], ["studio"])
+        response = await self.client.get(f'/prompt_manager/pairs/{pair["id"]}')
+        self.assertEqual((await response.json())["pair"]["title"], "Portrait")
+
+    async def test_negative_must_be_explicit_but_can_be_empty(self):
+        response = await self.client.post(
+            "/prompt_manager/pairs", json={"positive_text": "cat"}
+        )
+        self.assertEqual(response.status, 400)
+        response = await self._create(negative_text="")
+        self.assertEqual(response.status, 200)
+        self.assertIsNone((await response.json())["pair"]["negative_prompt_id"])
+
+    async def test_invalid_payloads_are_400_without_writes(self):
+        for body in (
+            [],
+            None,
+            {},
+            {"positive_text": 12, "negative_text": ""},
+            {"positive_text": "cat", "negative_text": None},
+            {"positive_text": "cat", "negative_text": "blur", "rating": True},
+            {"positive_text": "cat", "negative_text": "blur", "tags": [12]},
+        ):
+            response = await self.client.post("/prompt_manager/pairs", json=body)
+            self.assertEqual(response.status, 400, body)
+        response = await self.client.get("/prompt_manager/pairs")
+        self.assertEqual((await response.json())["total"], 0)
+
+    async def test_repeated_save_does_not_overwrite_metadata(self):
+        response = await self._create(title="Original", tags=["kept"])
+        pair = (await response.json())["pair"]
+        response = await self._create(title="Other", tags=["ignored"])
+        repeated = (await response.json())["pair"]
+        self.assertEqual(repeated["id"], pair["id"])
+        self.assertEqual(repeated["title"], "Original")
+        self.assertEqual(repeated["tags"], ["kept"])
+
+    async def test_metadata_update_and_archive(self):
+        pair = (await (await self._create()).json())["pair"]
+        url = f'/prompt_manager/pairs/{pair["id"]}'
+        response = await self.client.patch(url, json={"title": "New", "tags": ["a"]})
+        self.assertEqual(response.status, 200)
+        response = await self.client.patch(url, json={"positive_text": "dog"})
+        self.assertEqual(response.status, 400)
+        response = await self.client.patch(url, json={"archived": True})
+        self.assertEqual(response.status, 200)
+        self.assertEqual(
+            (await (await self.client.get("/prompt_manager/pairs")).json())["total"], 0
+        )
+        response = await self.client.get("/prompt_manager/pairs?include_archived=true")
+        self.assertEqual((await response.json())["total"], 1)
+        response = await self.client.delete(url)
+        self.assertEqual(response.status, 200)
+        self.assertIsNotNone(self.api.db.get_prompt_by_id(pair["positive_prompt_id"]))
+
+    async def test_export_is_versioned_and_contains_text_not_only_local_ids(self):
+        await self._create(tags=["a"])
+        response = await self.client.get("/prompt_manager/pairs/export")
+        self.assertEqual(response.status, 200)
+        data = await response.json()
+        self.assertEqual(data["schema_version"], 1)
+        self.assertEqual(data["pairs"][0]["positive_text"], "cat")
+        self.assertEqual(data["pairs"][0]["negative_text"], "blur")
+        self.assertNotIn("positive_prompt_id", data["pairs"][0])
+
+    async def test_missing_invalid_ids_and_pagination(self):
+        for verb in ("get", "delete"):
+            for suffix, status in (("abc", 400), ("9999", 404)):
+                response = await getattr(self.client, verb)(
+                    f"/prompt_manager/pairs/{suffix}"
+                )
+                self.assertEqual(response.status, status)
+        for query in ("limit=abc", "offset=abc", "include_archived=maybe"):
+            response = await self.client.get("/prompt_manager/pairs?" + query)
+            self.assertEqual(response.status, 400)
+        response = await self.client.get("/prompt_manager/pairs?limit=99999&offset=-1")
+        data = await response.json()
+        self.assertEqual(data["limit"], 500)
+        self.assertEqual(data["offset"], 0)
+
+    async def test_old_edit_and_delete_routes_report_pair_dependency(self):
+        pair = (await (await self._create()).json())["pair"]
+        prompt_id = pair["positive_prompt_id"]
+        response = await self.client.put(
+            f"/prompt_manager/prompts/{prompt_id}", json={"text": "dog"}
+        )
+        self.assertEqual(response.status, 409)
+        response = await self.client.delete(f"/prompt_manager/delete/{prompt_id}")
+        self.assertEqual(response.status, 409)
+        response = await self.client.post(
+            "/prompt_manager/bulk/delete", json={"prompt_ids": [prompt_id]}
+        )
+        self.assertEqual(response.status, 409)

tests/test_pair_ingestion.py

@@ -0,0 +1,123 @@
+"""Atomic, repeatable metadata enrichment without changing usage counts."""
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from database.operations import PromptDatabase
+from database.pair_ingestion import ingest_resolved_pair
+from utils.hashing import generate_prompt_hash
+
+
+def resolved(positive="cat", negative="blur"):
+    return {
+        "state": "resolved",
+        "positive": {"text": positive, "provenance": {"source": "api"}},
+        "negative": {"text": negative, "provenance": {"source": "api"}},
+    }
+
+
+class TestPairIngestion(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.db = PromptDatabase(str(Path(self.temp.name) / "test.db"))
+        self.path = str(Path(self.temp.name) / "image.png")
+
+    def tearDown(self):
+        self.db.close_all()
+        self.temp.cleanup()
+
+    def test_first_scan_repeat_and_negative_variants_reuse_components(self):
+        first = ingest_resolved_pair(self.db, self.path, {}, resolved())
+        self.assertEqual(first["status"], "added")
+        self.assertEqual(first["pairs_added"], 1)
+        second = ingest_resolved_pair(self.db, self.path, {}, resolved())
+        self.assertEqual(second["status"], "unchanged")
+        self.assertEqual(second["pairs_added"], 0)
+        pairs = self.db.list_prompt_pairs()["pairs"]
+        self.assertEqual(len(pairs), 1)
+        self.assertEqual(pairs[0]["image_count"], 1)
+        self.assertEqual(pairs[0]["run_count"], 0)
+        self.assertEqual(
+            self.db.get_prompt_by_hash(generate_prompt_hash("cat"))["run_count"], 0
+        )
+        ingest_resolved_pair(self.db, self.path + "2", {}, resolved(negative="noise"))
+        self.assertEqual(
+            self.db.model.get_connection()
+            .execute("SELECT count(*) FROM prompts")
+            .fetchone()[0],
+            3,
+        )
+
+    def test_existing_image_enriched_in_place_and_metadata_preserved(self):
+        pid = self.db.save_prompt(
+            "cat", category="Personal", prompt_hash=generate_prompt_hash("cat")
+        )
+        image_id = self.db.link_image_to_prompt(pid, self.path)
+        result = ingest_resolved_pair(
+            self.db, self.path, {"prompt": {"a": 1}}, resolved()
+        )
+        self.assertEqual(result["status"], "enriched")
+        self.assertEqual(result["image_id"], image_id)
+        self.assertEqual(self.db.get_prompt_by_id(pid)["category"], "Personal")
+        row = self.db.get_image_by_id(image_id)
+        self.assertIn("a", row["prompt_metadata"])
+
+    def test_conflicting_image_is_not_reassigned_or_duplicated(self):
+        pid = self.db.save_prompt("dog", prompt_hash=generate_prompt_hash("dog"))
+        self.db.link_image_to_prompt(pid, self.path)
+        result = ingest_resolved_pair(self.db, self.path, {}, resolved())
+        self.assertEqual(result["status"], "conflict")
+        self.assertEqual(self.db.list_prompt_pairs()["total"], 0)
+        self.assertEqual(
+            self.db.model.get_connection()
+            .execute("SELECT count(*) FROM generated_images")
+            .fetchone()[0],
+            1,
+        )
+
+    def test_unknown_and_missing_negative_never_become_empty_pairs(self):
+        for state in ("absent", "ambiguous", "unresolved"):
+            value = resolved()
+            value["state"] = state
+            result = ingest_resolved_pair(self.db, self.path, {}, value)
+            self.assertEqual(result["status"], state)
+        self.assertEqual(self.db.list_prompt_pairs()["total"], 0)
+        result = ingest_resolved_pair(self.db, self.path, {}, resolved(negative=""))
+        self.assertEqual(result["status"], "added")
+
+    def test_failure_after_inserts_rolls_back_all_rows(self):
+        with patch.object(
+            self.db, "_link_image_to_pair", side_effect=RuntimeError("injected")
+        ):
+            with self.assertRaises(RuntimeError):
+                ingest_resolved_pair(self.db, self.path, {}, resolved())
+        conn = self.db.model.get_connection()
+        for table in (
+            "prompts",
+            "prompt_pairs",
+            "generated_images",
+            "image_prompt_pairs",
+        ):
+            self.assertEqual(
+                conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0], 0
+            )
+
+    def test_distinct_paths_are_not_collapsed_and_effective_text_stays_in_provenance(
+        self,
+    ):
+        value = resolved("prefix cat suffix")
+        value["positive"]["storage_text"] = "cat"
+        a = ingest_resolved_pair(self.db, self.path, {}, value)
+        b = ingest_resolved_pair(
+            self.db, self.path.replace("image.png", "other/image.png"), {}, value
+        )
+        self.assertEqual(a["pair_id"], b["pair_id"])
+        pair = self.db.get_prompt_pair(a["pair_id"])
+        self.assertEqual(pair["positive_text"], "cat")
+        self.assertEqual(pair["image_count"], 2)
+        association = self.db.get_image_prompt_pairs(a["image_id"])[0]
+        self.assertEqual(
+            association["provenance"]["positive"]["text"], "prefix cat suffix"
+        )