Python: Improve function approval resume and replay (#7345)
* Python: Harden function approval resume and replay Make approval resume immutable and occurrence-aware, return grouped approved and rejected results consistently, preserve pending approval history without model-orphaned calls, and align streaming, non-streaming, and AG-UI result boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1ee1d250-d1e2-4c6f-8c36-aae0d94fe7a1 * Python: Clarify function invocation orchestration Simplify approval-resolution setup and add phase-level comments around the key function invocation orchestration paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1ee1d250-d1e2-4c6f-8c36-aae0d94fe7a1 EOF && git push origin python-approval-resume-contract --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 1ee1d250-d1e2-4c6f-8c36-aae0d94fe7a1
This commit is contained in:
Eduard van Valkenburg
committed by
GitHub
GitHub
parent
0e6a104192
commit
5987a6791b
@@ -0,0 +1,543 @@
|
||||
---
|
||||
status: proposed
|
||||
contact: eavanvalkenburg
|
||||
date: 2026-07-27
|
||||
deciders: eavanvalkenburg
|
||||
---
|
||||
|
||||
# Python function-calling loop contract and validation matrix
|
||||
|
||||
## Scope
|
||||
|
||||
This specification defines the required behavior and validation coverage for the Python function-calling loop.
|
||||
It covers:
|
||||
|
||||
- normal local function execution;
|
||||
- streaming and non-streaming response aggregation;
|
||||
- tool approval request and resume;
|
||||
- approved, rejected, mixed, and replayed approval rounds;
|
||||
- reasoning content and opaque reasoning signatures bound to function calls;
|
||||
- history persistence and service-side continuation;
|
||||
- error, user-input, middleware-termination, and loop-limit paths;
|
||||
- provider and transport serialization of function calls and results.
|
||||
|
||||
The primary implementation is in `python/packages/core/agent_framework/_tools.py`. History replay behavior in
|
||||
`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.
|
||||
|
||||
## Change sensitivity
|
||||
|
||||
This code is high risk. Small changes can produce duplicate side effects, orphaned calls or results, invalid
|
||||
provider histories, invisible streaming results, stale approval authority, or loops that never terminate.
|
||||
Dropping reasoning content that a service binds to a tool call can also make an otherwise balanced call/result
|
||||
transcript invalid.
|
||||
|
||||
Any change to the function-calling loop or its approval/history/serialization paths must:
|
||||
|
||||
1. identify every affected row in the scenario matrix below;
|
||||
2. add or update the corresponding regression tests;
|
||||
3. validate streaming updates, streaming finalization, and non-streaming output where applicable;
|
||||
4. validate both model-bound history and caller-visible responses;
|
||||
5. run the full core package tests plus every affected provider or transport package;
|
||||
6. run source typing, test typing, and syntax checks for every affected package;
|
||||
7. receive extra review focused on call/result pairing, exactly-once execution, and history replay.
|
||||
|
||||
A passing narrow regression test is not sufficient evidence for changes in this area.
|
||||
|
||||
### Contribution ownership
|
||||
|
||||
Issues involving this code must not be picked up by external contributors without first checking with the Agent
|
||||
Framework core team. The core team must confirm the intended behavior, affected scenario-matrix rows, ownership
|
||||
across core/providers/transports, and the required validation scope before implementation starts.
|
||||
|
||||
## Flow diagrams and code map
|
||||
|
||||
### Main function-calling flow
|
||||
|
||||
The main control flow deliberately has separate streaming and non-streaming methods. They share policy helpers, but
|
||||
their output mechanics differ: one returns an aggregated `ChatResponse`; the other yields `ChatResponseUpdate`
|
||||
items and is finalized by `ResponseStream`.
|
||||
|
||||
The diagrams use only the generic distinction between **local tools**, which Agent Framework executes, and
|
||||
**hosted-service tools**, whose calls and approval decisions are owned by a remote service. Provider-specific wire
|
||||
formats and regression tests appear later in the scenario matrix.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Entry["FunctionInvocationLayer.get_response(...)"]
|
||||
Setup["Prepare middleware, options, session, budget state,<br/>and execute_function_calls partial"]
|
||||
Enabled{"Function invocation enabled?"}
|
||||
Direct["Delegate directly to super().get_response(...)"]
|
||||
Mode{"stream?"}
|
||||
NonStream["_get_response_with_function_invocation(...)"]
|
||||
Stream["_stream_response_with_function_invocation(...)"]
|
||||
Resolve["_resolve_approval_responses(...)<br/>runs once before the model-iteration loop"]
|
||||
ApprovalAction{"approval action"}
|
||||
Immediate["Return/yield terminal result or user-input request<br/>without another model call"]
|
||||
ApprovalPolicy["Record approval executions;<br/>apply stop/function-call-limit policy"]
|
||||
Model["Call super_get_response(...)<br/>response may contain reasoning + function_call"]
|
||||
Process["_process_model_function_calls(...)"]
|
||||
FunctionAction{"function-processing action"}
|
||||
Execute["_execute_function_calls(...)"]
|
||||
Try["_try_execute_function_calls(...)"]
|
||||
Single["_execute_single_function_call(...)"]
|
||||
Handle["_handle_function_call_results(...)"]
|
||||
PostCallPolicy["Record executions; apply error/function-call-limit policy;<br/>reset required tool choice"]
|
||||
Advance["_prepare_messages_for_next_iteration(...)"]
|
||||
More{"iteration budget remains?"}
|
||||
Final["Final model call with tool_choice = none<br/>and deterministic fallback if needed"]
|
||||
Output["Return ChatResponse or complete ResponseStream"]
|
||||
|
||||
Entry --> Setup --> Enabled
|
||||
Enabled -- no --> Direct
|
||||
Enabled -- yes --> Mode
|
||||
Mode -- no --> NonStream
|
||||
Mode -- yes --> Stream
|
||||
NonStream --> Resolve
|
||||
Stream --> Resolve
|
||||
Resolve --> ApprovalAction
|
||||
ApprovalAction -- return --> Immediate --> Output
|
||||
ApprovalAction -- stop --> ApprovalPolicy
|
||||
ApprovalAction -- continue --> ApprovalPolicy
|
||||
ApprovalPolicy --> More
|
||||
Model --> Process
|
||||
Process --> Execute --> Try --> Single --> Handle --> FunctionAction
|
||||
FunctionAction -- return --> Output
|
||||
FunctionAction -- stop --> PostCallPolicy
|
||||
FunctionAction -- continue --> PostCallPolicy
|
||||
PostCallPolicy --> Advance
|
||||
Advance --> More
|
||||
More -- yes --> Model
|
||||
More -- no --> Final --> Output
|
||||
```
|
||||
|
||||
Code-reading landmarks:
|
||||
|
||||
- `get_response(...)` owns setup and selects the response mode.
|
||||
- `_get_response_with_function_invocation(...)` owns non-streaming aggregation.
|
||||
- `_stream_response_with_function_invocation(...)` owns streamed emission/finalization.
|
||||
- `_resolve_approval_responses(...)` handles only inbound approval decisions.
|
||||
- `_process_model_function_calls(...)` handles only calls from a completed model response.
|
||||
- `_try_execute_function_calls(...)` decides approval/declaration/execution behavior for a batch.
|
||||
- `_replace_approval_contents_with_results(...)` is the occurrence-aware approval transcript normalizer.
|
||||
|
||||
### Approval pause and resume
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Caller
|
||||
participant History as HistoryProvider
|
||||
participant Layer as FunctionInvocationLayer
|
||||
participant Tool
|
||||
participant Model
|
||||
|
||||
Caller->>Layer: Initial user request
|
||||
Layer->>Model: Messages + tools
|
||||
Model-->>Layer: reasoning content + function_call
|
||||
Layer->>Layer: Tool requires approval
|
||||
Layer-->>Caller: function_call + function_approval_request
|
||||
|
||||
Caller->>Layer: function_approval_response
|
||||
Layer->>Layer: Copy caller-owned messages
|
||||
Layer->>Layer: _resolve_approval_responses(...)
|
||||
|
||||
alt approved
|
||||
Layer->>Tool: Execute exactly once
|
||||
Tool-->>Layer: result or exception
|
||||
Layer->>Layer: Create terminal function_result
|
||||
else rejected
|
||||
Layer->>Layer: Create synthetic rejection function_result
|
||||
end
|
||||
|
||||
Layer-->>Caller: Terminal result message/update
|
||||
|
||||
alt tool requests more user input
|
||||
Layer-->>Caller: User-input request with assistant role
|
||||
else middleware terminates
|
||||
Layer-->>Caller: Termination result
|
||||
else error limit reached
|
||||
Layer->>Model: Normalized reasoning/call/result history, tools disabled
|
||||
Model-->>Layer: Final assistant response
|
||||
Layer-->>Caller: Final assistant response
|
||||
else continue normally
|
||||
Layer->>Model: Normalized reasoning/call/result history
|
||||
Model-->>Layer: Final assistant response or another function_call
|
||||
Layer-->>Caller: Final assistant response / continued loop
|
||||
end
|
||||
|
||||
Layer-->>History: Persist caller input + returned response
|
||||
Note over History: Later model replay filters approval request/response wrappers
|
||||
```
|
||||
|
||||
The terminal result is caller-visible in both modes. The private normalized message copy is model-visible. The
|
||||
original caller input and earlier response remain unchanged.
|
||||
|
||||
### Reasoning-bound function-call groups
|
||||
|
||||
Some hosted services bind reasoning content or an opaque reasoning signature to the function call that follows it.
|
||||
For those services, reasoning is not optional decoration; it is part of the provider-valid function-call group.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Response["Assistant response:<br/>reasoning content + function_call"]
|
||||
Group["One logical reasoning/function-call group"]
|
||||
Owner{"local or hosted-service tool?"}
|
||||
Local["Local execution"]
|
||||
Hosted["Hosted service owns tool execution/state"]
|
||||
Result["Terminal function_result or hosted result"]
|
||||
Continuation{"continuation mode"}
|
||||
Stateless["Stateless or framework-history replay"]
|
||||
Replayable{"reasoning payload/signature<br/>is replayable?"}
|
||||
Replay["Replay reasoning + call + result atomically"]
|
||||
Reject["Fail before the service call;<br/>do not send a lossy transcript"]
|
||||
Service["Hosted-service continuation"]
|
||||
Reference["Reference service-stored reasoning/call;<br/>send only the new result or approval decision"]
|
||||
Compact{"compaction needed?"}
|
||||
Atomic["Keep or exclude the complete<br/>reasoning/call/result group"]
|
||||
Caller["Caller-visible response retains reasoning<br/>with the function-call turn"]
|
||||
|
||||
Response --> Group --> Owner
|
||||
Group --> Caller
|
||||
Owner -- local --> Local --> Result
|
||||
Owner -- hosted service --> Hosted --> Result
|
||||
Result --> Compact
|
||||
Compact -- yes --> Atomic --> Continuation
|
||||
Compact -- no --> Continuation
|
||||
Continuation -- stateless / local history --> Stateless --> Replayable
|
||||
Replayable -- yes --> Replay
|
||||
Replayable -- no --> Reject
|
||||
Continuation -- service-managed --> Service --> Reference
|
||||
```
|
||||
|
||||
The generic contract is:
|
||||
|
||||
- reasoning content remains ordered immediately before or alongside the function call it explains;
|
||||
- a terminal result does not replace or discard the reasoning/call portion of the active group;
|
||||
- stateless replay includes the service-required reasoning payload or opaque signature;
|
||||
- if required reasoning cannot be reconstructed, the adapter fails before sending invalid or lossy history;
|
||||
- service-managed continuation may rely on the hosted service's stored reasoning/call items and send only new
|
||||
outputs or approval decisions;
|
||||
- compaction keeps or removes the entire reasoning/call/result group atomically.
|
||||
|
||||
In the code, core response aggregation preserves reasoning `Content` items, compaction annotations bind reasoning to
|
||||
the tool-call group, and provider adapters serialize or reconstruct the provider-specific reasoning representation.
|
||||
|
||||
### 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.
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
Scan["Scan normalized messages in order"]
|
||||
Kind{"content type"}
|
||||
Call["function_call:<br/>open a call occurrence"]
|
||||
Request["function_approval_request"]
|
||||
Bind{"unbound call occurrence<br/>with same call_id?"}
|
||||
BindExisting["Bind request id to existing occurrence<br/>and remove wrapper"]
|
||||
Duplicate{"same request identity<br/>already restored?"}
|
||||
DropDuplicate["Remove replayed duplicate wrapper"]
|
||||
Restore["Restore embedded function_call<br/>as a new occurrence"]
|
||||
Placeholder["function_result with APPROVAL_PENDING:<br/>attach placeholder to open occurrence"]
|
||||
Completed["terminal function_result:<br/>close earliest open occurrence"]
|
||||
Response["function_approval_response"]
|
||||
Pending{"response still pending?"}
|
||||
RemoveOld["Remove already-resolved historical response"]
|
||||
Decision{"approved?"}
|
||||
Approved["Pop next execution result for this call_id"]
|
||||
Rejected["Create synthetic rejection result"]
|
||||
HasPlaceholder{"occurrence has placeholder?"}
|
||||
Replace["Replace placeholder and remove response wrapper"]
|
||||
ReplaceResponse["Replace response wrapper with terminal content"]
|
||||
Close["Close occurrence; append terminal content<br/>to resumed response"]
|
||||
Next["Continue scan"]
|
||||
|
||||
Scan --> Kind
|
||||
Kind -- function_call --> Call --> Next
|
||||
Kind -- approval request --> Request --> Bind
|
||||
Bind -- yes --> BindExisting --> Next
|
||||
Bind -- no --> Duplicate
|
||||
Duplicate -- yes --> DropDuplicate --> Next
|
||||
Duplicate -- no --> Restore --> Next
|
||||
Kind -- pending placeholder --> Placeholder --> Next
|
||||
Kind -- terminal result --> Completed --> Next
|
||||
Kind -- approval response --> Response --> Pending
|
||||
Pending -- no --> RemoveOld --> Next
|
||||
Pending -- yes --> Decision
|
||||
Decision -- yes --> Approved --> HasPlaceholder
|
||||
Decision -- no --> Rejected --> HasPlaceholder
|
||||
HasPlaceholder -- yes --> Replace --> Close --> Next
|
||||
HasPlaceholder -- no --> ReplaceResponse --> Close --> Next
|
||||
Next --> Kind
|
||||
```
|
||||
|
||||
This flow corresponds to `_ApprovalCallOccurrence`, `_collect_approval_responses(...)`, and
|
||||
`_replace_approval_contents_with_results(...)`.
|
||||
|
||||
### History and service-side continuation
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
Store["History backing store<br/>(may retain approval wrappers for audit)"]
|
||||
Load{"HistoryProvider.load_messages?"}
|
||||
Filter["_filter_approval_control_messages(...)"]
|
||||
Context["SessionContext model history:<br/>function_call + terminal function_result"]
|
||||
Current["Current caller input:<br/>new function_approval_response"]
|
||||
Layer["FunctionInvocationLayer private copy"]
|
||||
Local{"local or hosted-service approval?"}
|
||||
LocalResult["Execute locally and normalize to function_result"]
|
||||
Hosted["Hosted-service adapter"]
|
||||
StoredRequest["Prior service-issued approval request"]
|
||||
NewResponse["Current hosted approval decision"]
|
||||
Skip["Do not replay the stored request inline"]
|
||||
Send["Send the approval decision exactly once"]
|
||||
Later["Later turn"]
|
||||
Manual["Manual-history caller"]
|
||||
|
||||
Store --> Load
|
||||
Load -- yes --> Filter --> Context --> Layer
|
||||
Load -- no --> Layer
|
||||
Current --> Layer
|
||||
Layer --> Local
|
||||
Local -- local --> LocalResult --> Later
|
||||
Local -- hosted service --> Hosted
|
||||
StoredRequest --> Hosted --> Skip
|
||||
NewResponse --> Hosted --> Send --> Later
|
||||
Later --> Store
|
||||
Manual -. owns equivalent filtering .-> Layer
|
||||
```
|
||||
|
||||
When `load_messages=False`, no history is replayed and the history filter is intentionally not invoked. Callers
|
||||
that manually replay messages own the equivalent rule: do not resend an approval response after its terminal result.
|
||||
|
||||
## Normative contract
|
||||
|
||||
### Function calls and results
|
||||
|
||||
- Every actionable local `function_call` produces exactly one terminal `function_result`, unless execution pauses
|
||||
for a new user-input request.
|
||||
- Parallel calls retain model order in the returned transcript.
|
||||
- Reused `call_id` values are correlated by logical occurrence, not one global value per id.
|
||||
- A completed function call/result pair is inert on later turns.
|
||||
- Informational-only and declaration-only calls are not executed as local tools.
|
||||
|
||||
### Reasoning-bound calls
|
||||
|
||||
- Reasoning content or opaque reasoning metadata that a service binds to a function call is part of the same logical
|
||||
group as that call and its terminal result.
|
||||
- Active function loops preserve the reasoning content, function call, function result, and final assistant output
|
||||
in caller-visible responses.
|
||||
- Framework-managed/stateless replay includes the service-required reasoning representation before the paired call.
|
||||
- Service-managed continuation may omit inline reasoning/call items only when the hosted service already owns them.
|
||||
- Missing non-reconstructable reasoning fails explicitly before a provider request instead of silently dropping the
|
||||
content.
|
||||
- Compaction preserves or excludes the complete reasoning/call/result group atomically.
|
||||
|
||||
### Approval request and resume
|
||||
|
||||
- A tool that requires approval does not execute before an approved response.
|
||||
- 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`.
|
||||
- The resumed response contains the newly resolved approved and rejected terminal results before any final assistant
|
||||
message.
|
||||
- Streaming yields the same logical result content and ordering as non-streaming output and
|
||||
`ResponseStream.get_final_response()`.
|
||||
- The function invocation layer normalizes a private copy of caller messages. It must not mutate the caller's
|
||||
approval `Message`, approval `Content`, or an earlier returned response.
|
||||
- Approval-time `UserInputRequiredException` and `MiddlewareTermination` return immediately without another model
|
||||
call.
|
||||
|
||||
### Approval control content
|
||||
|
||||
- `function_approval_request` and `function_approval_response` are control-plane contents, not durable model
|
||||
transcript items.
|
||||
- A current hosted approval response must be sent once on the immediate resume request.
|
||||
- A server-issued approval request must not be replayed inline during service-side continuation.
|
||||
- History providers may retain approval control contents in their backing store for audit, but base history replay
|
||||
filters them before later model calls.
|
||||
- Callers that manually own and replay message history without a loading `HistoryProvider` must likewise omit a
|
||||
previously submitted approval response from later continuation requests.
|
||||
|
||||
### History and continuation
|
||||
|
||||
- Model-bound history contains one function call/result pair per completed logical occurrence.
|
||||
- Append-only history must not replay stale approval request/response wrappers to the model.
|
||||
- Framework-managed and service-managed continuation must preserve the same logical call/result transcript.
|
||||
- A terminal result consumes the corresponding approval authority in explicit stateless replay.
|
||||
|
||||
## Scenario-to-test matrix
|
||||
|
||||
### Normal function invocation
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Single non-streaming call | Call, result, and final assistant message are returned in order. | `packages/core/tests/core/test_function_invocation_logic.py::test_base_client_with_function_calling` |
|
||||
| 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` |
|
||||
| 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` |
|
||||
| Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` |
|
||||
| Declaration-only call | The call is surfaced as user input and is not executed. | `test_declaration_only_tool` |
|
||||
| Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
|
||||
| Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
|
||||
|
||||
### Approval pause and resume
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Initial approval request | Assistant response contains the original call and approval request; tool does not execute. | `test_approval_requests_in_assistant_message`, `test_streaming_approval_request_generated`, `test_streaming_approval_requests_in_assistant_message` |
|
||||
| Approved non-streaming resume | Result precedes final text; tool executes once; inputs remain unchanged. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_returns_result_without_mutating_inputs[non-streaming-approved]` |
|
||||
| Rejected non-streaming resume | Rejection result precedes final text; tool executes zero times; inputs remain unchanged. | `test_approval_resume_returns_result_without_mutating_inputs[non-streaming-rejected]` |
|
||||
| Approved streaming resume | Result update precedes final text and final response matches non-streaming shape. | `test_approval_resume_returns_result_without_mutating_inputs[streaming-approved]`, `test_streaming_approval_resume_yields_terminal_result_before_model_text[approved]` |
|
||||
| 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` |
|
||||
| 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` |
|
||||
|
||||
### Approval correlation and replay
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Result matching without placeholders | Results match calls by id even when the result list is reordered. | `test_replace_approval_contents_with_results_uses_result_call_ids_without_placeholders` |
|
||||
| Reused id after completion | A later round with the same id creates a second valid pair. | `test_replace_approval_contents_with_results_allows_reused_call_id_after_completion` |
|
||||
| Replayed approval wrapper | A duplicated wrapper does not restore another function call. | `test_replace_approval_contents_with_results_deduplicates_replayed_approval_request` |
|
||||
| Historical resolved response plus new round | The old response is removed from normalized input and is not converted into a rejection result. | `test_replace_approval_contents_with_results_ignores_already_resolved_response` |
|
||||
| Multiple reused-id rounds | Approved and rejected rounds retain separate call/result occurrences. | `test_replace_approval_contents_with_results_correlates_reused_call_id_occurrences` |
|
||||
| Multi-content result with reused id | Every content produced by one execution stays with that approval occurrence and cannot bleed into the next reused-id round. | `test_replace_approval_contents_with_results_keeps_multi_content_group_with_reused_call_id` |
|
||||
| Follow-up request closes one occurrence | A user-input follow-up consumes only the preceding approval authority and leaves a later reused-id response pending. | `test_collect_approval_responses_consumes_matching_follow_up_request_occurrence` |
|
||||
| Reused-id placeholders | Placeholder results consume approved results by occurrence. | `test_replace_approval_contents_with_results_correlates_reused_call_id_placeholders` |
|
||||
| Rejected placeholder | Rejection replaces the pending placeholder instead of adding a second result. | `test_replace_approval_contents_with_results_replaces_rejected_placeholder` |
|
||||
| Results reordered with placeholders | Results still match the correct call ids. | `test_replace_approval_contents_with_results_uses_result_call_ids_for_placeholders` |
|
||||
| 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` |
|
||||
| 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` |
|
||||
|
||||
### Mixed batches and approval middleware
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Safe and approval-required calls in one batch | Hidden safe calls replay only with the matching visible approval. | `packages/core/tests/core/test_harness_tool_approval.py::test_mixed_batch_hides_already_approved_request_until_approval_replay` |
|
||||
| Restored approval state | Serialized `ToolApprovalState` restores mixed-batch behavior. | `test_mixed_batch_accepts_restored_tool_approval_state` |
|
||||
| Unrelated turn before approval | Hidden calls do not execute on an unrelated turn. | `test_hidden_mixed_batch_requests_do_not_replay_on_unrelated_turn` |
|
||||
| Multiple abandoned batches | Hidden calls replay only for the matching batch. | `test_hidden_mixed_batch_requests_replay_only_for_matching_visible_approval` |
|
||||
| Queued approvals | One unresolved approval is surfaced per run without premature execution. | `test_tool_approval_middleware_queues_multiple_approval_requests`, `test_tool_approval_middleware_queues_streamed_approval_requests` |
|
||||
| Middleware state plus hidden core state | State saves do not discard hidden mixed-batch calls. | `test_tool_approval_middleware_preserves_hidden_mixed_batch_requests` |
|
||||
| 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` |
|
||||
| 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` |
|
||||
|
||||
### Errors, control flow, and limits
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Rejected execution | Rejection is a normal terminal result, not an exception to the caller. | `test_unapproved_tool_execution_raises_exception` |
|
||||
| Approved tool exception | Generic and detailed error modes preserve one result and one execution. | `test_approved_function_call_with_error_without_detailed_errors`, `test_approved_function_call_with_error_with_detailed_errors` |
|
||||
| Approved validation error | Validation failure returns one result without invoking the function body. | `test_approved_function_call_with_validation_error` |
|
||||
| Approved success | Successful approved execution returns one result. | `test_approved_function_call_successful_execution` |
|
||||
| Consecutive error cap | Error threshold stops repeated failures, submits collected results, and makes only the required final no-tool model call. | `test_function_invocation_config_max_consecutive_errors`, `test_streaming_function_invocation_config_max_consecutive_errors`, `test_approval_resume_error_limit_forces_final_no_tool_response` |
|
||||
| Unknown call handling | Configured false returns an error result; configured true raises. | `test_function_invocation_config_terminate_on_unknown_calls_false`, `test_function_invocation_config_terminate_on_unknown_calls_true`, streaming equivalents |
|
||||
| Middleware termination | Normal non-approval loop stops without a second model call. | `test_terminate_loop_single_function_call`, `test_terminate_loop_multiple_function_calls_one_terminates`, `test_terminate_loop_streaming_single_function_call` |
|
||||
| Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents |
|
||||
| Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent |
|
||||
| Conversation continuation | Conversation id updates between iterations and is cleared on stop where required. | `test_conversation_id_updated_in_options_between_tool_iterations`, `test_function_invocation_stop_clears_conversation_id_non_stream`, `test_streaming_function_invocation_stop_clears_conversation_id` |
|
||||
|
||||
### History and provider serialization
|
||||
|
||||
| Scenario | Required invariant | Primary regression test |
|
||||
|---|---|---|
|
||||
| Append-only history replay | Resolved approval wrappers do not reach a later model call; one call/result pair remains. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_filters_resolved_control_items_from_file_history` |
|
||||
| Pending placeholder history | An approval response remains replayable while its only result is `[APPROVAL_PENDING]`. | `packages/core/tests/core/test_sessions.py::test_filter_approval_controls_keeps_response_for_pending_placeholder` |
|
||||
| Pending hosted history replay | Stateless hosted approval requests remain replayable until a response is recorded, then both controls become inert. | `packages/openai/tests/openai/test_openai_chat_client.py::test_stateless_history_preserves_pending_hosted_approval_request_until_response` |
|
||||
| Non-history provider plus session | Local history is still auto-injected for approval resume. | `packages/core/tests/core/test_agents.py::test_non_history_context_provider_still_injects_inmemory` |
|
||||
| OpenAI approval serialization | Approval id and decision serialize to `mcp_approval_response`. | `test_prepare_message_for_openai_with_function_approval_response`, `test_prepare_content_for_opentool_approval_response`, `test_function_approval_response_with_mcp_tool_call` |
|
||||
| OpenAI end-to-end hosted approval | Hosted request parses, response sends, and continuation completes. | `test_end_to_end_mcp_approval_flow` |
|
||||
| Stored function call/result | Service-side storage drops server-issued calls but keeps new outputs. | `test_prepare_options_with_conversation_id_strips_server_issued_items`, `test_prepare_messages_for_openai_full_conversation_with_reasoning` |
|
||||
| Stateless reasoning replay | Replay reconstructs reasoning, call, and result together; missing required reasoning fails before the request. | `test_tool_loop_store_false_replays_encrypted_reasoning_group`, `test_stateless_request_rejects_non_replayable_reasoning_bound_mcp_output`, `test_prepare_messages_for_openai_full_conversation_with_reasoning` |
|
||||
| Opaque reasoning signature replay | Provider-specific opaque reasoning metadata is captured and restored on reconstructed calls. | `packages/gemini/tests/test_gemini_client.py::test_function_call_part_captures_thought_signature_as_reasoning_content`, `test_reconstructed_function_call_replays_thought_signature_from_reasoning_content` |
|
||||
| Chat Completions approval wrappers | Framework approval wrappers are not sent as chat messages. | `packages/openai/tests/openai/test_openai_chat_completion_client.py` approval serialization tests |
|
||||
| AG-UI approval result event | Approved result emits once with content and persists in snapshot. | `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_approval_resume_emits_tool_call_result`, `test_approval_resume_result_has_content`, `test_approval_resume_snapshot_replaces_approval_payload_with_tool_result`, `test_approval_resume_zero_updates_emits_tool_result` |
|
||||
| AG-UI rejection/mixed decision | Transport emits only the events defined for approved and rejected calls without duplicates. | `test_rejection_does_not_emit_tool_call_result`, `test_mixed_approve_reject_emits_only_approved_tool_result`, `test_resolve_approval_responses_returns_only_approved` |
|
||||
| AG-UI approval-time follow-up | The full grouped user-input pause remains in message history and emits no synthetic `TOOL_CALL_RESULT`. | `test_resolve_approval_responses_preserves_follow_up_user_input_group` |
|
||||
| AG-UI approval execution failure | A grouped executor failure becomes one deterministic terminal error result for the approved call. | `test_resolve_approval_responses_returns_failure_when_grouped_execution_raises` |
|
||||
| AG-UI no-approval path | Ordinary tool results do not gain an extra approval result event. | `test_no_approval_no_extra_tool_result` |
|
||||
| Compaction pair integrity | Function call/result groups remain atomic. | `packages/core/tests/core/test_compaction.py::test_group_annotations_keep_tool_call_and_tool_result_atomic`, `test_group_annotations_include_reasoning_in_tool_call_group` |
|
||||
|
||||
## Required coverage gaps
|
||||
|
||||
These scenarios are required but are not fully covered by merged tests on `main`:
|
||||
|
||||
| Gap | Tracking |
|
||||
|---|---|
|
||||
| Non-adjacent and reused-id call/result occurrences remain atomic during compaction. | #7212 |
|
||||
| Provider-injected approval-required tools defer until `before_run` tools exist and still emit one result. | #7043 |
|
||||
| Service-side storage sends the current approval response while omitting the stored request. | #7125 |
|
||||
| Service-owned `previous_response_id` continuation cannot execute a terminal approval again on a later turn. | #6851 |
|
||||
| A provider that ignores `tool_choice="none"` after an invocation limit cannot expose an unanswered call. | #7045 |
|
||||
| Declaration-only streaming preserves request metadata without duplicating arguments. | #6973 |
|
||||
| AG-UI `confirm_changes` cleanup correlates one result by original function call id when several results exist. | #6828 |
|
||||
|
||||
Do not mark these rows covered by nearby tests; each needs a dedicated regression at the owning layer.
|
||||
|
||||
## Minimum validation commands
|
||||
|
||||
Run from `python/` for any core function-loop change:
|
||||
|
||||
```bash
|
||||
uv run poe test -P core
|
||||
uv run poe syntax -P core
|
||||
uv run poe pyright -P core
|
||||
uv run poe test-typing -P core
|
||||
```
|
||||
|
||||
Also run every affected package. Common approval-loop changes require:
|
||||
|
||||
```bash
|
||||
uv run poe test -P openai
|
||||
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 --directory packages/foundry_hosting poe test
|
||||
```
|
||||
|
||||
Run focused regression files first while iterating, but do not substitute them for the full package commands above.
|
||||
|
||||
## Review checklist
|
||||
|
||||
Before accepting an update, reviewers must confirm:
|
||||
|
||||
- the changed behavior is represented in this specification;
|
||||
- the matrix names a regression test for every affected scenario;
|
||||
- approved tools cannot execute twice;
|
||||
- rejected tools cannot execute;
|
||||
- no call or result becomes orphaned or duplicated;
|
||||
- call/result matching does not assume `call_id` is globally unique forever;
|
||||
- reasoning content or opaque signatures remain in the same logical group as the paired call/result, or replay fails
|
||||
explicitly before sending a lossy provider request;
|
||||
- caller messages and previous responses remain immutable;
|
||||
- streaming updates and final response agree with non-streaming output;
|
||||
- history replay does not reintroduce approval authority;
|
||||
- full package, syntax, source typing, and test typing checks were run.
|
||||
|
||||
## Related issues
|
||||
|
||||
- #7241 — approval-resolution result streaming
|
||||
- #7267 / #7271 and #7304 — replayed calls and reused ids
|
||||
- #6851 — duplicate side effects after approval continuation
|
||||
- #7383 — bind approval responses to framework-issued requests after this foundation merges
|
||||
- #6963 / #7095 — opaque reasoning-signature replay
|
||||
- #6074 / #7233 — reasoning-paired tool-call replay
|
||||
- #6450 / #6794 — provider message and tool-result serialization
|
||||
@@ -6,6 +6,8 @@ Instructions for AI coding agents working in the Python codebase.
|
||||
- [DEV_SETUP.md](DEV_SETUP.md) - Development environment setup and available poe tasks
|
||||
- [CODING_STANDARD.md](CODING_STANDARD.md) - Coding standards, docstring format, and performance guidelines
|
||||
- [samples/SAMPLE_GUIDELINES.md](samples/SAMPLE_GUIDELINES.md) - Sample structure and guidelines
|
||||
- [Python function-calling loop specification](../docs/specs/004-python-function-calling-loop.md) - Required
|
||||
behavior, scenario-to-test mapping, coverage gaps, and extra validation for function-loop changes
|
||||
|
||||
**Agent Skills** (`.github/skills/`) — detailed, task-specific instructions loaded on demand:
|
||||
- `python-development` — coding standards, type annotations, docstrings, logging, performance
|
||||
@@ -48,6 +50,16 @@ When preparing a PR description:
|
||||
|
||||
Run `uv run poe` from the `python/` directory to see available commands. See [DEV_SETUP.md](DEV_SETUP.md) for detailed usage.
|
||||
|
||||
## 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 whenever coverage or behavior changes. External contributors must check with the Agent Framework core team
|
||||
before picking up issues in this area.
|
||||
|
||||
## Project Structure
|
||||
|
||||
```
|
||||
|
||||
@@ -29,6 +29,8 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
|
||||
- Multimodal user inputs support both legacy (`text`, `binary`) and draft-style (`image`, `audio`, `video`, `document`) shapes.
|
||||
- Interrupted runs complete with `RUN_FINISHED.outcome.type == "interrupt"` and canonical `outcome.interrupts`; do not document or add new flows that depend on the legacy top-level `RUN_FINISHED.interrupt` field.
|
||||
- `Interrupt` and `ResumeEntry` come from the `ag-ui-protocol` package (`ag_ui.core`), not from an Agent Framework-specific interrupt model.
|
||||
- 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.
|
||||
- SSE keepalive is endpoint-owned transport behavior configured through
|
||||
`add_agent_framework_fastapi_endpoint(keepalive_seconds=...)`. It emits SSE comments only; do not add `PING`,
|
||||
`HEARTBEAT`, or `KEEPALIVE` AG-UI events, and do not add runner-level keepalive settings.
|
||||
|
||||
@@ -42,7 +42,7 @@ from agent_framework._tools import (
|
||||
_collect_approval_responses, # type: ignore
|
||||
_replace_approval_contents_with_results, # type: ignore
|
||||
_TOOL_APPROVAL_STATE_KEY, # type: ignore
|
||||
_try_execute_function_calls, # type: ignore
|
||||
_try_execute_function_call_groups, # type: ignore
|
||||
normalize_function_invocation_configuration,
|
||||
)
|
||||
from agent_framework._types import ResponseStream
|
||||
@@ -1306,7 +1306,7 @@ async def _resolve_approval_responses(
|
||||
approved_responses = validated
|
||||
rejected_responses = validated_rejected
|
||||
|
||||
approved_function_results: list[Any] = []
|
||||
approved_function_result_groups: list[list[Content]] = []
|
||||
|
||||
# Execute approved tool calls
|
||||
if approved_responses and tools:
|
||||
@@ -1319,36 +1319,30 @@ async def _resolve_approval_responses(
|
||||
# Filter out AG-UI-specific kwargs that should not be passed to tool execution
|
||||
tool_kwargs = {k: v for k, v in run_kwargs.items() if k != "options"}
|
||||
try:
|
||||
results, _ = await _try_execute_function_calls(
|
||||
approved_function_result_groups, _ = await _try_execute_function_call_groups(
|
||||
custom_args=tool_kwargs,
|
||||
attempt_idx=0,
|
||||
function_calls=approved_responses,
|
||||
tools=tools,
|
||||
middleware_pipeline=middleware_pipeline,
|
||||
config=config,
|
||||
)
|
||||
approved_function_results = list(results)
|
||||
except Exception as e:
|
||||
logger.exception("Failed to execute approved tool calls; injecting error results: %s", e)
|
||||
approved_function_results = []
|
||||
approved_function_result_groups = []
|
||||
|
||||
# Build results for approved responses (used for TOOL_CALL_RESULT event emission)
|
||||
# Normalize one group per approval and collect only terminal results for TOOL_CALL_RESULT events.
|
||||
replacement_groups: list[list[Content]] = []
|
||||
approved_results: list[Content] = []
|
||||
for idx, approval in enumerate(approved_responses):
|
||||
if (
|
||||
idx < len(approved_function_results)
|
||||
and getattr(approved_function_results[idx], "type", None) == "function_result"
|
||||
):
|
||||
approved_results.append(approved_function_results[idx])
|
||||
continue
|
||||
# Get call_id from function_call if present, otherwise use approval.id
|
||||
func_call = approval.function_call
|
||||
call_id = (func_call.call_id if func_call else None) or approval.id or ""
|
||||
approved_results.append(
|
||||
Content.from_function_result(call_id=call_id, result="Error: Tool call invocation failed.")
|
||||
)
|
||||
result_group = approved_function_result_groups[idx] if idx < len(approved_function_result_groups) else []
|
||||
if not result_group:
|
||||
func_call = approval.function_call
|
||||
call_id = (func_call.call_id if func_call else None) or approval.id or ""
|
||||
result_group = [Content.from_function_result(call_id=call_id, result="Error: Tool call invocation failed.")]
|
||||
replacement_groups.append(result_group)
|
||||
approved_results.extend(content for content in result_group if content.type == "function_result")
|
||||
|
||||
_replace_approval_contents_with_results(messages, fcc_todo, approved_results)
|
||||
_replace_approval_contents_with_results(messages, fcc_todo, replacement_groups)
|
||||
|
||||
# Post-process: Convert user messages with function_result content to proper tool messages.
|
||||
# After _replace_approval_contents_with_results, approved tool calls have their results
|
||||
|
||||
@@ -524,6 +524,90 @@ async def test_resolve_approval_responses_returns_only_approved() -> None:
|
||||
assert "rejected" in str(rejection_results[0].result).lower()
|
||||
|
||||
|
||||
async def test_resolve_approval_responses_preserves_follow_up_user_input_group() -> None:
|
||||
"""Approval-time follow-up requests stay grouped and do not emit a synthetic tool result."""
|
||||
from agent_framework import Message
|
||||
from agent_framework.exceptions import UserInputRequiredException
|
||||
|
||||
from agent_framework_ag_ui._agent_run import _resolve_approval_responses
|
||||
|
||||
def request_consent() -> str:
|
||||
raise UserInputRequiredException(
|
||||
contents=[
|
||||
Content.from_oauth_consent_request(consent_link="https://example.com/consent-1"),
|
||||
Content.from_oauth_consent_request(consent_link="https://example.com/consent-2"),
|
||||
]
|
||||
)
|
||||
|
||||
consent_tool = FunctionTool(
|
||||
name="request_consent",
|
||||
description="Request two consent steps",
|
||||
func=request_consent,
|
||||
approval_mode="always_require",
|
||||
)
|
||||
function_call = Content.from_function_call(call_id="call_consent", name="request_consent", arguments="{}")
|
||||
approval_request = Content.from_function_approval_request(id="approval_consent", function_call=function_call)
|
||||
messages: list[Any] = [
|
||||
Message(role="assistant", contents=[approval_request]),
|
||||
Message(role="user", contents=[approval_request.to_function_approval_response(approved=True)]),
|
||||
]
|
||||
agent = StubAgent(updates=[], default_options={"tools": [consent_tool]})
|
||||
|
||||
results = await _resolve_approval_responses(messages, [consent_tool], agent, {})
|
||||
|
||||
follow_up_requests = [content for message in messages for content in message.contents if content.user_input_request]
|
||||
assert results == []
|
||||
assert [request.consent_link for request in follow_up_requests] == [
|
||||
"https://example.com/consent-1",
|
||||
"https://example.com/consent-2",
|
||||
]
|
||||
assert not [content for message in messages for content in message.contents if content.type == "function_result"]
|
||||
assert any(message.role == "assistant" and message.contents == follow_up_requests for message in messages)
|
||||
|
||||
|
||||
async def test_resolve_approval_responses_returns_failure_when_grouped_execution_raises(
|
||||
monkeypatch: Any,
|
||||
) -> None:
|
||||
"""A grouped-execution failure produces one deterministic result for the approved call."""
|
||||
from agent_framework import Message
|
||||
|
||||
from agent_framework_ag_ui._agent_run import _resolve_approval_responses
|
||||
|
||||
async def fail_grouped_execution(**kwargs: Any) -> tuple[list[list[Content]], bool]:
|
||||
del kwargs
|
||||
raise RuntimeError("execution failed")
|
||||
|
||||
monkeypatch.setattr(
|
||||
"agent_framework_ag_ui._agent_run._try_execute_function_call_groups",
|
||||
fail_grouped_execution,
|
||||
)
|
||||
weather_tool = _make_weather_tool()
|
||||
function_call = Content.from_function_call(
|
||||
call_id="call_execution_failure",
|
||||
name="get_weather",
|
||||
arguments='{"city": "Seattle"}',
|
||||
)
|
||||
approval_request = Content.from_function_approval_request(
|
||||
id="approval_execution_failure",
|
||||
function_call=function_call,
|
||||
)
|
||||
messages: list[Any] = [
|
||||
Message(role="assistant", contents=[approval_request]),
|
||||
Message(role="user", contents=[approval_request.to_function_approval_response(approved=True)]),
|
||||
]
|
||||
agent = StubAgent(updates=[], default_options={"tools": [weather_tool]})
|
||||
|
||||
results = await _resolve_approval_responses(messages, [weather_tool], agent, {})
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].type == "function_result"
|
||||
assert results[0].call_id == "call_execution_failure"
|
||||
assert results[0].result == "Error: Tool call invocation failed."
|
||||
assert [
|
||||
content.result for message in messages for content in message.contents if content.type == "function_result"
|
||||
] == ["Error: Tool call invocation failed."]
|
||||
|
||||
|
||||
class TestApprovalToolResultDisplayChannel:
|
||||
"""Approved tools using ``state_update(..., tool_result=...)`` must route the
|
||||
display payload to the UI event while ``flow.tool_results`` still receives
|
||||
|
||||
@@ -145,6 +145,27 @@ agent_framework/
|
||||
available, approval requests for known non-approval-required tools are treated as already approved, hidden, stored
|
||||
in session state keyed to the visible approval request ids from that batch, and reinjected only when that visible
|
||||
approval flow resumes.
|
||||
- Approval resume is an immutable response boundary: the function invocation layer normalizes a private copy of
|
||||
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 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
|
||||
executions that pause for user input still consume `max_function_calls`.
|
||||
- `function_approval_request` and `function_approval_response` are control-plane contents. History providers may
|
||||
retain them in their backing store for audit. The base `HistoryProvider.before_run` filters resolved wrappers from
|
||||
later model replay, but preserves unresolved requests/responses until a terminal result or follow-up request closes
|
||||
the occurrence. On unrelated turns the function layer omits the complete pending call batch from model input while
|
||||
retaining it for a later approval response. Providers configured with `load_messages=False` do not replay history,
|
||||
so this filter is intentionally not invoked.
|
||||
- Reasoning content or opaque reasoning metadata bound to a function call is part of the same logical group as the
|
||||
call and terminal result. Function-loop replay and compaction must preserve that group atomically; adapters should
|
||||
fail before a stateless request when required reasoning cannot be reconstructed.
|
||||
### Agent Loop (`_harness/_loop.py`)
|
||||
|
||||
- **`AgentLoopMiddleware`** - `AgentMiddleware` that re-runs an agent in a loop by calling `call_next()` repeatedly (the pipeline re-reads `context.messages` each time). One configurable class covers two patterns: a required user `should_continue` predicate (sync or async, the first positional/keyword arg), and a chat-client judge built via the `.with_judge(...)` factory (a second chat client decides whether the original request was answered; loops while it is *not*, using a `JudgeVerdict` structured-output response — internally just an async `should_continue` predicate). The constructor covers the predicate pattern directly; only the judge has a convenience classmethod factory (`.with_judge(judge_client, ...)`) that forwards to `__init__`. Supports both streaming and non-streaming runs. By default a non-streaming run returns an aggregated `AgentResponse` containing every iteration's messages plus the injected `next_message` "nudge" messages (as `user` messages); set `return_final_only=True` to return only the last iteration's response. Streaming runs always yield each iteration's updates and emit the injected nudge messages as `user` updates between iterations (the `return_final_only` flag has no effect on streaming, and the final response reflects the last iteration; `MiddlewareTermination` is handled cleanly). `should_continue` is required; other constructor args are optional: `max_iterations` (safety cap; defaults to `DEFAULT_MAX_ITERATIONS`=10, explicit `None`→unbounded, positive int caps; `.with_judge` uses `DEFAULT_JUDGE_MAX_ITERATIONS`=5 as its default), `next_message` (defaults to a short "continue" nudge), `return_final_only`, and `additional_instructions` (an extra `system` message injected ahead of the input before the agent runs — becomes part of the original messages so it survives `fresh_context` resets and persists via a session). The judge is configured only through `.with_judge` (`judge_client`/`instructions`/`criteria`), not the constructor, and its `reasoning` is fed back to the agent as the next iteration's input; the judge forwards the original request messages and the agent's latest response messages verbatim so multi-modal content is preserved. `criteria` (a `list[str]`) is both injected as the agent's `additional_instructions` and rendered into the judge instructions wherever the `{{criteria}}` placeholder (`CRITERIA_PLACEHOLDER`) appears (`DEFAULT_JUDGE_INSTRUCTIONS` ends with it; custom `instructions` may include it, and it is stripped when no criteria are given). The `should_continue`/`next_message` callables are invoked with keyword args (`iteration`, `last_result`, `messages`, `original_messages`, `session`, `agent`, `progress`, `feedback`) and may be sync or async; declare only what you need plus `**kwargs`. `should_continue` may return a plain `bool` or a `(bool, str | None)` tuple whose second item is feedback surfaced to `next_message`/`record_feedback` via the `feedback` kwarg (the judge uses this to relay its `reasoning`). Stop precedence per iteration is `max_iterations` → `should_continue`, evaluated before `record_feedback` so the feedback is available to it.
|
||||
|
||||
@@ -22,6 +22,7 @@ import uuid
|
||||
import weakref
|
||||
from abc import abstractmethod
|
||||
from base64 import urlsafe_b64encode
|
||||
from collections import deque
|
||||
from collections.abc import AsyncIterable, Awaitable, Callable, Iterable, Mapping, Sequence
|
||||
from contextlib import suppress
|
||||
from pathlib import Path
|
||||
@@ -34,6 +35,7 @@ from ._types import (
|
||||
AgentRunInputs,
|
||||
ChatResponse,
|
||||
ChatResponseUpdate,
|
||||
Content,
|
||||
Message,
|
||||
ResponseStream,
|
||||
_build_agent_response_from_chat_response, # pyright: ignore[reportPrivateUsage]
|
||||
@@ -477,6 +479,76 @@ class ContextProvider:
|
||||
"""
|
||||
|
||||
|
||||
def _is_approval_placeholder_result(content: Content) -> bool:
|
||||
result = getattr(content, "result", None)
|
||||
return isinstance(result, str) and "[APPROVAL_PENDING]" in result
|
||||
|
||||
|
||||
def _approval_controls_to_keep(messages: Sequence[Message]) -> set[int]:
|
||||
unresolved_requests_by_id: dict[str, Content] = {}
|
||||
unresolved_local_responses_by_id: dict[str, Content] = {}
|
||||
local_response_ids_by_call_id: dict[str, deque[str]] = {}
|
||||
|
||||
for message in messages:
|
||||
for content in message.contents:
|
||||
if content.type == "function_approval_request":
|
||||
function_call = content.function_call
|
||||
if content.id is not None and function_call is not None and function_call.call_id is not None:
|
||||
unresolved_requests_by_id.setdefault(content.id, content)
|
||||
continue
|
||||
if content.type == "function_approval_response":
|
||||
function_call = content.function_call
|
||||
if content.id is not None:
|
||||
unresolved_requests_by_id.pop(content.id, None)
|
||||
if (
|
||||
content.id is not None
|
||||
and function_call is not None
|
||||
and function_call.call_id is not None
|
||||
and not function_call.additional_properties.get("server_label")
|
||||
and content.id not in unresolved_local_responses_by_id
|
||||
):
|
||||
unresolved_local_responses_by_id[content.id] = content
|
||||
local_response_ids_by_call_id.setdefault(function_call.call_id, deque()).append(content.id)
|
||||
continue
|
||||
if content.call_id is None:
|
||||
continue
|
||||
is_terminal_result = content.type == "function_result" and not _is_approval_placeholder_result(content)
|
||||
is_follow_up_request = content.user_input_request and content.type not in {
|
||||
"function_approval_request",
|
||||
"function_approval_response",
|
||||
}
|
||||
if not (is_terminal_result or is_follow_up_request):
|
||||
continue
|
||||
if response_ids := local_response_ids_by_call_id.get(content.call_id):
|
||||
unresolved_local_responses_by_id.pop(response_ids.popleft(), None)
|
||||
|
||||
return {
|
||||
id(content) for content in (*unresolved_requests_by_id.values(), *unresolved_local_responses_by_id.values())
|
||||
}
|
||||
|
||||
|
||||
def _filter_approval_control_messages(messages: Sequence[Message]) -> list[Message]:
|
||||
"""Remove resolved approval controls while preserving pending occurrences."""
|
||||
controls_to_keep = _approval_controls_to_keep(messages)
|
||||
filtered_messages: list[Message] = []
|
||||
for message in messages:
|
||||
filtered_contents = [
|
||||
content
|
||||
for content in message.contents
|
||||
if content.type not in {"function_approval_request", "function_approval_response"}
|
||||
or id(content) in controls_to_keep
|
||||
]
|
||||
if not filtered_contents:
|
||||
continue
|
||||
if len(filtered_contents) == len(message.contents):
|
||||
filtered_messages.append(message)
|
||||
continue
|
||||
filtered_message = copy.copy(message)
|
||||
filtered_message.contents = filtered_contents
|
||||
filtered_messages.append(filtered_message)
|
||||
return filtered_messages
|
||||
|
||||
|
||||
class HistoryProvider(ContextProvider):
|
||||
"""Base class for conversation history storage providers.
|
||||
|
||||
@@ -579,7 +651,7 @@ class HistoryProvider(ContextProvider):
|
||||
state: dict[str, Any],
|
||||
) -> None:
|
||||
"""Load history into context. Skipped by the agent when load_messages=False."""
|
||||
history = await self.get_messages(context.session_id, state=state)
|
||||
history = _filter_approval_control_messages(await self.get_messages(context.session_id, state=state))
|
||||
context.extend_messages(self, history)
|
||||
|
||||
async def after_run(
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -2,6 +2,13 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import warnings
|
||||
from collections.abc import MutableSequence
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from agent_framework import (
|
||||
DEFAULT_TOOL_APPROVAL_SOURCE_ID,
|
||||
Agent,
|
||||
@@ -9,6 +16,7 @@ from agent_framework import (
|
||||
ChatResponse,
|
||||
ChatResponseUpdate,
|
||||
Content,
|
||||
FileHistoryProvider,
|
||||
Message,
|
||||
ToolApprovalMiddleware,
|
||||
ToolApprovalState,
|
||||
@@ -16,6 +24,7 @@ from agent_framework import (
|
||||
create_always_approve_tool_with_arguments_response,
|
||||
tool,
|
||||
)
|
||||
from agent_framework._feature_stage import ExperimentalWarning
|
||||
|
||||
from .conftest import MockBaseChatClient
|
||||
|
||||
@@ -31,6 +40,381 @@ def _function_call(request: Content) -> Content:
|
||||
return request.function_call
|
||||
|
||||
|
||||
@pytest.mark.parametrize("approved", [True, False], ids=["approved", "rejected"])
|
||||
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
|
||||
async def test_approval_resume_returns_result_without_mutating_inputs(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
approved: bool,
|
||||
streaming: bool,
|
||||
) -> None:
|
||||
"""Approval resume should return its terminal result without changing caller-owned messages."""
|
||||
calls = 0
|
||||
|
||||
@tool(name="guarded_tool", approval_mode="always_require")
|
||||
def guarded_tool() -> str:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
return "approved result"
|
||||
|
||||
agent = Agent(client=chat_client_base, tools=[guarded_tool])
|
||||
session = AgentSession(session_id=f"immutable-approval-{streaming}-{approved}")
|
||||
function_call = Content.from_function_call(call_id="call_guarded", name="guarded_tool", arguments="{}")
|
||||
|
||||
if streaming:
|
||||
chat_client_base.streaming_responses = [[ChatResponseUpdate(role="assistant", contents=[function_call])]]
|
||||
first_stream = agent.run("run guarded", stream=True, session=session)
|
||||
first_updates = [update async for update in first_stream]
|
||||
first_response = await first_stream.get_final_response()
|
||||
approval_request = next(content for update in first_updates for content in update.user_input_requests)
|
||||
else:
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=[function_call]))]
|
||||
first_response = await agent.run("run guarded", session=session)
|
||||
approval_request = first_response.user_input_requests[0]
|
||||
|
||||
approval_response = approval_request.to_function_approval_response(approved=approved)
|
||||
approval_message = Message(role="user", contents=[approval_response])
|
||||
|
||||
if streaming:
|
||||
chat_client_base.streaming_responses = [
|
||||
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])]
|
||||
]
|
||||
second_stream = agent.run(approval_message, stream=True, session=session)
|
||||
second_updates = [update async for update in second_stream]
|
||||
second_response = await second_stream.get_final_response()
|
||||
assert [[content.type for content in update.contents] for update in second_updates] == [
|
||||
["function_result"],
|
||||
["text"],
|
||||
]
|
||||
else:
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
|
||||
second_response = await agent.run(approval_message, session=session)
|
||||
|
||||
assert [[content.type for content in message.contents] for message in second_response.messages] == [
|
||||
["function_result"],
|
||||
["text"],
|
||||
]
|
||||
result = second_response.messages[0].contents[0]
|
||||
assert result.call_id == "call_guarded"
|
||||
assert result.result == ("approved result" if approved else "Error: Tool call invocation was rejected by user.")
|
||||
assert calls == int(approved)
|
||||
assert approval_message.role == "user"
|
||||
assert approval_message.contents == [approval_response]
|
||||
assert [[content.type for content in message.contents] for message in first_response.messages] == [
|
||||
["function_call", "function_approval_request"]
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
|
||||
async def test_approval_resume_replays_reasoning_with_function_call_group(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
streaming: bool,
|
||||
) -> None:
|
||||
"""Reasoning bound to a function call must be replayed with its terminal result."""
|
||||
|
||||
@tool(name="reasoning_tool", approval_mode="always_require")
|
||||
def reasoning_tool() -> str:
|
||||
return "approved result"
|
||||
|
||||
captured_calls: list[list[tuple[str, list[tuple[str, str | None, str | None, str | None]]]]] = []
|
||||
|
||||
def capture(messages: MutableSequence[Message]) -> None:
|
||||
captured_calls.append([
|
||||
(
|
||||
message.role,
|
||||
[(content.type, content.id, content.call_id, content.protected_data) for content in message.contents],
|
||||
)
|
||||
for message in messages
|
||||
])
|
||||
|
||||
if streaming:
|
||||
original_get_streaming_response = chat_client_base._get_streaming_response
|
||||
|
||||
def capture_streaming_messages(
|
||||
*,
|
||||
messages: MutableSequence[Message],
|
||||
options: dict[str, Any],
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
capture(messages)
|
||||
return original_get_streaming_response(messages=messages, options=options, **kwargs)
|
||||
|
||||
chat_client_base._get_streaming_response = capture_streaming_messages # type: ignore[method-assign] # ty: ignore[invalid-assignment]
|
||||
else:
|
||||
original_get_non_streaming_response = chat_client_base._get_non_streaming_response
|
||||
|
||||
async def capture_non_streaming_messages(
|
||||
*,
|
||||
messages: MutableSequence[Message],
|
||||
options: dict[str, Any],
|
||||
**kwargs: Any,
|
||||
) -> ChatResponse:
|
||||
capture(messages)
|
||||
return await original_get_non_streaming_response(messages=messages, options=options, **kwargs)
|
||||
|
||||
chat_client_base._get_non_streaming_response = capture_non_streaming_messages # type: ignore[method-assign] # ty: ignore[invalid-assignment]
|
||||
|
||||
agent = Agent(client=chat_client_base, tools=[reasoning_tool])
|
||||
session = AgentSession(session_id=f"approval-reasoning-{streaming}")
|
||||
reasoning = Content.from_text_reasoning(
|
||||
id="reasoning_1",
|
||||
text="I need to run the tool",
|
||||
protected_data="encrypted-reasoning",
|
||||
additional_properties={"status": "completed"},
|
||||
)
|
||||
function_call = Content.from_function_call(
|
||||
call_id="call_reasoning",
|
||||
name="reasoning_tool",
|
||||
arguments="{}",
|
||||
)
|
||||
|
||||
if streaming:
|
||||
chat_client_base.streaming_responses = [
|
||||
[
|
||||
ChatResponseUpdate(role="assistant", contents=[reasoning]),
|
||||
ChatResponseUpdate(role="assistant", contents=[function_call]),
|
||||
],
|
||||
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("done")])],
|
||||
]
|
||||
first_stream = agent.run("run with reasoning", stream=True, session=session)
|
||||
first_updates = [update async for update in first_stream]
|
||||
first_response = await first_stream.get_final_response()
|
||||
approval_request = next(content for update in first_updates for content in update.user_input_requests)
|
||||
resumed_stream = agent.run(
|
||||
approval_request.to_function_approval_response(approved=True),
|
||||
stream=True,
|
||||
session=session,
|
||||
)
|
||||
_ = [update async for update in resumed_stream]
|
||||
resumed_response = await resumed_stream.get_final_response()
|
||||
else:
|
||||
chat_client_base.run_responses = [
|
||||
ChatResponse(messages=Message(role="assistant", contents=[reasoning, function_call])),
|
||||
ChatResponse(messages=Message(role="assistant", contents=["done"])),
|
||||
]
|
||||
first_response = await agent.run("run with reasoning", session=session)
|
||||
resumed_response = await agent.run(
|
||||
first_response.user_input_requests[0].to_function_approval_response(approved=True),
|
||||
session=session,
|
||||
)
|
||||
|
||||
replayed_contents = [content for _, contents in captured_calls[1] for content in contents]
|
||||
replayed_types = [content_type for content_type, _, _, _ in replayed_contents]
|
||||
reasoning_index = replayed_types.index("text_reasoning")
|
||||
call_index = replayed_types.index("function_call")
|
||||
result_index = replayed_types.index("function_result")
|
||||
|
||||
assert reasoning_index < call_index < result_index
|
||||
assert replayed_contents[reasoning_index] == (
|
||||
"text_reasoning",
|
||||
"reasoning_1",
|
||||
None,
|
||||
"encrypted-reasoning",
|
||||
)
|
||||
assert "function_approval_request" not in replayed_types
|
||||
assert "function_approval_response" not in replayed_types
|
||||
assert any(content.type == "text_reasoning" for content in first_response.messages[0].contents)
|
||||
assert [[content.type for content in message.contents] for message in resumed_response.messages] == [
|
||||
["function_result"],
|
||||
["text"],
|
||||
]
|
||||
|
||||
|
||||
async def test_approval_resume_filters_resolved_control_items_from_file_history(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Resolved approval wrappers should not be replayed from append-only history."""
|
||||
calls = 0
|
||||
|
||||
@tool(name="guarded_history_tool", approval_mode="always_require")
|
||||
def guarded_history_tool() -> str:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
return "approved result"
|
||||
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore", ExperimentalWarning)
|
||||
history_provider = FileHistoryProvider(tmp_path)
|
||||
agent = Agent(
|
||||
client=chat_client_base,
|
||||
tools=[guarded_history_tool],
|
||||
context_providers=[history_provider],
|
||||
)
|
||||
session = AgentSession(session_id="approval-file-history")
|
||||
chat_client_base.run_responses = [
|
||||
ChatResponse(
|
||||
messages=Message(
|
||||
role="assistant",
|
||||
contents=[
|
||||
Content.from_function_call(
|
||||
call_id="call_guarded_history",
|
||||
name="guarded_history_tool",
|
||||
arguments="{}",
|
||||
)
|
||||
],
|
||||
)
|
||||
)
|
||||
]
|
||||
first_response = await agent.run("run guarded", session=session)
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
|
||||
await agent.run(
|
||||
first_response.user_input_requests[0].to_function_approval_response(approved=True),
|
||||
session=session,
|
||||
)
|
||||
|
||||
captured_types: list[list[str]] = []
|
||||
original_get_response = chat_client_base._get_non_streaming_response
|
||||
|
||||
async def capture_messages(
|
||||
*,
|
||||
messages: MutableSequence[Message],
|
||||
options: dict[str, Any],
|
||||
**kwargs: Any,
|
||||
) -> ChatResponse:
|
||||
captured_types.extend([[content.type for content in message.contents] for message in messages])
|
||||
return await original_get_response(messages=messages, options=options, **kwargs)
|
||||
|
||||
chat_client_base._get_non_streaming_response = capture_messages # type: ignore[method-assign] # ty: ignore[invalid-assignment]
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["later"]))]
|
||||
await agent.run("unrelated later turn", session=session)
|
||||
|
||||
flattened_types = [content_type for message_types in captured_types for content_type in message_types]
|
||||
assert flattened_types.count("function_call") == 1
|
||||
assert flattened_types.count("function_result") == 1
|
||||
assert "function_approval_request" not in flattened_types
|
||||
assert "function_approval_response" not in flattened_types
|
||||
assert calls == 1
|
||||
|
||||
|
||||
async def test_pending_approval_from_file_history_stays_resumable_without_model_orphan(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""An unrelated turn hides a pending batch from the model without discarding its approval."""
|
||||
calls = 0
|
||||
|
||||
@tool(name="guarded_pending_tool", approval_mode="always_require")
|
||||
def guarded_pending_tool() -> str:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
return "approved result"
|
||||
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore", ExperimentalWarning)
|
||||
history_provider = FileHistoryProvider(tmp_path)
|
||||
agent = Agent(
|
||||
client=chat_client_base,
|
||||
tools=[guarded_pending_tool],
|
||||
context_providers=[history_provider],
|
||||
)
|
||||
session = AgentSession(session_id="pending-approval-file-history")
|
||||
function_call = Content.from_function_call(
|
||||
call_id="call_pending_history",
|
||||
name="guarded_pending_tool",
|
||||
arguments="{}",
|
||||
)
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=[function_call]))]
|
||||
first_response = await agent.run("run guarded", session=session)
|
||||
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["unrelated answer"]))]
|
||||
captured_types: list[str] = []
|
||||
original_get_response = chat_client_base._get_non_streaming_response
|
||||
|
||||
async def capture_messages(
|
||||
*,
|
||||
messages: MutableSequence[Message],
|
||||
options: dict[str, Any],
|
||||
**kwargs: Any,
|
||||
) -> ChatResponse:
|
||||
captured_types.extend(content.type for message in messages for content in message.contents)
|
||||
return await original_get_response(messages=messages, options=options, **kwargs)
|
||||
|
||||
chat_client_base._get_non_streaming_response = capture_messages # type: ignore[method-assign] # ty: ignore[invalid-assignment]
|
||||
unrelated_response = await agent.run("unrelated turn", session=session)
|
||||
|
||||
assert unrelated_response.text == "unrelated answer"
|
||||
assert "function_call" not in captured_types
|
||||
assert "function_approval_request" not in captured_types
|
||||
assert calls == 0
|
||||
|
||||
chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=["done"]))]
|
||||
resumed_response = await agent.run(
|
||||
first_response.user_input_requests[0].to_function_approval_response(approved=True),
|
||||
session=session,
|
||||
)
|
||||
|
||||
assert resumed_response.text == "done"
|
||||
assert calls == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
|
||||
async def test_approval_resume_returns_all_user_input_requests_without_another_model_call(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
streaming: bool,
|
||||
) -> None:
|
||||
"""All user input requested during approved execution should return before another model call."""
|
||||
from agent_framework.exceptions import UserInputRequiredException
|
||||
|
||||
calls = 0
|
||||
|
||||
@tool(name="oauth_tool", approval_mode="always_require")
|
||||
def oauth_tool() -> str:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
raise UserInputRequiredException(
|
||||
contents=[
|
||||
Content.from_oauth_consent_request(consent_link="https://example.com/consent-1"),
|
||||
Content.from_oauth_consent_request(consent_link="https://example.com/consent-2"),
|
||||
Content.from_oauth_consent_request(consent_link="https://example.com/consent-3"),
|
||||
]
|
||||
)
|
||||
|
||||
agent = Agent(client=chat_client_base, tools=[oauth_tool])
|
||||
session = AgentSession(session_id=f"approval-user-input-{streaming}")
|
||||
function_call = Content.from_function_call(call_id="call_oauth", name="oauth_tool", arguments="{}")
|
||||
|
||||
if streaming:
|
||||
chat_client_base.streaming_responses = [
|
||||
[ChatResponseUpdate(role="assistant", contents=[function_call])],
|
||||
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("unexpected model call")])],
|
||||
]
|
||||
first_stream = agent.run("run oauth", stream=True, session=session)
|
||||
first_updates = [update async for update in first_stream]
|
||||
approval_request = next(content for update in first_updates for content in update.user_input_requests)
|
||||
resumed_stream = agent.run(
|
||||
approval_request.to_function_approval_response(approved=True),
|
||||
stream=True,
|
||||
session=session,
|
||||
)
|
||||
resumed_updates = [update async for update in resumed_stream]
|
||||
resumed_response = await resumed_stream.get_final_response()
|
||||
assert len(chat_client_base.streaming_responses) == 1
|
||||
assert [content.consent_link for update in resumed_updates for content in update.user_input_requests] == [
|
||||
"https://example.com/consent-1",
|
||||
"https://example.com/consent-2",
|
||||
"https://example.com/consent-3",
|
||||
]
|
||||
else:
|
||||
chat_client_base.run_responses = [
|
||||
ChatResponse(messages=Message(role="assistant", contents=[function_call])),
|
||||
ChatResponse(messages=Message(role="assistant", contents=["unexpected model call"])),
|
||||
]
|
||||
first_response = await agent.run("run oauth", session=session)
|
||||
resumed_response = await agent.run(
|
||||
first_response.user_input_requests[0].to_function_approval_response(approved=True),
|
||||
session=session,
|
||||
)
|
||||
assert len(chat_client_base.run_responses) == 1
|
||||
|
||||
assert [content.consent_link for content in resumed_response.user_input_requests] == [
|
||||
"https://example.com/consent-1",
|
||||
"https://example.com/consent-2",
|
||||
"https://example.com/consent-3",
|
||||
]
|
||||
assert resumed_response.messages[0].role == "assistant"
|
||||
assert calls == 1
|
||||
|
||||
|
||||
async def test_mixed_batch_hides_already_approved_request_until_approval_replay(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
) -> None:
|
||||
@@ -482,6 +866,54 @@ async def test_tool_approval_middleware_auto_approved_loops_share_function_call_
|
||||
assert calls == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True], ids=["non-streaming", "streaming"])
|
||||
async def test_auto_approval_resolves_after_iteration_budget_is_exhausted(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
streaming: bool,
|
||||
) -> None:
|
||||
"""Approval resolution must run before the model-iteration budget is checked."""
|
||||
calls = 0
|
||||
|
||||
@tool(name="last_iteration_tool", approval_mode="always_require")
|
||||
def last_iteration_tool() -> str:
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
return "executed"
|
||||
|
||||
chat_client_base.function_invocation_configuration["max_iterations"] = 1
|
||||
agent = Agent(
|
||||
client=chat_client_base,
|
||||
tools=[last_iteration_tool],
|
||||
middleware=[ToolApprovalMiddleware(auto_approval_rules=[lambda function_call: True])],
|
||||
)
|
||||
session = AgentSession(session_id=f"approval-iteration-budget-{streaming}")
|
||||
function_call = Content.from_function_call(
|
||||
call_id="call_last_iteration",
|
||||
name="last_iteration_tool",
|
||||
arguments="{}",
|
||||
)
|
||||
|
||||
if streaming:
|
||||
chat_client_base.streaming_responses = [
|
||||
[ChatResponseUpdate(role="assistant", contents=[function_call])],
|
||||
[ChatResponseUpdate(role="assistant", contents=[Content.from_text("unused")])],
|
||||
]
|
||||
response_stream = agent.run("run once", stream=True, session=session)
|
||||
updates = [update async for update in response_stream]
|
||||
response = await response_stream.get_final_response()
|
||||
assert any(content.type == "function_result" for update in updates for content in update.contents)
|
||||
else:
|
||||
chat_client_base.run_responses = [
|
||||
ChatResponse(messages=Message(role="assistant", contents=[function_call])),
|
||||
ChatResponse(messages=Message(role="assistant", contents=["unused"])),
|
||||
]
|
||||
response = await agent.run("run once", session=session)
|
||||
|
||||
assert calls == 1
|
||||
assert any(content.type == "function_result" for message in response.messages for content in message.contents)
|
||||
assert response.text == "I broke out of the function invocation loop..."
|
||||
|
||||
|
||||
async def test_tool_approval_middleware_queues_streamed_approval_requests(
|
||||
chat_client_base: MockBaseChatClient,
|
||||
) -> None:
|
||||
|
||||
@@ -14,6 +14,7 @@ from agent_framework import (
|
||||
AgentContext,
|
||||
AgentSession,
|
||||
ChatContext,
|
||||
Content,
|
||||
ContextProvider,
|
||||
ExperimentalFeature,
|
||||
FileHistoryProvider,
|
||||
@@ -24,7 +25,11 @@ from agent_framework import (
|
||||
agent_middleware,
|
||||
chat_middleware,
|
||||
)
|
||||
from agent_framework._sessions import LOCAL_HISTORY_CONVERSATION_ID, is_local_history_conversation_id
|
||||
from agent_framework._sessions import (
|
||||
LOCAL_HISTORY_CONVERSATION_ID,
|
||||
_filter_approval_control_messages,
|
||||
is_local_history_conversation_id,
|
||||
)
|
||||
from agent_framework.exceptions import MiddlewareException
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -348,6 +353,83 @@ class ConcreteHistoryProvider(HistoryProvider):
|
||||
self.stored.extend(messages)
|
||||
|
||||
|
||||
def test_filter_approval_controls_preserves_only_unresolved_occurrences() -> None:
|
||||
local_call = Content.from_function_call(call_id="local", name="guarded", arguments="{}")
|
||||
local_request = Content.from_function_approval_request(id="local_approval", function_call=local_call)
|
||||
hosted_call = Content.from_function_call(
|
||||
call_id="hosted",
|
||||
name="hosted_tool",
|
||||
arguments="{}",
|
||||
additional_properties={"server_label": "server"},
|
||||
)
|
||||
hosted_request = Content.from_function_approval_request(id="hosted_approval", function_call=hosted_call)
|
||||
resolved_call = Content.from_function_call(call_id="resolved", name="guarded", arguments="{}")
|
||||
resolved_request = Content.from_function_approval_request(id="resolved_approval", function_call=resolved_call)
|
||||
resolved_response = resolved_request.to_function_approval_response(approved=True)
|
||||
|
||||
filtered = _filter_approval_control_messages([
|
||||
Message(role="assistant", contents=[local_call, local_request]),
|
||||
Message(role="assistant", contents=[hosted_request]),
|
||||
Message(role="assistant", contents=[resolved_call, resolved_request]),
|
||||
Message(role="user", contents=[resolved_response]),
|
||||
Message(role="tool", contents=[Content.from_function_result(call_id="resolved", result="done")]),
|
||||
])
|
||||
|
||||
controls = [
|
||||
content
|
||||
for message in filtered
|
||||
for content in message.contents
|
||||
if content.type in {"function_approval_request", "function_approval_response"}
|
||||
]
|
||||
assert controls == [local_request, hosted_request]
|
||||
assert any(
|
||||
content.type == "function_result" and content.call_id == "resolved"
|
||||
for message in filtered
|
||||
for content in message.contents
|
||||
)
|
||||
|
||||
|
||||
def test_filter_approval_controls_deduplicates_pending_request_replay() -> None:
|
||||
function_call = Content.from_function_call(call_id="call_1", name="guarded", arguments="{}")
|
||||
request = Content.from_function_approval_request(id="approval_1", function_call=function_call)
|
||||
replayed_request = Content.from_dict(request.to_dict())
|
||||
|
||||
filtered = _filter_approval_control_messages([
|
||||
Message(role="assistant", contents=[function_call, request]),
|
||||
Message(role="assistant", contents=[replayed_request]),
|
||||
])
|
||||
|
||||
requests = [
|
||||
content for message in filtered for content in message.contents if content.type == "function_approval_request"
|
||||
]
|
||||
assert requests == [request]
|
||||
|
||||
|
||||
def test_filter_approval_controls_keeps_response_for_pending_placeholder() -> None:
|
||||
function_call = Content.from_function_call(call_id="call_pending", name="guarded", arguments="{}")
|
||||
request = Content.from_function_approval_request(id="approval_pending", function_call=function_call)
|
||||
response = request.to_function_approval_response(approved=True)
|
||||
placeholder = Content.from_function_result(
|
||||
call_id="call_pending",
|
||||
result="[APPROVAL_PENDING] waiting for execution",
|
||||
)
|
||||
|
||||
filtered = _filter_approval_control_messages([
|
||||
Message(role="assistant", contents=[function_call, request]),
|
||||
Message(role="user", contents=[response]),
|
||||
Message(role="tool", contents=[placeholder]),
|
||||
])
|
||||
|
||||
controls = [
|
||||
content
|
||||
for message in filtered
|
||||
for content in message.contents
|
||||
if content.type in {"function_approval_request", "function_approval_response"}
|
||||
]
|
||||
assert controls == [response]
|
||||
assert any(placeholder in message.contents for message in filtered)
|
||||
|
||||
|
||||
class TestHistoryProviderBase:
|
||||
def test_default_flags(self) -> None:
|
||||
provider = ConcreteHistoryProvider("mem")
|
||||
|
||||
@@ -33,6 +33,7 @@ from agent_framework._sessions import (
|
||||
AgentSession,
|
||||
InMemoryHistoryProvider,
|
||||
SessionContext,
|
||||
_filter_approval_control_messages,
|
||||
)
|
||||
from agent_framework._workflows._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
|
||||
from agent_framework.exceptions import (
|
||||
@@ -7818,6 +7819,38 @@ def test_prepare_messages_strips_approval_items_under_storage() -> None:
|
||||
assert "mcp_approval_response" in storage_off_types
|
||||
|
||||
|
||||
def test_stateless_history_preserves_pending_hosted_approval_request_until_response() -> None:
|
||||
client = OpenAIChatClient(model="test-model", api_key="test-key")
|
||||
function_call = Content.from_function_call(
|
||||
call_id="mcp_pending",
|
||||
name="sensitive_action",
|
||||
arguments='{"action": "delete"}',
|
||||
additional_properties={"server_label": "hosted_server"},
|
||||
)
|
||||
approval_request = Content.from_function_approval_request(
|
||||
id="approval_pending",
|
||||
function_call=function_call,
|
||||
)
|
||||
approval_response = approval_request.to_function_approval_response(approved=True)
|
||||
|
||||
pending_history = _filter_approval_control_messages([Message(role="assistant", contents=[approval_request])])
|
||||
pending_items = client._prepare_messages_for_openai(
|
||||
pending_history,
|
||||
request_uses_service_side_storage=False,
|
||||
)
|
||||
assert [item.get("type") for item in pending_items] == ["mcp_approval_request"]
|
||||
|
||||
resolved_history = _filter_approval_control_messages([
|
||||
Message(role="assistant", contents=[approval_request]),
|
||||
Message(role="user", contents=[approval_response]),
|
||||
])
|
||||
resolved_items = client._prepare_messages_for_openai(
|
||||
resolved_history,
|
||||
request_uses_service_side_storage=False,
|
||||
)
|
||||
assert resolved_items == []
|
||||
|
||||
|
||||
def test_prepare_messages_strips_local_shell_call_under_storage() -> None:
|
||||
"""Local-shell-call function_results carry a server-issued local_shell_call_item_id and must
|
||||
be stripped under storage. Plain function_results (no shell ID) are kept either way (#3295)."""
|
||||
|
||||
@@ -1090,6 +1090,31 @@ def test_function_approval_content_is_skipped_in_preparation(
|
||||
assert prepared_mixed[0]["content"] == "I need approval for this action."
|
||||
|
||||
|
||||
def test_mixed_approval_resume_roles_serialize_function_result_as_tool(
|
||||
openai_unit_test_env: dict[str, str],
|
||||
) -> None:
|
||||
client = OpenAIChatCompletionClient()
|
||||
follow_up_request = Content.from_oauth_consent_request(consent_link="https://example.com/consent")
|
||||
follow_up_request.call_id = "call_paused"
|
||||
messages = [
|
||||
Message(
|
||||
role="tool",
|
||||
contents=[Content.from_function_result(call_id="call_completed", result="completed")],
|
||||
),
|
||||
Message(role="assistant", contents=[follow_up_request]),
|
||||
]
|
||||
|
||||
prepared = client._prepare_messages_for_openai(messages)
|
||||
|
||||
assert prepared[0] == {
|
||||
"role": "tool",
|
||||
"tool_call_id": "call_completed",
|
||||
"content": "completed",
|
||||
}
|
||||
assert prepared[1]["role"] == "assistant"
|
||||
assert "tool_call_id" not in prepared[1]
|
||||
|
||||
|
||||
def test_usage_content_in_streaming_response(
|
||||
openai_unit_test_env: dict[str, str],
|
||||
) -> None:
|
||||
|
||||
Reference in New Issue
Block a user