From be34abbeb1eefc0a326828906c7aae4f92d4668e Mon Sep 17 00:00:00 2001 From: John Niche Date: Tue, 14 Jul 2026 17:38:32 -0300 Subject: [PATCH 1/3] feat: public streaming + should_finalize contracts for final_turn_only An integrator adopting v1.2.0 final_turn_only in a streaming chat surface had to read core.py to learn two load-bearing behaviors; a refactor could have silently flipped them with zero signal. Both are now documented, stated in docstrings, and pinned by dedicated contract tests (#174): - Streaming guarantee: final_turn_only + astream NEVER emits the structured answer as content chunks, on any path. New: single_pass mode now SUPPRESSES content chunks (with the schema riding natively on every loop call, content is the JSON envelope by construction) - closing the leak that was previously a documented caveat. Tool-call chunks still stream. structured_synthesis_start fires iff a separate synthesis call is made (reuse and should_finalize-skip paths emit no event, pinned). - should_finalize messages contract: the predicate receives the run's conversation view at convergence - prior history (with memory), this turn's USER message(s), ASSISTANT tool-call messages, and one Message(role=TOOL) per executed tool with tool_name, tool_call_id, and tool_result ALWAYS populated. To back the guarantee, _append_tool_result now defaults tool_result to the message content, so error results and policy denials populate it too (previously success-path only). Closes #174 Claude-Session: https://claude.ai/code/session_01L1JWChKMs9yTyK92qquhJa --- CHANGELOG.md | 19 ++ docs/CHANGELOG.md | 19 ++ docs/modules/AGENT.md | 30 ++- src/selectools/agent/_tool_executor.py | 11 +- src/selectools/agent/config_groups.py | 21 +- src/selectools/agent/core.py | 12 +- .../test_structured_streaming_contract.py | 250 ++++++++++++++++++ 7 files changed, 345 insertions(+), 17 deletions(-) create mode 100644 tests/agent/test_structured_streaming_contract.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 142084e..e210bff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Public streaming + `should_finalize` contracts for `final_turn_only` + (#174).** Two previously source-only behaviors are now documented, typed + into docstrings, and pinned by dedicated contract tests + (`tests/agent/test_structured_streaming_contract.py`): + - `final_turn_only` + `astream()` NEVER emits the structured answer as + content chunks — on any path. New in this change: `single_pass` mode now + **suppresses** content chunks (they are the JSON envelope by construction + when the schema rides on every loop call), closing the leak that was + previously a documented caveat; tool-call chunks still stream. + `structured_synthesis_start` fires iff a separate synthesis call is made. + - `should_finalize(messages, text)` receives a documented conversation + view: this turn's TOOL messages are guaranteed to carry `tool_name`, + `tool_call_id`, and `tool_result`. To back that guarantee, + `tool_result` now defaults to the message content at every append site — + error results and policy denials populate it too (previously only + success-path messages did). + ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups Second 1.x minor, closing the loop on the v1.1.0 features from real diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 142084e..e210bff 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Public streaming + `should_finalize` contracts for `final_turn_only` + (#174).** Two previously source-only behaviors are now documented, typed + into docstrings, and pinned by dedicated contract tests + (`tests/agent/test_structured_streaming_contract.py`): + - `final_turn_only` + `astream()` NEVER emits the structured answer as + content chunks — on any path. New in this change: `single_pass` mode now + **suppresses** content chunks (they are the JSON envelope by construction + when the schema rides on every loop call), closing the leak that was + previously a documented caveat; tool-call chunks still stream. + `structured_synthesis_start` fires iff a separate synthesis call is made. + - `should_finalize(messages, text)` receives a documented conversation + view: this turn's TOOL messages are guaranteed to carry `tool_name`, + `tool_call_id`, and `tool_result`. To back that guarantee, + `tool_result` now defaults to the message content at every append site — + error results and policy denials populate it too (previously only + success-path messages did). + ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups Second 1.x minor, closing the loop on the v1.1.0 features from real diff --git a/docs/modules/AGENT.md b/docs/modules/AGENT.md index 69cd683..9f7d49c 100644 --- a/docs/modules/AGENT.md +++ b/docs/modules/AGENT.md @@ -1336,20 +1336,32 @@ The synthesis call is **conditional** (v1.2, #164/#166) — three paths avoid it Branch to a fallback deliberately on `"validation_failed"` instead of guessing from a bare `None`. -### Streaming and response_format +### Streaming and response_format — the public contract (v1.2.1, #174) + +These behaviors are load-bearing for chat surfaces, documented here as a +**tested public contract** (`tests/agent/test_structured_streaming_contract.py` +pins every clause): - **Default mode**: `astream()` streams the model's answer token-by-token, and with `response_format` set the answer IS the JSON — clients rendering chunks as chat text will see raw JSON fragments. Accumulate chunks and parse, or use `final_turn_only`. -- **`final_turn_only=True`**: the tool loop streams normally (prose chunks), - and the synthesis JSON is delivered ONLY via the terminal `AgentResult` - (`.content` / `.parsed`) — never leaked as content chunks. When a synthesis - call starts, `astream()` emits `StreamChunk(event="structured_synthesis_start")` - (no content) so clients can render a pending state. -- **`single_pass=True` caveat**: the final loop answer is the JSON itself, so - in `astream()` it streams as content chunks (like default mode) — the - trade-off for eliminating the synthesis round-trip. +- **`final_turn_only=True` guarantee**: the structured answer is NEVER emitted + as content chunks — on any path. It arrives only on the terminal + `AgentResult` (`.content` / `.parsed`). + - Without `single_pass`: the tool loop streams prose chunks normally; the + synthesis call is non-streaming; `StreamChunk(event="structured_synthesis_start")` + (content-free) fires **iff** a synthesis call is made — the reuse path and + `should_finalize` skips emit no event. + - With `single_pass`: content chunks are **suppressed** (the schema rides on + every loop call, so content is the JSON envelope by construction); + tool-call chunks still stream for activity display. +- **`should_finalize(messages, text)` contract**: `messages` is the run's + conversation view at convergence — prior history (with memory), this turn's + USER message(s), the ASSISTANT tool-call messages, and one + `Message(role=TOOL)` per executed tool with `tool_name`, `tool_call_id`, and + `tool_result` always populated (`tool_result` equals the result text the + model saw, including error/denial messages). Treat messages as read-only. ### Structured Retry Budget (v0.22.0 — BUG-34) diff --git a/src/selectools/agent/_tool_executor.py b/src/selectools/agent/_tool_executor.py index 448fe4c..bf9d482 100644 --- a/src/selectools/agent/_tool_executor.py +++ b/src/selectools/agent/_tool_executor.py @@ -784,12 +784,19 @@ def _append_tool_result( tool_result: Optional[str] = None, run_id: Optional[str] = None, ) -> None: - """Update history with tool output.""" + """Update history with tool output. + + ``tool_result`` defaults to ``tool_content`` so EVERY TOOL message — + success, error, or policy denial — carries a populated ``tool_result``. + This is a public contract for ``should_finalize`` consumers (#174); + callers pass an explicit ``tool_result`` only when the raw result + differs from the message content (e.g. compression). + """ tool_msg = Message( role=Role.TOOL, content=tool_content, tool_name=tool_name, - tool_result=tool_result, + tool_result=tool_result if tool_result is not None else tool_content, tool_call_id=tool_call_id, ) self._history.append(tool_msg) diff --git a/src/selectools/agent/config_groups.py b/src/selectools/agent/config_groups.py index 61b5213..1ed7f8f 100644 --- a/src/selectools/agent/config_groups.py +++ b/src/selectools/agent/config_groups.py @@ -213,16 +213,29 @@ class StructuredOutputConfig: answer wins before the predicate runs (even with ``reuse_loop_answer=False``). Like ``stop_condition``, an exception raised by the predicate propagates and aborts the run. - Default: None (always finalize, the v1.1 behavior). + + ``messages`` contract (public, tested — issue #174): the run's + conversation view at convergence, in chronological order — + prior history (when memory is configured), this turn's USER + message(s), each ASSISTANT message that requested tools + (``.tool_calls`` populated), and one ``Message(role=TOOL)`` per + executed tool with ``tool_name``, ``tool_call_id``, and + ``tool_result`` ALWAYS populated (``tool_result`` equals the + result text the model saw — success, error, or policy denial). + Treat the messages as read-only; mutating the list itself is + harmless. Default: None (always finalize, the v1.1 behavior). single_pass: In ``final_turn_only`` mode with a provider that advertises ``supports_native_structured_output_with_tools`` (OpenAI/Azure), carry the schema natively on the loop calls so the converged answer IS the structured object — combined with ``reuse_loop_answer`` this eliminates the synthesis call (v1.2, issue #166). Providers without combined support fall back to the - separate synthesis call. NOTE: in ``astream()`` the final answer - is then JSON and streams as content chunks (like plain native - mode). Default: False. + separate synthesis call. Streaming contract (public, tested — + issue #174): in ``astream()`` content chunks are SUPPRESSED in + this mode (with the schema riding on every loop call, content is + the JSON envelope by construction); tool-call chunks still + stream, and the structured answer arrives only on the terminal + ``AgentResult``. Default: False. """ native: bool = True diff --git a/src/selectools/agent/core.py b/src/selectools/agent/core.py index 510e1be..cd0fc25 100644 --- a/src/selectools/agent/core.py +++ b/src/selectools/agent/core.py @@ -2053,6 +2053,12 @@ async def astream( provider_extra: Dict[str, Any] = ( {"response_format": ctx.response_schema} if ctx.native_structured else {} ) + # single_pass streaming contract (#174): with the schema + # riding natively on loop calls, any content the model + # emits IS the structured JSON envelope — suppress it from + # the chunk stream; the answer arrives only on the + # terminal AgentResult. Tool-call chunks still stream. + suppress_content = ctx.native_structured and ctx.structured_final_only full_content = "" current_tool_calls: List[ToolCall] = [] @@ -2119,7 +2125,8 @@ async def astream( if _usage: self._notify_observers("on_usage", ctx.run_id, _usage) await self._anotify_observers("on_usage", ctx.run_id, _usage) - yield StreamChunk(content=_response_content) + if not suppress_content: + yield StreamChunk(content=_response_content) full_content = _response_content if response_msg.tool_calls: current_tool_calls = response_msg.tool_calls @@ -2144,7 +2151,8 @@ async def astream( ) as gen: async for item in gen: if isinstance(item, str): - yield StreamChunk(content=item) + if not suppress_content: + yield StreamChunk(content=item) full_content += item elif isinstance(item, ToolCall): current_tool_calls.append(item) diff --git a/tests/agent/test_structured_streaming_contract.py b/tests/agent/test_structured_streaming_contract.py new file mode 100644 index 0000000..b64cdfd --- /dev/null +++ b/tests/agent/test_structured_streaming_contract.py @@ -0,0 +1,250 @@ +"""Public contract tests for final_turn_only streaming + should_finalize (issue #174). + +Consumers of `final_turn_only` in streaming chat surfaces rely on two +behaviors that were previously source-only knowledge. These tests ARE the +contract — a refactor that breaks them breaks integrators: + +1. With ``final_turn_only=True``, ``astream()`` NEVER emits the structured + answer as content chunks — on any path. Non-single_pass: loop prose + streams, the synthesis call is non-streaming, and + ``StreamChunk(event="structured_synthesis_start")`` fires iff a synthesis + call is made. single_pass: content chunks are suppressed entirely (they + are the JSON envelope by construction), and the structured answer arrives + only on the terminal ``AgentResult``. +2. ``should_finalize(messages, last_response_text)`` receives the run's + conversation view at convergence, including this turn's TOOL messages + with ``tool_name`` and ``tool_result`` populated. +""" + +from __future__ import annotations + +import json +from typing import Any, List, Optional + +import pytest + +from selectools import Agent, AgentConfig, StructuredOutputConfig, tool +from selectools.types import AgentResult, Message, Role, StreamChunk, ToolCall +from tests.conftest import SharedFakeProvider + +from .test_structured_followups import ( # reuse the shared fakes + NativeWithToolsProvider, + _tool_call_message, +) + +SCHEMA = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], +} +JSON_ANSWER = '{"answer": "42"}' + + +@tool(description="A lookup tool.") +def lookup(query: str) -> str: + return json.dumps({"found": query}) + + +def _agent(provider: Any, **structured: Any) -> Agent: + return Agent( + [lookup], + provider=provider, + config=AgentConfig( + model="fake-model", + structured_output=StructuredOutputConfig(final_turn_only=True, **structured), + ), + ) + + +async def _collect(agent: Agent, **kwargs: Any) -> tuple: + chunks: List[StreamChunk] = [] + result: Optional[AgentResult] = None + async for item in agent.astream("go", **kwargs): + if isinstance(item, StreamChunk): + chunks.append(item) + else: + result = item + return chunks, result + + +class TestStreamingContract: + @pytest.mark.asyncio + async def test_single_pass_never_streams_json_content(self) -> None: + """single_pass + astream: the structured answer must NOT leak as + content chunks — it arrives only on the terminal AgentResult.""" + provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) + agent = _agent(provider, single_pass=True) + chunks, result = await _collect(agent, response_format=SCHEMA) + + streamed = "".join(c.content or "" for c in chunks) + assert "answer" not in streamed, f"JSON leaked into chunks: {streamed!r}" + assert result is not None + assert result.parsed == {"answer": "42"} + assert result.structured_status == "ok" + assert provider.calls == 2, "single_pass still avoids the synthesis call" + + @pytest.mark.asyncio + async def test_single_pass_still_streams_tool_call_chunks(self) -> None: + """Suppression covers content only — tool-call activity still streams.""" + provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) + agent = _agent(provider, single_pass=True) + chunks, _result = await _collect(agent, response_format=SCHEMA) + assert any(c.tool_calls for c in chunks), "tool-call chunks must still be visible" + + @pytest.mark.asyncio + async def test_synthesis_path_prose_streams_json_does_not(self) -> None: + """Non-single_pass: loop prose streams as chunks; the synthesis JSON + never appears in any chunk.""" + provider = SharedFakeProvider([_tool_call_message(), "Here is your answer.", JSON_ANSWER]) + agent = _agent(provider) + chunks, result = await _collect(agent, response_format=SCHEMA) + + streamed = "".join(c.content or "" for c in chunks) + assert "Here is your answer." in streamed + assert JSON_ANSWER not in streamed + assert result is not None and result.parsed == {"answer": "42"} + + @pytest.mark.asyncio + async def test_synthesis_start_event_fires_iff_synthesis_call_made(self) -> None: + # Case A: prose answer -> synthesis call -> event fires + provider = SharedFakeProvider([_tool_call_message(), "prose", JSON_ANSWER]) + agent = _agent(provider) + chunks, _ = await _collect(agent, response_format=SCHEMA) + assert sum(1 for c in chunks if c.event == "structured_synthesis_start") == 1 + + # Case B: reuse path (loop answer validates) -> no synthesis -> no event + provider = SharedFakeProvider([_tool_call_message(), JSON_ANSWER]) + agent = _agent(provider) + chunks, _ = await _collect(agent, response_format=SCHEMA) + assert all(c.event != "structured_synthesis_start" for c in chunks) + + # Case C: single_pass -> no synthesis call -> no event + provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) + agent = _agent(provider, single_pass=True) + chunks, _ = await _collect(agent, response_format=SCHEMA) + assert all(c.event != "structured_synthesis_start" for c in chunks) + + @pytest.mark.asyncio + async def test_event_chunks_carry_no_content(self) -> None: + provider = SharedFakeProvider([_tool_call_message(), "prose", JSON_ANSWER]) + agent = _agent(provider) + chunks, _ = await _collect(agent, response_format=SCHEMA) + for c in chunks: + if c.event: + assert not c.content, "event chunks must be content-free" + + +class TestShouldFinalizeMessagesContract: + def test_predicate_sees_tool_messages_with_name_and_result(self) -> None: + seen: List[Message] = [] + + def predicate(messages: List[Message], text: str) -> bool: + seen.extend(messages) + return False + + provider = SharedFakeProvider([_tool_call_message(), "converged prose"]) + agent = _agent(provider, should_finalize=predicate) + agent.run("go", response_format=SCHEMA) + + tool_msgs = [m for m in seen if m.role == Role.TOOL] + assert tool_msgs, "this turn's tool results must be visible to the predicate" + for m in tool_msgs: + assert m.tool_name == "lookup" + assert m.tool_result, "tool_result must be populated on TOOL messages" + assert json.loads(m.tool_result) == {"found": "x"} + + @pytest.mark.asyncio + async def test_predicate_sees_tool_messages_in_astream(self) -> None: + seen: List[Message] = [] + + def predicate(messages: List[Message], text: str) -> bool: + seen.extend(messages) + return False + + provider = SharedFakeProvider([_tool_call_message(), "converged prose"]) + agent = _agent(provider, should_finalize=predicate) + async for _item in agent.astream("go", response_format=SCHEMA): + pass + + tool_msgs = [m for m in seen if m.role == Role.TOOL] + assert tool_msgs + assert all(m.tool_name == "lookup" and m.tool_result for m in tool_msgs) + + def test_predicate_sees_user_and_assistant_turns(self) -> None: + seen: List[Message] = [] + + def predicate(messages: List[Message], text: str) -> bool: + seen.extend(messages) + return False + + provider = SharedFakeProvider([_tool_call_message(), "converged prose"]) + agent = _agent(provider, should_finalize=predicate) + agent.run("find x please", response_format=SCHEMA) + + roles = [m.role for m in seen] + assert Role.USER in roles + assert Role.ASSISTANT in roles, "the tool_call assistant message is part of the view" + user_msgs = [m for m in seen if m.role == Role.USER] + assert any("find x please" in (m.content or "") for m in user_msgs) + + +class TestToolResultFieldInvariant: + def test_every_tool_message_in_history_has_tool_result(self) -> None: + """The invariant behind the predicate contract: TOOL messages always + carry tool_result (defaulting to their content).""" + first = Message( + role=Role.ASSISTANT, + content="", + tool_calls=[ToolCall(tool_name="lookup", parameters={"query": "x"}, id="t1")], + ) + provider = SharedFakeProvider([first, "Done"]) + agent = Agent([lookup], provider=provider, config=AgentConfig(model="fake-model")) + agent.run("go") + tool_msgs = [m for m in agent._history if m.role == Role.TOOL] + assert tool_msgs + for m in tool_msgs: + assert m.tool_name + assert m.tool_result is not None + + def test_parallel_path_tool_messages_have_tool_result(self) -> None: + @tool(description="Second lookup.") + def lookup_b(query: str) -> str: + return f"b:{query}" + + first = Message( + role=Role.ASSISTANT, + content="", + tool_calls=[ + ToolCall(tool_name="lookup", parameters={"query": "x"}, id="t1"), + ToolCall(tool_name="lookup_b", parameters={"query": "y"}, id="t2"), + ], + ) + provider = SharedFakeProvider([first, "Done"]) + agent = Agent( + [lookup, lookup_b], + provider=provider, + config=AgentConfig(model="fake-model", parallel_tool_execution=True), + ) + agent.run("go") + tool_msgs = [m for m in agent._history if m.role == Role.TOOL] + assert len(tool_msgs) == 2 + for m in tool_msgs: + assert m.tool_result is not None, f"{m.tool_name} missing tool_result" + + def test_error_tool_messages_also_carry_tool_result(self) -> None: + @tool(description="Always fails.") + def broken(query: str) -> str: + raise ValueError("boom") + + first = Message( + role=Role.ASSISTANT, + content="", + tool_calls=[ToolCall(tool_name="broken", parameters={"query": "x"}, id="t1")], + ) + provider = SharedFakeProvider([first, "Done"]) + agent = Agent([broken], provider=provider, config=AgentConfig(model="fake-model")) + agent.run("go") + tool_msgs = [m for m in agent._history if m.role == Role.TOOL] + assert tool_msgs + for m in tool_msgs: + assert m.tool_result is not None, "error results must also populate tool_result" From d0a27a3f8b0d437855972bb64635376152fb2cb0 Mon Sep 17 00:00:00 2001 From: John Niche Date: Tue, 14 Jul 2026 17:58:57 -0300 Subject: [PATCH 2/3] fix: honest streaming contract (reuse-path signal), legacy-session normalization Self-review of the #174 PR (high-effort multi-agent review) caught the contract overpromising; all findings addressed: - The "structured answer NEVER streams on any path" claim was FALSE on the reuse path: without single_pass the loop streams model output as it always has, so a converged answer that validates has already streamed before reuse. The contract now states this precisely, and a new StreamChunk(event="structured_reuse") fires at the reuse decision so chat surfaces can convert/retract the just-streamed bubble. Pinned honestly in the contract tests (the reuse-leak assertion the old test avoided is now present). - tool_call_id was documented as always populated - false for text-parsed tool calls (ToolCallParser creates id=None). Contract corrected: tool_name + tool_result are the guaranteed pair; tool_call_id only for native tool-calling providers. - "tool_result ALWAYS populated" was false for histories persisted before this change (error/denial rows stored None). Message.from_dict now normalizes TOOL messages on load, with a test. - Changelog restructured: single_pass chunk suppression and the tool_result unification moved under "### Changed" with migration notes (read AgentResult.parsed; match denial/error text instead of tool_result is None). - _append_tool_result's now-redundant tool_result parameter removed (every call site passed content); policy-denial invariant test added; single_pass mode named once as ctx.single_pass_active instead of three inline reconstructions; contract test file made self-contained (no cross-test-module imports); version pin dropped from the AGENT.md section header. Claude-Session: https://claude.ai/code/session_01L1JWChKMs9yTyK92qquhJa --- CHANGELOG.md | 39 +++++--- docs/CHANGELOG.md | 39 +++++--- docs/modules/AGENT.md | 41 +++++---- src/selectools/agent/_tool_executor.py | 18 ++-- src/selectools/agent/config_groups.py | 20 +++-- src/selectools/agent/core.py | 16 +++- src/selectools/types.py | 15 +++- tests/agent/test_structured_followups.py | 7 +- .../test_structured_streaming_contract.py | 90 ++++++++++++++++--- 9 files changed, 200 insertions(+), 85 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e210bff..e2bf647 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,21 +10,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **Public streaming + `should_finalize` contracts for `final_turn_only` - (#174).** Two previously source-only behaviors are now documented, typed - into docstrings, and pinned by dedicated contract tests + (#174).** Previously source-only behaviors are now documented, stated in + docstrings, and pinned by dedicated contract tests (`tests/agent/test_structured_streaming_contract.py`): - - `final_turn_only` + `astream()` NEVER emits the structured answer as - content chunks — on any path. New in this change: `single_pass` mode now - **suppresses** content chunks (they are the JSON envelope by construction - when the schema rides on every loop call), closing the leak that was - previously a documented caveat; tool-call chunks still stream. - `structured_synthesis_start` fires iff a separate synthesis call is made. - - `should_finalize(messages, text)` receives a documented conversation - view: this turn's TOOL messages are guaranteed to carry `tool_name`, - `tool_call_id`, and `tool_result`. To back that guarantee, - `tool_result` now defaults to the message content at every append site — - error results and policy denials populate it too (previously only - success-path messages did). + - The synthesis call never streams — its JSON arrives only on the terminal + `AgentResult`; `structured_synthesis_start` fires iff a synthesis call is + made. + - New `StreamChunk(event="structured_reuse")`: fired when the converged + loop answer validates and is reused — the just-streamed loop chunks WERE + the structured answer, so chat surfaces can convert/retract the bubble. + - `should_finalize(messages, text)` message-view contract documented: TOOL + messages always carry `tool_name` and `tool_result`; `tool_call_id` only + for native tool-calling providers. + +### Changed + +- **`single_pass` + `astream()` no longer streams the structured answer as + content chunks** — content is suppressed (it is the JSON envelope by + construction when the schema rides on every loop call); tool-call chunks + still stream. Migration: read the answer from `AgentResult.parsed` / + `.content` instead of accumulating content chunks. (#174) +- **`Message.tool_result` is now always populated on TOOL messages** — + success, error, and policy-denial results all carry the text the model saw + (previously error/denial messages had `tool_result=None`). Sessions + persisted by older versions are normalized on load in `Message.from_dict`. + If you used `tool_result is None` to detect failed tools, match on the + `Error executing tool` / denial message text instead. (#174) ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e210bff..e2bf647 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -10,21 +10,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **Public streaming + `should_finalize` contracts for `final_turn_only` - (#174).** Two previously source-only behaviors are now documented, typed - into docstrings, and pinned by dedicated contract tests + (#174).** Previously source-only behaviors are now documented, stated in + docstrings, and pinned by dedicated contract tests (`tests/agent/test_structured_streaming_contract.py`): - - `final_turn_only` + `astream()` NEVER emits the structured answer as - content chunks — on any path. New in this change: `single_pass` mode now - **suppresses** content chunks (they are the JSON envelope by construction - when the schema rides on every loop call), closing the leak that was - previously a documented caveat; tool-call chunks still stream. - `structured_synthesis_start` fires iff a separate synthesis call is made. - - `should_finalize(messages, text)` receives a documented conversation - view: this turn's TOOL messages are guaranteed to carry `tool_name`, - `tool_call_id`, and `tool_result`. To back that guarantee, - `tool_result` now defaults to the message content at every append site — - error results and policy denials populate it too (previously only - success-path messages did). + - The synthesis call never streams — its JSON arrives only on the terminal + `AgentResult`; `structured_synthesis_start` fires iff a synthesis call is + made. + - New `StreamChunk(event="structured_reuse")`: fired when the converged + loop answer validates and is reused — the just-streamed loop chunks WERE + the structured answer, so chat surfaces can convert/retract the bubble. + - `should_finalize(messages, text)` message-view contract documented: TOOL + messages always carry `tool_name` and `tool_result`; `tool_call_id` only + for native tool-calling providers. + +### Changed + +- **`single_pass` + `astream()` no longer streams the structured answer as + content chunks** — content is suppressed (it is the JSON envelope by + construction when the schema rides on every loop call); tool-call chunks + still stream. Migration: read the answer from `AgentResult.parsed` / + `.content` instead of accumulating content chunks. (#174) +- **`Message.tool_result` is now always populated on TOOL messages** — + success, error, and policy-denial results all carry the text the model saw + (previously error/denial messages had `tool_result=None`). Sessions + persisted by older versions are normalized on load in `Message.from_dict`. + If you used `tool_result is None` to detect failed tools, match on the + `Error executing tool` / denial message text instead. (#174) ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups diff --git a/docs/modules/AGENT.md b/docs/modules/AGENT.md index 9f7d49c..aeef343 100644 --- a/docs/modules/AGENT.md +++ b/docs/modules/AGENT.md @@ -1336,7 +1336,7 @@ The synthesis call is **conditional** (v1.2, #164/#166) — three paths avoid it Branch to a fallback deliberately on `"validation_failed"` instead of guessing from a bare `None`. -### Streaming and response_format — the public contract (v1.2.1, #174) +### Streaming and response_format — the public contract (#174) These behaviors are load-bearing for chat surfaces, documented here as a **tested public contract** (`tests/agent/test_structured_streaming_contract.py` @@ -1346,22 +1346,31 @@ pins every clause): with `response_format` set the answer IS the JSON — clients rendering chunks as chat text will see raw JSON fragments. Accumulate chunks and parse, or use `final_turn_only`. -- **`final_turn_only=True` guarantee**: the structured answer is NEVER emitted - as content chunks — on any path. It arrives only on the terminal - `AgentResult` (`.content` / `.parsed`). - - Without `single_pass`: the tool loop streams prose chunks normally; the - synthesis call is non-streaming; `StreamChunk(event="structured_synthesis_start")` - (content-free) fires **iff** a synthesis call is made — the reuse path and - `should_finalize` skips emit no event. - - With `single_pass`: content chunks are **suppressed** (the schema rides on - every loop call, so content is the JSON envelope by construction); - tool-call chunks still stream for activity display. +- **`final_turn_only=True` guarantees**: + - The **synthesis call never streams**: its JSON arrives only on the + terminal `AgentResult` (`.content` / `.parsed`), and + `StreamChunk(event="structured_synthesis_start")` (content-free) fires + **iff** a synthesis call is made. + - With `single_pass`: content chunks are **suppressed entirely** (the schema + rides on every loop call, so content is the JSON envelope by + construction); tool-call chunks still stream for activity display. + - **Reuse caveat, signaled**: without `single_pass`, the loop streams model + output as it always has — so when the converged loop answer itself + validates and is reused (`reuse_loop_answer=True`), that answer has + already streamed as ordinary loop chunks. + `StreamChunk(event="structured_reuse")` fires at that moment so a chat + surface can convert or retract the just-streamed bubble. Clients that + must never display raw JSON should handle this event (or use + `single_pass` on supporting providers, where suppression is total). - **`should_finalize(messages, text)` contract**: `messages` is the run's - conversation view at convergence — prior history (with memory), this turn's - USER message(s), the ASSISTANT tool-call messages, and one - `Message(role=TOOL)` per executed tool with `tool_name`, `tool_call_id`, and - `tool_result` always populated (`tool_result` equals the result text the - model saw, including error/denial messages). Treat messages as read-only. + conversation view at convergence — prior history (with memory, including + restored sessions), this turn's USER message(s), the ASSISTANT tool-call + messages, and one `Message(role=TOOL)` per executed tool with `tool_name` + and `tool_result` always populated (`tool_result` equals the result text + the model saw, including error/denial messages; legacy persisted messages + are normalized on load). `tool_call_id` is populated for native + tool-calling providers but `None` for text-parsed tool calls. Treat + messages as read-only. ### Structured Retry Budget (v0.22.0 — BUG-34) diff --git a/src/selectools/agent/_tool_executor.py b/src/selectools/agent/_tool_executor.py index bf9d482..ad1f27d 100644 --- a/src/selectools/agent/_tool_executor.py +++ b/src/selectools/agent/_tool_executor.py @@ -781,22 +781,20 @@ def _append_tool_result( tool_content: str, tool_name: str, tool_call_id: Optional[str] = None, - tool_result: Optional[str] = None, run_id: Optional[str] = None, ) -> None: """Update history with tool output. - ``tool_result`` defaults to ``tool_content`` so EVERY TOOL message — - success, error, or policy denial — carries a populated ``tool_result``. - This is a public contract for ``should_finalize`` consumers (#174); - callers pass an explicit ``tool_result`` only when the raw result - differs from the message content (e.g. compression). + ``tool_result`` is always set to ``tool_content`` so EVERY TOOL + message — success, error, or policy denial — carries a populated + ``tool_result`` equal to the text the model saw. This is a public + contract for ``should_finalize`` consumers (#174). """ tool_msg = Message( role=Role.TOOL, content=tool_content, tool_name=tool_name, - tool_result=tool_result if tool_result is not None else tool_content, + tool_result=tool_content, tool_call_id=tool_call_id, ) self._history.append(tool_msg) @@ -1040,7 +1038,6 @@ def chunk_cb(chunk: str) -> None: message_content, r.tool_call.tool_name, r.tool_call.id, - tool_result=message_content, run_id=run_id, ) @@ -1318,7 +1315,6 @@ def chunk_cb(chunk: str) -> None: message_content, r.tool_call.tool_name, r.tool_call.id, - tool_result=message_content, run_id=run_id, ) @@ -1488,7 +1484,6 @@ def _execute_single_tool(self, ctx: _RunContext, tool_call: ToolCall) -> bool: cached_content, tool_name, tool_call.id, - tool_result=cached_content, run_id=ctx.run_id, ) if getattr(tool, "terminal", False) or self._stop_condition_hit(tool_name, cached_raw): @@ -1604,7 +1599,6 @@ def chunk_callback(chunk: str) -> None: message_content, tool_name, tool_call.id, - tool_result=message_content, run_id=ctx.run_id, ) @@ -1732,7 +1726,6 @@ async def _aexecute_single_tool(self, ctx: _RunContext, tool_call: ToolCall) -> cached_content, tool_name, tool_call.id, - tool_result=cached_content, run_id=ctx.run_id, ) if getattr(tool, "terminal", False) or self._stop_condition_hit(tool_name, cached_raw): @@ -1865,7 +1858,6 @@ def chunk_callback(chunk: str) -> None: message_content, tool_name, tool_call.id, - tool_result=message_content, run_id=ctx.run_id, ) diff --git a/src/selectools/agent/config_groups.py b/src/selectools/agent/config_groups.py index 1ed7f8f..af709d9 100644 --- a/src/selectools/agent/config_groups.py +++ b/src/selectools/agent/config_groups.py @@ -216,14 +216,18 @@ class StructuredOutputConfig: ``messages`` contract (public, tested — issue #174): the run's conversation view at convergence, in chronological order — - prior history (when memory is configured), this turn's USER - message(s), each ASSISTANT message that requested tools - (``.tool_calls`` populated), and one ``Message(role=TOOL)`` per - executed tool with ``tool_name``, ``tool_call_id``, and - ``tool_result`` ALWAYS populated (``tool_result`` equals the - result text the model saw — success, error, or policy denial). - Treat the messages as read-only; mutating the list itself is - harmless. Default: None (always finalize, the v1.1 behavior). + prior history (when memory is configured, including restored + sessions), this turn's USER message(s), the ASSISTANT messages + that requested tools, and one ``Message(role=TOOL)`` per + executed tool. On every TOOL message, ``tool_name`` and + ``tool_result`` are ALWAYS populated (``tool_result`` equals + the result text the model saw — success, error, or policy + denial; legacy persisted messages are normalized on load). + ``tool_call_id`` is populated for native tool-calling providers + but is ``None`` for text-parsed tool calls — key by it only if + you know your provider supplies ids. Treat the messages as + read-only; mutating the list itself is harmless. Default: None + (always finalize, the v1.1 behavior). single_pass: In ``final_turn_only`` mode with a provider that advertises ``supports_native_structured_output_with_tools`` (OpenAI/Azure), carry the schema natively on the loop calls so diff --git a/src/selectools/agent/core.py b/src/selectools/agent/core.py index cd0fc25..0779bf4 100644 --- a/src/selectools/agent/core.py +++ b/src/selectools/agent/core.py @@ -100,6 +100,9 @@ class _RunContext: structured_final_only: bool = False structured_final_pending: bool = False native_final: bool = False + # single_pass mode resolved once at run setup: the schema rides natively + # on loop calls AND final-turn-only semantics apply (issue #166/#174). + single_pass_active: bool = False # Budget allowance for the synthesis turn: set to 1 when the transition # into structured_final_pending consumes an iteration. structured_extra_turns: int = 0 @@ -582,6 +585,7 @@ def _prepare_run( native_structured=native_structured, structured_final_only=structured_final_only, native_final=native_final, + single_pass_active=native_structured and structured_final_only, ) @staticmethod @@ -862,7 +866,7 @@ def _process_response( # single_pass converged answers are native JSON — the text # parser's whole-text fallback would hijack schemas whose keys # look like a tool call (name/parameters). - and not ctx.native_structured + and not ctx.single_pass_active ): parse_result = self.parser.parse(response_text) if parse_result.tool_call: @@ -912,7 +916,7 @@ async def _aprocess_response( # single_pass converged answers are native JSON — the text # parser's whole-text fallback would hijack schemas whose keys # look like a tool call (name/parameters). - and not ctx.native_structured + and not ctx.single_pass_active ): parse_result = self.parser.parse(response_text) if parse_result.tool_call: @@ -2058,7 +2062,7 @@ async def astream( # emits IS the structured JSON envelope — suppress it from # the chunk stream; the answer arrives only on the # terminal AgentResult. Tool-call chunks still stream. - suppress_content = ctx.native_structured and ctx.structured_final_only + suppress_content = ctx.single_pass_active full_content = "" current_tool_calls: List[ToolCall] = [] @@ -2191,6 +2195,12 @@ async def astream( decision = self._resolve_structured_final(ctx, response_text) if decision == "skip": ctx.structured_skipped = True + elif decision == "reuse" and not ctx.single_pass_active: + # The converged loop answer validated and just streamed + # as ordinary loop content — signal clients that those + # chunks WERE the structured answer so a chat surface + # can convert/retract the bubble (#174). + yield StreamChunk(event="structured_reuse") elif decision == "synthesize": self._begin_structured_final_turn(ctx, response_text) self._notify_observers( diff --git a/src/selectools/types.py b/src/selectools/types.py index 87a2543..269156b 100644 --- a/src/selectools/types.py +++ b/src/selectools/types.py @@ -233,6 +233,12 @@ def from_dict(cls, data: Dict[str, Any]) -> "Message": msg.image_base64 = data.get("image_base64") msg.tool_name = data.get("tool_name") msg.tool_result = data.get("tool_result") + # Sessions persisted before v1.2.x stored tool_result=None on + # error/denial TOOL messages; normalize so the should_finalize + # contract (tool_result always populated on TOOL messages, #174) + # holds for restored histories too. + if msg.role == Role.TOOL and msg.tool_result is None: + msg.tool_result = msg.content msg.tool_call_id = data.get("tool_call_id") raw_calls = data.get("tool_calls") if raw_calls: @@ -385,10 +391,15 @@ class StreamChunk: content: The text content of this chunk (delta). role: The role emitting this chunk (usually ASSISTANT). tool_calls: Optional list of tool calls (if this chunk contains tool invocations). - event: Optional lifecycle signal (v1.2). Currently emitted: + event: Optional lifecycle signal (v1.2). Event chunks carry no + content. Currently emitted: ``"structured_synthesis_start"`` — the tool loop converged and a structured synthesis call is starting; clients can render a - pending state. Event chunks carry no content. + pending state. + ``"structured_reuse"`` — the converged loop answer validated and + was reused as the structured answer; the chunks already streamed + for this turn WERE that answer, so chat surfaces can convert or + retract the bubble (#174). """ content: str = "" diff --git a/tests/agent/test_structured_followups.py b/tests/agent/test_structured_followups.py index fb401b1..df5b8ab 100644 --- a/tests/agent/test_structured_followups.py +++ b/tests/agent/test_structured_followups.py @@ -275,14 +275,17 @@ async def test_astream_emits_synthesis_start_event(self) -> None: assert "structured_synthesis_start" in events @pytest.mark.asyncio - async def test_no_event_when_synthesis_skipped(self) -> None: + async def test_no_synthesis_event_when_synthesis_skipped(self) -> None: provider = SharedFakeProvider([_tool_call_message(), JSON_ANSWER]) agent = _final_turn_agent(provider) events: List[str] = [] async for chunk in agent.astream("go", response_format=SCHEMA): if isinstance(chunk, StreamChunk) and chunk.event: events.append(chunk.event) - assert events == [] + assert "structured_synthesis_start" not in events + # the reuse path signals that the streamed loop answer WAS the + # structured answer (#174) + assert events == ["structured_reuse"] # ── Review follow-ups (PR #169 self-review) ───────────────────────────── diff --git a/tests/agent/test_structured_streaming_contract.py b/tests/agent/test_structured_streaming_contract.py index b64cdfd..3427ec8 100644 --- a/tests/agent/test_structured_streaming_contract.py +++ b/tests/agent/test_structured_streaming_contract.py @@ -4,13 +4,12 @@ behaviors that were previously source-only knowledge. These tests ARE the contract — a refactor that breaks them breaks integrators: -1. With ``final_turn_only=True``, ``astream()`` NEVER emits the structured - answer as content chunks — on any path. Non-single_pass: loop prose - streams, the synthesis call is non-streaming, and - ``StreamChunk(event="structured_synthesis_start")`` fires iff a synthesis - call is made. single_pass: content chunks are suppressed entirely (they - are the JSON envelope by construction), and the structured answer arrives - only on the terminal ``AgentResult``. +1. With ``final_turn_only=True``: the SYNTHESIS call never streams + (``structured_synthesis_start`` fires iff it happens); ``single_pass`` + suppresses content chunks entirely; and on the reuse path — where the + converged loop answer already streamed as ordinary loop chunks — + ``StreamChunk(event="structured_reuse")`` signals that those chunks WERE + the structured answer. 2. ``should_finalize(messages, last_response_text)`` receives the run's conversation view at convergence, including this turn's TOOL messages with ``tool_name`` and ``tool_result`` populated. @@ -25,12 +24,37 @@ from selectools import Agent, AgentConfig, StructuredOutputConfig, tool from selectools.types import AgentResult, Message, Role, StreamChunk, ToolCall +from selectools.usage import UsageStats from tests.conftest import SharedFakeProvider -from .test_structured_followups import ( # reuse the shared fakes - NativeWithToolsProvider, - _tool_call_message, -) + +class NativeWithToolsProvider(SharedFakeProvider): + """Fake advertising tools+json_schema support; accepts response_format.""" + + supports_native_structured_output = True + supports_native_structured_output_with_tools = True + + def complete(self, **kwargs: Any) -> "tuple[Message, UsageStats]": + kwargs.pop("response_format", None) + return super().complete(**kwargs) + + async def acomplete(self, **kwargs: Any) -> "tuple[Message, UsageStats]": + kwargs.pop("response_format", None) + return await super().acomplete(**kwargs) + + async def astream(self, **kwargs: Any): + kwargs.pop("response_format", None) + async for chunk in super().astream(**kwargs): + yield chunk + + +def _tool_call_message() -> Message: + return Message( + role=Role.ASSISTANT, + content="", + tool_calls=[ToolCall(tool_name="lookup", parameters={"query": "x"}, id="tc1")], + ) + SCHEMA = { "type": "object", @@ -112,11 +136,17 @@ async def test_synthesis_start_event_fires_iff_synthesis_call_made(self) -> None chunks, _ = await _collect(agent, response_format=SCHEMA) assert sum(1 for c in chunks if c.event == "structured_synthesis_start") == 1 - # Case B: reuse path (loop answer validates) -> no synthesis -> no event + # Case B: reuse path (loop answer validates) -> no synthesis event, + # but the structured_reuse signal fires — and the answer HAS already + # streamed as ordinary loop chunks (documented caveat, pinned honest). provider = SharedFakeProvider([_tool_call_message(), JSON_ANSWER]) agent = _agent(provider) - chunks, _ = await _collect(agent, response_format=SCHEMA) + chunks, result = await _collect(agent, response_format=SCHEMA) assert all(c.event != "structured_synthesis_start" for c in chunks) + assert sum(1 for c in chunks if c.event == "structured_reuse") == 1 + streamed = "".join(c.content or "" for c in chunks) + assert JSON_ANSWER in streamed, "reuse-path answer streams as loop output" + assert result is not None and result.parsed == {"answer": "42"} # Case C: single_pass -> no synthesis call -> no event provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) @@ -231,6 +261,40 @@ def lookup_b(query: str) -> str: for m in tool_msgs: assert m.tool_result is not None, f"{m.tool_name} missing tool_result" + def test_policy_denied_tool_messages_carry_tool_result(self) -> None: + from selectools.policy import ToolPolicy + + first = Message( + role=Role.ASSISTANT, + content="", + tool_calls=[ToolCall(tool_name="lookup", parameters={"query": "x"}, id="t1")], + ) + provider = SharedFakeProvider([first, "Done"]) + agent = Agent( + [lookup], + provider=provider, + config=AgentConfig(model="fake-model", tool_policy=ToolPolicy(deny=["lookup"])), + ) + agent.run("go") + tool_msgs = [m for m in agent._history if m.role == Role.TOOL] + assert tool_msgs, "the denial must still produce a TOOL message" + for m in tool_msgs: + assert m.tool_result is not None, "policy denials must populate tool_result" + + def test_legacy_persisted_tool_messages_normalized_on_load(self) -> None: + """Sessions saved before v1.2.x stored tool_result=None on error + TOOL messages; from_dict must normalize so the contract holds for + restored histories.""" + legacy = { + "role": "tool", + "content": "Error executing tool 'x': boom", + "tool_name": "x", + "tool_result": None, + "tool_call_id": "t1", + } + restored = Message.from_dict(legacy) + assert restored.tool_result == "Error executing tool 'x': boom" + def test_error_tool_messages_also_carry_tool_result(self) -> None: @tool(description="Always fails.") def broken(query: str) -> str: From 30edc6c15ebca6ac79cc4038316c159a75297d7a Mon Sep 17 00:00:00 2001 From: John Niche Date: Tue, 14 Jul 2026 20:20:03 -0300 Subject: [PATCH 3/3] fix: enforce the streaming contract on every surface; sharpen tool_result semantics Max-effort self-review (39 agents, 15 verified findings incl. nits) showed the #174 contract was only implemented on the astream() chunk surface. All findings addressed: - stream_handler (run/arun): the synthesis turn and single_pass loop turns no longer feed the token callback - the second streaming surface now honors the same suppression contract (mutation-verified tests). - astream() non-streaming-provider fallback: tool-call chunks are now yielded there too, so single_pass suppression cannot produce a zero- chunk stream on providers without native async streaming (mutation-verified after fixing a fake whose class attribute was shadowed by an instance attribute). - serve SSE: content-free event chunks are forwarded as {"type": "event", "event": ...} records instead of being dropped - the documented convert/retract flow is reachable through selectools.serve (test added). - tool_result unification REVERTED in favor of the sharper v1.2-compatible contract: success -> tool_result carries the result text; error/policy/ coherence denial -> None (the structural failure discriminator, now documented as such). from_dict round-trip fidelity restored (@stable API no longer rewrites persisted data); caller-seeded TOOL messages scoped out of the guarantee; Message docstring updated. - Contract wording corrected everywhere: synthesis event is once-per-run (validation retries do not re-emit); structured_reuse qualified as non-single_pass in all three doc surfaces; SYSTEM context injections added to the messages enumeration; Since markers added (v1.3.0); changelog rewritten with an explicit Changed section scoped to the @beta single_pass surface; llms-full.txt regenerated. - Case C pins that structured_reuse is NOT emitted in single_pass mode. Pre-existing, unrelated: 5 starlette auth tests fail on clean main when starlette is installed (zero CI coverage for that surface) - filed #176. Claude-Session: https://claude.ai/code/session_01L1JWChKMs9yTyK92qquhJa --- CHANGELOG.md | 49 +++++----- docs/CHANGELOG.md | 49 +++++----- docs/llms-full.txt | 51 ++++++++-- docs/modules/AGENT.md | 48 ++++++---- src/selectools/agent/_tool_executor.py | 19 +++- src/selectools/agent/config_groups.py | 26 ++--- src/selectools/agent/core.py | 16 +++- src/selectools/serve/api.py | 7 ++ src/selectools/types.py | 33 ++++--- .../test_structured_streaming_contract.py | 94 ++++++++++++++++--- tests/test_agent_api.py | 25 +++++ 11 files changed, 300 insertions(+), 117 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e2bf647..42b987e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,30 +12,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Public streaming + `should_finalize` contracts for `final_turn_only` (#174).** Previously source-only behaviors are now documented, stated in docstrings, and pinned by dedicated contract tests - (`tests/agent/test_structured_streaming_contract.py`): - - The synthesis call never streams — its JSON arrives only on the terminal - `AgentResult`; `structured_synthesis_start` fires iff a synthesis call is - made. - - New `StreamChunk(event="structured_reuse")`: fired when the converged - loop answer validates and is reused — the just-streamed loop chunks WERE - the structured answer, so chat surfaces can convert/retract the bubble. - - `should_finalize(messages, text)` message-view contract documented: TOOL - messages always carry `tool_name` and `tool_result`; `tool_call_id` only - for native tool-calling providers. + (`tests/agent/test_structured_streaming_contract.py`). The contract covers + all consumption surfaces: `astream()` chunks, the `stream_handler` + callback in `run()`/`arun()`, and the `selectools.serve` SSE stream: + - The synthesis turn never streams — not as chunks and not through + `stream_handler`; its JSON arrives only on the terminal `AgentResult`. + `structured_synthesis_start` fires once per run when the synthesis turn + begins (validation retries within the turn do not re-emit it). + - New `StreamChunk(event="structured_reuse")`: fired (outside + `single_pass`) when the converged loop answer validates and is reused — + the just-streamed loop chunks WERE the structured answer, so chat + surfaces can convert/retract the bubble. + - The serve SSE endpoint forwards content-free event chunks as + `{"type": "event", "event": ...}` records (previously dropped). + - `should_finalize(messages, text)` message-view contract documented: + executor-appended TOOL messages always carry `tool_name`; + `tool_result` carries the result text for successful executions and is + `None` for errors/policy denials — the supported discriminator for + "this tool produced a real result". `tool_call_id` only for native + tool-calling providers. ### Changed -- **`single_pass` + `astream()` no longer streams the structured answer as - content chunks** — content is suppressed (it is the JSON envelope by - construction when the schema rides on every loop call); tool-call chunks - still stream. Migration: read the answer from `AgentResult.parsed` / - `.content` instead of accumulating content chunks. (#174) -- **`Message.tool_result` is now always populated on TOOL messages** — - success, error, and policy-denial results all carry the text the model saw - (previously error/denial messages had `tool_result=None`). Sessions - persisted by older versions are normalized on load in `Message.from_dict`. - If you used `tool_result is None` to detect failed tools, match on the - `Error executing tool` / denial message text instead. (#174) +- **`single_pass` structured output no longer streams the answer** (beta + surface): in `astream()` content chunks are suppressed and in + `run()`/`arun()` the `stream_handler` callback is not invoked for + single_pass loop turns or the synthesis turn — content on those calls is + the JSON envelope by construction. Tool-call chunks still stream + (including on providers without native async streaming, which previously + emitted no tool-call chunks at all). Migration: read the answer from + `AgentResult.parsed` / `.content` instead of accumulating streamed + content. (#174) ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e2bf647..42b987e 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -12,30 +12,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Public streaming + `should_finalize` contracts for `final_turn_only` (#174).** Previously source-only behaviors are now documented, stated in docstrings, and pinned by dedicated contract tests - (`tests/agent/test_structured_streaming_contract.py`): - - The synthesis call never streams — its JSON arrives only on the terminal - `AgentResult`; `structured_synthesis_start` fires iff a synthesis call is - made. - - New `StreamChunk(event="structured_reuse")`: fired when the converged - loop answer validates and is reused — the just-streamed loop chunks WERE - the structured answer, so chat surfaces can convert/retract the bubble. - - `should_finalize(messages, text)` message-view contract documented: TOOL - messages always carry `tool_name` and `tool_result`; `tool_call_id` only - for native tool-calling providers. + (`tests/agent/test_structured_streaming_contract.py`). The contract covers + all consumption surfaces: `astream()` chunks, the `stream_handler` + callback in `run()`/`arun()`, and the `selectools.serve` SSE stream: + - The synthesis turn never streams — not as chunks and not through + `stream_handler`; its JSON arrives only on the terminal `AgentResult`. + `structured_synthesis_start` fires once per run when the synthesis turn + begins (validation retries within the turn do not re-emit it). + - New `StreamChunk(event="structured_reuse")`: fired (outside + `single_pass`) when the converged loop answer validates and is reused — + the just-streamed loop chunks WERE the structured answer, so chat + surfaces can convert/retract the bubble. + - The serve SSE endpoint forwards content-free event chunks as + `{"type": "event", "event": ...}` records (previously dropped). + - `should_finalize(messages, text)` message-view contract documented: + executor-appended TOOL messages always carry `tool_name`; + `tool_result` carries the result text for successful executions and is + `None` for errors/policy denials — the supported discriminator for + "this tool produced a real result". `tool_call_id` only for native + tool-calling providers. ### Changed -- **`single_pass` + `astream()` no longer streams the structured answer as - content chunks** — content is suppressed (it is the JSON envelope by - construction when the schema rides on every loop call); tool-call chunks - still stream. Migration: read the answer from `AgentResult.parsed` / - `.content` instead of accumulating content chunks. (#174) -- **`Message.tool_result` is now always populated on TOOL messages** — - success, error, and policy-denial results all carry the text the model saw - (previously error/denial messages had `tool_result=None`). Sessions - persisted by older versions are normalized on load in `Message.from_dict`. - If you used `tool_result is None` to detect failed tools, match on the - `Error executing tool` / denial message text instead. (#174) +- **`single_pass` structured output no longer streams the answer** (beta + surface): in `astream()` content chunks are suppressed and in + `run()`/`arun()` the `stream_handler` callback is not invoked for + single_pass loop turns or the synthesis turn — content on those calls is + the JSON envelope by construction. Tool-call chunks still stream + (including on providers without native async streaming, which previously + emitted no tool-call chunks at all). Migration: read the answer from + `AgentResult.parsed` / `.content` instead of accumulating streamed + content. (#174) ## [1.2.0] - 2026-07-14 — Tool-Boundary Guardrails Complete & Structured Synthesis Follow-ups diff --git a/docs/llms-full.txt b/docs/llms-full.txt index d9863ad..f1c4071 100644 --- a/docs/llms-full.txt +++ b/docs/llms-full.txt @@ -4601,20 +4601,53 @@ The synthesis call is **conditional** (v1.2, #164/#166) — three paths avoid it Branch to a fallback deliberately on `"validation_failed"` instead of guessing from a bare `None`. -### Streaming and response_format +### Streaming and response_format — the public contract (#174) + +**Since: v1.3.0** (the `structured_synthesis_start` event since v1.2.0) + +These behaviors are load-bearing for chat surfaces, documented here as a +**tested public contract** (`tests/agent/test_structured_streaming_contract.py` +pins every clause). The contract covers `astream()` chunks, the +`stream_handler` callback in `run()`/`arun()`, and the `selectools.serve` +SSE stream (which forwards event chunks as `{"type": "event", ...}`): - **Default mode**: `astream()` streams the model's answer token-by-token, and with `response_format` set the answer IS the JSON — clients rendering chunks as chat text will see raw JSON fragments. Accumulate chunks and parse, or use `final_turn_only`. -- **`final_turn_only=True`**: the tool loop streams normally (prose chunks), - and the synthesis JSON is delivered ONLY via the terminal `AgentResult` - (`.content` / `.parsed`) — never leaked as content chunks. When a synthesis - call starts, `astream()` emits `StreamChunk(event="structured_synthesis_start")` - (no content) so clients can render a pending state. -- **`single_pass=True` caveat**: the final loop answer is the JSON itself, so - in `astream()` it streams as content chunks (like default mode) — the - trade-off for eliminating the synthesis round-trip. +- **`final_turn_only=True` guarantees**: + - The **synthesis turn never streams** — not as `astream()` chunks and not + through `stream_handler`. Its JSON arrives only on the terminal + `AgentResult` (`.content` / `.parsed`). + `StreamChunk(event="structured_synthesis_start")` (content-free) fires + once per run when the synthesis turn begins; validation retries within + the turn make additional (equally non-streamed) calls without re-emitting + the event. + - With `single_pass`: content is suppressed on **both** channels (chunks + and `stream_handler`) — the schema rides on every loop call, so content + is the JSON envelope by construction. Tool-call chunks still stream for + activity display (including on providers without native async + streaming). + - **Reuse caveat, signaled**: without `single_pass`, the loop streams model + output as it always has — so when the converged loop answer itself + validates and is reused (`reuse_loop_answer=True`), that answer has + already streamed as ordinary loop chunks. + `StreamChunk(event="structured_reuse")` fires at that moment so a chat + surface can convert or retract the just-streamed bubble (not emitted in + `single_pass` mode, where nothing streamed needs converting). Clients + that must never display raw JSON should handle this event, or use + `single_pass` on supporting providers. +- **`should_finalize(messages, text)` contract**: `messages` is the run's + conversation view at convergence — prior history (with memory), possible + `Role.SYSTEM` context injections (summaries, entity/knowledge memory), + this turn's USER message(s), the ASSISTANT tool-call messages, and one + `Message(role=TOOL)` per executed tool. On executor-appended TOOL + messages, `tool_name` is always populated; `tool_result` carries the + result text for **successful** executions and stays `None` for errors and + policy/coherence denials — `tool_result is not None` is the supported + test for "this tool produced a real result". `tool_call_id` is populated + for native tool-calling providers, `None` for text-parsed tool calls. + Treat messages as read-only. ### Structured Retry Budget (v0.22.0 — BUG-34) diff --git a/docs/modules/AGENT.md b/docs/modules/AGENT.md index aeef343..e39a974 100644 --- a/docs/modules/AGENT.md +++ b/docs/modules/AGENT.md @@ -1338,39 +1338,51 @@ guessing from a bare `None`. ### Streaming and response_format — the public contract (#174) +**Since: v1.3.0** (the `structured_synthesis_start` event since v1.2.0) + These behaviors are load-bearing for chat surfaces, documented here as a **tested public contract** (`tests/agent/test_structured_streaming_contract.py` -pins every clause): +pins every clause). The contract covers `astream()` chunks, the +`stream_handler` callback in `run()`/`arun()`, and the `selectools.serve` +SSE stream (which forwards event chunks as `{"type": "event", ...}`): - **Default mode**: `astream()` streams the model's answer token-by-token, and with `response_format` set the answer IS the JSON — clients rendering chunks as chat text will see raw JSON fragments. Accumulate chunks and parse, or use `final_turn_only`. - **`final_turn_only=True` guarantees**: - - The **synthesis call never streams**: its JSON arrives only on the - terminal `AgentResult` (`.content` / `.parsed`), and + - The **synthesis turn never streams** — not as `astream()` chunks and not + through `stream_handler`. Its JSON arrives only on the terminal + `AgentResult` (`.content` / `.parsed`). `StreamChunk(event="structured_synthesis_start")` (content-free) fires - **iff** a synthesis call is made. - - With `single_pass`: content chunks are **suppressed entirely** (the schema - rides on every loop call, so content is the JSON envelope by - construction); tool-call chunks still stream for activity display. + once per run when the synthesis turn begins; validation retries within + the turn make additional (equally non-streamed) calls without re-emitting + the event. + - With `single_pass`: content is suppressed on **both** channels (chunks + and `stream_handler`) — the schema rides on every loop call, so content + is the JSON envelope by construction. Tool-call chunks still stream for + activity display (including on providers without native async + streaming). - **Reuse caveat, signaled**: without `single_pass`, the loop streams model output as it always has — so when the converged loop answer itself validates and is reused (`reuse_loop_answer=True`), that answer has already streamed as ordinary loop chunks. `StreamChunk(event="structured_reuse")` fires at that moment so a chat - surface can convert or retract the just-streamed bubble. Clients that - must never display raw JSON should handle this event (or use - `single_pass` on supporting providers, where suppression is total). + surface can convert or retract the just-streamed bubble (not emitted in + `single_pass` mode, where nothing streamed needs converting). Clients + that must never display raw JSON should handle this event, or use + `single_pass` on supporting providers. - **`should_finalize(messages, text)` contract**: `messages` is the run's - conversation view at convergence — prior history (with memory, including - restored sessions), this turn's USER message(s), the ASSISTANT tool-call - messages, and one `Message(role=TOOL)` per executed tool with `tool_name` - and `tool_result` always populated (`tool_result` equals the result text - the model saw, including error/denial messages; legacy persisted messages - are normalized on load). `tool_call_id` is populated for native - tool-calling providers but `None` for text-parsed tool calls. Treat - messages as read-only. + conversation view at convergence — prior history (with memory), possible + `Role.SYSTEM` context injections (summaries, entity/knowledge memory), + this turn's USER message(s), the ASSISTANT tool-call messages, and one + `Message(role=TOOL)` per executed tool. On executor-appended TOOL + messages, `tool_name` is always populated; `tool_result` carries the + result text for **successful** executions and stays `None` for errors and + policy/coherence denials — `tool_result is not None` is the supported + test for "this tool produced a real result". `tool_call_id` is populated + for native tool-calling providers, `None` for text-parsed tool calls. + Treat messages as read-only. ### Structured Retry Budget (v0.22.0 — BUG-34) diff --git a/src/selectools/agent/_tool_executor.py b/src/selectools/agent/_tool_executor.py index ad1f27d..a3bfc48 100644 --- a/src/selectools/agent/_tool_executor.py +++ b/src/selectools/agent/_tool_executor.py @@ -781,20 +781,23 @@ def _append_tool_result( tool_content: str, tool_name: str, tool_call_id: Optional[str] = None, + tool_result: Optional[str] = None, run_id: Optional[str] = None, ) -> None: """Update history with tool output. - ``tool_result`` is always set to ``tool_content`` so EVERY TOOL - message — success, error, or policy denial — carries a populated - ``tool_result`` equal to the text the model saw. This is a public - contract for ``should_finalize`` consumers (#174). + Public contract for ``should_finalize`` consumers (#174): on + executor-appended TOOL messages, ``tool_result`` carries the result + text for SUCCESSFUL executions and stays ``None`` for errors and + policy/coherence denials (whose ``content`` carries the error text). + The ``None`` marker is the structural discriminator between "the + tool produced a result" and "the tool failed". """ tool_msg = Message( role=Role.TOOL, content=tool_content, tool_name=tool_name, - tool_result=tool_content, + tool_result=tool_result, tool_call_id=tool_call_id, ) self._history.append(tool_msg) @@ -1038,6 +1041,7 @@ def chunk_cb(chunk: str) -> None: message_content, r.tool_call.tool_name, r.tool_call.id, + tool_result=message_content, run_id=run_id, ) @@ -1315,6 +1319,7 @@ def chunk_cb(chunk: str) -> None: message_content, r.tool_call.tool_name, r.tool_call.id, + tool_result=message_content, run_id=run_id, ) @@ -1484,6 +1489,7 @@ def _execute_single_tool(self, ctx: _RunContext, tool_call: ToolCall) -> bool: cached_content, tool_name, tool_call.id, + tool_result=cached_content, run_id=ctx.run_id, ) if getattr(tool, "terminal", False) or self._stop_condition_hit(tool_name, cached_raw): @@ -1599,6 +1605,7 @@ def chunk_callback(chunk: str) -> None: message_content, tool_name, tool_call.id, + tool_result=message_content, run_id=ctx.run_id, ) @@ -1726,6 +1733,7 @@ async def _aexecute_single_tool(self, ctx: _RunContext, tool_call: ToolCall) -> cached_content, tool_name, tool_call.id, + tool_result=cached_content, run_id=ctx.run_id, ) if getattr(tool, "terminal", False) or self._stop_condition_hit(tool_name, cached_raw): @@ -1858,6 +1866,7 @@ def chunk_callback(chunk: str) -> None: message_content, tool_name, tool_call.id, + tool_result=message_content, run_id=ctx.run_id, ) diff --git a/src/selectools/agent/config_groups.py b/src/selectools/agent/config_groups.py index af709d9..539473c 100644 --- a/src/selectools/agent/config_groups.py +++ b/src/selectools/agent/config_groups.py @@ -216,18 +216,22 @@ class StructuredOutputConfig: ``messages`` contract (public, tested — issue #174): the run's conversation view at convergence, in chronological order — - prior history (when memory is configured, including restored - sessions), this turn's USER message(s), the ASSISTANT messages + prior history (when memory is configured), possible + ``Role.SYSTEM`` context injections (summaries, entity/knowledge + memory), this turn's USER message(s), the ASSISTANT messages that requested tools, and one ``Message(role=TOOL)`` per - executed tool. On every TOOL message, ``tool_name`` and - ``tool_result`` are ALWAYS populated (``tool_result`` equals - the result text the model saw — success, error, or policy - denial; legacy persisted messages are normalized on load). - ``tool_call_id`` is populated for native tool-calling providers - but is ``None`` for text-parsed tool calls — key by it only if - you know your provider supplies ids. Treat the messages as - read-only; mutating the list itself is harmless. Default: None - (always finalize, the v1.1 behavior). + executed tool. On executor-appended TOOL messages, + ``tool_name`` is always populated and ``tool_result`` carries + the result text for SUCCESSFUL executions while staying + ``None`` for errors and policy/coherence denials (whose + ``content`` carries the error text) — ``tool_result is not + None`` is the supported test for "this tool produced a real + result". Caller-seeded TOOL messages keep whatever fields the + caller set. ``tool_call_id`` is populated for native + tool-calling providers but is ``None`` for text-parsed tool + calls. Treat the messages as read-only; mutating the list + itself is harmless. Default: None (always finalize, the v1.1 + behavior). single_pass: In ``final_turn_only`` mode with a provider that advertises ``supports_native_structured_output_with_tools`` (OpenAI/Azure), carry the schema natively on the loop calls so diff --git a/src/selectools/agent/core.py b/src/selectools/agent/core.py index 0779bf4..36035f9 100644 --- a/src/selectools/agent/core.py +++ b/src/selectools/agent/core.py @@ -1681,8 +1681,12 @@ def run( "on_iteration_start", ctx.run_id, ctx.iteration, self._history ) + # The synthesis turn and single_pass loop turns carry the + # structured JSON — never feed it to the token callback + # (same contract as astream chunk suppression, #174). + _suppress_handler = ctx.structured_final_pending or ctx.single_pass_active response_msg = self._call_provider( - stream_handler=stream_handler, + stream_handler=None if _suppress_handler else stream_handler, trace=ctx.trace, run_id=ctx.run_id, **self._structured_call_kwargs(ctx), @@ -2134,6 +2138,10 @@ async def astream( full_content = _response_content if response_msg.tool_calls: current_tool_calls = response_msg.tool_calls + # Mirror the native-astream path: tool-call + # activity always streams, even when content is + # suppressed (single_pass contract, #174). + yield StreamChunk(tool_calls=current_tool_calls) else: # BUG-33: wrap provider.astream() generator in aclosing so # that a guardrail raise, validation failure, or caller @@ -2542,8 +2550,12 @@ async def arun( "on_iteration_start", ctx.run_id, ctx.iteration, self._history ) + # The synthesis turn and single_pass loop turns carry the + # structured JSON — never feed it to the token callback + # (same contract as astream chunk suppression, #174). + _suppress_handler = ctx.structured_final_pending or ctx.single_pass_active response_msg = await self._acall_provider( - stream_handler=stream_handler, + stream_handler=None if _suppress_handler else stream_handler, trace=ctx.trace, run_id=ctx.run_id, **self._structured_call_kwargs(ctx), diff --git a/src/selectools/serve/api.py b/src/selectools/serve/api.py index 525b57f..07fe797 100644 --- a/src/selectools/serve/api.py +++ b/src/selectools/serve/api.py @@ -446,6 +446,13 @@ async def _events() -> AsyncIterator[bytes]: async for item in runner.astream(messages): if isinstance(item, AgentResult): final = item + elif isinstance(item, StreamChunk) and getattr(item, "event", None): + # Content-free lifecycle signals (#174) — e.g. + # structured_synthesis_start / structured_reuse — + # must reach SSE clients so chat surfaces can + # render pending states or convert bubbles. + payload = {"type": "event", "event": item.event} + yield f"data: {json.dumps(payload)}\n\n".encode() elif isinstance(item, StreamChunk) and item.content: chunks.append(item.content) payload = {"type": "chunk", "content": item.content} diff --git a/src/selectools/types.py b/src/selectools/types.py index 269156b..33c335f 100644 --- a/src/selectools/types.py +++ b/src/selectools/types.py @@ -139,6 +139,12 @@ class Message: The image will be automatically base64-encoded on initialization. tool_name: Optional name of the tool that generated this message (for TOOL role). tool_result: Optional result from tool execution (for TOOL role). + On TOOL messages appended by the agent's tool executor + this carries the result text for successful executions + and stays None for errors/policy denials — the None + marker distinguishes failures (see the should_finalize + contract in StructuredOutputConfig, #174). Caller-built + TOOL messages may leave it unset. image_base64: Auto-populated base64-encoded image data (not set via __init__). Example: @@ -233,12 +239,6 @@ def from_dict(cls, data: Dict[str, Any]) -> "Message": msg.image_base64 = data.get("image_base64") msg.tool_name = data.get("tool_name") msg.tool_result = data.get("tool_result") - # Sessions persisted before v1.2.x stored tool_result=None on - # error/denial TOOL messages; normalize so the should_finalize - # contract (tool_result always populated on TOOL messages, #174) - # holds for restored histories too. - if msg.role == Role.TOOL and msg.tool_result is None: - msg.tool_result = msg.content msg.tool_call_id = data.get("tool_call_id") raw_calls = data.get("tool_calls") if raw_calls: @@ -391,15 +391,18 @@ class StreamChunk: content: The text content of this chunk (delta). role: The role emitting this chunk (usually ASSISTANT). tool_calls: Optional list of tool calls (if this chunk contains tool invocations). - event: Optional lifecycle signal (v1.2). Event chunks carry no - content. Currently emitted: - ``"structured_synthesis_start"`` — the tool loop converged and a - structured synthesis call is starting; clients can render a - pending state. - ``"structured_reuse"`` — the converged loop answer validated and - was reused as the structured answer; the chunks already streamed - for this turn WERE that answer, so chat surfaces can convert or - retract the bubble (#174). + event: Optional lifecycle signal. Event chunks carry no content. + Currently emitted: + ``"structured_synthesis_start"`` (since v1.2.0) — the structured + synthesis turn is starting; clients can render a pending state. + Fires once per run when the turn begins; validation retries + within the synthesis turn do not re-emit it. + ``"structured_reuse"`` (since v1.3.0) — the converged loop answer + validated and was reused as the structured answer; the chunks + already streamed for this turn WERE that answer, so chat surfaces + can convert or retract the bubble. Not emitted in ``single_pass`` + mode (content is suppressed there, so nothing streamed needs + converting). (#174) """ content: str = "" diff --git a/tests/agent/test_structured_streaming_contract.py b/tests/agent/test_structured_streaming_contract.py index 3427ec8..5a5815d 100644 --- a/tests/agent/test_structured_streaming_contract.py +++ b/tests/agent/test_structured_streaming_contract.py @@ -11,8 +11,10 @@ ``StreamChunk(event="structured_reuse")`` signals that those chunks WERE the structured answer. 2. ``should_finalize(messages, last_response_text)`` receives the run's - conversation view at convergence, including this turn's TOOL messages - with ``tool_name`` and ``tool_result`` populated. + conversation view at convergence, including this turn's TOOL messages: + ``tool_name`` always populated; ``tool_result`` populated for successful + executions and ``None`` for errors/policy denials (the structural + failure discriminator). """ from __future__ import annotations @@ -148,11 +150,14 @@ async def test_synthesis_start_event_fires_iff_synthesis_call_made(self) -> None assert JSON_ANSWER in streamed, "reuse-path answer streams as loop output" assert result is not None and result.parsed == {"answer": "42"} - # Case C: single_pass -> no synthesis call -> no event + # Case C: single_pass -> no synthesis call -> no synthesis event, and + # no structured_reuse either (content was suppressed; nothing + # streamed needs converting) provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) agent = _agent(provider, single_pass=True) chunks, _ = await _collect(agent, response_format=SCHEMA) assert all(c.event != "structured_synthesis_start" for c in chunks) + assert all(c.event != "structured_reuse" for c in chunks) @pytest.mark.asyncio async def test_event_chunks_carry_no_content(self) -> None: @@ -219,9 +224,9 @@ def predicate(messages: List[Message], text: str) -> bool: class TestToolResultFieldInvariant: - def test_every_tool_message_in_history_has_tool_result(self) -> None: - """The invariant behind the predicate contract: TOOL messages always - carry tool_result (defaulting to their content).""" + def test_every_successful_tool_message_has_tool_result(self) -> None: + """The invariant behind the predicate contract: successful executions + always populate tool_result; failures keep it None (tested below).""" first = Message( role=Role.ASSISTANT, content="", @@ -279,21 +284,22 @@ def test_policy_denied_tool_messages_carry_tool_result(self) -> None: tool_msgs = [m for m in agent._history if m.role == Role.TOOL] assert tool_msgs, "the denial must still produce a TOOL message" for m in tool_msgs: - assert m.tool_result is not None, "policy denials must populate tool_result" + assert m.tool_result is None, "denials keep the None failure marker" + assert m.content, "the denial text lives on content" - def test_legacy_persisted_tool_messages_normalized_on_load(self) -> None: - """Sessions saved before v1.2.x stored tool_result=None on error - TOOL messages; from_dict must normalize so the contract holds for - restored histories.""" - legacy = { + def test_from_dict_round_trip_preserves_none_marker(self) -> None: + """from_dict is @stable: the None failure marker must survive the + to_dict/from_dict round-trip unaltered (no silent rewriting).""" + persisted = { "role": "tool", "content": "Error executing tool 'x': boom", "tool_name": "x", "tool_result": None, "tool_call_id": "t1", } - restored = Message.from_dict(legacy) - assert restored.tool_result == "Error executing tool 'x': boom" + restored = Message.from_dict(persisted) + assert restored.tool_result is None + assert restored.to_dict()["tool_result"] is None def test_error_tool_messages_also_carry_tool_result(self) -> None: @tool(description="Always fails.") @@ -311,4 +317,62 @@ def broken(query: str) -> str: tool_msgs = [m for m in agent._history if m.role == Role.TOOL] assert tool_msgs for m in tool_msgs: - assert m.tool_result is not None, "error results must also populate tool_result" + assert m.tool_result is None, "errors keep the None failure marker" + assert "boom" in m.content, "the error text lives on content" + + +class TestStreamHandlerSuppression: + """run()/arun() stream_handler is the second streaming surface — the + synthesis turn and single_pass loop turns must never feed it JSON.""" + + def _streaming_agent(self, provider: Any, **structured: Any) -> Agent: + return Agent( + [lookup], + provider=provider, + config=AgentConfig( + model="fake-model", + stream=True, # the handler channel is only live in stream mode + structured_output=StructuredOutputConfig(final_turn_only=True, **structured), + ), + ) + + def test_run_synthesis_never_reaches_stream_handler(self) -> None: + seen: List[str] = [] + # (SharedFakeProvider.stream() yields content only, so the scenario + # converges on the first turn and then synthesizes.) + provider = SharedFakeProvider(["prose answer", JSON_ANSWER]) + agent = self._streaming_agent(provider) + result = agent.run("go", response_format=SCHEMA, stream_handler=seen.append) + assert result.parsed == {"answer": "42"} + assert any("prose answer" in t for t in seen), "loop prose still streams to handler" + assert not any(JSON_ANSWER in t for t in seen), "synthesis JSON leaked to handler" + + @pytest.mark.asyncio + async def test_arun_single_pass_never_invokes_handler_with_content(self) -> None: + seen: List[str] = [] + provider = NativeWithToolsProvider([_tool_call_message(), JSON_ANSWER]) + agent = self._streaming_agent(provider, single_pass=True) + result = await agent.arun("go", response_format=SCHEMA, stream_handler=seen.append) + assert result.parsed == {"answer": "42"} + assert not any("answer" in t for t in seen), f"JSON leaked to handler: {seen!r}" + + +class TestNonStreamingProviderFallback: + @pytest.mark.asyncio + async def test_single_pass_fallback_still_yields_tool_call_chunks(self) -> None: + """Providers without native async streaming route astream() through + the acomplete fallback; the contract's 'tool-call chunks still + stream' must hold there too.""" + + # NB: supports_streaming must go through the constructor — + # SharedFakeProvider.__init__ sets an instance attribute that would + # shadow a subclass class attribute. + provider = NativeWithToolsProvider( + [_tool_call_message(), JSON_ANSWER], supports_streaming=False + ) + agent = _agent(provider, single_pass=True) + chunks, result = await _collect(agent, response_format=SCHEMA) + assert any(c.tool_calls for c in chunks), "fallback path must yield tool-call chunks" + streamed = "".join(c.content or "" for c in chunks) + assert "answer" not in streamed + assert result is not None and result.parsed == {"answer": "42"} diff --git a/tests/test_agent_api.py b/tests/test_agent_api.py index fa96b28..2346ea5 100644 --- a/tests/test_agent_api.py +++ b/tests/test_agent_api.py @@ -17,6 +17,7 @@ from selectools.agent import Agent, AgentConfig # noqa: E402 from selectools.serve import AgentAPI # noqa: E402 from selectools.tools import Tool, ToolParameter # noqa: E402 +from selectools.types import Message, Role # noqa: E402 from tests.conftest import SharedFakeProvider # noqa: E402 # ─── Helpers ───────────────────────────────────────────────────────────────── @@ -313,6 +314,30 @@ def test_user_cannot_chat_into_another_users_session(self): class TestChatStream: + def test_stream_forwards_event_chunks(self): + """Content-free StreamChunk lifecycle events (#174) — e.g. + structured_synthesis_start / structured_reuse — must reach SSE + clients as {"type": "event", ...} records, not be dropped.""" + from selectools.types import AgentResult, StreamChunk + + agent = make_agent(responses=["final answer"]) + + class _EventStreamAgent(type(agent)): + async def astream(self, messages, **kwargs): + yield StreamChunk(event="structured_synthesis_start") + yield StreamChunk(content="final answer") + yield AgentResult(message=Message(role=Role.ASSISTANT, content="final answer")) + + agent.__class__ = _EventStreamAgent + client = make_client(AgentAPI(agents=[agent])) + r = client.post("/v1/chat/stream", json={"input": "go"}) + assert r.status_code == 200 + events = sse_events(r.text) + event_records = [e for e in events if e.get("type") == "event"] + assert event_records == [{"type": "event", "event": "structured_synthesis_start"}] + chunk_text = "".join(e["content"] for e in events if e["type"] == "chunk") + assert chunk_text == "final answer" + def test_stream_yields_chunks_and_result(self): client = make_client(AgentAPI(agents=[make_agent(responses=["streamed answer"])])) r = client.post("/v1/chat/stream", json={"input": "stream it"})