Chat process follow-ups: a mid-stream error raises, .events survives partial reads, bad show_process chokes first

pageindex/client.py

@@ -968,12 +968,14 @@ def chat(
                 prompted. Cloud documents: the managed chat scopes
                 server-side; own-model chat targets at the prompt level.
             stream: Answer lane: return a ``ChatStream`` — iterate it for
-                the answer as text chunks as they are produced, or read
-                its ``.events`` property instead for the run as typed
-                event dicts — thinking/answer deltas, each tool call and
-                its full result (own-model chat only; never clipped). One
-                run serves one view. Protocol lanes: the protocol's own
-                event stream.
+                the answer as text chunks as they are produced
+                (``show_process`` is on by default, so the run's process
+                arrives woven in; ``show_process=False`` gives the bare
+                answer), or read its ``.events`` property instead for the
+                run as typed event dicts — thinking/answer deltas, each
+                tool call and its full result (own-model chat only; never
+                clipped). One run serves one view. Protocol lanes: the
+                protocol's own event stream.
             model: Own-model chat only — backend model name (defaults
                 to ``chat_model``). ``protocol="messages"`` needs it named
                 — a Claude model; there is no cross-vendor default.
@@ -983,12 +985,11 @@ def chat(
                 spelling: LiteLLM's ``reasoning_effort``, Responses
                 ``reasoning.effort``, Messages ``output_config.effort``.
                 Unset sends nothing — the model's default applies.
-            show_process: Answer lane, streamed own-model chat — weave
-                the run into the text stream for display: thinking flows
-                as "[thinking] " sections, each tool call as a
-                "[tool_call] name arguments" line with its "[tool_result]"
-                line, and the answer unlabeled. **On by default**, weaving
-                what the mode
+            show_process: Answer lane, streamed chat — weave the run into
+                the text stream for display: thinking flows as
+                "[thinking] " sections, each tool call as a "[tool_call]
+                name arguments" line with its "[tool_result]" line, and
+                the answer unlabeled. **On by default**, weaving what the mode
                 serves: the in-process agent's full run; on a managed
                 client, the tool calls the endpoint streams (its wire
                 carries no thinking and no tool results). Pass ``False``
@@ -1043,9 +1044,10 @@ def chat(
         Returns:
             - answer lane, stream=False: the answer string
             - answer lane, stream=True: a ``ChatStream`` — iterating it
-              yields text chunks (with show_process, the run's process
-              woven in as labeled sections); ``.events`` yields typed
-              event dicts: ``{"type": "thinking"|"answer", "delta": ...}``,
+              yields text chunks (with show_process, on by default, the
+              run's process woven in as labeled sections); ``.events``
+              yields typed event dicts:
+              ``{"type": "thinking"|"answer", "delta": ...}``,
               ``{"type": "tool_call", "call_id", "name", "arguments"}``,
               ``{"type": "tool_result", "call_id", "name", "output"}``
             - protocol lane, stream=False: the protocol's response
@@ -1068,12 +1070,14 @@ def chat(
                 f"protocol={protocol!r} the run comes back as the protocol's "
                 "own transcript and events — drop show_process, or drop "
                 "protocol for the woven text stream.")
-        if (show_process is not False and show_process is not None
-                and not stream):
-            raise PageIndexAPIError(
-                "show_process shows the run as it happens and requires "
-                "stream=True; only show_process=False (or None) means "
-                f"off — got {show_process!r}.")
+        if show_process is not False and show_process is not None:
+            from .local_chat import _process_options
+            _process_options(show_process)  # a bad value chokes first
+            if not stream:
+                raise PageIndexAPIError(
+                    "show_process shows the run as it happens and requires "
+                    "stream=True; only show_process=False (or None) means "
+                    f"off — got {show_process!r}.")
         if isinstance(instructions, list) and protocol != "messages":
             raise PageIndexAPIError(
                 "instructions blocks are the Messages protocol's shape — "
@@ -1132,9 +1136,7 @@ def chat(
                                        max_turns=max_turns, backend=backend,
                                        extra_headers=extra_headers,
                                        extra_body=extra_body)
-            from .local_chat import _process_options, run_cloud_chat_stream
-            if resolved is not False:
-                _process_options(resolved)  # choke before the request is sent
+            from .local_chat import run_cloud_chat_stream
             chunks = self.chat_completions(messages, stream=True,
                                            stream_metadata=True,
                                            doc_id=doc_id, model=model,

pageindex/cloud_api.py

@@ -272,6 +272,10 @@ def _stream_chat_response(self, response: requests.Response) -> Iterator[str]:
 
                         try:
                             chunk = json.loads(data)
+                            if chunk.get("error"):
+                                raise PageIndexAPIError(
+                                    "Chat completion failed mid-stream: "
+                                    f"{chunk['error']}")
                             choices = chunk.get("choices") or [{}]
                             content = choices[0].get("delta", {}).get("content", "")
                             if content:
@@ -294,6 +298,10 @@ def _stream_chat_response_raw(self, response: requests.Response) -> Iterator[Dic
 
                         try:
                             chunk = json.loads(data)
+                            if chunk.get("error"):
+                                raise PageIndexAPIError(
+                                    "Chat completion failed mid-stream: "
+                                    f"{chunk['error']}")
                             yield chunk
                         except json.JSONDecodeError:
                             continue

pageindex/local_chat.py

@@ -742,14 +742,14 @@ def enter(kind, label: str = "") -> str:
                         if section != "thinking" else "")
                 yield head + ev["delta"]
             elif kind == "tool_call":
-                if not options["tool_call"]:
-                    continue
                 arguments = ev["arguments"]
                 if not isinstance(arguments, str):
                     arguments = json.dumps(arguments, ensure_ascii=False)
                 clipped = _clip(arguments, cap)
+                call_args[ev["call_id"]] = clipped  # even with call lines hidden
+                if not options["tool_call"]:
+                    continue
                 last_call = ev["call_id"]
-                call_args[last_call] = clipped
                 line = f"[tool_call] {ev['name']} {clipped}"
                 yield enter("tool") + line.rstrip()
             elif kind == "tool_result":
@@ -821,7 +821,9 @@ def consume():
                 if self._closed:
                     return
                 self._it = self._events()
-            yield from self._it
+            # no `yield from`: a dropped handle must not close the run
+            for ev in self._it:
+                yield ev
         return consume()
 
     def close(self) -> None:

tests/test_client.py

@@ -1503,6 +1503,30 @@ def test_cloud_chat_stream_parsing(cloud, monkeypatch):
     assert {"object": "chat.completion.citations", "citations": []} in chunks
 
 
+def test_cloud_chat_stream_error_chunk_raises(cloud, monkeypatch):
+    """A server-side failure mid-stream arrives as a final {"error": ...}
+    chunk after the partial answer: every streaming surface raises on it
+    instead of ending as a short, seemingly complete answer."""
+    client, calls, fake = cloud
+    lines = [
+        b'data: {"choices": [{"delta": {"content": "Partial"}}]}',
+        b'data: {"error": {"message": "boom", "type": "internal_error"}}',
+    ]
+    _patch_requests(monkeypatch, lambda m, url, kw: FakeResponse(lines=lines))
+    for stream in (
+        lambda: client.chat_completions("q", stream=True),
+        lambda: client.chat_completions("q", stream=True,
+                                        stream_metadata=True),
+        lambda: client.chat("q", stream=True),
+    ):
+        it = stream()
+        first = next(it)  # the partial answer is still delivered
+        assert first in ("Partial",
+                         {"choices": [{"delta": {"content": "Partial"}}]})
+        with pytest.raises(PageIndexAPIError, match="boom"):
+            list(it)
+
+
 def test_cloud_chat_accepts_query_string(cloud):
     client, calls, fake = cloud
     fake.payload = {"choices": [{"message": {"content": "ok"}}]}

tests/test_local_chat.py

@@ -513,7 +513,13 @@ def test_chat_process_requires_stream(client):
     # must say how to turn it off, not claim the caller passed True
     with pytest.raises(PageIndexAPIError,
                        match="only show_process=False"):
-        client.chat("q", show_process=0)
+        client.chat("q", show_process={})
+    # an invalid value is refused as such, with or without stream=True —
+    # never told to add stream=True first
+    for kwargs in ({}, {"stream": True}):
+        with pytest.raises(PageIndexAPIError,
+                           match="must be True, False, or a dict"):
+            client.chat("q", show_process=0, **kwargs)
 
 
 def _cloud_chunk(content=None, meta=None, choices=True):
@@ -646,7 +652,8 @@ def run(process):
 
     no_calls = run({"tool_call": False})
     assert "[tool_call]" not in no_calls
-    assert "\n\n[tool_result] get_document: " in no_calls  # results stand alone
+    # results stand alone, each echoing its call's arguments
+    assert '\n\n[tool_result] get_document {"doc_name": "report.pdf"}: ' in no_calls
     assert "[thinking] Need the report" in no_calls
 
     calls_only = run({"tool_result": False})
@@ -723,6 +730,23 @@ def test_chat_stream_events_read_is_inert(client, store_path, fake_model):
     assert "".join(stream) == "The answer"
 
 
+@needs_agents
+def test_chat_stream_events_survive_partial_reads(client, store_path,
+                                                  fake_model):
+    """Peek at one event, then read the rest: the dropped .events handle
+    must not close the run underneath."""
+    seed_doc(store_path, "pi-a", "report.pdf")
+    fake_model([
+        [_call_item("get_document", {"doc_name": "report.pdf"})],
+        [_msg_item("The answer")],
+    ])
+    stream = client.chat("What status?", stream=True)
+    first = next(stream.events)
+    rest = list(stream.events)
+    assert first["type"] == "tool_call"
+    assert [ev["type"] for ev in rest] == ["tool_result", "answer", "answer"]
+
+
 def test_chat_stream_events_refusal_waits_for_consumption(monkeypatch):
     """On managed, .events read is inert — getattr(stream, 'events',
     None) must not explode — and the refusal raises on first