fix: dedup race, cwd-relative image paths, unencoded legacy URLs

- SQLite dedup race: add UNIQUE(collection_name, file_hash) and switch
  save_document to a plain INSERT. add_document now catches the
  IntegrityError from a concurrent add of the same content, cleans up its
  managed files, and returns the winning doc_id — instead of two doc_ids
  for one file (each having paid for its own LLM indexing).

- PDF image paths: store the absolute path to each extracted image
  instead of a path relative to the indexing process's cwd. The
  ![image](...) references broke as soon as a query ran from a different
  directory.

- LegacyCloudAPI: URL-encode doc_id / retrieval_id path segments (added
  _enc()), matching CloudBackend. An id containing '/', '?', '#' or a
  space previously hit the wrong endpoint or produced a malformed URL.

Adds regression tests: UNIQUE enforcement, the add-race resolving to the
winner, absolute image paths, and encoded legacy URLs.

Claude-Session: https://claude.ai/code/session_01Kx5DgKbhK1N8autqXH8SmS

tests/test_legacy_sdk_contract.py

@@ -335,3 +335,17 @@ def test_is_retrieval_ready_propagates_api_errors(monkeypatch):
     monkeypatch.setattr("pageindex.cloud_api.requests.request", fake_request)
     with pytest.raises(PageIndexAPIError):
         PageIndexClient("pi-test").is_retrieval_ready("doc-1")
+
+
+def test_legacy_urls_encode_special_char_ids(monkeypatch):
+    """doc_id / retrieval_id must be URL-encoded into the path."""
+    urls = []
+    def fake_request(method, url, headers=None, **kwargs):
+        urls.append(url)
+        return FakeResponse(payload={"ok": True})
+    monkeypatch.setattr("pageindex.cloud_api.requests.request", fake_request)
+    client = PageIndexClient("pi-test")
+    client.get_document("a/b?c")
+    client.get_retrieval("x y")
+    assert "a%2Fb%3Fc" in urls[0] and "/a/b?c/" not in urls[0]
+    assert "x%20y" in urls[1]

tests/test_local_backend.py

@@ -180,3 +180,33 @@ def test_add_document_unknown_collection_fails_fast(backend, tmp_path):
     # Collection never created -> must raise before any parse/LLM work.
     with pytest.raises(CollectionNotFoundError, match="does not exist"):
         backend.add_document("ghost-collection", str(pdf))
+
+
+def test_add_document_race_returns_existing_id(backend, tmp_path, monkeypatch):
+    """If the pre-check misses but the INSERT hits UNIQUE (concurrent add),
+    add_document must clean up and return the winner's doc_id, not duplicate."""
+    import pageindex.backend.local as local_mod
+    pdf = tmp_path / "doc.pdf"
+    pdf.write_bytes(b"%PDF-1.4 body")
+    backend.get_or_create_collection("papers")
+
+    # Pretend a winning add already stored this content under "winner-id".
+    file_hash = backend._file_hash(str(pdf))
+    backend._storage.save_document("papers", "winner-id", {
+        "doc_name": "doc", "doc_type": "pdf", "file_hash": file_hash, "structure": [],
+    })
+    # Pre-check misses (returns None) so we reach the INSERT; the post-conflict
+    # lookup then returns the winner's id.
+    calls = {"n": 0}
+    def fake_find(col, h):
+        calls["n"] += 1
+        return None if calls["n"] == 1 else "winner-id"
+    monkeypatch.setattr(backend._storage, "find_document_by_hash", fake_find)
+    # avoid real parsing/LLM: stub parser + build_index
+    monkeypatch.setattr(backend, "_resolve_parser", lambda p: type("P", (), {
+        "parse": lambda self, fp, **k: type("PD", (), {"doc_name": "doc", "nodes": []})()
+    })())
+    monkeypatch.setattr(local_mod, "build_index", lambda parsed, model=None, opt=None: {"structure": [], "doc_description": ""})
+
+    result = backend.add_document("papers", str(pdf))
+    assert result == "winner-id"

tests/test_pdf_parser.py

@@ -27,3 +27,33 @@ def test_parse_nodes_are_flat_without_level():
         assert node.tokens >= 0
         assert node.index is not None
         assert node.level is None
+
+
+def test_image_paths_are_absolute(tmp_path):
+    """Image references must be absolute so they resolve regardless of cwd
+    (cwd-relative paths broke after the query ran from another directory)."""
+    import os
+    import pymupdf
+    from pageindex.parser.pdf import PdfParser
+
+    # Build a 1-page PDF with an embedded image (>= _MIN_IMAGE_SIZE).
+    pix = pymupdf.Pixmap(pymupdf.csRGB, pymupdf.IRect(0, 0, 64, 64), False)
+    pix.clear_with(128)
+    png = tmp_path / "img.png"
+    pix.save(str(png))
+
+    doc = pymupdf.open()
+    page = doc.new_page()
+    page.insert_image(pymupdf.Rect(20, 20, 180, 180), filename=str(png))
+    pdf_path = tmp_path / "withimg.pdf"
+    doc.save(str(pdf_path))
+    doc.close()
+
+    images_dir = tmp_path / "out" / "images"
+    result = PdfParser().parse(str(pdf_path), images_dir=str(images_dir))
+
+    img_paths = [im["path"] for n in result.nodes if n.images for im in n.images]
+    assert img_paths, "expected at least one extracted image"
+    for p in img_paths:
+        assert os.path.isabs(p), f"image path not absolute: {p}"
+        assert os.path.exists(p), f"image path does not resolve: {p}"

tests/test_sqlite_storage.py

@@ -79,3 +79,16 @@ def test_close_closes_connections_created_in_other_threads(storage):
     storage.close()  # main thread closes the worker's connection too
     with pytest.raises(sqlite3.ProgrammingError):
         conns["worker"].execute("SELECT 1")
+
+
+def test_duplicate_file_hash_in_collection_raises(storage):
+    """UNIQUE(collection_name, file_hash) guards the add-same-file race."""
+    import sqlite3
+    storage.create_collection("papers")
+    doc = {"doc_name": "a", "doc_type": "pdf", "file_hash": "HASH1", "structure": []}
+    storage.save_document("papers", "doc-1", doc)
+    with pytest.raises(sqlite3.IntegrityError):
+        storage.save_document("papers", "doc-2", {**doc, "doc_name": "b"})
+    # same hash in a DIFFERENT collection is fine
+    storage.create_collection("other")
+    storage.save_document("other", "doc-3", {**doc})