From 284ce005f55f278962f406dd806e9be498726cce Mon Sep 17 00:00:00 2001 From: mountain Date: Tue, 7 Jul 2026 09:40:09 +0800 Subject: [PATCH] fix(cloud): align cloud/local backend contracts and harden the HTTP layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 221 ++++++++++++++++++++++++------------ pageindex/backend/local.py | 8 +- pageindex/errors.py | 12 +- tests/test_cloud_backend.py | 185 ++++++++++++++++++++++++++++++ tests/test_local_backend.py | 14 +++ 5 files changed, 366 insertions(+), 74 deletions(-) diff --git a/pageindex/backend/cloud.py b/pageindex/backend/cloud.py index 144728d..b479e81 100644 --- a/pageindex/backend/cloud.py +++ b/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( - "Folders (collections) require a Max plan. " + import warnings + 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", []): - if folder.get("name") == name: - self._folder_id_cache[name] = folder["id"] - return folder["id"] - except CloudAPIError: - pass + except CloudAPIError as e: + if e.status_code in self._FOLDER_UNAVAILABLE: + self._warn_folder_upgrade() + self._folder_id_cache[name] = None + return None + 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)}/", - params={"type": "tree", "summary": "true"}) + tree_resp = self._doc_request(doc_id, "GET", f"/doc/{self._enc(doc_id)}/", + 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} - if folder_id: - params["folder_id"] = folder_id - data = self._request("GET", "/docs/", params=params) - return [ - { - "doc_id": d.get("id", ""), - "doc_name": d.get("name", ""), - "doc_description": d.get("description", ""), - "doc_type": "pdf", - } - for d in data.get("documents", []) - ] + # The API caps `limit` at 100; paginate with `offset` until a short + # page comes back so collections with >100 docs aren't silently + # truncated (queries over the whole collection rely on this list). + page_size = 100 + offset = 0 + docs: list[dict] = [] + while True: + params = {"limit": page_size, "offset": offset} + if folder_id: + params["folder_id"] = folder_id + data = self._request("GET", "/docs/", params=params) + 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() - loop = asyncio.get_event_loop() + # Queue carries QueryEvent, an Exception to re-raise, or None (end). + 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.""" - 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, - ) + """Background thread: read SSE and push events to queue. + + Everything — including the initial connect — runs inside try so a + failure can never die silently and leave the consumer awaiting a + sentinel that never arrives. Errors are forwarded as exceptions + (raised in the consumer), never disguised as answer events. + """ + resp = None + answer_parts: list[str] = [] 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( - queue.put_nowait, - QueryEvent(type="answer_done", - data=f"Cloud streaming error {resp.status_code}: {body}"), + raise CloudAPIError( + f"Cloud streaming error {resp.status_code}: {body}", + status_code=resp.status_code, ) - 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( - queue.put_nowait, - QueryEvent(type="tool_call", data={ - "name": current_tool_name, - "args": args_str, - }), - ) + _put(QueryEvent(type="tool_call", data={ + "name": current_tool_name, + "args": args_str, + })) current_tool_name = None current_tool_args = [] elif block_type == "text" and content: - loop.call_soon_threadsafe( - queue.put_nowait, - QueryEvent(type="answer_delta", data=content), - ) + answer_parts.append(content) + _put(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() - loop.call_soon_threadsafe(queue.put_nowait, None) # sentinel + if resp is not None: + resp.close() + _put(None) # sentinel thread = threading.Thread(target=_stream, daemon=True) thread.start() while True: - event = await queue.get() - if event is None: + item = await queue.get() + if item is None: break - yield event + if isinstance(item, Exception): + raise item + yield item thread.join(timeout=5) diff --git a/pageindex/backend/local.py b/pageindex/backend/local.py index 811b778..cfbd288 100644 --- a/pageindex/backend/local.py +++ b/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 diff --git a/pageindex/errors.py b/pageindex/errors.py index 045a9db..dc4b1c7 100644 --- a/pageindex/errors.py +++ b/pageindex/errors.py @@ -27,8 +27,16 @@ class PageIndexAPIError(PageIndexError): class CloudAPIError(PageIndexAPIError): - """Cloud API returned error.""" - pass + """Cloud API returned error. + + ``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): diff --git a/tests/test_cloud_backend.py b/tests/test_cloud_backend.py index cdaa4eb..9bcdfba 100644 --- a/tests/test_cloud_backend.py +++ b/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()) diff --git a/tests/test_local_backend.py b/tests/test_local_backend.py index 5854388..0155cdb 100644 --- a/tests/test_local_backend.py +++ b/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")