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..3309bd67e94 --- /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, while every assistant message from that uncommitted provider turn is 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..e39611d95e7 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. 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 ```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/ag-ui/agent_framework_ag_ui/_client.py b/python/packages/ag-ui/agent_framework_ag_ui/_client.py index 7aa43062427..78abea32473 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,8 @@ async def _streaming_impl( logger.debug(f"[AGUIChatClient] Client tool set: {client_tool_set}") converter = AGUIEventConverter() + has_function_calls = False + open_client_tool_call_ids: set[str] = set() available_interrupts = options.get("available_interrupts", options.get("availableInterrupts")) @@ -508,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( @@ -522,6 +549,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 +559,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=event_type == "RUN_FINISHED" and not open_client_tool_call_ids, + ) yield update diff --git a/python/packages/ag-ui/tests/ag_ui/conftest.py b/python/packages/ag-ui/tests/ag_ui/conftest.py index c4d4d972cde..68e1e7a6366 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) @@ -130,10 +140,28 @@ def _inner_get_response( ) -> 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 + 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 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 +170,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", + ) ) 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..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 @@ -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,197 @@ 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_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.""" + + @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 +795,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.""" 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 diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 564d78aa660..e9d25fe0915 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -63,6 +63,24 @@ logger = logging.getLogger("agent_framework") +_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY = "_agent_framework_uncommitted_function_calls" + + +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 + 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 @@ -374,7 +392,7 @@ async def _prepare_messages_for_model_call( compaction_strategy: CompactionStrategy | None = None, tokenizer: TokenizerProtocol | None = None, ) -> list[Message]: - prepared_messages = list(messages) + prepared_messages = _filter_uncommitted_function_call_messages(messages) if compaction_strategy is None: if tokenizer is None: return prepared_messages @@ -390,7 +408,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, @@ -516,10 +538,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/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index 9fcce18a071..3cdd530c4ce 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -1308,7 +1308,8 @@ 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..8e9a4d381a4 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1822,6 +1822,32 @@ 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] + + 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: + if message.role != "assistant": + continue + message.additional_properties[_UNCOMMITTED_FUNCTION_CALL_MESSAGE_KEY] = replay_metadata + 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 @@ -3392,6 +3418,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) @@ -3862,7 +3895,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 @@ -3920,6 +3956,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/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_agents.py b/python/packages/core/tests/core/test_agents.py index 4ef0d031d72..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 @@ -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,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: Literal["tool_calls", "stop"] = ( + "tool_calls" if any(content.type == "function_call" for content in contents) else "stop" + ) if stream: @@ -3311,7 +3316,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 +3333,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_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..4bb3fc0099b 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,8 @@ import logging import warnings from collections.abc import AsyncIterable, Awaitable, Callable, Sequence -from typing import Any, Literal +from typing import Any, Literal, cast +from unittest.mock import patch import pytest @@ -364,6 +365,196 @@ 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 + 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), + ), + ] + ] + else: + client.run_responses = [ + ChatResponse( + messages=[ + Message(role="assistant", contents=[reasoning]), + 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 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 + assert middleware_calls == 0 + assert client.call_count == 1 + + 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 == [] + + +@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 +6427,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 +6461,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_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_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"])), ] 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") diff --git a/python/packages/gemini/agent_framework_gemini/_chat_client.py b/python/packages/gemini/agent_framework_gemini/_chat_client.py index a6c6796b418..aab0be438fe 100644 --- a/python/packages/gemini/agent_framework_gemini/_chat_client.py +++ b/python/packages/gemini/agent_framework_gemini/_chat_client.py @@ -324,11 +324,25 @@ 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 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 + + class RawGeminiChatClient( BaseChatClient[GeminiChatOptionsT], Generic[GeminiChatOptionsT], @@ -585,12 +599,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 +1169,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 +1207,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..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 @@ -542,8 +543,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 +605,215 @@ 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"} + + +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() + 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 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, diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 0cc2bc3e34c..68b94964a0e 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -156,6 +156,27 @@ 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 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 + + # 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 +737,42 @@ 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) 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 + ) + 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, + local_shell_tool_name=self._get_local_shell_tool_name((validated_options or {}).get("tools")), + ) + return update def _finalize_with_captured_format(updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]: # ResponseStream only calls the finalizer after iterating or draining `_stream()`, @@ -745,12 +802,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 +827,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 +836,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 +2671,50 @@ 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, + 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) 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" + 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( + 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 + 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 + ) + 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, @@ -2950,7 +3024,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"): @@ -3197,7 +3274,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 7258c50cae9..38753f5f465 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,17 @@ SettingNotFoundError, ) from openai import AsyncOpenAI, BadRequestError +from openai.types.responses import ( + Response, + ResponseCompletedEvent, + ResponseFailedEvent, + ResponseFunctionCallArgumentsDeltaEvent, + 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 +71,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 +7377,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, "tool_calls"), + ("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 +7427,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,11 +7453,448 @@ 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 + 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( + 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] + + +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"), + [ + ({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 + 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: + """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"), [ ("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), @@ -7441,7 +7920,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={})