fix(ingest): never write a text capture through an existing entry

Binary captures pick their filename with lstat so they never replace an
existing directory entry (#4393). Text captures (webpages, tweets, arXiv)
still used Path.exists(), which follows symlinks: a dangling symlink named
like the capture reported "free" and write_text() followed it out of the
corpus directory. The loop also stopped at 999 numbered names and then
overwrote the last one. Share the lstat-based selection between both
paths.

graphify/ingest.py

@@ -221,22 +221,28 @@ Source: {url}
     return content, filename
 
 
-def _download_binary(url: str, suffix: str, target_dir: Path) -> Path:
-    """Save a sequential capture without replacing an existing directory entry.
+def _free_capture_path(target_dir: Path, filename: str) -> Path:
+    """Pick *filename*, or the first free numbered variant, in *target_dir*.
 
+    Uses lstat so no existing directory entry is replaced, including a
+    dangling symlink that a write would otherwise follow out of the corpus.
     Selection and writing are separate; concurrent captures are not atomic.
     """
-    filename = _safe_filename(url, suffix)
-    stem = Path(filename).stem
+    stem, suffix = Path(filename).stem, Path(filename).suffix
     out_path = target_dir / filename
     counter = 1
     while True:
         try:
             out_path.lstat()
         except FileNotFoundError:
-            break
+            return out_path
         out_path = target_dir / f"{stem}_{counter}{suffix}"
         counter += 1
+
+
+def _download_binary(url: str, suffix: str, target_dir: Path) -> Path:
+    """Save a sequential capture without replacing an existing directory entry."""
+    out_path = _free_capture_path(target_dir, _safe_filename(url, suffix))
     out_path.write_bytes(safe_fetch(url))
     return out_path
 
@@ -282,14 +288,7 @@ def ingest(url: str, target_dir: Path, author: str | None = None, contributor: s
     except (urllib.error.HTTPError, urllib.error.URLError, OSError) as exc:
         raise RuntimeError(f"ingest: failed to fetch {url!r}: {exc}") from exc
 
-    out_path = target_dir / filename
-    # Avoid overwriting - append counter if needed
-    counter = 1
-    while out_path.exists() and counter < 1000:
-        stem = Path(filename).stem
-        out_path = target_dir / f"{stem}_{counter}.md"
-        counter += 1
-
+    out_path = _free_capture_path(target_dir, filename)
     out_path.write_text(content, encoding="utf-8")
     print(f"Saved {url_type}: {out_path.name}")
     return out_path

tests/test_ingest_url_type.py

@@ -231,3 +231,31 @@ def test_binary_capture_fetch_failure_preserves_existing_file(tmp_path, monkeypa
         ingest_mod.ingest("https://example.org/report.pdf", tmp_path)
     assert retained.read_bytes() == b"retained capture"
     assert list(tmp_path.iterdir()) == [retained]
+
+
+@pytest.mark.parametrize("entry_kind", ["dangling_link", "file"])
+def test_text_captures_skip_existing_entries(entry_kind, tmp_path, monkeypatch):
+    """A webpage capture must not replace an existing entry either; a dangling
+    symlink in particular would otherwise be written through to its target."""
+    corpus = tmp_path / "corpus"
+    corpus.mkdir()
+    occupied = corpus / "example_org_page.md"
+    sentinel = tmp_path / "sentinel"
+    if entry_kind == "file":
+        occupied.write_text("retained file", encoding="utf-8")
+    else:
+        try:
+            occupied.symlink_to(sentinel)
+        except (OSError, NotImplementedError) as exc:
+            pytest.skip(f"host cannot create symlinks: {exc}")
+
+    monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url)
+    monkeypatch.setattr(ingest_mod, "safe_fetch_text", lambda url: "<title>T</title>body")
+    result = ingest_mod.ingest("https://example.org/page", corpus)
+
+    assert result == corpus / "example_org_page_1.md"
+    assert "body" in result.read_text(encoding="utf-8")
+    if entry_kind == "file":
+        assert occupied.read_text(encoding="utf-8") == "retained file"
+    else:
+        assert occupied.is_symlink() and not sentinel.exists()