From 3a2fc631c24b81a2930a4234affc9f1769844e62 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 11:15:44 +0200 Subject: [PATCH 1/6] Python: Handle provider-invalidated function calls Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...ovider-invalidated-local-function-calls.md | 71 +++++ .../specs/004-python-function-calling-loop.md | 15 + python/CODING_STANDARD.md | 2 + .../agent_framework_anthropic/_chat_client.py | 51 +++- .../anthropic/tests/test_anthropic_client.py | 277 ++++++++++++++++- .../packages/core/agent_framework/__init__.py | 2 + .../core/agent_framework/__init__.pyi | 2 + .../packages/core/agent_framework/_tools.py | 281 +++++++++++------- .../core/agent_framework/exceptions.py | 6 + .../core/tests/core/test_exceptions.py | 18 +- .../core/test_function_invocation_logic.py | 252 ++++++++++++++++ 11 files changed, 861 insertions(+), 116 deletions(-) create mode 100644 docs/decisions/0041-handle-provider-invalidated-local-function-calls.md diff --git a/docs/decisions/0041-handle-provider-invalidated-local-function-calls.md b/docs/decisions/0041-handle-provider-invalidated-local-function-calls.md new file mode 100644 index 00000000000..dc0eb3c42cb --- /dev/null +++ b/docs/decisions/0041-handle-provider-invalidated-local-function-calls.md @@ -0,0 +1,71 @@ +--- +status: proposed +contact: eavanvalkenburg +date: 2026-09-15 +deciders: eavanvalkenburg +--- + +# Execute usable function calls unless the provider invalidates them + +## Context and Problem Statement + +Chat providers do not use terminal response reasons uniformly. A response can stop because of an output limit while +still containing complete local function arguments that are valid for execution and accepted by the service on +continuation. Other provider protocols explicitly invalidate local function calls when a tool block is truncated, a +refusal replaces partial output, a stream fails, or required terminal events are absent. + +The function-calling loop needs a default that preserves usable calls without executing calls that a provider has +explicitly cancelled. + +## Decision Drivers + +- Preserve provider-neutral execution for complete, schema-usable local calls. +- Prevent local approvals and side effects for calls a provider protocol invalidates. +- Keep streaming deltas visible without buffering every provider response. +- Avoid coupling unrelated providers through a shared finish-reason interpretation layer. +- Keep local call invalidation distinct from caller cancellation. + +## Considered Options + +### Require one generic finish reason before execution + +- Good: provides one simple rule in the core loop. +- Bad: rejects complete calls from providers whose terminal reason describes the enclosing response rather than the + usability of its function arguments. +- Bad: makes custom clients depend on a provider-neutral meaning that their protocols may not have. + +### Execute every parseable function call + +- Good: preserves the current provider-neutral behavior. +- Bad: can execute partial output that a provider explicitly invalidated, even when accumulated JSON happens to parse. + +### Execute usable calls by default and let providers signal invalidation + +- Good: preserves the core loop's argument-driven policy. +- Good: lets an adapter enforce authoritative provider-specific terminal rules. +- Good: needs no cross-provider resolver or broad streaming buffer. +- Bad: each provider with invalidation semantics must implement and test its own protocol state. + +## Decision Outcome + +Chosen option: "Execute usable calls by default and let providers signal invalidation", because function argument +usability is the portable default while invalidation is provider-specific protocol knowledge. + +`FunctionInvocationLayer` remains independent of `finish_reason`: a newly completed actionable call proceeds when +argument preparation and schema validation succeed. A provider that knows its local calls were cancelled raises +`FunctionCallInvalidatedException`. The layer abandons that current iteration, clears request budget state, restores +the last valid continuation, avoids successful response persistence and local function side effects, and re-raises the +same exception. + +Anthropic applies the signal only to local actionable `tool_use` blocks. A valid stream has closed local blocks, a +terminal `stop_reason` of `tool_use`, and `message_stop`. Non-tool terminal reasons, an open block at `message_stop`, +EOF without `message_stop`, or a non-cancellation stream error after a local call starts invalidate those calls. +Hosted/server-only calls and `pause_turn` retain their existing behavior. Streaming call deltas may already have been +yielded; direct consumers must discard them when invalidation is raised. + +### Consequences + +- Complete calls may execute under an enclosing `length` finish reason. +- Anthropic local calls cannot execute after protocol evidence invalidates or fails to finalize them. +- Direct streaming consumers receive an explicit discard signal without added buffering latency. +- Other provider adapters remain unchanged unless their own protocol defines equivalent invalidation semantics. diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index bf7d1ed72a3..13e357482e2 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -350,6 +350,18 @@ that manually replay messages own the equivalent rule: do not resend an approval ### Function calls and results +- A newly completed, actionable local `function_call` is eligible for execution when argument preparation and schema + validation succeed. The enclosing response `finish_reason` is not a generic authorization gate: providers may + report another reason, including `length`, while still returning a complete and service-usable call. +- A provider adapter that knows its protocol invalidated or cancelled local function calls already exposed in response + content raises `FunctionCallInvalidatedException`. The function invocation layer then abandons only that current + model/tool iteration, clears its request budget state, restores the last valid service continuation, and re-raises + the same exception. It emits no new approval request or function result, invokes no function middleware or tool + body, makes no second model request or dangling-call settlement request, and does not success-persist that service + response. Approval decisions resolved before the invalidated provider call retain their existing semantics. + Streaming providers may yield local function-call deltas before discovering invalidation; direct stream consumers + must treat the exception as an explicit instruction to discard those calls and never execute them. Caller + cancellation remains cancellation rather than becoming this provider signal. - Every actionable local `function_call` produces exactly one terminal `function_result`, unless execution pauses for a new user-input request or the run is aborted by `MiddlewareFailure`. - An ordinary exception raised by function middleware or a tool body becomes one terminal error `function_result` @@ -511,6 +523,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Scenario | Required invariant | Primary regression test | |---|---|---| | Single non-streaming call | Call, result, and final assistant message are returned in order. | `packages/core/tests/core/test_function_invocation_logic.py::test_base_client_with_function_calling` | +| Usable call under a non-tool finish reason | A schema-usable call executes once and the normal loop continues in both response modes even when the enclosing response reports `finish_reason="length"`. | `packages/core/tests/core/test_function_invocation_logic.py::test_function_call_with_length_finish_reason_executes_and_continues`, `test_streamed_function_call_with_length_finish_reason_executes_and_continues` | | 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` | @@ -620,6 +633,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Middleware failure batch cancellation | A fatal signal fails the whole parallel batch: in-flight sibling tool invocations are cancelled and awaited before the failure propagates. Cancellation is cooperative — an async sibling stops at its next suspension point; a synchronous tool body already executing in a worker thread cannot be interrupted and may complete its side effects, but its result is discarded and never reaches the transcript, the model, or history, and failure propagation is not delayed behind it. | `TestMiddlewareFailure::test_failure_cancels_concurrent_sibling_tool`, `test_failure_with_sync_sibling_discards_late_result` | | Middleware failure on a service-managed conversation | The continuation state is already persisted when the batch fails, so before propagating, the loop settles the hosted thread: one error `function_result` per dangling call, sent with `tool_choice="none"` in one extra request; the persisted continuation advances to the settlement response (required for response-ID continuations, a no-op for conversation-object ids) and the settlement response is otherwise discarded; a settlement failure never masks the abort. Without a service-managed conversation no extra request is made. | `TestMiddlewareFailure::test_failure_settles_dangling_calls_on_service_conversation`, `test_failure_settles_service_conversation_streaming`, `test_failure_settlement_advances_response_id_continuation`, `test_failure_without_service_conversation_makes_no_settlement_request` | | Middleware failure during approved-tool replay | A fatal abort while the approval-resolution phase replays an approved tool escapes loudly (never absorbed into a rejection result), the tool's original — already service-persisted — call is settled the same way, and the continuation advances; both response modes. | `TestMiddlewareFailure::test_failure_during_approved_replay_settles_and_escapes`, `test_failure_during_approved_replay_streaming` | +| Provider-invalidated local function calls | The provider signal propagates unchanged in both response modes after clearing request budget state and restoring the last valid continuation. Partial streamed call deltas may remain visible, but no new approval, function middleware, tool body, result, follow-up model call, or settlement occurs. Anthropic requires every local call block to close, terminal `stop_reason="tool_use"`, and `message_stop`; non-tool terminal reasons, open blocks, missing `message_stop`, and non-cancellation stream errors after local call start invalidate the call. Hosted/server-only calls and caller cancellation remain unaffected. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_function_call_short_circuits_current_iteration`, `packages/anthropic/tests/test_anthropic_client.py::test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises`, `test_valid_streaming_local_tool_call_executes_and_continues`, `test_streaming_local_tool_call_invalid_terminal_sequences_raise`, `test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation`, `test_streaming_cancellation_after_local_call_is_not_wrapped`, `test_streaming_hosted_tool_pause_turn_is_not_invalidated` | | Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents | | Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent | | Provider tool content after an active limit | Locally actionable calls and local approval requests returned despite `tool_choice="none"` are removed in both response modes. Provider-executed informational call/result pairs, hosted approval requests, and metadata-only streaming updates remain visible; fallback text never replaces retained transcript content. | `test_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped`, `test_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_function_invocation_limit_preserves_hosted_approval_request`, `test_streaming_function_invocation_limit_preserves_hosted_approval_request` | @@ -634,6 +648,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Pending hosted history replay | Stateless hosted approval requests remain replayable until a response is recorded, then both controls become inert. | `packages/openai/tests/openai/test_openai_chat_client.py::test_stateless_history_preserves_pending_hosted_approval_request_until_response` | | Non-history provider plus session | Local history is still auto-injected for approval resume. | `packages/core/tests/core/test_agents.py::test_non_history_context_provider_still_injects_inmemory` | | Hosted per-service-call persistence | A host-managed transcript remains available throughout a local function-call loop without being persisted into the framework session and replayed on the next hosted request. | `packages/foundry_hosting/tests/test_responses.py::TestAgentSessionPersistence::test_per_service_call_persistence_preserves_function_loop_history` | +| Invalidated per-service-call persistence | An invalidated service response, its partial calls, and any nonexistent results are not success-persisted, and invalid content does not advance session continuation state. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_service_call_is_not_persisted` | | Streaming message injection with per-service-call persistence | A streaming response rebuilt from updates mirrors the inner conversation id exactly, including clearing it, and keeps its internal marker, so the next iteration appends only the latest message rather than replaying the whole turn on top of provider-held history, and never persists a conversation id from an earlier injected service call. | `packages/core/tests/core/test_middleware_with_chat.py::TestChatMiddleware::test_message_injection_middleware_streaming_preserves_inner_continuation_state`, `test_message_injection_middleware_streaming_keeps_service_conversation_id_external`, `test_message_injection_middleware_streaming_clears_conversation_id_when_final_call_has_none`, `test_message_injection_middleware_conversation_id_matches_across_streaming_modes`, `packages/core/tests/core/test_harness_agent.py::test_streaming_harness_tool_call_does_not_duplicate_transcript` | | Service-side approval decision | Stored hosted request is skipped; the current approved or rejected hosted response is sent, while local approval controls are omitted from provider input. | `packages/openai/tests/openai/test_openai_chat_client.py::test_prepare_messages_strips_approval_request_but_keeps_response_under_storage`, `test_prepare_messages_drops_local_approval_controls` | | OpenAI approval serialization | Hosted approval id and decision serialize to `mcp_approval_response`; local approvals remain in-process. | `test_prepare_message_for_openai_with_function_approval_response`, `test_prepare_content_for_opentool_approval_response`, `test_function_approval_response_with_mcp_tool_call` | diff --git a/python/CODING_STANDARD.md b/python/CODING_STANDARD.md index 6756c860f96..71ca6225df8 100644 --- a/python/CODING_STANDARD.md +++ b/python/CODING_STANDARD.md @@ -240,6 +240,7 @@ AgentFrameworkException # Base for all AF exceptions │ └── AgentContentFilterException # Agent content filter triggered │ ├── ChatClientException # Chat client lifecycle and communication failures +│ ├── FunctionCallInvalidatedException # Provider invalidated emitted local function calls │ ├── ChatClientInvalidAuthException # Chat client auth failures │ ├── ChatClientInvalidRequestException # Invalid request to chat client │ ├── ChatClientInvalidResponseException # Invalid/unexpected response from chat client @@ -284,6 +285,7 @@ AgentFrameworkException # Base for all AF exceptions | Object in wrong state (e.g., client not initialized) | `RuntimeError` | | External service returns 401/403 | `IntegrationInvalidAuthException` (or `ChatClient`/`Agent` variant) | | External service returns unexpected response | `IntegrationInvalidResponseException` (or variant) | +| Chat provider invalidates local function calls already present in response content | `FunctionCallInvalidatedException` | | Content filter blocks a request | `IntegrationContentFilterException` (or variant) | | Request validation fails before sending to service | `IntegrationInvalidRequestException` (or variant) | | Agent not found in registry | `AgentInvalidRequestException` | diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index aeb4822411e..ca8be5c9247 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import logging import sys from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence @@ -38,6 +39,7 @@ ChatClientException, ChatClientInvalidAuthException, ChatClientInvalidRequestException, + FunctionCallInvalidatedException, ) from agent_framework.observability import ChatTelemetryLayer from anthropic import APIError as AnthropicAPIError @@ -599,15 +601,52 @@ 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] = {} + local_function_call_started = False + open_local_function_call_blocks: set[int] = set() + latest_stop_reason: str | None = None + message_stop_received = False 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, request_state=request_state) + if chunk.type == "content_block_start" and parsed_chunk is not None: + if any( + content.type == "function_call" and not content.informational_only + for content in parsed_chunk.contents + ): + local_function_call_started = True + open_local_function_call_blocks.add(chunk.index) + elif chunk.type == "content_block_stop": + open_local_function_call_blocks.discard(chunk.index) + elif chunk.type == "message_delta" and chunk.delta.stop_reason is not None: + latest_stop_reason = chunk.delta.stop_reason + elif chunk.type == "message_stop": + message_stop_received = True + if local_function_call_started and ( + open_local_function_call_blocks or latest_stop_reason != "tool_use" + ): + raise FunctionCallInvalidatedException( + "Anthropic invalidated local function calls before completing the response." + ) if parsed_chunk: yield parsed_chunk - except AgentFrameworkException: + if local_function_call_started and not message_stop_received: + raise FunctionCallInvalidatedException( + "Anthropic invalidated local function calls because the response ended " + "without message_stop." + ) + except asyncio.CancelledError: + raise + except FunctionCallInvalidatedException: raise except Exception as ex: + if local_function_call_started: + raise FunctionCallInvalidatedException( + "Anthropic invalidated local function calls because the response stream failed.", + inner_exception=ex, + ) from ex + if isinstance(ex, AgentFrameworkException): + raise raise _wrap_anthropic_error(ex) from ex return self._build_response_stream(_stream(), response_format=options.get("response_format")) @@ -1157,12 +1196,20 @@ def _process_message( Returns: A ChatResponse object containing the processed response. """ + contents = self._parse_contents_from_anthropic(message.content, request_state=request_state) + if message.stop_reason != "tool_use" and any( + content.type == "function_call" and not content.informational_only for content in contents + ): + raise FunctionCallInvalidatedException( + "Anthropic invalidated local function calls before completing the response." + ) + return ChatResponse( response_id=message.id, messages=[ Message( role="assistant", - contents=self._parse_contents_from_anthropic(message.content, request_state=request_state), + contents=contents, raw_representation=message, ) ], diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 36a65228abd..31e80a377af 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -16,6 +16,7 @@ ChatResponse, ChatResponseUpdate, Content, + FunctionCallInvalidatedException, FunctionInvocationLayer, FunctionTool, InlineSkill, @@ -2083,9 +2084,10 @@ def _message_start_event(response_id: str) -> MagicMock: return event -def _tool_start_event(*, block_type: str, call_id: str, name: str) -> MagicMock: +def _tool_start_event(*, block_type: str, call_id: str, name: str, index: int = 0) -> MagicMock: event = MagicMock() event.type = "content_block_start" + event.index = index event.content_block.type = block_type event.content_block.id = call_id event.content_block.name = name @@ -2093,6 +2095,19 @@ def _tool_start_event(*, block_type: str, call_id: str, name: str) -> MagicMock: return event +def _content_block_stop_event(index: int = 0) -> MagicMock: + event = MagicMock() + event.type = "content_block_stop" + event.index = index + return event + + +def _message_stop_event() -> MagicMock: + event = MagicMock() + event.type = "message_stop" + return event + + def _argument_delta_event(partial_json: str) -> MagicMock: event = MagicMock() event.type = "content_block_delta" @@ -2109,6 +2124,262 @@ def _message_delta_event(stop_reason: str) -> MagicMock: return event +def _local_tool_message(stop_reason: str) -> MagicMock: + message = MagicMock(spec=BetaMessage) + message.id = "msg_local_tool" + message.model = "claude-test" + message.content = [ + BetaToolUseBlock( + type="tool_use", + id="local-call", + name="local_tool", + input={"value": "a"}, + ) + ] + message.usage = None + message.stop_reason = stop_reason + return message + + +@pytest.mark.parametrize("stop_reason", ["max_tokens", "refusal"]) +async def test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises( + mock_anthropic_client: MagicMock, + stop_reason: str, +) -> None: + """Anthropic terminal reasons other than tool_use invalidate local calls even when arguments parse.""" + executions = 0 + + @tool(name="local_tool", approval_mode="never_require") + def local_tool(value: str) -> str: + nonlocal executions + executions += 1 + return value + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.return_value = _local_tool_message(stop_reason) + + with pytest.raises(FunctionCallInvalidatedException, match="Anthropic invalidated"): + await client.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [local_tool], "max_tokens": 64}, + ) + + assert executions == 0 + assert mock_anthropic_client.beta.messages.create.call_count == 1 + + +def test_non_streaming_hosted_tool_pause_turn_is_not_invalidated( + mock_anthropic_client: MagicMock, +) -> None: + """Hosted server-tool content keeps pause_turn semantics and is not a local invalidation.""" + client = create_test_anthropic_client(mock_anthropic_client) + message = MagicMock(spec=BetaMessage) + message.id = "msg_hosted" + message.model = "claude-test" + hosted_block = MagicMock() + hosted_block.type = "server_tool_use" + hosted_block.id = "hosted-call" + hosted_block.name = "local_tool" + hosted_block.input = {"value": "a"} + message.content = [hosted_block] + message.usage = None + message.stop_reason = "pause_turn" + + response = client._process_message(message, {}) + + assert response.finish_reason == "stop" + assert response.messages[0].contents[0].type == "function_call" + assert response.messages[0].contents[0].informational_only is True + + +async def test_valid_streaming_local_tool_call_executes_and_continues( + mock_anthropic_client: MagicMock, +) -> None: + """Closed local tool_use streams with message_stop retain the normal function loop.""" + executions: list[str] = [] + + @tool(name="local_tool", approval_mode="never_require") + def local_tool(value: str) -> str: + executions.append(value) + return f"found {value}" + + async def local_events() -> Any: + yield _message_start_event("tool-response") + yield _tool_start_event(block_type="tool_use", call_id="local-call", name="local_tool") + yield _argument_delta_event('{"value": "a"}') + yield _content_block_stop_event() + yield _message_delta_event("tool_use") + yield _message_stop_event() + + async def final_events() -> Any: + yield _message_start_event("final-response") + text_start = MagicMock() + text_start.type = "content_block_start" + text_start.index = 0 + text_start.content_block.type = "text" + text_start.content_block.text = "done" + text_start.content_block.citations = None + yield text_start + yield _content_block_stop_event() + yield _message_delta_event("end_turn") + yield _message_stop_event() + + request_count = 0 + + async def create(**kwargs: Any) -> Any: + nonlocal request_count + del kwargs + request_count += 1 + return local_events() if request_count == 1 else final_events() + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.side_effect = create + stream = client.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [local_tool], "max_tokens": 64}, + stream=True, + ) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert executions == ["a"] + assert request_count == 2 + assert any(content.type == "function_result" for update in updates for content in update.contents) + assert response.text == "done" + + +@pytest.mark.parametrize( + ("close_block", "stop_reason", "include_message_stop"), + [ + (True, "max_tokens", True), + (True, "refusal", True), + (False, "tool_use", True), + (True, "tool_use", False), + ], + ids=["max_tokens", "refusal", "open_block", "missing_message_stop"], +) +async def test_streaming_local_tool_call_invalid_terminal_sequences_raise( + mock_anthropic_client: MagicMock, + close_block: bool, + stop_reason: str, + include_message_stop: bool, +) -> None: + """Local streamed calls require closed blocks, tool_use, and message_stop.""" + + async def events() -> Any: + yield _message_start_event("invalid-response") + start = _tool_start_event(block_type="tool_use", call_id="local-call", name="local_tool") + start.content_block.input = {"value": "a"} + yield start + if close_block: + yield _content_block_stop_event() + yield _message_delta_event(stop_reason) + if include_message_stop: + yield _message_stop_event() + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.return_value = events() + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["run"])], + options={"max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + updates: list[ChatResponseUpdate] = [] + + with pytest.raises(FunctionCallInvalidatedException, match="Anthropic invalidated"): + async for update in stream: + updates.append(update) + + assert any(content.type == "function_call" for update in updates for content in update.contents) + + +async def test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation( + mock_anthropic_client: MagicMock, +) -> None: + """A provider failure after local call content preserves the original error as the inner exception.""" + original = RuntimeError("provider stream failed") + + async def events() -> Any: + yield _message_start_event("invalid-response") + yield _tool_start_event(block_type="tool_use", call_id="local-call", name="local_tool") + raise original + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.return_value = events() + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["run"])], + options={"max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + async for _ in stream: + pass + + assert exc_info.value.args[1] is original + assert exc_info.value.__cause__ is original + + +async def test_streaming_cancellation_after_local_call_is_not_wrapped( + mock_anthropic_client: MagicMock, +) -> None: + """Caller cancellation remains asyncio.CancelledError after local call content.""" + + async def events() -> Any: + yield _message_start_event("cancelled-response") + yield _tool_start_event(block_type="tool_use", call_id="local-call", name="local_tool") + raise asyncio.CancelledError + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.return_value = events() + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["run"])], + options={"max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + with pytest.raises(asyncio.CancelledError): + async for _ in stream: + pass + + +async def test_streaming_hosted_tool_pause_turn_is_not_invalidated( + mock_anthropic_client: MagicMock, +) -> None: + """Hosted-only calls keep pause_turn behavior even when their name matches a local tool.""" + + @tool(name="local_tool", approval_mode="never_require") + def local_tool(value: str) -> str: + return value + + async def events() -> Any: + yield _message_start_event("hosted-response") + yield _tool_start_event(block_type="server_tool_use", call_id="hosted-call", name="local_tool") + yield _content_block_stop_event() + yield _message_delta_event("pause_turn") + yield _message_stop_event() + + client = create_test_anthropic_client(mock_anthropic_client) + mock_anthropic_client.beta.messages.create.return_value = events() + stream = client._inner_get_response( # type: ignore[attr-defined] + messages=[Message(role="user", contents=["run"])], + options={"tools": [local_tool], "max_tokens": 64}, + stream=True, + ) + assert isinstance(stream, ResponseStream) + + updates = [update async for update in stream] + response = await stream.get_final_response() + + hosted_calls = [content for update in updates for content in update.contents if content.type == "function_call"] + assert len(hosted_calls) == 1 + assert hosted_calls[0].informational_only is True + assert response.finish_reason == "stop" + + async def test_concurrent_streams_isolate_tool_state_when_peer_fails( mock_anthropic_client: MagicMock, ) -> None: @@ -2136,7 +2407,9 @@ async def primary_events() -> Any: yield _argument_delta_event('{"command":') await asyncio.wait_for(hosted_failed.wait(), timeout=1) yield _argument_delta_event('"pwd"}') + yield _content_block_stop_event() yield _message_delta_event("tool_use") + yield _message_stop_event() async def hosted_events() -> Any: await asyncio.wait_for(primary_started.wait(), timeout=1) @@ -2226,7 +2499,9 @@ async def approval_events(response_id: str, command: str) -> Any: yield _tool_start_event(block_type="tool_use", call_id="shared-approval-call", name="bash") yield _argument_delta_event('{"command":') yield _argument_delta_event(f'"{command}"}}') + yield _content_block_stop_event() yield _message_delta_event("tool_use") + yield _message_stop_event() async def create(**kwargs: Any) -> Any: nonlocal request_count diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index 691c35e9110..c7c05b01a79 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -32,6 +32,7 @@ ) from .exceptions import ( AgentFrameworkException, + FunctionCallInvalidatedException, MiddlewareException, UserInputRequiredException, WorkflowCheckpointException, @@ -513,6 +514,7 @@ "FinalT", "FinishReason", "FinishReasonLiteral", + "FunctionCallInvalidatedException", "FunctionExecutor", "FunctionInvocationConfiguration", "FunctionInvocationContext", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index b6ad94bdf11..3800400eb74 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -334,6 +334,7 @@ from ._workflows._workflow_context import WorkflowContext from ._workflows._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor from .exceptions import ( AgentFrameworkException, + FunctionCallInvalidatedException, MiddlewareException, UserInputRequiredException, WorkflowCheckpointException, @@ -468,6 +469,7 @@ __all__ = [ "FinalT", "FinishReason", "FinishReasonLiteral", + "FunctionCallInvalidatedException", "FunctionExecutor", "FunctionInvocationConfiguration", "FunctionInvocationContext", diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 805c3df5849..289ace6758f 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -46,7 +46,7 @@ from pydantic import BaseModel, Field, ValidationError, create_model from ._serialization import SerializationMixin -from .exceptions import ToolException, UserInputRequiredException +from .exceptions import FunctionCallInvalidatedException, ToolException, UserInputRequiredException from .observability import ( OPERATION_DURATION_BUCKET_BOUNDARIES, OtelAttr, @@ -77,7 +77,7 @@ FunctionMiddlewareTypes, MiddlewareTypes, ) - from ._sessions import AgentSession + from ._sessions import AgentSession, ServiceSessionId from ._types import ( ChatOptions, ChatResponse, @@ -3378,6 +3378,19 @@ def _clear_budget_state_from_session(invocation_session: AgentSession | None) -> invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) +def _clear_invalidated_function_invocation_state( + invocation_session: AgentSession | None, + budget_state: dict[str, Any], + service_session_id: str | ServiceSessionId | None, +) -> None: + """Discard invalidated call state and restore the last valid service continuation.""" + budget_state.clear() + if invocation_session is None: + return + invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) + invocation_session.service_session_id = service_session_id + + def _apply_batch_limit_decision( action: Literal["continue", "return", "stop"], options: dict[str, Any], @@ -3965,17 +3978,28 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: alternate model turns and local execution until a terminal response or safety limit is reached. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - response = cast( - ChatResponse[Any], - await super_get_response( - messages=prepared_messages, - stream=False, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), + service_session_id_before_call = ( + invocation_session.service_session_id if invocation_session is not None else None ) + try: + response = cast( + ChatResponse[Any], + await super_get_response( + messages=prepared_messages, + stream=False, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), + ) + except FunctionCallInvalidatedException: + _clear_invalidated_function_invocation_state( + invocation_session, + budget_state, + service_session_id_before_call, + ) + raise if options.get("tool_choice") == "none" and budget_state.get("truncated"): _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) @@ -4040,17 +4064,28 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - response = cast( - ChatResponse[Any], - await super_get_response( - messages=prepared_messages, - stream=False, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), + service_session_id_before_call = ( + invocation_session.service_session_id if invocation_session is not None else None ) + try: + response = cast( + ChatResponse[Any], + await super_get_response( + messages=prepared_messages, + stream=False, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), + ) + except FunctionCallInvalidatedException: + _clear_invalidated_function_invocation_state( + invocation_session, + budget_state, + service_session_id_before_call, + ) + raise _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) self._update_function_invocation_continuation_state( @@ -4147,81 +4182,92 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: stream each model turn, finalize it, execute its calls, then advance the transcript. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - inner_stream = cast( - "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( - messages=prepared_messages, - stream=True, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), + service_session_id_before_call = ( + invocation_session.service_session_id if invocation_session is not None else None ) - await inner_stream - drop_unexecutable_calls = options.get("tool_choice") == "none" and budget_state.get("truncated") - streamed_identities_by_call_id: dict[str, tuple[str, str]] = {} - streamed_names_by_call_id: dict[str, str] = {} - last_streamed_identity: tuple[str, str] | None = None - warned_empty_call_ids: set[str] = set() - async for update in inner_stream: - for content in update.contents: - if content.type != "function_call": - continue - if not _is_actionable_function_call(content): - continue - had_occurrence_id = content.id is not None - provider_call_id = content.call_id - identity = streamed_identities_by_call_id.get(provider_call_id) if provider_call_id else None - if ( - identity is not None - and provider_call_id is not None - and content.id is None - and content.name - and ( - streamed_names_by_call_id.get(provider_call_id) != content.name - or isinstance(content.arguments, Mapping) - ) - ): - identity = None - if identity is None and not provider_call_id and not content.name: - identity = last_streamed_identity - - if identity is None: - occurrence_id = content.id or _generate_function_call_occurrence_id() - effective_call_id = provider_call_id or ("" if had_occurrence_id else occurrence_id) - else: - occurrence_id, effective_call_id = identity - if content.id is not None: - occurrence_id = content.id - if provider_call_id: - effective_call_id = provider_call_id - - content.id = occurrence_id - if not content.call_id and not had_occurrence_id: - content.call_id = effective_call_id - if identity is None and occurrence_id not in warned_empty_call_ids: - warnings.warn( - "An actionable function_call had an empty call_id. Agent Framework used its generated " - "Content.id for local correlation. Providers should supply and preserve their service " - "call_id; this fallback will be removed in a future release.", - FutureWarning, - stacklevel=3, + try: + inner_stream = cast( + "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", + super_get_response( + messages=prepared_messages, + stream=True, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), + ) + await inner_stream + drop_unexecutable_calls = options.get("tool_choice") == "none" and budget_state.get("truncated") + streamed_identities_by_call_id: dict[str, tuple[str, str]] = {} + streamed_names_by_call_id: dict[str, str] = {} + last_streamed_identity: tuple[str, str] | None = None + warned_empty_call_ids: set[str] = set() + async for update in inner_stream: + for content in update.contents: + if content.type != "function_call": + continue + if not _is_actionable_function_call(content): + continue + had_occurrence_id = content.id is not None + provider_call_id = content.call_id + identity = streamed_identities_by_call_id.get(provider_call_id) if provider_call_id else None + if ( + identity is not None + and provider_call_id is not None + and content.id is None + and content.name + and ( + streamed_names_by_call_id.get(provider_call_id) != content.name + or isinstance(content.arguments, Mapping) ) - warned_empty_call_ids.add(occurrence_id) - identity = (occurrence_id, effective_call_id) - if effective_call_id: - streamed_identities_by_call_id[effective_call_id] = identity - if content.name: - streamed_names_by_call_id[effective_call_id] = content.name - last_streamed_identity = identity - if drop_unexecutable_calls: - update = _drop_unexecutable_tool_contents_from_update(update) - if update is None: - continue - yield update - - response = await inner_stream.get_final_response() + ): + identity = None + if identity is None and not provider_call_id and not content.name: + identity = last_streamed_identity + + if identity is None: + occurrence_id = content.id or _generate_function_call_occurrence_id() + effective_call_id = provider_call_id or ("" if had_occurrence_id else occurrence_id) + else: + occurrence_id, effective_call_id = identity + if content.id is not None: + occurrence_id = content.id + if provider_call_id: + effective_call_id = provider_call_id + + content.id = occurrence_id + if not content.call_id and not had_occurrence_id: + content.call_id = effective_call_id + if identity is None and occurrence_id not in warned_empty_call_ids: + warnings.warn( + "An actionable function_call had an empty call_id. Agent Framework used its " + "generated Content.id for local correlation. Providers should supply and preserve " + "their service call_id; this fallback will be removed in a future release.", + FutureWarning, + stacklevel=3, + ) + warned_empty_call_ids.add(occurrence_id) + identity = (occurrence_id, effective_call_id) + if effective_call_id: + streamed_identities_by_call_id[effective_call_id] = identity + if content.name: + streamed_names_by_call_id[effective_call_id] = content.name + last_streamed_identity = identity + if drop_unexecutable_calls: + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue + yield update + + response = await inner_stream.get_final_response() + except FunctionCallInvalidatedException: + _clear_invalidated_function_invocation_state( + invocation_session, + budget_state, + service_session_id_before_call, + ) + raise fallback_added = False if options.get("tool_choice") == "none" and budget_state.get("truncated"): fallback_added = _ensure_function_invocation_limit_fallback_response(response) @@ -4296,24 +4342,35 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - final_inner_stream = cast( - "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( - messages=prepared_messages, - stream=True, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), + service_session_id_before_call = ( + invocation_session.service_session_id if invocation_session is not None else None ) - await final_inner_stream - async for update in final_inner_stream: - update = _drop_unexecutable_tool_contents_from_update(update) - if update is None: - continue - yield update - final_response = await final_inner_stream.get_final_response() + try: + final_inner_stream = cast( + "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", + super_get_response( + messages=prepared_messages, + stream=True, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), + ) + await final_inner_stream + async for update in final_inner_stream: + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue + yield update + final_response = await final_inner_stream.get_final_response() + except FunctionCallInvalidatedException: + _clear_invalidated_function_invocation_state( + invocation_session, + budget_state, + service_session_id_before_call, + ) + raise fallback_added = _ensure_function_invocation_limit_fallback_response(final_response) self._update_function_invocation_continuation_state( request_kwargs, diff --git a/python/packages/core/agent_framework/exceptions.py b/python/packages/core/agent_framework/exceptions.py index b96aab6b916..88bd8b27d43 100644 --- a/python/packages/core/agent_framework/exceptions.py +++ b/python/packages/core/agent_framework/exceptions.py @@ -82,6 +82,12 @@ class ChatClientException(AgentFrameworkException): pass +class FunctionCallInvalidatedException(ChatClientException): + """A provider invalidated local function calls already present in response content.""" + + pass + + class ChatClientInvalidAuthException(ChatClientException): """An authentication error occurred in a chat client.""" diff --git a/python/packages/core/tests/core/test_exceptions.py b/python/packages/core/tests/core/test_exceptions.py index bf164cdbe24..e19a6e6c243 100644 --- a/python/packages/core/tests/core/test_exceptions.py +++ b/python/packages/core/tests/core/test_exceptions.py @@ -2,7 +2,8 @@ """Tests for AgentFrameworkException inner_exception handling.""" -from agent_framework import AgentFrameworkException +from agent_framework import AgentFrameworkException, FunctionCallInvalidatedException +from agent_framework.exceptions import ChatClientException def test_exception_with_inner_exception(): @@ -25,3 +26,18 @@ def test_exception_inner_exception_none_explicit(): exc = AgentFrameworkException("test message", inner_exception=None) assert exc.args == ("test message",) assert len(exc.args) == 1 + + +def test_function_call_invalidated_exception_is_public_and_picklable() -> None: + """The provider invalidation signal is a public chat-client exception.""" + import pickle + + inner = RuntimeError("provider stream failed") + exc = FunctionCallInvalidatedException("local function calls were invalidated", inner_exception=inner) + restored = pickle.loads(pickle.dumps(exc)) + + assert isinstance(exc, ChatClientException) + assert isinstance(restored, FunctionCallInvalidatedException) + assert restored.args[0] == "local function calls were invalidated" + assert isinstance(restored.args[1], RuntimeError) + assert str(restored.args[1]) == "provider stream failed" 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 d829aaa4c31..10b1d85f7e4 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -19,6 +19,7 @@ ChatResponse, ChatResponseUpdate, Content, + FunctionCallInvalidatedException, Message, ResponseStream, SupportsChatGetResponse, @@ -1001,6 +1002,257 @@ def ai_func(arg1: str) -> str: assert response.messages[2].text == "done" +async def test_function_call_with_length_finish_reason_executes_and_continues( + chat_client_base: SupportsChatGetResponse, +) -> None: + """Usable function arguments remain actionable regardless of the enclosing finish reason.""" + executions: list[str] = [] + + @tool(name="lookup", approval_mode="never_require") + def lookup(value: str) -> str: + executions.append(value) + return f"found {value}" + + chat_client_base.run_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + ChatResponse( + messages=Message( + role="assistant", + contents=[Content.from_function_call(call_id="call-1", name="lookup", arguments={"value": "a"})], + ), + finish_reason="length", + ), + ChatResponse(messages=Message(role="assistant", contents=["done"]), finish_reason="stop"), + ] + + response = await chat_client_base.get_response( + [Message(role="user", contents=["look up a"])], + options={"tools": [lookup]}, + ) + + assert executions == ["a"] + assert chat_client_base.call_count == 2 # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + assert [message.role for message in response.messages] == ["assistant", "tool", "assistant"] + assert response.messages[1].contents[0].result == "found a" + assert response.text == "done" + + +async def test_streamed_function_call_with_length_finish_reason_executes_and_continues( + chat_client_base: SupportsChatGetResponse, +) -> None: + """Streaming uses argument validity, not the enclosing finish reason, to decide execution.""" + executions: list[str] = [] + + @tool(name="lookup", approval_mode="never_require") + def lookup(value: str) -> str: + executions.append(value) + return f"found {value}" + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call(call_id="call-1", name="lookup", arguments={"value": "a"})], + finish_reason="length", + ) + ], + [ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")], finish_reason="stop")], + ] + + stream = chat_client_base.get_response( + [Message(role="user", contents=["look up a"])], + options={"tools": [lookup]}, + stream=True, + ) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert executions == ["a"] + assert chat_client_base.call_count == 2 # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + assert [content.type for update in updates for content in update.contents] == [ + "function_call", + "function_result", + "text", + ] + assert response.text == "done" + + +@pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) +async def test_invalidated_function_call_short_circuits_current_iteration( + chat_client_base: SupportsChatGetResponse, + streaming: bool, +) -> None: + """Provider invalidation propagates without creating any local function side effect.""" + from agent_framework._sessions import AgentSession + from agent_framework._tools import _FUNCTION_INVOCATION_BUDGET_STATE_KEY + + middleware_calls = 0 + tool_calls = 0 + provider_calls = 0 + invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + + class TrackingMiddleware(FunctionMiddleware): + async def process( + self, + context: FunctionInvocationContext, + call_next: Callable[[], Awaitable[None]], + ) -> None: + nonlocal middleware_calls + middleware_calls += 1 + await call_next() + + @tool(name="guarded", approval_mode="always_require") + def guarded(value: str) -> str: + nonlocal tool_calls + tool_calls += 1 + return value + + session = AgentSession() + session.service_session_id = "existing-session" + budget_state: dict[str, Any] = {"attempt_count": 0, "total_function_calls": 2} + session.state[_FUNCTION_INVOCATION_BUDGET_STATE_KEY] = budget_state + yielded: list[ChatResponseUpdate] = [] + + if streaming: + + def invalid_stream(**kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + del kwargs + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + nonlocal provider_calls + provider_calls += 1 + yield ChatResponseUpdate( + role="assistant", + contents=[ + Content.from_function_call( + call_id="invalid-call", + name="guarded", + arguments={"value": "x"}, + ) + ], + ) + raise invalidated + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + chat_client_base._get_streaming_response = invalid_stream # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + response_stream = chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [guarded]}, + stream=True, + client_kwargs={ + "middleware": [TrackingMiddleware()], + "session": session, + _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, + }, + ) + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + async for update in response_stream: + yielded.append(update) + else: + + async def invalid_response(**kwargs: Any) -> ChatResponse: + nonlocal provider_calls + del kwargs + provider_calls += 1 + raise invalidated + + chat_client_base._get_non_streaming_response = invalid_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + await chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [guarded]}, + client_kwargs={ + "middleware": [TrackingMiddleware()], + "session": session, + _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, + }, + ) + + assert exc_info.value is invalidated + assert provider_calls == 1 + assert middleware_calls == 0 + assert tool_calls == 0 + assert budget_state == {} + assert _FUNCTION_INVOCATION_BUDGET_STATE_KEY not in session.state + assert session.service_session_id == "existing-session" + assert not any( + content.type in {"function_approval_request", "function_result"} + for update in yielded + for content in update.contents + ) + if streaming: + assert [content.type for update in yielded for content in update.contents] == ["function_call"] + + +@pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) +async def test_invalidated_service_call_is_not_persisted( + chat_client_base: SupportsChatGetResponse, + streaming: bool, +) -> None: + """Per-service-call history never success-persists invalidated calls or partial results.""" + from agent_framework import InMemoryHistoryProvider + from agent_framework._sessions import AgentSession + + provider = InMemoryHistoryProvider() + session = AgentSession() + seed = Message(role="assistant", contents=["existing history"]) + session.state[provider.source_id] = {"messages": [seed]} + invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + tool_calls = 0 + + @tool(name="local_tool", approval_mode="never_require") + def local_tool() -> str: + nonlocal tool_calls + tool_calls += 1 + return "result" + + if streaming: + + class InvalidRawUpdate: + conversation_id = "invalid-session" + + def invalid_stream(**kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + del kwargs + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call(call_id="invalid-call", name="local_tool", arguments={})], + raw_representation=InvalidRawUpdate(), + ) + raise invalidated + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + chat_client_base._get_streaming_response = invalid_stream # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + else: + + async def invalid_response(**kwargs: Any) -> ChatResponse: + del kwargs + raise invalidated + + chat_client_base._get_non_streaming_response = invalid_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + + agent = Agent( + client=chat_client_base, + tools=[local_tool], + context_providers=[provider], + require_per_service_call_history_persistence=True, + ) + + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + if streaming: + async for _ in agent.run("run", session=session, stream=True): + pass + else: + await agent.run("run", session=session) + + assert exc_info.value is invalidated + assert tool_calls == 0 + assert session.state[provider.source_id]["messages"] == [seed] + assert session.service_session_id is None + + async def test_base_client_with_function_calling_string_input(chat_client_base: SupportsChatGetResponse): exec_counter = 0 From 0d6a2ede852fab485853d8450c0de461fc97f919 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 14:04:31 +0200 Subject: [PATCH 2/6] Python: Simplify invalidated call cleanup Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../specs/004-python-function-calling-loop.md | 2 +- .../packages/core/agent_framework/_tools.py | 340 +++++++++--------- .../core/test_function_invocation_logic.py | 120 ++++++- 3 files changed, 292 insertions(+), 170 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 13e357482e2..2e6fa4328bc 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -633,7 +633,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Middleware failure batch cancellation | A fatal signal fails the whole parallel batch: in-flight sibling tool invocations are cancelled and awaited before the failure propagates. Cancellation is cooperative — an async sibling stops at its next suspension point; a synchronous tool body already executing in a worker thread cannot be interrupted and may complete its side effects, but its result is discarded and never reaches the transcript, the model, or history, and failure propagation is not delayed behind it. | `TestMiddlewareFailure::test_failure_cancels_concurrent_sibling_tool`, `test_failure_with_sync_sibling_discards_late_result` | | Middleware failure on a service-managed conversation | The continuation state is already persisted when the batch fails, so before propagating, the loop settles the hosted thread: one error `function_result` per dangling call, sent with `tool_choice="none"` in one extra request; the persisted continuation advances to the settlement response (required for response-ID continuations, a no-op for conversation-object ids) and the settlement response is otherwise discarded; a settlement failure never masks the abort. Without a service-managed conversation no extra request is made. | `TestMiddlewareFailure::test_failure_settles_dangling_calls_on_service_conversation`, `test_failure_settles_service_conversation_streaming`, `test_failure_settlement_advances_response_id_continuation`, `test_failure_without_service_conversation_makes_no_settlement_request` | | Middleware failure during approved-tool replay | A fatal abort while the approval-resolution phase replays an approved tool escapes loudly (never absorbed into a rejection result), the tool's original — already service-persisted — call is settled the same way, and the continuation advances; both response modes. | `TestMiddlewareFailure::test_failure_during_approved_replay_settles_and_escapes`, `test_failure_during_approved_replay_streaming` | -| Provider-invalidated local function calls | The provider signal propagates unchanged in both response modes after clearing request budget state and restoring the last valid continuation. Partial streamed call deltas may remain visible, but no new approval, function middleware, tool body, result, follow-up model call, or settlement occurs. Anthropic requires every local call block to close, terminal `stop_reason="tool_use"`, and `message_stop`; non-tool terminal reasons, open blocks, missing `message_stop`, and non-cancellation stream errors after local call start invalidate the call. Hosted/server-only calls and caller cancellation remain unaffected. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_function_call_short_circuits_current_iteration`, `packages/anthropic/tests/test_anthropic_client.py::test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises`, `test_valid_streaming_local_tool_call_executes_and_continues`, `test_streaming_local_tool_call_invalid_terminal_sequences_raise`, `test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation`, `test_streaming_cancellation_after_local_call_is_not_wrapped`, `test_streaming_hosted_tool_pause_turn_is_not_invalidated` | +| Provider-invalidated local function calls | The provider signal propagates unchanged in both response modes after clearing request budget state and restoring the last valid continuation. Partial streamed call deltas may remain visible, but no new approval, function middleware, tool body, result, follow-up model call, or settlement occurs. Anthropic requires every local call block to close, terminal `stop_reason="tool_use"`, and `message_stop`; non-tool terminal reasons, open blocks, missing `message_stop`, and non-cancellation stream errors after local call start invalidate the call. Hosted/server-only calls and caller cancellation remain unaffected. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_function_call_short_circuits_current_iteration`, `test_invalidated_final_no_tool_call_preserves_prior_continuation`, `packages/anthropic/tests/test_anthropic_client.py::test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises`, `test_valid_streaming_local_tool_call_executes_and_continues`, `test_streaming_local_tool_call_invalid_terminal_sequences_raise`, `test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation`, `test_streaming_cancellation_after_local_call_is_not_wrapped`, `test_streaming_hosted_tool_pause_turn_is_not_invalidated` | | Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents | | Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent | | Provider tool content after an active limit | Locally actionable calls and local approval requests returned despite `tool_choice="none"` are removed in both response modes. Provider-executed informational call/result pairs, hosted approval requests, and metadata-only streaming updates remain visible; fallback text never replaces retained transcript content. | `test_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped`, `test_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_function_invocation_limit_preserves_hosted_approval_request`, `test_streaming_function_invocation_limit_preserves_hosted_approval_request` | diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 289ace6758f..cfa8528abf6 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -77,7 +77,7 @@ FunctionMiddlewareTypes, MiddlewareTypes, ) - from ._sessions import AgentSession, ServiceSessionId + from ._sessions import AgentSession from ._types import ( ChatOptions, ChatResponse, @@ -3378,17 +3378,53 @@ def _clear_budget_state_from_session(invocation_session: AgentSession | None) -> invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) -def _clear_invalidated_function_invocation_state( +def _function_call_invalidation_cleanup( invocation_session: AgentSession | None, budget_state: dict[str, Any], - service_session_id: str | ServiceSessionId | None, -) -> None: - """Discard invalidated call state and restore the last valid service continuation.""" - budget_state.clear() - if invocation_session is None: - return - invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) - invocation_session.service_session_id = service_session_id + stream_error: list[FunctionCallInvalidatedException] | None = None, +) -> Callable[[FunctionCallInvalidatedException], None]: + """Capture the last valid continuation and return invalidation cleanup.""" + service_session_id = invocation_session.service_session_id if invocation_session is not None else None + + def cleanup(error: FunctionCallInvalidatedException) -> None: + if stream_error is not None: + stream_error[:] = [error] + budget_state.clear() + if invocation_session is None: + return + invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) + invocation_session.service_session_id = service_session_id + + return cleanup + + +_InvalidationResultT = TypeVar("_InvalidationResultT") + + +async def _await_provider_call( + operation: Callable[..., Awaitable[_InvalidationResultT]], + on_invalidated: Callable[[FunctionCallInvalidatedException], None], + **kwargs: Any, +) -> _InvalidationResultT: + """Await one provider operation and clean up invalidated local calls.""" + try: + return await operation(**kwargs) + except FunctionCallInvalidatedException as error: + on_invalidated(error) + raise + + +async def _iterate_provider_stream( + stream: AsyncIterable[_InvalidationResultT], + on_invalidated: Callable[[FunctionCallInvalidatedException], None], +) -> AsyncIterable[_InvalidationResultT]: + """Yield one provider stream and clean up invalidated local calls.""" + try: + async for item in stream: + yield item + except FunctionCallInvalidatedException as error: + on_invalidated(error) + raise def _apply_batch_limit_decision( @@ -3978,28 +4014,20 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: alternate model turns and local execution until a terminal response or safety limit is reached. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - service_session_id_before_call = ( - invocation_session.service_session_id if invocation_session is not None else None + on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state) + response = cast( + ChatResponse[Any], + await _await_provider_call( + super_get_response, + on_invalidated, + messages=prepared_messages, + stream=False, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), ) - try: - response = cast( - ChatResponse[Any], - await super_get_response( - messages=prepared_messages, - stream=False, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), - ) - except FunctionCallInvalidatedException: - _clear_invalidated_function_invocation_state( - invocation_session, - budget_state, - service_session_id_before_call, - ) - raise if options.get("tool_choice") == "none" and budget_state.get("truncated"): _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) @@ -4064,28 +4092,20 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - service_session_id_before_call = ( - invocation_session.service_session_id if invocation_session is not None else None + on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state) + response = cast( + ChatResponse[Any], + await _await_provider_call( + super_get_response, + on_invalidated, + messages=prepared_messages, + stream=False, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), ) - try: - response = cast( - ChatResponse[Any], - await super_get_response( - messages=prepared_messages, - stream=False, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), - ) - except FunctionCallInvalidatedException: - _clear_invalidated_function_invocation_state( - invocation_session, - budget_state, - service_session_id_before_call, - ) - raise _ensure_function_invocation_limit_fallback_response(response) aggregated_usage = add_usage_details(aggregated_usage, response.usage_details) self._update_function_invocation_continuation_state( @@ -4113,6 +4133,7 @@ async def _stream_response_with_function_invocation( budget_state: dict[str, Any], max_errors: int, middleware_pipeline: FunctionMiddlewarePipeline | None = None, + invalidation_error: list[FunctionCallInvalidatedException], ) -> AsyncIterable[ChatResponseUpdate]: """Run the streaming function invocation loop.""" from ._middleware import MiddlewareFailure @@ -4182,92 +4203,83 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: stream each model turn, finalize it, execute its calls, then advance the transcript. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - service_session_id_before_call = ( - invocation_session.service_session_id if invocation_session is not None else None + on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state, invalidation_error) + inner_stream = cast( + "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", + await _await_provider_call( + super_get_response, + on_invalidated, + messages=prepared_messages, + stream=True, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), ) - try: - inner_stream = cast( - "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( - messages=prepared_messages, - stream=True, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), - ) - await inner_stream - drop_unexecutable_calls = options.get("tool_choice") == "none" and budget_state.get("truncated") - streamed_identities_by_call_id: dict[str, tuple[str, str]] = {} - streamed_names_by_call_id: dict[str, str] = {} - last_streamed_identity: tuple[str, str] | None = None - warned_empty_call_ids: set[str] = set() - async for update in inner_stream: - for content in update.contents: - if content.type != "function_call": - continue - if not _is_actionable_function_call(content): - continue - had_occurrence_id = content.id is not None - provider_call_id = content.call_id - identity = streamed_identities_by_call_id.get(provider_call_id) if provider_call_id else None - if ( - identity is not None - and provider_call_id is not None - and content.id is None - and content.name - and ( - streamed_names_by_call_id.get(provider_call_id) != content.name - or isinstance(content.arguments, Mapping) + drop_unexecutable_calls = options.get("tool_choice") == "none" and budget_state.get("truncated") + streamed_identities_by_call_id: dict[str, tuple[str, str]] = {} + streamed_names_by_call_id: dict[str, str] = {} + last_streamed_identity: tuple[str, str] | None = None + warned_empty_call_ids: set[str] = set() + async for update in _iterate_provider_stream(inner_stream, on_invalidated): + for content in update.contents: + if content.type != "function_call": + continue + if not _is_actionable_function_call(content): + continue + had_occurrence_id = content.id is not None + provider_call_id = content.call_id + identity = streamed_identities_by_call_id.get(provider_call_id) if provider_call_id else None + if ( + identity is not None + and provider_call_id is not None + and content.id is None + and content.name + and ( + streamed_names_by_call_id.get(provider_call_id) != content.name + or isinstance(content.arguments, Mapping) + ) + ): + identity = None + if identity is None and not provider_call_id and not content.name: + identity = last_streamed_identity + + if identity is None: + occurrence_id = content.id or _generate_function_call_occurrence_id() + effective_call_id = provider_call_id or ("" if had_occurrence_id else occurrence_id) + else: + occurrence_id, effective_call_id = identity + if content.id is not None: + occurrence_id = content.id + if provider_call_id: + effective_call_id = provider_call_id + + content.id = occurrence_id + if not content.call_id and not had_occurrence_id: + content.call_id = effective_call_id + if identity is None and occurrence_id not in warned_empty_call_ids: + warnings.warn( + "An actionable function_call had an empty call_id. Agent Framework used its generated " + "Content.id for local correlation. Providers should supply and preserve their service " + "call_id; this fallback will be removed in a future release.", + FutureWarning, + stacklevel=3, ) - ): - identity = None - if identity is None and not provider_call_id and not content.name: - identity = last_streamed_identity - - if identity is None: - occurrence_id = content.id or _generate_function_call_occurrence_id() - effective_call_id = provider_call_id or ("" if had_occurrence_id else occurrence_id) - else: - occurrence_id, effective_call_id = identity - if content.id is not None: - occurrence_id = content.id - if provider_call_id: - effective_call_id = provider_call_id - - content.id = occurrence_id - if not content.call_id and not had_occurrence_id: - content.call_id = effective_call_id - if identity is None and occurrence_id not in warned_empty_call_ids: - warnings.warn( - "An actionable function_call had an empty call_id. Agent Framework used its " - "generated Content.id for local correlation. Providers should supply and preserve " - "their service call_id; this fallback will be removed in a future release.", - FutureWarning, - stacklevel=3, - ) - warned_empty_call_ids.add(occurrence_id) - identity = (occurrence_id, effective_call_id) - if effective_call_id: - streamed_identities_by_call_id[effective_call_id] = identity - if content.name: - streamed_names_by_call_id[effective_call_id] = content.name - last_streamed_identity = identity - if drop_unexecutable_calls: - update = _drop_unexecutable_tool_contents_from_update(update) - if update is None: - continue - yield update - - response = await inner_stream.get_final_response() - except FunctionCallInvalidatedException: - _clear_invalidated_function_invocation_state( - invocation_session, - budget_state, - service_session_id_before_call, - ) - raise + warned_empty_call_ids.add(occurrence_id) + identity = (occurrence_id, effective_call_id) + if effective_call_id: + streamed_identities_by_call_id[effective_call_id] = identity + if content.name: + streamed_names_by_call_id[effective_call_id] = content.name + last_streamed_identity = identity + if drop_unexecutable_calls: + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue + yield update + + response = await _await_provider_call(inner_stream.get_final_response, on_invalidated) fallback_added = False if options.get("tool_choice") == "none" and budget_state.get("truncated"): fallback_added = _ensure_function_invocation_limit_fallback_response(response) @@ -4342,35 +4354,26 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - service_session_id_before_call = ( - invocation_session.service_session_id if invocation_session is not None else None + on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state, invalidation_error) + final_inner_stream = cast( + "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", + await _await_provider_call( + super_get_response, + on_invalidated, + messages=prepared_messages, + stream=True, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ), ) - try: - final_inner_stream = cast( - "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( - messages=prepared_messages, - stream=True, - options=options, - compaction_strategy=compaction_strategy, - tokenizer=tokenizer, - client_kwargs=request_kwargs, - ), - ) - await final_inner_stream - async for update in final_inner_stream: - update = _drop_unexecutable_tool_contents_from_update(update) - if update is None: - continue - yield update - final_response = await final_inner_stream.get_final_response() - except FunctionCallInvalidatedException: - _clear_invalidated_function_invocation_state( - invocation_session, - budget_state, - service_session_id_before_call, - ) - raise + async for update in _iterate_provider_stream(final_inner_stream, on_invalidated): + update = _drop_unexecutable_tool_contents_from_update(update) + if update is None: + continue + yield update + final_response = await _await_provider_call(final_inner_stream.get_final_response, on_invalidated) fallback_added = _ensure_function_invocation_limit_fallback_response(final_response) self._update_function_invocation_continuation_state( request_kwargs, @@ -4553,6 +4556,12 @@ def get_response( ) response_format = mutable_options.get("response_format") + invalidation_error: list[FunctionCallInvalidatedException] = [] + + def finalize_stream(updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]: + if invalidation_error: + raise invalidation_error[0] + return ChatResponse.from_updates(updates, output_format_type=response_format) return ResponseStream( self._stream_response_with_function_invocation( @@ -4567,8 +4576,9 @@ def get_response( budget_state=budget_state, max_errors=max_errors, middleware_pipeline=function_middleware_pipeline, + invalidation_error=invalidation_error, ), - finalizer=partial(ChatResponse.from_updates, output_format_type=response_format), + finalizer=finalize_stream, ) 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 10b1d85f7e4..5b3806780d0 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -1240,11 +1240,16 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: require_per_service_call_history_persistence=True, ) - with pytest.raises(FunctionCallInvalidatedException) as exc_info: - if streaming: - async for _ in agent.run("run", session=session, stream=True): + if streaming: + stream = agent.run("run", session=session, stream=True) + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + async for _ in stream: pass - else: + with pytest.raises(FunctionCallInvalidatedException) as final_exc_info: + await stream.get_final_response() + assert final_exc_info.value is invalidated + else: + with pytest.raises(FunctionCallInvalidatedException) as exc_info: await agent.run("run", session=session) assert exc_info.value is invalidated @@ -1253,6 +1258,113 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: assert session.service_session_id is None +@pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) +async def test_invalidated_final_no_tool_call_preserves_prior_continuation( + chat_client_base: SupportsChatGetResponse, + streaming: bool, +) -> None: + """Providers can invalidate unexpected local calls returned after tools were disabled.""" + from agent_framework._sessions import AgentSession + from agent_framework._tools import _FUNCTION_INVOCATION_BUDGET_STATE_KEY + + tool_calls = 0 + provider_calls = 0 + invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + + @tool(name="local_tool", approval_mode="never_require") + def local_tool() -> str: + nonlocal tool_calls + tool_calls += 1 + return "result" + + function_call = Content.from_function_call(call_id="valid-call", name="local_tool", arguments={}) + session = AgentSession() + session.service_session_id = "starting-continuation" + budget_state: dict[str, Any] = {"attempt_count": 0} + session.state[_FUNCTION_INVOCATION_BUDGET_STATE_KEY] = budget_state + chat_client_base.function_invocation_configuration["max_iterations"] = 1 # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + + if streaming: + + def scripted_stream( + *, + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + nonlocal provider_calls + del kwargs + provider_calls += 1 + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + if provider_calls == 1: + yield ChatResponseUpdate( + role="assistant", + contents=[function_call], + finish_reason="tool_calls", + conversation_id="valid-continuation", + ) + return + assert options["tool_choice"] == "none" + yield ChatResponseUpdate( + role="assistant", + contents=[Content.from_function_call(call_id="invalid-call", name="local_tool", arguments={})], + conversation_id="invalid-continuation", + ) + raise invalidated + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + chat_client_base._get_streaming_response = scripted_stream # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + stream = chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [local_tool]}, + stream=True, + client_kwargs={ + "session": session, + _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, + }, + ) + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + async for _ in stream: + pass + else: + + async def scripted_response( + *, + options: dict[str, Any], + **kwargs: Any, + ) -> ChatResponse: + nonlocal provider_calls + del kwargs + provider_calls += 1 + if provider_calls == 1: + return ChatResponse( + messages=Message(role="assistant", contents=[function_call]), + finish_reason="tool_calls", + conversation_id="valid-continuation", + ) + assert options["tool_choice"] == "none" + raise invalidated + + chat_client_base._get_non_streaming_response = scripted_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + with pytest.raises(FunctionCallInvalidatedException) as exc_info: + await chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": [local_tool]}, + client_kwargs={ + "session": session, + _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, + }, + ) + + assert exc_info.value is invalidated + assert provider_calls == 2 + assert tool_calls == 1 + assert budget_state == {} + assert _FUNCTION_INVOCATION_BUDGET_STATE_KEY not in session.state + assert session.service_session_id == "valid-continuation" + + async def test_base_client_with_function_calling_string_input(chat_client_base: SupportsChatGetResponse): exec_counter = 0 From cff936c2b17bf4286966c5220887aae6d5b6046a Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 14:21:09 +0200 Subject: [PATCH 3/6] Python: Rename response invalidation exception Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...-handle-provider-invalidated-responses.md} | 12 ++++---- .../specs/004-python-function-calling-loop.md | 11 +++---- python/CODING_STANDARD.md | 4 +-- .../agent_framework_anthropic/_chat_client.py | 25 +++++++++------- .../anthropic/tests/test_anthropic_client.py | 8 ++--- .../packages/core/agent_framework/__init__.py | 4 +-- .../core/agent_framework/__init__.pyi | 4 +-- .../packages/core/agent_framework/_tools.py | 30 +++++++++---------- .../core/agent_framework/exceptions.py | 4 +-- .../core/tests/core/test_exceptions.py | 15 +++++----- .../core/test_function_invocation_logic.py | 28 ++++++++--------- 11 files changed, 75 insertions(+), 70 deletions(-) rename docs/decisions/{0041-handle-provider-invalidated-local-function-calls.md => 0041-handle-provider-invalidated-responses.md} (87%) diff --git a/docs/decisions/0041-handle-provider-invalidated-local-function-calls.md b/docs/decisions/0041-handle-provider-invalidated-responses.md similarity index 87% rename from docs/decisions/0041-handle-provider-invalidated-local-function-calls.md rename to docs/decisions/0041-handle-provider-invalidated-responses.md index dc0eb3c42cb..fdefd5a6762 100644 --- a/docs/decisions/0041-handle-provider-invalidated-local-function-calls.md +++ b/docs/decisions/0041-handle-provider-invalidated-responses.md @@ -5,13 +5,13 @@ date: 2026-09-15 deciders: eavanvalkenburg --- -# Execute usable function calls unless the provider invalidates them +# Execute usable function calls unless the provider invalidates the response ## Context and Problem Statement Chat providers do not use terminal response reasons uniformly. A response can stop because of an output limit while still containing complete local function arguments that are valid for execution and accepted by the service on -continuation. Other provider protocols explicitly invalidate local function calls when a tool block is truncated, a +continuation. Other provider protocols explicitly invalidate partial response output when a tool block is truncated, a refusal replaces partial output, a stream fails, or required terminal events are absent. The function-calling loop needs a default that preserves usable calls without executing calls that a provider has @@ -52,10 +52,10 @@ Chosen option: "Execute usable calls by default and let providers signal invalid usability is the portable default while invalidation is provider-specific protocol knowledge. `FunctionInvocationLayer` remains independent of `finish_reason`: a newly completed actionable call proceeds when -argument preparation and schema validation succeed. A provider that knows its local calls were cancelled raises -`FunctionCallInvalidatedException`. The layer abandons that current iteration, clears request budget state, restores -the last valid continuation, avoids successful response persistence and local function side effects, and re-raises the -same exception. +argument preparation and schema validation succeed. A provider that knows partial response output was invalidated +raises `ResponseInvalidatedException`; any local function calls from that response must not execute. The layer +abandons that current iteration, clears request budget state, restores the last valid continuation, avoids successful +response persistence and local function side effects, and re-raises the same exception. Anthropic applies the signal only to local actionable `tool_use` blocks. A valid stream has closed local blocks, a terminal `stop_reason` of `tool_use`, and `message_stop`. Non-tool terminal reasons, an open block at `message_stop`, diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 2e6fa4328bc..be0c4d96ae2 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -353,9 +353,10 @@ that manually replay messages own the equivalent rule: do not resend an approval - A newly completed, actionable local `function_call` is eligible for execution when argument preparation and schema validation succeed. The enclosing response `finish_reason` is not a generic authorization gate: providers may report another reason, including `length`, while still returning a complete and service-usable call. -- A provider adapter that knows its protocol invalidated or cancelled local function calls already exposed in response - content raises `FunctionCallInvalidatedException`. The function invocation layer then abandons only that current - model/tool iteration, clears its request budget state, restores the last valid service continuation, and re-raises +- A provider adapter that knows its protocol invalidated or cancelled partial response output raises + `ResponseInvalidatedException`; any local function calls already exposed from that response must not execute. The + function invocation layer then abandons only that current model/tool iteration, clears its request budget state, + restores the last valid service continuation, and re-raises the same exception. It emits no new approval request or function result, invokes no function middleware or tool body, makes no second model request or dangling-call settlement request, and does not success-persist that service response. Approval decisions resolved before the invalidated provider call retain their existing semantics. @@ -633,7 +634,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Middleware failure batch cancellation | A fatal signal fails the whole parallel batch: in-flight sibling tool invocations are cancelled and awaited before the failure propagates. Cancellation is cooperative — an async sibling stops at its next suspension point; a synchronous tool body already executing in a worker thread cannot be interrupted and may complete its side effects, but its result is discarded and never reaches the transcript, the model, or history, and failure propagation is not delayed behind it. | `TestMiddlewareFailure::test_failure_cancels_concurrent_sibling_tool`, `test_failure_with_sync_sibling_discards_late_result` | | Middleware failure on a service-managed conversation | The continuation state is already persisted when the batch fails, so before propagating, the loop settles the hosted thread: one error `function_result` per dangling call, sent with `tool_choice="none"` in one extra request; the persisted continuation advances to the settlement response (required for response-ID continuations, a no-op for conversation-object ids) and the settlement response is otherwise discarded; a settlement failure never masks the abort. Without a service-managed conversation no extra request is made. | `TestMiddlewareFailure::test_failure_settles_dangling_calls_on_service_conversation`, `test_failure_settles_service_conversation_streaming`, `test_failure_settlement_advances_response_id_continuation`, `test_failure_without_service_conversation_makes_no_settlement_request` | | Middleware failure during approved-tool replay | A fatal abort while the approval-resolution phase replays an approved tool escapes loudly (never absorbed into a rejection result), the tool's original — already service-persisted — call is settled the same way, and the continuation advances; both response modes. | `TestMiddlewareFailure::test_failure_during_approved_replay_settles_and_escapes`, `test_failure_during_approved_replay_streaming` | -| Provider-invalidated local function calls | The provider signal propagates unchanged in both response modes after clearing request budget state and restoring the last valid continuation. Partial streamed call deltas may remain visible, but no new approval, function middleware, tool body, result, follow-up model call, or settlement occurs. Anthropic requires every local call block to close, terminal `stop_reason="tool_use"`, and `message_stop`; non-tool terminal reasons, open blocks, missing `message_stop`, and non-cancellation stream errors after local call start invalidate the call. Hosted/server-only calls and caller cancellation remain unaffected. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_function_call_short_circuits_current_iteration`, `test_invalidated_final_no_tool_call_preserves_prior_continuation`, `packages/anthropic/tests/test_anthropic_client.py::test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises`, `test_valid_streaming_local_tool_call_executes_and_continues`, `test_streaming_local_tool_call_invalid_terminal_sequences_raise`, `test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation`, `test_streaming_cancellation_after_local_call_is_not_wrapped`, `test_streaming_hosted_tool_pause_turn_is_not_invalidated` | +| Provider-invalidated partial response output | `ResponseInvalidatedException` propagates unchanged in both response modes after clearing request budget state and restoring the last valid continuation. Partial streamed call deltas may remain visible, but no new approval, function middleware, tool body, result, follow-up model call, or settlement occurs. Anthropic requires every local call block to close, terminal `stop_reason="tool_use"`, and `message_stop`; non-tool terminal reasons, open blocks, missing `message_stop`, and non-cancellation stream errors after local call start invalidate the call. Hosted/server-only calls and caller cancellation remain unaffected. | `packages/core/tests/core/test_function_invocation_logic.py::test_response_invalidation_short_circuits_current_iteration`, `test_invalidated_final_no_tool_response_preserves_prior_continuation`, `packages/anthropic/tests/test_anthropic_client.py::test_non_streaming_local_tool_call_with_invalidating_stop_reason_raises`, `test_valid_streaming_local_tool_call_executes_and_continues`, `test_streaming_local_tool_call_invalid_terminal_sequences_raise`, `test_streaming_provider_error_after_local_call_is_wrapped_as_invalidation`, `test_streaming_cancellation_after_local_call_is_not_wrapped`, `test_streaming_hosted_tool_pause_turn_is_not_invalidated` | | Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents | | Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent | | Provider tool content after an active limit | Locally actionable calls and local approval requests returned despite `tool_choice="none"` are removed in both response modes. Provider-executed informational call/result pairs, hosted approval requests, and metadata-only streaming updates remain visible; fallback text never replaces retained transcript content. | `test_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped`, `test_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_function_invocation_limit_preserves_hosted_approval_request`, `test_streaming_function_invocation_limit_preserves_hosted_approval_request` | @@ -648,7 +649,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Pending hosted history replay | Stateless hosted approval requests remain replayable until a response is recorded, then both controls become inert. | `packages/openai/tests/openai/test_openai_chat_client.py::test_stateless_history_preserves_pending_hosted_approval_request_until_response` | | Non-history provider plus session | Local history is still auto-injected for approval resume. | `packages/core/tests/core/test_agents.py::test_non_history_context_provider_still_injects_inmemory` | | Hosted per-service-call persistence | A host-managed transcript remains available throughout a local function-call loop without being persisted into the framework session and replayed on the next hosted request. | `packages/foundry_hosting/tests/test_responses.py::TestAgentSessionPersistence::test_per_service_call_persistence_preserves_function_loop_history` | -| Invalidated per-service-call persistence | An invalidated service response, its partial calls, and any nonexistent results are not success-persisted, and invalid content does not advance session continuation state. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_service_call_is_not_persisted` | +| Invalidated per-service-call persistence | An invalidated service response, its partial calls, and any nonexistent results are not success-persisted, and invalid content does not advance session continuation state. | `packages/core/tests/core/test_function_invocation_logic.py::test_invalidated_response_is_not_persisted` | | Streaming message injection with per-service-call persistence | A streaming response rebuilt from updates mirrors the inner conversation id exactly, including clearing it, and keeps its internal marker, so the next iteration appends only the latest message rather than replaying the whole turn on top of provider-held history, and never persists a conversation id from an earlier injected service call. | `packages/core/tests/core/test_middleware_with_chat.py::TestChatMiddleware::test_message_injection_middleware_streaming_preserves_inner_continuation_state`, `test_message_injection_middleware_streaming_keeps_service_conversation_id_external`, `test_message_injection_middleware_streaming_clears_conversation_id_when_final_call_has_none`, `test_message_injection_middleware_conversation_id_matches_across_streaming_modes`, `packages/core/tests/core/test_harness_agent.py::test_streaming_harness_tool_call_does_not_duplicate_transcript` | | Service-side approval decision | Stored hosted request is skipped; the current approved or rejected hosted response is sent, while local approval controls are omitted from provider input. | `packages/openai/tests/openai/test_openai_chat_client.py::test_prepare_messages_strips_approval_request_but_keeps_response_under_storage`, `test_prepare_messages_drops_local_approval_controls` | | OpenAI approval serialization | Hosted approval id and decision serialize to `mcp_approval_response`; local approvals remain in-process. | `test_prepare_message_for_openai_with_function_approval_response`, `test_prepare_content_for_opentool_approval_response`, `test_function_approval_response_with_mcp_tool_call` | diff --git a/python/CODING_STANDARD.md b/python/CODING_STANDARD.md index 71ca6225df8..145fef81fca 100644 --- a/python/CODING_STANDARD.md +++ b/python/CODING_STANDARD.md @@ -240,7 +240,7 @@ AgentFrameworkException # Base for all AF exceptions │ └── AgentContentFilterException # Agent content filter triggered │ ├── ChatClientException # Chat client lifecycle and communication failures -│ ├── FunctionCallInvalidatedException # Provider invalidated emitted local function calls +│ ├── ResponseInvalidatedException # Provider invalidated partial response output │ ├── ChatClientInvalidAuthException # Chat client auth failures │ ├── ChatClientInvalidRequestException # Invalid request to chat client │ ├── ChatClientInvalidResponseException # Invalid/unexpected response from chat client @@ -285,7 +285,7 @@ AgentFrameworkException # Base for all AF exceptions | Object in wrong state (e.g., client not initialized) | `RuntimeError` | | External service returns 401/403 | `IntegrationInvalidAuthException` (or `ChatClient`/`Agent` variant) | | External service returns unexpected response | `IntegrationInvalidResponseException` (or variant) | -| Chat provider invalidates local function calls already present in response content | `FunctionCallInvalidatedException` | +| Chat provider invalidates partial response output containing local function calls | `ResponseInvalidatedException` | | Content filter blocks a request | `IntegrationContentFilterException` (or variant) | | Request validation fails before sending to service | `IntegrationInvalidRequestException` (or variant) | | Agent not found in registry | `AgentInvalidRequestException` | diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index ca8be5c9247..431276c0656 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -39,7 +39,7 @@ ChatClientException, ChatClientInvalidAuthException, ChatClientInvalidRequestException, - FunctionCallInvalidatedException, + ResponseInvalidatedException, ) from agent_framework.observability import ChatTelemetryLayer from anthropic import APIError as AnthropicAPIError @@ -625,24 +625,26 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: if local_function_call_started and ( open_local_function_call_blocks or latest_stop_reason != "tool_use" ): - raise FunctionCallInvalidatedException( - "Anthropic invalidated local function calls before completing the response." + raise ResponseInvalidatedException( + "Anthropic invalidated partial response output; " + "local function calls must not execute." ) if parsed_chunk: yield parsed_chunk if local_function_call_started and not message_stop_received: - raise FunctionCallInvalidatedException( - "Anthropic invalidated local function calls because the response ended " - "without message_stop." + raise ResponseInvalidatedException( + "Anthropic invalidated partial response output after the response ended without " + "message_stop; local function calls must not execute." ) except asyncio.CancelledError: raise - except FunctionCallInvalidatedException: + except ResponseInvalidatedException: raise except Exception as ex: if local_function_call_started: - raise FunctionCallInvalidatedException( - "Anthropic invalidated local function calls because the response stream failed.", + raise ResponseInvalidatedException( + "Anthropic invalidated partial response output after the response stream failed; " + "local function calls must not execute.", inner_exception=ex, ) from ex if isinstance(ex, AgentFrameworkException): @@ -1200,8 +1202,9 @@ def _process_message( if message.stop_reason != "tool_use" and any( content.type == "function_call" and not content.informational_only for content in contents ): - raise FunctionCallInvalidatedException( - "Anthropic invalidated local function calls before completing the response." + raise ResponseInvalidatedException( + "Anthropic invalidated partial response output; " + "local function calls must not execute." ) return ChatResponse( diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 31e80a377af..24597877291 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -16,11 +16,11 @@ ChatResponse, ChatResponseUpdate, Content, - FunctionCallInvalidatedException, FunctionInvocationLayer, FunctionTool, InlineSkill, Message, + ResponseInvalidatedException, ResponseStream, SkillFrontmatter, SkillsProvider, @@ -2158,7 +2158,7 @@ def local_tool(value: str) -> str: client = create_test_anthropic_client(mock_anthropic_client) mock_anthropic_client.beta.messages.create.return_value = _local_tool_message(stop_reason) - with pytest.raises(FunctionCallInvalidatedException, match="Anthropic invalidated"): + with pytest.raises(ResponseInvalidatedException, match="Anthropic invalidated"): await client.get_response( [Message(role="user", contents=["run"])], options={"tools": [local_tool], "max_tokens": 64}, @@ -2287,7 +2287,7 @@ async def events() -> Any: assert isinstance(stream, ResponseStream) updates: list[ChatResponseUpdate] = [] - with pytest.raises(FunctionCallInvalidatedException, match="Anthropic invalidated"): + with pytest.raises(ResponseInvalidatedException, match="Anthropic invalidated"): async for update in stream: updates.append(update) @@ -2314,7 +2314,7 @@ async def events() -> Any: ) assert isinstance(stream, ResponseStream) - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: async for _ in stream: pass diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index c7c05b01a79..ea9e4577a22 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -32,8 +32,8 @@ ) from .exceptions import ( AgentFrameworkException, - FunctionCallInvalidatedException, MiddlewareException, + ResponseInvalidatedException, UserInputRequiredException, WorkflowCheckpointException, WorkflowConvergenceException, @@ -514,7 +514,6 @@ "FinalT", "FinishReason", "FinishReasonLiteral", - "FunctionCallInvalidatedException", "FunctionExecutor", "FunctionInvocationConfiguration", "FunctionInvocationContext", @@ -567,6 +566,7 @@ "Param", "RawAgent", "ReleaseCandidateFeature", + "ResponseInvalidatedException", "ResponseStream", "Role", "RoleLiteral", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index 3800400eb74..350a458a389 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -334,8 +334,8 @@ from ._workflows._workflow_context import WorkflowContext from ._workflows._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor from .exceptions import ( AgentFrameworkException, - FunctionCallInvalidatedException, MiddlewareException, + ResponseInvalidatedException, UserInputRequiredException, WorkflowCheckpointException, WorkflowConvergenceException, @@ -469,7 +469,6 @@ __all__ = [ "FinalT", "FinishReason", "FinishReasonLiteral", - "FunctionCallInvalidatedException", "FunctionExecutor", "FunctionInvocationConfiguration", "FunctionInvocationContext", @@ -522,6 +521,7 @@ __all__ = [ "Param", "RawAgent", "ReleaseCandidateFeature", + "ResponseInvalidatedException", "ResponseStream", "Role", "RoleLiteral", diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index cfa8528abf6..1eb757cec4b 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -46,7 +46,7 @@ from pydantic import BaseModel, Field, ValidationError, create_model from ._serialization import SerializationMixin -from .exceptions import FunctionCallInvalidatedException, ToolException, UserInputRequiredException +from .exceptions import ResponseInvalidatedException, ToolException, UserInputRequiredException from .observability import ( OPERATION_DURATION_BUCKET_BOUNDARIES, OtelAttr, @@ -3378,15 +3378,15 @@ def _clear_budget_state_from_session(invocation_session: AgentSession | None) -> invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) -def _function_call_invalidation_cleanup( +def _response_invalidation_cleanup( invocation_session: AgentSession | None, budget_state: dict[str, Any], - stream_error: list[FunctionCallInvalidatedException] | None = None, -) -> Callable[[FunctionCallInvalidatedException], None]: + stream_error: list[ResponseInvalidatedException] | None = None, +) -> Callable[[ResponseInvalidatedException], None]: """Capture the last valid continuation and return invalidation cleanup.""" service_session_id = invocation_session.service_session_id if invocation_session is not None else None - def cleanup(error: FunctionCallInvalidatedException) -> None: + def cleanup(error: ResponseInvalidatedException) -> None: if stream_error is not None: stream_error[:] = [error] budget_state.clear() @@ -3403,26 +3403,26 @@ def cleanup(error: FunctionCallInvalidatedException) -> None: async def _await_provider_call( operation: Callable[..., Awaitable[_InvalidationResultT]], - on_invalidated: Callable[[FunctionCallInvalidatedException], None], + on_invalidated: Callable[[ResponseInvalidatedException], None], **kwargs: Any, ) -> _InvalidationResultT: """Await one provider operation and clean up invalidated local calls.""" try: return await operation(**kwargs) - except FunctionCallInvalidatedException as error: + except ResponseInvalidatedException as error: on_invalidated(error) raise async def _iterate_provider_stream( stream: AsyncIterable[_InvalidationResultT], - on_invalidated: Callable[[FunctionCallInvalidatedException], None], + on_invalidated: Callable[[ResponseInvalidatedException], None], ) -> AsyncIterable[_InvalidationResultT]: """Yield one provider stream and clean up invalidated local calls.""" try: async for item in stream: yield item - except FunctionCallInvalidatedException as error: + except ResponseInvalidatedException as error: on_invalidated(error) raise @@ -4014,7 +4014,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: alternate model turns and local execution until a terminal response or safety limit is reached. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state) + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state) response = cast( ChatResponse[Any], await _await_provider_call( @@ -4092,7 +4092,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state) + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state) response = cast( ChatResponse[Any], await _await_provider_call( @@ -4133,7 +4133,7 @@ async def _stream_response_with_function_invocation( budget_state: dict[str, Any], max_errors: int, middleware_pipeline: FunctionMiddlewarePipeline | None = None, - invalidation_error: list[FunctionCallInvalidatedException], + invalidation_error: list[ResponseInvalidatedException], ) -> AsyncIterable[ChatResponseUpdate]: """Run the streaming function invocation loop.""" from ._middleware import MiddlewareFailure @@ -4203,7 +4203,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non # Phase 2: stream each model turn, finalize it, execute its calls, then advance the transcript. for attempt_idx in range(attempt_start, max_iterations): budget_state["attempt_count"] = attempt_idx + 1 - on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state, invalidation_error) + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state, invalidation_error) inner_stream = cast( "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", await _await_provider_call( @@ -4354,7 +4354,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" - on_invalidated = _function_call_invalidation_cleanup(invocation_session, budget_state, invalidation_error) + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state, invalidation_error) final_inner_stream = cast( "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", await _await_provider_call( @@ -4556,7 +4556,7 @@ def get_response( ) response_format = mutable_options.get("response_format") - invalidation_error: list[FunctionCallInvalidatedException] = [] + invalidation_error: list[ResponseInvalidatedException] = [] def finalize_stream(updates: Sequence[ChatResponseUpdate]) -> ChatResponse[Any]: if invalidation_error: diff --git a/python/packages/core/agent_framework/exceptions.py b/python/packages/core/agent_framework/exceptions.py index 88bd8b27d43..03f950e331a 100644 --- a/python/packages/core/agent_framework/exceptions.py +++ b/python/packages/core/agent_framework/exceptions.py @@ -82,8 +82,8 @@ class ChatClientException(AgentFrameworkException): pass -class FunctionCallInvalidatedException(ChatClientException): - """A provider invalidated local function calls already present in response content.""" +class ResponseInvalidatedException(ChatClientException): + """A provider invalidated partial response output, so its local function calls must not execute.""" pass diff --git a/python/packages/core/tests/core/test_exceptions.py b/python/packages/core/tests/core/test_exceptions.py index e19a6e6c243..1605268cc56 100644 --- a/python/packages/core/tests/core/test_exceptions.py +++ b/python/packages/core/tests/core/test_exceptions.py @@ -2,8 +2,8 @@ """Tests for AgentFrameworkException inner_exception handling.""" -from agent_framework import AgentFrameworkException, FunctionCallInvalidatedException -from agent_framework.exceptions import ChatClientException +from agent_framework import AgentFrameworkException, ResponseInvalidatedException +from agent_framework.exceptions import ChatClientException, ChatClientInvalidResponseException def test_exception_with_inner_exception(): @@ -28,16 +28,17 @@ def test_exception_inner_exception_none_explicit(): assert len(exc.args) == 1 -def test_function_call_invalidated_exception_is_public_and_picklable() -> None: - """The provider invalidation signal is a public chat-client exception.""" +def test_response_invalidated_exception_is_public_and_picklable() -> None: + """The partial-response invalidation signal is a public chat-client exception.""" import pickle inner = RuntimeError("provider stream failed") - exc = FunctionCallInvalidatedException("local function calls were invalidated", inner_exception=inner) + exc = ResponseInvalidatedException("partial response output was invalidated", inner_exception=inner) restored = pickle.loads(pickle.dumps(exc)) assert isinstance(exc, ChatClientException) - assert isinstance(restored, FunctionCallInvalidatedException) - assert restored.args[0] == "local function calls were invalidated" + assert not isinstance(exc, ChatClientInvalidResponseException) + assert isinstance(restored, ResponseInvalidatedException) + assert restored.args[0] == "partial response output was invalidated" assert isinstance(restored.args[1], RuntimeError) assert str(restored.args[1]) == "provider stream failed" 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 5b3806780d0..25c944592f2 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -19,8 +19,8 @@ ChatResponse, ChatResponseUpdate, Content, - FunctionCallInvalidatedException, Message, + ResponseInvalidatedException, ResponseStream, SupportsChatGetResponse, chat_middleware, @@ -1077,7 +1077,7 @@ def lookup(value: str) -> str: @pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) -async def test_invalidated_function_call_short_circuits_current_iteration( +async def test_response_invalidation_short_circuits_current_iteration( chat_client_base: SupportsChatGetResponse, streaming: bool, ) -> None: @@ -1088,7 +1088,7 @@ async def test_invalidated_function_call_short_circuits_current_iteration( middleware_calls = 0 tool_calls = 0 provider_calls = 0 - invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + invalidated = ResponseInvalidatedException("provider invalidated partial response output") class TrackingMiddleware(FunctionMiddleware): async def process( @@ -1145,7 +1145,7 @@ async def updates() -> AsyncIterable[ChatResponseUpdate]: _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, }, ) - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: async for update in response_stream: yielded.append(update) else: @@ -1157,7 +1157,7 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: raise invalidated chat_client_base._get_non_streaming_response = invalid_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: await chat_client_base.get_response( [Message(role="user", contents=["run"])], options={"tools": [guarded]}, @@ -1185,7 +1185,7 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: @pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) -async def test_invalidated_service_call_is_not_persisted( +async def test_invalidated_response_is_not_persisted( chat_client_base: SupportsChatGetResponse, streaming: bool, ) -> None: @@ -1197,7 +1197,7 @@ async def test_invalidated_service_call_is_not_persisted( session = AgentSession() seed = Message(role="assistant", contents=["existing history"]) session.state[provider.source_id] = {"messages": [seed]} - invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + invalidated = ResponseInvalidatedException("provider invalidated partial response output") tool_calls = 0 @tool(name="local_tool", approval_mode="never_require") @@ -1242,14 +1242,14 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: if streaming: stream = agent.run("run", session=session, stream=True) - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: async for _ in stream: pass - with pytest.raises(FunctionCallInvalidatedException) as final_exc_info: + with pytest.raises(ResponseInvalidatedException) as final_exc_info: await stream.get_final_response() assert final_exc_info.value is invalidated else: - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: await agent.run("run", session=session) assert exc_info.value is invalidated @@ -1259,7 +1259,7 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: @pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) -async def test_invalidated_final_no_tool_call_preserves_prior_continuation( +async def test_invalidated_final_no_tool_response_preserves_prior_continuation( chat_client_base: SupportsChatGetResponse, streaming: bool, ) -> None: @@ -1269,7 +1269,7 @@ async def test_invalidated_final_no_tool_call_preserves_prior_continuation( tool_calls = 0 provider_calls = 0 - invalidated = FunctionCallInvalidatedException("provider invalidated local function calls") + invalidated = ResponseInvalidatedException("provider invalidated partial response output") @tool(name="local_tool", approval_mode="never_require") def local_tool() -> str: @@ -1324,7 +1324,7 @@ async def updates() -> AsyncIterable[ChatResponseUpdate]: _FUNCTION_INVOCATION_BUDGET_STATE_KEY: budget_state, }, ) - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: async for _ in stream: pass else: @@ -1347,7 +1347,7 @@ async def scripted_response( raise invalidated chat_client_base._get_non_streaming_response = scripted_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] - with pytest.raises(FunctionCallInvalidatedException) as exc_info: + with pytest.raises(ResponseInvalidatedException) as exc_info: await chat_client_base.get_response( [Message(role="user", contents=["run"])], options={"tools": [local_tool]}, From 536df66bc12a77e72543b5faa67800317edeb3b8 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 16:03:32 +0200 Subject: [PATCH 4/6] Python: Raise Anthropic Core dependency floor Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/anthropic/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/anthropic/pyproject.toml b/python/packages/anthropic/pyproject.toml index 182be3e99b5..1e47c8c867c 100644 --- a/python/packages/anthropic/pyproject.toml +++ b/python/packages/anthropic/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "agent-framework-core>=1.15.0,<2", + "agent-framework-core>=1.18.0,<2", "anthropic>=0.80.0,<0.117.0", ] From 819a562d54bae8532d2307931bdf862acd3ce393 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 16:04:19 +0200 Subject: [PATCH 5/6] Apply Anthropic formatting Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../anthropic/agent_framework_anthropic/_chat_client.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py index 431276c0656..b3b0458457c 100644 --- a/python/packages/anthropic/agent_framework_anthropic/_chat_client.py +++ b/python/packages/anthropic/agent_framework_anthropic/_chat_client.py @@ -1203,8 +1203,7 @@ def _process_message( content.type == "function_call" and not content.informational_only for content in contents ): raise ResponseInvalidatedException( - "Anthropic invalidated partial response output; " - "local function calls must not execute." + "Anthropic invalidated partial response output; local function calls must not execute." ) return ChatResponse( From 83007f60a7d53707bb55a2c9dd3a4c2a55f2624c Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 15 Sep 2026 16:26:11 +0200 Subject: [PATCH 6/6] Python: Address response invalidation review Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- python/packages/anthropic/pyproject.toml | 2 +- .../packages/core/agent_framework/_tools.py | 4 +- .../core/test_function_invocation_logic.py | 61 +++++++++++++++++++ 3 files changed, 65 insertions(+), 2 deletions(-) diff --git a/python/packages/anthropic/pyproject.toml b/python/packages/anthropic/pyproject.toml index 1e47c8c867c..3045222d7ef 100644 --- a/python/packages/anthropic/pyproject.toml +++ b/python/packages/anthropic/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "agent-framework-core>=1.18.0,<2", + "agent-framework-core>=1.19.0,<2", "anthropic>=0.80.0,<0.117.0", ] diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 1eb757cec4b..64f8d0f9557 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -3384,7 +3384,9 @@ def _response_invalidation_cleanup( stream_error: list[ResponseInvalidatedException] | None = None, ) -> Callable[[ResponseInvalidatedException], None]: """Capture the last valid continuation and return invalidation cleanup.""" - service_session_id = invocation_session.service_session_id if invocation_session is not None else None + service_session_id = ( + copy.deepcopy(invocation_session.service_session_id) if invocation_session is not None else None + ) def cleanup(error: ResponseInvalidatedException) -> None: if stream_error is not None: 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 25c944592f2..f09dab17485 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -1184,6 +1184,67 @@ async def invalid_response(**kwargs: Any) -> ChatResponse: assert [content.type for update in yielded for content in update.contents] == ["function_call"] +@pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) +async def test_response_invalidation_restores_structured_continuation_snapshot( + chat_client_base: SupportsChatGetResponse, + streaming: bool, +) -> None: + """Cleanup restores a copy when a provider mutates structured continuation state in place.""" + from agent_framework._sessions import AgentSession + + original_service_session_id = {"conversation_id": "valid", "metadata": {"generation": 1}} + session = AgentSession(service_session_id=original_service_session_id) + invalidated = ResponseInvalidatedException("provider invalidated partial response output") + + def mutate_continuation() -> None: + service_session_id = session.service_session_id + assert isinstance(service_session_id, dict) + service_session_id["conversation_id"] = "invalid" + metadata = service_session_id["metadata"] + assert isinstance(metadata, dict) + metadata["generation"] = 2 + + if streaming: + + def invalid_stream(**kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + del kwargs + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + mutate_continuation() + raise invalidated + yield # pragma: no cover + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + chat_client_base._get_streaming_response = invalid_stream # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + stream = chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": []}, + stream=True, + client_kwargs={"session": session}, + ) + with pytest.raises(ResponseInvalidatedException): + async for _ in stream: + pass + else: + + async def invalid_response(**kwargs: Any) -> ChatResponse: + del kwargs + mutate_continuation() + raise invalidated + + chat_client_base._get_non_streaming_response = invalid_response # type: ignore[attr-defined, method-assign] # ty: ignore[unresolved-attribute] + with pytest.raises(ResponseInvalidatedException): + await chat_client_base.get_response( + [Message(role="user", contents=["run"])], + options={"tools": []}, + client_kwargs={"session": session}, + ) + + assert session.service_session_id == {"conversation_id": "valid", "metadata": {"generation": 1}} + assert session.service_session_id is not original_service_session_id + + @pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) async def test_invalidated_response_is_not_persisted( chat_client_base: SupportsChatGetResponse,