Python: Add approval storage and improve truth checks

docs/specs/004-python-function-calling-loop.md

@@ -342,6 +342,21 @@ that manually replay messages own the equivalent rule: do not resend an approval
 ### Approval request and resume
 
 - 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.
+- 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.
+- 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.
+- Direct chat-client invocation without an `AgentSession` preserves pass-through compatibility, matching .NET;
+  authorization sinks still require strict `True`.
 - An approved tool executes exactly once.
 - A rejected tool executes zero times and produces one synthetic rejection `function_result` using the original
   function `call_id`.
@@ -421,12 +436,17 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Rejected streaming resume | Rejection result update precedes final text and tool executes zero times. | `test_approval_resume_returns_result_without_mutating_inputs[streaming-rejected]`, `test_streaming_approval_resume_yields_terminal_result_before_model_text[rejected]` |
 | Mixed approved/rejected batch | Every call gets one correctly correlated terminal result. | `packages/core/tests/core/test_function_invocation_logic.py::test_rejected_approval` |
 | Persisted approval replay | Resume executes with the prior call available. | `test_persisted_approval_messages_replay_correctly` |
-| Hosted approval pass-through | Hosted requests/responses are not processed as local calls. | `test_hosted_tool_approval_response`, `test_hosted_mcp_approval_response_passthrough`, `test_mixed_local_and_hosted_approval_flow` |
+| Hosted approval pass-through | Hosted requests/responses are bound to the recorded provider request and are not processed as local calls. | `test_hosted_tool_approval_response`, `test_hosted_mcp_approval_response_passthrough`, `test_session_approval_binding_reconstructs_hosted_response`, `test_mixed_local_and_hosted_approval_flow` |
 | Approval-time user input | Every user-input request from one approved execution returns in order with assistant role and no extra model call; the execution consumes one call-budget unit. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_returns_all_user_input_requests_without_another_model_call`, `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_user_input_counts_toward_function_call_budget` |
 | Mixed terminal result and follow-up input | Completed siblings remain tool-role while only follow-up input requests use assistant-role messages/updates. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_separates_terminal_results_from_follow_up_requests`, `packages/openai/tests/openai/test_openai_chat_completion_client.py::test_mixed_approval_resume_roles_serialize_function_result_as_tool` |
 | Approval-time middleware termination | Terminal result returns with no extra model call in either response mode. | `packages/core/tests/core/test_function_invocation_logic.py::test_approval_resume_honors_middleware_termination` |
 | 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` |
+| 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` |
+| Tool registry changes | Same-name upgrades may execute the recorded operation; removing the recorded name executes nothing. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_allows_same_name_tool_upgrade`, `test_approval_resume_does_not_execute_when_recorded_tool_disappears` |
 
 ### Approval correlation and replay
 
@@ -445,6 +465,8 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Missing result call id | A malformed result does not steal another approval's result. | `test_replace_approval_contents_with_results_skips_results_without_call_id` |
 | Empty approval message cleanup | Fully consumed approval messages are removed from normalized model input. | `test_replace_approval_contents_with_results_prunes_emptied_messages` |
 | Later stateless turn | A prior terminal approval response cannot execute again. | `test_resolved_approval_response_is_inert_on_later_stateless_turn` |
+| Unbound or duplicate response | A response with no pending session request is removed; one request authorizes at most one response. | `test_session_approval_binding_rebinds_consumes_and_rejects_duplicates` |
+| Forged inbound request history | A caller-supplied request wrapper cannot replace the server snapshot or resurrect consumed authority. | `test_session_approval_binding_does_not_trust_inbound_request_history` |
 | Pending history turn | An unresolved approval batch is omitted atomically from unrelated model input while a later decision can still resume it once. | `packages/core/tests/core/test_harness_tool_approval.py::test_pending_approval_from_file_history_stays_resumable_without_model_orphan` |
 | Duplicate function-call prevention | Approval normalization does not create a second call for one round. | `test_no_duplicate_function_calls_after_approval_processing` |
 | Rejection call id | Rejection result uses the function call id, not only the approval id. | `test_rejection_result_uses_function_call_id` |
@@ -462,6 +484,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Auto-approval callback | Callback receives the original function call and executes the approved set once. | `test_tool_approval_middleware_auto_approval_rule_receives_function_call` |
 | Shared call budget | Auto-approved re-entry does not reset `max_function_calls`, and every executed approval group counts even when it pauses for input. | `test_tool_approval_middleware_auto_approved_loops_share_function_call_budget`, `test_approval_resume_user_input_counts_toward_function_call_budget` |
 | Standing tool rule | Tool-level approval applies only to later matching tools. | `test_tool_approval_middleware_always_approve_tool_rule` |
+| Forged standing rule | An unbound or substituted hosted response cannot create a standing middleware approval rule for caller-selected metadata. | `test_tool_approval_middleware_drops_forged_standing_approval`, `test_tool_approval_middleware_rebinds_hosted_standing_approval` |
 | Hosted server boundary | Standing approval does not cross `server_label`. | `test_tool_approval_middleware_standing_rules_include_hosted_server_boundary` |
 | Argument-scoped rule | Exact arguments are required; empty arguments are not tool-wide. | `test_tool_approval_middleware_always_approve_tool_with_arguments_rule`, `test_tool_approval_middleware_empty_arguments_rule_is_not_tool_wide` |
 | Provider-injected approval tool | A tool added during `before_run` defers to in-run resolution, executes once, and emits one result. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_deferred_provider_tool_executes` |
@@ -544,6 +567,7 @@ uv run poe syntax -P openai
 uv run poe pyright -P openai
 uv run poe test-typing -P openai
 uv run poe test -P ag-ui
+uv run poe test -P declarative
 uv run --directory packages/foundry_hosting poe test
 ```
 

python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py

@@ -636,7 +636,7 @@ def _handle_step_based_approval(messages: list[Any]) -> list[BaseEvent]:
         try:
             parsed_result = json.loads(approval_text)
             result: dict[str, Any] = cast(dict[str, Any], parsed_result) if isinstance(parsed_result, dict) else {}
-            accepted = bool(result.get("accepted", False))
+            accepted = result.get("accepted") is True
             steps_raw = result.get("steps", [])
             steps: list[dict[str, Any]] = []
             if isinstance(steps_raw, list):
@@ -880,7 +880,7 @@ def _register_server_generated_approval_response(
         aliases=[str(response.function_call.call_id)] if response.function_call.call_id else None,
         server_label=_function_call_server_label(response.function_call),
     )
-    if not response.approved:
+    if response.approved is not True:
         lifecycle.claim_batch(
             thread_id=thread_id,
             decisions=[
@@ -1565,6 +1565,12 @@ async def _resolve_approval_responses(
         # stale replay controls and must not authorize a malformed fresh one.
         primary_response = responses[-1]
         response_content_ids_to_strip.update(id(response) for response in responses[:-1])
+        if not isinstance(primary_response.approved, bool):
+            logger.warning(
+                "Treating approval response id=%s as rejected: approved must be a boolean",
+                primary_response.id,
+            )
+            primary_response.approved = False
         resp_id = primary_response.id
         id_entry = (
             lifecycle.occurrence_for_alias(thread_id=thread_id, interrupt_id=str(resp_id))
@@ -1612,7 +1618,7 @@ async def _resolve_approval_responses(
             else:
                 primary_response.function_call.additional_properties.pop("server_label", None)
         if (
-            primary_response.approved
+            primary_response.approved is True
             and lifecycle is not None
             and authorized_executions is not None
             and primary_response.function_call is not None
@@ -1626,14 +1632,14 @@ async def _resolve_approval_responses(
             intents_by_response_content_id[id(primary_response)] = intent
         valid_response_content_ids.add(id(primary_response))
         if (
-            primary_response.approved
+            primary_response.approved is True
             and intent is not None
             and intent.owner in {ApprovalExecutionOwner.HOSTED, ApprovalExecutionOwner.DEFERRED}
         ):
             validated_forwarded_approvals.append(primary_response)
         if not server_label:
             pending_local_response_content_ids.add(id(primary_response))
-        if validated_approved_responses is not None and primary_response.approved and not server_label:
+        if validated_approved_responses is not None and primary_response.approved is True and not server_label:
             validated_approved_responses.append(primary_response)
 
     if response_content_ids_to_strip:
@@ -1698,7 +1704,7 @@ async def forward_hosted_decision(approval: Content = approval) -> list[Content]
     if not fcc_todo:
         return []
 
-    approved_responses = [resp for resp in fcc_todo.values() if resp.approved]
+    approved_responses = [resp for resp in fcc_todo.values() if resp.approved is True]
 
     approved_function_result_groups: list[list[Content]] = []
 
@@ -1909,7 +1915,7 @@ def _clean_resolved_approvals_from_snapshot(
             )
             if target_call_id is None:
                 continue
-            if parsed.get("accepted"):
+            if parsed.get("accepted") is True:
                 replacement = result_by_call_id.get(target_call_id)
                 if replacement is None:
                     continue

python/packages/ag-ui/agent_framework_ag_ui/_approval_lifecycle.py

@@ -901,7 +901,7 @@ def settle_forwarded(
             result
             for result in results
             if result.type == "function_approval_response"
-            and result.approved
+            and result.approved is True
             and result.function_call is not None
             and result.function_call.call_id == occurrence.identity.call_id
         ]

python/packages/ag-ui/agent_framework_ag_ui/_message_adapters.py

@@ -152,9 +152,9 @@ def _sanitize_tool_history(
                     if content.function_call and content.function_call.call_id:
                         approval_call_ids.add(str(content.function_call.call_id))
                     if approval_accepted is None:
-                        approval_accepted = bool(content.approved)
+                        approval_accepted = content.approved is True
                     else:
-                        approval_accepted = approval_accepted and bool(content.approved)
+                        approval_accepted = approval_accepted and content.approved is True
 
             if approval_call_ids and pending_tool_call_ids:
                 pending_tool_call_ids = [
@@ -203,7 +203,7 @@ def _sanitize_tool_history(
                             contents=[
                                 Content.from_function_result(
                                     call_id=pending_confirm_changes_id,
-                                    result="Confirmed" if parsed.get("accepted") else "Rejected",
+                                    result="Confirmed" if parsed.get("accepted") is True else "Rejected",
                                 )
                             ],
                         )
@@ -724,7 +724,7 @@ def _filter_modified_args(
                 # Look for the matching function call in previous messages to create
                 # proper function_approval_response content. This enables the agent framework
                 # to execute the approved tool (fix for GitHub issue #3034).
-                accepted = parsed.get("accepted", False) if parsed is not None else False
+                accepted = parsed.get("accepted") is True if parsed is not None else False
                 approval_payload_text = result_content if isinstance(result_content, str) else json.dumps(parsed)
 
                 # Log the full approval payload to debug modified arguments
@@ -932,7 +932,7 @@ def _filter_modified_args(
 
                 # Create the approval response
                 approval_response = Content.from_function_approval_response(
-                    approved=approval.get("approved", True),
+                    approved=approval.get("approved") is True,
                     id=approval.get("id", ""),
                     function_call=func_call,
                 )

python/packages/ag-ui/tests/ag_ui/test_approval_result_event.py

@@ -14,7 +14,7 @@
 
 from agent_framework_ag_ui._agent import AgentConfig
 from agent_framework_ag_ui._agent_run import run_agent_stream
-from agent_framework_ag_ui._approval_lifecycle import ApprovalExecutionOwner
+from agent_framework_ag_ui._approval_lifecycle import ApprovalExecutionOwner, ApprovalStatus
 from agent_framework_ag_ui._approval_state import InMemoryAGUIApprovalStateStore
 
 
@@ -148,6 +148,69 @@ async def test_rejected_call_does_not_execute_or_emit_live_result() -> None:
     assert not [event for event in events if getattr(event, "type", None) == "TOOL_CALL_RESULT"]
 
 
+async def test_resolve_approval_responses_treats_non_boolean_decision_as_rejection() -> None:
+    """A malformed decision completes the pending call as an explicit rejection."""
+    executions: list[str] = []
+
+    def guarded_write(value: str) -> str:
+        executions.append(value)
+        return f"wrote:{value}"
+
+    tool = FunctionTool(name="guarded_write", description="Write", func=guarded_write)
+    function_call = Content.from_function_call(
+        call_id="call-bool",
+        name="guarded_write",
+        arguments={"value": "safe"},
+    )
+    response = Content.from_function_approval_response(
+        approved=True,
+        id="approval-bool",
+        function_call=function_call,
+    )
+    response.approved = "true"  # type: ignore[assignment]  # ty: ignore[invalid-assignment]
+    store = InMemoryAGUIApprovalStateStore()
+    store.set_tool_approval_state(
+        "thread-bool",
+        {"collected_approval_responses": [response]},
+    )
+    agent = StubAgent(
+        updates=[AgentResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")],
+        default_options={"tools": [tool]},
+    )
+
+    events = [
+        event
+        async for event in run_agent_stream(
+            {
+                "thread_id": "thread-bool",
+                "run_id": "run-bool",
+                "messages": [{"role": "user", "content": "Continue"}],
+            },
+            agent,
+            AgentConfig(),
+            approval_state_store=store,
+        )
+    ]
+
+    occurrence = store.lifecycle.occurrence_for_alias(
+        thread_id="thread-bool",
+        interrupt_id="approval-bool",
+    )
+    rejection_results = [
+        content
+        for message in agent.messages_received
+        for content in message.contents
+        if content.type == "function_result" and content.call_id == "call-bool"
+    ]
+    assert executions == []
+    assert occurrence is not None
+    assert occurrence.status is ApprovalStatus.REJECTED
+    assert [(result.call_id, result.result) for result in rejection_results] == [
+        ("call-bool", "Error: Tool call invocation was rejected by user.")
+    ]
+    assert not [event for event in events if getattr(event, "type", None) == "TOOL_CALL_RESULT"]
+
+
 async def test_mixed_batch_preserves_approved_result_identity_and_order() -> None:
     executions: list[str] = []