fix(cloud): align cloud/local backend contracts and harden the HTTP layer

Fixes the cloud/local contract mismatches from the PR #272 review
(verified against the official API docs — the cloud API has no
folder/collection endpoints publicly, GET /docs supports limit<=100
with offset):

- query_stream: emit a terminal answer_done event with the full answer
  (same contract as the local backend); raise CloudAPIError instead of
  disguising HTTP errors as answer events; move the initial connect
  inside try so a connection failure can no longer strand the consumer
  awaiting a sentinel that never arrives
- _request: rewind file objects before retrying so a transient 5xx/429
  during upload no longer re-sends an empty multipart body; carry the
  HTTP status on CloudAPIError (status_code) and keep the last status
  in the max-retries error
- list_documents: paginate with limit/offset instead of a hard-coded
  limit=100, so >100-doc collections are no longer silently truncated
  (whole-collection queries rely on this list)
- folders: treat only 403/404 as "folders unavailable" (warned via
  warnings.warn instead of an invisible logger.warning, matched on
  status_code instead of a "403" substring); transient errors now
  propagate instead of being permanently cached as folder_id=None
- error taxonomy parity: cloud doc endpoints map HTTP 404 to
  DocumentNotFoundError; local get_document raises DocumentNotFoundError
  instead of returning {}; local delete_document raises on missing
  doc_id instead of silently deleting nothing
- cloud get_document warns that include_text is not supported instead
  of silently ignoring it

Adds regression tests for each fix (11 new tests).

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

pageindex/backend/cloud.py

@@ -13,7 +13,7 @@ import urllib.parse
 import requests
 from typing import AsyncIterator
 
-from ..errors import CloudAPIError, PageIndexError
+from ..errors import CloudAPIError, DocumentNotFoundError, PageIndexError
 from ..events import QueryEvent
 
 logger = logging.getLogger(__name__)
@@ -32,33 +32,54 @@ class CloudBackend:
 
     # ── HTTP helpers ──────────────────────────────────────────────────────
 
+    # Folder API statuses meaning "folders are not available on this account"
+    # (403: requires Max plan; 404: endpoint not exposed). Anything else is a
+    # real error and must propagate rather than silently degrade.
+    _FOLDER_UNAVAILABLE = (403, 404)
+
     def _warn_folder_upgrade(self) -> None:
         if not self._folder_warning_shown:
-            logger.warning(
+            import warnings
-                "Folders (collections) require a Max plan. "
+            warnings.warn(
+                "Folders (collections) are not available on this plan. "
                 "All documents are stored in a single global space — collection names are ignored. "
-                "Upgrade at https://dash.pageindex.ai/subscription"
+                "Upgrade at https://dash.pageindex.ai/subscription",
+                UserWarning,
+                stacklevel=4,
             )
             self._folder_warning_shown = True
 
     def _request(self, method: str, path: str, **kwargs) -> dict:
         url = f"{API_BASE}{path}"
+        last_status: int | None = None
         for attempt in range(3):
+            if attempt and "files" in kwargs:
+                # Rewind file objects before a retry — the previous attempt
+                # consumed them, and re-sending without seek(0) would upload
+                # an empty multipart body.
+                for value in kwargs["files"].values():
+                    fobj = value[1] if isinstance(value, tuple) else value
+                    if hasattr(fobj, "seek"):
+                        fobj.seek(0)
             try:
                 resp = requests.request(method, url, headers=self._headers, timeout=30, **kwargs)
                 if resp.status_code in (429, 500, 502, 503):
                     logger.warning("Cloud API %s %s returned %d, retrying...", method, path, resp.status_code)
+                    last_status = resp.status_code
                     time.sleep(2 ** attempt)
                     continue
                 if resp.status_code != 200:
                     body = resp.text[:500] if resp.text else ""
-                    raise CloudAPIError(f"Cloud API error {resp.status_code}: {body}")
+                    raise CloudAPIError(f"Cloud API error {resp.status_code}: {body}",
+                                        status_code=resp.status_code)
                 return resp.json() if resp.content else {}
             except requests.RequestException as e:
                 if attempt == 2:
                     raise CloudAPIError(f"Cloud API request failed: {e}") from e
                 time.sleep(2 ** attempt)
-        raise CloudAPIError("Max retries exceeded")
+        raise CloudAPIError(f"Cloud API {method} {path} failed after retries"
+                            + (f" (last status {last_status})" if last_status else ""),
+                            status_code=last_status)
 
     @staticmethod
     def _validate_collection_name(name: str) -> None:
@@ -80,7 +101,7 @@ class CloudBackend:
             resp = self._request("POST", "/folder/", json={"name": name})
             self._folder_id_cache[name] = resp.get("folder", {}).get("id")
         except CloudAPIError as e:
-            if "403" in str(e):
+            if e.status_code in self._FOLDER_UNAVAILABLE:
                 self._warn_folder_upgrade()
                 self._folder_id_cache[name] = None
             else:
@@ -97,24 +118,33 @@ class CloudBackend:
             resp = self._request("POST", "/folder/", json={"name": name})
             self._folder_id_cache[name] = resp.get("folder", {}).get("id")
         except CloudAPIError as e:
-            if "403" in str(e):
+            if e.status_code in self._FOLDER_UNAVAILABLE:
                 self._warn_folder_upgrade()
                 self._folder_id_cache[name] = None
             else:
                 raise
 
     def _get_folder_id(self, name: str) -> str | None:
-        """Resolve collection name to folder ID. Returns None if folders not available."""
+        """Resolve collection name to folder ID. Returns None if folders not available.
+
+        Only "folders unavailable on this plan" (403/404) is cached as None —
+        transient errors (network, 5xx) propagate so a blip can't silently
+        drop documents into the global space forever.
+        """
         if name in self._folder_id_cache:
             return self._folder_id_cache.get(name)
         try:
             data = self._request("GET", "/folders/")
-            for folder in data.get("folders", []):
+        except CloudAPIError as e:
-                if folder.get("name") == name:
+            if e.status_code in self._FOLDER_UNAVAILABLE:
-                    self._folder_id_cache[name] = folder["id"]
+                self._warn_folder_upgrade()
-                    return folder["id"]
+                self._folder_id_cache[name] = None
-        except CloudAPIError:
+                return None
-            pass
+            raise
+        for folder in data.get("folders", []):
+            if folder.get("name") == name:
+                self._folder_id_cache[name] = folder["id"]
+                return folder["id"]
         self._folder_id_cache[name] = None
         return None
 
@@ -153,11 +183,30 @@ class CloudBackend:
 
         raise CloudAPIError(f"Document {doc_id} indexing timed out")
 
+    def _doc_request(self, doc_id: str, method: str, path: str, **kwargs) -> dict:
+        """Doc-scoped request: maps HTTP 404 to DocumentNotFoundError for
+        parity with the local backend's error taxonomy."""
+        try:
+            return self._request(method, path, **kwargs)
+        except CloudAPIError as e:
+            if e.status_code == 404:
+                raise DocumentNotFoundError(f"Document {doc_id} not found") from e
+            raise
+
     def get_document(self, collection: str, doc_id: str, include_text: bool = False) -> dict:
-        resp = self._request("GET", f"/doc/{self._enc(doc_id)}/metadata/")
+        if include_text:
+            import warnings
+            warnings.warn(
+                "include_text is not supported by the cloud backend; "
+                "returning the structure without node text. "
+                "Use get_page_content(doc_id, pages) to fetch content.",
+                UserWarning,
+                stacklevel=3,
+            )
+        resp = self._doc_request(doc_id, "GET", f"/doc/{self._enc(doc_id)}/metadata/")
         # Fetch structure in the same call via tree endpoint
-        tree_resp = self._request("GET", f"/doc/{self._enc(doc_id)}/",
+        tree_resp = self._doc_request(doc_id, "GET", f"/doc/{self._enc(doc_id)}/",
-                                  params={"type": "tree", "summary": "true"})
+                                      params={"type": "tree", "summary": "true"})
         raw_tree = tree_resp.get("tree", tree_resp.get("structure", tree_resp.get("result", [])))
         return {
             "doc_id": resp.get("id", doc_id),
@@ -169,12 +218,14 @@ class CloudBackend:
         }
 
     def get_document_structure(self, collection: str, doc_id: str) -> list:
-        resp = self._request("GET", f"/doc/{self._enc(doc_id)}/", params={"type": "tree", "summary": "true"})
+        resp = self._doc_request(doc_id, "GET", f"/doc/{self._enc(doc_id)}/",
+                                 params={"type": "tree", "summary": "true"})
         raw_tree = resp.get("tree", resp.get("structure", resp.get("result", [])))
         return self._normalize_tree(raw_tree)
 
     def get_page_content(self, collection: str, doc_id: str, pages: str) -> list:
-        resp = self._request("GET", f"/doc/{self._enc(doc_id)}/", params={"type": "ocr", "format": "page"})
+        resp = self._doc_request(doc_id, "GET", f"/doc/{self._enc(doc_id)}/",
+                                 params={"type": "ocr", "format": "page"})
         # Filter to requested pages
         from ..index.utils import parse_pages
         page_nums = set(parse_pages(pages))
@@ -210,22 +261,33 @@ class CloudBackend:
 
     def list_documents(self, collection: str) -> list[dict]:
         folder_id = self._get_folder_id(collection)
-        params = {"limit": 100}
+        # The API caps `limit` at 100; paginate with `offset` until a short
-        if folder_id:
+        # page comes back so collections with >100 docs aren't silently
-            params["folder_id"] = folder_id
+        # truncated (queries over the whole collection rely on this list).
-        data = self._request("GET", "/docs/", params=params)
+        page_size = 100
-        return [
+        offset = 0
-            {
+        docs: list[dict] = []
-                "doc_id": d.get("id", ""),
+        while True:
-                "doc_name": d.get("name", ""),
+            params = {"limit": page_size, "offset": offset}
-                "doc_description": d.get("description", ""),
+            if folder_id:
-                "doc_type": "pdf",
+                params["folder_id"] = folder_id
-            }
+            data = self._request("GET", "/docs/", params=params)
-            for d in data.get("documents", [])
+            batch = data.get("documents", [])
-        ]
+            docs.extend(
+                {
+                    "doc_id": d.get("id", ""),
+                    "doc_name": d.get("name", ""),
+                    "doc_description": d.get("description", ""),
+                    "doc_type": "pdf",
+                }
+                for d in batch
+            )
+            if len(batch) < page_size:
+                return docs
+            offset += page_size
 
     def delete_document(self, collection: str, doc_id: str) -> None:
-        self._request("DELETE", f"/doc/{self._enc(doc_id)}/")
+        self._doc_request(doc_id, "DELETE", f"/doc/{self._enc(doc_id)}/")
 
     # ── Query (uses cloud chat/completions, no LLM key needed) ────────────
 
@@ -269,32 +331,45 @@ class CloudBackend:
             )
         doc_id = doc_ids if doc_ids else self._get_all_doc_ids(collection)
         headers = self._headers
-        queue: asyncio.Queue[QueryEvent | None] = asyncio.Queue()
+        # Queue carries QueryEvent, an Exception to re-raise, or None (end).
-        loop = asyncio.get_event_loop()
+        queue: asyncio.Queue[QueryEvent | Exception | None] = asyncio.Queue()
+        loop = asyncio.get_running_loop()
+
+        def _put(item: QueryEvent | Exception | None) -> None:
+            try:
+                loop.call_soon_threadsafe(queue.put_nowait, item)
+            except RuntimeError:
+                pass  # event loop already closed; consumer is gone
 
         def _stream():
-            """Background thread: read SSE and push events to queue."""
+            """Background thread: read SSE and push events to queue.
-            resp = requests.post(
+
-                f"{API_BASE}/chat/completions/",
+            Everything — including the initial connect — runs inside try so a
-                headers=headers,
+            failure can never die silently and leave the consumer awaiting a
-                json={
+            sentinel that never arrives. Errors are forwarded as exceptions
-                    "messages": [{"role": "user", "content": question}],
+            (raised in the consumer), never disguised as answer events.
-                    "doc_id": doc_id,
+            """
-                    "stream": True,
+            resp = None
-                    "stream_metadata": True,
+            answer_parts: list[str] = []
-                },
-                stream=True,
-                timeout=120,
-            )
             try:
+                resp = requests.post(
+                    f"{API_BASE}/chat/completions/",
+                    headers=headers,
+                    json={
+                        "messages": [{"role": "user", "content": question}],
+                        "doc_id": doc_id,
+                        "stream": True,
+                        "stream_metadata": True,
+                    },
+                    stream=True,
+                    timeout=120,
+                )
                 if resp.status_code != 200:
                     body = resp.text[:500] if resp.text else ""
-                    loop.call_soon_threadsafe(
+                    raise CloudAPIError(
-                        queue.put_nowait,
+                        f"Cloud streaming error {resp.status_code}: {body}",
-                        QueryEvent(type="answer_done",
+                        status_code=resp.status_code,
-                                   data=f"Cloud streaming error {resp.status_code}: {body}"),
                     )
-                    return
 
                 current_tool_name = None
                 current_tool_args: list[str] = []
@@ -327,34 +402,40 @@ class CloudBackend:
                     elif block_type == "tool_use_stop":
                         if current_tool_name and current_tool_name not in _INTERNAL_TOOLS:
                             args_str = "".join(current_tool_args)
-                            loop.call_soon_threadsafe(
+                            _put(QueryEvent(type="tool_call", data={
-                                queue.put_nowait,
+                                "name": current_tool_name,
-                                QueryEvent(type="tool_call", data={
+                                "args": args_str,
-                                    "name": current_tool_name,
+                            }))
-                                    "args": args_str,
-                                }),
-                            )
                         current_tool_name = None
                         current_tool_args = []
 
                     elif block_type == "text" and content:
-                        loop.call_soon_threadsafe(
+                        answer_parts.append(content)
-                            queue.put_nowait,
+                        _put(QueryEvent(type="answer_delta", data=content))
-                            QueryEvent(type="answer_delta", data=content),
-                        )
 
+                # Same terminal contract as the local backend: a final
+                # answer_done event carrying the full answer text.
+                _put(QueryEvent(type="answer_done", data="".join(answer_parts)))
+
+            except requests.RequestException as e:
+                _put(CloudAPIError(f"Cloud streaming request failed: {e}"))
+            except Exception as e:
+                _put(e)
             finally:
-                resp.close()
+                if resp is not None:
-                loop.call_soon_threadsafe(queue.put_nowait, None)  # sentinel
+                    resp.close()
+                _put(None)  # sentinel
 
         thread = threading.Thread(target=_stream, daemon=True)
         thread.start()
 
         while True:
-            event = await queue.get()
+            item = await queue.get()
-            if event is None:
+            if item is None:
                 break
-            yield event
+            if isinstance(item, Exception):
+                raise item
+            yield item
 
         thread.join(timeout=5)
 

pageindex/backend/local.py

@@ -143,7 +143,7 @@ class LocalBackend:
         """
         doc = self._storage.get_document(collection, doc_id)
         if not doc:
-            return {}
+            raise DocumentNotFoundError(f"Document {doc_id} not found")
         doc["structure"] = self._storage.get_document_structure(collection, doc_id)
         if include_text:
             pages = self._storage.get_pages(collection, doc_id) or []
@@ -190,7 +190,11 @@ class LocalBackend:
 
     def delete_document(self, collection: str, doc_id: str) -> None:
         doc = self._storage.get_document(collection, doc_id)
-        if doc and doc.get("file_path"):
+        if not doc:
+            # Parity with the cloud backend, which surfaces HTTP 404 as
+            # DocumentNotFoundError — a typo'd doc_id should not pass silently.
+            raise DocumentNotFoundError(f"Document {doc_id} not found")
+        if doc.get("file_path"):
             Path(doc["file_path"]).unlink(missing_ok=True)
         # Clean up images directory: files/{collection}/{doc_id}/
         doc_dir = self._files_dir / collection / doc_id

pageindex/errors.py

@@ -27,8 +27,16 @@ class PageIndexAPIError(PageIndexError):
 
 
 class CloudAPIError(PageIndexAPIError):
-    """Cloud API returned error."""
+    """Cloud API returned error.
-    pass
+
+    ``status_code`` carries the HTTP status when the error came from an HTTP
+    response (None for transport-level failures), so callers can branch on it
+    instead of parsing the message.
+    """
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
 
 
 class FileTypeError(PageIndexError):

tests/test_cloud_backend.py

@@ -1,6 +1,12 @@
+import asyncio
+import io
+import json
+
 import pytest
 
+import pageindex.backend.cloud as cloud_mod
 from pageindex.backend.cloud import CloudBackend, API_BASE
+from pageindex.errors import CloudAPIError, DocumentNotFoundError
 
 
 def test_cloud_backend_init():
@@ -17,3 +23,182 @@ def test_query_rejects_empty_doc_ids():
     backend = CloudBackend(api_key="pi-test")
     with pytest.raises(ValueError, match="cannot be empty"):
         backend.query("col", "q", doc_ids=[])
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+class FakeResponse:
+    def __init__(self, status_code=200, json_data=None, text="", lines=None):
+        self.status_code = status_code
+        self._json = json_data if json_data is not None else {}
+        self.text = text
+        self.content = json.dumps(self._json).encode() if json_data is not None else b""
+        self._lines = lines or []
+
+    def json(self):
+        return self._json
+
+    def iter_lines(self, decode_unicode=True):
+        yield from self._lines
+
+    def close(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch):
+    monkeypatch.setattr(cloud_mod.time, "sleep", lambda *_: None)
+
+
+# ── _request: retry must rewind file objects (empty-upload regression) ──────
+
+def test_request_rewinds_file_on_retry(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    payload = b"%PDF-1.4 fake body"
+    fobj = io.BytesIO(payload)
+    uploads = []
+
+    def fake_request(method, url, headers=None, timeout=None, **kwargs):
+        # Simulate requests consuming the file body on every attempt.
+        uploads.append(kwargs["files"]["file"].read())
+        if len(uploads) == 1:
+            return FakeResponse(status_code=500)
+        return FakeResponse(status_code=200, json_data={"doc_id": "d1"})
+
+    monkeypatch.setattr(cloud_mod.requests, "request", fake_request)
+    resp = backend._request("POST", "/doc/", files={"file": fobj}, data={})
+    assert resp == {"doc_id": "d1"}
+    assert uploads[0] == payload
+    # Without seek(0) the retry would upload an empty body.
+    assert uploads[1] == payload
+
+
+# ── list_documents: pagination beyond the API's 100-doc page cap ─────────────
+
+def test_list_documents_paginates(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    backend._folder_id_cache["col"] = None  # skip folder lookup
+    calls = []
+
+    def fake_request(method, path, **kwargs):
+        offset = kwargs["params"]["offset"]
+        calls.append(offset)
+        n = 100 if offset == 0 else 30
+        return {"documents": [{"id": f"d{offset + i}", "name": ""} for i in range(n)]}
+
+    monkeypatch.setattr(backend, "_request", fake_request)
+    docs = backend.list_documents("col")
+    assert len(docs) == 130
+    assert calls == [0, 100]
+    assert docs[-1]["doc_id"] == "d129"
+
+
+# ── folder resolution: plan-limit vs transient errors ────────────────────────
+
+def test_folder_unavailable_warns_and_caches(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+
+    def fake_request(method, path, **kwargs):
+        raise CloudAPIError("Cloud API error 403: upgrade", status_code=403)
+
+    monkeypatch.setattr(backend, "_request", fake_request)
+    with pytest.warns(UserWarning, match="not available on this plan"):
+        assert backend._get_folder_id("col") is None
+    assert backend._folder_id_cache["col"] is None
+
+
+def test_folder_transient_error_propagates_and_is_not_cached(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+
+    def fake_request(method, path, **kwargs):
+        raise CloudAPIError("Cloud API request failed: connection reset")
+
+    monkeypatch.setattr(backend, "_request", fake_request)
+    with pytest.raises(CloudAPIError):
+        backend._get_folder_id("col")
+    # A blip must not permanently route documents to the global space.
+    assert "col" not in backend._folder_id_cache
+
+
+# ── doc endpoints: 404 maps to DocumentNotFoundError (local parity) ──────────
+
+def test_doc_404_maps_to_document_not_found(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+
+    def fake_request(method, path, **kwargs):
+        raise CloudAPIError("Cloud API error 404: not found", status_code=404)
+
+    monkeypatch.setattr(backend, "_request", fake_request)
+    with pytest.raises(DocumentNotFoundError):
+        backend.get_document_structure("col", "missing")
+    with pytest.raises(DocumentNotFoundError):
+        backend.delete_document("col", "missing")
+
+
+def test_get_document_include_text_warns(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    monkeypatch.setattr(backend, "_doc_request", lambda *a, **k: {"tree": []})
+    with pytest.warns(UserWarning, match="include_text is not supported"):
+        backend.get_document("col", "d1", include_text=True)
+
+
+# ── query_stream: terminal contract and error propagation ───────────────────
+
+def _collect_events(backend, **kwargs):
+    async def _run():
+        events = []
+        async for ev in backend.query_stream("col", "q", doc_ids=["d1"], **kwargs):
+            events.append(ev)
+        return events
+    return asyncio.run(_run())
+
+
+def _sse(block_type, content):
+    return "data: " + json.dumps({
+        "block_metadata": {"type": block_type},
+        "choices": [{"delta": {"content": content}}],
+    })
+
+
+def test_query_stream_emits_answer_done(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    lines = [_sse("text", "Hello "), _sse("text", "world"), "data: [DONE]"]
+    monkeypatch.setattr(
+        cloud_mod.requests, "post",
+        lambda *a, **k: FakeResponse(status_code=200, lines=lines),
+    )
+    events = _collect_events(backend)
+    assert [e.type for e in events] == ["answer_delta", "answer_delta", "answer_done"]
+    # Same contract as the local backend: answer_done carries the full text.
+    assert events[-1].data == "Hello world"
+
+
+def test_query_stream_http_error_raises(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    monkeypatch.setattr(
+        cloud_mod.requests, "post",
+        lambda *a, **k: FakeResponse(status_code=401, text="unauthorized"),
+    )
+
+    async def _run():
+        async for _ in backend.query_stream("col", "q", doc_ids=["d1"]):
+            pass
+
+    with pytest.raises(CloudAPIError, match="401"):
+        asyncio.run(_run())
+
+
+def test_query_stream_connect_failure_raises_instead_of_hanging(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+
+    def fake_post(*a, **k):
+        raise cloud_mod.requests.ConnectionError("dns failure")
+
+    monkeypatch.setattr(cloud_mod.requests, "post", fake_post)
+
+    async def _run():
+        async for _ in backend.query_stream("col", "q", doc_ids=["d1"]):
+            pass
+
+    with pytest.raises(CloudAPIError, match="request failed"):
+        asyncio.run(_run())

tests/test_local_backend.py

@@ -141,3 +141,17 @@ def test_normalize_doc_ids():
 def test_normalize_doc_ids_rejects_empty_list():
     with pytest.raises(ValueError, match="cannot be empty"):
         LocalBackend._normalize_doc_ids([])
+
+
+# ── error taxonomy: missing docs raise DocumentNotFoundError ─────────────────
+
+def test_get_document_missing_raises(backend):
+    backend.get_or_create_collection("papers")
+    with pytest.raises(DocumentNotFoundError, match="ghost"):
+        backend.get_document("papers", "ghost")
+
+
+def test_delete_document_missing_raises(backend):
+    backend.get_or_create_collection("papers")
+    with pytest.raises(DocumentNotFoundError, match="ghost"):
+        backend.delete_document("papers", "ghost")