From 6eff2ba2f11700c471cba79d14904973d59b0cb3 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 15:51:07 +0200 Subject: [PATCH 01/16] Require committed finish reason for tool calls Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...orize-function-calls-with-finish-reason.md | 78 +++++++++++++ .../specs/004-python-function-calling-loop.md | 32 +++++- .../packages/core/agent_framework/_clients.py | 15 ++- .../core/agent_framework/_sessions.py | 7 +- .../packages/core/agent_framework/_tools.py | 20 ++++ .../packages/core/agent_framework/_types.py | 2 +- python/packages/core/tests/core/conftest.py | 22 ++++ .../packages/core/tests/core/test_clients.py | 2 +- .../core/test_function_invocation_logic.py | 107 +++++++++++++++++- .../tests/core/test_middleware_with_chat.py | 3 +- 10 files changed, 281 insertions(+), 7 deletions(-) create mode 100644 docs/decisions/0041-authorize-function-calls-with-finish-reason.md diff --git a/docs/decisions/0041-authorize-function-calls-with-finish-reason.md b/docs/decisions/0041-authorize-function-calls-with-finish-reason.md new file mode 100644 index 00000000000..1ac2228c1be --- /dev/null +++ b/docs/decisions/0041-authorize-function-calls-with-finish-reason.md @@ -0,0 +1,78 @@ +--- +status: proposed +contact: eavanvalkenburg +date: 2026-09-11 +deciders: eavanvalkenburg +--- + +# Authorize function calls with the finish reason + +## Context and Problem Statement + +Streaming providers expose function-call content before they commit the complete model turn. Agent Framework already +waits for the provider stream to finish before local invocation, but iterator completion alone does not prove that the +provider emitted its terminal event or committed every call. A syntactically complete call fragment can therefore be +mistaken for an executable request after an interrupted or failed response. + +## Decision Drivers + +- Execute local side effects only after an authoritative provider completion signal. +- Keep one provider-neutral authorization rule for streaming and non-streaming responses. +- Preserve incremental function-call updates for callers without adding buffering or another stream drain. +- Avoid adding a second response lifecycle field or changing the meaning of `informational_only`. +- Keep provider-specific protocol evidence in the provider adapter. + +## Considered Options + +### Add a response completion field + +- Good: separates lifecycle state from generation finish reason. +- Bad: adds a public concept that overlaps existing finish-reason behavior. +- Bad: requires every response and update mapping to carry another field. + +### Mark speculative calls as informational + +- Good: reuses the existing core execution filter. +- Bad: `informational_only` describes calls the framework does not own, not uncommitted local calls. +- Bad: changes the meaning of caller-visible function-call content. + +### Infer commitment from content or stream exhaustion + +- Good: requires no provider changes. +- Bad: complete JSON and clean iterator exhaustion do not prove that the provider committed the turn. +- Bad: different providers expose different authoritative terminal events. + +### Authorize with `finish_reason == "tool_calls"` + +- Good: uses the existing provider-neutral response field. +- Good: gives core one fail-closed rule for streaming and non-streaming responses. +- Good: lets providers continue yielding speculative content immediately and resolve the reason on their existing + terminal event. +- Bad: custom clients that return local function calls without `finish_reason="tool_calls"` must be updated. +- Bad: provider adapters must assign `"tool_calls"` only after their own protocol commitment evidence is satisfied. + +## Decision Outcome + +Chosen option: "Authorize with `finish_reason == "tool_calls"`", because it establishes one explicit execution +boundary without another public lifecycle model. + +Core approves or executes model-issued local calls only when the finalized response has +`finish_reason == "tool_calls"`. Missing and all other finish reasons are non-authorizing. + +Each provider adapter owns its commitment evidence. Adapters that require correction use the same small local +`_resolve_finish_reason(...)` pattern: committed calls override the ordinary reason to `"tool_calls"`; calls without +commitment cannot retain or manufacture `"tool_calls"`; responses without calls preserve their ordinary reason. The +function remains local to each provider module because the evidence is protocol-specific and does not justify a shared +base class, mixin, registry, or cross-provider helper. + +Streaming adapters resolve the finish reason on the provider's existing terminal event. They do not add another drain, +buffer speculative content, or wait after iterator exhaustion. Calls returned without authorization remain visible to +the caller but are excluded from later model-bound replay. + +### Consequences + +- Local tool bodies, approval requests, function middleware, function results, and follow-up model calls require an + authoritatively committed `"tool_calls"` response. +- Custom clients must set `finish_reason="tool_calls"` for completed local function-call turns. +- Provider adapters must not infer commitment from parseable arguments, observed call content, or iterator exhaustion. +- Interrupted call-bearing responses remain inspectable without becoming executable or replayable model history. diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 81019bb5bbe..145443fbac5 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -128,7 +128,8 @@ Code-reading landmarks: - `_get_response_with_function_invocation(...)` owns non-streaming aggregation. - `_stream_response_with_function_invocation(...)` owns streamed emission/finalization. - `_resolve_approval_responses(...)` handles only inbound approval decisions. -- `_process_model_function_calls(...)` handles only calls from a completed model response. +- `_process_model_function_calls(...)` handles only calls from a provider-committed model response whose + `finish_reason` is `"tool_calls"`. - `_try_execute_function_calls(...)` decides approval/declaration/execution behavior for a batch. - `_replace_approval_contents_with_results(...)` is the occurrence-aware approval transcript normalizer. - `FunctionInvocationLayer._update_function_invocation_continuation_state(...)` updates continuation state after @@ -136,6 +137,33 @@ Code-reading landmarks: the next service call, but must delegate to the base implementation so generic conversation continuation remains synchronized with the active `AgentSession`. +### Function-call commitment boundary + +`finish_reason == "tool_calls"` is the provider-neutral authorization signal for local function invocation. A +call-bearing response with a missing or different finish reason is caller-visible but must not create a local approval +request, enter function middleware, execute a tool body, create a function result, or trigger another model call. + +Provider adapters may set `"tool_calls"` only after the provider's authoritative terminal evidence commits every +actionable call in that response. For non-streaming APIs, the final SDK response must prove completion. For streaming +APIs, adapters resolve the reason on the existing terminal event; parseable arguments and iterator exhaustion are not +commitment evidence. Providers must continue yielding incremental call updates without adding a second drain, +buffering the complete stream, or waiting after EOF. + +Adapters that need to combine an ordinary provider reason with protocol-specific call commitment use a small local +`_resolve_finish_reason(provider_finish_reason, *, has_function_calls, function_calls_committed)` function: + +- committed calls return `"tool_calls"` regardless of the ordinary provider reason; +- calls without commitment never return `"tool_calls"`; +- responses without calls preserve their ordinary reason. + +The helper is intentionally repeated in affected provider modules. The commitment evidence differs by protocol and +does not justify a shared provider abstraction. Custom clients that return completed local function calls must set +`finish_reason="tool_calls"` explicitly. + +Uncommitted call-bearing assistant turns remain in the caller-visible response but are marked as excluded from future +model input. Reasoning and call content from that turn are omitted together so later stateless replay cannot create a +dangling provider call. + ### Approval pause and resume ```mermaid @@ -489,6 +517,8 @@ that manually replay messages own the equivalent rule: do not resend an approval | String input | Flexible string input follows the same loop behavior. | `test_base_client_with_function_calling_string_input` | | Multiple sequential rounds | Each round retains one call/result pair. | `test_base_client_with_function_calling_resets` | | Streaming call | Call chunks, one result update, and final text are emitted in order. | `test_base_client_with_streaming_function_calling` | +| Missing or non-authorizing finish reason | Call content remains caller-visible, but no approval, middleware, tool body, result, or follow-up model call occurs; the uncommitted turn is excluded from later model input. | `test_function_calls_require_tool_calls_finish_reason`, `test_missing_tool_calls_finish_reason_does_not_request_approval` | +| Provider terminal commitment | Streaming providers emit incremental calls immediately and assign `"tool_calls"` only on their authoritative terminal event; incomplete, failed, malformed, or prematurely exhausted streams remain non-authorizing. | Provider-specific completion tests in OpenAI Responses, Anthropic, Gemini, and AG-UI | | Function-call occurrence identity | Actionable calls gain one stable `Content.id`; a safe local empty-`call_id` fallback uses that id with a migration warning, and streaming aggregation preserves provider-assigned occurrence ids across interleaved fragments. OpenAI Chat Completions scopes fragment correlation to each request and `(choice.index, tool.index)`. | `test_actionable_function_call_gets_stable_occurrence_identity`, `test_actionable_function_call_uses_occurrence_identity_for_empty_call_id`, `test_streaming_empty_call_id_keeps_occurrence_identity_through_approval`, `test_streaming_empty_call_id_delta_reuses_opening_call_identity`, `test_streaming_interleaved_indexed_call_fragments_coalesce_by_occurrence`, `packages/core/tests/core/test_types.py::test_function_call_occurrence_id_roundtrips_without_regeneration`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_streaming_tool_call_identity_is_request_local_and_scoped_by_choice_index` | | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 564d78aa660..95553ce4125 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -63,6 +63,8 @@ logger = logging.getLogger("agent_framework") +_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY = "_agent_framework_uncommitted_function_calls" + # region SupportsChatGetResponse Protocol @@ -374,7 +376,18 @@ async def _prepare_messages_for_model_call( compaction_strategy: CompactionStrategy | None = None, tokenizer: TokenizerProtocol | None = None, ) -> list[Message]: - prepared_messages = list(messages) + prepared_messages = [ + message + for message in messages + if message.role != "assistant" + or ( + _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY not in message.additional_properties + and not any( + _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY in content.additional_properties + for content in message.contents + ) + ) + ] if compaction_strategy is None: if tokenizer is None: return prepared_messages diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index 9fcce18a071..df2b8e057aa 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -1308,7 +1308,12 @@ def is_local_history_conversation_id(conversation_id: str | None) -> bool: def _response_contains_follow_up_request(response: ChatResponse) -> bool: """Return whether a response requires another model call in the current run.""" return any( - item.type == "function_approval_request" or (item.type == "function_call" and not item.informational_only) + item.type == "function_approval_request" + or ( + response.finish_reason == "tool_calls" + and item.type == "function_call" + and not item.informational_only + ) for message in response.messages for item in message.contents ) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index d09078ea434..ae54cb55480 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1822,6 +1822,19 @@ def _is_actionable_function_call(content: Content) -> bool: return content.type == "function_call" and not content.informational_only +def _mark_uncommitted_function_call_messages(response: ChatResponse) -> None: + from ._clients import _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY # pyright: ignore[reportPrivateUsage] + + for message in response.messages: + uncommitted_calls = [content for content in message.contents if _is_actionable_function_call(content)] + if not uncommitted_calls: + continue + replay_metadata = {"finish_reason": response.finish_reason} + message.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata + for content in uncommitted_calls: + content.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata + + def _underlying_function_call(content: Content) -> Content: if content.type == "function_approval_response" and content.function_call is not None: return content.function_call @@ -3392,6 +3405,13 @@ async def _process_model_function_calls( # 1. Extract only actionable, unanswered calls from this model turn. tools = _extract_tools(options) function_calls = _extract_function_calls(response) + if function_calls and response.finish_reason != "tool_calls": + _mark_uncommitted_function_call_messages(response) + if function_call_messages is not None: + _prepend_function_call_messages(response, function_call_messages) + if approval_requests: + _store_pending_approval_requests(invocation_session, approval_requests) + return _FunctionProcessingResult(errors_in_a_row=errors_in_a_row, action="return") if not (function_calls and tools): if function_call_messages is not None: _prepend_function_call_messages(response, function_call_messages) diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 7b0efa634e8..c66f83ef3a5 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -1782,7 +1782,7 @@ def _combine_annotations( Known values: - "stop": Normal completion - "length": Max tokens reached - - "tool_calls": Tool calls triggered + - "tool_calls": The provider completed the turn and committed the returned tool calls - "content_filter": Content filter triggered Examples: diff --git a/python/packages/core/tests/core/conftest.py b/python/packages/core/tests/core/conftest.py index 70361935082..0a7145a7d8e 100644 --- a/python/packages/core/tests/core/conftest.py +++ b/python/packages/core/tests/core/conftest.py @@ -173,6 +173,7 @@ def __init__(self, **kwargs: Any): self.run_responses: list[ChatResponse] = [] self.streaming_responses: list[list[ChatResponseUpdate]] = [] self.call_count: int = 0 + self.auto_finish_function_calls: bool = True @override def _inner_get_response( # pyrefly: ignore[bad-override] # ty: ignore[invalid-method-override] @@ -226,6 +227,16 @@ async def _get_non_streaming_response( conversation_id=response.conversation_id, ) + if ( + self.auto_finish_function_calls + and response.finish_reason is None + and any( + content.type == "function_call" and not content.informational_only + for message in response.messages + for content in message.contents + ) + ): + response.finish_reason = "tool_calls" return response def _get_streaming_response( @@ -253,6 +264,17 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: ) return response = self.streaming_responses.pop(0) + if ( + self.auto_finish_function_calls + and response + and not any(update.finish_reason is not None for update in response) + and any( + content.type == "function_call" and not content.informational_only + for update in response + for content in update.contents + ) + ): + response[-1].finish_reason = "tool_calls" for update in response: yield update await asyncio.sleep(0) diff --git a/python/packages/core/tests/core/test_clients.py b/python/packages/core/tests/core/test_clients.py index b5c4fc8853e..77fb0f640ff 100644 --- a/python/packages/core/tests/core/test_clients.py +++ b/python/packages/core/tests/core/test_clients.py @@ -389,7 +389,7 @@ def _tool_call_update(call_id: str, location: str) -> list[ChatResponseUpdate]: ) ], role="assistant", - finish_reason="stop", + finish_reason="tool_calls", response_id=f"resp_{call_id}", ) ] diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 7e688a2e13c..6eca05a00fc 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -5,7 +5,7 @@ import logging import warnings from collections.abc import AsyncIterable, Awaitable, Callable, Sequence -from typing import Any, Literal +from typing import Any, Literal, cast import pytest @@ -364,6 +364,109 @@ def test_actionable_function_call_uses_occurrence_identity_for_empty_call_id() - assert function_call.call_id == function_call.id +@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"]) +@pytest.mark.parametrize( + "finish_reason", + [None, "stop", "length", "content_filter"], + ids=["missing", "stop", "length", "content_filter"], +) +async def test_function_calls_require_tool_calls_finish_reason( + chat_client_base: SupportsChatGetResponse, + stream: bool, + finish_reason: str | None, +) -> None: + tool_calls = 0 + middleware_calls = 0 + + class RecordingMiddleware(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Any) -> None: + nonlocal middleware_calls + middleware_calls += 1 + await call_next() + + @tool(name="guarded_write", approval_mode="never_require") + def guarded_write() -> str: + nonlocal tool_calls + tool_calls += 1 + return "done" + + client = cast(Any, chat_client_base) + client.auto_finish_function_calls = False + function_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}) + if stream: + client.streaming_responses = [ + [ + ChatResponseUpdate( + contents=[function_call], + role="assistant", + finish_reason=cast(Any, finish_reason), + ) + ] + ] + else: + client.run_responses = [ + ChatResponse( + messages=[Message(role="assistant", contents=[function_call])], + finish_reason=cast(Any, finish_reason), + ) + ] + + result = client.get_response( + [Message(role="user", contents=["write"])], + options={"tools": [guarded_write]}, + middleware=[RecordingMiddleware()], + stream=stream, + ) + if stream: + updates = [update async for update in result] + assert any(content is function_call for update in updates for content in update.contents) + response = await result.get_final_response() + else: + response = await result + + contents = [content for message in response.messages for content in message.contents] + assert function_call in contents + assert not any(content.type in {"function_approval_request", "function_result"} for content in contents) + assert tool_calls == 0 + assert middleware_calls == 0 + assert client.call_count == 1 + + prepared_messages = await client._prepare_messages_for_model_call(response.messages) + assert prepared_messages == [] + + +@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"]) +async def test_missing_tool_calls_finish_reason_does_not_request_approval( + chat_client_base: SupportsChatGetResponse, + stream: bool, +) -> None: + @tool(name="guarded_write", approval_mode="always_require") + def guarded_write() -> str: + raise AssertionError("tool body must not run") + + client = cast(Any, chat_client_base) + client.auto_finish_function_calls = False + function_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}) + if stream: + client.streaming_responses = [[ChatResponseUpdate(contents=[function_call], role="assistant")]] + else: + client.run_responses = [ChatResponse(messages=[Message(role="assistant", contents=[function_call])])] + + result = client.get_response( + [Message(role="user", contents=["write"])], + options={"tools": [guarded_write]}, + stream=stream, + ) + response = await result.get_final_response() if stream else await result + + assert not any( + content.type == "function_approval_request" + for message in response.messages + for content in message.contents + ) + assert client.call_count == 1 + + async def test_streaming_empty_call_id_keeps_occurrence_identity_through_approval( chat_client_base: SupportsChatGetResponse, ) -> None: @@ -6236,6 +6339,7 @@ def test_func(arg1: str) -> str: contents=[Content.from_function_call(call_id="call_1", name="test_func", arguments='{"arg1": "v1"}')], ), conversation_id="conv_after_first_call", + finish_reason="tool_calls", ), ChatResponse( messages=Message(role="assistant", contents=["done"]), @@ -6269,6 +6373,7 @@ def test_func(arg1: str) -> str: contents=[Content.from_function_call(call_id="call_2", name="test_func", arguments='{"arg1": "v2"}')], role="assistant", conversation_id="stream_conv_after_first", + finish_reason="tool_calls", ), ], [ diff --git a/python/packages/core/tests/core/test_middleware_with_chat.py b/python/packages/core/tests/core/test_middleware_with_chat.py index bb5f0c80b35..d09f1bda6c7 100644 --- a/python/packages/core/tests/core/test_middleware_with_chat.py +++ b/python/packages/core/tests/core/test_middleware_with_chat.py @@ -497,7 +497,8 @@ async def test_message_injection_middleware_tool_enqueued_messages_wait_for_func messages=Message( role="assistant", contents=[Content.from_function_call(call_id="call-1", name="inject_message", arguments={})], - ) + ), + finish_reason="tool_calls", ), ChatResponse(messages=Message(role="assistant", contents=["done"])), ] From 2c3b994ff6f90646135b49c3741201c2f7a621b6 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 15:51:11 +0200 Subject: [PATCH 02/16] Python: harden Anthropic tool call commitment Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../agent_framework_anthropic/_chat_client.py | 59 +++- .../anthropic/tests/test_anthropic_client.py | 322 ++++++++++++++++-- 2 files changed, 346 insertions(+), 35 deletions(-) diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index cefc50b1223..36b645c54d5 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -255,6 +255,22 @@ def _map_finish_reason(stop_reason: str | None) -> FinishReason | None: return FinishReason(FINISH_REASON_MAP.get(stop_reason, stop_reason)) +def _resolve_finish_reason( + provider_finish_reason: str | None, + *, + has_function_calls: bool, + function_calls_committed: bool, +) -> FinishReason | None: + """Resolve the framework finish reason from provider and function-call completion evidence.""" + if function_calls_committed: + return FinishReason("tool_calls") + + finish_reason = _map_finish_reason(provider_finish_reason) + if has_function_calls and finish_reason == "tool_calls": + return None + return finish_reason + + class AnthropicSettings(TypedDict, total=False): """Anthropic Project settings. @@ -592,10 +608,41 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: # each message_delta carries the running total), so thread a per-stream # accumulator to _process_stream_event to emit increments instead. emitted_usage: dict[str, int] = {} + provider_finish_reason: str | None = None + has_function_calls = False + open_function_call_blocks: set[int] = set() mark_feature_used(FeatureIndex.ANTHROPIC) try: async for chunk in await self.anthropic_client.beta.messages.create(**run_options, stream=True): parsed_chunk = self._process_stream_event(chunk, emitted_usage) + if chunk.type == "message_delta": + if chunk.delta.stop_reason is not None: + provider_finish_reason = chunk.delta.stop_reason + if parsed_chunk: + parsed_chunk.finish_reason = None + elif chunk.type == "content_block_start" and parsed_chunk: + if any( + content.type == "function_call" and not content.informational_only + for content in parsed_chunk.contents + ): + has_function_calls = True + open_function_call_blocks.add(chunk.index) + elif chunk.type == "content_block_stop": + open_function_call_blocks.discard(chunk.index) + elif chunk.type == "message_stop": + yield ChatResponseUpdate( + finish_reason=_resolve_finish_reason( + provider_finish_reason, + has_function_calls=has_function_calls, + function_calls_committed=( + has_function_calls + and not open_function_call_blocks + and provider_finish_reason == "tool_use" + ), + ), + raw_representation=chunk, + ) + return if parsed_chunk: yield parsed_chunk except AgentFrameworkException: @@ -1134,18 +1181,26 @@ def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) -> Returns: A ChatResponse object containing the processed response. """ + contents = self._parse_contents_from_anthropic(message.content) + has_function_calls = any( + content.type == "function_call" and not content.informational_only for content in contents + ) return ChatResponse( response_id=message.id, messages=[ Message( role="assistant", - contents=self._parse_contents_from_anthropic(message.content), + contents=contents, raw_representation=message, ) ], usage_details=self._parse_usage_from_anthropic(message.usage), model=message.model, - finish_reason=_map_finish_reason(message.stop_reason), + finish_reason=_resolve_finish_reason( + message.stop_reason, + has_function_calls=has_function_calls, + function_calls_committed=has_function_calls and message.stop_reason == "tool_use", + ), response_format=options.get("response_format"), raw_representation=message, ) diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 133656e197f..8e628e99df0 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -41,7 +41,7 @@ from pydantic import BaseModel, Field from agent_framework_anthropic import AnthropicChatOptions, AnthropicClient, RawAnthropicClient -from agent_framework_anthropic._chat_client import AnthropicSettings +from agent_framework_anthropic._chat_client import AnthropicSettings, _resolve_finish_reason from agent_framework_anthropic._feature_usage import FeatureIndex # Test constants @@ -598,32 +598,26 @@ def test_streaming_replay_preserves_empty_signed_thinking_block( client = create_test_anthropic_client(mock_anthropic_client) events: list[BetaRawMessageStreamEvent] = [ - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 0, - "content_block": {"type": "thinking", "thinking": "", "signature": ""}, - } - ), - BetaRawContentBlockDeltaEvent.model_validate( - { - "type": "content_block_delta", - "index": 0, - "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, - } - ), - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 1, - "content_block": { - "type": "tool_use", - "id": "toolu_test", - "name": "lookup", - "input": {}, - }, - } - ), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }), + BetaRawContentBlockDeltaEvent.model_validate({ + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, + }), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_test", + "name": "lookup", + "input": {}, + }, + }), ] updates = [client._process_stream_event(event) for event in events] @@ -1635,6 +1629,77 @@ def test_process_message_with_tool_use(mock_anthropic_client: MagicMock) -> None assert response.finish_reason == "tool_calls" +@pytest.mark.parametrize( + ("stop_reason", "expected"), + [ + ("end_turn", "stop"), + ("max_tokens", "length"), + (None, None), + ], +) +def test_process_message_does_not_commit_incomplete_tool_use( + mock_anthropic_client: MagicMock, + stop_reason: str | None, + expected: str | None, +) -> None: + """A tool block is locally actionable only when the completed message authoritatively reports tool_use.""" + client = create_test_anthropic_client(mock_anthropic_client) + mock_message = MagicMock(spec=BetaMessage) + mock_message.id = "msg_123" + mock_message.model = "claude-3-5-sonnet-20241022" + mock_message.content = [ + BetaToolUseBlock( + type="tool_use", + id="call_123", + name="get_weather", + input={"location": "San Francisco"}, + ) + ] + mock_message.usage = BetaUsage(input_tokens=10, output_tokens=5) + mock_message.stop_reason = stop_reason + + response = client._process_message(mock_message, {}) + + assert response.messages[0].contents[0].type == "function_call" + assert response.finish_reason == expected + + +def test_resolve_finish_reason_uses_function_call_commitment() -> None: + """Committed calls override the provider reason while incomplete call evidence fails closed.""" + assert ( + _resolve_finish_reason( + "end_turn", + has_function_calls=True, + function_calls_committed=True, + ) + == "tool_calls" + ) + assert ( + _resolve_finish_reason( + "tool_use", + has_function_calls=True, + function_calls_committed=False, + ) + is None + ) + assert ( + _resolve_finish_reason( + "end_turn", + has_function_calls=False, + function_calls_committed=False, + ) + == "stop" + ) + assert ( + _resolve_finish_reason( + "tool_use", + has_function_calls=False, + function_calls_committed=False, + ) + == "tool_calls" + ) + + @pytest.mark.parametrize( ("stop_reason", "expected"), [ @@ -1854,6 +1919,199 @@ def test_parse_contents_server_tool_use_input_json_delta_ignored( # Stream Processing Tests +def _anthropic_stream_event( + event_type: str, + *, + index: int | None = None, + stop_reason: str | None = None, + content_block: Any | None = None, +) -> MagicMock: + event = MagicMock() + event.type = event_type + if index is not None: + event.index = index + if event_type == "message_start": + event.message.id = "msg_stream" + event.message.role = "assistant" + event.message.model = "claude-3-5-sonnet-20241022" + event.message.content = [] + event.message.stop_reason = None + event.message.usage = None + elif event_type == "message_delta": + event.delta.stop_reason = stop_reason + event.usage = None + elif event_type == "content_block_start": + event.content_block = content_block + return event + + +async def _collect_anthropic_stream_updates( + client: AnthropicClient, + mock_anthropic_client: MagicMock, + events: list[MagicMock], +) -> list[ChatResponseUpdate]: + async def mock_stream(): + for event in events: + yield event + + mock_anthropic_client.beta.messages.create.return_value = mock_stream() + return [ + update + async for update in client._inner_get_response( # type: ignore[attr-defined] # ty: ignore[not-iterable] + messages=[Message(role="user", contents=["Hi"])], + options=ChatOptions(max_tokens=10), + stream=True, + ) + ] + + +def _local_tool_use_block() -> MagicMock: + content_block = MagicMock() + content_block.type = "tool_use" + content_block.id = "call_stream" + content_block.name = "get_weather" + content_block.input = {} + return content_block + + +async def test_streaming_tool_call_updates_precede_message_stop_commitment( + mock_anthropic_client: MagicMock, +) -> None: + """Tool content remains incremental, but tool_calls is emitted only on message_stop.""" + client = create_test_anthropic_client(mock_anthropic_client) + argument_delta = MagicMock() + argument_delta.type = "input_json_delta" + argument_delta.partial_json = '{"location":"Paris"}' + delta_event = _anthropic_stream_event("content_block_delta", index=0) + delta_event.delta = argument_delta + + updates = await _collect_anthropic_stream_updates( + client, + mock_anthropic_client, + [ + _anthropic_stream_event("message_start"), + _anthropic_stream_event("content_block_start", index=0, content_block=_local_tool_use_block()), + delta_event, + _anthropic_stream_event("content_block_stop", index=0), + _anthropic_stream_event("message_delta", stop_reason="tool_use"), + _anthropic_stream_event("message_stop"), + ], + ) + + assert [update.finish_reason for update in updates] == [None, None, None, None, "tool_calls"] + assert updates[1].contents[0].call_id == "call_stream" + assert updates[2].contents[0].call_id == "call_stream" + assert updates[2].contents[0].arguments == '{"location":"Paris"}' + + +async def test_streaming_tool_call_without_message_stop_is_not_committed( + mock_anthropic_client: MagicMock, +) -> None: + """EOF after the provisional tool_use reason must not authorize the streamed call.""" + client = create_test_anthropic_client(mock_anthropic_client) + + updates = await _collect_anthropic_stream_updates( + client, + mock_anthropic_client, + [ + _anthropic_stream_event("message_start"), + _anthropic_stream_event("content_block_start", index=0, content_block=_local_tool_use_block()), + _anthropic_stream_event("content_block_stop", index=0), + _anthropic_stream_event("message_delta", stop_reason="tool_use"), + ], + ) + + assert any(content.type == "function_call" for update in updates for content in update.contents) + assert all(update.finish_reason != "tool_calls" for update in updates) + + +async def test_streaming_preserves_last_non_null_finish_reason( + mock_anthropic_client: MagicMock, +) -> None: + """A later metadata delta without a reason must not erase the pending terminal reason.""" + client = create_test_anthropic_client(mock_anthropic_client) + + updates = await _collect_anthropic_stream_updates( + client, + mock_anthropic_client, + [ + _anthropic_stream_event("message_start"), + _anthropic_stream_event("content_block_start", index=0, content_block=_local_tool_use_block()), + _anthropic_stream_event("content_block_stop", index=0), + _anthropic_stream_event("message_delta", stop_reason="tool_use"), + _anthropic_stream_event("message_delta", stop_reason=None), + _anthropic_stream_event("message_stop"), + ], + ) + + assert updates[-1].finish_reason == "tool_calls" + assert sum(update.finish_reason == "tool_calls" for update in updates) == 1 + + +@pytest.mark.parametrize( + ("close_tool_block", "stop_reason", "expected"), + [ + (True, "tool_use", "tool_calls"), + (False, "tool_use", None), + (True, "max_tokens", "length"), + ], +) +async def test_streaming_tool_call_commitment_requires_closed_block_and_completed_output( + mock_anthropic_client: MagicMock, + close_tool_block: bool, + stop_reason: str, + expected: str | None, +) -> None: + """Only a closed tool block followed by an authoritative tool_use stop is committed.""" + client = create_test_anthropic_client(mock_anthropic_client) + events = [ + _anthropic_stream_event("message_start"), + _anthropic_stream_event("content_block_start", index=0, content_block=_local_tool_use_block()), + ] + if close_tool_block: + events.append(_anthropic_stream_event("content_block_stop", index=0)) + events.extend([ + _anthropic_stream_event("message_delta", stop_reason=stop_reason), + _anthropic_stream_event("message_stop"), + ]) + + updates = await _collect_anthropic_stream_updates(client, mock_anthropic_client, events) + + assert updates[-1].finish_reason == expected + assert sum(update.finish_reason == "tool_calls" for update in updates) == (1 if expected == "tool_calls" else 0) + + +async def test_streaming_message_stop_emits_finish_reason_once_without_draining( + mock_anthropic_client: MagicMock, +) -> None: + """The terminal event resolves once and does not consume provider events after message_stop.""" + client = create_test_anthropic_client(mock_anthropic_client) + events_consumed = 0 + + async def mock_stream(): + nonlocal events_consumed + for event in ( + _anthropic_stream_event("message_delta", stop_reason="end_turn"), + _anthropic_stream_event("message_stop"), + _anthropic_stream_event("message_stop"), + ): + events_consumed += 1 + yield event + + mock_anthropic_client.beta.messages.create.return_value = mock_stream() + updates = [ + update + async for update in client._inner_get_response( # type: ignore[attr-defined] # ty: ignore[not-iterable] + messages=[Message(role="user", contents=["Hi"])], + options=ChatOptions(max_tokens=10), + stream=True, + ) + ] + + assert [update.finish_reason for update in updates] == [None, "stop"] + assert events_consumed == 2 + + def test_process_stream_event_simple(mock_anthropic_client: MagicMock) -> None: """Test _process_stream_event with simple mock event.""" client = create_test_anthropic_client(mock_anthropic_client) @@ -1944,8 +2202,8 @@ async def mock_stream(): if chunk: chunks.append(chunk) - # We should get at least some response (even if empty due to message_stop) - assert isinstance(chunks, list) + assert len(chunks) == 1 + assert chunks[0].finish_reason is None async def test_inner_get_response_ignores_options_stream_streaming( @@ -2033,11 +2291,9 @@ async def test_inner_get_response_streaming_wraps_sdk_errors(mock_anthropic_clie ): pass - # 2. Failure raised mid-stream, after at least one event has been yielded. + # 2. Failure raised mid-stream, after at least one non-terminal event has been yielded. async def _raise_after_first_event() -> Any: - event = MagicMock() - event.type = "message_stop" - yield event + yield _anthropic_stream_event("message_delta", stop_reason=None) raise _anthropic_status_error(anthropic_sdk.PermissionDeniedError, 403, "permission denied") mock_anthropic_client.beta.messages.create.side_effect = None From 118cc1f6d22354e722f67d96736f6d6c389f30e1 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 15:51:44 +0200 Subject: [PATCH 03/16] Fix AG-UI client tool call commitment Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ag-ui/agent_framework_ag_ui/_client.py | 24 +++ .../ag-ui/tests/ag_ui/test_ag_ui_client.py | 186 +++++++++++++++++- 2 files changed, 209 insertions(+), 1 deletion(-) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_client.py b/python/packages/ag-ui/agent_framework_ag_ui/_client.py index 7aa43062427..2c340b4160d 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_client.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_client.py @@ -21,6 +21,7 @@ ChatResponse, ChatResponseUpdate, Content, + FinishReason, FunctionTool, Message, ResponseStream, @@ -58,6 +59,22 @@ logger: logging.Logger = logging.getLogger("agent_framework.ag_ui") +def _resolve_finish_reason( + provider_finish_reason: FinishReason | None, + *, + has_function_calls: bool, + function_calls_committed: bool, +) -> FinishReason | None: + """Resolve the provider finish reason after function-call commitment.""" + if not has_function_calls: + return provider_finish_reason + if function_calls_committed: + return FinishReason("tool_calls") + if provider_finish_reason == "tool_calls": + return None + return provider_finish_reason + + def _unwrap_server_function_call_contents(contents: MutableSequence[Content | dict[str, Any]]) -> None: """Replace server_function_call instances with their underlying call content.""" for idx, content in enumerate(contents): @@ -495,6 +512,7 @@ async def _streaming_impl( logger.debug(f"[AGUIChatClient] Client tool set: {client_tool_set}") converter = AGUIEventConverter() + has_function_calls = False available_interrupts = options.get("available_interrupts", options.get("availableInterrupts")) @@ -522,6 +540,7 @@ async def _streaming_impl( ) if content.name in client_tool_set: # Client tool - let function invocation execute it + has_function_calls = True if not content.additional_properties: content.additional_properties = {} content.additional_properties["agui_thread_id"] = thread_id @@ -531,4 +550,9 @@ async def _streaming_impl( self._register_server_tool_placeholder(content.name) # type: ignore[arg-type] update.contents[i] = Content(type="server_function_call", function_call=content) # type: ignore + update.finish_reason = _resolve_finish_reason( + cast(FinishReason | None, update.finish_reason), + has_function_calls=has_function_calls, + function_calls_committed=str(event.get("type", "")).upper() == "RUN_FINISHED", + ) yield update diff --git a/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py b/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py index acb6af67e8a..3e7adbfefa0 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py +++ b/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py @@ -14,13 +14,14 @@ ChatResponse, ChatResponseUpdate, Content, + FinishReason, Message, ResponseStream, tool, ) from pytest import MonkeyPatch -from agent_framework_ag_ui._client import AGUIChatClient +from agent_framework_ag_ui._client import AGUIChatClient, _resolve_finish_reason from agent_framework_ag_ui._http_service import AGUIHttpService @@ -58,6 +59,33 @@ def inner_get_response( class TestAGUIChatClient: """Test suite for AGUIChatClient.""" + @pytest.mark.parametrize( + ("provider_finish_reason", "has_function_calls", "function_calls_committed", "expected"), + [ + (None, False, False, None), + (FinishReason("tool_calls"), False, False, "tool_calls"), + (FinishReason("stop"), True, False, "stop"), + (FinishReason("tool_calls"), True, False, None), + (FinishReason("stop"), True, True, "tool_calls"), + ], + ) + def test_resolve_finish_reason_requires_function_call_commitment( + self, + provider_finish_reason: FinishReason | None, + has_function_calls: bool, + function_calls_committed: bool, + expected: str | None, + ) -> None: + """Only committed function calls resolve to an authorizing finish reason.""" + assert ( + _resolve_finish_reason( + provider_finish_reason, + has_function_calls=has_function_calls, + function_calls_committed=function_calls_committed, + ) + == expected + ) + async def test_client_initialization(self) -> None: """Test client initialization.""" client = StubAGUIChatClient(endpoint="http://localhost:8888/") @@ -546,6 +574,161 @@ async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str assert response is not None + async def test_client_tool_updates_precede_run_finished_commit(self, monkeypatch: MonkeyPatch) -> None: + """Client tool updates stream immediately before RUN_FINISHED commits them.""" + + @tool + def client_tool(value: int) -> str: + """Return the supplied value.""" + return str(value) + + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"}, + {"type": "TOOL_CALL_START", "toolCallId": "call_1", "toolName": "client_tool"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call_1", "delta": '{"value": 1}'}, + {"type": "TOOL_CALL_END", "toolCallId": "call_1"}, + {"type": "RUN_FINISHED", "threadId": "thread_1", "runId": "run_1"}, + ] + + async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]: + for event in mock_events: + yield event + + client = StubAGUIChatClient(endpoint="http://localhost:8888/") + monkeypatch.setattr(client.http_service, "post_run", mock_post_run) + + stream = client.inner_get_response( + messages=[Message(role="user", contents=["Test"])], + options={"tools": [client_tool]}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + updates = [cast(ChatResponseUpdate, update) async for update in stream] + + function_updates = [ + (index, content.arguments) + for index, update in enumerate(updates) + for content in update.contents + if content.type == "function_call" + ] + assert function_updates == [(1, ""), (2, '{"value": 1}')] + assert [update.finish_reason for update in updates] == [None, None, None, "tool_calls"] + + async def test_client_tool_run_error_is_not_committed(self, monkeypatch: MonkeyPatch) -> None: + """RUN_ERROR leaves streamed client calls non-authorizing.""" + + @tool + def client_tool(value: int) -> str: + """Return the supplied value.""" + return str(value) + + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"}, + {"type": "TOOL_CALL_START", "toolCallId": "call_1", "toolName": "client_tool"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call_1", "delta": '{"value": 1}'}, + {"type": "TOOL_CALL_END", "toolCallId": "call_1"}, + {"type": "RUN_ERROR", "threadId": "thread_1", "runId": "run_1", "message": "failed"}, + ] + + async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]: + for event in mock_events: + yield event + + client = StubAGUIChatClient(endpoint="http://localhost:8888/") + monkeypatch.setattr(client.http_service, "post_run", mock_post_run) + + stream = client.inner_get_response( + messages=[Message(role="user", contents=["Test"])], + options={"tools": [client_tool]}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + updates = [cast(ChatResponseUpdate, update) async for update in stream] + + assert all(update.finish_reason != "tool_calls" for update in updates) + assert updates[-1].contents[0].type == "error" + + async def test_client_tool_transport_eof_is_not_committed(self, monkeypatch: MonkeyPatch) -> None: + """Transport EOF without RUN_FINISHED leaves client calls non-authorizing.""" + + @tool + def client_tool(value: int) -> str: + """Return the supplied value.""" + return str(value) + + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"}, + {"type": "TOOL_CALL_START", "toolCallId": "call_1", "toolName": "client_tool"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call_1", "delta": '{"value": 1}'}, + {"type": "TOOL_CALL_END", "toolCallId": "call_1"}, + ] + + async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]: + for event in mock_events: + yield event + + client = StubAGUIChatClient(endpoint="http://localhost:8888/") + monkeypatch.setattr(client.http_service, "post_run", mock_post_run) + + stream = client.inner_get_response( + messages=[Message(role="user", contents=["Test"])], + options={"tools": [client_tool]}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + updates = [cast(ChatResponseUpdate, update) async for update in stream] + + assert all(update.finish_reason != "tool_calls" for update in updates) + + async def test_committed_client_tool_executes_exactly_once(self, monkeypatch: MonkeyPatch) -> None: + """A committed client tool call executes once before the follow-up response.""" + executions: list[int] = [] + + @tool + def client_tool(value: int) -> str: + """Record and return the supplied value.""" + executions.append(value) + return str(value) + + call_count = 0 + + async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]: + nonlocal call_count + call_count += 1 + if call_count == 1: + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"}, + {"type": "TOOL_CALL_START", "toolCallId": "call_1", "toolName": "client_tool"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call_1", "delta": '{"value": 1}'}, + {"type": "TOOL_CALL_END", "toolCallId": "call_1"}, + {"type": "RUN_FINISHED", "threadId": "thread_1", "runId": "run_1"}, + ] + else: + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_2"}, + {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg_1", "delta": "done"}, + {"type": "RUN_FINISHED", "threadId": "thread_1", "runId": "run_2"}, + ] + for event in mock_events: + yield event + + client = StubAGUIChatClient(endpoint="http://localhost:8888/") + monkeypatch.setattr(client.http_service, "post_run", mock_post_run) + + response = await client.get_response( + [Message(role="user", contents=["Test"])], + options={"tools": [client_tool]}, + ) + content_types = [content.type for message in response.messages for content in message.contents] + + assert executions == [1] + assert call_count == 2 + assert response.text == "done" + assert content_types.count("function_result") == 1 + assert ( + content_types.index("function_call") < content_types.index("function_result") < content_types.index("text") + ) + async def test_server_tool_calls_unwrapped_after_invocation(self, monkeypatch: MonkeyPatch) -> None: """Ensure server-side tool calls are exposed as FunctionCallContent after processing.""" @@ -576,6 +759,7 @@ async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str assert function_calls[0].name == "get_time_zone" assert not any(content.type == "server_function_call" for update in updates for content in update.contents) + assert updates[-1].finish_reason == "stop" async def test_server_tool_calls_not_executed_locally(self, monkeypatch: MonkeyPatch) -> None: """Server tools should not trigger local function invocation even when client tools exist.""" From 481d5bb8295297c97efa19a50041df3aa9ddc6e0 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 15:57:04 +0200 Subject: [PATCH 04/16] Update tool call fixtures with finish reasons Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/core/tests/core/test_agents.py | 12 ++++++++---- .../core/tests/core/test_harness_agent.py | 1 + .../core/tests/core/test_observability.py | 4 ++++ .../workflow/test_agent_executor_tool_calls.py | 16 ++++++++++++---- 4 files changed, 25 insertions(+), 8 deletions(-) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index 4ef0d031d72..af802db74d2 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -638,7 +638,7 @@ def lookup_weather(location: str) -> str: ) ], role="assistant", - finish_reason="stop", + finish_reason="tool_calls", response_id="resp_call_1", ) ], @@ -698,7 +698,7 @@ def lookup_weather(location: str) -> str: ) ], role="assistant", - finish_reason="stop", + finish_reason="tool_calls", response_id="resp_call_1", ) ], @@ -818,6 +818,7 @@ def lookup_weather(location: str) -> str: ), conversation_id="resp_call_1", response_id="resp_call_1", + finish_reason="tool_calls", ) mock_get_non_streaming_response = AsyncMock( side_effect=[first_response, RuntimeError("service down")], @@ -864,7 +865,7 @@ async def _first_stream_updates() -> AsyncIterable[ChatResponseUpdate]: ) ], role="assistant", - finish_reason="stop", + finish_reason="tool_calls", ) def _finalize_first_stream(_updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]: @@ -881,6 +882,7 @@ def _finalize_first_stream(_updates: Sequence[ChatResponseUpdate]) -> ChatRespon ), conversation_id="resp_call_1", response_id="resp_call_1", + finish_reason="tool_calls", ) first_stream = ResponseStream(_first_stream_updates(), finalizer=_finalize_first_stream) @@ -3303,6 +3305,7 @@ def _inner_get_response( # type: ignore[override] store_and_echo = self._effective_store(options) and self._echo_conversation_id conv_id = _PSC_SERVICE_CONVERSATION_ID if store_and_echo else None contents = self._next_contents() + finish_reason = "tool_calls" if any(content.type == "function_call" for content in contents) else "stop" if stream: @@ -3311,7 +3314,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: yield ChatResponseUpdate( contents=contents, role="assistant", - finish_reason="stop", + finish_reason=finish_reason, conversation_id=conv_id, ) @@ -3328,6 +3331,7 @@ async def _get() -> ChatResponse: return ChatResponse( messages=Message(role="assistant", contents=contents), conversation_id=conv_id, + finish_reason=finish_reason, ) return _get() diff --git a/python/packages/core/tests/core/test_harness_agent.py b/python/packages/core/tests/core/test_harness_agent.py index 6535a6fc68f..a2b6ad3af04 100644 --- a/python/packages/core/tests/core/test_harness_agent.py +++ b/python/packages/core/tests/core/test_harness_agent.py @@ -1527,6 +1527,7 @@ def build_updates(call_index: int) -> list[ChatResponseUpdate]: Content.from_function_call(call_id="call_1", name="lookup", arguments={"query": "widgets"}) ], role="assistant", + finish_reason="tool_calls", ), ] return [ChatResponseUpdate(contents=[Content.from_text("Done.")], role="assistant", finish_reason="stop")] diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index f496e44d07f..9fb3b3cbaea 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -4825,6 +4825,7 @@ async def _get() -> ChatResponse: ], ) ], + finish_reason="tool_calls", ) return ChatResponse( messages=[Message(role="assistant", contents=["The weather in Seattle is sunny!"])], @@ -6277,6 +6278,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: ) ], role="assistant", + finish_reason="tool_calls", ) else: yield ChatResponseUpdate( @@ -6305,6 +6307,7 @@ async def _get() -> ChatResponse: ], ) ], + finish_reason="tool_calls", ) return ChatResponse( messages=[Message(role="assistant", contents=["The weather in Seattle is sunny!"])], @@ -6405,6 +6408,7 @@ async def _get_tool_calls() -> ChatResponse: ], ) ], + finish_reason="tool_calls", ) return _get_tool_calls() diff --git a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py index cee0b1570e8..cd4856f1b29 100644 --- a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py +++ b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py @@ -209,7 +209,8 @@ def _create_response(self) -> ChatResponse: call_id="2", name="mock_tool_requiring_approval", arguments='{"query": "test"}' ), ], - ) + ), + finish_reason="tool_calls", ) else: response = ChatResponse( @@ -220,7 +221,8 @@ def _create_response(self) -> ChatResponse: call_id="1", name="mock_tool_requiring_approval", arguments='{"query": "test"}' ) ], - ) + ), + finish_reason="tool_calls", ) else: response = ChatResponse(messages=Message("assistant", ["Tool executed successfully."])) @@ -242,6 +244,7 @@ async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: ), ], role="assistant", + finish_reason="tool_calls", ) else: yield ChatResponseUpdate( @@ -251,6 +254,7 @@ async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: ) ], role="assistant", + finish_reason="tool_calls", ) else: yield ChatResponseUpdate(contents=[Content.from_text(text="Tool executed ")], role="assistant") @@ -549,7 +553,8 @@ def _create_response(self) -> ChatResponse: call_id="2", name="client_side_tool", arguments='{"query": "test2"}' ), ], - ) + ), + finish_reason="tool_calls", ) else: response = ChatResponse( @@ -560,7 +565,8 @@ def _create_response(self) -> ChatResponse: call_id="1", name="client_side_tool", arguments='{"query": "test"}' ) ], - ) + ), + finish_reason="tool_calls", ) else: response = ChatResponse(messages=Message("assistant", ["Tool executed successfully."])) @@ -579,6 +585,7 @@ async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: ), ], role="assistant", + finish_reason="tool_calls", ) else: yield ChatResponseUpdate( @@ -586,6 +593,7 @@ async def _stream_response(self) -> AsyncIterable[ChatResponseUpdate]: Content.from_function_call(call_id="1", name="client_side_tool", arguments='{"query": "test"}') ], role="assistant", + finish_reason="tool_calls", ) else: yield ChatResponseUpdate(contents=[Content.from_text(text="Tool executed ")], role="assistant") From 1b15c40118931ea24be4e59fb53c0a06cc38fee3 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:01:24 +0200 Subject: [PATCH 05/16] Mark committed calls in AG-UI test client Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/ag-ui/tests/ag_ui/conftest.py | 36 +++++++++++++++---- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/python/packages/ag-ui/tests/ag_ui/conftest.py b/python/packages/ag-ui/tests/ag_ui/conftest.py index c4d4d972cde..9208776be8f 100644 --- a/python/packages/ag-ui/tests/ag_ui/conftest.py +++ b/python/packages/ag-ui/tests/ag_ui/conftest.py @@ -38,6 +38,16 @@ ResponseFn = Callable[..., Awaitable[ChatResponse]] +def _finish_committed_test_calls(response: ChatResponse) -> ChatResponse: + if response.finish_reason is None and any( + content.type == "function_call" and not content.informational_only + for message in response.messages + for content in message.contents + ): + response.finish_reason = "tool_calls" + return response + + def pytest_configure() -> None: """Ensure this test directory is on sys.path so helper modules can be imported by name.""" test_dir = str(Path(__file__).resolve().parent) @@ -129,11 +139,23 @@ def _inner_get_response( **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: if stream: + async def _stream() -> AsyncIterator[ChatResponseUpdate]: + has_function_calls = False + has_finish_reason = False + async for update in self._stream_fn(messages, options, **kwargs): + has_function_calls = has_function_calls or any( + content.type == "function_call" and not content.informational_only + for content in update.contents + ) + has_finish_reason = has_finish_reason or update.finish_reason is not None + yield update + if has_function_calls and not has_finish_reason: + yield ChatResponseUpdate(finish_reason="tool_calls") def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: - return ChatResponse.from_updates(updates) + return _finish_committed_test_calls(ChatResponse.from_updates(updates)) - return ResponseStream(self._stream_fn(messages, options, **kwargs), finalizer=_finalize) + return ResponseStream(_stream(), finalizer=_finalize) return self._get_response_impl(messages, options, **kwargs) @@ -142,15 +164,17 @@ async def _get_response_impl( ) -> ChatResponse: """Non-streaming implementation.""" if self._response_fn is not None: - return await self._response_fn(messages, options, **kwargs) + return _finish_committed_test_calls(await self._response_fn(messages, options, **kwargs)) contents: list[Any] = [] async for update in self._stream_fn(list(messages), dict(options), **kwargs): contents.extend(update.contents) - return ChatResponse( - messages=[Message(role="assistant", contents=contents)], - response_id="stub-response", + return _finish_committed_test_calls( + ChatResponse( + messages=[Message(role="assistant", contents=contents)], + response_id="stub-response", + ) ) From 62c2ff84488bd6ce58778d8d95c6c32b406c8485 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:00:17 +0200 Subject: [PATCH 06/16] fix(gemini): require committed tool calls Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../agent_framework_gemini/_chat_client.py | 50 ++++- .../gemini/tests/test_gemini_client.py | 202 +++++++++++++++++- 2 files changed, 242 insertions(+), 10 deletions(-) diff --git a/python/packages/gemini/agent_framework_gemini/_chat_client.py b/python/packages/gemini/agent_framework_gemini/_chat_client.py index a6c6796b418..601b1d27c87 100644 --- a/python/packages/gemini/agent_framework_gemini/_chat_client.py +++ b/python/packages/gemini/agent_framework_gemini/_chat_client.py @@ -324,11 +324,21 @@ def _validate_client_auth_configuration( "IMAGE_SAFETY": "content_filter", "IMAGE_PROHIBITED_CONTENT": "content_filter", "IMAGE_RECITATION": "content_filter", - "MALFORMED_FUNCTION_CALL": "tool_calls", - "UNEXPECTED_TOOL_CALL": "tool_calls", } +def _resolve_finish_reason( + provider_finish_reason: FinishReasonLiteral | FinishReason | None, + *, + has_function_calls: bool, + function_calls_committed: bool, +) -> FinishReasonLiteral | FinishReason | None: + """Resolve the public finish reason without authorizing uncommitted function calls.""" + if has_function_calls and function_calls_committed: + return "tool_calls" + return provider_finish_reason + + class RawGeminiChatClient( BaseChatClient[GeminiChatOptionsT], Generic[GeminiChatOptionsT], @@ -585,12 +595,18 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: cast(Any, self._genai_client.aio.models).generate_content_stream, ) try: + has_function_calls = False async for chunk in await generate_content_stream( model=model, contents=contents, config=config, ): - yield self._process_chunk(chunk) + update = self._process_chunk(chunk, has_function_calls=has_function_calls) + has_function_calls = has_function_calls or any( + content.type == "function_call" and not content.informational_only + for content in update.contents + ) + yield update except AgentFrameworkException: raise except Exception as ex: @@ -1149,25 +1165,37 @@ def _process_generate_response( candidate = response.candidates[0] if response.candidates else None parts: list[types.Part] = (candidate.content.parts or []) if candidate and candidate.content else [] contents = self._parse_parts(parts) + has_function_calls = any( + content.type == "function_call" and not content.informational_only for content in contents + ) + provider_finish_reason = candidate.finish_reason.name if candidate and candidate.finish_reason else None return ChatResponse( response_id=None, messages=[Message(role="assistant", contents=contents, raw_representation=candidate)], usage_details=self._parse_usage(response.usage_metadata), model=response.model_version or self.model, - finish_reason=self._map_finish_reason( - candidate.finish_reason.name if candidate and candidate.finish_reason else None + finish_reason=_resolve_finish_reason( + self._map_finish_reason(provider_finish_reason), + has_function_calls=has_function_calls, + function_calls_committed=has_function_calls and provider_finish_reason == "STOP", ), response_format=response_format, raw_representation=response, ) - def _process_chunk(self, chunk: types.GenerateContentResponse) -> ChatResponseUpdate: + def _process_chunk( + self, + chunk: types.GenerateContentResponse, + *, + has_function_calls: bool = False, + ) -> ChatResponseUpdate: """Convert a single streaming chunk to a framework ChatResponseUpdate. Usage details are attached only to the final chunk, identified by a non-None finish reason. Args: chunk: A streaming ``GenerateContentResponse`` chunk from the Gemini API. + has_function_calls: Whether an earlier streaming update contained an actionable function call. Returns: A ``ChatResponseUpdate`` with parsed contents, finish reason, and model ID. @@ -1175,9 +1203,15 @@ def _process_chunk(self, chunk: types.GenerateContentResponse) -> ChatResponseUp candidate = chunk.candidates[0] if chunk.candidates else None parts: list[types.Part] = (candidate.content.parts or []) if candidate and candidate.content else [] contents = self._parse_parts(parts) + has_function_calls = has_function_calls or any( + content.type == "function_call" and not content.informational_only for content in contents + ) + provider_finish_reason = candidate.finish_reason.name if candidate and candidate.finish_reason else None - finish_reason = self._map_finish_reason( - candidate.finish_reason.name if candidate and candidate.finish_reason else None + finish_reason = _resolve_finish_reason( + self._map_finish_reason(provider_finish_reason), + has_function_calls=has_function_calls, + function_calls_committed=has_function_calls and provider_finish_reason == "STOP", ) # Attach usage to the final chunk only (when finish_reason is set). diff --git a/python/packages/gemini/tests/test_gemini_client.py b/python/packages/gemini/tests/test_gemini_client.py index a2b4908d7ad..a875ebbe686 100644 --- a/python/packages/gemini/tests/test_gemini_client.py +++ b/python/packages/gemini/tests/test_gemini_client.py @@ -542,8 +542,8 @@ async def test_get_response_no_usage_when_metadata_absent() -> None: ("IMAGE_SAFETY", "content_filter"), ("IMAGE_PROHIBITED_CONTENT", "content_filter"), ("IMAGE_RECITATION", "content_filter"), - ("MALFORMED_FUNCTION_CALL", "tool_calls"), - ("UNEXPECTED_TOOL_CALL", "tool_calls"), + ("MALFORMED_FUNCTION_CALL", "MALFORMED_FUNCTION_CALL"), + ("UNEXPECTED_TOOL_CALL", "UNEXPECTED_TOOL_CALL"), # Real google-genai FinishReason values with no entry in _FINISH_REASON_MAP: must now # pass through as the raw string instead of being silently dropped to None. ("OTHER", "OTHER"), @@ -604,6 +604,204 @@ async def test_unmapped_finish_reason_still_attaches_usage_on_streamed_final_chu assert any(c.type == "usage" for c in updates[-1].contents) +@pytest.mark.parametrize( + ("provider_finish_reason", "expected_finish_reason"), + [ + ("STOP", "tool_calls"), + ("MALFORMED_FUNCTION_CALL", "MALFORMED_FUNCTION_CALL"), + ("UNEXPECTED_TOOL_CALL", "UNEXPECTED_TOOL_CALL"), + ("SAFETY", "content_filter"), + ("MAX_TOKENS", "length"), + ("TOO_MANY_TOOL_CALLS", "TOO_MANY_TOOL_CALLS"), + ("OTHER", "OTHER"), + (None, None), + ], +) +async def test_non_streaming_function_calls_require_authoritative_terminal_commitment( + provider_finish_reason: str | None, + expected_finish_reason: str | None, +) -> None: + """Only a normal terminal candidate authoritatively commits local function calls.""" + client, mock = _make_gemini_client() + mock.aio.models.generate_content = AsyncMock( + return_value=_make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason=provider_finish_reason, + ) + ) + + response = await client.get_response(messages=[Message(role="user", contents=[Content.from_text("Search")])]) + + assert response.finish_reason == expected_finish_reason + assert len(response.messages[0].contents) == 1 + function_call = response.messages[0].contents[0] + assert function_call.type == "function_call" + assert function_call.call_id == "call-1" + assert function_call.name == "search" + assert function_call.arguments == {"q": "framework"} + + +async def test_non_streaming_server_side_tool_call_does_not_override_finish_reason() -> None: + """Informational server-side calls are not actionable local function calls.""" + client, mock = _make_gemini_client() + mock.aio.models.generate_content = AsyncMock( + return_value=_make_response( + [_make_part(tool_call=("search-1", types.ToolType.FILE_SEARCH, {"query": "framework"}))], + finish_reason="STOP", + ) + ) + + response = await client.get_response(messages=[Message(role="user", contents=[Content.from_text("Search")])]) + + assert response.finish_reason == "stop" + assert response.messages[0].contents[0].informational_only is True + + +async def test_streaming_committed_function_call_precedes_terminal_tool_calls_update() -> None: + """A streamed call remains immediate and a later normal terminal candidate commits it.""" + client, mock = _make_gemini_client() + chunks = [ + _make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason=None, + prompt_tokens=None, + output_tokens=None, + ), + _make_response([_make_part(text="Searching")], finish_reason="STOP"), + ] + mock.aio.models.generate_content_stream = AsyncMock(return_value=_async_iter(chunks)) + + stream = client.get_response( + messages=[Message(role="user", contents=[Content.from_text("Search")])], + stream=True, + ) + updates = [update async for update in stream] + final = await stream.get_final_response() + + assert [update.finish_reason for update in updates] == [None, "tool_calls"] + assert [content.type for content in updates[0].contents] == ["function_call"] + assert updates[0].contents[0].call_id == "call-1" + assert updates[0].contents[0].arguments == {"q": "framework"} + assert updates[1].text == "Searching" + assert final.finish_reason == "tool_calls" + assert [content.type for content in final.messages[0].contents] == ["function_call", "text"] + + +async def test_streaming_terminal_candidate_commits_function_call_in_same_update() -> None: + """A normal terminal candidate can commit an actionable call carried in the same update.""" + client, mock = _make_gemini_client() + chunks = [ + _make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason="STOP", + ) + ] + mock.aio.models.generate_content_stream = AsyncMock(return_value=_async_iter(chunks)) + + stream = client.get_response( + messages=[Message(role="user", contents=[Content.from_text("Search")])], + stream=True, + ) + updates = [update async for update in stream] + final = await stream.get_final_response() + + assert len(updates) == 1 + assert updates[0].finish_reason == "tool_calls" + assert updates[0].contents[0].call_id == "call-1" + assert final.finish_reason == "tool_calls" + + +@pytest.mark.parametrize( + ("provider_finish_reason", "expected_finish_reason"), + [ + ("MALFORMED_FUNCTION_CALL", "MALFORMED_FUNCTION_CALL"), + ("UNEXPECTED_TOOL_CALL", "UNEXPECTED_TOOL_CALL"), + ("SAFETY", "content_filter"), + ("MAX_TOKENS", "length"), + ("TOO_MANY_TOOL_CALLS", "TOO_MANY_TOOL_CALLS"), + ("OTHER", "OTHER"), + ], +) +async def test_streaming_function_calls_are_not_committed_by_abnormal_terminal_reason( + provider_finish_reason: str, + expected_finish_reason: str, +) -> None: + """Abnormal terminal candidates do not authorize streamed local function calls.""" + client, mock = _make_gemini_client() + chunks = [ + _make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason=None, + prompt_tokens=None, + output_tokens=None, + ), + _make_response([], finish_reason=provider_finish_reason), + ] + mock.aio.models.generate_content_stream = AsyncMock(return_value=_async_iter(chunks)) + + stream = client.get_response( + messages=[Message(role="user", contents=[Content.from_text("Search")])], + stream=True, + ) + updates = [update async for update in stream] + final = await stream.get_final_response() + + assert [update.finish_reason for update in updates] == [None, expected_finish_reason] + assert final.finish_reason == expected_finish_reason + + +async def test_streaming_function_call_without_terminal_candidate_is_not_committed_at_eof() -> None: + """EOF alone does not synthesize authority for a streamed local function call.""" + client, mock = _make_gemini_client() + chunks = [ + _make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason=None, + prompt_tokens=None, + output_tokens=None, + ) + ] + mock.aio.models.generate_content_stream = AsyncMock(return_value=_async_iter(chunks)) + + stream = client.get_response( + messages=[Message(role="user", contents=[Content.from_text("Search")])], + stream=True, + ) + updates = [update async for update in stream] + final = await stream.get_final_response() + + assert len(updates) == 1 + assert updates[0].finish_reason is None + assert final.finish_reason is None + + +async def test_streaming_function_call_is_not_committed_by_chunk_without_candidate() -> None: + """A blocked or otherwise candidate-less chunk provides no terminal commitment evidence.""" + client, mock = _make_gemini_client() + candidate_less_chunk = _make_response([]) + candidate_less_chunk.candidates = [] + chunks = [ + _make_response( + [_make_part(function_call=("call-1", "search", {"q": "framework"}))], + finish_reason=None, + prompt_tokens=None, + output_tokens=None, + ), + candidate_less_chunk, + ] + mock.aio.models.generate_content_stream = AsyncMock(return_value=_async_iter(chunks)) + + stream = client.get_response( + messages=[Message(role="user", contents=[Content.from_text("Search")])], + stream=True, + ) + updates = [update async for update in stream] + final = await stream.get_final_response() + + assert [update.finish_reason for update in updates] == [None, None] + assert final.finish_reason is None + + # message conversion From 519b2114568d2e06afc7133ecedf07fe636b1fa7 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:03:47 +0200 Subject: [PATCH 07/16] Align Gemini finish reason resolver Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../gemini/agent_framework_gemini/_chat_client.py | 6 +++++- python/packages/gemini/tests/test_gemini_client.py | 14 +++++++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/python/packages/gemini/agent_framework_gemini/_chat_client.py b/python/packages/gemini/agent_framework_gemini/_chat_client.py index 601b1d27c87..aab0be438fe 100644 --- a/python/packages/gemini/agent_framework_gemini/_chat_client.py +++ b/python/packages/gemini/agent_framework_gemini/_chat_client.py @@ -334,8 +334,12 @@ def _resolve_finish_reason( function_calls_committed: bool, ) -> FinishReasonLiteral | FinishReason | None: """Resolve the public finish reason without authorizing uncommitted function calls.""" - if has_function_calls and function_calls_committed: + if not has_function_calls: + return provider_finish_reason + if function_calls_committed: return "tool_calls" + if provider_finish_reason == "tool_calls": + return None return provider_finish_reason diff --git a/python/packages/gemini/tests/test_gemini_client.py b/python/packages/gemini/tests/test_gemini_client.py index a875ebbe686..9c89121abc0 100644 --- a/python/packages/gemini/tests/test_gemini_client.py +++ b/python/packages/gemini/tests/test_gemini_client.py @@ -11,7 +11,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from agent_framework import Agent, Content, FunctionTool, Message +from agent_framework import Agent, Content, FinishReason, FunctionTool, Message from agent_framework._settings import SecretString from agent_framework.exceptions import ( ChatClientException, @@ -24,6 +24,7 @@ from typing_extensions import NotRequired, TypedDict from agent_framework_gemini import GeminiChatClient, GeminiChatOptions, RawGeminiChatClient, ThinkingConfig +from agent_framework_gemini._chat_client import _resolve_finish_reason from agent_framework_gemini._feature_usage import FeatureIndex @@ -641,6 +642,17 @@ async def test_non_streaming_function_calls_require_authoritative_terminal_commi assert function_call.arguments == {"q": "framework"} +def test_resolve_finish_reason_does_not_trust_uncommitted_tool_calls_reason() -> None: + assert ( + _resolve_finish_reason( + FinishReason("tool_calls"), + has_function_calls=True, + function_calls_committed=False, + ) + is None + ) + + async def test_non_streaming_server_side_tool_call_does_not_override_finish_reason() -> None: """Informational server-side calls are not actionable local function calls.""" client, mock = _make_gemini_client() From 4507afd289e6666502865047448f089ca0d26b28 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 15:54:56 +0200 Subject: [PATCH 08/16] Python: Harden Responses tool call commitment Require terminal completion evidence before exposing tool_calls as the finalized finish reason in OpenAI Responses streams and non-streaming responses. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../agent_framework_openai/_chat_client.py | 114 +++++-- .../tests/openai/test_openai_chat_client.py | 293 +++++++++++++++++- 2 files changed, 376 insertions(+), 31 deletions(-) diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 0cc2bc3e34c..0028ffd5a2c 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -156,6 +156,25 @@ def _is_refusal_text_content(content: Content) -> bool: _AZURE_AI_SEARCH_OUTPUT_EVENT_TYPES = {"response.output_item.added", "response.output_item.done"} _AZURE_AI_SEARCH_OUTPUT_EVENT_PREFIX = "response.azure_ai_search_call_output." +_FUNCTION_CALL_ARGUMENTS_DONE = 1 +_FUNCTION_CALL_OUTPUT_ITEM_DONE = 2 +_FUNCTION_CALL_COMMITTED = _FUNCTION_CALL_ARGUMENTS_DONE | _FUNCTION_CALL_OUTPUT_ITEM_DONE + + +def _resolve_finish_reason( + provider_finish_reason: FinishReason | None, + *, + has_function_calls: bool, + function_calls_committed: bool, +) -> FinishReason | None: + """Resolve function-call authorization without trusting a speculative provider reason.""" + if has_function_calls and function_calls_committed: + return FinishReason("tool_calls") + if provider_finish_reason == "tool_calls": + return None + return provider_finish_reason + + # Internal marker emitted by `_prepare_content_for_openai` for an # `mcp_server_tool_result` Content. The Responses API expects an `mcp_call` # input item to carry both arguments and output as one item, so result @@ -716,6 +735,37 @@ def _inner_get_response( # Captured once request options are validated/prepared so the streaming finalizer can # still parse the aggregated response into structured output after the stream completes. response_format: Any | None = None + function_call_commitments: dict[tuple[int, str], int] = {} + + def _parse_stream_chunk(chunk: OpenAIResponseStreamEvent) -> ChatResponseUpdate: + terminal_response: Any | None = None + match chunk.type: + case "response.function_call_arguments.done": + key = (chunk.output_index, chunk.item_id) + function_call_commitments[key] = ( + function_call_commitments.get(key, 0) | _FUNCTION_CALL_ARGUMENTS_DONE + ) + case "response.output_item.done" if getattr(chunk.item, "type", None) == "function_call": + key = (chunk.output_index, getattr(chunk.item, "id", None) or "") + function_call_commitments[key] = ( + function_call_commitments.get(key, 0) | _FUNCTION_CALL_OUTPUT_ITEM_DONE + ) + case "response.completed" | "response.incomplete" | "response.failed": + terminal_response = chunk.response + case _: + pass + update = self._parse_chunk_from_openai( + chunk, + options=validated_options or {}, + function_call_ids=function_call_ids, + seen_reasoning_delta_item_ids=seen_reasoning_delta_item_ids, + ) + if terminal_response is not None: + update.finish_reason = self._get_finish_reason_from_openai_response( + terminal_response, + function_call_commitments=function_call_commitments, + ) + return update def _finalize_with_captured_format(updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]: # ResponseStream only calls the finalizer after iterating or draining `_stream()`, @@ -745,12 +795,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: served_model = self._extract_served_model(getattr(raw_stream_response, "headers", None)) async with _open_event_stream(raw_stream_response) as stream_response: async for chunk in stream_response: - update = self._parse_chunk_from_openai( - chunk, - options=validated_options, - function_call_ids=function_call_ids, - seen_reasoning_delta_item_ids=seen_reasoning_delta_item_ids, - ) + update = _parse_stream_chunk(chunk) if served_model is not None: update.model = served_model yield update @@ -775,12 +820,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: # surface the served-model header. async with client.responses.stream(**run_options) as response: async for chunk in response: - yield self._parse_chunk_from_openai( - chunk, - options=validated_options, - function_call_ids=function_call_ids, - seen_reasoning_delta_item_ids=seen_reasoning_delta_item_ids, - ) + yield _parse_stream_chunk(chunk) else: raw_create_response = await client.responses.with_raw_response.create( stream=True, **run_options @@ -789,12 +829,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: served_model = self._extract_served_model(getattr(raw_create_response, "headers", None)) async with _open_event_stream(raw_create_response) as stream_response: async for chunk in stream_response: - update = self._parse_chunk_from_openai( - chunk, - options=validated_options, - function_call_ids=function_call_ids, - seen_reasoning_delta_item_ids=seen_reasoning_delta_item_ids, - ) + update = _parse_stream_chunk(chunk) if served_model is not None: update.model = served_model yield update @@ -2629,18 +2664,43 @@ def _parse_hosted_function_call_content( ) # region Parse methods - def _get_finish_reason_from_openai_response(self, response: Any) -> FinishReason | None: + def _get_finish_reason_from_openai_response( + self, + response: Any, + *, + function_call_commitments: Mapping[tuple[int, str], int] | None = None, + ) -> FinishReason | None: """Get the framework finish reason from a terminal Responses API response.""" incomplete_reason = getattr(getattr(response, "incomplete_details", None), "reason", None) if incomplete_reason == "content_filter": - return FinishReason("content_filter") - if incomplete_reason == "max_output_tokens": - return FinishReason("length") - if getattr(response, "status", None) != "completed": - return None - if any(getattr(item, "type", None) == "function_call" for item in getattr(response, "output", ())): - return FinishReason("tool_calls") - return FinishReason("stop") + provider_finish_reason = FinishReason("content_filter") + elif incomplete_reason == "max_output_tokens": + provider_finish_reason = FinishReason("length") + elif getattr(response, "status", None) == "completed": + provider_finish_reason = FinishReason("stop") + else: + provider_finish_reason = None + + function_calls = [ + (output_index, item) + for output_index, item in enumerate(getattr(response, "output", ())) + if getattr(item, "type", None) == "function_call" + ] + if function_call_commitments is None: + function_calls_committed = getattr(response, "status", None) == "completed" and all( + getattr(item, "status", None) == "completed" for _, item in function_calls + ) + else: + function_calls_committed = getattr(response, "status", None) == "completed" and all( + function_call_commitments.get((output_index, getattr(item, "id", None) or ""), 0) + == _FUNCTION_CALL_COMMITTED + for output_index, item in function_calls + ) + return _resolve_finish_reason( + provider_finish_reason, + has_function_calls=bool(function_calls), + function_calls_committed=function_calls_committed, + ) def _parse_response_from_openai( self, diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 7258c50cae9..15cdeec28d8 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -7,7 +7,7 @@ from collections.abc import AsyncGenerator, Iterator, Sequence from datetime import datetime, timezone from pathlib import Path -from typing import Annotated, Any, cast +from typing import Annotated, Any, Literal, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -42,6 +42,16 @@ SettingNotFoundError, ) from openai import AsyncOpenAI, BadRequestError +from openai.types.responses import ( + Response, + ResponseCompletedEvent, + ResponseFailedEvent, + ResponseFunctionCallArgumentsDoneEvent, + ResponseFunctionToolCall, + ResponseIncompleteEvent, + ResponseOutputItemAddedEvent, + ResponseOutputItemDoneEvent, +) from openai.types.responses.response_reasoning_item import Summary from openai.types.responses.response_reasoning_summary_text_delta_event import ( ResponseReasoningSummaryTextDeltaEvent, @@ -60,7 +70,10 @@ from pytest import param from agent_framework_openai import OpenAIChatClient, OpenAIChatOptions, RawOpenAIChatClient -from agent_framework_openai._chat_client import OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY +from agent_framework_openai._chat_client import ( + OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY, + _resolve_finish_reason, +) from agent_framework_openai._exceptions import OpenAIContentFilterException skip_if_openai_integration_tests_disabled = pytest.mark.skipif( @@ -7363,6 +7376,34 @@ def test_streaming_response_completed_sets_created_at() -> None: assert update.created_at == "2001-09-09T01:46:40.000000Z" +@pytest.mark.parametrize( + ("provider_finish_reason", "has_function_calls", "function_calls_committed", "expected_finish_reason"), + [ + ("stop", False, False, "stop"), + ("length", False, True, "length"), + ("tool_calls", False, False, None), + ("stop", True, True, "tool_calls"), + ("length", True, True, "tool_calls"), + ("tool_calls", True, False, None), + ("stop", True, False, "stop"), + ], +) +def test_resolve_finish_reason_requires_committed_function_calls( + provider_finish_reason: str, + has_function_calls: bool, + function_calls_committed: bool, + expected_finish_reason: str | None, +) -> None: + """Function-call evidence controls authorization while ordinary reasons pass through.""" + finish_reason = _resolve_finish_reason( + cast(Any, provider_finish_reason), + has_function_calls=has_function_calls, + function_calls_committed=function_calls_committed, + ) + + assert finish_reason == expected_finish_reason + + @pytest.mark.parametrize( ("status", "incomplete_reason", "output_type", "expected_finish_reason"), [ @@ -7385,7 +7426,7 @@ def test_get_finish_reason_from_openai_response( mock_response = MagicMock() mock_response.status = status mock_response.incomplete_details = MagicMock(reason=incomplete_reason) if incomplete_reason is not None else None - mock_response.output = [MagicMock(type=output_type)] if output_type is not None else [] + mock_response.output = [MagicMock(type=output_type, status="completed")] if output_type is not None else [] finish_reason = client._get_finish_reason_from_openai_response(mock_response) @@ -7411,6 +7452,250 @@ def test_parse_response_from_openai_sets_finish_reason() -> None: assert response.finish_reason == "stop" +def _typed_function_call( + index: int, + *, + status: Literal["in_progress", "completed", "incomplete"] = "completed", +) -> ResponseFunctionToolCall: + return ResponseFunctionToolCall( + arguments=json.dumps({"index": index}), + call_id=f"call_{index}", + id=f"fc_{index}", + name="lookup", + status=status, + type="function_call", + ) + + +def _typed_terminal_response( + function_calls: Sequence[ResponseFunctionToolCall], + *, + status: str = "completed", + incomplete_reason: str | None = None, +) -> Response: + return Response.model_validate({ + "id": "resp_done", + "created_at": 1000000000, + "incomplete_details": {"reason": incomplete_reason} if incomplete_reason else None, + "model": "test-model", + "object": "response", + "output": function_calls, + "parallel_tool_calls": True, + "status": status, + "tool_choice": "auto", + "tools": [], + }) + + +def test_parse_response_requires_completed_function_call_items_for_tool_calls() -> None: + """A completed response must not commit an incomplete local function call.""" + client = OpenAIChatClient(model="test-model", api_key="test-key") + response = _typed_terminal_response([ + _typed_function_call(0), + _typed_function_call(1, status="incomplete"), + ]) + + parsed = client._parse_response_from_openai(response, options={}) # type: ignore[arg-type] + + assert parsed.finish_reason == "stop" + + +def _typed_function_call_events( + function_calls: Sequence[ResponseFunctionToolCall], + *, + arguments_done: set[int], + output_items_done: set[int], +) -> list[object]: + events: list[object] = [] + sequence_number = 0 + for index, function_call in enumerate(function_calls): + events.append( + ResponseOutputItemAddedEvent( + item=function_call, + output_index=index, + sequence_number=sequence_number, + type="response.output_item.added", + ) + ) + sequence_number += 1 + if index in arguments_done: + events.append( + ResponseFunctionCallArgumentsDoneEvent( + arguments=function_call.arguments, + item_id=function_call.id or "", + output_index=index, + sequence_number=sequence_number, + type="response.function_call_arguments.done", + ) + ) + sequence_number += 1 + if index in output_items_done: + events.append( + ResponseOutputItemDoneEvent( + item=function_call, + output_index=index, + sequence_number=sequence_number, + type="response.output_item.done", + ) + ) + sequence_number += 1 + return events + + +async def _parse_typed_response_stream(events: Sequence[object]) -> list[ChatResponseUpdate]: + client = OpenAIChatClient(model="test-model", api_key="test-key") + with ( + patch.object(client, "_prepare_request", new=AsyncMock(return_value=(client.client, {}, {}))), + patch.object(client.client.responses, "create", new=AsyncMock(return_value=_FakeAsyncEventStream(events))), + ): + stream = _as_chat_response_stream( + client._inner_get_response(messages=[Message(role="user", contents=["Hi"])], options={}, stream=True) + ) + return [update async for update in stream] + + +@pytest.mark.parametrize( + ("arguments_done", "output_items_done", "expected_finish_reason"), + [ + ({0, 1}, {0, 1}, "tool_calls"), + ({0}, {0, 1}, "stop"), + ({0, 1}, {0}, "stop"), + (set(), set(), "stop"), + ], +) +async def test_streaming_parallel_function_calls_commit_all_or_nothing( + arguments_done: set[int], + output_items_done: set[int], + expected_finish_reason: str, +) -> None: + """Every parallel call needs both done events before the terminal response commits the batch.""" + function_calls = [_typed_function_call(0), _typed_function_call(1)] + events = _typed_function_call_events( + function_calls, + arguments_done=arguments_done, + output_items_done=output_items_done, + ) + events.append( + ResponseCompletedEvent( + response=_typed_terminal_response(function_calls), + sequence_number=len(events), + type="response.completed", + ) + ) + + updates = await _parse_typed_response_stream(events) + + assert [update.raw_representation for update in updates] == events + assert updates[-1].finish_reason == expected_finish_reason + + +async def test_streaming_function_call_done_event_order_is_independent() -> None: + """The two required done events may arrive in either order without changing commitment.""" + function_call = _typed_function_call(0) + events: list[object] = [ + ResponseOutputItemAddedEvent( + item=function_call, + output_index=0, + sequence_number=0, + type="response.output_item.added", + ), + ResponseOutputItemDoneEvent( + item=function_call, + output_index=0, + sequence_number=1, + type="response.output_item.done", + ), + ResponseFunctionCallArgumentsDoneEvent( + arguments=function_call.arguments, + item_id=function_call.id or "", + output_index=0, + sequence_number=2, + type="response.function_call_arguments.done", + ), + ResponseCompletedEvent( + response=_typed_terminal_response([function_call]), + sequence_number=3, + type="response.completed", + ), + ] + + updates = await _parse_typed_response_stream(events) + + assert [update.raw_representation for update in updates] == events + assert updates[-1].finish_reason == "tool_calls" + + +async def test_streaming_terminal_event_does_not_wait_for_late_function_call_evidence() -> None: + """The existing terminal update resolves immediately and later events cannot retroactively commit it.""" + function_call = _typed_function_call(0) + commitment_events = _typed_function_call_events( + [function_call], + arguments_done={0}, + output_items_done=set(), + ) + terminal_event = ResponseCompletedEvent( + response=_typed_terminal_response([function_call]), + sequence_number=len(commitment_events), + type="response.completed", + ) + late_done_event = ResponseOutputItemDoneEvent( + item=function_call, + output_index=0, + sequence_number=len(commitment_events) + 1, + type="response.output_item.done", + ) + events = [*commitment_events, terminal_event, late_done_event] + + updates = await _parse_typed_response_stream(events) + + assert [update.raw_representation for update in updates] == events + assert updates[-2].finish_reason == "stop" + assert updates[-1].finish_reason is None + + +@pytest.mark.parametrize(("terminal_status", "event_type"), [("incomplete", "incomplete"), ("failed", "failed")]) +async def test_streaming_noncompleted_terminal_never_commits_function_calls( + terminal_status: str, + event_type: str, +) -> None: + """Done evidence cannot authorize calls when the provider response did not complete.""" + function_call = _typed_function_call(0) + events = _typed_function_call_events([function_call], arguments_done={0}, output_items_done={0}) + response = _typed_terminal_response( + [function_call], + status=terminal_status, + incomplete_reason="max_output_tokens" if terminal_status == "incomplete" else None, + ) + terminal_event: object + if event_type == "incomplete": + terminal_event = ResponseIncompleteEvent( + response=response, + sequence_number=len(events), + type="response.incomplete", + ) + else: + terminal_event = ResponseFailedEvent( + response=response, + sequence_number=len(events), + type="response.failed", + ) + events.append(terminal_event) + + updates = await _parse_typed_response_stream(events) + + assert updates[-1].finish_reason != "tool_calls" + + +async def test_streaming_missing_terminal_event_never_commits_function_calls() -> None: + """A stream ending after both done events has no finish reason to authorize execution.""" + function_call = _typed_function_call(0) + events = _typed_function_call_events([function_call], arguments_done={0}, output_items_done={0}) + + updates = await _parse_typed_response_stream(events) + + assert all(update.finish_reason != "tool_calls" for update in updates) + + @pytest.mark.parametrize( ("event_type", "status", "incomplete_reason", "output_type", "expected_finish_reason"), [ @@ -7441,7 +7726,7 @@ def test_streaming_terminal_response_sets_finish_reason( mock_event.response.incomplete_details = ( MagicMock(reason=incomplete_reason) if incomplete_reason is not None else None ) - mock_event.response.output = [MagicMock(type=output_type)] if output_type is not None else [] + mock_event.response.output = [MagicMock(type=output_type, status="completed")] if output_type is not None else [] update = client._parse_chunk_from_openai(mock_event, options={}, function_call_ids={}) From 32e6be0b94e946014aaa3abbb24eaec4d2209951 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:10:05 +0200 Subject: [PATCH 09/16] Exercise Responses commitment through tool loop Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../agent_framework_openai/_chat_client.py | 5 +- .../tests/openai/test_openai_chat_client.py | 196 +++++++++++++++++- 2 files changed, 199 insertions(+), 2 deletions(-) diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 0028ffd5a2c..965e02ea827 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -168,7 +168,9 @@ def _resolve_finish_reason( function_calls_committed: bool, ) -> FinishReason | None: """Resolve function-call authorization without trusting a speculative provider reason.""" - if has_function_calls and function_calls_committed: + if not has_function_calls: + return provider_finish_reason + if function_calls_committed: return FinishReason("tool_calls") if provider_finish_reason == "tool_calls": return None @@ -2694,6 +2696,7 @@ def _get_finish_reason_from_openai_response( function_calls_committed = getattr(response, "status", None) == "completed" and all( function_call_commitments.get((output_index, getattr(item, "id", None) or ""), 0) == _FUNCTION_CALL_COMMITTED + and getattr(item, "status", None) == "completed" for output_index, item in function_calls ) return _resolve_finish_reason( diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 15cdeec28d8..a73b1d5f957 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -46,6 +46,7 @@ Response, ResponseCompletedEvent, ResponseFailedEvent, + ResponseFunctionCallArgumentsDeltaEvent, ResponseFunctionCallArgumentsDoneEvent, ResponseFunctionToolCall, ResponseIncompleteEvent, @@ -7381,7 +7382,7 @@ def test_streaming_response_completed_sets_created_at() -> None: [ ("stop", False, False, "stop"), ("length", False, True, "length"), - ("tool_calls", False, False, None), + ("tool_calls", False, False, "tool_calls"), ("stop", True, True, "tool_calls"), ("length", True, True, "tool_calls"), ("tool_calls", True, False, None), @@ -7518,6 +7519,16 @@ def _typed_function_call_events( ) ) sequence_number += 1 + events.append( + ResponseFunctionCallArgumentsDeltaEvent( + delta=function_call.arguments, + item_id=function_call.id or "", + output_index=index, + sequence_number=sequence_number, + type="response.function_call_arguments.delta", + ) + ) + sequence_number += 1 if index in arguments_done: events.append( ResponseFunctionCallArgumentsDoneEvent( @@ -7554,6 +7565,39 @@ async def _parse_typed_response_stream(events: Sequence[object]) -> list[ChatRes return [update async for update in stream] +async def _run_typed_function_loop( + first_events: Sequence[object], + tool_instance: FunctionTool, +) -> tuple[list[ChatResponseUpdate], ChatResponse, AsyncMock]: + client = OpenAIChatClient(model="test-model", api_key="test-key") + final_events = [ + ResponseCompletedEvent( + response=_typed_terminal_response([]), + sequence_number=0, + type="response.completed", + ) + ] + create = AsyncMock( + side_effect=[ + _FakeAsyncEventStream(first_events), + _FakeAsyncEventStream(final_events), + ] + ) + with ( + patch.object(client, "_prepare_request", new=AsyncMock(return_value=(client.client, {}, {}))), + patch.object(client.client.responses, "create", new=create), + ): + stream = _as_chat_response_stream( + client.get_response( + [Message(role="user", contents=["Call the tool"])], + options={"tools": [tool_instance]}, + stream=True, + ) + ) + updates = [update async for update in stream] + return updates, await stream.get_final_response(), create + + @pytest.mark.parametrize( ("arguments_done", "output_items_done", "expected_finish_reason"), [ @@ -7587,6 +7631,156 @@ async def test_streaming_parallel_function_calls_commit_all_or_nothing( assert [update.raw_representation for update in updates] == events assert updates[-1].finish_reason == expected_finish_reason + calls = [content for update in updates for content in update.contents if content.type == "function_call"] + assert [(call.call_id, call.arguments) for call in calls] == [ + ("call_0", '{"index": 0}'), + ("call_1", '{"index": 1}'), + ] + + +async def test_streaming_completed_response_requires_completed_function_call_item() -> None: + function_call = _typed_function_call(0, status="incomplete") + events = _typed_function_call_events([function_call], arguments_done={0}, output_items_done={0}) + events.append( + ResponseCompletedEvent( + response=_typed_terminal_response([function_call]), + sequence_number=len(events), + type="response.completed", + ) + ) + + updates = await _parse_typed_response_stream(events) + + assert updates[-1].finish_reason == "stop" + + +@pytest.mark.parametrize( + ("arguments_done", "output_items_done", "terminal_status", "expected_executions", "expected_provider_calls"), + [ + ({0}, {0}, "completed", 1, 2), + (set(), {0}, "completed", 0, 1), + ({0}, set(), "completed", 0, 1), + (set(), set(), "completed", 0, 1), + ({0}, {0}, "incomplete", 0, 1), + ({0}, {0}, "failed", 0, 1), + ], +) +async def test_openai_function_loop_executes_only_committed_streamed_calls( + arguments_done: set[int], + output_items_done: set[int], + terminal_status: str, + expected_executions: int, + expected_provider_calls: int, +) -> None: + executions = 0 + + @tool(name="lookup", approval_mode="never_require") + def lookup(index: int) -> str: + nonlocal executions + executions += 1 + return str(index) + + function_call = _typed_function_call(0) + events = _typed_function_call_events( + [function_call], + arguments_done=arguments_done, + output_items_done=output_items_done, + ) + response = _typed_terminal_response( + [function_call], + status=terminal_status, + incomplete_reason="max_output_tokens" if terminal_status == "incomplete" else None, + ) + if terminal_status == "completed": + terminal_event: object = ResponseCompletedEvent( + response=response, + sequence_number=len(events), + type="response.completed", + ) + elif terminal_status == "incomplete": + terminal_event = ResponseIncompleteEvent( + response=response, + sequence_number=len(events), + type="response.incomplete", + ) + else: + terminal_event = ResponseFailedEvent( + response=response, + sequence_number=len(events), + type="response.failed", + ) + events.append(terminal_event) + + updates, final, create = await _run_typed_function_loop(events, lookup) + + assert executions == expected_executions + assert create.await_count == expected_provider_calls + assert any(content.type == "function_call" for update in updates for content in update.contents) + result_contents = [ + content for message in final.messages for content in message.contents if content.type == "function_result" + ] + assert len(result_contents) == expected_executions + + +async def test_uncommitted_streamed_call_does_not_request_approval() -> None: + @tool(name="lookup", approval_mode="always_require") + def lookup(index: int) -> str: + raise AssertionError("uncommitted call must not execute") + + function_call = _typed_function_call(0) + events = _typed_function_call_events([function_call], arguments_done=set(), output_items_done=set()) + events.append( + ResponseCompletedEvent( + response=_typed_terminal_response([function_call]), + sequence_number=len(events), + type="response.completed", + ) + ) + + updates, final, create = await _run_typed_function_loop(events, lookup) + + assert create.await_count == 1 + assert not any(content.type == "function_approval_request" for update in updates for content in update.contents) + assert not any( + content.type == "function_approval_request" + for message in final.messages + for content in message.contents + ) + + +async def test_committed_malformed_arguments_fail_before_tool_body() -> None: + executions = 0 + + @tool(name="lookup", approval_mode="never_require") + def lookup(index: int) -> str: + nonlocal executions + executions += 1 + return str(index) + + function_call = ResponseFunctionToolCall( + arguments='{"index":', + call_id="call_0", + id="fc_0", + name="lookup", + status="completed", + type="function_call", + ) + events = _typed_function_call_events([function_call], arguments_done={0}, output_items_done={0}) + events.append( + ResponseCompletedEvent( + response=_typed_terminal_response([function_call]), + sequence_number=len(events), + type="response.completed", + ) + ) + + _, final, create = await _run_typed_function_loop(events, lookup) + + assert executions == 0 + assert create.await_count == 2 + results = [content for message in final.messages for content in message.contents if content.type == "function_result"] + assert len(results) == 1 + assert results[0].exception is not None async def test_streaming_function_call_done_event_order_is_independent() -> None: From 95bdd7359a0a008ca92169b2d9933ff2ff7b3942 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:19:14 +0200 Subject: [PATCH 10/16] Cover replay and shell commitment edges Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...orize-function-calls-with-finish-reason.md | 2 +- .../specs/004-python-function-calling-loop.md | 4 +-- python/packages/ag-ui/tests/ag_ui/conftest.py | 12 +++++-- .../packages/core/agent_framework/_tools.py | 31 ++++++++++++++++--- .../core/test_function_invocation_logic.py | 13 +++++++- .../agent_framework_openai/_chat_client.py | 25 +++++++++++++-- 6 files changed, 73 insertions(+), 14 deletions(-) diff --git a/docs/decisions/0041-authorize-function-calls-with-finish-reason.md b/docs/decisions/0041-authorize-function-calls-with-finish-reason.md index 1ac2228c1be..3309bd67e94 100644 --- a/docs/decisions/0041-authorize-function-calls-with-finish-reason.md +++ b/docs/decisions/0041-authorize-function-calls-with-finish-reason.md @@ -67,7 +67,7 @@ base class, mixin, registry, or cross-provider helper. Streaming adapters resolve the finish reason on the provider's existing terminal event. They do not add another drain, buffer speculative content, or wait after iterator exhaustion. Calls returned without authorization remain visible to -the caller but are excluded from later model-bound replay. +the caller, while every assistant message from that uncommitted provider turn is excluded from later model-bound replay. ### Consequences diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 145443fbac5..e39611d95e7 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -161,8 +161,8 @@ does not justify a shared provider abstraction. Custom clients that return compl `finish_reason="tool_calls"` explicitly. Uncommitted call-bearing assistant turns remain in the caller-visible response but are marked as excluded from future -model input. Reasoning and call content from that turn are omitted together so later stateless replay cannot create a -dangling provider call. +model input. All assistant messages in that provider turn are omitted together, including reasoning that arrived in a +separate streamed message, so later stateless replay cannot create a dangling provider call. ### Approval pause and resume diff --git a/python/packages/ag-ui/tests/ag_ui/conftest.py b/python/packages/ag-ui/tests/ag_ui/conftest.py index 9208776be8f..68e1e7a6366 100644 --- a/python/packages/ag-ui/tests/ag_ui/conftest.py +++ b/python/packages/ag-ui/tests/ag_ui/conftest.py @@ -139,18 +139,24 @@ def _inner_get_response( **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: if stream: + async def _stream() -> AsyncIterator[ChatResponseUpdate]: has_function_calls = False has_finish_reason = False + pending_update: ChatResponseUpdate | None = None async for update in self._stream_fn(messages, options, **kwargs): has_function_calls = has_function_calls or any( content.type == "function_call" and not content.informational_only for content in update.contents ) has_finish_reason = has_finish_reason or update.finish_reason is not None - yield update - if has_function_calls and not has_finish_reason: - yield ChatResponseUpdate(finish_reason="tool_calls") + if pending_update is not None: + yield pending_update + pending_update = update + if pending_update is not None: + if has_function_calls and not has_finish_reason: + pending_update.finish_reason = "tool_calls" + yield pending_update def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: return _finish_committed_test_calls(ChatResponse.from_updates(updates)) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index ae54cb55480..ea35af538ea 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1825,16 +1825,35 @@ def _is_actionable_function_call(content: Content) -> bool: def _mark_uncommitted_function_call_messages(response: ChatResponse) -> None: from ._clients import _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY # pyright: ignore[reportPrivateUsage] + if not any( + _is_actionable_function_call(content) + for message in response.messages + for content in message.contents + ): + return + replay_metadata = {"finish_reason": response.finish_reason} for message in response.messages: - uncommitted_calls = [content for content in message.contents if _is_actionable_function_call(content)] - if not uncommitted_calls: + if message.role != "assistant": continue - replay_metadata = {"finish_reason": response.finish_reason} message.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata - for content in uncommitted_calls: + for content in message.contents: content.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata +def _mark_uncommitted_streaming_contents(contents: Sequence[Content], response: ChatResponse) -> None: + from ._clients import _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY # pyright: ignore[reportPrivateUsage] + + if response.finish_reason == "tool_calls" or not any( + _is_actionable_function_call(content) + for message in response.messages + for content in message.contents + ): + return + replay_metadata = {"finish_reason": response.finish_reason} + for content in contents: + content.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata + + def _underlying_function_call(content: Content) -> Content: if content.type == "function_approval_response" and content.function_call is not None: return content.function_call @@ -3882,7 +3901,10 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non streamed_names_by_call_id: dict[str, str] = {} last_streamed_identity: tuple[str, str] | None = None warned_empty_call_ids: set[str] = set() + streamed_contents: list[Content] = [] async for update in inner_stream: + if update.role in (None, "assistant"): + streamed_contents.extend(update.contents) for content in update.contents: if content.type != "function_call": continue @@ -3940,6 +3962,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non yield update response = await inner_stream.get_final_response() + _mark_uncommitted_streaming_contents(streamed_contents, response) fallback_added = False if options.get("tool_choice") == "none" and budget_state.get("truncated"): fallback_added = _ensure_function_invocation_limit_fallback_response(response) diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 6eca05a00fc..82328154bcd 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -392,13 +392,20 @@ def guarded_write() -> str: client = cast(Any, chat_client_base) client.auto_finish_function_calls = False + reasoning = Content.from_text_reasoning(text="I should write") function_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}) if stream: client.streaming_responses = [ [ + ChatResponseUpdate( + contents=[reasoning], + role="assistant", + message_id="reasoning-message", + ), ChatResponseUpdate( contents=[function_call], role="assistant", + message_id="function-call-message", finish_reason=cast(Any, finish_reason), ) ] @@ -406,7 +413,10 @@ def guarded_write() -> str: else: client.run_responses = [ ChatResponse( - messages=[Message(role="assistant", contents=[function_call])], + messages=[ + Message(role="assistant", contents=[reasoning]), + Message(role="assistant", contents=[function_call]), + ], finish_reason=cast(Any, finish_reason), ) ] @@ -425,6 +435,7 @@ def guarded_write() -> str: response = await result contents = [content for message in response.messages for content in message.contents] + assert reasoning in contents assert function_call in contents assert not any(content.type in {"function_approval_request", "function_result"} for content in contents) assert tool_calls == 0 diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 965e02ea827..34fd7c10e57 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -747,7 +747,11 @@ def _parse_stream_chunk(chunk: OpenAIResponseStreamEvent) -> ChatResponseUpdate: function_call_commitments[key] = ( function_call_commitments.get(key, 0) | _FUNCTION_CALL_ARGUMENTS_DONE ) - case "response.output_item.done" if getattr(chunk.item, "type", None) == "function_call": + case "response.output_item.done" if getattr(chunk.item, "type", None) in { + "function_call", + "shell_call", + "local_shell_call", + }: key = (chunk.output_index, getattr(chunk.item, "id", None) or "") function_call_commitments[key] = ( function_call_commitments.get(key, 0) | _FUNCTION_CALL_OUTPUT_ITEM_DONE @@ -766,6 +770,9 @@ def _parse_stream_chunk(chunk: OpenAIResponseStreamEvent) -> ChatResponseUpdate: update.finish_reason = self._get_finish_reason_from_openai_response( terminal_response, function_call_commitments=function_call_commitments, + local_shell_tool_name=self._get_local_shell_tool_name( + (validated_options or {}).get("tools") + ), ) return update @@ -2671,6 +2678,7 @@ def _get_finish_reason_from_openai_response( response: Any, *, function_call_commitments: Mapping[tuple[int, str], int] | None = None, + local_shell_tool_name: str | None = None, ) -> FinishReason | None: """Get the framework finish reason from a terminal Responses API response.""" incomplete_reason = getattr(getattr(response, "incomplete_details", None), "reason", None) @@ -2687,6 +2695,10 @@ def _get_finish_reason_from_openai_response( (output_index, item) for output_index, item in enumerate(getattr(response, "output", ())) if getattr(item, "type", None) == "function_call" + or ( + local_shell_tool_name is not None + and getattr(item, "type", None) in {"shell_call", "local_shell_call"} + ) ] if function_call_commitments is None: function_calls_committed = getattr(response, "status", None) == "completed" and all( @@ -2695,7 +2707,11 @@ def _get_finish_reason_from_openai_response( else: function_calls_committed = getattr(response, "status", None) == "completed" and all( function_call_commitments.get((output_index, getattr(item, "id", None) or ""), 0) - == _FUNCTION_CALL_COMMITTED + == ( + _FUNCTION_CALL_COMMITTED + if getattr(item, "type", None) == "function_call" + else _FUNCTION_CALL_OUTPUT_ITEM_DONE + ) and getattr(item, "status", None) == "completed" for output_index, item in function_calls ) @@ -3013,7 +3029,10 @@ def _parse_response_from_openai( args["value"] = structured_response elif response_format := options.get("response_format"): args["response_format"] = response_format - if finish_reason := self._get_finish_reason_from_openai_response(response): + if finish_reason := self._get_finish_reason_from_openai_response( + response, + local_shell_tool_name=local_shell_tool_name, + ): args["finish_reason"] = finish_reason # Set continuation_token when background operation is still in progress if response.status and response.status in ("in_progress", "queued"): From f583414c69d774a70606733f578d45c1a7a82276 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:21:56 +0200 Subject: [PATCH 11/16] Format OpenAI commitment regression Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../openai/tests/openai/test_openai_chat_client.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index a73b1d5f957..2c8a7cebc8b 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -7778,7 +7778,12 @@ def lookup(index: int) -> str: assert executions == 0 assert create.await_count == 2 - results = [content for message in final.messages for content in message.contents if content.type == "function_result"] + results = [ + content + for message in final.messages + for content in message.contents + if content.type == "function_result" + ] assert len(results) == 1 assert results[0].exception is not None From 2e4b7eb6ee38365b55ba91225d4b964a5c14f50e Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:24:28 +0200 Subject: [PATCH 12/16] Type scripted finish reasons precisely Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/core/tests/core/test_agents.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index af802db74d2..dbe110aceaa 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -5,7 +5,7 @@ import json import logging from collections.abc import AsyncIterable, Awaitable, Callable, MutableSequence, Sequence -from typing import Any, cast +from typing import Any, Literal, cast from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 @@ -3305,7 +3305,9 @@ def _inner_get_response( # type: ignore[override] store_and_echo = self._effective_store(options) and self._echo_conversation_id conv_id = _PSC_SERVICE_CONVERSATION_ID if store_and_echo else None contents = self._next_contents() - finish_reason = "tool_calls" if any(content.type == "function_call" for content in contents) else "stop" + finish_reason: Literal["tool_calls", "stop"] = ( + "tool_calls" if any(content.type == "function_call" for content in contents) else "stop" + ) if stream: From 25de4b6ad0eb5adffa5915cd12a54ce81f1cb78e Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:26:22 +0200 Subject: [PATCH 13/16] Model Ollama terminal tool call completion Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ollama/tests/test_ollama_chat_client.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/python/packages/ollama/tests/test_ollama_chat_client.py b/python/packages/ollama/tests/test_ollama_chat_client.py index a8245bf3ef0..0ee761067a2 100644 --- a/python/packages/ollama/tests/test_ollama_chat_client.py +++ b/python/packages/ollama/tests/test_ollama_chat_client.py @@ -135,6 +135,8 @@ def mock_streaming_chat_completion_tool_call() -> AsyncStream[OllamaChatResponse tool_calls=cast(Any, [{"function": {"name": "hello_world", "arguments": {"arg1": "value1"}}}]), ), model="test", + done=True, + done_reason="stop", ) stream = MagicMock(spec=AsyncStream) stream.__aiter__.return_value = [ollama_tool_call] @@ -512,6 +514,25 @@ async def test_cmc_streaming_ignores_done_reason_and_usage_before_final_chunk( assert final_response.usage_details is None +def test_streaming_tool_call_before_done_is_not_authorized( + ollama_unit_test_env: dict[str, str], +) -> None: + response = OllamaChatResponse( + message=OllamaMessage( + content="", + role="assistant", + tool_calls=cast(Any, [{"function": {"name": "hello_world", "arguments": {"arg1": "value1"}}}]), + ), + model="test", + done=False, + ) + + update = OllamaChatClient()._parse_streaming_response_from_ollama(response) + + assert update.contents[0].type == "function_call" + assert update.finish_reason is None + + @patch.object(AsyncClient, "chat", new_callable=AsyncMock) async def test_cmc_streaming_reasoning( mock_chat: AsyncMock, From f2ab412cef7b8e6a0fac80b61fc35a47f7298899 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:38:24 +0200 Subject: [PATCH 14/16] Close finish reason review gaps Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ag-ui/agent_framework_ag_ui/_client.py | 11 ++- .../ag-ui/tests/ag_ui/test_ag_ui_client.py | 36 +++++++++ .../anthropic/tests/test_anthropic_client.py | 46 ++++++----- .../packages/core/agent_framework/_clients.py | 37 +++++---- .../core/test_function_invocation_logic.py | 80 +++++++++++++++++++ .../agent_framework_openai/_chat_client.py | 5 +- .../tests/openai/test_openai_chat_client.py | 2 +- 7 files changed, 180 insertions(+), 37 deletions(-) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_client.py b/python/packages/ag-ui/agent_framework_ag_ui/_client.py index 2c340b4160d..78abea32473 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_client.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_client.py @@ -513,6 +513,7 @@ async def _streaming_impl( converter = AGUIEventConverter() has_function_calls = False + open_client_tool_call_ids: set[str] = set() available_interrupts = options.get("available_interrupts", options.get("availableInterrupts")) @@ -526,6 +527,14 @@ async def _streaming_impl( resume=_serialize_resume(options.get("resume")), ): logger.debug(f"[AGUIChatClient] Raw AG-UI event: {event}") + event_type = str(event.get("type", "")).upper() + event_tool_call_id = str(event.get("toolCallId") or event.get("tool_call_id") or "") + if event_type == "TOOL_CALL_START": + event_tool_name = event.get("toolName") or event.get("toolCallName") or event.get("tool_call_name") + if event_tool_name in client_tool_set: + open_client_tool_call_ids.add(event_tool_call_id) + elif event_type == "TOOL_CALL_END": + open_client_tool_call_ids.discard(event_tool_call_id) update = converter.convert_event(event) if update is not None: logger.debug( @@ -553,6 +562,6 @@ async def _streaming_impl( update.finish_reason = _resolve_finish_reason( cast(FinishReason | None, update.finish_reason), has_function_calls=has_function_calls, - function_calls_committed=str(event.get("type", "")).upper() == "RUN_FINISHED", + function_calls_committed=event_type == "RUN_FINISHED" and not open_client_tool_call_ids, ) yield update diff --git a/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py b/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py index 3e7adbfefa0..aec8e840269 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py +++ b/python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py @@ -648,6 +648,42 @@ async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str assert all(update.finish_reason != "tool_calls" for update in updates) assert updates[-1].contents[0].type == "error" + async def test_client_tool_without_end_is_not_committed_by_run_finished( + self, + monkeypatch: MonkeyPatch, + ) -> None: + """RUN_FINISHED does not commit a client tool call whose TOOL_CALL_END is missing.""" + + @tool + def client_tool(value: int) -> str: + """Return the supplied value.""" + return str(value) + + mock_events = [ + {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"}, + {"type": "TOOL_CALL_START", "toolCallId": "call_1", "toolName": "client_tool"}, + {"type": "TOOL_CALL_ARGS", "toolCallId": "call_1", "delta": '{"value": 1}'}, + {"type": "RUN_FINISHED", "threadId": "thread_1", "runId": "run_1"}, + ] + + async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]: + for event in mock_events: + yield event + + client = StubAGUIChatClient(endpoint="http://localhost:8888/") + monkeypatch.setattr(client.http_service, "post_run", mock_post_run) + + stream = client.inner_get_response( + messages=[Message(role="user", contents=["Test"])], + options={"tools": [client_tool]}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + updates = [cast(ChatResponseUpdate, update) async for update in stream] + + assert any(content.type == "function_call" for update in updates for content in update.contents) + assert all(update.finish_reason != "tool_calls" for update in updates) + async def test_client_tool_transport_eof_is_not_committed(self, monkeypatch: MonkeyPatch) -> None: """Transport EOF without RUN_FINISHED leaves client calls non-authorizing.""" diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 8e628e99df0..c065002e978 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -598,26 +598,32 @@ def test_streaming_replay_preserves_empty_signed_thinking_block( client = create_test_anthropic_client(mock_anthropic_client) events: list[BetaRawMessageStreamEvent] = [ - BetaRawContentBlockStartEvent.model_validate({ - "type": "content_block_start", - "index": 0, - "content_block": {"type": "thinking", "thinking": "", "signature": ""}, - }), - BetaRawContentBlockDeltaEvent.model_validate({ - "type": "content_block_delta", - "index": 0, - "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, - }), - BetaRawContentBlockStartEvent.model_validate({ - "type": "content_block_start", - "index": 1, - "content_block": { - "type": "tool_use", - "id": "toolu_test", - "name": "lookup", - "input": {}, - }, - }), + BetaRawContentBlockStartEvent.model_validate( + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + } + ), + BetaRawContentBlockDeltaEvent.model_validate( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, + } + ), + BetaRawContentBlockStartEvent.model_validate( + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_test", + "name": "lookup", + "input": {}, + }, + } + ), ] updates = [client._process_stream_event(event) for event in events] diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 95553ce4125..50cdd174d29 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -66,6 +66,21 @@ _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY = "_agent_framework_uncommitted_function_calls" +def _filter_uncommitted_function_call_messages(messages: Sequence[Message]) -> list[Message]: + return [ + message + for message in messages + if message.role != "assistant" + or ( + _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY not in message.additional_properties + and not any( + _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY in content.additional_properties + for content in message.contents + ) + ) + ] + + # region SupportsChatGetResponse Protocol # Contravariant for the Protocol @@ -376,18 +391,7 @@ async def _prepare_messages_for_model_call( compaction_strategy: CompactionStrategy | None = None, tokenizer: TokenizerProtocol | None = None, ) -> list[Message]: - prepared_messages = [ - message - for message in messages - if message.role != "assistant" - or ( - _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY not in message.additional_properties - and not any( - _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY in content.additional_properties - for content in message.contents - ) - ) - ] + prepared_messages = _filter_uncommitted_function_call_messages(messages) if compaction_strategy is None: if tokenizer is None: return prepared_messages @@ -403,7 +407,11 @@ async def _prepare_messages_for_model_call( # the function-invocation tool loop reuses across iterations; otherwise inserted # summaries would be lost on a throwaway copy while exclusions persisted, silently # dropping older groups (issue #4991). - working_messages = messages if isinstance(messages, list) else prepared_messages + if isinstance(messages, list): + messages[:] = prepared_messages + working_messages = messages + else: + working_messages = prepared_messages return await apply_compaction( working_messages, strategy=compaction_strategy, @@ -529,10 +537,11 @@ def get_response( tokenizer=tokenizer, ) merged_client_kwargs = dict(client_kwargs) if client_kwargs is not None else {} + filtered_messages = _filter_uncommitted_function_call_messages(messages) if not compaction_overrides: return self._inner_get_response( - messages=messages, + messages=filtered_messages, stream=stream, options=options or {}, **merged_client_kwargs, diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 82328154bcd..72ec1190866 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -6,6 +6,7 @@ import warnings from collections.abc import AsyncIterable, Awaitable, Callable, Sequence from typing import Any, Literal, cast +from unittest.mock import patch import pytest @@ -445,6 +446,85 @@ def guarded_write() -> str: prepared_messages = await client._prepare_messages_for_model_call(response.messages) assert prepared_messages == [] + replayed_messages: list[Message] = [] + if stream: + + def capture_stream( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + replayed_messages.extend(messages) + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate( + contents=[Content.from_text("done")], + role="assistant", + finish_reason="stop", + ) + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + with patch.object(client, "_get_streaming_response", side_effect=capture_stream): + replay_stream = client.get_response(response.messages, stream=True) + _ = [update async for update in replay_stream] + else: + + async def capture_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ChatResponse: + replayed_messages.extend(messages) + return ChatResponse( + messages=[Message(role="assistant", contents=["done"])], + finish_reason="stop", + ) + + with patch.object(client, "_get_non_streaming_response", side_effect=capture_response): + await client.get_response(response.messages) + + assert replayed_messages == [] + + +async def test_uncommitted_function_call_turn_is_filtered_before_compaction( + chat_client_base: SupportsChatGetResponse, +) -> None: + client = cast(Any, chat_client_base) + client.compaction_strategy = SlidingWindowStrategy(keep_last_groups=2) + client.tokenizer = CharacterEstimatorTokenizer() + uncommitted_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}) + response = ChatResponse( + messages=[ + Message(role="assistant", contents=[Content.from_text_reasoning(text="I should write")]), + Message(role="assistant", contents=[uncommitted_call]), + ] + ) + from agent_framework._tools import _mark_uncommitted_function_call_messages + + _mark_uncommitted_function_call_messages(response) + replayed_messages: list[Message] = [] + + async def capture_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ChatResponse: + replayed_messages.extend(messages) + return ChatResponse( + messages=[Message(role="assistant", contents=["done"])], + finish_reason="stop", + ) + + with patch.object(client, "_get_non_streaming_response", side_effect=capture_response): + await client.get_response(response.messages) + + assert replayed_messages == [] + assert response.messages == [] + @pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"]) async def test_missing_tool_calls_finish_reason_does_not_request_approval( diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 34fd7c10e57..3a4e4ac33ab 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -3279,7 +3279,10 @@ def output_text_properties(output: Any) -> dict[str, Any] | None: created_at = datetime.fromtimestamp(event.response.created_at, tz=timezone.utc).strftime( "%Y-%m-%dT%H:%M:%S.%fZ" ) - finish_reason = self._get_finish_reason_from_openai_response(event.response) + finish_reason = self._get_finish_reason_from_openai_response( + event.response, + function_call_commitments={}, + ) if event.response.usage: usage = self._parse_usage_from_openai(event.response.usage) if usage: diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 2c8a7cebc8b..c169bc5e37c 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -7899,7 +7899,7 @@ async def test_streaming_missing_terminal_event_never_commits_function_calls() - ("event_type", "status", "incomplete_reason", "output_type", "expected_finish_reason"), [ ("response.completed", "completed", None, None, "stop"), - ("response.completed", "completed", None, "function_call", "tool_calls"), + ("response.completed", "completed", None, "function_call", "stop"), ("response.incomplete", "incomplete", "max_output_tokens", None, "length"), ("response.incomplete", "incomplete", "content_filter", None, "content_filter"), ("response.failed", "failed", None, None, None), From cbdf396163c0638779b4414a803ab59fb0060647 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:41:09 +0200 Subject: [PATCH 15/16] Filter uncommitted turns at model boundary Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/core/agent_framework/_clients.py | 2 ++ .../packages/core/tests/core/test_function_invocation_logic.py | 1 - 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 50cdd174d29..8a38e9c48f4 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -67,6 +67,8 @@ def _filter_uncommitted_function_call_messages(messages: Sequence[Message]) -> list[Message]: + if any(not isinstance(message, Message) for message in messages): + return cast(list[Message], messages) return [ message for message in messages diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 72ec1190866..3561bd096da 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -523,7 +523,6 @@ async def capture_response( await client.get_response(response.messages) assert replayed_messages == [] - assert response.messages == [] @pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"]) From dde9ffb5f8739d763873571407b3ea36e037cbe5 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 11 Sep 2026 16:51:33 +0200 Subject: [PATCH 16/16] Apply repository formatting Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../anthropic/tests/test_anthropic_client.py | 46 ++++++++----------- .../packages/core/agent_framework/_clients.py | 3 +- .../core/agent_framework/_sessions.py | 6 +-- .../packages/core/agent_framework/_tools.py | 10 +--- .../core/test_function_invocation_logic.py | 6 +-- .../agent_framework_openai/_chat_client.py | 9 +--- .../tests/openai/test_openai_chat_client.py | 9 +--- 7 files changed, 30 insertions(+), 59 deletions(-) diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index c065002e978..8e628e99df0 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -598,32 +598,26 @@ def test_streaming_replay_preserves_empty_signed_thinking_block( client = create_test_anthropic_client(mock_anthropic_client) events: list[BetaRawMessageStreamEvent] = [ - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 0, - "content_block": {"type": "thinking", "thinking": "", "signature": ""}, - } - ), - BetaRawContentBlockDeltaEvent.model_validate( - { - "type": "content_block_delta", - "index": 0, - "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, - } - ), - BetaRawContentBlockStartEvent.model_validate( - { - "type": "content_block_start", - "index": 1, - "content_block": { - "type": "tool_use", - "id": "toolu_test", - "name": "lookup", - "input": {}, - }, - } - ), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }), + BetaRawContentBlockDeltaEvent.model_validate({ + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": "synthetic-signature"}, + }), + BetaRawContentBlockStartEvent.model_validate({ + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_test", + "name": "lookup", + "input": {}, + }, + }), ] updates = [client._process_stream_event(event) for event in events] diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 8a38e9c48f4..e9d25fe0915 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -76,8 +76,7 @@ def _filter_uncommitted_function_call_messages(messages: Sequence[Message]) -> l or ( _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY not in message.additional_properties and not any( - _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY in content.additional_properties - for content in message.contents + _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY in content.additional_properties for content in message.contents ) ) ] diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index df2b8e057aa..3cdd530c4ce 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -1309,11 +1309,7 @@ def _response_contains_follow_up_request(response: ChatResponse) -> bool: """Return whether a response requires another model call in the current run.""" return any( item.type == "function_approval_request" - or ( - response.finish_reason == "tool_calls" - and item.type == "function_call" - and not item.informational_only - ) + or (response.finish_reason == "tool_calls" and item.type == "function_call" and not item.informational_only) for message in response.messages for item in message.contents ) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index ea35af538ea..8e9a4d381a4 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1825,11 +1825,7 @@ def _is_actionable_function_call(content: Content) -> bool: def _mark_uncommitted_function_call_messages(response: ChatResponse) -> None: from ._clients import _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY # pyright: ignore[reportPrivateUsage] - if not any( - _is_actionable_function_call(content) - for message in response.messages - for content in message.contents - ): + if not any(_is_actionable_function_call(content) for message in response.messages for content in message.contents): return replay_metadata = {"finish_reason": response.finish_reason} for message in response.messages: @@ -1844,9 +1840,7 @@ def _mark_uncommitted_streaming_contents(contents: Sequence[Content], response: from ._clients import _UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY # pyright: ignore[reportPrivateUsage] if response.finish_reason == "tool_calls" or not any( - _is_actionable_function_call(content) - for message in response.messages - for content in message.contents + _is_actionable_function_call(content) for message in response.messages for content in message.contents ): return replay_metadata = {"finish_reason": response.finish_reason} diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 3561bd096da..4bb3fc0099b 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -408,7 +408,7 @@ def guarded_write() -> str: role="assistant", message_id="function-call-message", finish_reason=cast(Any, finish_reason), - ) + ), ] ] else: @@ -550,9 +550,7 @@ def guarded_write() -> str: response = await result.get_final_response() if stream else await result assert not any( - content.type == "function_approval_request" - for message in response.messages - for content in message.contents + content.type == "function_approval_request" for message in response.messages for content in message.contents ) assert client.call_count == 1 diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 3a4e4ac33ab..68b94964a0e 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -770,9 +770,7 @@ def _parse_stream_chunk(chunk: OpenAIResponseStreamEvent) -> ChatResponseUpdate: update.finish_reason = self._get_finish_reason_from_openai_response( terminal_response, function_call_commitments=function_call_commitments, - local_shell_tool_name=self._get_local_shell_tool_name( - (validated_options or {}).get("tools") - ), + local_shell_tool_name=self._get_local_shell_tool_name((validated_options or {}).get("tools")), ) return update @@ -2695,10 +2693,7 @@ def _get_finish_reason_from_openai_response( (output_index, item) for output_index, item in enumerate(getattr(response, "output", ())) if getattr(item, "type", None) == "function_call" - or ( - local_shell_tool_name is not None - and getattr(item, "type", None) in {"shell_call", "local_shell_call"} - ) + or (local_shell_tool_name is not None and getattr(item, "type", None) in {"shell_call", "local_shell_call"}) ] if function_call_commitments is None: function_calls_committed = getattr(response, "status", None) == "completed" and all( diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index c169bc5e37c..38753f5f465 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -7742,9 +7742,7 @@ def lookup(index: int) -> str: assert create.await_count == 1 assert not any(content.type == "function_approval_request" for update in updates for content in update.contents) assert not any( - content.type == "function_approval_request" - for message in final.messages - for content in message.contents + content.type == "function_approval_request" for message in final.messages for content in message.contents ) @@ -7779,10 +7777,7 @@ def lookup(index: int) -> str: assert executions == 0 assert create.await_count == 2 results = [ - content - for message in final.messages - for content in message.contents - if content.type == "function_result" + content for message in final.messages for content in message.contents if content.type == "function_result" ] assert len(results) == 1 assert results[0].exception is not None