diff --git a/CHANGELOG.md b/CHANGELOG.md index 142084e..42b987e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,43 @@ 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).** Previously source-only behaviors are now documented, stated in + docstrings, and pinned by dedicated contract tests + (`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` 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 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..42b987e 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,43 @@ 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).** Previously source-only behaviors are now documented, stated in + docstrings, and pinned by dedicated contract tests + (`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` 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 Second 1.x minor, closing the loop on the v1.1.0 features from real 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 69cd683..e39a974 100644 --- a/docs/modules/AGENT.md +++ b/docs/modules/AGENT.md @@ -1336,20 +1336,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/src/selectools/agent/_tool_executor.py b/src/selectools/agent/_tool_executor.py index 448fe4c..a3bfc48 100644 --- a/src/selectools/agent/_tool_executor.py +++ b/src/selectools/agent/_tool_executor.py @@ -784,7 +784,15 @@ 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. + + 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, diff --git a/src/selectools/agent/config_groups.py b/src/selectools/agent/config_groups.py index 61b5213..539473c 100644 --- a/src/selectools/agent/config_groups.py +++ b/src/selectools/agent/config_groups.py @@ -213,16 +213,37 @@ 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), 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 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 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..36035f9 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: @@ -1677,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), @@ -2053,6 +2061,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.single_pass_active full_content = "" current_tool_calls: List[ToolCall] = [] @@ -2119,10 +2133,15 @@ 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 + # 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 @@ -2144,7 +2163,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) @@ -2183,6 +2203,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( @@ -2524,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 87a2543..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: @@ -385,10 +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). 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. + 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_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 new file mode 100644 index 0000000..5a5815d --- /dev/null +++ b/tests/agent/test_structured_streaming_contract.py @@ -0,0 +1,378 @@ +"""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``: 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: + ``tool_name`` always populated; ``tool_result`` populated for successful + executions and ``None`` for errors/policy denials (the structural + failure discriminator). +""" + +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 selectools.usage import UsageStats +from tests.conftest import SharedFakeProvider + + +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", + "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 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, 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 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: + 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_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="", + 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_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 None, "denials keep the None failure marker" + assert m.content, "the denial text lives on content" + + 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(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.") + 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 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"})