Merge commit '3dfb828' into feat/247-foreign-workflows-pr3

TESTING.md

@@ -5076,7 +5076,12 @@ workflows without a PromptManager node. It does no file or database I/O.
   (`found`); two writers of one text store one prompt. A recorded run's stored
   prompt links the same way (`link_stored_prompt_or_raise`). A rescan therefore
   never adds a second link to a file an older scan linked to the legacy answer;
-  the row reads "Kept: already linked to a different prompt". Every scan write
+  the row reads "Kept: already linked to a different prompt". Link rows are
+  matched under the path and its `realpath` spelling (`file_link_rows`), so a
+  file the scan stored through the resolved folder is found when it arrives
+  through a symlinked or junction folder. The reverse order (a row stored under
+  the watcher's raw spelling, then a scan whose root resolves differently) is
+  matched only when that raw spelling equals the realpath. Every scan write
   runs through `survive_library_switch`, so a library move reruns it.
 - **Marks.** Every prompt ingestion creates, pair components included, carries
   category `scanned`, tag `auto-scanned` and the note `Auto-scanned from

database/pair_ingestion.py

@@ -70,13 +70,29 @@ def mark_created(conn: Any, db: Any, prompt_id: int, marks: Marks, name: str) ->
 
 
 def file_link_rows(conn: Any, image_path: str) -> List[Any]:
-    """The file's link rows (either spelling), each with its prompt's hash."""
-    return 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=?",
-        (normalize_image_path(image_path), image_path),
-    ).fetchall()
+    """The file's link rows, each with its prompt's hash.
+
+    Matched under the path and its resolved spelling (an output folder
+    reached through a symlink or junction), each normalized and raw;
+    deduplicated by row id.
+    """
+    rows: Dict[Any, Any] = {}
+    for spelling in dict.fromkeys((image_path, _resolved(image_path))):
+        for row in 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=?",
+            (normalize_image_path(spelling), spelling),
+        ).fetchall():
+            rows.setdefault(row["id"], row)
+    return list(rows.values())
+
+
+def _resolved(path: str) -> str:
+    try:
+        return os.path.realpath(path)
+    except (OSError, ValueError):
+        return path
 
 
 def link_conflict(rows: List[Any], prompt_hash: str) -> bool:

tests/test_scanned_prompts.py

@@ -187,6 +187,41 @@ class TestStoreScannedPrompt(StoreCase):
             )
         self.assertEqual((status, calls), ("added", [1]))
 
+    def _linked_folder(self):
+        """``(out_real, out_link)``: a symlink to the same output folder."""
+        from tests.test_api_duplicate_files import TestReview
+
+        if not TestReview._can_symlink(self.temp.name):  # pragma: no cover
+            self.skipTest("symlinks are not available")
+        real = Path(self.temp.name) / "out_real"
+        real.mkdir()
+        link = Path(self.temp.name) / "out_link"
+        os.symlink(str(real), str(link))
+        (real / "a.png").write_bytes(b"png")
+        return os.path.realpath(str(real)), str(link)
+
+    def test_a_file_linked_through_its_real_folder_is_found_through_a_link(self):
+        """The scan stores realpath spellings, the monitor the watcher's."""
+        real, link = self._linked_folder()
+        other = self.save("a dog")
+        self.db.link_image_to_prompt(other, os.path.join(real, "a.png"))
+        through_link = os.path.join(link, "a.png")
+        self.assertEqual(
+            store_scanned_prompt_or_raise(
+                self.db, through_link, META, "a cat", SCAN_MARKS
+            ),
+            ("conflict", None),
+        )
+        self.assertEqual(
+            store_scanned_prompt_or_raise(
+                self.db, through_link, META, "a dog", SCAN_MARKS
+            ),
+            ("unchanged", other),
+        )
+        pair = ingest_resolved_pair(self.db, through_link, META, resolved())
+        self.assertEqual(pair["status"], "conflict")
+        self.assertEqual(len(self.links()), 1)
+
 
 class TestPairMarks(StoreCase):
     def test_marks_reach_only_created_components(self):