From 4d2e822d8dbef06056b13e532d75bd91b2f91f8e Mon Sep 17 00:00:00 2001 From: aah20 Date: Mon, 17 Aug 2026 15:06:48 +0300 Subject: [PATCH 1/2] Python: Execute every approved tool in a parallel AG-UI HITL batch AG-UI resume synthesizes one user message per interrupt, and the sanitizer treated leftover always_require siblings as abandoned follow-up. Co-authored-by: Cursor --- .../specs/004-python-function-calling-loop.md | 2 + python/packages/ag-ui/AGENTS.md | 6 +- .../_message_adapters.py | 45 ++++++- .../ag-ui/tests/ag_ui/test_endpoint.py | 126 ++++++++++++++++++ .../tests/ag_ui/test_message_adapters.py | 102 ++++++++++++++ 5 files changed, 276 insertions(+), 5 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 6db7d1a9f00..d632d7e0565 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -491,6 +491,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | AG-UI provider boundary | Completed local approval controls from AG-UI request and snapshot replay are absent from raw chat-client input while deferred and hosted approvals keep their respective in-run/provider paths. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_does_not_forward_resolved_local_approval_control_to_chat_client`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_deferred_provider_tool_executes`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_canonical_resume_preserves_hosted_approval_for_provider`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_removes_duplicate_completed_controls`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_pairs_reused_call_ids_by_occurrence`, `packages/ag-ui/tests/ag_ui/test_run.py::test_canonical_hosted_approval_resume_rejects_edited_arguments_without_mutating_pending` | | AG-UI standard approval payload | Agent and workflow tool approvals emit canonical `tool_call` interrupts. `approved` plus full-replacement `editedArgs` executes once and replays idempotently, while legacy `accepted` plus direct partial edits remains supported. Hosted approvals remain decision-only. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_entry_applies_standard_full_replacement_edited_args`, `test_endpoint_agent_approval_replayed_standard_edited_resume_is_idempotent`, `test_endpoint_agent_approval_resume_entry_applies_edited_arguments`, `test_workflow_endpoint_emits_canonical_tool_approval_interrupt`, `test_workflow_endpoint_accepts_canonical_tool_approval_resume`, `test_workflow_endpoint_applies_canonical_approval_edited_args`, `test_workflow_endpoint_accepts_legacy_partial_approval_edits`, `test_workflow_endpoint_hosted_approval_rejects_argument_edits` | | AG-UI cancellation | A cancelled interrupt executes zero times and completes normally, including an identical retry during retained cancellation state; resolved siblings in the same complete resume still execute once. Workflow cancellation clears both runner correlation and the owning agent executor's pending request so later approvals remain resumable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_cancelled_resume_entry_completes_without_execution`, `test_endpoint_agent_approval_replayed_cancellation_completes_idempotently`, `test_endpoint_agent_approval_mixed_cancelled_and_resolved_resume_executes_resolved_tool`, `test_endpoint_workflow_request_info_cancelled_resume_completes_normally`, `test_workflow_endpoint_cancelled_agent_approval_does_not_block_next_approval` | +| AG-UI parallel gated batch resume | Consecutive approval-control user messages for one assistant batch keep leftover gated call ids open. Every accepted call executes once, mixed reject/accept reaches the resolver without skip injection, and ID-less approval-control messages for distinct call ids are not collapsed. Real follow-up user text still skips leftover siblings. | `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_sanitize_consecutive_approval_controls_keep_gated_siblings`, `test_sanitize_consecutive_approval_controls_keep_mixed_decisions`, `test_sanitize_approval_then_followup_still_skips_leftover_sibling`, `test_deduplicate_idless_approval_controls_for_distinct_calls`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_executes_every_gated_call_in_parallel_batch` | | AG-UI approval retention and capacity | Pending authority expires automatically, indeterminate outcomes remain non-retryable until their safety window permits reclamation, and one trusted scope cannot consume another scope's occurrence quota. | `packages/ag-ui/tests/ag_ui/test_approval_lifecycle.py::test_abandoned_pending_occurrence_expires_and_releases_capacity`, `test_indeterminate_occurrence_is_reclaimed_after_its_safety_window`, `test_capacity_is_enforced_per_trusted_scope` | | AG-UI local executor unavailable on resume | A claimed local occurrence whose executor disappeared releases its unstarted claim, reports temporary unavailability, and remains safely retryable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_remains_retryable_when_local_tool_is_temporarily_unavailable` | | AG-UI forwarded execution interruption | A provider failure, cancellation, or stream close after forwarding an approval recovers the open occurrence as indeterminate when no idempotency key proves retry safety. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_hosted_approval_becomes_indeterminate_when_provider_stream_fails` | @@ -596,6 +597,7 @@ Before accepting an update, reviewers must confirm: - #7267 / #7271 and #7304 — replayed calls and reused ids - #7043 — provider-injected approval execution - #6828 — AG-UI `confirm_changes` snapshot correlation +- #7569 — AG-UI parallel gated approval resume executes every approved call - #7212 — non-adjacent and reused-id compaction integrity - #7125 — service-side approval response serialization - #7045 — post-limit tool-content transcript integrity diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index c1b6c328969..6092ecb269b 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -31,8 +31,10 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. - `Interrupt` and `ResumeEntry` come from the `ag-ui-protocol` package (`ag_ui.core`), not from an Agent Framework-specific interrupt model. - Tool approval interrupts, including approvals surfaced through workflow `request_info`, advertise standard `approved` and full-replacement `editedArgs` responses while retaining the existing `accepted` alias and direct - partial edits for MAF client compatibility. A `cancelled` resume completes normally without executing that call; - resolved siblings in the same complete resume still proceed. + partial edits for MAF client compatibility. A `cancelled` resume completes normally without executing that + call; resolved siblings in the same complete resume still proceed. Consecutive approval-control user messages + for a parallel `always_require` batch keep leftover gated call ids open; they are not treated as follow-up that + skip remaining siblings. - Approval-time execution preserves each call's complete result group. Follow-up user-input requests remain in the resumed messages, while `TOOL_CALL_RESULT` events are emitted only for terminal `function_result` contents. - Approval responses for tools injected during `before_run` are deferred to the in-run approval middleware rather 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 c5a4d04cb35..43e6e2a20c8 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 @@ -75,6 +75,41 @@ def _function_result_call_ids(messages: list[Message]) -> set[str]: return result_ids +def _is_approval_control_user_message(msg: Message) -> bool: + """Whether a user message is only tool-approval controls, not a new turn. + + Resume synthesis emits one user message per interrupt. A later sibling + approval in the same batch must still consume its own call id; it is not a + follow-up that abandons remaining gated calls. + """ + has_approval = False + for content in msg.contents or []: + if content.type == "function_approval_response": + has_approval = True + continue + if content.type == "text" and (content.text or "").strip(): + return False + if content.type not in {"function_approval_response", "text"}: + return False + return has_approval + + +def _approval_control_identity(msg: Message) -> tuple[tuple[str | None, str | None], ...] | None: + """Stable identity for an approval-control user message. + + ID-less approval messages hashing only ``str(Content)`` can collapse two + distinct call occurrences from the same assistant batch. + """ + contents = msg.contents or [] + if not contents or any(content.type != "function_approval_response" for content in contents): + return None + identities: list[tuple[str | None, str | None]] = [] + for content in contents: + call_id = content.function_call.call_id if content.function_call is not None else None + identities.append((content.id, str(call_id) if call_id else None)) + return tuple(identities) + + def _sanitize_tool_history( messages: list[Message], *, @@ -217,7 +252,7 @@ def _sanitize_tool_history( except (json.JSONDecodeError, KeyError) as exc: logger.debug(f"Could not parse user message as confirm_changes response: {type(exc).__name__}") - if pending_tool_call_ids: + if pending_tool_call_ids and not _is_approval_control_user_message(msg): logger.info( f"User message arrived with {len(pending_tool_call_ids)} pending tool calls - " "injecting synthetic results" @@ -342,8 +377,12 @@ def _deduplicate_messages(messages: list[Message]) -> list[Message]: if msg.message_id: key = ("id", msg.message_id) else: - content_str = str([str(c) for c in msg.contents]) if msg.contents else "" - key = ("content", role_value, hash(content_str)) + approval_identity = _approval_control_identity(msg) + if approval_identity is not None: + key = ("approval", approval_identity) + else: + content_str = str([str(c) for c in msg.contents]) if msg.contents else "" + key = ("content", role_value, hash(content_str)) if key in seen_keys: logger.info(f"Skipping duplicate message at index {idx}: role={role_value}") 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 da84d6d2eba..7522bdd3eea 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -2049,6 +2049,63 @@ async def stream_fn( return TestClient(app), executed, messages_received, state +def _build_parallel_gated_approval_batch_endpoint( + streaming_chat_client_stub: Any, + *, + snapshot_store: InMemoryAGUIThreadSnapshotStore | None = None, +) -> tuple[TestClient, list[str], list[Message], dict[str, str]]: + executed: list[str] = [] + messages_received: list[Message] = [] + state = {"phase": "pause"} + + def first_tool() -> str: + executed.append("first") + return "first result" + + def second_tool() -> str: + executed.append("second") + return "second result" + + async def stream_fn( + messages: list[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> AsyncIterator[ChatResponseUpdate]: + del options, kwargs + if state["phase"] == "pause": + yield ChatResponseUpdate( + contents=[ + Content.from_function_call(call_id="call_first", name="first_tool", arguments="{}"), + Content.from_function_call(call_id="call_second", name="second_tool", arguments="{}"), + ], + role="assistant", + ) + return + messages_received[:] = list(messages) + 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="first_tool", description="First tool", func=first_tool, approval_mode="always_require"), + FunctionTool( + name="second_tool", description="Second tool", func=second_tool, approval_mode="always_require" + ), + ], + ) + app = FastAPI() + add_agent_framework_fastapi_endpoint( + app, + AgentFrameworkAgent(agent=agent, require_confirmation=False), + path="/approval", + snapshot_store=snapshot_store, + snapshot_scope_resolver=(lambda _request: "tenant-a") if snapshot_store is not None else None, + ) + return TestClient(app), executed, messages_received, state + + def _build_tool_approval_queue_endpoint( streaming_chat_client_stub: Any, ) -> tuple[TestClient, list[str], list[Message], dict[str, str], AgentFrameworkAgent]: @@ -2286,6 +2343,75 @@ 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_executes_every_gated_call_in_parallel_batch(streaming_chat_client_stub): + """A complete resume for a parallel gated batch must execute every approved call.""" + client, executed, messages_received, state = _build_parallel_gated_approval_batch_endpoint( + streaming_chat_client_stub, + snapshot_store=InMemoryAGUIThreadSnapshotStore(), + ) + + pause_response = client.post( + "/approval", + json={ + "runId": "run-pause", + "threadId": "thread-parallel-gated", + "messages": [{"id": "user-1", "role": "user", "content": "Run both 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"] for interrupt in interrupts] == ["call_first", "call_second"] + assert not [event for event in pause_events if event.get("type") == "TOOL_CALL_RESULT"] + + state["phase"] = "resume" + resume_response = client.post( + "/approval", + json={ + "runId": "run-resume", + "threadId": "thread-parallel-gated", + "messages": [], + "resume": [ + {"interruptId": "call_first", "status": "resolved", "payload": {"accepted": True}}, + {"interruptId": "call_second", "status": "resolved", "payload": {"accepted": True}}, + ], + }, + ) + + assert resume_response.status_code == 200 + resume_events = _decode_sse_events(resume_response) + tool_results = [ + (event["toolCallId"], event["content"]) for event in resume_events if event.get("type") == "TOOL_CALL_RESULT" + ] + assert tool_results == [("call_first", "first result"), ("call_second", "second result")] + assert executed == ["first", "second"] + assert not any("Tool execution skipped" in str(event.get("content")) for event in resume_events) + + hydrate_response = client.post( + "/approval", + json={"runId": "run-hydrate", "threadId": "thread-parallel-gated", "messages": []}, + ) + assert hydrate_response.status_code == 200 + hydrated_messages = _latest_messages_snapshot(hydrate_response) + replayed_results = [ + (message.get("toolCallId"), message.get("content")) + for message in hydrated_messages + if message.get("role") == "tool" and message.get("toolCallId") in {"call_first", "call_second"} + ] + assert replayed_results == [("call_first", "first result"), ("call_second", "second result")] + assert not any("Tool execution skipped" in str(message.get("content")) for message in hydrated_messages) + + replayed_provider_results = [ + content for message in messages_received for content in message.contents if content.type == "function_result" + ] + assert [(content.call_id, content.result) for content in replayed_provider_results] == [ + ("call_first", "first result"), + ("call_second", "second result"), + ] + + 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( diff --git a/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py b/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py index c78bec32930..ecbd0d91001 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py +++ b/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py @@ -1204,9 +1204,111 @@ def test_sanitize_json_confirm_changes_response(): assert len(result) >= 1 +def _approval_control_user_message(*, call_id: str, name: str, approved: bool) -> Message: + function_call = Content.from_function_call(call_id=call_id, name=name, arguments="{}") + return Message( + role="user", + contents=[ + Content.from_function_approval_response( + approved=approved, + id=call_id, + function_call=function_call, + ) + ], + ) + + +def test_sanitize_consecutive_approval_controls_keep_gated_siblings(): + """Consecutive AG-UI approval-control messages must not skip leftover gated calls.""" + from agent_framework_ag_ui._message_adapters import _sanitize_tool_history + + assistant_msg = Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="c1", name="first_tool", arguments="{}"), + Content.from_function_call(call_id="c2", name="second_tool", arguments="{}"), + ], + ) + first_approval = _approval_control_user_message(call_id="c1", name="first_tool", approved=True) + second_approval = _approval_control_user_message(call_id="c2", name="second_tool", approved=True) + + result = _sanitize_tool_history([assistant_msg, first_approval, second_approval]) + + assert [msg.role for msg in result] == ["assistant", "user", "user"] + assert not [msg for msg in result if msg.role == "tool"] + assert not any( + content.type == "function_result" and "skipped" in str(content.result).lower() + for msg in result + for content in (msg.contents or []) + ) + + +def test_sanitize_consecutive_approval_controls_keep_mixed_decisions(): + """Approve-then-reject sibling controls must reach the resolver without skip injection.""" + from agent_framework_ag_ui._message_adapters import _sanitize_tool_history + + assistant_msg = Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="c1", name="first_tool", arguments="{}"), + Content.from_function_call(call_id="c2", name="second_tool", arguments="{}"), + ], + ) + first_approval = _approval_control_user_message(call_id="c1", name="first_tool", approved=True) + second_rejection = _approval_control_user_message(call_id="c2", name="second_tool", approved=False) + + result = _sanitize_tool_history([assistant_msg, first_approval, second_rejection]) + + assert [msg.role for msg in result] == ["assistant", "user", "user"] + assert not [msg for msg in result if msg.role == "tool"] + + +def test_sanitize_approval_then_followup_still_skips_leftover_sibling(): + """Real follow-up user text after one approval still abandons leftover siblings.""" + from agent_framework_ag_ui._message_adapters import _sanitize_tool_history + + assistant_msg = Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="c1", name="first_tool", arguments="{}"), + Content.from_function_call(call_id="c2", name="second_tool", arguments="{}"), + ], + ) + first_approval = _approval_control_user_message(call_id="c1", name="first_tool", approved=True) + followup = Message(role="user", contents=[Content.from_text(text="Actually, never mind")]) + + result = _sanitize_tool_history([assistant_msg, first_approval, followup]) + + tool_results = [msg for msg in result if msg.role == "tool"] + assert len(tool_results) == 1 + assert tool_results[0].contents[0].call_id == "c2" + assert "skipped" in str(tool_results[0].contents[0].result).lower() + assert [msg.role for msg in result] == ["assistant", "user", "tool", "user"] + + # ── Deduplication edge cases ── +def test_deduplicate_idless_approval_controls_for_distinct_calls(): + """ID-less approval-control messages for distinct call ids are preserved.""" + from agent_framework_ag_ui._message_adapters import _deduplicate_messages + + first_approval = _approval_control_user_message(call_id="c1", name="first_tool", approved=True) + second_approval = _approval_control_user_message(call_id="c2", name="second_tool", approved=True) + first_approval.message_id = None + second_approval.message_id = None + + result = _deduplicate_messages([first_approval, second_approval]) + + assert len(result) == 2 + assert [ + content.function_call.call_id + for msg in result + for content in (msg.contents or []) + if content.type == "function_approval_response" and content.function_call is not None + ] == ["c1", "c2"] + + def test_deduplicate_tool_results(): """Duplicate tool results for same call_id are deduplicated.""" from agent_framework_ag_ui._message_adapters import _deduplicate_messages From 81464f0ee5484324f22bcbf734732f600830f2e6 Mon Sep 17 00:00:00 2001 From: aah20 Date: Tue, 18 Aug 2026 09:02:56 +0300 Subject: [PATCH 2/2] Keep later AG-UI reject/edit when approval IDs are reused. Dedup identity now includes the decision and canonical payload so a later reject or argument edit is not collapsed before occurrence-aware matching. Co-authored-by: Cursor --- .../specs/004-python-function-calling-loop.md | 2 +- python/packages/ag-ui/AGENTS.md | 4 +- .../_message_adapters.py | 37 +++++++- .../tests/ag_ui/test_message_adapters.py | 85 ++++++++++++++++++- 4 files changed, 119 insertions(+), 9 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index d632d7e0565..e9b756c725a 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -491,7 +491,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | AG-UI provider boundary | Completed local approval controls from AG-UI request and snapshot replay are absent from raw chat-client input while deferred and hosted approvals keep their respective in-run/provider paths. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_does_not_forward_resolved_local_approval_control_to_chat_client`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_deferred_provider_tool_executes`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_canonical_resume_preserves_hosted_approval_for_provider`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_removes_duplicate_completed_controls`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_pairs_reused_call_ids_by_occurrence`, `packages/ag-ui/tests/ag_ui/test_run.py::test_canonical_hosted_approval_resume_rejects_edited_arguments_without_mutating_pending` | | AG-UI standard approval payload | Agent and workflow tool approvals emit canonical `tool_call` interrupts. `approved` plus full-replacement `editedArgs` executes once and replays idempotently, while legacy `accepted` plus direct partial edits remains supported. Hosted approvals remain decision-only. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_entry_applies_standard_full_replacement_edited_args`, `test_endpoint_agent_approval_replayed_standard_edited_resume_is_idempotent`, `test_endpoint_agent_approval_resume_entry_applies_edited_arguments`, `test_workflow_endpoint_emits_canonical_tool_approval_interrupt`, `test_workflow_endpoint_accepts_canonical_tool_approval_resume`, `test_workflow_endpoint_applies_canonical_approval_edited_args`, `test_workflow_endpoint_accepts_legacy_partial_approval_edits`, `test_workflow_endpoint_hosted_approval_rejects_argument_edits` | | AG-UI cancellation | A cancelled interrupt executes zero times and completes normally, including an identical retry during retained cancellation state; resolved siblings in the same complete resume still execute once. Workflow cancellation clears both runner correlation and the owning agent executor's pending request so later approvals remain resumable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_cancelled_resume_entry_completes_without_execution`, `test_endpoint_agent_approval_replayed_cancellation_completes_idempotently`, `test_endpoint_agent_approval_mixed_cancelled_and_resolved_resume_executes_resolved_tool`, `test_endpoint_workflow_request_info_cancelled_resume_completes_normally`, `test_workflow_endpoint_cancelled_agent_approval_does_not_block_next_approval` | -| AG-UI parallel gated batch resume | Consecutive approval-control user messages for one assistant batch keep leftover gated call ids open. Every accepted call executes once, mixed reject/accept reaches the resolver without skip injection, and ID-less approval-control messages for distinct call ids are not collapsed. Real follow-up user text still skips leftover siblings. | `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_sanitize_consecutive_approval_controls_keep_gated_siblings`, `test_sanitize_consecutive_approval_controls_keep_mixed_decisions`, `test_sanitize_approval_then_followup_still_skips_leftover_sibling`, `test_deduplicate_idless_approval_controls_for_distinct_calls`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_executes_every_gated_call_in_parallel_batch` | +| AG-UI parallel gated batch resume | Consecutive approval-control user messages for one assistant batch keep leftover gated call ids open. Every accepted call executes once, mixed reject/accept reaches the resolver without skip injection, and ID-less approval-control messages for distinct call ids are not collapsed. A later reject or edit that reuses the same approval and call ids is kept until occurrence matching. Real follow-up user text still skips leftover siblings. | `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_sanitize_consecutive_approval_controls_keep_gated_siblings`, `test_sanitize_consecutive_approval_controls_keep_mixed_decisions`, `test_sanitize_approval_then_followup_still_skips_leftover_sibling`, `test_deduplicate_idless_approval_controls_for_distinct_calls`, `test_deduplicate_idless_approval_controls_keep_later_reject_for_reused_ids`, `test_deduplicate_idless_approval_controls_keep_later_edit_for_reused_ids`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_executes_every_gated_call_in_parallel_batch` | | AG-UI approval retention and capacity | Pending authority expires automatically, indeterminate outcomes remain non-retryable until their safety window permits reclamation, and one trusted scope cannot consume another scope's occurrence quota. | `packages/ag-ui/tests/ag_ui/test_approval_lifecycle.py::test_abandoned_pending_occurrence_expires_and_releases_capacity`, `test_indeterminate_occurrence_is_reclaimed_after_its_safety_window`, `test_capacity_is_enforced_per_trusted_scope` | | AG-UI local executor unavailable on resume | A claimed local occurrence whose executor disappeared releases its unstarted claim, reports temporary unavailability, and remains safely retryable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_remains_retryable_when_local_tool_is_temporarily_unavailable` | | AG-UI forwarded execution interruption | A provider failure, cancellation, or stream close after forwarding an approval recovers the open occurrence as indeterminate when no idempotency key proves retry safety. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_hosted_approval_becomes_indeterminate_when_provider_stream_fails` | diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index 6092ecb269b..bd917c3ca64 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -34,7 +34,9 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. partial edits for MAF client compatibility. A `cancelled` resume completes normally without executing that call; resolved siblings in the same complete resume still proceed. Consecutive approval-control user messages for a parallel `always_require` batch keep leftover gated call ids open; they are not treated as follow-up that - skip remaining siblings. + skip remaining siblings. ID-less approval-control messages are keyed by approval + id, call id, decision, and function-call payload so a later reject or edit that + reuses those ids is not collapsed before occurrence matching. - Approval-time execution preserves each call's complete result group. Follow-up user-input requests remain in the resumed messages, while `TOOL_CALL_RESULT` events are emitted only for terminal `function_result` contents. - Approval responses for tools injected during `before_run` are deferred to the in-run approval middleware rather 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 43e6e2a20c8..4c3fad9ff8e 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 @@ -8,6 +8,7 @@ import binascii import json import logging +from collections.abc import Mapping from typing import Any, cast from agent_framework import ( @@ -94,19 +95,47 @@ def _is_approval_control_user_message(msg: Message) -> bool: return has_approval -def _approval_control_identity(msg: Message) -> tuple[tuple[str | None, str | None], ...] | None: +def _function_call_payload_identity(function_call: Content | None) -> tuple[str | None, str]: + """Canonical name plus arguments for one approval-bound function call.""" + if function_call is None: + return None, "" + name = function_call.name + arguments = function_call.arguments + if isinstance(arguments, Mapping): + payload = json.dumps(dict(arguments), sort_keys=True, default=str) + elif arguments is None: + payload = "" + else: + payload = str(arguments) + return (str(name) if name else None, payload) + + +def _approval_control_identity( + msg: Message, +) -> tuple[tuple[str | None, str | None, bool | None, str | None, str], ...] | None: """Stable identity for an approval-control user message. ID-less approval messages hashing only ``str(Content)`` can collapse two - distinct call occurrences from the same assistant batch. + distinct call occurrences from the same assistant batch. Dedup must also + keep a later reject or edit that reuses the same approval and call ids; + ``(content.id, call_id)`` alone would drop it before occurrence matching. """ contents = msg.contents or [] if not contents or any(content.type != "function_approval_response" for content in contents): return None - identities: list[tuple[str | None, str | None]] = [] + identities: list[tuple[str | None, str | None, bool | None, str | None, str]] = [] for content in contents: call_id = content.function_call.call_id if content.function_call is not None else None - identities.append((content.id, str(call_id) if call_id else None)) + name, arguments = _function_call_payload_identity(content.function_call) + identities.append( + ( + content.id, + str(call_id) if call_id else None, + content.approved, + name, + arguments, + ) + ) return tuple(identities) diff --git a/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py b/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py index ecbd0d91001..592cfc5eb17 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py +++ b/python/packages/ag-ui/tests/ag_ui/test_message_adapters.py @@ -1204,14 +1204,21 @@ def test_sanitize_json_confirm_changes_response(): assert len(result) >= 1 -def _approval_control_user_message(*, call_id: str, name: str, approved: bool) -> Message: - function_call = Content.from_function_call(call_id=call_id, name=name, arguments="{}") +def _approval_control_user_message( + *, + call_id: str, + name: str, + approved: bool, + arguments: str = "{}", + approval_id: str | None = None, +) -> Message: + function_call = Content.from_function_call(call_id=call_id, name=name, arguments=arguments) return Message( role="user", contents=[ Content.from_function_approval_response( approved=approved, - id=call_id, + id=approval_id if approval_id is not None else call_id, function_call=function_call, ) ], @@ -1309,6 +1316,78 @@ def test_deduplicate_idless_approval_controls_for_distinct_calls(): ] == ["c1", "c2"] +def test_deduplicate_idless_approval_controls_keep_later_reject_for_reused_ids(): + """A later reject reusing the same approval and call ids must not be dropped.""" + from agent_framework_ag_ui._message_adapters import _deduplicate_messages + + first_approval = _approval_control_user_message( + call_id="c1", + name="gated_tool", + approved=True, + approval_id="approval-1", + ) + later_reject = _approval_control_user_message( + call_id="c1", + name="gated_tool", + approved=False, + approval_id="approval-1", + ) + first_approval.message_id = None + later_reject.message_id = None + + result = _deduplicate_messages([first_approval, later_reject]) + + assert len(result) == 2 + assert [content.approved for msg in result for content in (msg.contents or [])] == [True, False] + + +def test_deduplicate_idless_approval_controls_keep_later_edit_for_reused_ids(): + """A later edit reusing the same approval and call ids must keep the new payload.""" + from agent_framework_ag_ui._message_adapters import _deduplicate_messages + + first_approval = _approval_control_user_message( + call_id="c1", + name="gated_tool", + approved=True, + arguments='{"target":"prod"}', + approval_id="approval-1", + ) + later_edit = _approval_control_user_message( + call_id="c1", + name="gated_tool", + approved=True, + arguments='{"target":"staging"}', + approval_id="approval-1", + ) + first_approval.message_id = None + later_edit.message_id = None + + result = _deduplicate_messages([first_approval, later_edit]) + + assert len(result) == 2 + assert [ + content.function_call.arguments + for msg in result + for content in (msg.contents or []) + if content.type == "function_approval_response" and content.function_call is not None + ] == ['{"target":"prod"}', '{"target":"staging"}'] + + +def test_deduplicate_idless_approval_controls_collapse_identical_replays(): + """Identical ID-less approval-control replays still collapse.""" + from agent_framework_ag_ui._message_adapters import _deduplicate_messages + + first_approval = _approval_control_user_message(call_id="c1", name="gated_tool", approved=True) + replay = _approval_control_user_message(call_id="c1", name="gated_tool", approved=True) + first_approval.message_id = None + replay.message_id = None + + result = _deduplicate_messages([first_approval, replay]) + + assert len(result) == 1 + assert result[0] is first_approval + + def test_deduplicate_tool_results(): """Duplicate tool results for same call_id are deduplicated.""" from agent_framework_ag_ui._message_adapters import _deduplicate_messages