diff --git a/docs/decisions/0041-handle-provider-invalidated-responses.md b/docs/decisions/0041-handle-provider-invalidated-responses.md new file mode 100644 index 00000000000..fdefd5a6762 --- /dev/null +++ b/docs/decisions/0041-handle-provider-invalidated-responses.md @@ -0,0 +1,71 @@ +--- +status: proposed +contact: eavanvalkenburg +date: 2026-09-15 +deciders: eavanvalkenburg +--- + +# 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 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 +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 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`, +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..be0c4d96ae2 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -350,6 +350,19 @@ 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 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. + 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 +524,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 +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 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` | @@ -634,6 +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_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 6756c860f96..145fef81fca 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 +│ ├── ResponseInvalidatedException # Provider invalidated partial response output │ ├── 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 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 aeb4822411e..b3b0458457c 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, + ResponseInvalidatedException, ) from agent_framework.observability import ChatTelemetryLayer from anthropic import APIError as AnthropicAPIError @@ -599,15 +601,54 @@ 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 ResponseInvalidatedException( + "Anthropic invalidated partial response output; " + "local function calls must not execute." + ) if parsed_chunk: yield parsed_chunk - except AgentFrameworkException: + if local_function_call_started and not message_stop_received: + 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 ResponseInvalidatedException: raise except Exception as ex: + if local_function_call_started: + 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): + raise raise _wrap_anthropic_error(ex) from ex return self._build_response_stream(_stream(), response_format=options.get("response_format")) @@ -1157,12 +1198,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 ResponseInvalidatedException( + "Anthropic invalidated partial response output; local function calls must not execute." + ) + 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/pyproject.toml b/python/packages/anthropic/pyproject.toml index 182be3e99b5..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.15.0,<2", + "agent-framework-core>=1.19.0,<2", "anthropic>=0.80.0,<0.117.0", ] diff --git a/python/packages/anthropic/tests/test_anthropic_client.py b/python/packages/anthropic/tests/test_anthropic_client.py index 36a65228abd..24597877291 100644 --- a/python/packages/anthropic/tests/test_anthropic_client.py +++ b/python/packages/anthropic/tests/test_anthropic_client.py @@ -20,6 +20,7 @@ FunctionTool, InlineSkill, Message, + ResponseInvalidatedException, ResponseStream, SkillFrontmatter, SkillsProvider, @@ -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(ResponseInvalidatedException, 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(ResponseInvalidatedException, 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(ResponseInvalidatedException) 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..ea9e4577a22 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -33,6 +33,7 @@ from .exceptions import ( AgentFrameworkException, MiddlewareException, + ResponseInvalidatedException, UserInputRequiredException, WorkflowCheckpointException, WorkflowConvergenceException, @@ -565,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 b6ad94bdf11..350a458a389 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -335,6 +335,7 @@ from ._workflows._workflow_executor import SubWorkflowRequestMessage, SubWorkflo from .exceptions import ( AgentFrameworkException, MiddlewareException, + ResponseInvalidatedException, UserInputRequiredException, WorkflowCheckpointException, WorkflowConvergenceException, @@ -520,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 805c3df5849..64f8d0f9557 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 ResponseInvalidatedException, ToolException, UserInputRequiredException from .observability import ( OPERATION_DURATION_BUCKET_BOUNDARIES, OtelAttr, @@ -3378,6 +3378,57 @@ def _clear_budget_state_from_session(invocation_session: AgentSession | None) -> invocation_session.state.pop(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, None) +def _response_invalidation_cleanup( + invocation_session: AgentSession | None, + budget_state: dict[str, Any], + stream_error: list[ResponseInvalidatedException] | None = None, +) -> Callable[[ResponseInvalidatedException], None]: + """Capture the last valid continuation and return invalidation cleanup.""" + 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: + 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[[ResponseInvalidatedException], None], + **kwargs: Any, +) -> _InvalidationResultT: + """Await one provider operation and clean up invalidated local calls.""" + try: + return await operation(**kwargs) + except ResponseInvalidatedException as error: + on_invalidated(error) + raise + + +async def _iterate_provider_stream( + stream: AsyncIterable[_InvalidationResultT], + 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 ResponseInvalidatedException as error: + on_invalidated(error) + raise + + def _apply_batch_limit_decision( action: Literal["continue", "return", "stop"], options: dict[str, Any], @@ -3965,9 +4016,12 @@ 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 = _response_invalidation_cleanup(invocation_session, budget_state) response = cast( ChatResponse[Any], - await super_get_response( + await _await_provider_call( + super_get_response, + on_invalidated, messages=prepared_messages, stream=False, options=options, @@ -4040,9 +4094,12 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state) response = cast( ChatResponse[Any], - await super_get_response( + await _await_provider_call( + super_get_response, + on_invalidated, messages=prepared_messages, stream=False, options=options, @@ -4078,6 +4135,7 @@ async def _stream_response_with_function_invocation( budget_state: dict[str, Any], max_errors: int, middleware_pipeline: FunctionMiddlewarePipeline | None = None, + invalidation_error: list[ResponseInvalidatedException], ) -> AsyncIterable[ChatResponseUpdate]: """Run the streaming function invocation loop.""" from ._middleware import MiddlewareFailure @@ -4147,9 +4205,12 @@ 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 = _response_invalidation_cleanup(invocation_session, budget_state, invalidation_error) inner_stream = cast( "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( + await _await_provider_call( + super_get_response, + on_invalidated, messages=prepared_messages, stream=True, options=options, @@ -4158,13 +4219,12 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non 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: + async for update in _iterate_provider_stream(inner_stream, on_invalidated): for content in update.contents: if content.type != "function_call": continue @@ -4221,7 +4281,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non continue yield update - response = await inner_stream.get_final_response() + 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) @@ -4296,9 +4356,12 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non max_iterations, ) options["tool_choice"] = "none" + on_invalidated = _response_invalidation_cleanup(invocation_session, budget_state, invalidation_error) final_inner_stream = cast( "ResponseStream[ChatResponseUpdate, ChatResponse[Any]]", - super_get_response( + await _await_provider_call( + super_get_response, + on_invalidated, messages=prepared_messages, stream=True, options=options, @@ -4307,13 +4370,12 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non client_kwargs=request_kwargs, ), ) - await final_inner_stream - async for update in final_inner_stream: + 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 final_inner_stream.get_final_response() + 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, @@ -4496,6 +4558,12 @@ def get_response( ) response_format = mutable_options.get("response_format") + invalidation_error: list[ResponseInvalidatedException] = [] + + 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( @@ -4510,8 +4578,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/agent_framework/exceptions.py b/python/packages/core/agent_framework/exceptions.py index b96aab6b916..03f950e331a 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 ResponseInvalidatedException(ChatClientException): + """A provider invalidated partial response output, so its local function calls must not execute.""" + + 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..1605268cc56 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, ResponseInvalidatedException +from agent_framework.exceptions import ChatClientException, ChatClientInvalidResponseException def test_exception_with_inner_exception(): @@ -25,3 +26,19 @@ 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_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 = ResponseInvalidatedException("partial response output was invalidated", inner_exception=inner) + restored = pickle.loads(pickle.dumps(exc)) + + assert isinstance(exc, ChatClientException) + 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 d829aaa4c31..f09dab17485 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -20,6 +20,7 @@ ChatResponseUpdate, Content, Message, + ResponseInvalidatedException, ResponseStream, SupportsChatGetResponse, chat_middleware, @@ -1001,6 +1002,430 @@ 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_response_invalidation_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 = ResponseInvalidatedException("provider invalidated partial response output") + + 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(ResponseInvalidatedException) 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(ResponseInvalidatedException) 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_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, + 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 = ResponseInvalidatedException("provider invalidated partial response output") + 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, + ) + + if streaming: + stream = agent.run("run", session=session, stream=True) + with pytest.raises(ResponseInvalidatedException) as exc_info: + async for _ in stream: + pass + with pytest.raises(ResponseInvalidatedException) as final_exc_info: + await stream.get_final_response() + assert final_exc_info.value is invalidated + else: + with pytest.raises(ResponseInvalidatedException) as exc_info: + 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 + + +@pytest.mark.parametrize("streaming", [False, True], ids=["non_streaming", "streaming"]) +async def test_invalidated_final_no_tool_response_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 = ResponseInvalidatedException("provider invalidated partial response output") + + @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(ResponseInvalidatedException) 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(ResponseInvalidatedException) 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