diff --git a/pageindex/__init__.py b/pageindex/__init__.py
index e95bce1..0670332 100644
--- a/pageindex/__init__.py
+++ b/pageindex/__init__.py
@@ -64,6 +64,11 @@ __all__ = [
     # Legacy top-level exports (pre-SDK API), kept so `from pageindex import *`
     # still binds them.
     "page_index",
+    "page_index_main",
+    "tree_parser",
+    "ConfigLoader",
+    "llm_completion",
+    "llm_acompletion",
     "md_to_tree",
     "get_document",
     "get_document_structure",
diff --git a/pageindex/backend/cloud.py b/pageindex/backend/cloud.py
index 09f095b..6d2b6f2 100644
--- a/pageindex/backend/cloud.py
+++ b/pageindex/backend/cloud.py
@@ -14,7 +14,8 @@ import requests
 from typing import AsyncIterator
 
 from ..cloud_api import API_BASE  # single source of truth for the cloud base URL
-from ..errors import AUTH_HINT, CloudAPIError, DocumentNotFoundError, PageIndexError
+from ..errors import (AUTH_HINT, CloudAPIError, CollectionNotFoundError,
+                      DocumentNotFoundError, PageIndexError)
 from ..events import QueryEvent
 
 logger = logging.getLogger(__name__)
@@ -30,8 +31,9 @@ def _as_int(value):
 
 
 class CloudBackend:
-    def __init__(self, api_key: str):
+    def __init__(self, api_key: str, base_url: str | None = None):
         self._api_key = api_key
+        self._base_url = base_url or API_BASE
         self._headers = {"api_key": api_key}
         self._folder_id_cache: dict[str, str | None] = {}
         self._folder_warning_shown = False
@@ -59,7 +61,7 @@ class CloudBackend:
         """HTTP helper. ``retries`` caps total attempts — pass 1 for
         non-idempotent, expensive calls (e.g. chat completions) where a
         retry would redo the full server-side work."""
-        url = f"{API_BASE}{path}"
+        url = f"{self._base_url}{path}"
         kwargs.setdefault("timeout", 30)
         last_status: int | None = None
         for attempt in range(retries):
@@ -150,7 +152,9 @@ class CloudBackend:
                 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 are
+        not available on this plan; raises CollectionNotFoundError if folders
+        are available but no folder has this name.
 
         Only "folders unavailable on this plan" (403/404) is cached as None —
         transient errors (network, 5xx) propagate so a blip can't silently
@@ -170,7 +174,10 @@ class CloudBackend:
             if folder.get("name") == name:
                 self._folder_id_cache[name] = folder["id"]
                 return folder["id"]
-        return None
+        raise CollectionNotFoundError(
+            f"Collection '{name}' does not exist; "
+            f"create it first (e.g. client.collection('{name}'))."
+        )
 
     def list_collections(self) -> list[str]:
         try:
@@ -183,7 +190,10 @@ class CloudBackend:
         return [f["name"] for f in data.get("folders", []) or []]
 
     def delete_collection(self, name: str) -> None:
-        folder_id = self._get_folder_id(name)
+        try:
+            folder_id = self._get_folder_id(name)
+        except CollectionNotFoundError:
+            return  # already gone — delete is idempotent
         if folder_id:
             self._request("DELETE", f"/folder/{self._enc(folder_id)}/")
             # Drop the cached id so a later same-name op re-resolves instead of
@@ -382,10 +392,13 @@ class CloudBackend:
             raise ValueError(
                 "doc_ids cannot be empty; pass None to query the whole collection"
             )
-        doc_id = doc_ids if doc_ids else self._get_all_doc_ids(collection)
+        doc_id = doc_ids if doc_ids else await asyncio.to_thread(
+            self._get_all_doc_ids, collection
+        )
         if not doc_id:
             raise ValueError("collection has no documents to query")
         headers = self._headers
+        base_url = self._base_url
         # Queue carries QueryEvent, an Exception to re-raise, or None (end).
         queue: asyncio.Queue[QueryEvent | Exception | None] = asyncio.Queue()
         loop = asyncio.get_running_loop()
@@ -413,7 +426,7 @@ class CloudBackend:
             answer_parts: list[str] = []
             try:
                 resp = requests.post(
-                    f"{API_BASE}/chat/completions/",
+                    f"{base_url}/chat/completions/",
                     headers=headers,
                     json={
                         "messages": [{"role": "user", "content": question}],
@@ -521,7 +534,11 @@ class CloudBackend:
                     # during teardown; closing is just to unblock the thread.
                     logger.debug("Ignoring error closing streaming response during cleanup",
                                  exc_info=True)
-            thread.join(timeout=5)
+            try:
+                await asyncio.to_thread(thread.join, 5)
+            except RuntimeError:
+                # event loop already shutting down — fall back to a bounded sync join
+                thread.join(timeout=5)
 
     def _get_all_doc_ids(self, collection: str) -> list[str]:
         """Get all document IDs in a collection."""
diff --git a/pageindex/backend/local.py b/pageindex/backend/local.py
index feb695c..1cbd47a 100644
--- a/pageindex/backend/local.py
+++ b/pageindex/backend/local.py
@@ -137,7 +137,7 @@ class LocalBackend:
                 "structure": result["structure"],
                 "pages": pages,
             })
-        except sqlite3.IntegrityError:
+        except sqlite3.IntegrityError as e:
             # Lost a concurrent add of the same content (UNIQUE collection+hash).
             # Discard our managed files and return the winner's doc_id.
             managed_path.unlink(missing_ok=True)
@@ -147,7 +147,12 @@ class LocalBackend:
             existing_id = self._storage.find_document_by_hash(collection, file_hash)
             if existing_id:
                 return existing_id
-            raise
+            # No winner — likely an FK violation from a concurrent collection delete.
+            if collection not in self._storage.list_collections():
+                raise CollectionNotFoundError(
+                    f"Collection '{collection}' was deleted while indexing {file_path}"
+                ) from e
+            raise IndexingError(f"Failed to index {file_path}: {e}") from e
         except Exception as e:
             managed_path.unlink(missing_ok=True)
             doc_dir = col_dir / doc_id
diff --git a/pageindex/client.py b/pageindex/client.py
index de0dd83..1f8ea5b 100644
--- a/pageindex/client.py
+++ b/pageindex/client.py
@@ -74,7 +74,7 @@ class PageIndexClient:
     def _init_cloud(self, api_key: str):
         from .backend.cloud import CloudBackend
         from .cloud_api import LegacyCloudAPI
-        self._backend = CloudBackend(api_key=api_key)
+        self._backend = CloudBackend(api_key=api_key, base_url=self.BASE_URL)
         self._legacy_cloud_api = LegacyCloudAPI(api_key=api_key, base_url=self.BASE_URL)
 
     def _init_local(self, model: str = None, retrieve_model: str = None,
diff --git a/pageindex/cloud_api.py b/pageindex/cloud_api.py
index ef31746..109141e 100644
--- a/pageindex/cloud_api.py
+++ b/pageindex/cloud_api.py
@@ -166,12 +166,15 @@ class LegacyCloudAPI:
         if stream_metadata:
             payload["stream_metadata"] = stream_metadata
 
+        # Non-streaming completions return no bytes until server-side
+        # generation finishes — far longer than the default 30s read timeout.
         response = self._request(
             "POST",
             "/chat/completions/",
             "Failed to get chat completion",
             json=payload,
             stream=stream,
+            **({} if stream else {"timeout": 300}),
         )
 
         if stream:
diff --git a/pageindex/config.py b/pageindex/config.py
index 92de8dd..4fe08b8 100644
--- a/pageindex/config.py
+++ b/pageindex/config.py
@@ -28,7 +28,8 @@ class IndexConfig(BaseModel):
     if_add_node_text: bool = False
     # Max concurrent in-flight LLM calls during indexing. None = use the global
     # default (get_max_concurrency(), overridable via PAGEINDEX_MAX_CONCURRENCY).
-    # An explicit value here wins for this client.
+    # An explicit value can only lower the cap; raise the ceiling itself via
+    # set_max_concurrency() or PAGEINDEX_MAX_CONCURRENCY.
     max_concurrency: int | None = None
     # Per-call litellm completion kwargs for this client's indexing calls only
     # (e.g. {"temperature": 1}). None = use the process-wide defaults
@@ -217,6 +218,16 @@ def max_concurrency_scope(value: int | None):
     """
     if value is not None:
         _validate_max_concurrency(value)
+        if value > _MAX_CONCURRENCY:
+            import warnings
+            warnings.warn(
+                f"max_concurrency={value} exceeds the process-wide ceiling "
+                f"({_MAX_CONCURRENCY}), which still applies — a per-index value "
+                f"can only lower the cap. Raise the ceiling with "
+                f"set_max_concurrency({value}) or PAGEINDEX_MAX_CONCURRENCY.",
+                UserWarning,
+                stacklevel=3,
+            )
     scoped_sem = threading.Semaphore(value) if value is not None else None
     token = _MAX_CONCURRENCY_OVERRIDE.set(value)
     sem_token = _MAX_CONCURRENCY_SCOPE_SEMAPHORE.set(scoped_sem)
diff --git a/pageindex/index/page_index.py b/pageindex/index/page_index.py
index 1cba5f2..17cc7e4 100644
--- a/pageindex/index/page_index.py
+++ b/pageindex/index/page_index.py
@@ -17,6 +17,9 @@ async def check_title_appearance(item, page_list, start_index=1, model=None):
     
     
     page_number = item['physical_index']
+    if page_number < start_index:
+        # a below-range index would wrap to a negative Python index (the last page)
+        return {'list_index': item.get('list_index'), 'answer': 'no', 'title': title, 'page_number': None}
     page_text = page_list[page_number-start_index][0]
 
     
@@ -1085,6 +1088,13 @@ async def tree_parser(page_list, opt, doc=None, logger=None):
 
 
 def page_index_main(doc, opt=None):
+    # accept legacy 'yes'/'no' string flags (a bare 'no' is truthy)
+    from .page_index_md import _coerce_bool
+    for flag in ('if_add_node_id', 'if_add_node_text',
+                 'if_add_node_summary', 'if_add_doc_description'):
+        if hasattr(opt, flag):
+            setattr(opt, flag, _coerce_bool(getattr(opt, flag)))
+
     logger = JsonLogger(doc)
     
     is_valid_pdf = (
diff --git a/pageindex/utils.py b/pageindex/utils.py
index fe403d2..ff6d890 100644
--- a/pageindex/utils.py
+++ b/pageindex/utils.py
@@ -12,3 +12,6 @@ warnings.warn(
 )
 
 from .index.utils import *  # noqa: F401,F403,E402
+# Legacy 0.2.x alias. index.utils keeps it private (_config) so its star-export
+# can't shadow the pageindex.config submodule; re-expose it only in this shim.
+from types import SimpleNamespace as config  # noqa: E402,F401
diff --git a/tests/test_cloud_backend.py b/tests/test_cloud_backend.py
index 6154473..1fb0cfa 100644
--- a/tests/test_cloud_backend.py
+++ b/tests/test_cloud_backend.py
@@ -7,7 +7,7 @@ import pytest
 
 import pageindex.backend.cloud as cloud_mod
 from pageindex.backend.cloud import CloudBackend, API_BASE
-from pageindex.errors import CloudAPIError, DocumentNotFoundError
+from pageindex.errors import CloudAPIError, CollectionNotFoundError, DocumentNotFoundError
 
 # Real sleep captured at import, before the _no_sleep autouse fixture patches
 # time.sleep on the shared module object. Tests that need genuine pacing (e.g.
@@ -100,6 +100,47 @@ def test_list_documents_paginates(monkeypatch):
     assert docs[-1]["doc_id"] == "d129"
 
 
+# ── base_url override must reach every request path ──────────────────────────
+
+def test_base_url_override_routes_requests(monkeypatch):
+    backend = CloudBackend(api_key="pi-test", base_url="https://staging.example.com")
+    urls = []
+
+    def fake_request(method, url, headers=None, **kwargs):
+        urls.append(url)
+        return FakeResponse(status_code=200, json_data={"folders": []})
+
+    monkeypatch.setattr(cloud_mod.requests, "request", fake_request)
+    backend.list_collections()
+    assert urls == ["https://staging.example.com/folders/"]
+
+
+def test_base_url_override_routes_streaming(monkeypatch):
+    backend = CloudBackend(api_key="pi-test", base_url="https://staging.example.com")
+    urls = []
+
+    def fake_post(url, **kwargs):
+        urls.append(url)
+        return FakeResponse(status_code=200, lines=["data: [DONE]"])
+
+    monkeypatch.setattr(cloud_mod.requests, "post", fake_post)
+    _collect_events(backend)
+    assert urls == ["https://staging.example.com/chat/completions/"]
+
+
+def test_client_base_url_override_reaches_both_backends():
+    # PageIndexClient.BASE_URL is the documented override point; it must apply
+    # to the Collection API (CloudBackend), not just the legacy SDK methods.
+    from pageindex.client import PageIndexClient
+
+    class StagingClient(PageIndexClient):
+        BASE_URL = "https://staging.example.com"
+
+    client = StagingClient(api_key="pi-test")
+    assert client._backend._base_url == "https://staging.example.com"
+    assert client._legacy_cloud_api.base_url == "https://staging.example.com"
+
+
 # ── folder resolution: plan-limit vs transient errors ────────────────────────
 
 def test_folder_unavailable_warns_and_caches(monkeypatch):
@@ -127,6 +168,42 @@ def test_folder_transient_error_propagates_and_is_not_cached(monkeypatch):
     assert "col" not in backend._folder_id_cache
 
 
+def test_missing_folder_raises_collection_not_found(monkeypatch):
+    # Folders ARE available but the name has no match (e.g. deleted): must
+    # raise, never fall back to the account-wide global space.
+    backend = CloudBackend(api_key="pi-test")
+    monkeypatch.setattr(
+        backend, "_request",
+        lambda *a, **k: {"folders": [{"name": "other", "id": "f1"}]},
+    )
+    with pytest.raises(CollectionNotFoundError):
+        backend._get_folder_id("gone")
+    assert "gone" not in backend._folder_id_cache
+
+
+def test_list_documents_raises_for_missing_collection(monkeypatch):
+    # Guards the query path: a stale collection must error out, not silently
+    # list (and query over) every document in the account.
+    backend = CloudBackend(api_key="pi-test")
+    monkeypatch.setattr(backend, "_request", lambda *a, **k: {"folders": []})
+    with pytest.raises(CollectionNotFoundError):
+        backend.list_documents("gone")
+
+
+def test_delete_collection_missing_is_noop(monkeypatch):
+    backend = CloudBackend(api_key="pi-test")
+    calls = []
+
+    def fake_request(method, path, **kwargs):
+        calls.append((method, path))
+        return {"folders": []}
+
+    monkeypatch.setattr(backend, "_request", fake_request)
+    backend.delete_collection("gone")  # idempotent, matching the local backend
+    assert ("GET", "/folders/") in calls
+    assert not any(method == "DELETE" for method, _ in calls)
+
+
 def test_list_collections_degrades_when_folders_unavailable(monkeypatch):
     backend = CloudBackend(api_key="pi-test")
 
diff --git a/tests/test_review_fixes_3.py b/tests/test_review_fixes_3.py
new file mode 100644
index 0000000..5ca40e8
--- /dev/null
+++ b/tests/test_review_fixes_3.py
@@ -0,0 +1,215 @@
+"""Regression tests for the PR #272 review findings #9-#15."""
+import asyncio
+import sqlite3
+import threading
+import warnings
+
+import pytest
+
+from pageindex.errors import CollectionNotFoundError, IndexingError
+
+
+# ── #9: below-range physical_index must not wrap to the last page ────────────
+
+def test_check_title_appearance_rejects_below_range_index(monkeypatch):
+    from pageindex.index import page_index as pi
+
+    async def _fail(*a, **k):
+        raise AssertionError("LLM must not be called for a below-range index")
+
+    monkeypatch.setattr(pi, "llm_acompletion", _fail)
+    item = {"title": "Intro", "physical_index": 0, "list_index": 3}
+    # A single-page page_list: without the guard, page_list[0-1] silently
+    # reads this (last) page instead of erroring.
+    result = asyncio.run(
+        pi.check_title_appearance(item, [("last page text", 10)], start_index=1)
+    )
+    assert result["answer"] == "no"
+    assert result["page_number"] is None
+
+
+# ── #6: page_index_main must coerce legacy 'yes'/'no' string flags ───────────
+
+def test_page_index_main_coerces_legacy_string_flags():
+    from types import SimpleNamespace
+    from pageindex.index.page_index import page_index_main
+
+    opt = SimpleNamespace(model=None, if_add_node_id='yes', if_add_node_text='no',
+                          if_add_node_summary='no', if_add_doc_description='no')
+    # Coercion runs before input validation, which rejects the non-PDF path.
+    with pytest.raises(ValueError, match="Unsupported input type"):
+        page_index_main('not-a-pdf.txt', opt)
+    assert opt.if_add_node_id is True
+    assert opt.if_add_node_text is False
+    assert opt.if_add_node_summary is False
+    assert opt.if_add_doc_description is False
+
+
+# ── #10: per-index max_concurrency above the ceiling must warn ────────────────
+
+def test_max_concurrency_scope_warns_above_ceiling():
+    from pageindex.config import (
+        max_concurrency_scope,
+        set_max_concurrency,
+        _process_wide_max_concurrency,
+    )
+
+    original = _process_wide_max_concurrency()
+    set_max_concurrency(5)
+    try:
+        with pytest.warns(UserWarning, match="exceeds the process-wide ceiling"):
+            with max_concurrency_scope(20):
+                pass
+        # A narrowing value is the supported use and must stay silent.
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")
+            with max_concurrency_scope(3):
+                pass
+    finally:
+        set_max_concurrency(original)
+
+
+# ── #11: non-streaming legacy chat_completions needs a long read timeout ─────
+
+def test_legacy_chat_completions_nonstream_timeout(monkeypatch):
+    from pageindex import cloud_api as ca
+
+    captured = {}
+
+    class FakeResp:
+        status_code = 200
+        text = ""
+
+        def json(self):
+            return {"choices": []}
+
+    def fake_request(method, url, headers=None, **kwargs):
+        captured.clear()
+        captured.update(kwargs)
+        return FakeResp()
+
+    monkeypatch.setattr(ca.requests, "request", fake_request)
+    api = ca.LegacyCloudAPI(api_key="pi-test")
+
+    api.chat_completions(messages=[{"role": "user", "content": "q"}], stream=False)
+    assert captured["timeout"] == 300
+
+    api.chat_completions(messages=[{"role": "user", "content": "q"}], stream=True)
+    assert captured["timeout"] == 120  # between-chunks timeout, unchanged
+
+
+# ── #12: query_stream must not run the doc-id listing on the loop thread ─────
+
+def test_query_stream_lists_docs_off_event_loop(monkeypatch):
+    import pageindex.backend.cloud as cloud_mod
+    from pageindex.backend.cloud import CloudBackend
+
+    backend = CloudBackend(api_key="pi-test")
+    seen = {}
+
+    def fake_get_all(collection):
+        seen["thread"] = threading.current_thread()
+        return ["d1"]
+
+    monkeypatch.setattr(backend, "_get_all_doc_ids", fake_get_all)
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+        def iter_lines(self, decode_unicode=True):
+            yield "data: [DONE]"
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(cloud_mod.requests, "post", lambda *a, **k: FakeResponse())
+
+    async def _run():
+        async for _ in backend.query_stream("col", "q"):  # doc_ids=None → list all
+            pass
+        return threading.current_thread()
+
+    loop_thread = asyncio.run(_run())
+    assert seen["thread"] is not loop_thread
+
+
+# ── #13: an unexplained IntegrityError must surface as a PageIndexError ──────
+
+class _RaceStorage:
+    """Collection exists at the fail-fast check, then save trips the FK."""
+
+    def __init__(self, collections_after_start):
+        self._after = collections_after_start
+        self._calls = 0
+
+    def list_collections(self):
+        self._calls += 1
+        return ["col"] if self._calls == 1 else self._after
+
+    def find_document_by_hash(self, collection, file_hash):
+        return None
+
+    def save_document(self, *a, **k):
+        raise sqlite3.IntegrityError("FOREIGN KEY constraint failed")
+
+
+def _make_backend(tmp_path, monkeypatch, storage):
+    import pageindex.backend.local as local_mod
+    from pageindex.backend.local import LocalBackend
+
+    backend = LocalBackend(storage=storage, files_dir=str(tmp_path / "files"))
+
+    class FakeParsed:
+        doc_name = "doc"
+        nodes = []
+
+    class FakeParser:
+        def parse(self, path, model=None, images_dir=None):
+            return FakeParsed()
+
+    monkeypatch.setattr(backend, "_resolve_parser", lambda p: FakeParser())
+    monkeypatch.setattr(
+        local_mod, "build_index", lambda parsed, model=None, opt=None: {"structure": []}
+    )
+    pdf = tmp_path / "doc.pdf"
+    pdf.write_bytes(b"%PDF-1.4 fake")
+    return backend, str(pdf)
+
+
+def test_concurrent_collection_delete_raises_collection_not_found(tmp_path, monkeypatch):
+    backend, pdf = _make_backend(tmp_path, monkeypatch, _RaceStorage([]))
+    with pytest.raises(CollectionNotFoundError):
+        backend.add_document("col", pdf)
+
+
+def test_unexplained_integrity_error_wrapped_as_indexing_error(tmp_path, monkeypatch):
+    backend, pdf = _make_backend(tmp_path, monkeypatch, _RaceStorage(["col"]))
+    with pytest.raises(IndexingError):
+        backend.add_document("col", pdf)
+
+
+# ── #14: legacy `from pageindex.utils import config` must keep working ───────
+
+def test_legacy_config_alias_importable():
+    from types import SimpleNamespace
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")  # module-level deprecation shim warning
+        from pageindex.utils import config
+    assert config is SimpleNamespace
+
+
+# ── #15: star import must bind the legacy pre-SDK names ──────────────────────
+
+def test_star_import_binds_legacy_names():
+    ns = {}
+    exec("from pageindex import *", ns)
+    for name in (
+        "page_index",
+        "page_index_main",
+        "tree_parser",
+        "ConfigLoader",
+        "llm_completion",
+        "llm_acompletion",
+    ):
+        assert name in ns, f"{name} missing from star import"