fix(sdk): follow batch pagination and stop on cancel in both SDKs

sdks/python/src/crw/client.py

@@ -8,7 +8,7 @@
 import subprocess
 import time
 from typing import Any, cast
-from urllib.parse import quote, urlencode
+from urllib.parse import parse_qs, quote, urlencode, urlsplit
 
 from crw._binary import ensure_binary
 from crw.exceptions import CrwApiError, CrwError, CrwExtractCancelledError, CrwTimeoutError
@@ -47,6 +47,20 @@ def _next_id() -> int:
     return _REQUEST_ID
 
 
+def _page_path(base_path: str, next_url: str) -> str:
+    """Next page request, rebuilt from OUR base path + the cursor's skip/limit.
+
+    The server's `next` is an absolute URL built from its own public origin/prefix
+    (e.g. `https://host/api/v2/batch/scrape/{id}?skip=100`). Reusing that path
+    verbatim double-prefixes when the configured `api_url` already carries a path
+    (`.../api`), and trusts a host that may not match `api_url`. We only borrow the
+    cursor position (`skip`/`limit`) and re-issue against our own path.
+    """
+    q = parse_qs(urlsplit(next_url).query)
+    params = [f"{k}={q[k][0]}" for k in ("skip", "limit") if q.get(k)]
+    return base_path + ("?" + "&".join(params) if params else "")
+
+
 def _read_json_response(req: Any) -> dict:
     """Send a urllib request and parse the JSON body.
 
@@ -616,19 +630,49 @@ def batch_scrape(
             raise CrwError(f"Batch scrape did not return job ID: {start}")
 
         deadline = time.monotonic() + timeout
+        status_path = f"/v2/batch/scrape/{job_id}"
+
+        # Poll the first page until the job reaches a terminal state.
         while True:
             if time.monotonic() > deadline:
                 raise CrwTimeoutError(f"Batch scrape {job_id} timed out after {timeout}s")
             status_result = self._http_request(
-                "GET", f"/v2/batch/scrape/{job_id}", raw=True, check_success=False
+                "GET", status_path, raw=True, check_success=False
             )
             status = status_result.get("status")
             if status == "completed":
-                return status_result.get("data", [])
+                break
             if status == "failed":
                 raise CrwError(f"Batch scrape failed: {status_result.get('error', 'unknown')}")
+            if status == "cancelled":
+                # Terminal: a cancelled job never completes, so returning here
+                # instead of looping avoids hanging until `timeout`.
+                raise CrwError(f"Batch scrape {job_id} was cancelled")
             time.sleep(poll_interval)
 
+        # Page through the full result set. The batch status returns at most ~100
+        # documents per page and sets `next` to the cursor for the following page
+        # (null on the last). Returning only page 1 silently truncated any batch
+        # larger than one page.
+        # `.get("data") or []`, not `.get("data", [])`: a page may carry
+        # `"data": null`, and `list(None)` would raise instead of yielding [].
+        docs: list[dict] = list(status_result.get("data") or [])
+        next_url = status_result.get("next")
+        # Seed with page 1's path: a `next` that rebuilds to it (skip=0, or no
+        # cursor at all) is a repeat, not a new page — stop rather than re-append.
+        seen: set[str] = {status_path}
+        while next_url:
+            page_path = _page_path(status_path, next_url)
+            if page_path in seen:  # repeating/empty cursor; nothing new to fetch
+                break
+            if time.monotonic() > deadline:
+                raise CrwTimeoutError(f"Batch scrape {job_id} timed out after {timeout}s")
+            seen.add(page_path)
+            page = self._http_request("GET", page_path, raw=True, check_success=False)
+            docs.extend(page.get("data") or [])
+            next_url = page.get("next")
+        return docs
+
     def capabilities(self) -> dict:
         """Return what this engine instance supports (HTTP mode only).
 
@@ -756,6 +800,8 @@ def _poll_crawl(self, job_id: str, poll_interval: float, timeout: float) -> list
                 return result.get("data", [])
             if status == "failed":
                 raise CrwError(f"Crawl failed: {result.get('error', 'unknown')}")
+            if status == "cancelled":
+                raise CrwError(f"Crawl {job_id} was cancelled")
 
             time.sleep(poll_interval)
 
@@ -830,5 +876,7 @@ def _http_crawl(self, args: dict, poll_interval: float, timeout: float) -> list[
                 return status_result.get("data", [])
             if status == "failed":
                 raise CrwError(f"Crawl failed: {status_result.get('error', 'unknown')}")
+            if status == "cancelled":
+                raise CrwError(f"Crawl {job_id} was cancelled")
 
             time.sleep(poll_interval)

sdks/python/tests/test_client_unit.py

@@ -382,6 +382,51 @@ def test_extract_polls_until_complete(self) -> None:
         assert req.call_args_list[0][0][:2] == ("POST", "/v1/extract")
         assert req.call_args_list[0].kwargs["headers"] == {"Prefer": "respond-async"}
 
+    def test_batch_scrape_follows_pagination_to_completion(self) -> None:
+        # A batch larger than one page must return EVERY document, not just the
+        # first ~100 the completed status page carries. The SDK follows `next`.
+        client = CrwClient(api_url="https://fastcrw.com/api")
+        start = {"success": True, "id": "job-1"}
+        page1 = {
+            "success": True,
+            "status": "completed",
+            "data": [{"i": 0}],
+            "next": "https://fastcrw.com/api/v2/batch/scrape/job-1?skip=1",
+        }
+        page2 = {"success": True, "status": "completed", "data": [{"i": 1}], "next": None}
+        with patch.object(
+            client, "_http_request", side_effect=[start, page1, page2]
+        ) as req:
+            with patch("time.sleep"):
+                result = client.batch_scrape(["a", "b"])
+        assert result == [{"i": 0}, {"i": 1}]
+        # Page 2 fetched via the cursor's path+query, host stripped.
+        assert req.call_args_list[2][0][:2] == ("GET", "/v2/batch/scrape/job-1?skip=1")
+
+    def test_batch_scrape_tolerates_null_data_page(self) -> None:
+        # A page may serialize `"data": null`; `list(None)` would crash. The
+        # completed job here has no docs yet — must yield [], not raise.
+        client = CrwClient(api_url="https://fastcrw.com/api")
+        start = {"success": True, "id": "job-1"}
+        done = {"success": True, "status": "completed", "data": None, "next": None}
+        with patch.object(client, "_http_request", side_effect=[start, done]):
+            with patch("time.sleep"):
+                assert client.batch_scrape(["a"]) == []
+
+    def test_batch_scrape_raises_on_cancel_without_hanging(self) -> None:
+        # A cancelled job never reaches "completed"; the poll must stop, not spin
+        # until `timeout`.
+        client = CrwClient(api_url="https://fastcrw.com/api")
+        start = {"success": True, "id": "job-1"}
+        cancelled = {"success": True, "status": "cancelled"}
+        with patch.object(
+            client, "_http_request", side_effect=[start, cancelled]
+        ) as req:
+            with patch("time.sleep"):
+                with pytest.raises(CrwError, match="cancelled"):
+                    client.batch_scrape(["a"])
+        assert req.call_count == 2  # start + one status poll, no spinning
+
     def test_start_extract_prefer_managed_and_self_hosted_fixtures(self) -> None:
         accepted = {"id": "job-1", "status": "processing", "urls": 1}
         for client in (

sdks/typescript/src/client.ts

@@ -53,6 +53,26 @@ function httpOnlyHint(name: string, reason: string): string {
 
 const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
 
+/**
+ * Next page request, rebuilt from OUR base path + the cursor's skip/limit. The
+ * server's `next` is an absolute URL from its own public origin/prefix; reusing
+ * it verbatim double-prefixes when apiUrl carries a path (`.../api`) and trusts a
+ * host that may not match apiUrl. We only borrow the cursor position.
+ */
+function pageCursorPath(basePath: string, nextUrl: string): string {
+  let search = "";
+  try {
+    search = new URL(nextUrl, "http://placeholder.invalid").search;
+  } catch {
+    return basePath;
+  }
+  const q = new URLSearchParams(search);
+  const params = ["skip", "limit"]
+    .filter((k) => q.get(k) !== null)
+    .map((k) => `${k}=${q.get(k)}`);
+  return params.length ? `${basePath}?${params.join("&")}` : basePath;
+}
+
 export class CrwClient {
   private apiUrl: string | null;
   private apiKey: string | undefined;
@@ -318,13 +338,38 @@ export class CrwClient {
     const jobId = start.id as string | undefined;
     if (!jobId) throw new CrwError(`Batch scrape did not return job ID: ${JSON.stringify(start)}`);
     const deadline = Date.now() + timeout * 1000;
+    const statusPath = `/v2/batch/scrape/${jobId}`;
+
+    // Poll the first page until the job reaches a terminal state.
+    let status: Json;
     for (;;) {
       if (Date.now() > deadline) throw new CrwTimeoutError(`Batch scrape ${jobId} timed out after ${timeout}s`);
-      const status = await this.httpRequest("GET", `/v2/batch/scrape/${jobId}`, undefined, { raw: true, checkSuccess: false });
-      if (status.status === "completed") return (status.data as BatchResult) ?? [];
+      status = await this.httpRequest("GET", statusPath, undefined, { raw: true, checkSuccess: false });
+      if (status.status === "completed") break;
       if (status.status === "failed") throw new CrwError(`Batch scrape failed: ${status.error ?? "unknown"}`);
+      // A cancelled job never completes; stop instead of spinning until timeout.
+      if (status.status === "cancelled") throw new CrwError(`Batch scrape ${jobId} was cancelled`);
       await sleep(pollInterval * 1000);
     }
+
+    // Page through the full result set. The batch status returns at most ~100
+    // documents per page and sets `next` to the cursor for the next page (null on
+    // the last). Returning only page 1 silently truncated batches over one page.
+    const docs: BatchResult = [...((status.data as BatchResult) ?? [])];
+    let nextUrl = status.next as string | undefined;
+    // Seed with page 1's path: a `next` that rebuilds to it (skip=0, or no cursor
+    // at all) is a repeat, not a new page — stop rather than re-append.
+    const seen = new Set<string>([statusPath]);
+    while (nextUrl) {
+      const pagePath = pageCursorPath(statusPath, nextUrl);
+      if (seen.has(pagePath)) break; // repeating/empty cursor; nothing new
+      if (Date.now() > deadline) throw new CrwTimeoutError(`Batch scrape ${jobId} timed out after ${timeout}s`);
+      seen.add(pagePath);
+      const page = await this.httpRequest("GET", pagePath, undefined, { raw: true, checkSuccess: false });
+      docs.push(...((page.data as BatchResult) ?? []));
+      nextUrl = page.next as string | undefined;
+    }
+    return docs;
   }
 
   /** Feature-detect the engine (HTTP mode only). */
@@ -365,6 +410,7 @@ export class CrwClient {
       const result = await this.localTransport().toolCall("crw_check_crawl_status", { id: jobId });
       if (result.status === "completed") return (result.data as CrawlResult) ?? [];
       if (result.status === "failed") throw new CrwError(`Crawl failed: ${result.error ?? "unknown"}`);
+      if (result.status === "cancelled") throw new CrwError(`Crawl ${jobId} was cancelled`);
       await sleep(pollInterval * 1000);
     }
   }
@@ -439,6 +485,7 @@ export class CrwClient {
       const status = await this.httpRequest("GET", `/v1/crawl/${jobId}`, undefined, { raw: true });
       if (status.status === "completed") return (status.data as CrawlResult) ?? [];
       if (status.status === "failed") throw new CrwError(`Crawl failed: ${status.error ?? "unknown"}`);
+      if (status.status === "cancelled") throw new CrwError(`Crawl ${jobId} was cancelled`);
       await sleep(pollInterval * 1000);
     }
   }

sdks/typescript/test/client.test.ts

@@ -149,6 +149,46 @@ test("non-2xx body surfaces engine error as CrwApiError", async () => {
   await assert.rejects(() => c.scrape("https://example.com"), /boom/);
 });
 
+test("batchScrape follows the next cursor to return every page", async () => {
+  const responses: unknown[] = [
+    { success: true, id: "b1" }, // start
+    {
+      success: true,
+      status: "completed",
+      data: [{ markdown: "p1" }],
+      next: "https://fastcrw.com/api/v2/batch/scrape/b1?skip=1",
+    },
+    { success: true, status: "completed", data: [{ markdown: "p2" }], next: null },
+  ];
+  const calls: string[] = [];
+  globalThis.fetch = (async (url: string) => {
+    calls.push(String(url));
+    const body = responses.shift();
+    return { ok: true, status: 200, statusText: "OK", text: async () => JSON.stringify(body) } as Response;
+  }) as typeof fetch;
+  const c = new CrwClient({ apiUrl: "https://fastcrw.com/api" });
+  const docs = await c.batchScrape(["a", "b"]);
+  assert.deepEqual(docs, [{ markdown: "p1" }, { markdown: "p2" }]);
+  // Page 2 rebuilt from our base path + the cursor, no double `/api` prefix.
+  assert.equal(calls[2], "https://fastcrw.com/api/v2/batch/scrape/b1?skip=1");
+});
+
+test("batchScrape stops on a cancelled job instead of hanging", async () => {
+  const responses: unknown[] = [
+    { success: true, id: "b1" },
+    { success: true, status: "cancelled" },
+  ];
+  let n = 0;
+  globalThis.fetch = (async () => {
+    n++;
+    const body = responses.shift();
+    return { ok: true, status: 200, statusText: "OK", text: async () => JSON.stringify(body) } as Response;
+  }) as typeof fetch;
+  const c = new CrwClient({ apiUrl: "https://fastcrw.com/api" });
+  await assert.rejects(() => c.batchScrape(["a"]), /cancelled/);
+  assert.equal(n, 2); // start + one poll, no spinning
+});
+
 test("extract starts a /v1/extract job and returns per-URL results", async () => {
   let n = 0;
   const calls: Array<{ url: string; init?: RequestInit }> = [];