refactor(workflows): move stream error hook into _hooks and cover logs streams

- Move WorkflowStreamErrorHook to client/_hooks alongside the other hooks
- Cover deployment + execution logs SSE streams (same error-frame contract)
- Derive valid reasons from the StreamDisconnectReason Literal
- Add tests: error frame without trailing boundary; all stream operations

Author: mistralai-nfau

src/mistralai/client/_hooks/registration.py

@@ -3,8 +3,8 @@
 from .traceparent import TraceparentInjectionHook
 from .tracing import TracingHook
 from .types import Hooks
+from .stream_error_hook import WorkflowStreamErrorHook
 from .workflow_encoding_hook import WorkflowEncodingHook
-from mistralai.extra.workflows.stream_error_hook import WorkflowStreamErrorHook
 
 # This file is only ever generated once on the first generation and then is free to be modified.
 # Any hooks you wish to add should be registered in the init_hooks function. Feel free to define them

…lai/extra/workflows/stream_error_hook.py …ralai/client/_hooks/stream_error_hook.pysrc/mistralai/extra/workflows/stream_error_hook.py renamed to src/mistralai/client/_hooks/stream_error_hook.py src/mistralai/extra/workflows/stream_error_hook.py renamed to src/mistralai/client/_hooks/stream_error_hook.py

@@ -1,26 +1,27 @@
-from __future__ import annotations
-
 import json
 import re
-from typing import Any, AsyncIterator, Iterator, Optional, Tuple, Union
+from typing import Any, AsyncIterator, Dict, Iterator, Optional, Tuple, Union, get_args
 
 import httpx
 from httpx._types import AsyncByteStream, SyncByteStream
 
-from mistralai.client._hooks.types import AfterSuccessContext, AfterSuccessHook
+from .types import AfterSuccessContext, AfterSuccessHook
 from mistralai.extra.exceptions import (
     StreamDisconnectReason,
     StreamDisconnectedError,
 )
 
-# Operation IDs of the two SSE-backed workflow stream endpoints.
+# Operation IDs of the SSE-backed workflow stream endpoints that can emit a
+# terminal ``event: error`` frame (event, execution, and logs streams).
 STREAM_OPERATIONS = {
     "get_stream_events_v1_workflows_events_stream_get",
     "stream_v1_workflows_executions__execution_id__stream_get",
+    "stream_deployment_logs",
+    "stream_workflow_execution_logs",
 }
 
 _ERROR_EVENT = "error"
-_VALID_REASONS = ("read_error", "stream_error", "internal_error")
+_VALID_REASONS = get_args(StreamDisconnectReason)
 _DEFAULT_REASON: StreamDisconnectReason = "stream_error"
 
 # SSE frame boundaries (blank line), longest first so the full separator is consumed.
@@ -59,7 +60,7 @@ def _find_boundary(buffer: bytearray) -> Optional[Tuple[int, int]]:
 
 
 def _parse_error_payload(data: str) -> Tuple[str, StreamDisconnectReason]:
-    payload: dict[str, Any] = {}
+    payload: Dict[str, Any] = {}
     try:
         parsed = json.loads(data.strip())
         if isinstance(parsed, dict):
@@ -156,7 +157,7 @@ async def aclose(self) -> None:
 class WorkflowStreamErrorHook(AfterSuccessHook):
     """Raise StreamDisconnectedError when a workflow SSE stream sends an error frame.
 
-    Wraps the response byte stream for the two workflow SSE operations so that an
+    Wraps the response byte stream for the workflow SSE operations so that an
     ``event: error`` frame raises during iteration, terminating the consumer's
     ``for event in stream`` loop instead of yielding the error as a normal event.
     """

src/mistralai/extra/tests/test_stream_error_hook.py

@@ -5,7 +5,10 @@
 from mistralai.client import Mistral
 from mistralai.client._hooks.types import AfterSuccessContext, HookContext
 from mistralai.extra.exceptions import StreamDisconnectedError
-from mistralai.extra.workflows.stream_error_hook import WorkflowStreamErrorHook
+from mistralai.client._hooks.stream_error_hook import (
+    STREAM_OPERATIONS,
+    WorkflowStreamErrorHook,
+)
 
 STREAM_OPERATION_ID = "get_stream_events_v1_workflows_events_stream_get"
 NON_STREAM_OPERATION_ID = "chat_completion_v1_chat_completions_post"
@@ -119,6 +122,21 @@ def test_hook_detects_error_frame_split_across_chunks():
     assert exc_info.value.error == "x"
 
 
+def test_hook_raises_on_error_frame_without_trailing_boundary():
+    frame = b'event: error\ndata: {"error": "boom", "reason": "read_error"}'
+    response = _sse_response(_SyncSource([frame]))
+    result = WorkflowStreamErrorHook().after_success(
+        _hook_ctx(STREAM_OPERATION_ID), response
+    )
+    assert isinstance(result, httpx.Response)
+
+    with pytest.raises(StreamDisconnectedError) as exc_info:
+        list(result.iter_bytes())
+
+    assert exc_info.value.reason == "read_error"
+    assert exc_info.value.error == "boom"
+
+
 def test_hook_defaults_reason_when_missing_or_invalid():
     frame = b'event: error\ndata: {"error": "no reason given"}\n\n'
     response = _sse_response(_SyncSource([frame]))
@@ -145,6 +163,19 @@ def test_hook_passes_normal_stream_through_without_raising():
     assert body.count(b"workflow.event") == 2
 
 
+@pytest.mark.parametrize("operation_id", sorted(STREAM_OPERATIONS))
+def test_hook_raises_for_every_workflow_stream_operation(operation_id: str):
+    response = _sse_response(_SyncSource([ERROR_FRAME]))
+    result = WorkflowStreamErrorHook().after_success(_hook_ctx(operation_id), response)
+    assert isinstance(result, httpx.Response)
+
+    with pytest.raises(StreamDisconnectedError) as exc_info:
+        list(result.iter_bytes())
+
+    assert exc_info.value.reason == "read_error"
+    assert exc_info.value.error == "boom"
+
+
 def test_hook_ignores_non_stream_operations():
     source = _SyncSource([ERROR_FRAME])
     response = _sse_response(source)