diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0930ab3..e0fee9f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,24 @@
 # Bright Data Python SDK Changelog
 
+## Version 2.6.0 - Structured errors
+
+- **Every non-2xx is now a typed exception.** `AsyncEngine` previously converted only 401/403 and handed every other failure back as a normal response, leaving each subsystem to improvise. Classification now happens once, at the single point every request passes through: `429` → the new **`RateLimitError`** (with `retry_after` parsed from the header), other 4xx/5xx → `APIError`.
+- **Exceptions carry data, not just prose.** `BrightDataError` gained `status_code`, `url`, `method`, `retry_after`, `retryable` and `raw` (the response body, bounded to 4 KB). Callers can branch on the failure instead of parsing its message.
+- **Fixed**: a rate-limited dataset request surfaced as an aiohttp *content-type* error, because the 429 body is a bare string and the response was parsed as JSON before its status was checked. It now raises `RateLimitError`.
+- **Fixed**: a token expiring mid-poll was reported as `"Job failed with status: error"` — blaming the user's scrape for an authentication problem. `DatasetAPIClient.get_status` no longer collapses every non-200 into the status string `"error"`.
+- **Results carry the failure too.** `ScrapeResult` / `CrawlResult` gained `cause`, the originating exception, populated wherever one is converted into a result. `error` remains the human-readable message:
+  ```python
+  result = await client.scrape.x.posts(url)
+  if not result.success and isinstance(result.cause, RateLimitError):
+      await asyncio.sleep(result.cause.retry_after or 60)
+  ```
+- **Retry is now opt-in.** `retry_with_backoff` no longer retries by exception type. An error with no status code is raised locally — sometimes *after* the server accepted the work — so repeating it could create a duplicate billed job; those are never retried. Explicit 5xx still is. **429 never is**, because those responses consume quota and retrying extends the lockout.
+- `DatasetError` now subclasses `BrightDataError` (still catchable as before). `RateLimitError` and `DataNotReadyError` are exported from `brightdata` top level.
+- **Note on messages**: error messages are shorter and more uniform, with detail moved to attributes. Code matching on message *text* may need updating; use `status_code` instead.
+- **Known limit**: this is status-based, so it cannot see HTTP 200 responses carrying an error in the body (SERP's inner envelope, Web Unlocker error content). Those remain string-shaped.
+
+---
+
 ## Version 2.4.0 - Sync parity, colorless job verbs, dataset error reporting
 
 - **Sync client parity**: `SyncBrightDataClient` now mirrors the async surface. Added `client.datasets` (fixes the `SyncBrightDataClient` `datasets` `AttributeError`), the 5 missing scrapers (`scrape.tiktok` / `youtube` / `reddit` / `perplexity` / `digikey`), the 2 missing search verticals (`search.tiktok` / `youtube`), Pinterest trigger/status/fetch, and Instagram-search `profiles` / `reels_all`.
diff --git a/README.md b/README.md
index 0c45c75..5bbe837 100644
--- a/README.md
+++ b/README.md
@@ -22,6 +22,34 @@ export BRIGHTDATA_API_TOKEN="your_api_token_here"
 **Already logged in with the CLI?** The SDK works with no configuration — it automatically
 falls back to the credentials stored by `brightdata login`.
 
+## Handling errors
+
+Failures carry structured data, not just a message:
+
+```python
+from brightdata import BrightDataClient, RateLimitError, APIError
+
+async with BrightDataClient() as client:
+    try:
+        data = await client.datasets.instagram_profiles.download(snapshot_id)
+    except RateLimitError as e:
+        await asyncio.sleep(e.retry_after or 60)   # the API told us how long
+    except APIError as e:
+        print(e.status_code, e.raw)                # not a message to parse
+```
+
+Methods that return a result instead of raising expose the same information on `cause`:
+
+```python
+result = await client.scrape.x.posts(url)
+if not result.success and isinstance(result.cause, RateLimitError):
+    await asyncio.sleep(result.cause.retry_after or 60)
+```
+
+Use `e.retryable` to decide whether repeating the call is safe. Rate limits are never
+retryable — a 429 response itself consumes quota, so retrying extends the lockout; wait
+`retry_after` instead.
+
 ## Quick Start
 
 This SDK is **async-native**. A sync client is also available (see [Sync Client](#sync-client)).
diff --git a/pyproject.toml b/pyproject.toml
index 0583e5b..0aa5a36 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,7 +7,7 @@ where = ["src"]
 
 [project]
 name = "brightdata-sdk"
-version = "2.5.0"
+version = "2.6.0"
 description = "Modern async-first Python SDK for Bright Data APIs"
 authors = [{name = "Bright Data", email = "support@brightdata.com"}]
 license = {text = "MIT"}
diff --git a/src/brightdata/__init__.py b/src/brightdata/__init__.py
index 83ae8c0..d09a7f8 100644
--- a/src/brightdata/__init__.py
+++ b/src/brightdata/__init__.py
@@ -69,6 +69,8 @@
     ValidationError,
     AuthenticationError,
     APIError,
+    RateLimitError,
+    DataNotReadyError,
     ZoneError,
     NetworkError,
     SSLError,
@@ -127,6 +129,8 @@
     "ValidationError",
     "AuthenticationError",
     "APIError",
+    "RateLimitError",
+    "DataNotReadyError",
     "ZoneError",
     "NetworkError",
     "SSLError",
diff --git a/src/brightdata/core/engine.py b/src/brightdata/core/engine.py
index 32490fa..4a44ebe 100644
--- a/src/brightdata/core/engine.py
+++ b/src/brightdata/core/engine.py
@@ -6,9 +6,16 @@
 import warnings
 from typing import Optional, Dict, Any
 from .. import __version__
-from ..exceptions import AuthenticationError, NetworkError, SSLError
+from ..exceptions import (
+    APIError,
+    AuthenticationError,
+    NetworkError,
+    RateLimitError,
+    SSLError,
+)
 from http import HTTPStatus
 from ..utils.ssl_helpers import is_ssl_certificate_error, get_ssl_error_message
+from ..utils.http import parse_retry_after, status_phrase
 
 # Rate limiting support
 try:
@@ -409,19 +416,43 @@ async def __aenter__(self):
                         headers=self._headers,
                         timeout=self._timeout,
                     )
-                    # Check status codes that should raise exceptions
-                    if self._response.status == HTTPStatus.UNAUTHORIZED:
-                        text = await self._response.text()
-                        await self._response.release()
-                        raise AuthenticationError(
-                            f"Unauthorized ({HTTPStatus.UNAUTHORIZED}): {text}"
+                    status = self._response.status
+
+                    # 202 MUST pass through. It is a success for
+                    # scraper_studio.trigger_immediate, and elsewhere it means
+                    # "ready but still building", which fetch_result turns into
+                    # DataNotReadyError -- the SDK's only recovery path.
+                    if status < 400 or status == HTTPStatus.ACCEPTED:
+                        return self._response
+
+                    text = await self._response.text()
+                    await self._response.release()
+
+                    context = {
+                        "status_code": status,
+                        "url": self._url,
+                        "method": self._method,
+                        "raw": text,
+                    }
+                    # A short body excerpt keeps a bare print(exc) useful; the
+                    # full (bounded) body stays on .raw.
+                    detail = " ".join(text.split())[:200]
+                    suffix = f": {detail}" if detail else ""
+
+                    if status in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN):
+                        raise AuthenticationError(f"{status_phrase(status)} ({status})", **context)
+
+                    if status == HTTPStatus.TOO_MANY_REQUESTS:
+                        # Never retryable: a 429 response itself consumes quota,
+                        # so retrying extends the lockout instead of waiting it out.
+                        raise RateLimitError(
+                            f"Rate limited ({status}){suffix}",
+                            retry_after=parse_retry_after(self._response.headers),
+                            **context,
                         )
-                    elif self._response.status == HTTPStatus.FORBIDDEN:
-                        text = await self._response.text()
-                        await self._response.release()
-                        raise AuthenticationError(f"Forbidden ({HTTPStatus.FORBIDDEN}): {text}")
 
-                    return self._response
+                    # retryable resolves from status_code: 5xx yes, 4xx no.
+                    raise APIError(f"Request failed (HTTP {status}){suffix}", **context)
                 except asyncio.TimeoutError as e:
                     # Must be caught before OSError — on Python 3.11+,
                     # TimeoutError is a subclass of OSError
diff --git a/src/brightdata/crawler/models.py b/src/brightdata/crawler/models.py
index 54d72e2..e01f53f 100644
--- a/src/brightdata/crawler/models.py
+++ b/src/brightdata/crawler/models.py
@@ -4,6 +4,8 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+from ..exceptions import BrightDataError
+
 
 @dataclass
 class CrawlResult:
@@ -22,6 +24,9 @@ class CrawlResult:
     trigger_sent_at: Optional[datetime] = None
     data_fetched_at: Optional[datetime] = None
     error: Optional[str] = None
+    # Underlying exception when one was converted into this result; `error`
+    # stays the message, `cause` is what code branches on.
+    cause: Optional[BrightDataError] = field(default=None, repr=False, compare=False)
 
     def __repr__(self) -> str:
         sid = f" snapshot_id={self.snapshot_id}" if self.snapshot_id else ""
diff --git a/src/brightdata/crawler/service.py b/src/brightdata/crawler/service.py
index 50d1f44..9bba3b9 100644
--- a/src/brightdata/crawler/service.py
+++ b/src/brightdata/crawler/service.py
@@ -22,7 +22,7 @@
 from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 
 from .models import CrawlJob, CrawlResult
-from ..exceptions import APIError, ValidationError
+from ..exceptions import APIError, BrightDataError, ValidationError
 from ..utils.function_detection import get_caller_function_name
 from ..utils.validation import validate_url, validate_url_list
 
@@ -222,6 +222,7 @@ async def download(
                     trigger_sent_at=trigger_sent_at,
                     data_fetched_at=datetime.now(timezone.utc),
                     error=f"Status check failed: {exc}",
+                    cause=exc,
                 )
 
             if current == "ready":
@@ -298,8 +299,19 @@ async def _scrape_sync(
                     trigger_sent_at=trigger_sent_at,
                     data_fetched_at=data_fetched_at,
                 )
-        except (ValidationError, APIError):
+        except ValidationError:
             raise
+        except APIError as exc:
+            # The engine now raises for non-2xx, so this is the same condition
+            # the status check above used to handle inline. Keep returning a
+            # CrawlResult rather than raising, which is crawl()'s contract.
+            return CrawlResult(
+                success=False,
+                trigger_sent_at=trigger_sent_at,
+                data_fetched_at=datetime.now(timezone.utc),
+                error=f"HTTP {exc.status_code}: {exc.raw or exc.message}",
+                cause=exc,
+            )
         except Exception as exc:
             return CrawlResult(
                 success=False,
@@ -347,6 +359,7 @@ async def _fetch_snapshot(
                 trigger_sent_at=trigger_sent_at,
                 data_fetched_at=datetime.now(timezone.utc),
                 error=f"Snapshot fetch error: {exc}",
+                cause=exc if isinstance(exc, BrightDataError) else None,
             )
 
     @staticmethod
diff --git a/src/brightdata/datasets/base.py b/src/brightdata/datasets/base.py
index 1a913f2..de029f5 100644
--- a/src/brightdata/datasets/base.py
+++ b/src/brightdata/datasets/base.py
@@ -3,21 +3,44 @@
 """
 
 import asyncio
+import json
 import time
 from typing import Dict, List, Any, Optional, Literal, TYPE_CHECKING
 
 from .models import DatasetMetadata, SnapshotStatus
+from ..exceptions import APIError, BrightDataError, RateLimitError
 
 if TYPE_CHECKING:
     from ..core.engine import AsyncEngine
 
 
-class DatasetError(Exception):
+class DatasetError(BrightDataError):
     """Error related to dataset operations."""
 
     pass
 
 
+def _as_dataset_error(exc: APIError, what: str) -> DatasetError:
+    """
+    Re-type an engine-raised APIError as a DatasetError, preserving context.
+
+    The engine classifies every non-2xx centrally, so by the time a failure
+    reaches this layer it is already structured. Callers of the datasets API
+    catch DatasetError, though, so convert rather than let a sibling type
+    escape. RateLimitError is deliberately NOT converted: it is the more
+    specific, actionable type and users are told to catch it directly.
+    """
+    return DatasetError(
+        f"{what} failed (HTTP {exc.status_code})",
+        status_code=exc.status_code,
+        url=exc.url,
+        method=exc.method,
+        retry_after=exc.retry_after,
+        retryable=exc.retryable,
+        raw=exc.raw,
+    )
+
+
 class BaseDataset:
     """
     Base class for all dataset types.
@@ -95,11 +118,17 @@ async def __call__(
         if records_limit is not None:
             payload["records_limit"] = records_limit
 
-        async with self._engine.post_to_url(
-            f"{self.BASE_URL}/datasets/filter",
-            json_data=payload,
-        ) as response:
-            data = await response.json()
+        try:
+            async with self._engine.post_to_url(
+                f"{self.BASE_URL}/datasets/filter",
+                json_data=payload,
+            ) as response:
+                body = await response.text()
+                data = json.loads(body) if body.strip() else {}
+        except RateLimitError:
+            raise
+        except APIError as exc:
+            raise _as_dataset_error(exc, "Filter request") from exc
 
         if "snapshot_id" not in data:
             error_msg = (
@@ -142,10 +171,16 @@ async def get_status(self, snapshot_id: str) -> SnapshotStatus:
         Returns:
             SnapshotStatus with status field: "scheduled", "building", "ready", or "failed"
         """
-        async with self._engine.get_from_url(
-            f"{self.BASE_URL}/datasets/snapshots/{snapshot_id}"
-        ) as response:
-            data = await response.json()
+        try:
+            async with self._engine.get_from_url(
+                f"{self.BASE_URL}/datasets/snapshots/{snapshot_id}"
+            ) as response:
+                body = await response.text()
+                data = json.loads(body) if body.strip() else {}
+        except RateLimitError:
+            raise
+        except APIError as exc:
+            raise _as_dataset_error(exc, "Snapshot status check") from exc
         return SnapshotStatus.from_dict(data)
 
     async def download(
@@ -197,8 +232,6 @@ async def download(
             f"{self.BASE_URL}/datasets/snapshots/{snapshot_id}/download",
             params={"format": format},
         ) as response:
-            import json
-
             # Check for HTTP errors
             if response.status >= 400:
                 error_text = await response.text()
diff --git a/src/brightdata/exceptions/__init__.py b/src/brightdata/exceptions/__init__.py
index 9974381..f0a746e 100644
--- a/src/brightdata/exceptions/__init__.py
+++ b/src/brightdata/exceptions/__init__.py
@@ -5,6 +5,7 @@
     ValidationError,
     AuthenticationError,
     APIError,
+    RateLimitError,
     DataNotReadyError,
     ZoneError,
     NetworkError,
@@ -16,6 +17,7 @@
     "ValidationError",
     "AuthenticationError",
     "APIError",
+    "RateLimitError",
     "DataNotReadyError",
     "ZoneError",
     "NetworkError",
diff --git a/src/brightdata/exceptions/errors.py b/src/brightdata/exceptions/errors.py
index fc476d3..91004b8 100644
--- a/src/brightdata/exceptions/errors.py
+++ b/src/brightdata/exceptions/errors.py
@@ -2,13 +2,62 @@
 
 from __future__ import annotations
 
+from typing import Any
+
+# Bound how much of a response body an exception retains. Error pages from
+# proxies (trust_env=True is on) or CDNs can be hundreds of KB, and exceptions
+# are routinely captured by loggers and error trackers.
+RAW_LIMIT = 4096
+
+
+def _truncate(raw: Any) -> Any:
+    """Bound a retained response body, marking it when shortened."""
+    if isinstance(raw, str) and len(raw) > RAW_LIMIT:
+        return f"{raw[:RAW_LIMIT]}…[truncated {len(raw) - RAW_LIMIT} bytes]"
+    return raw
+
 
 class BrightDataError(Exception):
-    """Base exception for all Bright Data errors."""
+    """
+    Base exception for all Bright Data errors.
+
+    Carries structured context alongside the human message so callers can react
+    programmatically instead of parsing the message string.
+
+    Attributes:
+        message: Short human-readable description.
+        status_code: HTTP status, when the failure came from a response.
+        url: URL of the failing request, when known.
+        method: HTTP method of the failing request, when known.
+        retry_after: Seconds parsed from a Retry-After header, when present.
+        retryable: Whether repeating the operation is safe and worthwhile.
+                   Defaults to False — only a raiser with enough knowledge
+                   (currently the engine, for 5xx) may set it True.
+        raw: Response body, truncated to RAW_LIMIT.
+    """
 
-    def __init__(self, message: str, *args, **kwargs):
+    def __init__(
+        self,
+        message: str,
+        *args,
+        status_code: int | None = None,
+        url: str | None = None,
+        method: str | None = None,
+        retry_after: float | None = None,
+        retryable: bool | None = None,
+        raw: Any = None,
+        **kwargs,
+    ):
         super().__init__(message, *args)
         self.message = message
+        self.status_code = status_code
+        self.url = url
+        self.method = method
+        self.retry_after = retry_after
+        if retryable is None:
+            retryable = status_code is not None and status_code >= 500
+        self.retryable = retryable
+        self.raw = _truncate(raw)
 
 
 class ValidationError(BrightDataError):
@@ -34,11 +83,28 @@ def __init__(
         *args,
         **kwargs,
     ):
-        super().__init__(message, *args, **kwargs)
-        self.status_code = status_code
+        # status_code stays the second positional parameter for backwards
+        # compatibility (APIError("msg", 429) is legal); drop any duplicate
+        # keyword before forwarding it to the base.
+        kwargs.pop("status_code", None)
+        super().__init__(message, *args, status_code=status_code, **kwargs)
         self.response_text = response_text
 
 
+class RateLimitError(APIError):
+    """
+    HTTP 429 — request rate or quota exceeded.
+
+    Never retryable: on the Bright Data API a 429 response itself consumes
+    quota, so retrying extends the lockout rather than waiting it out. Use
+    `retry_after` to decide how long to pause instead.
+    """
+
+    def __init__(self, message: str, *args, **kwargs):
+        kwargs["retryable"] = False  # not overridable — see the class docstring
+        super().__init__(message, *args, **kwargs)
+
+
 class DataNotReadyError(BrightDataError):
     """Data is not ready yet (HTTP 202). Should retry."""
 
diff --git a/src/brightdata/models.py b/src/brightdata/models.py
index d2ebee1..fa3e5ea 100644
--- a/src/brightdata/models.py
+++ b/src/brightdata/models.py
@@ -8,6 +8,8 @@
 import json
 from pathlib import Path
 
+from .exceptions import BrightDataError
+
 StatusType = Literal["ready", "error", "timeout", "in_progress"]
 PlatformType = Optional[Literal["linkedin", "amazon", "chatgpt", "instagram", "facebook"]]
 SearchEngineType = Optional[Literal["google", "bing", "yandex"]]
@@ -27,6 +29,10 @@ class BaseResult:
         error: Error message if operation failed, None otherwise.
         trigger_sent_at: Timestamp when the trigger request was sent to Bright Data (UTC-aware).
         data_fetched_at: Timestamp when data was fetched after polling completed (UTC-aware).
+        cause: The underlying exception when one was converted into this result.
+               `error` stays the human-readable message; `cause` is what code
+               branches on -- e.g. `isinstance(result.cause, RateLimitError)`
+               then `result.cause.retry_after`.
     """
 
     success: bool
@@ -34,6 +40,7 @@ class BaseResult:
     error: Optional[str] = None
     trigger_sent_at: Optional[datetime] = None
     data_fetched_at: Optional[datetime] = None
+    cause: Optional[BrightDataError] = field(default=None, repr=False, compare=False)
 
     def __post_init__(self) -> None:
         """Validate data after initialization."""
@@ -83,6 +90,17 @@ def to_dict(self) -> Dict[str, Any]:
                 result[key] = value.isoformat()
             elif isinstance(value, list) and value and isinstance(value[0], datetime):
                 result[key] = [v.isoformat() if isinstance(v, datetime) else v for v in value]
+
+        # An exception is not JSON-serializable; keep a summary so to_json()
+        # and save_to_file() still work.
+        if self.cause is not None:
+            result["cause"] = {
+                "type": type(self.cause).__name__,
+                "message": self.cause.message,
+                "status_code": self.cause.status_code,
+                "retry_after": self.cause.retry_after,
+                "retryable": self.cause.retryable,
+            }
         return result
 
     def to_json(self, indent: Optional[int] = None) -> str:
diff --git a/src/brightdata/scrapers/api_client.py b/src/brightdata/scrapers/api_client.py
index 2443517..54af39c 100644
--- a/src/brightdata/scrapers/api_client.py
+++ b/src/brightdata/scrapers/api_client.py
@@ -11,7 +11,7 @@
 
 from ..core.engine import AsyncEngine
 from http import HTTPStatus
-from ..exceptions import APIError, DataNotReadyError
+from ..exceptions import APIError, DataNotReadyError, RateLimitError
 
 
 class DatasetAPIClient:
@@ -108,12 +108,22 @@ async def get_status(self, snapshot_id: str) -> str:
         """
         url = f"{self.STATUS_URL}/{snapshot_id}"
 
-        async with self.engine.get_from_url(url) as response:
-            if response.status == HTTPStatus.OK:
+        # Do NOT collapse transport failures into a job status. Returning
+        # "error" here made an expired token look like a failed scrape; the
+        # engine raises for non-2xx, and re-raising keeps the endpoint context.
+        try:
+            async with self.engine.get_from_url(url) as response:
                 data = await response.json()
                 return data.get("status", "unknown")
-            else:
-                return "error"
+        except RateLimitError:
+            raise
+        except APIError as exc:
+            raise APIError(
+                f"Status check failed (HTTP {exc.status_code})",
+                status_code=exc.status_code,
+                raw=exc.raw,
+                retryable=exc.retryable,
+            ) from exc
 
     async def fetch_result(self, snapshot_id: str, format: str = "json") -> Any:
         """
diff --git a/src/brightdata/scrapers/workflow.py b/src/brightdata/scrapers/workflow.py
index 02d0952..40bd751 100644
--- a/src/brightdata/scrapers/workflow.py
+++ b/src/brightdata/scrapers/workflow.py
@@ -89,6 +89,7 @@ async def execute(
                 url="",
                 status="error",
                 error=f"Trigger failed: {str(e)}",
+                cause=e,
                 platform=self.platform_name,
                 method="web_scraper",
                 trigger_sent_at=trigger_sent_at,
diff --git a/src/brightdata/utils/http.py b/src/brightdata/utils/http.py
new file mode 100644
index 0000000..b42bc31
--- /dev/null
+++ b/src/brightdata/utils/http.py
@@ -0,0 +1,55 @@
+"""HTTP header helpers shared by the engine and the datasets layer."""
+
+from __future__ import annotations
+
+from email.utils import parsedate_to_datetime
+from datetime import datetime, timezone
+from http import HTTPStatus
+from typing import Any, Optional
+
+
+def parse_retry_after(headers: Any) -> Optional[float]:
+    """
+    Parse a Retry-After header into seconds.
+
+    Supports both documented forms: delta-seconds ("30") and an HTTP-date
+    ("Wed, 21 Oct 2015 07:28:00 GMT"). Returns None when the header is absent
+    or unparseable — callers treat that as "the server did not say".
+    """
+    if not headers:
+        return None
+    try:
+        value = headers.get("Retry-After")
+    except AttributeError:
+        return None
+    if not value:
+        return None
+
+    value = str(value).strip()
+    try:
+        return max(0.0, float(value))
+    except ValueError:
+        pass
+
+    try:
+        when = parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    if when is None:
+        return None
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=timezone.utc)
+    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
+
+
+def status_phrase(status: int) -> str:
+    """
+    Human-readable phrase for an HTTP status.
+
+    HTTPStatus(...) raises ValueError on non-standard codes, and those are real
+    in the wild (Cloudflare uses 520-526), so fall back rather than raise.
+    """
+    try:
+        return HTTPStatus(status).phrase
+    except ValueError:
+        return "HTTP error"
diff --git a/src/brightdata/utils/polling.py b/src/brightdata/utils/polling.py
index 94c4678..fc51059 100644
--- a/src/brightdata/utils/polling.py
+++ b/src/brightdata/utils/polling.py
@@ -16,7 +16,7 @@
 
 from ..models import ScrapeResult
 from ..constants import DEFAULT_POLL_INTERVAL, DEFAULT_POLL_TIMEOUT
-from ..exceptions import DataNotReadyError
+from ..exceptions import BrightDataError, DataNotReadyError
 
 
 async def poll_until_ready(
@@ -108,6 +108,7 @@ async def poll_until_ready(
                 url="",
                 status="error",
                 error=f"Failed to get status: {str(e)}",
+                cause=e if isinstance(e, BrightDataError) else None,
                 snapshot_id=snapshot_id,
                 platform=platform,
                 method=method or "web_scraper",
@@ -135,6 +136,7 @@ async def poll_until_ready(
                     url="",
                     status="error",
                     error=f"Failed to fetch results: {str(e)}",
+                    cause=e if isinstance(e, BrightDataError) else None,
                     snapshot_id=snapshot_id,
                     platform=platform,
                     method=method or "web_scraper",
diff --git a/src/brightdata/utils/retry.py b/src/brightdata/utils/retry.py
index 6cb91d6..55f4c1c 100644
--- a/src/brightdata/utils/retry.py
+++ b/src/brightdata/utils/retry.py
@@ -2,11 +2,32 @@
 
 import asyncio
 from typing import Callable, Awaitable, TypeVar, Optional, List, Type
-from ..exceptions import APIError, NetworkError
+from ..exceptions import BrightDataError, NetworkError
 
 T = TypeVar("T")
 
 
+def is_retryable(exc: Exception) -> bool:
+    """
+    Decide whether repeating an operation is safe and worthwhile.
+
+    Retryability is NOT inferred from the exception type or from a missing
+    status code. Several SDK errors are raised *after* the server accepted the
+    work (e.g. "Failed to trigger scrape - no snapshot_id returned"), and
+    repeating those creates a duplicate billed job. So anything raised without
+    an explicit opinion defaults to not-retryable; only a raiser with enough
+    knowledge -- currently the engine, for 5xx -- opts in.
+
+    Rate limits are never retryable: on this API a 429 response itself consumes
+    quota, so retrying extends the lockout. Use `retry_after` to pause instead.
+    """
+    if isinstance(exc, (NetworkError, TimeoutError)):
+        return True
+    if isinstance(exc, BrightDataError):
+        return exc.retryable
+    return False
+
+
 async def retry_with_backoff(
     func: Callable[[], Awaitable[T]],
     max_retries: int = 3,
@@ -32,9 +53,6 @@ async def retry_with_backoff(
     Raises:
         Last exception if all retries fail
     """
-    if retryable_exceptions is None:
-        retryable_exceptions = [NetworkError, TimeoutError, APIError]
-
     last_exception = None
     delay = initial_delay
 
@@ -45,7 +63,10 @@ async def retry_with_backoff(
             last_exception = e
 
             # Check if exception is retryable
-            if not any(isinstance(e, exc_type) for exc_type in retryable_exceptions):
+            if retryable_exceptions is None:
+                if not is_retryable(e):
+                    raise
+            elif not any(isinstance(e, exc_type) for exc_type in retryable_exceptions):
                 raise
 
             # Don't retry on last attempt
diff --git a/src/brightdata/web_unlocker/async_client.py b/src/brightdata/web_unlocker/async_client.py
index 3b6d295..3b7bccf 100644
--- a/src/brightdata/web_unlocker/async_client.py
+++ b/src/brightdata/web_unlocker/async_client.py
@@ -141,16 +141,21 @@ async def get_status(self, zone: str, response_id: str, customer: Optional[str]
         if customer:
             params["customer"] = customer
 
-        async with self.engine.get_from_url(
-            f"{self.engine.BASE_URL}{self.FETCH_ENDPOINT}", params=params
-        ) as response:
-            if response.status == 200:
-                return "ready"
-            elif response.status == 202:
-                return "pending"
-            else:
-                # Any other status (4xx, 5xx) is treated as error
+        try:
+            async with self.engine.get_from_url(
+                f"{self.engine.BASE_URL}{self.FETCH_ENDPOINT}", params=params
+            ) as response:
+                if response.status == 200:
+                    return "ready"
+                elif response.status == 202:
+                    return "pending"
+                # Unreachable in practice: the engine raises for other statuses.
                 return "error"
+        except APIError:
+            # This method's contract is a status string, not an exception -- its
+            # caller is a poll loop. The engine now raises for 4xx/5xx, so map
+            # that back to the documented "error" value.
+            return "error"
 
     async def fetch_result(
         self,
diff --git a/tests/unit/test_error_translation.py b/tests/unit/test_error_translation.py
new file mode 100644
index 0000000..d3dc8c5
--- /dev/null
+++ b/tests/unit/test_error_translation.py
@@ -0,0 +1,370 @@
+"""
+Tests for engine-level HTTP error translation and structured exceptions.
+
+Covers: per-status translation at the choke point, the 202 pass-through that
+protects the SDK's only recovery path, `retryable` correctness (the
+duplicate-job guard), export surface, and back-compat of the exception
+constructors.
+"""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+import brightdata
+import brightdata.exceptions as ex
+from brightdata.core.engine import AsyncEngine
+from brightdata.exceptions import (
+    APIError,
+    AuthenticationError,
+    BrightDataError,
+    DataNotReadyError,
+    RateLimitError,
+)
+from brightdata.utils.http import parse_retry_after, status_phrase
+from brightdata.utils.retry import is_retryable
+
+
+def _resp(status, text="body", headers=None):
+    r = AsyncMock()
+    r.status = status
+    r.text = AsyncMock(return_value=text)
+    r.release = AsyncMock()
+    r.close = MagicMock()  # aiohttp's close() is sync
+    r.headers = headers or {}
+    return r
+
+
+async def _enter(engine, response):
+    """Send one request through the engine with a canned response."""
+    engine._session.request = AsyncMock(return_value=response)
+    async with engine.get("/test") as r:
+        return r
+
+
+# ---------------------------------------------------------------------------
+# Per-status translation
+# ---------------------------------------------------------------------------
+
+
+class TestStatusTranslation:
+    @pytest.mark.parametrize("status", [401, 403])
+    async def test_auth_statuses(self, status):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(AuthenticationError) as ei:
+                await _enter(engine, _resp(status, "nope"))
+            assert ei.value.status_code == status
+            assert ei.value.raw == "nope"
+            assert ei.value.retryable is False
+
+    @pytest.mark.parametrize("status", [400, 404, 422])
+    async def test_client_errors_are_api_error_not_retryable(self, status):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(APIError) as ei:
+                await _enter(engine, _resp(status))
+            assert not isinstance(ei.value, RateLimitError)
+            assert ei.value.status_code == status
+            assert ei.value.retryable is False
+
+    @pytest.mark.parametrize("status", [500, 502, 503])
+    async def test_server_errors_are_retryable(self, status):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(APIError) as ei:
+                await _enter(engine, _resp(status))
+            assert ei.value.status_code == status
+            assert ei.value.retryable is True
+
+    async def test_429_becomes_rate_limit_error_with_retry_after(self):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(RateLimitError) as ei:
+                await _enter(engine, _resp(429, "slow down", {"Retry-After": "30"}))
+            assert ei.value.status_code == 429
+            assert ei.value.retry_after == 30.0
+            assert ei.value.retryable is False
+
+    async def test_429_with_non_json_body_is_not_a_parse_error(self):
+        """The field case: a bare string under a non-JSON content type."""
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(RateLimitError) as ei:
+                await _enter(engine, _resp(429, "too_many_parallel_jobs"))
+            assert "too_many_parallel_jobs" in ei.value.raw
+            assert ei.value.retry_after is None
+
+    async def test_non_standard_status_does_not_crash_phrase_lookup(self):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(APIError) as ei:
+                await _enter(engine, _resp(520))  # Cloudflare
+            assert ei.value.status_code == 520
+
+    async def test_error_carries_url_and_method(self):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(APIError) as ei:
+                await _enter(engine, _resp(500))
+            assert ei.value.method == "GET"
+            assert ei.value.url.endswith("/test")
+
+    async def test_raw_is_bounded(self):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            with pytest.raises(APIError) as ei:
+                await _enter(engine, _resp(500, "x" * 20000))
+            assert len(ei.value.raw) < 20000
+            assert "truncated" in ei.value.raw
+
+
+# ---------------------------------------------------------------------------
+# 202 must pass through — the SDK's only recovery path depends on it
+# ---------------------------------------------------------------------------
+
+
+class TestAcceptedPassesThrough:
+    @pytest.mark.parametrize("status", [200, 201, 202, 204])
+    async def test_success_and_accepted_are_returned_not_raised(self, status):
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            r = await _enter(engine, _resp(status))
+            assert r.status == status
+
+    async def test_fetch_result_still_raises_data_not_ready_on_202(self):
+        from brightdata.scrapers.api_client import DatasetAPIClient
+
+        resp = _resp(202, "still building")
+        engine = MagicMock()
+        engine.get_from_url = MagicMock(return_value=_ctx(resp))
+        with pytest.raises(DataNotReadyError):
+            await DatasetAPIClient(engine).fetch_result("snap_1")
+
+
+def _ctx(response):
+    cm = MagicMock()
+    cm.__aenter__ = AsyncMock(return_value=response)
+    cm.__aexit__ = AsyncMock(return_value=False)
+    return cm
+
+
+# ---------------------------------------------------------------------------
+# retryable — the duplicate-job guard
+# ---------------------------------------------------------------------------
+
+
+class TestRetryable:
+    def test_missing_status_is_never_retryable(self):
+        """Raised locally, possibly after the server accepted the work."""
+        assert is_retryable(APIError("Failed to trigger scrape - no snapshot_id returned")) is False
+
+    def test_explicit_5xx_is_retryable(self):
+        assert is_retryable(APIError("x", status_code=500)) is True
+
+    def test_4xx_is_not_retryable(self):
+        assert is_retryable(APIError("x", status_code=400)) is False
+
+    def test_rate_limit_is_never_retryable_even_if_flagged(self):
+        assert is_retryable(RateLimitError("x", status_code=429, retryable=True)) is False
+
+    def test_network_and_timeout_are_retryable(self):
+        from brightdata.exceptions import NetworkError
+
+        assert is_retryable(NetworkError("down")) is True
+        assert is_retryable(TimeoutError("slow")) is True
+
+
+# ---------------------------------------------------------------------------
+# Behaviour preserved at repaired call sites
+# ---------------------------------------------------------------------------
+
+
+class TestRepairedCallSites:
+    async def test_crawler_scrape_returns_result_on_http_error(self):
+        from brightdata.crawler.service import CrawlerService
+
+        client = MagicMock()
+        cm = MagicMock()
+        cm.__aenter__ = AsyncMock(side_effect=APIError("boom", status_code=500, raw="oops"))
+        cm.__aexit__ = AsyncMock(return_value=False)
+        client.engine.post_to_url = MagicMock(return_value=cm)
+
+        result = await CrawlerService(client).crawl(urls="https://example.com")
+        assert result.success is False
+        assert "500" in result.error
+
+    @pytest.mark.parametrize(
+        "raised,expected",
+        [
+            (APIError("x", status_code=500), "error"),
+            (RateLimitError("x", status_code=429), "error"),
+        ],
+    )
+    async def test_unlocker_status_returns_string_never_raises(self, raised, expected):
+        from brightdata.web_unlocker.async_client import AsyncUnblockerClient
+
+        engine = MagicMock()
+        engine.BASE_URL = "https://api.brightdata.com"
+        cm = MagicMock()
+        cm.__aenter__ = AsyncMock(side_effect=raised)
+        cm.__aexit__ = AsyncMock(return_value=False)
+        engine.get_from_url = MagicMock(return_value=cm)
+
+        got = await AsyncUnblockerClient(engine).get_status(zone="z", response_id="r")
+        assert got == expected
+
+    async def test_poll_reports_transport_failure_not_job_failure(self):
+        from brightdata.utils.polling import poll_until_ready
+
+        async def bad_status(_):
+            raise AuthenticationError("Unauthorized (401)", status_code=401)
+
+        async def never(_):
+            raise AssertionError("should not fetch")
+
+        r = await poll_until_ready(bad_status, never, "snap_1", poll_interval=0, poll_timeout=5)
+        assert r.success is False
+        assert "Failed to get status" in r.error
+        assert "Job failed" not in r.error
+
+
+# ---------------------------------------------------------------------------
+# Exports and back-compat
+# ---------------------------------------------------------------------------
+
+
+class TestExportsAndBackCompat:
+    def test_rate_limit_error_importable_at_both_levels(self):
+        assert brightdata.RateLimitError is ex.RateLimitError
+
+    def test_every_exception_name_is_exported_top_level(self):
+        missing = [n for n in ex.__all__ if n not in brightdata.__all__]
+        assert not missing, f"not re-exported from brightdata: {missing}"
+
+    def test_legacy_constructors_still_work(self):
+        assert BrightDataError("m").message == "m"
+        assert APIError("m").status_code is None
+        assert APIError("m", 429).status_code == 429  # positional, as before
+        assert APIError("m", 429, "body").response_text == "body"
+
+    def test_rate_limit_is_catchable_as_api_error(self):
+        with pytest.raises(APIError):
+            raise RateLimitError("x", status_code=429)
+
+    def test_dataset_error_is_catchable_both_ways(self):
+        from brightdata.datasets import DatasetError
+
+        assert issubclass(DatasetError, BrightDataError)
+        assert issubclass(DatasetError, Exception)
+
+
+class TestHttpHelpers:
+    @pytest.mark.parametrize(
+        "value,expected", [("30", 30.0), ("0", 0.0), (None, None), ("soon", None)]
+    )
+    def test_parse_retry_after(self, value, expected):
+        headers = {"Retry-After": value} if value is not None else {}
+        assert parse_retry_after(headers) == expected
+
+    def test_status_phrase_handles_non_standard(self):
+        assert status_phrase(429) == "Too Many Requests"
+        assert status_phrase(520)  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# Structure must survive to the caller (the point of the whole change)
+# ---------------------------------------------------------------------------
+
+
+class TestCausePropagation:
+    async def test_rate_limit_during_polling_reaches_the_caller(self):
+        from brightdata.utils.polling import poll_until_ready
+
+        async def rate_limited(_):
+            raise RateLimitError("Rate limited (429)", status_code=429, retry_after=45.0)
+
+        async def never(_):
+            raise AssertionError("should not fetch")
+
+        r = await poll_until_ready(rate_limited, never, "snap_1", poll_interval=0, poll_timeout=5)
+
+        assert r.success is False
+        assert isinstance(r.cause, RateLimitError)
+        assert r.cause.status_code == 429
+        assert r.cause.retry_after == 45.0
+        assert r.cause.retryable is False
+
+    async def test_trigger_failure_carries_cause(self):
+        from brightdata.scrapers.workflow import WorkflowExecutor
+
+        api = MagicMock()
+        api.trigger = AsyncMock(side_effect=APIError("boom", status_code=503))
+        r = await WorkflowExecutor(api).execute(payload=[{"url": "u"}], dataset_id="gd_x")
+
+        assert r.success is False
+        assert isinstance(r.cause, APIError)
+        assert r.cause.status_code == 503
+        assert r.cause.retryable is True
+
+    async def test_crawler_result_carries_cause(self):
+        from brightdata.crawler.service import CrawlerService
+
+        client = MagicMock()
+        cm = MagicMock()
+        cm.__aenter__ = AsyncMock(side_effect=RateLimitError("rl", status_code=429, retry_after=10))
+        cm.__aexit__ = AsyncMock(return_value=False)
+        client.engine.post_to_url = MagicMock(return_value=cm)
+
+        r = await CrawlerService(client).crawl(urls="https://example.com")
+        assert isinstance(r.cause, RateLimitError)
+        assert r.cause.retry_after == 10
+
+    def test_result_with_cause_is_still_serializable(self):
+        from brightdata.models import ScrapeResult
+
+        r = ScrapeResult(
+            success=False,
+            error="rate limited",
+            cause=RateLimitError("rl", status_code=429, retry_after=30.0),
+        )
+        payload = r.to_dict()
+        assert payload["cause"]["type"] == "RateLimitError"
+        assert payload["cause"]["status_code"] == 429
+        r.to_json()  # must not raise
+
+
+class TestDatasetErrorContract:
+    """The datasets layer must keep raising DatasetError after central translation."""
+
+    async def _call(self, status, body, headers=None):
+        from brightdata.datasets.base import BaseDataset
+
+        class DS(BaseDataset):
+            DATASET_ID = "gd_x"
+            NAME = "x"
+
+        engine = AsyncEngine(bearer_token="tok")
+        async with engine:
+            engine._session.request = AsyncMock(return_value=_resp(status, body, headers))
+            return await DS(engine)(filter={"name": "f", "operator": "is_not_null"})
+
+    async def test_client_error_is_still_a_dataset_error(self):
+        from brightdata.datasets import DatasetError
+
+        with pytest.raises(DatasetError) as ei:
+            await self._call(400, '{"error":"bad filter"}')
+        assert ei.value.status_code == 400
+        assert "bad filter" in ei.value.raw
+
+    async def test_rate_limit_stays_rate_limit_error(self):
+        with pytest.raises(RateLimitError) as ei:
+            await self._call(429, "too_many_parallel_jobs", {"Retry-After": "60"})
+        assert ei.value.retry_after == 60.0
+
+    async def test_message_keeps_a_body_excerpt(self):
+        with pytest.raises(APIError) as ei:
+            engine = AsyncEngine(bearer_token="tok")
+            async with engine:
+                await _enter(engine, _resp(400, "helpful detail"))
+        assert "helpful detail" in str(ei.value)
+        assert len(str(ei.value)) < 300  # bounded