clone/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-05-18 10:12:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add high-level Python SDK approval mode

Browse Source 
Expose approval_mode with deny_all and auto_review options on the high-level Python SDK, and map those choices to generated app-server approval params internally.

Update examples, docs, notebooks, and public API tests to use the new mode instead of raw generated approval fields.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Ahmed Ibrahim
2026-05-10 11:44:34 +03:00
parent 7edbdc555c
commit ffe6e44a03
11 changed files with 314 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
sdk/python/tests/test_public_api_runtime_behavior.py
View File

@@ -20,6 +20,7 @@ from openai_codex.generated.v2_all import (
)
from openai_codex.models import InitializeResponse, Notification
from openai_codex.api import (
ApprovalMode,
AsyncCodex,
AsyncThread,
AsyncTurnHandle,
@@ -34,10 +35,18 @@ from openai_codex.types import AskForApproval
ROOT = Path(__file__).resolve().parents[1]
def _approval_policy_values(params: list[Any]) -> list[object]:
"""Return serialized approval policies from captured Pydantic params."""
def _approval_settings(params: list[Any]) -> list[dict[str, object]]:
"""Return serialized approval settings from captured Pydantic params."""
return [
param.model_dump(by_alias=True, mode="json").get("approvalPolicy")
{
key: value
for key, value in param.model_dump(
by_alias=True,
exclude_none=True,
mode="json",
).items()
if key in {"approvalPolicy", "approvalsReviewer"}
}
for param in params
]
@@ -255,8 +264,8 @@ def test_ask_for_approval_exposes_simple_policy_constants() -> None:
}
def test_sync_api_forces_approval_policy_never_for_started_work() -> None:
"""Sync start methods should send never until approval handling exists."""
def test_sync_api_maps_approval_modes_for_started_work() -> None:
"""Sync start methods should serialize only supported approval modes."""
captured: list[Any] = []
class FakeClient:
@@ -294,19 +303,33 @@ def test_sync_api_forces_approval_policy_never_for_started_work() -> None:
codex = object.__new__(Codex)
codex._client = client
codex.thread_start(approval_policy=AskForApproval.on_request)
codex.thread_resume("thread-1", approval_policy=AskForApproval.on_request)
codex.thread_fork("thread-1", approval_policy=AskForApproval.on_request)
codex.thread_start()
codex.thread_resume("thread-1")
codex.thread_fork("thread-1")
Thread(client, "thread-1").turn(TextInput("hello"))
codex.thread_start(approval_mode=ApprovalMode.auto_review)
codex.thread_resume("thread-1", approval_mode=ApprovalMode.auto_review)
codex.thread_fork("thread-1", approval_mode=ApprovalMode.auto_review)
Thread(client, "thread-1").turn(
TextInput("hello"),
approval_policy=AskForApproval.on_request,
approval_mode=ApprovalMode.auto_review,
)
assert _approval_policy_values(captured) == ["never", "never", "never", "never"]
assert _approval_settings(captured) == [
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
]
def test_async_api_forces_approval_policy_never_for_started_work() -> None:
"""Async start methods should send never until approval handling exists."""
def test_async_api_maps_approval_modes_for_started_work() -> None:
"""Async start methods should serialize only supported approval modes."""
async def scenario() -> None:
"""Exercise the async wrappers without spawning a real app server."""
captured: list[Any] = []
@@ -346,19 +369,30 @@ def test_async_api_forces_approval_policy_never_for_started_work() -> None:
codex._client = FakeAsyncClient()
codex._initialized = True
await codex.thread_start(approval_policy=AskForApproval.on_request)
await codex.thread_resume("thread-1", approval_policy=AskForApproval.on_request)
await codex.thread_fork("thread-1", approval_policy=AskForApproval.on_request)
await codex.thread_start()
await codex.thread_resume("thread-1")
await codex.thread_fork("thread-1")
await AsyncThread(codex, "thread-1").turn(TextInput("hello"))
await codex.thread_start(approval_mode=ApprovalMode.auto_review)
await codex.thread_resume(
"thread-1",
approval_mode=ApprovalMode.auto_review,
)
await codex.thread_fork("thread-1", approval_mode=ApprovalMode.auto_review)
await AsyncThread(codex, "thread-1").turn(
TextInput("hello"),
approval_policy=AskForApproval.on_request,
approval_mode=ApprovalMode.auto_review,
)
assert _approval_policy_values(captured) == [
"never",
"never",
"never",
"never",
assert _approval_settings(captured) == [
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "never"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"},
]
asyncio.run(scenario())
@@ -397,6 +431,7 @@ def test_turn_streams_can_consume_multiple_turns_on_one_client() -> None:
def test_async_turn_streams_can_consume_multiple_turns_on_one_client() -> None:
"""Two async TurnHandle streams should advance independently on one client."""
async def scenario() -> None:
"""Interleave two async streams backed by separate per-turn queues."""
codex = AsyncCodex()
@@ -479,14 +514,25 @@ def test_thread_run_accepts_string_input_and_returns_run_result() -> None:
client.turn_start = fake_turn_start # type: ignore[method-assign]
result = Thread(client, "thread-1").run("hello")
result = Thread(client, "thread-1").run(
"hello",
approval_mode=ApprovalMode.auto_review,
)
assert seen["thread_id"] == "thread-1"
assert seen["wire_input"] == [{"type": "text", "text": "hello"}]
assert result == RunResult(
final_response="Hello.",
items=[item_notification.payload.item],
usage=usage_notification.payload.token_usage,
assert (
seen["thread_id"],
seen["wire_input"],
_approval_settings([seen["params"]]),
result,
) == (
"thread-1",
[{"type": "text", "text": "hello"}],
[{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"}],
RunResult(
final_response="Hello.",
items=[item_notification.payload.item],
usage=usage_notification.payload.token_usage,
),
)
@@ -658,6 +704,7 @@ def test_stream_text_registers_and_consumes_turn_notifications() -> None:
def test_async_thread_run_accepts_string_input_and_returns_run_result() -> None:
"""Async Thread.run should normalize string input and collect routed results."""
async def scenario() -> None:
"""Feed item, usage, and completion events through the async turn stream."""
codex = AsyncCodex()
@@ -692,14 +739,25 @@ def test_async_thread_run_accepts_string_input_and_returns_run_result() -> None:
codex._client.turn_start = fake_turn_start # type: ignore[method-assign]
codex._client.next_turn_notification = fake_next_notification # type: ignore[method-assign]
result = await AsyncThread(codex, "thread-1").run("hello")
result = await AsyncThread(codex, "thread-1").run(
"hello",
approval_mode=ApprovalMode.auto_review,
)
assert seen["thread_id"] == "thread-1"
assert seen["wire_input"] == [{"type": "text", "text": "hello"}]
assert result == RunResult(
final_response="Hello async.",
items=[item_notification.payload.item],
usage=usage_notification.payload.token_usage,
assert (
seen["thread_id"],
seen["wire_input"],
_approval_settings([seen["params"]]),
result,
) == (
"thread-1",
[{"type": "text", "text": "hello"}],
[{"approvalPolicy": "on-request", "approvalsReviewer": "auto_review"}],
RunResult(
final_response="Hello async.",
items=[item_notification.payload.item],
usage=usage_notification.payload.token_usage,
),
)
asyncio.run(scenario())
@@ -709,6 +767,7 @@ def test_async_thread_run_uses_last_completed_assistant_message_as_final_respons
None
):
"""Async run should use the last final assistant message as the response text."""
async def scenario() -> None:
"""Feed two completed agent messages through the async per-turn stream."""
codex = AsyncCodex()
@@ -756,6 +815,7 @@ def test_async_thread_run_uses_last_completed_assistant_message_as_final_respons
def test_async_thread_run_returns_none_when_only_commentary_messages_complete() -> None:
"""Async Thread.run should ignore commentary-only messages for final text."""
async def scenario() -> None:
"""Feed a commentary item and completion through the async turn stream."""
codex = AsyncCodex()