github-featured/openai--openai-agents-python
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix: honor exact call approval decisions (#4447)

Browse Source 
Co-authored-by: chiruu12 <103719146+chiruu12@users.noreply.github.com>
This commit is contained in:
Kazuhiro Sera
2026-08-16 08:40:44 +09:00
committed by GitHub
parent 05c789fe4d
commit dde0bc99fd
9 changed files with 564 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/run_context.py
+23 -13
View File
@@ -629,17 +629,6 @@ class RunContextWrapper(Generic[TContext]):
if approval_entry is None:
return None
# Check for permanent approval/rejection
if approval_entry.approved is True and approval_entry.rejected is True:
# Approval takes precedence
return True
if approval_entry.approved is True:
return True
if approval_entry.rejected is True:
return False
approved_ids = (
set(approval_entry.approved) if isinstance(approval_entry.approved, list) else set()
)
@@ -651,6 +640,18 @@ class RunContextWrapper(Generic[TContext]):
return True
if call_id in rejected_ids:
return False
# Exact call decisions override sticky defaults for the same approval key.
if approval_entry.approved is True and approval_entry.rejected is True:
# Approval takes precedence when sticky decisions conflict.
return True
if approval_entry.approved is True:
return True
if approval_entry.rejected is True:
return False
# Per-call approvals are scoped to the exact call ID, so other calls require a new decision.
return None
@@ -685,6 +686,8 @@ class RunContextWrapper(Generic[TContext]):
@staticmethod
def _get_rejection_message_for_key(record: _ApprovalRecord, call_id: str) -> str | None:
if isinstance(record.approved, list) and call_id in record.approved:
return None
if record.rejected is True:
if call_id in record.rejection_messages:
return record.rejection_messages[call_id]
@@ -1011,8 +1014,15 @@ class RunContextWrapper(Generic[TContext]):
opposite.remove(call_id)
target = approval_entry.approved if approve else approval_entry.rejected
if isinstance(target, list) and call_id not in target:
target.append(call_id)
if target is not True:
if not isinstance(target, list):
target = []
if approve:
approval_entry.approved = target
else:
approval_entry.rejected = target
if call_id not in target:
target.append(call_id)
if approve:
self._clear_rejection_message(approval_entry, call_id)
elif call_id is not None:
src/agents/run_state.py
+2 -1
View File
@@ -179,7 +179,7 @@ def _default_run_state_validation_error(
# 3. to_json() always emits CURRENT_SCHEMA_VERSION.
# 4. Forward compatibility is intentionally fail-fast (older SDKs reject newer or unsupported
# versions).
CURRENT_SCHEMA_VERSION = "1.15"
CURRENT_SCHEMA_VERSION = "1.16"
_PROGRAMMATIC_TOOL_CALLING_MIN_SCHEMA_VERSION = "1.13"
_HOSTED_MCP_APPROVALS_MIN_SCHEMA_VERSION = "1.14"
# Keep this mapping in chronological order. Every schema bump must add a one-line summary here.
@@ -209,6 +209,7 @@ SCHEMA_VERSION_SUMMARIES: dict[str, str] = {
"Persists canonical tool invocation identity plus sanitized mount authority and trusted "
"rebind metadata, durable pending input, and resumable next-model-call state."
),
"1.16": "Lets an exact call approval decision override a sticky decision for the same tool.",
}
SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES)
tests/fixtures/run_state/README.md
Vendored
+2 -2
View File
@@ -1,6 +1,6 @@
# RunState compatibility corpus
The `minimal/` fixtures cover every schema version accepted by the current reader. The `features/` fixtures cover the schema-bearing behavior introduced in versions 1.2 through 1.15. The `resume/` fixture records an actual pending function-tool approval emitted by the v0.19.4 writer, and the `security/` fixture records that writer's credential-bearing sandbox state. `sources.json` records the source commit and provenance for every fixture.
The `minimal/` fixtures cover every schema version accepted by the current reader. The `features/` fixtures cover the schema-bearing behavior introduced in versions 1.2 through 1.16. The `resume/` fixture records an actual pending function-tool approval emitted by the v0.19.4 writer, and the `security/` fixture records that writer's credential-bearing sandbox state. `sources.json` records the source commit and provenance for every fixture.
Regenerate the feature corpus from the recorded historical source trees with:
@@ -10,6 +10,6 @@ UV_DEFAULT_INDEX=https://pypi.org/simple uv run python tests/fixtures/run_state/
The generator extracts each recorded commit with `git archive` and runs that commit's writer in a fresh locked environment. It does not import the current checkout.
Versions 1.7 and 1.8 are explicit exceptions. Release-boundary schema renumbering assigned duplicate-agent/sandbox state to 1.7 and prompt-cache state to 1.8 without any writer commit that emitted those final version numbers. Their fixtures are therefore marked `canonical_compatibility`: the recorded 1.9 writer produces the payload, and the generator changes only the schema label so the corresponding reader branch remains covered. They must not be represented as historical-writer output.
Versions 1.7, 1.8, and 1.16 are explicit exceptions. Release-boundary schema renumbering assigned duplicate-agent/sandbox state to 1.7 and prompt-cache state to 1.8 without any writer commit that emitted those final version numbers. The 1.16 schema transition likewise has no retained writer commit, but the retained 1.15 writer produces the same minimal and per-call-override payloads. These fixtures are therefore marked `canonical_compatibility`: the recorded writer produces the payload, and the generator changes only the schema label so the corresponding reader branch remains covered. They must not be represented as historical-writer output.
Ordinary tests never run the generator. They read the frozen payloads, compare every durable field emitted by the historical writer across the upgrade, rewrite to the current schema, and verify that the rewritten form is idempotent. They also approve and reject the historical pending interruption through actual `Runner` resumes. The schema version itself is the only normalization for the ordinary corpus; fields added by newer writers may be absent from an older payload, but every field present in that payload must survive. The security fixture has one explicit migration normalization: persisted mount credentials and opaque driver options are removed and the trusted-rebind marker is added. All non-authority topology remains part of the comparison.
tests/fixtures/run_state/features/v1_16_per_call_approval_override.json
Vendored
+86
View File
@@ -0,0 +1,86 @@
{
"$schemaVersion": "1.16",
"auto_previous_response_id": false,
"context": {
"approvals": {
"sensitive_tool": {
"approved": true,
"rejected": [
"exception-call"
],
"rejection_messages": {
"exception-call": "Denied exactly"
},
"sticky_scope": "b6a3e5c0378ea20a4f9b146318a050924153377109c454065f977537d8d58300"
}
},
"context": {},
"context_meta": {
"omitted": false,
"original_type": "mapping",
"requires_deserializer": false,
"serialized_via": "mapping"
},
"tool_invocations": {
"exception-call": {
"approval_scope": "b6a3e5c0378ea20a4f9b146318a050924153377109c454065f977537d8d58300",
"completed": false,
"executed": false,
"fingerprint": "23eb3e956e31f397bdbf5fbcc53b72f004e4f61a546ba7ae501c038a0d29f203",
"type": "function_call"
},
"sticky-call": {
"approval_scope": "b6a3e5c0378ea20a4f9b146318a050924153377109c454065f977537d8d58300",
"completed": false,
"executed": false,
"fingerprint": "23eb3e956e31f397bdbf5fbcc53b72f004e4f61a546ba7ae501c038a0d29f203",
"type": "function_call"
}
},
"usage": {
"input_tokens": 0,
"input_tokens_details": [
{
"cache_write_tokens": 0,
"cached_tokens": 0
}
],
"output_tokens": 0,
"output_tokens_details": [
{
"reasoning_tokens": 0
}
],
"request_usage_entries": [],
"requests": 0,
"total_tokens": 0
}
},
"conversation_id": null,
"current_agent": {
"name": "compat-agent"
},
"current_step": null,
"current_turn": 0,
"current_turn_persisted_item_count": 0,
"generated_items": [],
"generated_prompt_cache_key": null,
"generated_session_item_indexes": [],
"input_guardrail_results": [],
"last_model_response": null,
"last_processed_response": null,
"max_turns": 10,
"model_responses": [],
"nested_history_owned_session_item_refs": [],
"no_active_agent_run": true,
"original_input": "historical input",
"output_guardrail_results": [],
"pending_input": [],
"previous_response_id": null,
"reasoning_item_id_policy": null,
"session_items": [],
"tool_input_guardrail_results": [],
"tool_output_guardrail_results": [],
"tool_use_tracker": {},
"trace": null
}
tests/fixtures/run_state/generate_corpus.py
Vendored
+75 -5
View File
@@ -12,6 +12,7 @@ from typing import cast
ROOT = Path(__file__).resolve().parents[3]
OUTPUT = Path(__file__).resolve().parent / "features"
MINIMAL_OUTPUT = Path(__file__).resolve().parent / "minimal"
SECURITY_OUTPUT = Path(__file__).resolve().parent / "security"
RESUME_OUTPUT = Path(__file__).resolve().parent / "resume"
@@ -29,6 +30,12 @@ state = RunState(
)
"""
LEGACY_CANONICAL_COMPATIBILITY_NOTE = (
"The release-boundary schema renumbering introduced this reader version without a writer "
"that emitted it. The recorded writer emitted 1.9; only the schema label is changed to "
"exercise the canonical compatibility branch."
)
@dataclass(frozen=True)
class Scenario:
@@ -38,6 +45,7 @@ class Scenario:
code: str
provenance: str = "historical_writer"
emitted_version: str | None = None
note: str | None = None
SCENARIOS = (
@@ -387,6 +395,54 @@ approval = ToolApprovalItem(
state.approve(approval)
""",
),
Scenario(
"1.16",
"1c3b72019e547fe1cf1530419dc6fc687cc4df39",
"per_call_approval_override",
"""
from agents.items import ToolApprovalItem
from openai.types.responses import ResponseFunctionToolCall
def approval(call_id):
return ToolApprovalItem(
agent=agent,
raw_item=ResponseFunctionToolCall(
type="function_call",
name="sensitive_tool",
call_id=call_id,
status="completed",
arguments="{}",
),
)
state.approve(approval("sticky-call"), always_approve=True)
state.reject(approval("exception-call"), rejection_message="Denied exactly")
""",
provenance="canonical_compatibility",
emitted_version="1.15",
note=(
"The schema transition introduced this reader version without a retained writer "
"commit that emitted it. The recorded writer emitted 1.15; only the schema label is "
"changed to exercise the canonical compatibility branch."
),
),
)
MINIMAL_SCENARIOS = (
Scenario(
"1.16",
"1c3b72019e547fe1cf1530419dc6fc687cc4df39",
"minimal",
"",
provenance="canonical_compatibility",
emitted_version="1.15",
note=(
"The schema transition introduced this reader version without a retained writer "
"commit that emitted it. The recorded writer emitted 1.15; only the schema label is "
"changed to exercise the canonical compatibility branch."
),
),
)
@@ -546,17 +602,31 @@ def main() -> None:
}
if scenario.emitted_version is not None:
source["emitted_version"] = scenario.emitted_version
source["note"] = (
"The release-boundary schema renumbering introduced this reader version "
"without a writer that emitted it. The recorded writer emitted 1.9; only "
"the schema label is changed to exercise the canonical compatibility branch."
)
source["note"] = scenario.note or LEGACY_CANONICAL_COMPATIBILITY_NOTE
feature_sources.append(source)
sources_path = OUTPUT.parent / "sources.json"
sources = json.loads(sources_path.read_text(encoding="utf-8"))
sources["features"] = feature_sources
MINIMAL_OUTPUT.mkdir(parents=True, exist_ok=True)
for scenario in MINIMAL_SCENARIOS:
minimal_payload = _generate(scenario)
minimal_filename = f"v{scenario.version.replace('.', '_')}.json"
(MINIMAL_OUTPUT / minimal_filename).write_text(
json.dumps(minimal_payload, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
minimal_source = {
"commit": scenario.commit,
"fixture": f"minimal/{minimal_filename}",
}
if scenario.emitted_version is not None:
minimal_source["emitted_version"] = scenario.emitted_version
minimal_source["provenance"] = scenario.provenance
minimal_source["note"] = scenario.note or LEGACY_CANONICAL_COMPATIBILITY_NOTE
sources["versions"][scenario.version] = minimal_source
SECURITY_OUTPUT.mkdir(parents=True, exist_ok=True)
security_payload = _generate(LEGACY_MOUNT_CREDENTIALS)
security_filename = "v1_13_legacy_mount_credentials.json"
tests/fixtures/run_state/minimal/v1_16.json
Vendored
+60
View File
@@ -0,0 +1,60 @@
{
"$schemaVersion": "1.16",
"auto_previous_response_id": false,
"context": {
"approvals": {},
"context": {},
"context_meta": {
"omitted": false,
"original_type": "mapping",
"requires_deserializer": false,
"serialized_via": "mapping"
},
"tool_invocations": {},
"usage": {
"input_tokens": 0,
"input_tokens_details": [
{
"cache_write_tokens": 0,
"cached_tokens": 0
}
],
"output_tokens": 0,
"output_tokens_details": [
{
"reasoning_tokens": 0
}
],
"request_usage_entries": [],
"requests": 0,
"total_tokens": 0
}
},
"conversation_id": null,
"current_agent": {
"name": "compat-agent"
},
"current_step": null,
"current_turn": 0,
"current_turn_persisted_item_count": 0,
"generated_items": [],
"generated_prompt_cache_key": null,
"generated_session_item_indexes": [],
"input_guardrail_results": [],
"last_model_response": null,
"last_processed_response": null,
"max_turns": 10,
"model_responses": [],
"nested_history_owned_session_item_refs": [],
"no_active_agent_run": true,
"original_input": "historical input",
"output_guardrail_results": [],
"pending_input": [],
"previous_response_id": null,
"reasoning_item_id_policy": null,
"session_items": [],
"tool_input_guardrail_results": [],
"tool_output_guardrail_results": [],
"tool_use_tracker": {},
"trace": null
}
tests/fixtures/run_state/sources.json
Vendored
+16
View File
@@ -109,6 +109,15 @@
"fixture": "features/v1_15_canonical_invocation_identity.json",
"provenance": "historical_writer",
"version": "1.15"
},
{
"commit": "1c3b72019e547fe1cf1530419dc6fc687cc4df39",
"emitted_version": "1.15",
"feature": "per_call_approval_override",
"fixture": "features/v1_16_per_call_approval_override.json",
"note": "The schema transition introduced this reader version without a retained writer commit that emitted it. The recorded writer emitted 1.15; only the schema label is changed to exercise the canonical compatibility branch.",
"provenance": "canonical_compatibility",
"version": "1.16"
}
],
"resume": {
@@ -163,6 +172,13 @@
"commit": "4720150fde047baa4e88b16082b282bee3a5e87d",
"fixture": "minimal/v1_15.json"
},
"1.16": {
"commit": "1c3b72019e547fe1cf1530419dc6fc687cc4df39",
"emitted_version": "1.15",
"fixture": "minimal/v1_16.json",
"note": "The schema transition introduced this reader version without a retained writer commit that emitted it. The recorded writer emitted 1.15; only the schema label is changed to exercise the canonical compatibility branch.",
"provenance": "canonical_compatibility"
},
"1.2": {
"commit": "74e8c1e22d7441bd42c58bcd4270937ccc2dca8c",
"fixture": "minimal/v1_2.json"
tests/test_run_context_wrapper.py
+99
View File
@@ -207,3 +207,102 @@ def test_tool_approval_item_preserves_positional_type_argument() -> None:
assert approval.type == "tool_approval_item"
assert approval.tool_name == "lookup_account"
assert approval.tool_namespace == "billing"
def test_exact_call_decisions_override_sticky_defaults() -> None:
agent = make_agent()
def approval(call_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item={
"type": "function_call",
"name": "tool_call",
"call_id": call_id,
"arguments": "{}",
},
)
approved: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
approved.approve_tool(approval("approve-sticky"), always_approve=True)
approved.reject_tool(approval("approve-exception"), rejection_message="denied by user")
assert approved.is_tool_approved("tool_call", "approve-exception") is False
assert approved.get_rejection_message("tool_call", "approve-exception") == "denied by user"
assert approved.is_tool_approved("tool_call", "approve-other") is True
rejected: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
rejected.reject_tool(
approval("reject-sticky"),
always_reject=True,
rejection_message="denied by default",
)
rejected.approve_tool(approval("reject-exception"))
assert rejected.is_tool_approved("tool_call", "reject-exception") is True
assert rejected.get_rejection_message("tool_call", "reject-exception") is None
assert rejected.is_tool_approved("tool_call", "reject-other") is False
assert rejected.get_rejection_message("tool_call", "reject-other") == "denied by default"
def test_matching_exact_call_decisions_preserve_sticky_defaults() -> None:
agent = make_agent()
def approval(call_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item={
"type": "function_call",
"name": "tool_call",
"call_id": call_id,
"arguments": "{}",
},
)
approved: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
approved.approve_tool(approval("approve-sticky"), always_approve=True)
approved.approve_tool(approval("approve-match"))
assert approved.is_tool_approved("tool_call", "approve-other") is True
rejected: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
rejected.reject_tool(approval("reject-sticky"), always_reject=True)
rejected.reject_tool(approval("reject-match"))
assert rejected.is_tool_approved("tool_call", "reject-other") is False
def test_exact_call_reversals_keep_other_calls_on_sticky_default() -> None:
agent = make_agent()
def approval(call_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item={
"type": "function_call",
"name": "tool_call",
"call_id": call_id,
"arguments": "{}",
},
)
approved: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
approved.approve_tool(approval("approve-sticky"), always_approve=True)
approved.reject_tool(approval("approve-exception"), rejection_message="denied")
approved.approve_tool(approval("approve-exception"))
assert approved.is_tool_approved("tool_call", "approve-exception") is True
assert approved.get_rejection_message("tool_call", "approve-exception") is None
assert approved.is_tool_approved("tool_call", "approve-other") is True
rejected: RunContextWrapper[dict[str, object]] = RunContextWrapper(context={})
rejected.reject_tool(
approval("reject-sticky"),
always_reject=True,
rejection_message="denied by default",
)
rejected.approve_tool(approval("reject-exception"))
rejected.reject_tool(approval("reject-exception"), rejection_message="denied exactly")
assert rejected.is_tool_approved("tool_call", "reject-exception") is False
assert rejected.get_rejection_message("tool_call", "reject-exception") == "denied exactly"
assert rejected.is_tool_approved("tool_call", "reject-other") is False
assert rejected.get_rejection_message("tool_call", "reject-other") == "denied by default"
tests/test_run_state.py
+201
View File
@@ -2967,6 +2967,89 @@ class TestRunState:
assert new_state._context.is_tool_approved(tool_name="tool2", call_id="cid2") is False
assert new_state._context.get_rejection_message("tool2", "cid2") is None
@pytest.mark.parametrize("sticky_approved", [True, False], ids=["approve", "reject"])
async def test_exact_call_override_round_trips_with_sticky_default(
self,
sticky_approved: bool,
) -> None:
"""A current snapshot preserves an exact exception and the sticky default."""
context: RunContextWrapper[dict[str, str]] = RunContextWrapper(context={})
agent = Agent(name="MixedApprovalAgent")
state = make_state(agent, context=context, original_input="test")
def approval(call_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item=ResponseFunctionToolCall(
type="function_call",
name="tool1",
call_id=call_id,
status="completed",
arguments="{}",
),
)
if sticky_approved:
state.approve(approval("sticky"), always_approve=True)
state.reject(approval("exception"), rejection_message="denied exactly")
else:
state.reject(
approval("sticky"),
always_reject=True,
rejection_message="denied by default",
)
state.approve(approval("exception"))
serialized = state.to_json()
assert serialized["$schemaVersion"] == "1.16"
restored = await RunState.from_json(agent, serialized)
assert restored._context is not None
expected_exact = not sticky_approved
assert restored._context.is_tool_approved("tool1", "exception") is expected_exact
assert restored._context.is_tool_approved("tool1", "other") is sticky_approved
assert restored._context.get_rejection_message("tool1", "exception") == (
"denied exactly" if sticky_approved else None
)
@pytest.mark.parametrize("sticky_approved", [True, False], ids=["approve", "reject"])
async def test_schema_1_15_mixed_approval_record_keeps_exact_decision(
self,
sticky_approved: bool,
) -> None:
"""An explicit decision in a legacy snapshot remains authoritative."""
context: RunContextWrapper[dict[str, str]] = RunContextWrapper(context={})
agent = Agent(name="LegacyMixedApprovalAgent")
state = make_state(agent, context=context, original_input="test")
def approval(call_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item=ResponseFunctionToolCall(
type="function_call",
name="tool1",
call_id=call_id,
status="completed",
arguments="{}",
),
)
if sticky_approved:
state.approve(approval("sticky"), always_approve=True)
state.reject(approval("exception"), rejection_message="denied exactly")
else:
state.reject(approval("sticky"), always_reject=True)
state.approve(approval("exception"))
serialized = state.to_json()
serialized["$schemaVersion"] = "1.15"
restored = await RunState.from_json(agent, serialized)
assert restored._context is not None
expected_exact = not sticky_approved
assert restored._context.is_tool_approved("tool1", "exception") is expected_exact
assert restored._context.is_tool_approved("tool1", "other") is sticky_approved
async def test_schema_1_13_restores_pending_approval_binding_from_interruption(self):
"""A 1.13 snapshot may resume only the exact invocation that was approved."""
agent = Agent(name="ApprovalLegacyAgent")
@@ -6344,6 +6427,74 @@ class TestDeserializeHelpers:
class TestRunStateResumption:
"""Test resuming runs from RunState using Runner.run()."""
@pytest.mark.parametrize("streamed", [False, True], ids=["run", "run_streamed"])
@pytest.mark.parametrize("sticky_approved", [True, False], ids=["approve", "reject"])
@pytest.mark.asyncio
async def test_resume_executes_only_exact_override_result(
self,
streamed: bool,
sticky_approved: bool,
) -> None:
"""Public resume paths execute only calls authorized by mixed decisions."""
model = ScriptedModel()
executions: list[str] = []
@function_tool(needs_approval=True)
async def approval_tool(value: str) -> str:
executions.append(value)
return f"approved:{value}"
agent = Agent(name="MixedApprovalAgent", model=model, tools=[approval_tool])
model.extend(
[
[
get_function_tool_call(
"approval_tool",
json.dumps({"value": "sticky"}),
call_id="sticky-call",
),
get_function_tool_call(
"approval_tool",
json.dumps({"value": "exception"}),
call_id="exception-call",
),
],
[get_final_output_message("done")],
]
)
initial = await Runner.run(agent, "start")
state = initial.to_state()
interruptions = {
cast(str, interruption.raw_item.call_id): interruption
for interruption in state.get_interruptions()
}
if sticky_approved:
state.approve(interruptions["sticky-call"], always_approve=True)
state.reject(
interruptions["exception-call"],
rejection_message="denied exactly",
)
else:
state.reject(
interruptions["sticky-call"],
always_reject=True,
rejection_message="denied by default",
)
state.approve(interruptions["exception-call"])
restored = await RunState.from_string(agent, state.to_string())
if streamed:
resumed = Runner.run_streamed(agent, restored)
async for _ in resumed.stream_events():
pass
else:
resumed = await Runner.run(agent, restored)
assert resumed.final_output == "done"
assert resumed.interruptions == []
assert executions == (["sticky"] if sticky_approved else ["exception"])
@pytest.mark.asyncio
async def test_resume_from_run_state(self):
"""Test resuming a run from a RunState."""
@@ -8936,6 +9087,7 @@ class TestRunStateSerializationEdgeCases:
"1.12",
"1.13",
"1.14",
"1.15",
CURRENT_SCHEMA_VERSION,
}
)
@@ -11667,6 +11819,55 @@ async def test_hosted_mcp_approval_round_trip_uses_typed_identity_records() -> N
)
@pytest.mark.asyncio
async def test_hosted_mcp_exact_rejection_overrides_sticky_approval_after_round_trip() -> None:
agent = Agent(name="test")
context: RunContextWrapper[dict[str, str]] = RunContextWrapper(context={})
state = make_state(agent, context=context)
def approval(request_id: str) -> ToolApprovalItem:
return ToolApprovalItem(
agent=agent,
raw_item=McpApprovalRequest(
id=request_id,
type="mcp_approval_request",
arguments="{}",
name="lookup_account",
server_label="server-a",
),
)
state.approve(approval("sticky-request"), always_approve=True)
state.reject(approval("exception-request"), rejection_message="denied exactly")
restored = await RunState.from_json(agent, state.to_json())
assert restored._context is not None
assert (
restored._context.get_approval_status(
"lookup_account",
"exception-request",
existing_pending=approval("exception-request"),
)
is False
)
assert (
restored._context.get_rejection_message(
"lookup_account",
"exception-request",
existing_pending=approval("exception-request"),
)
== "denied exactly"
)
assert (
restored._context.get_approval_status(
"lookup_account",
"other-request",
existing_pending=approval("other-request"),
)
is True
)
@pytest.mark.asyncio
async def test_incomplete_hosted_mcp_query_cannot_create_approval_authority() -> None:
agent = Agent(name="test")