From 0eb970084d9cf6d92f20dab9f52ea8ba600fd654 Mon Sep 17 00:00:00 2001 From: Ray Date: Wed, 2 Sep 2026 19:16:14 +0800 Subject: [PATCH 1/2] fix: .events survives partial reads; hidden call lines still label results; bad show_process chokes first - ChatStream.events delegated with `yield from`, so a dropped handle (next(stream.events), for ... break) closed the shared run on GC and the rest of the run silently vanished. A plain loop leaves it alone. - _weave filled call_args only past the tool_call visibility guard, so with call lines hidden the standalone result lines never carried the arguments they promise. - show_process is validated before the stream check: an invalid value is refused as such instead of being told to add stream=True and then refused again; the managed lane's duplicate choke goes with it. - Docstring: show_process is not own-model-only. Claude-Session: https://claude.ai/code/session_016M3qaQedSK7L4DwysFRmk2 --- pageindex/client.py | 29 ++++++++++++++--------------- pageindex/local_chat.py | 10 ++++++---- tests/test_local_chat.py | 28 ++++++++++++++++++++++++++-- 3 files changed, 46 insertions(+), 21 deletions(-) diff --git a/pageindex/client.py b/pageindex/client.py index 4e7ac2453..3fccd4976 100644 --- a/pageindex/client.py +++ b/pageindex/client.py @@ -983,12 +983,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`` @@ -1068,12 +1067,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 +1133,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, diff --git a/pageindex/local_chat.py b/pageindex/local_chat.py index 4b167f120..ef9c6eabd 100644 --- a/pageindex/local_chat.py +++ b/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: diff --git a/tests/test_local_chat.py b/tests/test_local_chat.py index 2c987bbf3..4d16cd19a 100644 --- a/tests/test_local_chat.py +++ b/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 From 0791a2c0b0dc24f7756be87d8afd0b0778750e5d Mon Sep 17 00:00:00 2001 From: Ray Date: Thu, 3 Sep 2026 14:54:21 +0800 Subject: [PATCH 2/2] fix: mid-stream error chunk raises instead of ending as a short answer; stream docstring says show_process is on by default - The managed endpoint reports a server-side failure as a final {"error": ...} chunk after the partial answer (api.py refunds the credits, then yields it). Neither chunk decoder looked at it, so chat(stream=True) and chat_completions(stream=True) in both modes ended as an apparently complete short answer with no exception. One guard in each decoder raises PageIndexAPIError; the partial answer is still delivered first. - The `stream:` arg and the Returns block still described the pre-PR contract (bare text chunks); only the show_process paragraph said it is on by default. Claude-Session: https://claude.ai/code/session_01PYr9yG1FPQxKCA9m7ECQWY --- pageindex/client.py | 21 ++++++++++++--------- pageindex/cloud_api.py | 8 ++++++++ tests/test_client.py | 24 ++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 9 deletions(-) diff --git a/pageindex/client.py b/pageindex/client.py index 3fccd4976..c3859080a 100644 --- a/pageindex/client.py +++ b/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. @@ -1042,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 diff --git a/pageindex/cloud_api.py b/pageindex/cloud_api.py index df3c0088c..a676cf3f0 100644 --- a/pageindex/cloud_api.py +++ b/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 diff --git a/tests/test_client.py b/tests/test_client.py index 3bdadde6e..f423419e8 100644 --- a/tests/test_client.py +++ b/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"}}]}