diff --git a/.github/skills/pull-requests/SKILL.md b/.github/skills/pull-requests/SKILL.md index b8bc31472e..cd724366bf 100644 --- a/.github/skills/pull-requests/SKILL.md +++ b/.github/skills/pull-requests/SKILL.md @@ -52,6 +52,16 @@ Check every item that applies. For the breaking-change item: the checklist already cover validation status. - Do **not** remove or reorder the template's headings. +### Stable specifications + +For Python function-calling loop changes, read +[`docs/specs/004-python-function-calling-loop.md`](../../../docs/specs/004-python-function-calling-loop.md) +and validate the PR against it. Do not edit that specification by default. It +is a stable cross-package contract, not a per-PR changelog. A specification +change is warranted only when the PR intentionally changes normative behavior, +the scenario inventory, an acknowledged coverage gap, or the authoritative +scenario-to-test mapping; keep any such edit to the smallest affected sections. + ### Creating the PR Open new PRs as **drafts** until they are ready for review. Example: diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 85e9e0c657..d7d4605818 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -25,6 +25,17 @@ The primary implementation is in `python/packages/core/agent_framework/_tools.py `python/packages/core/agent_framework/_sessions.py`, provider serializers, hosting packages, and UI transports are part of the same contract when they carry function-call loop content. +## Maintenance policy + +This document is the stable cross-package contract for the function-calling loop, not a per-PR changelog. Every change +in scope must be reviewed against it, but most such PRs should not edit it. Update only the smallest affected sections +when a change intentionally alters normative behavior, the scenario inventory, an acknowledged coverage gap, or the +authoritative scenario-to-test mapping. + +Do not add implementation narration, temporary debugging guidance, or tests that merely preserve an already documented +contract. Temporary compatibility behavior belongs here only when it is itself part of the user-visible contract; its +later removal requires another deliberate contract update. + ## Change sensitivity This code is high risk. Small changes can produce duplicate side effects, orphaned calls or results, invalid @@ -228,8 +239,29 @@ the tool-call group, and provider adapters serialize or reconstruct the provider ### Approval correlation, replay, and reused ids -`call_id` is not globally unique forever. The normalizer therefore tracks open logical occurrences in transcript -order instead of keeping one global result per id. +`call_id` is the provider/service correlation identifier and is not globally unique forever. `Content.id` on a +`function_call` is the Agent Framework identity for one locally actionable occurrence. New actionable calls receive +that occurrence id once and preserve it through streaming aggregation, serialization, and replay; existing ids are +never regenerated. Deserializing a legacy stored call without `Content.id` does not manufacture a new identity. + +New local approval requests use the function call occurrence id as their request id. Provider-issued hosted approval +request ids remain unchanged and continue to follow the hosted service protocol. A response carrying the occurrence +id can bind without embedding a function call because the trusted pending snapshot is authoritative. For an +occurrence-aware local snapshot, a missing or mismatched occurrence identity fails closed without consuming the +pending request; matching a nested provider `call_id` is not a compatibility alias. + +Legacy stored local pending snapshots whose function call lacks `Content.id` retain their exact request-id binding for +one resume. Taking that compatibility path emits a migration warning and consumes the matching request once. The +warning marks the staged path for removal after stored legacy approvals have drained. An empty provider `call_id` may +fall back to the generated occurrence id only when the framework is about to correlate a local actionable call; this +also warns so provider adapters can supply a real service id. Deserialization itself never warns or rewrites either id. + +Pending approval state is a trusted session-state boundary: hosts must authorize and tenant-scope the session store and +must prevent untrusted callers from replacing snapshots. Consume-on-bind prevents replay within one authoritative +session state, but it is not a durable exactly-once guarantee across crashes or concurrent workers without external +transactional coordination. + +The normalizer tracks open logical occurrences in transcript order instead of keeping one global result per id. ```mermaid flowchart TD @@ -338,7 +370,11 @@ that manually replay messages own the equivalent rule: do not resend an approval discarded either way and never reaches the transcript, the model, or history. Middleware must not catch `MiddlewareFailure` — swallowing it converts a fail-closed abort back into a running, possibly unguarded loop. - Parallel calls retain model order in the returned transcript. +- `call_id` remains the provider/service correlation id; a locally actionable `function_call` also carries a stable + Agent Framework occurrence identity in `Content.id`. - Reused `call_id` values are correlated by logical occurrence, not one global value per id. +- Existing `Content.id` values survive aggregation and replay and are never regenerated; legacy deserialization does + not invent one. - A completed function call/result pair is inert on later turns. - Informational-only and declaration-only calls are not executed as local tools. @@ -361,14 +397,22 @@ that manually replay messages own the equivalent rule: do not resend an approval - A tool that requires approval does not execute before an approved response. - With an `AgentSession`, every surfaced local or hosted approval request is stored as an immutable snapshot in one active model batch. A new surfaced batch replaces an abandoned batch instead of accumulating session state. -- Approval request IDs use the provider function `call_id`, whose conversation-level uniqueness is required for - function-call/result correlation. Duplicate request IDs within one batch are rejected as malformed. -- An inbound response is honored only when its request id matches the pending server-held snapshot. +- New local approval request IDs use the recorded `function_call.id` occurrence identity. Provider-issued hosted + approval request IDs remain unchanged. Duplicate request IDs within one batch are rejected as malformed. +- An inbound response is honored only when its occurrence identity matches the pending server-held snapshot. A new + local response may omit its embedded function call because the snapshot is authoritative; a mismatched embedded + occurrence identity fails closed. +- Legacy stored local snapshots without `function_call.id` retain exact request-id matching for one consume-on-bind + resume and emit a migration warning. Deserialization does not rewrite the snapshot or emit that warning. - Approval requests replayed in inbound message history do not create, replace, or resurrect approval authority. - The executable call id, tool name, arguments, and local or hosted tool metadata are sourced from the recorded request, never from the response payload. - A matched approval response consumes its pending entry once. Unmatched, duplicate, and replayed responses do not reach local execution. +- Unmatched occurrence-aware responses leave the pending request intact for a corrected retry and produce an + observable warning/log. A nested `call_id` is never accepted as an occurrence-identity alias. +- Session-backed pending snapshots are trusted host state and require tenant-scoped, authorized storage. Consume-on-bind + does not claim durable exactly-once behavior across crashes or concurrent workers. - Tool lookup uses the recorded name against the current registry. A same-name implementation upgrade is allowed; removing the name prevents local execution. - Only the strict boolean `True` grants approval. Missing decisions and non-boolean values are rejection, not consent. @@ -434,6 +478,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | String input | Flexible string input follows the same loop behavior. | `test_base_client_with_function_calling_string_input` | | Multiple sequential rounds | Each round retains one call/result pair. | `test_base_client_with_function_calling_resets` | | Streaming call | Call chunks, one result update, and final text are emitted in order. | `test_base_client_with_streaming_function_calling` | +| Function-call occurrence identity | Actionable calls gain one stable `Content.id`; a safe local empty-`call_id` fallback uses that id with a migration warning, and streaming aggregation preserves provider-assigned occurrence ids across interleaved fragments. OpenAI Chat Completions scopes fragment correlation to each request and `(choice.index, tool.index)`. | `test_actionable_function_call_gets_stable_occurrence_identity`, `test_actionable_function_call_uses_occurrence_identity_for_empty_call_id`, `test_streaming_empty_call_id_keeps_occurrence_identity_through_approval`, `test_streaming_empty_call_id_delta_reuses_opening_call_identity`, `test_streaming_interleaved_indexed_call_fragments_coalesce_by_occurrence`, `packages/core/tests/core/test_types.py::test_function_call_occurrence_id_roundtrips_without_regeneration`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_streaming_tool_call_identity_is_request_local_and_scoped_by_choice_index` | | Reasoning-bound call | Finalized output retains reasoning, function call, function result, and final text. | `test_streaming_function_calling_response_includes_reasoning_and_tool_results` | | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` | | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` | @@ -460,6 +505,9 @@ that manually replay messages own the equivalent rule: do not resend an approval | Approval re-entry after iteration budget | Pending approved calls resolve once even when prior model calls consumed `max_iterations`. | `packages/core/tests/core/test_harness_tool_approval.py::test_auto_approval_resolves_after_iteration_budget_is_exhausted` | | Approval resume with reasoning | Model-bound resume history retains reasoning before the call and terminal result in both modes. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_replays_reasoning_with_function_call_group` | | Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, or arguments. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates` | +| Occurrence-aware local binding | New local requests use `function_call.id`; missing, mismatched, or stale occurrence ids do not execute or consume pending state, while the canonical occurrence id binds without an embedded call. | `test_occurrence_aware_approval_rejects_stale_reused_call_id_response`, `test_occurrence_aware_approval_mismatched_identity_does_not_consume_pending`, `test_occurrence_aware_approval_binds_without_embedded_function_call` | +| Legacy stored approval | A serialized pending request without `function_call.id` retains exact request-id binding once and warns only when resumed. | `test_legacy_serialized_pending_approval_resumes_once_with_migration_warning`, `packages/core/tests/core/test_types.py::test_legacy_function_call_deserialization_does_not_generate_an_occurrence_id` | +| Hosted approval identity | Provider-issued hosted approval request ids are unchanged by local occurrence correlation. | `test_hosted_approval_keeps_provider_issued_request_id` | | Truthy non-boolean decision | Strings, integers, null, and other non-booleans do not authorize execution. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_treats_truthy_non_boolean_as_rejection`, `packages/core/tests/core/test_types.py::test_function_approval_response_deserialization_rejects_non_boolean_decisions`, `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_function_approval_requires_real_boolean`, `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_resolve_approval_responses_treats_non_boolean_decision_as_rejection` | | Active batch replacement | A newly surfaced model batch replaces abandoned approval authority instead of growing session state. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_replaces_abandoned_batch` | | Duplicate request id | Ambiguous request IDs within one active batch fail explicitly. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_batch_rejects_duplicate_request_ids` | diff --git a/python/AGENTS.md b/python/AGENTS.md index 8e1f0819b2..a55bec29f9 100644 --- a/python/AGENTS.md +++ b/python/AGENTS.md @@ -60,12 +60,14 @@ Run `uv run poe` from the `python/` directory to see available commands. See [DE ## Function-Calling Loop Changes Changes to the Python function-calling loop, approval resume behavior, function-call history, provider -serialization, or transport result handling must follow -[the function-calling loop specification](../docs/specs/004-python-function-calling-loop.md). This area requires -extra validation because small changes can duplicate side effects, orphan call/result pairs, replay stale approval -authority, or make streaming and non-streaming behavior diverge. Update the specification and its scenario-to-test -mapping only when the documented contract, scenario inventory, or authoritative scenario-to-test mapping materially -changes. Adding or modifying tests that preserve existing documented behavior does not require a specification update. +serialization, or transport result handling must be reviewed against +[the function-calling loop specification](../docs/specs/004-python-function-calling-loop.md). Reading and validating +against the specification is required; editing it is not the default. It is a stable contract, not a per-PR changelog. +Update only the smallest affected sections when normative behavior, the scenario inventory, an acknowledged coverage +gap, or the authoritative scenario-to-test mapping intentionally changes. Do not update it for implementation details +or tests that preserve existing documented behavior. This area requires extra validation because small changes can +duplicate side effects, orphan call/result pairs, replay stale approval authority, or make streaming and non-streaming +behavior diverge. External contributors must check with the Agent Framework core team before picking up issues in this area. ## Project Structure diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index 6d4b86b40f..ac157998dc 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -40,6 +40,10 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. - `_approval_lifecycle.py` is the sole owner of approval occurrence registration, trusted aliases, authority validation, claims, terminal outcomes, and retry deduplication. Runner code normalizes AG-UI protocol values and projects lifecycle outcomes but must not maintain a parallel pending-approval registry. +- Local tool-approval interrupt ids use the Agent Framework `function_call.id` occurrence identity; `toolCallId` + remains the provider/service `call_id`. Hosted approvals preserve the provider-issued approval request id. The + lifecycle stores these identities separately so resume responses carry the authoritative approval occurrence id + without rewriting tool-result correlation ids. - Default stateless conversation history is client-controlled, including historical tool calls and results. Never document conversational tool results as authorization or policy evidence; use deterministic server-side checks, server-validated approvals, or scoped authoritative snapshots. diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py index 1a1be43956..f577cab705 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py @@ -102,6 +102,8 @@ ) from ._snapshot_session import ThreadSnapshotSession, _event_messages_to_snapshot_dicts from ._utils import ( + _approval_interrupt_id, + _function_call_server_label, canonical_function_arguments, convert_agui_tools_to_agent_framework, generate_event_id, @@ -715,13 +717,6 @@ def _pending_approval_server_label(entry: ApprovalOccurrence) -> str | None: return entry.server_label -def _function_call_server_label(function_call: Content | None) -> str | None: - if function_call is None: - return None - server_label = function_call.additional_properties.get("server_label") - return server_label if isinstance(server_label, str) and server_label else None - - def _function_call_execution_owner( function_call: Content, tools: list[Any] | None, @@ -1434,7 +1429,6 @@ def _canonical_approval_resume_messages( call_id=sibling_call_id, name=function_call.name, arguments=sibling_arguments, - aliases=[sibling_call_id], response_id=str(response_id), server_label=_function_call_server_label(function_call), ) @@ -1505,6 +1499,7 @@ async def _resolve_approval_responses( tools: list[Any], agent: SupportsAgentRun, run_kwargs: dict[str, Any], + invocation_session: AgentSession, thread_id: str = "", validated_approved_responses: list[Content] | None = None, *, @@ -1524,6 +1519,7 @@ async def _resolve_approval_responses( tools: List of available tools agent: The agent instance (to get client and config) run_kwargs: Kwargs for tool execution + invocation_session: Restored session for middleware and tool execution. thread_id: The conversation thread ID used to scope registry keys. validated_approved_responses: Optional collector for validated local approval responses, including controls removed because the matching @@ -1762,6 +1758,7 @@ async def execute_local_call(approval: Content = approval, call_id: str = call_i tools=tools, middleware_pipeline=middleware_pipeline, config=config, + invocation_session=invocation_session, ) except Exception as exc: logger.exception("Failed to execute approved tool call; injecting error result: %s", exc) @@ -2293,6 +2290,121 @@ def _split_service_session_input( return snapshot_messages[len(stored_snapshot_messages) :], snapshot_messages +def _legacy_tool_message_approval_resume( + messages: Sequence[Mapping[str, Any]], + *, + lifecycle: ApprovalLifecycle, + thread_id: str, + submitted_messages: Sequence[Mapping[str, Any]] | None = None, +) -> tuple[list[dict[str, Any]], set[int], str | None] | None: + """Translate unambiguous legacy tool-result approvals to canonical resume entries.""" + entries: list[dict[str, Any]] = [] + translated_message_ids: set[int] = set() + error: str | None = None + seen_interrupt_ids: set[str] = set() + retained_occurrences_by_call_id: dict[str, list[ApprovalOccurrence]] = {} + pending_local_occurrences_by_call_id: dict[str, list[ApprovalOccurrence]] = {} + for occurrence in lifecycle.occurrences_for_thread(thread_id=thread_id): + retained_occurrences_by_call_id.setdefault(occurrence.identity.call_id, []).append(occurrence) + if occurrence.status is ApprovalStatus.PENDING and occurrence.server_label is None: + pending_local_occurrences_by_call_id.setdefault(occurrence.identity.call_id, []).append(occurrence) + + submitted = messages if submitted_messages is None else submitted_messages + trailing_tool_message_ids: set[int] = set() + for message in reversed(submitted): + if str(message.get("role", "")).lower() != "tool": + break + trailing_tool_message_ids.add(id(message)) + + latest_function_call_by_id: dict[str, tuple[str, str] | None] = {} + for message in messages: + if str(message.get("role", "")).lower() != "assistant": + continue + raw_tool_calls = message.get("tool_calls") or message.get("toolCalls") + if not isinstance(raw_tool_calls, list): + continue + for raw_tool_call in raw_tool_calls: + if not isinstance(raw_tool_call, Mapping): + continue + function = raw_tool_call.get("function") + if isinstance(function, Mapping) and function.get("name") == "confirm_changes" and raw_tool_call.get("id"): + latest_function_call_by_id[str(raw_tool_call["id"])] = None + continue + if not isinstance(function, Mapping) or not raw_tool_call.get("id") or not function.get("name"): + continue + call_id = str(raw_tool_call["id"]) + parsed_call = Content.from_function_call( + call_id=call_id, + name=str(function["name"]), + arguments=function.get("arguments"), + ) + latest_function_call_by_id[call_id] = ( + str(function["name"]), + canonical_function_arguments(parsed_call) or "{}", + ) + + for message in submitted: + if id(message) not in trailing_tool_message_ids: + continue + call_id = message.get("tool_call_id") or message.get("toolCallId") or message.get("actionExecutionId") + if not call_id: + continue + raw_content = message.get("content") + if raw_content is None: + raw_content = message.get("result") + if isinstance(raw_content, str): + try: + payload = json.loads(raw_content) + except json.JSONDecodeError: + continue + else: + payload = raw_content + if not isinstance(payload, Mapping) or "accepted" not in payload: + continue + call_id_string = str(call_id) + if call_id_string in latest_function_call_by_id and latest_function_call_by_id[call_id_string] is None: + continue + matching_occurrences = retained_occurrences_by_call_id.get(call_id_string, []) + pending_occurrences = pending_local_occurrences_by_call_id.get(call_id_string, []) + if not pending_occurrences: + continue + translated_message_ids.add(id(message)) + submitted_call = latest_function_call_by_id.get(call_id_string) + if submitted_call != ( + pending_occurrences[0].name, + pending_occurrences[0].arguments, + ): + error = ( + f"Legacy AG-UI tool-message approval call_id '{call_id}' does not match the pending tool operation. " + "Retry with the canonical approval interrupt id." + ) + continue + if len(matching_occurrences) != 1 or len(pending_occurrences) != 1: + error = ( + f"Legacy AG-UI tool-message approval call_id '{call_id}' does not identify exactly one retained " + "pending local occurrence. Retry with the canonical approval interrupt id." + ) + continue + interrupt_id = pending_occurrences[0].identity.interrupt_id + if interrupt_id in seen_interrupt_ids: + error = ( + f"Legacy AG-UI tool-message approval repeats call_id '{call_id}'. " + "Retry with one canonical resume entry." + ) + continue + seen_interrupt_ids.add(interrupt_id) + entries.append({"interruptId": interrupt_id, "status": "resolved", "payload": dict(payload)}) + logger.warning( + "Translated a legacy AG-UI tool-message approval for call_id=%s to interrupt id=%s; " + "clients must send canonical resume entries because this compatibility path will be removed.", + call_id, + interrupt_id, + ) + if not entries and not translated_message_ids: + return None + return entries, translated_message_ids, error + + async def run_agent_stream( input_data: dict[str, Any], agent: SupportsAgentRun, @@ -2427,6 +2539,20 @@ async def run_agent_stream( server_tools = collect_server_tools(agent) tools = merge_tools(server_tools, client_tools) workflow_agent_owns_approval = isinstance(agent, WorkflowAgent) + if resume_payload is None: + legacy_resume = _legacy_tool_message_approval_resume( + snapshot_seed_messages if snapshot_seed_messages is not None else raw_messages, + lifecycle=approval_state_store.lifecycle, + thread_id=approval_thread_id, + submitted_messages=raw_messages, + ) + if legacy_resume is not None: + resume_payload, translated_message_ids, legacy_resume_error = legacy_resume + raw_messages[:] = [message for message in raw_messages if id(message) not in translated_message_ids] + if legacy_resume_error is not None: + yield RunStartedEvent(run_id=run_id, thread_id=thread_id) + yield RunErrorEvent(message=legacy_resume_error, code="APPROVAL_RESUME_INVALID") + return approval_resume_messages, handled_resume_ids, cancelled_resume_ids, resume_error = ( _canonical_approval_resume_messages( resume_payload, @@ -2708,6 +2834,7 @@ async def run_agent_stream( tools_for_execution, agent, run_kwargs, + session, approval_thread_id, validated_approved_responses, lifecycle=approval_state_store.lifecycle, @@ -2852,12 +2979,12 @@ async def run_agent_stream( # Register pending approval requests so we can validate responses later if content_type == "function_approval_request": if content.id and content.function_call and content.function_call.name: - canonical_interrupt_id = content.function_call.call_id or content.id + server_label = _function_call_server_label(content.function_call) + canonical_interrupt_id = _approval_interrupt_id(content) provider_approval_thread_id = approval_state_thread_id( scope=approval_scope, thread_id=provider_thread_id or thread_id, ) - server_label = _function_call_server_label(content.function_call) already_approved_requests = _stored_already_approved_requests_for_visible_approval( session, str(content.id), @@ -2875,6 +3002,7 @@ async def run_agent_stream( "arguments": canonical_function_arguments(content.function_call) or "{}", "request_id": str(content.id), "interrupt_id": str(canonical_interrupt_id), + "call_id": str(content.function_call.call_id or canonical_interrupt_id), "already_approved_requests": already_approved_requests, } approval_state_store.register( diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_approval_state.py b/python/packages/ag-ui/agent_framework_ag_ui/_approval_state.py index 1a5bc7b2a1..63193b57fe 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_approval_state.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_approval_state.py @@ -84,6 +84,7 @@ def register( request_id: str, interrupt_id: str, owner: ApprovalExecutionOwner, + call_id: str | None = None, scope: ApprovalScope | None = None, already_approved_requests: list[dict[str, Any]] | None = None, server_label: str | None = None, @@ -95,7 +96,7 @@ def register( scope=scope, thread_ids=unique_thread_ids, interrupt_id=interrupt_id, - call_id=interrupt_id, + call_id=call_id or interrupt_id, name=name, arguments=arguments, aliases=[request_id], diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py b/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py index b9db16161f..48e2f45e35 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py @@ -815,6 +815,7 @@ def _filter_modified_args( name=matching_func_call.name, # type: ignore[arg-type] arguments=json.dumps(filtered_args), ) + func_call_for_approval.id = matching_func_call.id or str(approval_call_id) logger.info(f"Using modified arguments from approval: {filtered_args}") else: # No modified arguments - use the original function call @@ -823,7 +824,7 @@ def _filter_modified_args( # Create function_approval_response content for the agent framework approval_response = Content.from_function_approval_response( approved=accepted, - id=str(approval_call_id), + id=func_call_for_approval.id or str(approval_call_id), function_call=func_call_for_approval, additional_properties={"ag_ui_state_args": state_args} if state_args else None, ) @@ -930,6 +931,7 @@ def _filter_modified_args( name=approval.get("name", ""), arguments=approval.get("arguments", {}), ) + func_call.id = approval.get("id") or None # Create the approval response approval_response = Content.from_function_approval_response( diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py index ef57168f73..5f73c58d53 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py @@ -37,7 +37,7 @@ from ._predictive_state import PredictiveStateHandler from ._state import TOOL_RESULT_DISPLAY_KEY, TOOL_RESULT_STATE_KEY -from ._utils import generate_event_id, make_json_safe, normalize_agui_role +from ._utils import _approval_interrupt_id, generate_event_id, make_json_safe, normalize_agui_role logger = logging.getLogger(__name__) @@ -890,11 +890,12 @@ def _emit_approval_request( events.append(ToolCallEndEvent(tool_call_id=func_call_id)) flow.tool_calls_ended.add(func_call_id) + interrupt_id = _approval_interrupt_id(content) events.append( CustomEvent( name="function_approval_request", value={ - "id": content.id, + "id": interrupt_id, "function_call": { "call_id": func_call_id, "name": func_name, @@ -903,7 +904,6 @@ def _emit_approval_request( }, ) ) - interrupt_id = func_call_id or content.id if interrupt_id: response_schema = _approval_response_schema() if func_call.additional_properties.get("server_label") else None flow.interrupts.append( diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_utils.py b/python/packages/ag-ui/agent_framework_ag_ui/_utils.py index fe93051d6f..62d4f39c26 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_utils.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_utils.py @@ -71,6 +71,31 @@ def canonical_function_arguments(function_call: Any) -> str | None: return json.dumps(make_json_safe(parsed_arguments), sort_keys=True, separators=(",", ":")) +def _function_call_server_label(function_call: Any) -> str | None: + """Return a normalized hosted-tool server label.""" + if function_call is None: + return None + server_label = getattr(function_call, "additional_properties", {}).get("server_label") + return server_label if isinstance(server_label, str) and server_label else None + + +def _approval_interrupt_id(content: Any) -> str | None: + """Return the canonical client and lifecycle identity for an approval request.""" + function_call = getattr(content, "function_call", None) + if function_call is None: + return None + request_id = getattr(content, "id", None) + if _function_call_server_label(function_call) is not None: + return request_id if isinstance(request_id, str) and request_id else None + occurrence_id = getattr(function_call, "id", None) + if isinstance(occurrence_id, str) and occurrence_id: + return occurrence_id + call_id = getattr(function_call, "call_id", None) + if isinstance(call_id, str) and call_id: + return call_id + return request_id if isinstance(request_id, str) and request_id else None + + def get_role_value(message: Any) -> str: """Extract role string from a message object. diff --git a/python/packages/ag-ui/pyproject.toml b/python/packages/ag-ui/pyproject.toml index 7314eb2811..26343ba019 100644 --- a/python/packages/ag-ui/pyproject.toml +++ b/python/packages/ag-ui/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "agent-framework-core>=1.15.0,<2", + "agent-framework-core>=1.17.0,<2", "ag-ui-protocol>=0.1.19,<0.2", "fastapi>=0.121.0,<0.140.0", "httpx>=0.28.1,<1", diff --git a/python/packages/ag-ui/tests/ag_ui/test_agent_wrapper_comprehensive.py b/python/packages/ag-ui/tests/ag_ui/test_agent_wrapper_comprehensive.py index d2b7e9d647..b32cea9af8 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_agent_wrapper_comprehensive.py +++ b/python/packages/ag-ui/tests/ag_ui/test_agent_wrapper_comprehensive.py @@ -14,6 +14,15 @@ from agent_framework_ag_ui._approval_lifecycle import ApprovalExecutionOwner +def _approval_request_id(events: list[Any]) -> str: + event = next( + event + for event in events + if getattr(event, "type", None) == "CUSTOM" and getattr(event, "name", None) == "function_approval_request" + ) + return str(event.value["id"]) + + async def test_agent_initialization_basic(streaming_chat_client_stub): """Test basic agent initialization without state schema.""" from agent_framework.ag_ui import AgentFrameworkAgent @@ -795,6 +804,8 @@ async def stream_fn_turn1( ] assert len(approval_events) == 1, "Expected one approval request event" + approval_id = _approval_request_id(events1) + # --- Turn 2: Client approves → tool executes --- async def stream_fn_turn2( messages: MutableSequence[Message], options: ChatOptions, **kwargs: Any @@ -813,7 +824,7 @@ async def stream_fn_turn2( input_data: dict[str, Any] = { "thread_id": thread_id, "messages": [], - "resume": [{"interruptId": "call_get_datetime_123", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], } events2: list[Any] = [] @@ -1058,7 +1069,8 @@ async def stream_fn_approval( if getattr(e, "type", None) == "CUSTOM" and getattr(e, "name", None) == "function_approval_request" ] assert len(approval_events) == 1, "Expected one approval request event" - assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id="call_sens_001") + approval_id = _approval_request_id(events1) + assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id) # --- Turn 2: legitimate approval --- async def stream_fn_post_approval( @@ -1078,7 +1090,7 @@ async def stream_fn_post_approval( turn2_input: dict[str, Any] = { "thread_id": thread_id, "messages": [], - "resume": [{"interruptId": "call_sens_001", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], } events2: list[Any] = [] @@ -1086,9 +1098,7 @@ async def stream_fn_post_approval( events2.append(event) assert call_count == 1, "Tool should have been executed once" - assert not wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id=thread_id, interrupt_id="call_sens_001" - ) + assert not wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id) # --- Turn 3: replay attempt with the same approval ID --- call_count = 0 # reset @@ -1096,7 +1106,7 @@ async def stream_fn_post_approval( turn3_input: dict[str, Any] = { "thread_id": thread_id, "messages": [], - "resume": [{"interruptId": "call_sens_001", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], } events3: list[Any] = [] @@ -1156,14 +1166,19 @@ async def approval_stream( ) ) - async for _ in wrapper.run({"thread_id": "client-thread", "messages": [{"role": "user", "content": "do it"}]}): - pass + approval_events = [ + event + async for event in wrapper.run( + {"thread_id": "client-thread", "messages": [{"role": "user", "content": "do it"}]} + ) + ] + approval_id = _approval_request_id(approval_events) assert wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id="client-thread", interrupt_id="call_sensitive" + thread_id="client-thread", interrupt_id=approval_id ) assert wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id="provider-conversation", interrupt_id="call_sensitive" + thread_id="provider-conversation", interrupt_id=approval_id ) async def completion_stream( @@ -1182,7 +1197,7 @@ def approval_input(thread_id: str) -> dict[str, Any]: return { "thread_id": thread_id, "messages": [], - "resume": [{"interruptId": "call_sensitive", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], } async for _ in wrapper.run(approval_input(resume_thread_id)): @@ -1190,10 +1205,10 @@ def approval_input(thread_id: str) -> dict[str, Any]: assert execution_count == 1 assert not wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id="client-thread", interrupt_id="call_sensitive" + thread_id="client-thread", interrupt_id=approval_id ) assert not wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id="provider-conversation", interrupt_id="call_sensitive" + thread_id="provider-conversation", interrupt_id=approval_id ) replay_thread_id = "provider-conversation" if resume_thread_id == "client-thread" else "client-thread" @@ -1278,7 +1293,9 @@ async def stream_fn_approval( async for event in wrapper.run({"thread_id": thread_id, "messages": [{"role": "user", "content": "do safe"}]}): events1.append(event) - assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id="call_safe_001") + approval_id = _approval_request_id(events1) + + assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id) # Turn 2: try to approve with a different function name (function name spoofing) async def stream_fn_post( @@ -1301,7 +1318,7 @@ async def stream_fn_post( "content": "approve", "function_approvals": [ { - "id": "call_safe_001", + "id": approval_id, "call_id": "call_safe_001", "name": "dangerous_action", # Mismatch! "approved": True, @@ -1317,9 +1334,9 @@ async def stream_fn_post( events2.append(event) assert not tool_executed, "Function name spoofing should be blocked" - assert wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id=thread_id, interrupt_id="call_safe_001" - ), "Pending approval should be preserved after mismatch for legitimate retry" + assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id), ( + "Pending approval should be preserved after mismatch for legitimate retry" + ) async def test_approval_bypass_via_fabricated_tool_result_is_blocked(streaming_chat_client_stub): @@ -1516,9 +1533,9 @@ async def stream_fn_approval( async for event in wrapper.run({"thread_id": thread_id, "messages": [{"role": "user", "content": "update"}]}): events1.append(event) - assert wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id=thread_id, interrupt_id="call_update_001" - ) + approval_id = _approval_request_id(events1) + + assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id) async def stream_fn_post( messages: MutableSequence[Message], options: ChatOptions, **kwargs: Any @@ -1540,7 +1557,7 @@ async def stream_fn_post( "content": "approve", "function_approvals": [ { - "id": "call_update_001", + "id": approval_id, "call_id": "call_update_001", "name": "update_record", "approved": True, @@ -1556,9 +1573,9 @@ async def stream_fn_post( events2.append(event) assert executed_args == [] - assert wrapper._approval_state_store.lifecycle.pending_occurrence( - thread_id=thread_id, interrupt_id="call_update_001" - ), "Pending approval should be preserved after argument mismatch for legitimate retry" + assert wrapper._approval_state_store.lifecycle.pending_occurrence(thread_id=thread_id, interrupt_id=approval_id), ( + "Pending approval should be preserved after argument mismatch for legitimate retry" + ) async def test_state_update_end_to_end_via_real_tool_invocation(streaming_chat_client_stub): diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index dc4efc6802..120e387817 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -4,6 +4,7 @@ import asyncio import json +import logging import subprocess import sys from collections import Counter @@ -701,7 +702,8 @@ async def stream_fn(messages: Any, options: Any, **kwargs: Any) -> AsyncIterator pause_events = _decode_sse_events(pause_response) pause_finished = [event for event in pause_events if event.get("type") == "RUN_FINISHED"] pause_interrupts = _run_finished_interrupts(pause_finished[-1]) - assert {interrupt["id"] for interrupt in pause_interrupts} == { + interrupt_ids_by_call_id = {interrupt["toolCallId"]: interrupt["id"] for interrupt in pause_interrupts} + assert set(interrupt_ids_by_call_id) == { "refund-call-1", "refund-call-2", }, pause_events @@ -713,8 +715,12 @@ async def stream_fn(messages: Any, options: Any, **kwargs: Any) -> AsyncIterator "threadId": "thread-nested-mixed", "messages": [], "resume": [ - {"interruptId": "refund-call-1", "status": "cancelled"}, - {"interruptId": "refund-call-2", "status": "resolved", "payload": {"approved": True}}, + {"interruptId": interrupt_ids_by_call_id["refund-call-1"], "status": "cancelled"}, + { + "interruptId": interrupt_ids_by_call_id["refund-call-2"], + "status": "resolved", + "payload": {"approved": True}, + }, ], }, ) @@ -2849,6 +2855,392 @@ async def test_endpoint_agent_approval_resume_entry_executes_approved_tool(): assert "outcome" not in [event for event in events if event.get("type") == "RUN_FINISHED"][-1] +async def test_endpoint_agent_legacy_tool_message_uses_unique_pending_call_id( + caplog: pytest.LogCaptureFixture, +) -> None: + """Legacy tool messages remain usable only while their provider call id is unambiguous.""" + executed: list[str] = [] + + def get_weather(city: str) -> str: + executed.append(city) + return f"Sunny in {city}" + + weather_tool = FunctionTool( + name="get_weather", + description="Get the weather for a city", + func=get_weather, + approval_mode="always_require", + ) + function_call = Content.from_function_call( + call_id="provider-weather", + name="get_weather", + arguments={"city": "Seattle"}, + id="af-call-weather", + ) + approval_request = Content.from_function_approval_request( + id="af-call-weather", + function_call=function_call, + ) + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[approval_request], role="assistant")], + default_options={"tools": [weather_tool]}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + client = TestClient(app) + pause = client.post( + "/approval", + json={ + "runId": "run-pause", + "threadId": "thread-legacy-message", + "messages": [{"role": "user", "content": "Weather?"}], + }, + ) + assert pause.status_code == 200 + agent.updates = [AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")] + + with caplog.at_level(logging.WARNING, logger="agent_framework"): + response = client.post( + "/approval", + json={ + "runId": "run-resume", + "threadId": "thread-legacy-message", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-weather", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"city":"Seattle"}', + }, + } + ], + }, + { + "role": "tool", + "toolCallId": "provider-weather", + "content": '{"accepted":true}', + }, + ], + }, + ) + + assert response.status_code == 200 + events = _decode_sse_events(response) + assert executed == ["Seattle"], events + assert "Translated a legacy AG-UI tool-message approval" in caplog.text + + +async def test_endpoint_agent_legacy_tool_message_reuses_historical_confirm_changes_call_id() -> None: + """A sole pending local call may reuse an older synthetic confirmation call id.""" + executed: list[str] = [] + + def guarded_tool(value: str) -> str: + executed.append(value) + return value + + tool = FunctionTool(name="guarded_tool", description="Guarded", func=guarded_tool) + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")], + default_options={"tools": [tool]}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + wrapped_agent._approval_state_store.lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-reused-confirm-id", + interrupt_id="af-call-current", + call_id="provider-reused", + name="guarded_tool", + arguments='{"value":"current"}', + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + + response = TestClient(app).post( + "/approval", + json={ + "runId": "run-reused-confirm-id", + "threadId": "thread-reused-confirm-id", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-reused", + "type": "function", + "function": {"name": "confirm_changes", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "toolCallId": "provider-reused", "content": '{"accepted":true}'}, + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-reused", + "type": "function", + "function": {"name": "guarded_tool", "arguments": '{"value":"current"}'}, + } + ], + }, + {"role": "tool", "toolCallId": "provider-reused", "content": '{"accepted":true}'}, + ], + }, + ) + + assert response.status_code == 200 + events = _decode_sse_events(response) + assert executed == ["current"], events + assert not [event for event in events if event.get("type") == "RUN_ERROR"] + + +async def test_endpoint_agent_legacy_tool_message_rejects_reused_call_id( + caplog: pytest.LogCaptureFixture, +) -> None: + """A provider call id shared by retained occurrences cannot authorize either one.""" + executed: list[str] = [] + + def first_tool() -> str: + executed.append("first") + return "first" + + def second_tool() -> str: + executed.append("second") + return "second" + + tools = [ + FunctionTool(name="first_tool", description="First", func=first_tool), + FunctionTool(name="second_tool", description="Second", func=second_tool), + ] + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")], + default_options={"tools": tools}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + lifecycle = wrapped_agent._approval_state_store.lifecycle + lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-legacy-reused", + interrupt_id="approval-first", + call_id="provider-reused", + name="first_tool", + arguments="{}", + ) + lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-legacy-reused", + interrupt_id="approval-second", + call_id="provider-reused", + name="second_tool", + arguments="{}", + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + + with caplog.at_level(logging.WARNING, logger="agent_framework"): + response = TestClient(app).post( + "/approval", + json={ + "runId": "run-legacy-reused", + "threadId": "thread-legacy-reused", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-reused", + "type": "function", + "function": {"name": "first_tool", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "toolCallId": "provider-reused", + "content": '{"accepted":true}', + }, + ], + }, + ) + + assert response.status_code == 200 + events = _decode_sse_events(response) + assert executed == [] + assert [event for event in events if event.get("type") == "RUN_ERROR"] + assert "does not identify exactly one retained pending local occurrence" in events[-1]["message"] + + +async def test_endpoint_agent_legacy_tool_message_cannot_collide_with_interrupt_id() -> None: + """An unknown provider call id cannot be reinterpreted as a canonical interrupt id.""" + executed: list[str] = [] + + def guarded_tool() -> str: + executed.append("ran") + return "done" + + tool = FunctionTool(name="guarded_tool", description="Guarded", func=guarded_tool) + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")], + default_options={"tools": [tool]}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + wrapped_agent._approval_state_store.lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-legacy-collision", + interrupt_id="af-call-secret", + call_id="provider-real", + name="guarded_tool", + arguments="{}", + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + + response = TestClient(app).post( + "/approval", + json={ + "runId": "run-legacy-collision", + "threadId": "thread-legacy-collision", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "af-call-secret", + "type": "function", + "function": {"name": "guarded_tool", "arguments": "{}"}, + } + ], + }, + { + "role": "tool", + "actionExecutionId": "af-call-secret", + "content": None, + "result": {"accepted": True}, + }, + ], + }, + ) + + assert response.status_code == 200 + assert executed == [] + events = _decode_sse_events(response) + assert events[-1]["code"] == "APPROVAL_RESUME_REQUIRED" + + +async def test_endpoint_agent_legacy_tool_message_rejects_duplicate_decisions() -> None: + """Conflicting legacy decisions cannot select an earlier approval.""" + executed: list[str] = [] + + def guarded_tool() -> str: + executed.append("ran") + return "done" + + tool = FunctionTool(name="guarded_tool", description="Guarded", func=guarded_tool) + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")], + default_options={"tools": [tool]}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + wrapped_agent._approval_state_store.lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-legacy-duplicate", + interrupt_id="af-call-current", + call_id="provider-call", + name="guarded_tool", + arguments="{}", + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + + response = TestClient(app).post( + "/approval", + json={ + "runId": "run-legacy-duplicate", + "threadId": "thread-legacy-duplicate", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-call", + "type": "function", + "function": {"name": "guarded_tool", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "toolCallId": "provider-call", "content": '{"accepted":true}'}, + {"role": "tool", "toolCallId": "provider-call", "content": '{"accepted":false}'}, + ], + }, + ) + + assert response.status_code == 200 + events = _decode_sse_events(response) + assert executed == [] + assert events[-1]["code"] == "APPROVAL_RESUME_INVALID" + assert "repeats call_id" in events[-1]["message"] + + +async def test_endpoint_agent_historical_legacy_approval_cannot_authorize_newer_reused_call() -> None: + """A historical approval outside the submitted turn suffix remains inert.""" + executed: list[str] = [] + + def dangerous_tool(value: str) -> str: + executed.append(value) + return value + + tool = FunctionTool(name="dangerous_tool", description="Dangerous", func=dangerous_tool) + agent = StubAgent( + updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")], + default_options={"tools": [tool]}, + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + wrapped_agent._approval_state_store.lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="thread-historical-legacy", + interrupt_id="af-call-current", + call_id="provider-reused", + name="dangerous_tool", + arguments='{"value":"new"}', + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + + response = TestClient(app).post( + "/approval", + json={ + "runId": "run-historical-legacy", + "threadId": "thread-historical-legacy", + "messages": [ + { + "role": "assistant", + "toolCalls": [ + { + "id": "provider-reused", + "type": "function", + "function": { + "name": "old_tool", + "arguments": '{"value":"old"}', + }, + } + ], + }, + {"role": "tool", "toolCallId": "provider-reused", "content": '{"accepted":true}'}, + {"role": "user", "content": "Continue without approving anything."}, + ], + }, + ) + + assert response.status_code == 200 + events = _decode_sse_events(response) + assert executed == [] + assert events[-1]["code"] == "APPROVAL_RESUME_REQUIRED" + + async def test_endpoint_agent_approval_resume_remains_retryable_when_local_tool_is_temporarily_unavailable(): """A local approval can be retried after its executor disappears before resume.""" client, agent, executed_cities = _build_weather_approval_endpoint(snapshot_store=InMemoryAGUIThreadSnapshotStore()) @@ -2910,7 +3302,10 @@ async def test_endpoint_agent_approval_resume_releases_already_approved_sibling( pause_events = _decode_sse_events(pause_response) pause_finished = [event for event in pause_events if event.get("type") == "RUN_FINISHED"] interrupts = _run_finished_interrupts(pause_finished[-1]) - assert [interrupt["id"] for interrupt in interrupts] == ["call_sensitive"] + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_sensitive" assert not [event for event in pause_events if event.get("type") == "TOOL_CALL_RESULT"] state["phase"] = "resume" @@ -2920,7 +3315,7 @@ async def test_endpoint_agent_approval_resume_releases_already_approved_sibling( "runId": "run-resume", "threadId": "thread-mixed-batch", "messages": [], - "resume": [{"interruptId": "call_sensitive", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) @@ -2944,6 +3339,137 @@ async def test_endpoint_agent_approval_resume_releases_already_approved_sibling( assert sorted(replayed_call_ids) == ["call_sensitive", "call_weather"] +async def test_endpoint_agent_approval_resume_distinguishes_hidden_siblings_with_reused_call_id( + streaming_chat_client_stub, +) -> None: + """Distinct hidden occurrences sharing a provider call ID resume and execute once.""" + executed: list[str] = [] + state = {"phase": "pause"} + + def guarded_tool() -> str: + executed.append("guarded") + return "guarded result" + + def first_safe_tool() -> str: + executed.append("first-safe") + return "first safe result" + + def second_safe_tool() -> str: + executed.append("second-safe") + return "second safe result" + + async def stream_fn( + messages: list[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> AsyncIterator[ChatResponseUpdate]: + del messages, options, kwargs + if state["phase"] == "pause": + yield ChatResponseUpdate( + contents=[ + Content.from_function_call( + id="guarded-occurrence", + call_id="provider-shared", + name="guarded_tool", + arguments="{}", + ), + Content.from_function_call( + id="first-safe-occurrence", + call_id="provider-shared", + name="first_safe_tool", + arguments="{}", + ), + Content.from_function_call( + id="second-safe-occurrence", + call_id="provider-shared", + name="second_safe_tool", + arguments="{}", + ), + ], + role="assistant", + ) + return + yield ChatResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant") + + agent = Agent( + name="test_agent", + instructions="Test", + client=streaming_chat_client_stub(stream_fn), + tools=[ + FunctionTool( + name="guarded_tool", + description="Guarded tool", + func=guarded_tool, + approval_mode="always_require", + ), + FunctionTool(name="first_safe_tool", description="First safe tool", func=first_safe_tool), + FunctionTool(name="second_safe_tool", description="Second safe tool", func=second_safe_tool), + ], + ) + wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) + app = FastAPI() + add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/approval") + client = TestClient(app) + thread_id = "thread-shared-provider-call" + + pause_response = client.post( + "/approval", + json={ + "runId": "run-pause", + "threadId": thread_id, + "messages": [{"role": "user", "content": "Run all three tools"}], + }, + ) + + assert pause_response.status_code == 200 + pause_events = _decode_sse_events(pause_response) + pause_finished = [event for event in pause_events if event.get("type") == "RUN_FINISHED"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert [(interrupt["id"], interrupt["toolCallId"]) for interrupt in interrupts] == [ + ("guarded-occurrence", "provider-shared") + ] + assert executed == [] + + state["phase"] = "resume" + resume_response = client.post( + "/approval", + json={ + "runId": "run-resume", + "threadId": thread_id, + "messages": [], + "resume": [ + { + "interruptId": "guarded-occurrence", + "status": "resolved", + "payload": {"accepted": True}, + } + ], + }, + ) + + assert resume_response.status_code == 200 + resume_events = _decode_sse_events(resume_response) + assert not [event for event in resume_events if event.get("type") == "RUN_ERROR"] + assert Counter(executed) == {"guarded": 1, "first-safe": 1, "second-safe": 1} + tool_results = [event for event in resume_events if event.get("type") == "TOOL_CALL_RESULT"] + assert Counter(event["content"] for event in tool_results) == { + "guarded result": 1, + "first safe result": 1, + "second safe result": 1, + } + assert {event["toolCallId"] for event in tool_results} == {"provider-shared"} + + occurrences = wrapped_agent._approval_state_store.lifecycle.occurrences_for_thread(thread_id=thread_id) + assert {occurrence.identity.interrupt_id for occurrence in occurrences} == { + "guarded-occurrence", + "first-safe-occurrence", + "second-safe-occurrence", + } + assert {occurrence.identity.call_id for occurrence in occurrences} == {"provider-shared"} + assert len({occurrence.identity.occurrence_id for occurrence in occurrences}) == 3 + assert {occurrence.status for occurrence in occurrences} == {ApprovalStatus.SETTLED} + + async def test_endpoint_agent_approval_resume_persists_replayable_tool_results(streaming_chat_client_stub): """Approved batches should hydrate with real results under original tool call ids.""" client, executed, messages_received, state = _build_mixed_approval_batch_endpoint( @@ -2961,7 +3487,11 @@ async def test_endpoint_agent_approval_resume_persists_replayable_tool_results(s ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_sensitive"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_sensitive" state["phase"] = "resume" resume_response = client.post( @@ -2970,7 +3500,7 @@ async def test_endpoint_agent_approval_resume_persists_replayable_tool_results(s "runId": "run-resume", "threadId": "thread-mixed-replay", "messages": [], - "resume": [{"interruptId": "call_sensitive", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) @@ -3040,7 +3570,11 @@ async def test_endpoint_agent_approval_resume_surfaces_queued_tool_approval(stre ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_first"] + first_interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(first_interrupts) == 1 + first_approval_id = first_interrupts[0]["id"] + assert first_approval_id.startswith("af-call-") + assert first_interrupts[0]["toolCallId"] == "call_first" assert executed == [] state["phase"] = "resume" @@ -3050,7 +3584,7 @@ async def test_endpoint_agent_approval_resume_surfaces_queued_tool_approval(stre "runId": "run-resume-first", "threadId": "thread-queued-approval", "messages": [], - "resume": [{"interruptId": "call_first", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": first_approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) @@ -3059,7 +3593,11 @@ async def test_endpoint_agent_approval_resume_surfaces_queued_tool_approval(stre tool_results = [event for event in first_resume_events if event.get("type") == "TOOL_CALL_RESULT"] assert [(event["toolCallId"], event["content"]) for event in tool_results] == [("call_first", "first result")] first_resume_finished = [event for event in first_resume_events if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(first_resume_finished[-1])] == ["call_second"] + second_interrupts = _run_finished_interrupts(first_resume_finished[-1]) + assert len(second_interrupts) == 1 + second_approval_id = second_interrupts[0]["id"] + assert second_approval_id.startswith("af-call-") + assert second_interrupts[0]["toolCallId"] == "call_second" assert not [ event for event in first_resume_events @@ -3074,7 +3612,7 @@ async def test_endpoint_agent_approval_resume_surfaces_queued_tool_approval(stre "runId": "run-resume-second", "threadId": "thread-queued-approval", "messages": [], - "resume": [{"interruptId": "call_second", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": second_approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) @@ -3104,7 +3642,11 @@ async def test_endpoint_agent_approval_cancel_discards_queued_tool_approval(stre ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_first"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_first" assert executed == [] state["phase"] = "resume" @@ -3114,7 +3656,7 @@ async def test_endpoint_agent_approval_cancel_discards_queued_tool_approval(stre "runId": "run-cancel", "threadId": "thread-queued-cancel", "messages": [], - "resume": [{"interruptId": "call_first", "status": "cancelled"}], + "resume": [{"interruptId": approval_id, "status": "cancelled"}], }, ) @@ -3163,7 +3705,11 @@ async def test_endpoint_agent_approval_cancel_clears_queued_state_when_visible_e ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_first"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_first" stored_state = wrapped_agent._approval_state_store.get_tool_approval_state("thread-queued-cancel-evicted") assert stored_state is not None assert "call_second" in json.dumps(stored_state) @@ -3176,7 +3722,7 @@ async def test_endpoint_agent_approval_cancel_clears_queued_state_when_visible_e "runId": "run-cancel", "threadId": "thread-queued-cancel-evicted", "messages": [], - "resume": [{"interruptId": "call_first", "status": "cancelled"}], + "resume": [{"interruptId": approval_id, "status": "cancelled"}], }, ) @@ -3223,7 +3769,11 @@ async def test_endpoint_agent_approval_resume_processes_collected_auto_approved_ ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_manual"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_manual" assert executed == [] state["phase"] = "resume" @@ -3233,7 +3783,7 @@ async def test_endpoint_agent_approval_resume_processes_collected_auto_approved_ "runId": "run-resume", "threadId": "thread-auto-approval", "messages": [], - "resume": [{"interruptId": "call_manual", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) @@ -3264,7 +3814,11 @@ async def test_endpoint_agent_approval_rejection_releases_already_approved_sibli ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_sensitive"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_sensitive" state["phase"] = "resume" resume_response = client.post( @@ -3273,7 +3827,7 @@ async def test_endpoint_agent_approval_rejection_releases_already_approved_sibli "runId": "run-resume", "threadId": "thread-mixed-reject", "messages": [], - "resume": [{"interruptId": "call_sensitive", "status": "resolved", "payload": {"accepted": False}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": False}}], }, ) @@ -3308,7 +3862,11 @@ async def test_endpoint_agent_approval_cancellation_does_not_release_already_app ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_sensitive"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_sensitive" state["phase"] = "resume" cancel_response = client.post( @@ -3317,7 +3875,7 @@ async def test_endpoint_agent_approval_cancellation_does_not_release_already_app "runId": "run-cancel", "threadId": "thread-mixed-cancel", "messages": [], - "resume": [{"interruptId": "call_sensitive", "status": "cancelled"}], + "resume": [{"interruptId": approval_id, "status": "cancelled"}], }, ) @@ -7890,7 +8448,11 @@ async def stream_fn( ) assert pause_response.status_code == 200 pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] - assert [interrupt["id"] for interrupt in _run_finished_interrupts(pause_finished[-1])] == ["call_provider"] + interrupts = _run_finished_interrupts(pause_finished[-1]) + assert len(interrupts) == 1 + approval_id = interrupts[0]["id"] + assert approval_id.startswith("af-call-") + assert interrupts[0]["toolCallId"] == "call_provider" assert side_effects == [] # Resume with approval: the deferred provider tool runs during agent.run. @@ -7901,7 +8463,7 @@ async def stream_fn( "runId": "run-resume", "threadId": "thread-provider", "messages": [], - "resume": [{"interruptId": "call_provider", "status": "resolved", "payload": {"accepted": True}}], + "resume": [{"interruptId": approval_id, "status": "resolved", "payload": {"accepted": True}}], }, ) assert resume_response.status_code == 200 diff --git a/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py b/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py index ae2d7f2ece..e002058609 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py +++ b/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py @@ -12,7 +12,7 @@ import json from typing import Any -from agent_framework import AgentResponseUpdate, Content +from agent_framework import AgentResponseUpdate, Content, FunctionTool from conftest import StubAgent # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] from fastapi import FastAPI from fastapi.testclient import TestClient @@ -27,6 +27,7 @@ InMemoryAGUIThreadSnapshotStore, add_agent_framework_fastapi_endpoint, ) +from agent_framework_ag_ui._approval_lifecycle import ApprovalExecutionOwner def _build_app_with_agent(updates: list[AgentResponseUpdate], **kwargs: Any) -> FastAPI: @@ -705,6 +706,77 @@ async def test_service_session_keeps_incremental_trusted_tool_result() -> None: ) +async def test_service_session_legacy_approval_validates_against_authoritative_snapshot_history() -> None: + executed: list[str] = [] + + def guarded_write(value: str) -> str: + executed.append(value) + return value + + tool = FunctionTool( + name="guarded_write", + description="Write a guarded value", + func=guarded_write, + approval_mode="always_require", + ) + stub = StubAgent(default_options={"tools": [tool], "response_format": None}) + store = InMemoryAGUIThreadSnapshotStore() + runner = AgentFrameworkAgent(agent=stub, use_service_session=True, snapshot_store=store) + runner._approval_state_store.lifecycle.register( + owner=ApprovalExecutionOwner.LOCAL, + thread_id="service-legacy-approval", + interrupt_id="af-call-write", + call_id="provider-call-write", + name="guarded_write", + arguments='{"value":"approved"}', + ) + await store.save( + scope="service", + thread_id="service-legacy-approval", + snapshot=AGUIThreadSnapshot( + messages=[ + {"role": "user", "content": "Write it"}, + { + "role": "assistant", + "tool_calls": [ + { + "id": "provider-call-write", + "type": "function", + "function": { + "name": "guarded_write", + "arguments": '{"value":"approved"}', + }, + } + ], + }, + ] + ), + ) + + events = await _run_service_turn( + runner, + thread_id="service-legacy-approval", + scope="service", + messages=[ + { + "role": "tool", + "toolCallId": "provider-call-write", + "content": '{"accepted":true}', + } + ], + ) + + assert executed == ["approved"] + assert not any(getattr(event, "type", None) == "RUN_ERROR" for event in events) + received_results = [ + content + for message in stub.messages_received + for content in message.contents + if content.type == "function_result" + ] + assert [(result.call_id, result.result) for result in received_results] == [("provider-call-write", "approved")] + + async def test_service_session_empty_generic_resume_invokes_agent_with_only_new_result() -> None: runner, stub, store = _service_session_runner() await store.save( diff --git a/python/packages/ag-ui/tests/ag_ui/test_run.py b/python/packages/ag-ui/tests/ag_ui/test_run.py index b593f0ef69..f63d52423c 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run.py @@ -990,6 +990,71 @@ def test_emit_approval_request_populates_interrupt_metadata(): } +def test_emit_local_approval_request_prefers_function_call_occurrence_id() -> None: + """Local approval interrupts use occurrence identity without rewriting tool correlation.""" + flow = FlowState(message_id="msg-1") + function_call = Content.from_function_call( + call_id="call_123", + name="write_doc", + arguments={"content": "x"}, + id="af-call-occurrence", + ) + with pytest.warns(FutureWarning, match="id differs from function_call.id.*legacy"): + approval_content = Content.from_function_approval_request(id="call_123", function_call=function_call) + + events = _emit_approval_request(approval_content, flow) + + custom_event = next(event for event in events if isinstance(event, CustomEvent)) + assert custom_event.value["id"] == "af-call-occurrence" + assert flow.interrupts[0]["id"] == "af-call-occurrence" + assert flow.interrupts[0]["toolCallId"] == "call_123" + + +def test_emit_approval_request_normalizes_empty_server_label_for_identity() -> None: + """Client events and lifecycle registration treat an empty server label as local.""" + flow = FlowState(message_id="msg-1") + function_call = Content.from_function_call( + call_id="provider-call", + name="write_doc", + arguments={"content": "x"}, + id="af-call-occurrence", + additional_properties={"server_label": ""}, + ) + approval_content = Content.from_function_approval_request( + id="provider-approval-request", + function_call=function_call, + ) + + events = _emit_approval_request(approval_content, flow) + + custom_event = next(event for event in events if getattr(event, "name", None) == "function_approval_request") + assert custom_event.value["id"] == "af-call-occurrence" # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + assert flow.interrupts[0]["id"] == "af-call-occurrence" + + +def test_emit_hosted_approval_request_preserves_provider_request_id() -> None: + """Hosted approval interrupts retain the provider protocol request identity.""" + flow = FlowState(message_id="msg-1") + function_call = Content.from_function_call( + call_id="provider-call", + name="hosted_search", + arguments={"query": "x"}, + id="af-call-occurrence", + additional_properties={"server_label": "provider"}, + ) + approval_content = Content.from_function_approval_request( + id="provider-approval-request", + function_call=function_call, + ) + + events = _emit_approval_request(approval_content, flow) + + custom_event = next(event for event in events if isinstance(event, CustomEvent)) + assert custom_event.value["id"] == "provider-approval-request" + assert flow.interrupts[0]["id"] == "provider-approval-request" + assert flow.interrupts[0]["toolCallId"] == "provider-call" + + def test_emit_approval_request_reuses_confirmation_message_id_in_snapshot(): """Confirmation tool events and snapshots share the same message ID.""" flow = FlowState() diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index f1e3af8280..257782f7c3 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -156,10 +156,13 @@ agent_framework/ caller messages, returns approved and rejected terminal results in the resumed response (and stream) before any final assistant message, and does not mutate the caller's approval `Message` or the earlier approval-request response. -- Approval/result correlation is occurrence-aware. A `call_id` may be reused after a completed round, so approval - normalization matches ordered call occurrences and consumes approved results per occurrence rather than using one - global result per `call_id`. All contents produced by one execution remain one result group and are consumed - together, including multiple user-input requests. +- Approval/result correlation is occurrence-aware. Provider/service correlation stays in `function_call.call_id`, + while new locally actionable calls carry one stable Agent Framework occurrence identity in `function_call.id`. + New local approval request ids use that occurrence id; hosted provider-issued approval ids remain unchanged. Legacy + stored pending calls without `function_call.id` retain exact request-id binding for one warned compatibility resume. + A `call_id` may be reused after a completed round, so approval normalization matches ordered call occurrences and + consumes approved results per occurrence rather than using one global result per `call_id`. All contents produced by + one execution remain one result group and are consumed together, including multiple user-input requests. - Approval resume keeps terminal `function_result` contents in tool-role messages and follow-up user-input requests in assistant-role messages, including mixed sibling batches. - Function-call budget accounting counts one unit per executed result group, not per emitted `function_result`, so diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 1e4089808a..9b40754760 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -10,6 +10,7 @@ import logging import sys import typing +import warnings from collections import deque from collections.abc import ( AsyncIterable, @@ -38,6 +39,7 @@ get_origin, overload, ) +from uuid import uuid4 from opentelemetry.metrics import Histogram, NoOpHistogram from pydantic import BaseModel, Field, ValidationError, create_model @@ -92,6 +94,11 @@ logger = logging.getLogger("agent_framework") +def _generate_function_call_occurrence_id() -> str: + """Generate an Agent Framework identity for one function-call occurrence.""" + return f"af-call-{uuid4().hex}" + + DEFAULT_MAX_ITERATIONS: Final[int] = 40 DEFAULT_MAX_CONSECUTIVE_ERRORS_PER_REQUEST: Final[int] = 3 SHELL_TOOL_KIND_VALUE: Final[str] = "shell" @@ -1585,8 +1592,10 @@ async def _auto_invoke_function( if call_id is None: raise KeyError(f'Function "{function_call_content.name}" is missing call_id.') - # Always pass call_id to middleware for policy violation approval flow + # Pass both provider correlation and framework occurrence identity to middleware. middleware_context.metadata["call_id"] = call_id + if function_call_content.id is not None: + middleware_context.metadata["function_call_occurrence_id"] = function_call_content.id # Pass through the original approval response so middleware can decide whether # this replay corresponds to a middleware-specific approval flow. @@ -1803,7 +1812,7 @@ async def _try_execute_function_call_groups( if function_call.type != "function_call": continue approval_request = Content.from_function_approval_request( - id=function_call.call_id, # type: ignore[arg-type] + id=function_call.id or function_call.call_id, # type: ignore[arg-type] function_call=function_call, ) tool_name = function_call.name @@ -1838,7 +1847,8 @@ async def _try_execute_function_call_groups( for function_call in function_calls: if function_call.type == "function_call": function_call.user_input_request = True - function_call.id = function_call.call_id + if function_call.id is None: + function_call.id = function_call.call_id declaration_only_calls.append(function_call) return [[function_call] for function_call in declaration_only_calls], False @@ -2190,25 +2200,69 @@ def _bind_approval_response_to_pending_request( if invocation_session is None: return response - if response.id is None: - return None pending = _load_pending_approval_requests(invocation_session) - request = pending.get(response.id) - if request is None or request.function_call is None: + request_key = response.id + request = pending.get(request_key) if request_key is not None else None + + # During the staged migration, accept the occurrence id even if an intermediate + # producer still stored the provider call_id as the request id. This is not a + # call_id alias: the lookup uses the stored function_call.id only. + if request is None and response.id is not None: + matching_occurrences = [ + (pending_id, candidate) + for pending_id, candidate in pending.items() + if not _is_hosted_tool_approval(candidate) + and candidate.function_call is not None + and candidate.function_call.id == response.id + ] + if len(matching_occurrences) == 1: + request_key, request = matching_occurrences[0] + + if request is None or request.function_call is None or request_key is None: return None - rebound_call = _content_from_state(request.function_call.to_dict()) + + stored_call = request.function_call + is_hosted = _is_hosted_tool_approval(request) + occurrence_id = stored_call.id + if not is_hosted and occurrence_id is not None: + embedded_call = response.function_call + uses_occurrence_id = response.id == occurrence_id + uses_legacy_request_id = response.id == request.id + if not uses_occurrence_id: + if not (uses_legacy_request_id and embedded_call is not None and embedded_call.id == occurrence_id): + return None + warnings.warn( + "An occurrence-aware approval used the legacy provider call_id request binding. " + "Return function_call.id as the approval response id; legacy request-id binding will be removed " + "in a future release.", + FutureWarning, + stacklevel=3, + ) + elif embedded_call is not None and embedded_call.id != occurrence_id: + return None + elif not is_hosted: + warnings.warn( + "Resuming a legacy stored approval whose function_call has no Content.id. This exact request-id " + "compatibility path is deprecated; complete the pending approval and store occurrence-aware snapshots " + "before support is removed in a future release.", + FutureWarning, + stacklevel=3, + ) + + rebound_call = _content_from_state(stored_call.to_dict()) if rebound_call is None: return None + rebound_id = occurrence_id if not is_hosted and occurrence_id is not None else response.id rebound = Content.from_function_approval_response( approved=_is_approval_granted(response.approved), - id=response.id, + id=rebound_id, # type: ignore[arg-type] function_call=rebound_call, annotations=response.annotations, additional_properties=copy.deepcopy(response.additional_properties), raw_representation=response.raw_representation, ) if consume: - pending.pop(response.id, None) + pending.pop(request_key, None) _save_pending_approval_requests(invocation_session, pending) return rebound @@ -2235,7 +2289,8 @@ def _bind_approval_responses_to_pending_requests( ) if rebound is None: logger.warning( - "Ignored an approval response with request id %r because no pending approval request exists.", + "Ignored an approval response with id %r because it did not match the active approval " + "occurrence identity; the pending request was retained for retry.", content.id, ) continue @@ -2673,26 +2728,35 @@ def find_approval_occurrence(approval_id: str) -> _ApprovalCallOccurrence | None def _extract_function_calls(response: ChatResponse) -> list[Content]: - completed_call_ids: set[str] = set() - seen_call_ids: set[str] = set() + completed_occurrence_ids: set[str] = set() + open_occurrence_ids_by_call_id: dict[str, deque[str]] = {} + seen_occurrence_ids: set[str] = set() candidate_calls: list[Content] = [] for message in response.messages: for item in message.contents: if item.type == "function_result" and item.call_id: - completed_call_ids.add(item.call_id) + if open_occurrence_ids := open_occurrence_ids_by_call_id.get(item.call_id): + completed_occurrence_ids.add(open_occurrence_ids.popleft()) continue if not _is_actionable_function_call(item): continue - if item.call_id and item.call_id in seen_call_ids: + if item.id is None: + item.id = _generate_function_call_occurrence_id() + if not item.call_id: + item.call_id = item.id + 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, + ) + if item.id in seen_occurrence_ids: continue - if item.call_id: - seen_call_ids.add(item.call_id) + seen_occurrence_ids.add(item.id) candidate_calls.append(item) - return [ - function_call - for function_call in candidate_calls - if not function_call.call_id or function_call.call_id not in completed_call_ids - ] + open_occurrence_ids_by_call_id.setdefault(item.call_id, deque()).append(item.id) + return [function_call for function_call in candidate_calls if function_call.id not in completed_occurrence_ids] def _prepend_function_call_messages(response: ChatResponse, function_call_messages: list[Message]) -> None: @@ -3414,7 +3478,61 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non total_function_calls, max_function_calls, ) + 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, + ) + 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: diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 7481d13e13..3456dd407f 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -8,6 +8,7 @@ import logging import re import sys +import warnings from asyncio import iscoroutine from collections.abc import ( AsyncGenerator, @@ -816,6 +817,7 @@ def from_function_call( arguments: str | Mapping[str, Any] | None = None, exception: str | None = None, informational_only: bool = False, + id: str | None = None, annotations: Sequence[Annotation] | None = None, additional_properties: MutableMapping[str, Any] | None = None, raw_representation: Any = None, @@ -834,6 +836,8 @@ def from_function_call( error state. informational_only: Whether the function call is present only for transcript fidelity and should not be executed by Agent Framework function invocation. + id: Stable Agent Framework identity for this occurrence. When omitted, the function invocation layer + assigns one before a locally actionable call is processed. annotations: Optional annotations attached to this content item. additional_properties: Extra provider-specific properties to preserve with the content item. raw_representation: The original provider-specific object or payload this content item was created from. @@ -848,6 +852,7 @@ def from_function_call( arguments=arguments, exception=exception, informational_only=informational_only, + id=id, annotations=annotations, additional_properties=additional_properties, raw_representation=raw_representation, @@ -1282,6 +1287,19 @@ def from_function_approval_request( raw_representation: Any = None, ) -> ContentT: """Create function approval request content.""" + if ( + function_call.type == "function_call" + and function_call.id is not None + and id != function_call.id + and function_call.additional_properties.get("server_label") is None + ): + warnings.warn( + "Creating a local function_approval_request whose id differs from function_call.id uses the legacy " + "provider call_id binding. Use function_call.id as the approval request id; legacy binding support " + "will be removed in a future release.", + FutureWarning, + stacklevel=2, + ) return cls( "function_approval_request", id=id, @@ -1541,9 +1559,11 @@ def _add_text_reasoning_content(self, other: Content) -> Content: def _add_function_call_content(self, other: Content) -> Content: """Add two FunctionCallContent instances.""" + if self.id and other.id and self.id != other.id: + raise AdditionItemMismatch("Cannot merge function calls with different ids") other_call_id = getattr(other, "call_id", None) self_call_id = getattr(self, "call_id", None) - if other_call_id and self_call_id != other_call_id: + if self_call_id and other_call_id and self_call_id != other_call_id: raise ContentError("Cannot add function calls with different call_ids") self_arguments = getattr(self, "arguments", None) @@ -1562,7 +1582,7 @@ def _add_function_call_content(self, other: Content) -> Content: return Content( "function_call", - call_id=self_call_id, + call_id=self_call_id or other_call_id, name=getattr(self, "name", None) or getattr(other, "name", None), arguments=arguments, id=self.id or other.id, @@ -2148,6 +2168,30 @@ def _finalize_response(response: ChatResponse | AgentResponse) -> None: _coalesce_text_content(msg.contents, "text") _coalesce_text_content(msg.contents, "text_reasoning") _coalesce_code_interpreter_content(msg.contents) + _coalesce_function_call_occurrences(response) + + +def _coalesce_function_call_occurrences(response: ChatResponse | AgentResponse) -> None: + """Merge streamed function-call fragments that share a stable occurrence id.""" + occurrences: dict[str, tuple[list[Content], int, Content]] = {} + for message in response.messages: + original_contents = message.contents + coalesced_contents: list[Content] = [] + message.contents = coalesced_contents + for content in original_contents: + if content.type != "function_call" or content.id is None: + coalesced_contents.append(content) + continue + existing = occurrences.get(content.id) + if existing is None: + coalesced_contents.append(content) + occurrences[content.id] = (coalesced_contents, len(coalesced_contents) - 1, content) + continue + contents, index, accumulated = existing + merged = accumulated + content + contents[index] = merged + occurrences[content.id] = (contents, index, merged) + response.messages[:] = [message for message in response.messages if message.contents] # region ContinuationToken diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index 1aed8c0596..0df47d2b34 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -772,7 +772,18 @@ def _extract_function_responses( request_id = content.call_id if request_id is None: raise AgentInvalidResponseException("Function result is missing its call ID.") - pending_request = pending_requests.get(request_id) + response_request_id = request_id + pending_request = pending_requests.get(response_request_id) + if pending_request is None: + matching_requests = [ + (pending_id, pending_event) + for pending_id, pending_event in pending_requests.items() + if isinstance(pending_event.data, Content) + and pending_event.data.type == "function_call" + and pending_event.data.call_id == request_id + ] + if len(matching_requests) == 1: + response_request_id, pending_request = matching_requests[0] response_data = ( content if pending_request is not None @@ -781,7 +792,7 @@ def _extract_function_responses( and pending_request.data.type == "function_call" else content.result ) - function_responses[request_id] = response_data + function_responses[response_request_id] = response_data else: raise AgentInvalidResponseException( "Unexpected content type while awaiting request info responses." diff --git a/python/packages/core/agent_framework/security.py b/python/packages/core/agent_framework/security.py index 397f9af62f..d9d857c9c7 100644 --- a/python/packages/core/agent_framework/security.py +++ b/python/packages/core/agent_framework/security.py @@ -1696,11 +1696,11 @@ def __init__( self.block_on_violation = block_on_violation if not approval_on_violation else False self.enable_audit_log = enable_audit_log self.audit_log: list[dict[str, Any]] = [] - # Track call_ids awaiting approval, each mapped to a binding record capturing the exact - # invocation the approval was requested for: the function name + arguments, the security - # label (integrity/confidentiality) shown for review, and the session. Combined with the - # call_id key and consume-on-use, an approval cannot re-authorize a repeated call, a - # different function, changed arguments, a different security label, or a different session. + # Track occurrence-aware approval ids, each mapped to a binding record capturing the exact + # invocation the approval was requested for: the provider call id, function name + arguments, + # security label shown for review, and session. Combined with consume-on-use, an approval + # cannot re-authorize a repeated call, a different function, changed arguments, a different + # security label, or a different session. self._pending_policy_approvals: dict[str, _PendingPolicyApproval] = {} def _get_call_id(self, context: FunctionInvocationContext) -> str: @@ -1708,6 +1708,13 @@ def _get_call_id(self, context: FunctionInvocationContext) -> str: call_id = context.metadata.get("call_id", "") return call_id if isinstance(call_id, str) else "" + def _get_approval_id(self, context: FunctionInvocationContext) -> str: + """Get the occurrence-aware approval id, falling back for legacy direct callers.""" + occurrence_id = context.metadata.get("function_call_occurrence_id") + if isinstance(occurrence_id, str) and occurrence_id: + return occurrence_id + return self._get_call_id(context) + def _current_arguments(self, context: FunctionInvocationContext) -> dict[str, Any]: """Resolve the current call arguments, preferring unexpanded ([var_xxx]) originals.""" # Use original unexpanded arguments if available (preserves [var_xxx] placeholders) @@ -1728,6 +1735,7 @@ def _build_function_call_content(self, context: FunctionInvocationContext) -> Co call_id=self._get_call_id(context), name=context.function.name, arguments=self._current_arguments(context), + id=self._get_approval_id(context), ) def _signature_from_parts(self, name: str | None, arguments: dict[str, Any]) -> str: @@ -1802,16 +1810,17 @@ def _signature_from_function_call(self, function_call: Any) -> str | None: def _response_matches_pending( self, approval_response: Content, + approval_id: str, call_id: str, body_signature: str, ) -> bool: """Validate that the approval response itself corresponds to the pending request. - The response must carry the request id that was shown for review and embed the exact - function call (name + arguments) that was requested. Both the response id and the embedded - function-call id are **required** to be present and equal to the pending ``call_id`` — a - crafted response that omits either identifier (``id=None`` / ``function_call.call_id=None``) - is rejected rather than allowed to skip the binding. + The response must carry the request id shown for review and embed the exact function call + (name + arguments) reconstructed from the authoritative pending snapshot. The embedded + ``call_id`` must match provider correlation; an occurrence-aware response and embedded call + must also carry the Agent Framework approval id. Legacy direct middleware callers continue + to use ``call_id`` for both identities. """ embedded = getattr(approval_response, "function_call", None) if self._signature_from_function_call(embedded) != body_signature: @@ -1819,7 +1828,12 @@ def _response_matches_pending( # Both identifiers must be present and name the pending request (no None bypass). response_id = getattr(approval_response, "id", None) embedded_call_id = getattr(embedded, "call_id", None) - return response_id == call_id and embedded_call_id == call_id + embedded_occurrence_id = getattr(embedded, "id", None) + return ( + response_id == approval_id + and embedded_call_id == call_id + and (approval_id == call_id or embedded_occurrence_id == approval_id) + ) def _matches_pending_approval( self, @@ -1838,9 +1852,10 @@ def _matches_pending_approval( :meth:`_consume_pending_approval` once the approval actually waves the detected violations. """ call_id = self._get_call_id(context) - if not call_id: + approval_id = self._get_approval_id(context) + if not call_id or not approval_id: return False - pending = self._pending_policy_approvals.get(call_id) + pending = self._pending_policy_approvals.get(approval_id) if pending is None: return False approval_response = context.metadata.get("approval_response") @@ -1854,7 +1869,7 @@ def _matches_pending_approval( # and the invocation about to execute must match every recorded binding dimension, including # the exact set of violations that was disclosed for review. return ( - self._response_matches_pending(approval_response, call_id, pending.body_signature) + self._response_matches_pending(approval_response, approval_id, call_id, pending.body_signature) and self._call_body_signature(context) == pending.body_signature and self._context_label_key(context) == pending.label_key and self._session_key(context) == pending.session_key @@ -1867,7 +1882,7 @@ def _consume_pending_approval(self, context: FunctionInvocationContext) -> None: Idempotent: safe to call for both the integrity and confidentiality checks of a single invocation. """ - self._pending_policy_approvals.pop(self._get_call_id(context), None) + self._pending_policy_approvals.pop(self._get_approval_id(context), None) def _mark_policy_violation_approved( self, @@ -1898,9 +1913,9 @@ def _request_policy_violation_approval( f"APPROVAL REQUESTED: Tool '{context.function.name}' requires user approval " f"due to policy violation(s): {disclosed}." ) - call_id = self._get_call_id(context) - if call_id: - self._pending_policy_approvals[call_id] = self._pending_record(context, violations) + approval_id = self._get_approval_id(context) + if approval_id: + self._pending_policy_approvals[approval_id] = self._pending_record(context, violations) additional_properties: dict[str, Any] = { "policy_violation": True, "violation_type": primary["violation_type"], @@ -1917,7 +1932,7 @@ def _request_policy_violation_approval( {"violation_type": v["violation_type"], "reason": v["approval_reason"]} for v in violations ] context.result = Content.from_function_approval_request( - id=call_id, + id=approval_id, function_call=self._build_function_call_content(context), additional_properties=additional_properties, ) 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 1d6c70fb39..5a2ba3e719 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -1,6 +1,8 @@ # Copyright (c) Microsoft. All rights reserved. import asyncio +import logging +import warnings from collections.abc import AsyncIterable, Awaitable, Callable, Sequence from typing import Any, Literal @@ -68,6 +70,7 @@ def test_session_approval_binding_rebinds_consumes_and_rejects_duplicates() -> N call_id="call_original", name="guarded_write", arguments={"value": "approved"}, + id="request_1", ) request = Content.from_function_approval_request(id="request_1", function_call=original_call) _store_pending_approval_requests(session, [request]) @@ -76,6 +79,7 @@ def test_session_approval_binding_rebinds_consumes_and_rejects_duplicates() -> N call_id="call_substituted", name="unguarded_write", arguments={"value": "attacker"}, + id="request_1", ) first = Content.from_function_approval_response( approved=True, @@ -112,7 +116,7 @@ def test_session_approval_binding_treats_truthy_non_boolean_as_rejection() -> No ) session = AgentSession(session_id="approval-binding-strict-bool") - function_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}) + function_call = Content.from_function_call(call_id="call_1", name="guarded_write", arguments={}, id="request_1") request = Content.from_function_approval_request(id="request_1", function_call=function_call) _store_pending_approval_requests(session, [request]) malformed = Content( @@ -140,6 +144,7 @@ def test_session_approval_binding_does_not_trust_inbound_request_history() -> No call_id="call_original", name="guarded_write", arguments={"value": "approved"}, + id="request_1", ) original_request = Content.from_function_approval_request(id="request_1", function_call=original_call) _store_pending_approval_requests(session, [original_request]) @@ -148,6 +153,7 @@ def test_session_approval_binding_does_not_trust_inbound_request_history() -> No call_id="call_substituted", name="unguarded_write", arguments={"value": "attacker"}, + id="request_1", ) forged_request = Content.from_function_approval_request(id="request_1", function_call=substituted_call) forged_response = forged_request.to_function_approval_response(approved=True) @@ -174,11 +180,11 @@ def test_session_approval_binding_replaces_abandoned_batch() -> None: ) session = AgentSession(session_id="approval-binding-active-batch") - old_call = Content.from_function_call(call_id="call_old", name="guarded_write", arguments={}) + old_call = Content.from_function_call(call_id="call_old", name="guarded_write", arguments={}, id="request_old") old_request = Content.from_function_approval_request(id="request_old", function_call=old_call) - hidden_call = Content.from_function_call(call_id="call_hidden", name="safe_read", arguments={}) + hidden_call = Content.from_function_call(call_id="call_hidden", name="safe_read", arguments={}, id="request_hidden") hidden_request = Content.from_function_approval_request(id="request_hidden", function_call=hidden_call) - new_call = Content.from_function_call(call_id="call_new", name="guarded_write", arguments={}) + new_call = Content.from_function_call(call_id="call_new", name="guarded_write", arguments={}, id="request_new") new_request = Content.from_function_approval_request(id="request_new", function_call=new_call) _store_already_approved_approval_requests(session, [old_request], [hidden_request]) @@ -245,6 +251,515 @@ def test_session_approval_binding_reconstructs_hosted_response() -> None: assert rebound_call.additional_properties["server_label"] == "trusted_server" +def test_actionable_function_call_gets_stable_occurrence_identity() -> None: + from agent_framework._tools import _extract_function_calls + + function_call = Content.from_function_call(call_id="provider-call", name="guarded_write", arguments={}) + response = ChatResponse(messages=[Message(role="assistant", contents=[function_call])]) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + first = _extract_function_calls(response) + second = _extract_function_calls(response) + + assert len(first) == 1 + assert first[0].id + assert second[0].id == first[0].id + assert function_call.call_id == "provider-call" + assert caught == [] + + +def test_extract_function_calls_preserves_distinct_occurrences_with_reused_call_id() -> None: + from agent_framework._tools import _extract_function_calls + + calls = [ + Content.from_function_call(call_id="provider-reused", name="first", arguments={}, id="occurrence-1"), + Content.from_function_call(call_id="provider-reused", name="second", arguments={}, id="occurrence-2"), + ] + response = ChatResponse(messages=[Message(role="assistant", contents=calls)]) + + assert _extract_function_calls(response) == calls + + +def test_extract_function_calls_keeps_later_occurrence_after_reused_call_id_result() -> None: + from agent_framework._tools import _extract_function_calls + + completed_call = Content.from_function_call( + call_id="provider-reused", + name="first", + arguments={}, + id="occurrence-1", + ) + later_call = Content.from_function_call( + call_id="provider-reused", + name="second", + arguments={}, + id="occurrence-2", + ) + response = ChatResponse( + messages=[ + Message(role="assistant", contents=[completed_call]), + Message(role="tool", contents=[Content.from_function_result(call_id="provider-reused", result="done")]), + Message(role="assistant", contents=[later_call]), + ] + ) + + assert _extract_function_calls(response) == [later_call] + + +def test_actionable_function_call_uses_occurrence_identity_for_empty_call_id() -> None: + from agent_framework._tools import _extract_function_calls + + function_call = Content.from_function_call(call_id="", name="guarded_write", arguments={}) + response = ChatResponse(messages=[Message(role="assistant", contents=[function_call])]) + + with pytest.warns(FutureWarning, match="empty.*call_id.*Content.id"): + extracted = _extract_function_calls(response) + + assert extracted == [function_call] + assert function_call.id + assert function_call.call_id == function_call.id + + +async def test_streaming_empty_call_id_keeps_occurrence_identity_through_approval( + chat_client_base: SupportsChatGetResponse, +) -> None: + @tool(name="guarded_write", approval_mode="always_require") + def guarded_write() -> str: + return "done" + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="", name="guarded_write", arguments={})], + role="assistant", + ) + ] + ] + streamed_calls: list[tuple[str | None, str | None]] = [] + approval_requests: list[Content] = [] + + with pytest.warns(FutureWarning, match="empty.*call_id.*Content.id"): + async for update in chat_client_base.get_response( + [Message(role="user", contents=["hello"])], + options={"tool_choice": "auto", "tools": [guarded_write]}, + stream=True, + ): + for content in update.contents: + if content.type == "function_call": + streamed_calls.append((content.id, content.call_id)) + elif content.type == "function_approval_request": + approval_requests.append(content) + + assert len(streamed_calls) == 1 + occurrence_id, streamed_call_id = streamed_calls[0] + assert occurrence_id + assert streamed_call_id == occurrence_id + assert len(approval_requests) == 1 + assert approval_requests[0].id == occurrence_id + assert approval_requests[0].function_call is not None + assert approval_requests[0].function_call.id == occurrence_id + assert approval_requests[0].function_call.call_id == occurrence_id + + +async def test_streaming_empty_call_id_delta_reuses_opening_call_identity( + chat_client_base: SupportsChatGetResponse, +) -> None: + @tool(name="guarded_write", approval_mode="always_require") + def guarded_write(value: str) -> str: + return value + + opening = Content.from_function_call( + call_id="provider-call", + name="guarded_write", + arguments='{"value":', + additional_properties={"tool_call_index": 0}, + ) + continuation = Content.from_function_call( + call_id="", + name="", + arguments='"done"}', + additional_properties={"tool_call_index": 0}, + ) + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate(contents=[opening], role="assistant"), + ChatResponseUpdate(contents=[continuation], role="assistant"), + ] + ] + streamed_calls: list[tuple[str | None, str | None]] = [] + approval_requests: list[Content] = [] + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + async for update in chat_client_base.get_response( + [Message(role="user", contents=["hello"])], + options={"tool_choice": "auto", "tools": [guarded_write]}, + stream=True, + ): + for content in update.contents: + if content.type == "function_call": + streamed_calls.append((content.id, content.call_id)) + elif content.type == "function_approval_request": + approval_requests.append(content) + + assert len(streamed_calls) == 2 + assert streamed_calls[0][0] + assert streamed_calls[0] == streamed_calls[1] + assert streamed_calls[0][1] == "provider-call" + assert len(approval_requests) == 1 + assert approval_requests[0].id == streamed_calls[0][0] + assert approval_requests[0].function_call is not None + assert approval_requests[0].function_call.call_id == "provider-call" + assert caught == [] + + +async def test_streaming_named_calls_with_reused_call_id_get_distinct_occurrences( + chat_client_base: SupportsChatGetResponse, +) -> None: + @tool(name="first_write", approval_mode="always_require") + def first_write() -> str: + return "first" + + @tool(name="second_write", approval_mode="always_require") + def second_write() -> str: + return "second" + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="provider-reused", name="first_write", arguments={})], + role="assistant", + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="provider-reused", name="second_write", arguments={})], + role="assistant", + ), + ] + ] + + stream = chat_client_base.get_response( + [Message(role="user", contents=["hello"])], + options={"tool_choice": "auto", "tools": [first_write, second_write]}, + stream=True, + ) + streamed_calls = [ + content async for update in stream for content in update.contents if content.type == "function_call" + ] + final_response = await stream.get_final_response() + final_calls = [ + content + for message in final_response.messages + for content in message.contents + if content.type == "function_call" + ] + + assert [call.name for call in streamed_calls] == ["first_write", "second_write"] + assert all(call.call_id == "provider-reused" for call in streamed_calls) + assert all(call.id for call in streamed_calls) + assert streamed_calls[0].id != streamed_calls[1].id + assert [(call.id, call.name) for call in final_calls] == [ + (streamed_calls[0].id, "first_write"), + (streamed_calls[1].id, "second_write"), + ] + + +async def test_streaming_interleaved_indexed_call_fragments_coalesce_by_occurrence( + chat_client_base: SupportsChatGetResponse, +) -> None: + @tool(name="first_write", approval_mode="always_require") + def first_write(value: str) -> str: + return value + + @tool(name="second_write", approval_mode="always_require") + def second_write(value: str) -> str: + return value + + def fragment(call_id: str, name: str, arguments: str, index: int) -> Content: + return Content.from_function_call( + call_id=call_id, + name=name, + arguments=arguments, + id=f"occurrence-{index}", + ) + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate(contents=[fragment("provider-a", "first_write", '{"value":', 0)], role="assistant"), + ChatResponseUpdate(contents=[fragment("provider-b", "second_write", '{"value":', 1)], role="assistant"), + ChatResponseUpdate(contents=[fragment("provider-a", "", '"first"}', 0)], role="assistant"), + ChatResponseUpdate(contents=[fragment("provider-b", "", '"second"}', 1)], role="assistant"), + ] + ] + streamed_by_index: dict[int, list[tuple[str | None, str | None]]] = {0: [], 1: []} + approval_requests: list[Content] = [] + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + stream = chat_client_base.get_response( + [Message(role="user", contents=["hello"])], + options={"tool_choice": "auto", "tools": [first_write, second_write]}, + stream=True, + ) + async for update in stream: + for content in update.contents: + if content.type == "function_call": + assert content.id is not None + index = int(content.id.removeprefix("occurrence-")) + streamed_by_index[index].append((content.id, content.call_id)) + elif content.type == "function_approval_request": + approval_requests.append(content) + + final_response = await stream.get_final_response() + final_calls = [ + content + for message in final_response.messages + for content in message.contents + if content.type == "function_call" + ] + assert [(call.id, call.call_id, call.parse_arguments()) for call in final_calls] == [ + ("occurrence-0", "provider-a", {"value": "first"}), + ("occurrence-1", "provider-b", {"value": "second"}), + ] + + assert len(approval_requests) == 2 + requests_by_call_id = { + request.function_call.call_id: request for request in approval_requests if request.function_call is not None + } + assert set(requests_by_call_id) == {"provider-a", "provider-b"} + assert requests_by_call_id["provider-a"].function_call is not None + assert requests_by_call_id["provider-a"].function_call.parse_arguments() == {"value": "first"} + assert requests_by_call_id["provider-b"].function_call is not None + assert requests_by_call_id["provider-b"].function_call.parse_arguments() == {"value": "second"} + for index, provider_call_id in ((0, "provider-a"), (1, "provider-b")): + assert len(streamed_by_index[index]) == 2 + assert streamed_by_index[index][0] == streamed_by_index[index][1] + assert streamed_by_index[index][0][1] == provider_call_id + assert caught == [] + + +def test_occurrence_aware_approval_rejects_stale_reused_call_id_response(caplog: pytest.LogCaptureFixture) -> None: + from agent_framework._tools import ( + _bind_approval_responses_to_pending_requests, + _load_pending_approval_requests, + _store_pending_approval_requests, + ) + + session = AgentSession(session_id="approval-binding-reused-provider-id") + first_call = Content( + "function_call", id="af-call-first", call_id="provider-reused", name="guarded_write", arguments={"value": 1} + ) + second_call = Content( + "function_call", id="af-call-second", call_id="provider-reused", name="guarded_write", arguments={"value": 2} + ) + first_request = Content.from_function_approval_request(id="af-call-first", function_call=first_call) + second_request = Content.from_function_approval_request(id="af-call-second", function_call=second_call) + _store_pending_approval_requests(session, [first_request]) + _store_pending_approval_requests(session, [second_request]) + stale_response = Content.from_function_approval_response( + approved=True, id="af-call-first", function_call=first_call + ) + messages = [Message(role="user", contents=[stale_response])] + + with caplog.at_level(logging.WARNING, logger="agent_framework"): + _bind_approval_responses_to_pending_requests(messages, session) + + assert messages == [] + assert list(_load_pending_approval_requests(session)) == ["af-call-second"] + assert "occurrence identity" in caplog.text + + +@pytest.mark.parametrize("response_id", [None, "provider-reused", "af-call-other"]) +def test_occurrence_aware_approval_mismatched_identity_does_not_consume_pending(response_id: str | None) -> None: + from agent_framework._tools import ( + _bind_approval_responses_to_pending_requests, + _load_pending_approval_requests, + _store_pending_approval_requests, + ) + + session = AgentSession(session_id="approval-binding-occurrence-mismatch") + function_call = Content( + "function_call", id="af-call-current", call_id="provider-reused", name="guarded_write", arguments={} + ) + request = Content.from_function_approval_request(id="af-call-current", function_call=function_call) + _store_pending_approval_requests(session, [request]) + response = Content( + "function_approval_response", + approved=True, + id=response_id, + function_call=Content.from_function_call(call_id="provider-reused", name="guarded_write", arguments={}), + ) + messages = [Message(role="user", contents=[response])] + + _bind_approval_responses_to_pending_requests(messages, session) + + assert messages == [] + assert list(_load_pending_approval_requests(session)) == ["af-call-current"] + + +def test_occurrence_aware_approval_can_retry_after_mismatched_identity() -> None: + from agent_framework._tools import ( + _bind_approval_responses_to_pending_requests, + _load_pending_approval_requests, + _store_pending_approval_requests, + ) + + session = AgentSession(session_id="approval-binding-corrected-retry") + function_call = Content.from_function_call( + call_id="provider-call", + name="guarded_write", + arguments={"value": "trusted"}, + id="af-call-current", + ) + request = Content.from_function_approval_request(id="af-call-current", function_call=function_call) + _store_pending_approval_requests(session, [request]) + mismatched_messages = [ + Message( + role="user", + contents=[Content("function_approval_response", approved=True, id="af-call-stale")], + ) + ] + + _bind_approval_responses_to_pending_requests(mismatched_messages, session) + + assert mismatched_messages == [] + assert list(_load_pending_approval_requests(session)) == ["af-call-current"] + + corrected_messages = [ + Message( + role="user", + contents=[Content("function_approval_response", approved=True, id="af-call-current")], + ) + ] + _bind_approval_responses_to_pending_requests(corrected_messages, session) + + assert len(corrected_messages) == 1 + assert corrected_messages[0].contents[0].id == "af-call-current" + assert corrected_messages[0].contents[0].function_call is not None + assert corrected_messages[0].contents[0].function_call.call_id == "provider-call" + assert _load_pending_approval_requests(session) == {} + + +def test_occurrence_aware_approval_binds_without_embedded_function_call() -> None: + from agent_framework._tools import ( + _bind_approval_responses_to_pending_requests, + _load_pending_approval_requests, + _store_pending_approval_requests, + ) + + session = AgentSession(session_id="approval-binding-occurrence-only") + function_call = Content( + "function_call", + id="af-call-current", + call_id="provider-call", + name="guarded_write", + arguments={"value": "trusted"}, + ) + request = Content.from_function_approval_request(id="af-call-current", function_call=function_call) + _store_pending_approval_requests(session, [request]) + response = Content("function_approval_response", approved=True, id="af-call-current") + messages = [Message(role="user", contents=[response])] + + _bind_approval_responses_to_pending_requests(messages, session) + + rebound = messages[0].contents[0] + assert rebound.function_call == function_call + assert _load_pending_approval_requests(session) == {} + + +def test_occurrence_aware_legacy_request_id_rebounds_to_occurrence_id() -> None: + from agent_framework._tools import ( + _bind_approval_responses_to_pending_requests, + _store_pending_approval_requests, + ) + + session = AgentSession(session_id="approval-binding-legacy-request-id") + function_call = Content.from_function_call( + call_id="provider-call", + name="guarded_write", + arguments={"value": "trusted"}, + id="af-call-current", + ) + with pytest.warns(FutureWarning, match="id differs from function_call.id.*legacy"): + request = Content.from_function_approval_request(id="provider-call", function_call=function_call) + _store_pending_approval_requests(session, [request]) + response = Content.from_function_approval_response( + approved=True, + id="provider-call", + function_call=Content.from_dict(function_call.to_dict()), + ) + messages = [Message(role="user", contents=[response])] + + with pytest.warns(FutureWarning, match="legacy provider call_id request binding"): + _bind_approval_responses_to_pending_requests(messages, session) + + assert len(messages) == 1 + assert len(messages[0].contents) == 1 + assert messages[0].contents[0].id == "af-call-current" + assert messages[0].contents[0].function_call is not None + assert messages[0].contents[0].function_call.call_id == "provider-call" + + +def test_legacy_serialized_pending_approval_resumes_once_with_migration_warning() -> None: + from agent_framework._tools import _bind_approval_responses_to_pending_requests, _load_pending_approval_requests + + session = AgentSession(session_id="approval-binding-legacy") + session.state["tool_approval"] = { + "pending_approval_requests": [ + { + "type": "function_approval_request", + "id": "legacy-call", + "function_call": { + "type": "function_call", + "call_id": "legacy-call", + "name": "guarded_write", + "arguments": {"value": "stored"}, + }, + "user_input_request": True, + } + ] + } + response = Content.from_function_approval_response( + approved=True, + id="legacy-call", + function_call=Content.from_function_call( + call_id="legacy-call", name="guarded_write", arguments={"value": "client"} + ), + ) + messages = [Message(role="user", contents=[response])] + + with pytest.warns(FutureWarning, match="legacy stored approval.*Content.id"): + _bind_approval_responses_to_pending_requests(messages, session) + + rebound = messages[0].contents[0] + assert rebound.function_call is not None + assert rebound.function_call.id is None + assert rebound.function_call.parse_arguments() == {"value": "stored"} + assert _load_pending_approval_requests(session) == {} + + +def test_hosted_approval_keeps_provider_issued_request_id() -> None: + from agent_framework._tools import _bind_approval_responses_to_pending_requests, _store_pending_approval_requests + + session = AgentSession(session_id="approval-binding-hosted-id") + hosted_call = Content( + "function_call", + id="af-call-local-occurrence", + call_id="hosted-call", + name="hosted_search", + arguments={}, + additional_properties={"server_label": "hosted"}, + ) + request = Content.from_function_approval_request(id="provider-approval-id", function_call=hosted_call) + _store_pending_approval_requests(session, [request]) + response = Content("function_approval_response", approved=True, id="provider-approval-id") + messages = [Message(role="user", contents=[response])] + + _bind_approval_responses_to_pending_requests(messages, session) + + assert messages[0].contents[0].id == "provider-approval-id" + + def test_session_approval_batch_rejects_duplicate_request_ids() -> None: """Ambiguous request IDs in one provider batch must not overwrite authority.""" from agent_framework._tools import _store_pending_approval_requests @@ -636,7 +1151,7 @@ def ai_func(arg1: str) -> str: role="assistant", ), ChatResponseUpdate( - contents=[Content.from_function_call(call_id="1", name="test_function", arguments='"value1"}')], + contents=[Content.from_function_call(call_id="1", name="", arguments='"value1"}')], role="assistant", ), ], @@ -933,7 +1448,7 @@ def func_with_approval(arg1: str) -> str: role="assistant", ), ChatResponseUpdate( - contents=[Content.from_function_call(call_id="1", name=function_name, arguments='"value1"}')], + contents=[Content.from_function_call(call_id="1", name="", arguments='"value1"}')], role="assistant", ), ] @@ -1150,7 +1665,7 @@ def func_no_approval(arg1: str) -> str: return f"Processed {arg1}" informational_call = Content.from_function_call( - call_id="1", + call_id="", name="no_approval_func", arguments='{"arg1": "value1"}', informational_only=True, @@ -1159,18 +1674,23 @@ def func_no_approval(arg1: str) -> str: [ChatResponseUpdate(contents=[informational_call], role="assistant")] ] - updates = [ - update - async for update in chat_client_base.get_response( - [Message(role="user", contents=["hello"])], - options={"tool_choice": "auto", "tools": [func_no_approval]}, - stream=True, - ) - ] + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + updates = [ + update + async for update in chat_client_base.get_response( + [Message(role="user", contents=["hello"])], + options={"tool_choice": "auto", "tools": [func_no_approval]}, + stream=True, + ) + ] assert exec_counter == 0 assert len(updates) == 1 assert updates[0].contents == [informational_call] + assert informational_call.id is None + assert informational_call.call_id == "" + assert caught == [] async def test_rejected_approval(chat_client_base: SupportsChatGetResponse): @@ -3612,13 +4132,13 @@ async def test_streaming_declaration_only_tool_preserves_metadata_without_duplic contents=[ Content.from_function_call( call_id="call_weather", - name="get_weather", + name="get_weather" if index == 0 else "", arguments=arguments, ) ], role="assistant", ) - for arguments in argument_chunks + for index, arguments in enumerate(argument_chunks) ] ] @@ -3627,10 +4147,14 @@ async def test_streaming_declaration_only_tool_preserves_metadata_without_duplic options={"tool_choice": "auto", "tools": [declaration_tool]}, stream=True, ) + streamed_occurrence_ids: list[str | None] = [] metadata_updates: list[tuple[Any, str | None]] = [] async for update in stream: for content in update.contents: - if content.type == "function_call" and content.call_id == "call_weather" and content.user_input_request: + if content.type != "function_call" or content.call_id != "call_weather": + continue + streamed_occurrence_ids.append(content.id) + if content.user_input_request: metadata_updates.append((content.arguments, content.id)) final_response = await stream.get_final_response() function_calls = [ @@ -3640,11 +4164,15 @@ async def test_streaming_declaration_only_tool_preserves_metadata_without_duplic if content.type == "function_call" and content.call_id == "call_weather" ] - assert metadata_updates == [(None, "call_weather")] + assert len(metadata_updates) == 1 + assert metadata_updates[0][0] is None + assert metadata_updates[0][1] + assert streamed_occurrence_ids + assert set(streamed_occurrence_ids) == {metadata_updates[0][1]} assert len(function_calls) == 1 assert function_calls[0].arguments == '{"location":"Seattle"}' assert function_calls[0].user_input_request is True - assert function_calls[0].id == "call_weather" + assert function_calls[0].id == metadata_updates[0][1] async def test_multiple_function_calls_parallel_execution(chat_client_base: SupportsChatGetResponse): @@ -3911,7 +4439,7 @@ def ai_func(arg1: str) -> str: role="assistant", ), ChatResponseUpdate( - contents=[Content.from_function_call(call_id="1", name="test_function", arguments='"value1"}')], + contents=[Content.from_function_call(call_id="1", name="", arguments='"value1"}')], role="assistant", ), ], @@ -3921,7 +4449,7 @@ def ai_func(arg1: str) -> str: role="assistant", ), ChatResponseUpdate( - contents=[Content.from_function_call(call_id="2", name="test_function", arguments='"value2"}')], + contents=[Content.from_function_call(call_id="2", name="", arguments='"value2"}')], role="assistant", ), ], @@ -3967,7 +4495,7 @@ def ai_func(arg1: str) -> str: role="assistant", ), ChatResponseUpdate( - contents=[Content.from_function_call(call_id="call_1", name="test_function", arguments='"v1"}')], + contents=[Content.from_function_call(call_id="call_1", name="", arguments='"v1"}')], role="assistant", ), ], diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 23902d17b9..59aaa30ae1 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -2,6 +2,7 @@ import base64 import json +import warnings from collections.abc import AsyncIterable, Sequence from dataclasses import dataclass from datetime import datetime, timezone @@ -594,6 +595,12 @@ def test_function_call_content_add_merging_and_errors(): with raises(ContentError): _ = a + b + # incompatible occurrence ids + a = Content.from_function_call(call_id="1", name="f", arguments="abc", id="occurrence-a") + b = Content.from_function_call(call_id="1", name="f", arguments="def", id="occurrence-b") + with raises(AdditionItemMismatch, match="different ids"): + _ = a + b + # name merging: when the first chunk has no name (e.g. a streaming delta where # the function name arrives later), the merged content must keep the name from # whichever side provides it, regardless of order. @@ -740,6 +747,67 @@ def test_function_approval_serialization_roundtrip(): # The Content union will need to be handled differently when we fully migrate +def test_function_call_occurrence_id_roundtrips_without_regeneration(): + function_call = Content.from_function_call( + call_id="provider-call", + name="f", + arguments={"x": 1}, + id="af-call-existing", + ) + + restored = Content.from_dict(function_call.to_dict()) + + assert restored.id == "af-call-existing" + assert restored.call_id == "provider-call" + + +def test_local_function_approval_request_warns_for_legacy_occurrence_identity() -> None: + function_call = Content.from_function_call( + call_id="provider-call", + name="f", + id="af-call-occurrence", + ) + + with pytest.warns(FutureWarning, match="id differs from function_call.id.*legacy"): + request = Content.from_function_approval_request(id="provider-call", function_call=function_call) + + assert request.id == "provider-call" + assert request.function_call is function_call + + +def test_hosted_function_approval_request_allows_provider_request_identity_without_warning() -> None: + function_call = Content.from_function_call( + call_id="provider-call", + name="hosted", + id="af-call-occurrence", + additional_properties={"server_label": "provider"}, + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + request = Content.from_function_approval_request( + id="provider-approval-request", + function_call=function_call, + ) + + assert request.id == "provider-approval-request" + assert caught == [] + + +def test_legacy_function_call_deserialization_does_not_generate_an_occurrence_id(): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + restored = Content.from_dict({ + "type": "function_call", + "call_id": "legacy-call", + "name": "f", + "arguments": {}, + }) + + assert restored.id is None + assert caught == [] + + def test_function_approval_request_function_call_none_guard(): """Test that accessing function_call attributes is safe when function_call is None.""" # Construct a Content with type "function_approval_request" but no function_call. diff --git a/python/packages/core/tests/test_security.py b/python/packages/core/tests/test_security.py index 3b9932e910..20cbff84ba 100644 --- a/python/packages/core/tests/test_security.py +++ b/python/packages/core/tests/test_security.py @@ -748,6 +748,7 @@ async def test_policy_violation_approval_preserves_type_through_auto_invoke(self call_id="call-policy-violation", name=mock_function.name, arguments='{"arg": "test"}', + id="af-call-policy-violation", ) with pytest.raises(MiddlewareTermination) as exc_info: @@ -766,7 +767,9 @@ async def test_policy_violation_approval_preserves_type_through_auto_invoke(self "MiddlewareTermination handler must not wrap approval requests in function_result" ) assert result.function_call is not None + assert result.id == "af-call-policy-violation" assert result.function_call.call_id == "call-policy-violation" + assert result.function_call.id == "af-call-policy-violation" assert result.additional_properties["policy_violation"] is True assert result.additional_properties["violation_type"] == "untrusted_context" diff --git a/python/packages/devui/agent_framework_devui/_executor.py b/python/packages/devui/agent_framework_devui/_executor.py index 35fcf8e680..ec38920191 100644 --- a/python/packages/devui/agent_framework_devui/_executor.py +++ b/python/packages/devui/agent_framework_devui/_executor.py @@ -135,7 +135,9 @@ def _track_approval_request(self, event: dict[str, Any]) -> None: request_id = event.get("request_id") fc = event.get("function_call", {}) if isinstance(request_id, str) and request_id: + occurrence_id = fc.get("occurrence_id") self._pending_approvals[request_id] = { + "id": occurrence_id if isinstance(occurrence_id, str) and occurrence_id else request_id, "call_id": fc.get("id", ""), "name": fc.get("name", ""), "arguments": fc.get("arguments", {}), @@ -830,6 +832,7 @@ def _convert_openai_input_to_chat_message(self, input_items: list[Any], Message: name=stored_fc["name"], arguments=stored_fc["arguments"], ) + function_call.id = stored_fc.get("id", request_id) # Create approval response using server-validated data approval_response = Content.from_function_approval_response( diff --git a/python/packages/devui/agent_framework_devui/_mapper.py b/python/packages/devui/agent_framework_devui/_mapper.py index 21cad790cf..f74922c2c4 100644 --- a/python/packages/devui/agent_framework_devui/_mapper.py +++ b/python/packages/devui/agent_framework_devui/_mapper.py @@ -1827,6 +1827,9 @@ async def _map_approval_request_content(self, content: Any, context: dict[str, A "request_id": getattr(content, "id", "unknown"), "function_call": { "id": getattr(content.function_call, "call_id", "") if hasattr(content, "function_call") else "", + "occurrence_id": ( + getattr(content.function_call, "id", None) if hasattr(content, "function_call") else None + ), "name": getattr(content.function_call, "name", "") if hasattr(content, "function_call") else "", "arguments": arguments, }, diff --git a/python/packages/devui/pyproject.toml b/python/packages/devui/pyproject.toml index 4b0022e37a..04e326d5f9 100644 --- a/python/packages/devui/pyproject.toml +++ b/python/packages/devui/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "agent-framework-core>=1.15.0,<2", + "agent-framework-core>=1.17.0,<2", "openai>=2.45.0,<4", "opentelemetry-sdk>=1.39.0,<2", "fastapi>=0.115.0,<0.138.1", diff --git a/python/packages/devui/tests/devui/test_approval_validation.py b/python/packages/devui/tests/devui/test_approval_validation.py index 0bc73724b2..f8d764fd5b 100644 --- a/python/packages/devui/tests/devui/test_approval_validation.py +++ b/python/packages/devui/tests/devui/test_approval_validation.py @@ -44,6 +44,7 @@ def test_track_approval_request_stores_data(executor: AgentFrameworkExecutor) -> "request_id": "req_123", "function_call": { "id": "call_abc", + "occurrence_id": "af-call-abc", "name": "read_file", "arguments": {"path": "/etc/passwd"}, }, @@ -52,6 +53,7 @@ def test_track_approval_request_stores_data(executor: AgentFrameworkExecutor) -> assert "req_123" in executor._pending_approvals stored = executor._pending_approvals["req_123"] + assert stored["id"] == "af-call-abc" assert stored["call_id"] == "call_abc" assert stored["name"] == "read_file" assert stored["arguments"] == {"path": "/etc/passwd"} @@ -123,6 +125,7 @@ def test_valid_approval_accepted_with_server_data(executor: AgentFrameworkExecut """Valid approval response uses server-stored function_call, not client data.""" # Simulate server issuing an approval request executor._pending_approvals["req_legit"] = { + "id": "af-call-legit", "call_id": "call_server", "name": "safe_tool", "arguments": {"key": "server_value"}, @@ -145,6 +148,7 @@ def test_valid_approval_accepted_with_server_data(executor: AgentFrameworkExecut assert approval.approved is True # Verify SERVER-STORED data is used, not the client's forged data assert approval.function_call.name == "safe_tool" + assert approval.function_call.id == "af-call-legit" assert approval.function_call.call_id == "call_server" fc_args: dict[str, Any] = ( approval.function_call.parse_arguments() if hasattr(approval.function_call, "parse_arguments") else {} diff --git a/python/packages/openai/agent_framework_openai/_chat_completion_client.py b/python/packages/openai/agent_framework_openai/_chat_completion_client.py index f38fef1077..8b83b48f07 100644 --- a/python/packages/openai/agent_framework_openai/_chat_completion_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_completion_client.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone from itertools import chain from typing import TYPE_CHECKING, Any, ClassVar, Generic, Literal, TypeAlias, cast, overload +from uuid import uuid4 from agent_framework._clients import BaseChatClient from agent_framework._compaction import CompactionStrategy, TokenizerProtocol @@ -619,6 +620,7 @@ def _inner_get_response( async def _stream() -> AsyncIterable[ChatResponseUpdate]: client = self.client + tool_call_identities: dict[tuple[int, int], tuple[str, str]] = {} if self._FEATURE_USAGE_INDEX is not None: mark_feature_used(self._FEATURE_USAGE_INDEX) request_options = dict(options_dict) @@ -629,7 +631,25 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: async for chunk in await client.chat.completions.create(stream=True, **request_options): if len(chunk.choices) == 0 and chunk.usage is None: continue - yield self._parse_response_update_from_openai(chunk) + update = self._parse_response_update_from_openai(chunk) + for content in update.contents: + if content.type != "function_call": + continue + choice_index = content.additional_properties.get("tool_call_choice_index") + tool_index = content.additional_properties.get("tool_call_index") + if not isinstance(choice_index, int) or not isinstance(tool_index, int): + continue + index_key = (choice_index, tool_index) + identity = tool_call_identities.get(index_key) + if identity is None: + identity = (f"af-call-{uuid4().hex}", content.call_id or "") + occurrence_id, provider_call_id = identity + if content.call_id: + provider_call_id = content.call_id + tool_call_identities[index_key] = (occurrence_id, provider_call_id) + content.id = occurrence_id + content.call_id = provider_call_id + yield update except BadRequestError as ex: if ex.code == "content_filter": raise OpenAIContentFilterException( @@ -995,6 +1015,9 @@ def _parse_tool_calls_from_openai(self, choice: Choice | ChunkChoice) -> list[Co tool_index = getattr(tool, "index", None) if tool_index is not None: fcc.additional_properties["tool_call_index"] = tool_index + choice_index = getattr(choice, "index", None) + if choice_index is not None: + fcc.additional_properties["tool_call_choice_index"] = choice_index resp.append(fcc) # When you enable asynchronous content filtering in Azure OpenAI, you may receive empty deltas diff --git a/python/packages/openai/pyproject.toml b/python/packages/openai/pyproject.toml index b2e086fb47..26aed2adc1 100644 --- a/python/packages/openai/pyproject.toml +++ b/python/packages/openai/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "agent-framework-core>=1.15.0,<2", + "agent-framework-core>=1.17.0,<2", "openai>=2.25.0,<4", ] diff --git a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py b/python/packages/openai/tests/openai/test_openai_chat_completion_client.py index fe628d3e0a..fd499a726d 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_completion_client.py @@ -2356,6 +2356,275 @@ def test_streaming_chunk_with_null_delta_no_tool_calls_parsed( assert not any(c.type == "function_call" for c in update.contents) +def test_streaming_tool_call_preserves_choice_local_index_scope( + openai_unit_test_env: dict[str, str], +) -> None: + from openai.types.chat.chat_completion_chunk import ChatCompletionChunk + + client = OpenAIChatCompletionClient() + chunk = ChatCompletionChunk.model_validate({ + "id": "test-tool-chunk", + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "test-model", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call-a", + "type": "function", + "function": {"name": "first", "arguments": ""}, + } + ] + }, + "finish_reason": None, + }, + { + "index": 1, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call-b", + "type": "function", + "function": {"name": "second", "arguments": ""}, + } + ] + }, + "finish_reason": None, + }, + ], + }) + + update = client._parse_response_update_from_openai(chunk) + function_calls = [content for content in update.contents if content.type == "function_call"] + + assert [content.additional_properties["tool_call_index"] for content in function_calls] == [0, 0] + assert [content.additional_properties["tool_call_choice_index"] for content in function_calls] == [0, 1] + + +async def test_streaming_tool_call_identity_is_request_local_and_scoped_by_choice_index( + openai_unit_test_env: dict[str, str], +) -> None: + from openai.types.chat.chat_completion_chunk import ChatCompletionChunk + + client = RawOpenAIChatCompletionClient() + + def chunks() -> list[ChatCompletionChunk]: + common = { + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "test-model", + } + return [ + ChatCompletionChunk.model_validate({ + **common, + "id": "opening", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "reused-provider-id", + "type": "function", + "function": {"name": "first", "arguments": '{"a":'}, + }, + { + "index": 1, + "id": "reused-provider-id", + "type": "function", + "function": {"name": "second", "arguments": '{"b":'}, + }, + ] + }, + "finish_reason": None, + }, + { + "index": 1, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "reused-provider-id", + "type": "function", + "function": {"name": "third", "arguments": '{"c":'}, + } + ] + }, + "finish_reason": None, + }, + ], + }), + ChatCompletionChunk.model_validate({ + **common, + "id": "continuation", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + {"index": 0, "type": "function", "function": {"arguments": "1}"}}, + {"index": 1, "type": "function", "function": {"arguments": "2}"}}, + ] + }, + "finish_reason": None, + }, + { + "index": 1, + "delta": {"tool_calls": [{"index": 0, "type": "function", "function": {"arguments": "3}"}}]}, + "finish_reason": None, + }, + ], + }), + ] + + async def create(**kwargs: Any) -> Any: + async def stream_chunks() -> Any: + for chunk in chunks(): + yield chunk + + return stream_chunks() + + request_occurrence_ids: list[dict[tuple[int, int], str | None]] = [] + with patch.object(client.client.chat.completions, "create", side_effect=create): + for _ in range(2): + response_stream = client._inner_get_response( + messages=[Message(role="user", contents=["test"])], stream=True, options={} + ) + assert isinstance(response_stream, ResponseStream) + calls = [ + content + async for update in response_stream + for content in update.contents + if content.type == "function_call" + ] + by_index: dict[tuple[int, int], list[Content]] = {} + for call in calls: + key = ( + call.additional_properties["tool_call_choice_index"], + call.additional_properties["tool_call_index"], + ) + by_index.setdefault(key, []).append(call) + + assert set(by_index) == {(0, 0), (0, 1), (1, 0)} + assert all(len(fragments) == 2 for fragments in by_index.values()) + assert all( + fragments[0].id == fragments[1].id + and fragments[0].call_id == fragments[1].call_id == "reused-provider-id" + for fragments in by_index.values() + ) + occurrence_ids = {key: fragments[0].id for key, fragments in by_index.items()} + assert len(set(occurrence_ids.values())) == 3 + final_response = await response_stream.get_final_response() + final_calls = [ + content + for message in final_response.messages + for content in message.contents + if content.type == "function_call" + ] + assert [(call.name, call.call_id, call.parse_arguments()) for call in final_calls] == [ + ("first", "reused-provider-id", {"a": 1}), + ("second", "reused-provider-id", {"b": 2}), + ("third", "reused-provider-id", {"c": 3}), + ] + request_occurrence_ids.append(occurrence_ids) + + assert set(request_occurrence_ids[0].values()).isdisjoint(request_occurrence_ids[1].values()) + + +async def test_streaming_tool_call_adopts_late_provider_id_without_changing_occurrence( + openai_unit_test_env: dict[str, str], +) -> None: + from openai.types.chat.chat_completion_chunk import ChatCompletionChunk + + client = RawOpenAIChatCompletionClient() + common = { + "object": "chat.completion.chunk", + "created": 1234567890, + "model": "test-model", + } + chunks = [ + ChatCompletionChunk.model_validate({ + **common, + "id": "opening", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "type": "function", + "function": {"name": "lookup", "arguments": '{"value":'}, + } + ] + }, + "finish_reason": None, + } + ], + }), + ChatCompletionChunk.model_validate({ + **common, + "id": "continuation", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "late-service-id", + "type": "function", + "function": {"arguments": "1}"}, + } + ] + }, + "finish_reason": "tool_calls", + } + ], + }), + ] + + async def create(**kwargs: Any) -> Any: + async def stream_chunks() -> Any: + for chunk in chunks: + yield chunk + + return stream_chunks() + + with patch.object(client.client.chat.completions, "create", side_effect=create): + response_stream = client._inner_get_response( + messages=[Message(role="user", contents=["test"])], stream=True, options={} + ) + assert isinstance(response_stream, ResponseStream) + fragments = [ + content + async for update in response_stream + for content in update.contents + if content.type == "function_call" + ] + + assert len(fragments) == 2 + assert fragments[0].id + assert fragments[0].id == fragments[1].id + assert [fragment.call_id for fragment in fragments] == ["", "late-service-id"] + final_response = await response_stream.get_final_response() + final_calls = [ + content + for message in final_response.messages + for content in message.contents + if content.type == "function_call" + ] + assert [(call.id, call.call_id, call.name, call.parse_arguments()) for call in final_calls] == [ + (fragments[0].id, "late-service-id", "lookup", {"value": 1}) + ] + + # endregion