Preserve provider error details in high-level streams

README.md

@@ -63,6 +63,11 @@ Terminal outcomes are immutable `RunSuccess`, `RunInterrupted`, or
 do not raise. Setup, connection, process, protocol, timeout, and cancellation
 failures do. `asyncio.CancelledError` is never wrapped.
 
+`ErrorEvent.error_name` exposes the nested provider error name when available,
+so applications can classify failures without matching message text.
+`error_type` retains the protocol's top-level category; `error_detail` contains
+the immutable nested payload. Both fields are `None` when detail is absent.
+
 The wheel contains `py.typed`. Result and event unions narrow with
 `isinstance()`. `stream.result` is cached after completion and raises
 `StreamIncompleteError` before then.

src/droid_sdk/_high_level/messages.py

@@ -210,12 +210,22 @@ class HookExecution:
 
 @dataclass(frozen=True, slots=True)
 class ErrorEvent:
+    """Preserve provider error identity without requiring message-text parsing."""
+
     message: str
     error_type: ErrorType
     timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    error_name: str | None = field(default=None, kw_only=True)
+    error_detail: Mapping[str, object] | None = field(default=None, kw_only=True)
 
     def __post_init__(self) -> None:
         ensure_aware(self.timestamp, "timestamp")
+        if self.error_detail is not None:
+            object.__setattr__(
+                self,
+                "error_detail",
+                freeze_json_object(self.error_detail, where="error detail"),
+            )
 
 
 Message: TypeAlias = (

src/droid_sdk/_high_level/streaming.py

@@ -523,6 +523,14 @@ def _convert(
                     message=notification.message,
                     error_type=ErrorType(notification.error_type.value),
                     timestamp=_parse_timestamp(notification.timestamp),
+                    error_name=(
+                        notification.error.name if notification.error else None
+                    ),
+                    error_detail=(
+                        notification.error.model_dump(by_alias=True)
+                        if notification.error
+                        else None
+                    ),
                 )
             ]
         if isinstance(notification, PermissionResolvedNotification):

tests/test_high_level_error_details.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+import pytest
+
+from droid_sdk import ErrorEvent, ErrorType, RunStream, RunSuccess
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+# --- Notification fixture helpers ---
+def _complete() -> dict[str, object]:
+    return {
+        "type": "agent_turn_completed",
+        "reason": "completed",
+        "turnId": "turn",
+        "durationMs": 1,
+        "tokenUsage": {
+            "inputTokens": 0,
+            "outputTokens": 0,
+            "cacheCreationTokens": 0,
+            "cacheReadTokens": 0,
+            "thinkingTokens": 0,
+        },
+    }
+
+
+# --- End notification fixture helpers ---
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("name", ["LLMInvalidRequestError", "LLMServiceError"])
+async def test_high_level_stream_preserves_error_discriminator(name: str) -> None:
+    """Applications must classify equal-message errors without parsing prose."""
+    detail: dict[str, object] = {
+        "name": name,
+        "message": "Provider request failed",
+        "cause": {"status": 404, "attempts": [1, 2]},
+    }
+    stream = RunStream[None](expected_turn_id="turn", session_id="session")
+    stream.feed_notification(
+        {
+            "type": "error",
+            "message": "Provider request failed",
+            "errorType": "Error",
+            "timestamp": "2026-08-02T11:00:13.647Z",
+            "error": detail,
+        }
+    )
+    stream.feed_notification(_complete())
+
+    events = [event async for event in stream]
+    error = events[0]
+    assert isinstance(error, ErrorEvent)
+    assert error.error_type is ErrorType.ERROR
+    assert error.message == "Provider request failed"
+    assert error.error_name == name
+    assert error.error_detail is not None
+    assert error.error_detail["name"] == name
+    cause = cast("Mapping[str, object]", error.error_detail["cause"])
+    assert cause["status"] == 404
+    assert cause["attempts"] == (1, 2)
+    assert error.timestamp.isoformat() == "2026-08-02T11:00:13.647000+00:00"
+    assert isinstance(events[-1], RunSuccess)
+    assert stream.result.messages == (error,)
+    with pytest.raises(TypeError):
+        cast("dict[str, object]", cause)["status"] = 500
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("extra", [{}, {"error": None}])
+async def test_high_level_error_without_detail_remains_supported(
+    extra: dict[str, object],
+) -> None:
+    stream = RunStream[None](expected_turn_id="turn", session_id="session")
+    stream.feed_notification(
+        {
+            "type": "error",
+            "message": "recoverable",
+            "errorType": "Error",
+            "timestamp": "2025-01-01T00:00:00Z",
+            **extra,
+        }
+    )
+    stream.feed_notification(_complete())
+
+    events = [event async for event in stream]
+    error = events[0]
+    assert isinstance(error, ErrorEvent)
+    assert error.error_name is None
+    assert error.error_detail is None
+
+
+def test_error_detail_is_detached_from_mutable_constructor_input() -> None:
+    detail: dict[str, object] = {"cause": {"attempts": [1, 2]}}
+    error = ErrorEvent(
+        message="failed",
+        error_type=ErrorType.ERROR,
+        error_detail=detail,
+    )
+    cast("dict[str, object]", detail["cause"])["attempts"] = [3]
+
+    assert error.error_detail is not None
+    cause = cast("Mapping[str, object]", error.error_detail["cause"])
+    assert cause["attempts"] == (1, 2)