Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion python/packages/core/agent_framework/_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -3029,6 +3029,7 @@ def _bind_approval_responses_to_pending_requests(
invocation_session: AgentSession | None,
*,
consume: bool = True,
staged_response_ids: set[int] | None = None,
) -> set[int]:
"""Rebind approval responses and remove unissued or duplicate responses."""
if invocation_session is None:
Expand All @@ -3049,6 +3050,8 @@ def _bind_approval_responses_to_pending_requests(
consume=consume,
)
if rebound is None:
if staged_response_ids and id(content) in staged_response_ids:
continue
logger.warning(
"Ignored an approval response with id %r because it did not match the active approval "
"occurrence identity; the pending request was retained for retry.",
Expand Down Expand Up @@ -3107,8 +3110,13 @@ def _store_already_approved_approval_requests(
def _stage_approval_batch_responses(
invocation_session: AgentSession | None,
approval_responses: Sequence[Content],
*,
staged_response_ids: set[int] | None = None,
) -> tuple[list[Content], list[dict[str, str | None]], list[Content] | None]:
"""Accumulate approval decisions and release a batch only when every decision is present."""
"""Accumulate approval decisions and release a batch only when every decision is present.

``staged_response_ids`` records matched source objects for diagnostic-only warning suppression.
"""
if not approval_responses:
return [], [], None
state = _get_tool_approval_state(invocation_session)
Expand Down Expand Up @@ -3154,6 +3162,8 @@ def _stage_approval_batch_responses(
continue
if request_id not in stored_responses:
stored_responses[request_id] = rebound
if staged_response_ids is not None:
staged_response_ids.add(id(response))
matched = True

if not matched:
Expand Down Expand Up @@ -4593,9 +4603,11 @@ async def _resolve_approval_responses(
pending_responses_before_binding = list(
_collect_approval_responses(prepared_messages, non_approval_result_ids=host_result_ids).values()
)
staged_response_ids: set[int] = set()
staged_responses, function_call_order, waiting_requests = _stage_approval_batch_responses(
approval_session,
pending_responses_before_binding,
staged_response_ids=staged_response_ids,
)
if waiting_requests is not None:
response_messages, streaming_updates = _messages_and_updates_for_terminal_contents(waiting_requests)
Expand All @@ -4610,6 +4622,7 @@ async def _resolve_approval_responses(
prepared_messages,
approval_session,
consume=False,
staged_response_ids=staged_response_ids,
)
active_pending_ids = (
set(_load_pending_approval_requests(approval_session))
Expand Down
82 changes: 65 additions & 17 deletions python/packages/core/tests/core/test_harness_tool_approval.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import json
import warnings
from collections.abc import Awaitable, Callable, MutableSequence
from contextlib import nullcontext
from datetime import timedelta
from enum import Enum
from pathlib import Path
Expand Down Expand Up @@ -1119,12 +1120,17 @@ async def capture_response(
assert "function_approval_response" not in replayed_types


@pytest.mark.parametrize("legacy_request_id", [False, True], ids=["occurrence-id", "legacy-id"])
@pytest.mark.parametrize("conflicting_duplicate", [False, True], ids=["single-response", "conflicting-duplicate"])
@pytest.mark.parametrize("approved", [True, False], ids=["approved", "rejected"])
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
async def test_approval_resume_returns_result_without_mutating_inputs(
chat_client_base: MockBaseChatClient,
caplog: pytest.LogCaptureFixture,
approved: bool,
streaming: bool,
legacy_request_id: bool,
conflicting_duplicate: bool,
) -> None:
"""Approval resume should return its terminal result without changing caller-owned messages."""
calls = 0
Expand All @@ -1150,23 +1156,49 @@ def guarded_tool() -> str:
first_response = await agent.run("run guarded", session=session)
approval_request = first_response.user_input_requests[0]

approval_response = approval_request.to_function_approval_response(approved=approved)
approval_message = Message(role="user", contents=[approval_response])
if legacy_request_id:
from agent_framework._tools import _store_already_approved_approval_requests, _store_pending_approval_requests

if streaming:
chat_client_base.streaming_responses = [
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])]
]
second_stream = agent.run(approval_message, stream=True, session=session)
second_updates = [update async for update in second_stream]
second_response = await second_stream.get_final_response()
assert [[content.type for content in update.contents] for update in second_updates] == [
["function_result"],
["text"],
]
else:
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
second_response = await agent.run(approval_message, session=session)
approval_request = Content.from_dict(approval_request.to_dict())
approval_request.id = "call_guarded"
assert approval_request.id != _function_call(approval_request).id
_store_pending_approval_requests(session, [approval_request])
_store_already_approved_approval_requests(session, [approval_request], [])

approval_response = approval_request.to_function_approval_response(approved=approved)
approval_contents = [approval_response]
if conflicting_duplicate:
conflicting_call = Content.from_dict(_function_call(approval_request).to_dict())
conflicting_call.id = "af-call-conflicting-duplicate"
assert approval_response.id is not None
approval_contents.append(
Content.from_function_approval_response(
approved=not approved,
id=approval_response.id,
function_call=conflicting_call,
)
)
approval_message = Message(role="user", contents=approval_contents.copy())

with (
pytest.warns(FutureWarning, match="legacy provider call_id request binding")
if legacy_request_id
else nullcontext[None]()
):
if streaming:
chat_client_base.streaming_responses = [
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])]
]
second_stream = agent.run(approval_message, stream=True, session=session)
second_updates = [update async for update in second_stream]
second_response = await second_stream.get_final_response()
assert [[content.type for content in update.contents] for update in second_updates] == [
["function_result"],
["text"],
]
else:
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
second_response = await agent.run(approval_message, session=session)

assert [[content.type for content in message.contents] for message in second_response.messages] == [
["function_result"],
Expand All @@ -1176,12 +1208,28 @@ def guarded_tool() -> str:
assert result.call_id == "call_guarded"
assert result.result == ("approved result" if approved else "Error: Tool call invocation was rejected by user.")
assert calls == int(approved)
if conflicting_duplicate:
assert len(caplog.records) == 1
assert "occurrence identity" in caplog.text
else:
assert caplog.records == []
assert approval_message.role == "user"
assert approval_message.contents == [approval_response]
assert approval_message.contents == approval_contents
assert [[content.type for content in message.contents] for message in first_response.messages] == [
["function_call", "function_approval_request"]
]

caplog.clear()
replay_response = approval_request.to_function_approval_response(approved=True)
if streaming:
replay_stream = agent.run(replay_response, stream=True, session=session)
_ = [update async for update in replay_stream]
await replay_stream.get_final_response()
else:
await agent.run(replay_response, session=session)
assert calls == int(approved)
assert "occurrence identity" in caplog.text


@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
async def test_approval_resume_replays_reasoning_with_function_call_group(
Expand Down
Loading