Compare commits

...

6 Commits

Author SHA1 Message Date
Ahmed Ibrahim
a1ef35b7b0 Format Python SDK dangerous bypass updates 2026-05-16 09:49:22 -07:00
Ahmed Ibrahim
d8ea048b5e Fix Python SDK approvals import order 2026-05-16 09:49:22 -07:00
Ahmed Ibrahim
79a9adbf44 Fix dangerous bypass integration coverage 2026-05-16 09:49:22 -07:00
Ahmed Ibrahim
3cc8e7b30f Format dangerous bypass SDK updates 2026-05-16 09:49:22 -07:00
Ahmed Ibrahim
2eb61fa741 Add Python SDK dangerous bypass mode 2026-05-16 09:49:21 -07:00
Ahmed Ibrahim
a280248021 [codex] Split Python SDK helper logic (#22939)
## Summary
- Move approval-mode mapping into
`sdk/python/src/openai_codex/_approval_mode.py`.
- Move initialize metadata parsing and normalization into
`sdk/python/src/openai_codex/_initialize_metadata.py`.
- Keep the public `ApprovalMode` export stable and retarget direct
metadata helper coverage.

## Integration coverage
- Add an app-server harness smoke that exercises sync and async SDK
initialization plus thread creation.

## Validation
- Local tests were not run per repo guidance. CI should validate this
branch once the PR is online.
2026-05-16 09:47:51 -07:00
10 changed files with 528 additions and 104 deletions

View File

@@ -3,7 +3,7 @@
Public surface of `openai_codex` for app-server v2.
This SDK surface is experimental. Turn streams are routed by turn ID so one client can consume multiple active turns concurrently.
Thread and turn starts expose `approval_mode`. `ApprovalMode.auto_review` is the default; use `ApprovalMode.deny_all` to deny escalated permissions.
Thread and turn starts expose `approval_mode`. `ApprovalMode.auto_review` is the default; use `ApprovalMode.deny_all` to deny escalated permissions. `ApprovalMode.dangerously_bypass_approvals_and_sandbox` disables approvals and requests full sandbox bypass. That explicit bypass mode cannot be combined with `sandbox` on thread APIs or `sandbox_policy` on turn APIs.
## Package Entry

View File

@@ -867,6 +867,21 @@ def _approval_mode_model_arg_lines(*, indent: str = " ") -> list[str]
]
def _approval_mode_thread_sandbox_line(*, indent: str = " ") -> str:
"""Return the approval-mode sandbox preset for thread operations."""
return f"{indent}sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)"
def _approval_mode_turn_sandbox_policy_lines(*, indent: str = " ") -> list[str]:
"""Return the approval-mode sandbox preset for turn operations."""
return [
f"{indent}sandbox_policy = _turn_sandbox_policy_for_approval_mode(",
f"{indent} approval_mode,",
f"{indent} sandbox_policy,",
f"{indent})",
]
def _model_arg_lines(fields: list[PublicFieldSpec], *, indent: str = " ") -> list[str]:
return [f"{indent}{field.wire_name}={field.py_name}," for field in fields]
@@ -896,6 +911,7 @@ def _render_codex_block(
*_kw_signature_lines(thread_start_fields),
" ) -> Thread:",
_approval_mode_assignment_line("_approval_mode_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadStartParams(",
*_approval_mode_model_arg_lines(),
*_model_arg_lines(thread_start_fields),
@@ -921,6 +937,7 @@ def _render_codex_block(
*_kw_signature_lines(resume_fields),
" ) -> Thread:",
_approval_mode_assignment_line("_approval_mode_override_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadResumeParams(",
" thread_id=thread_id,",
*_approval_mode_model_arg_lines(),
@@ -937,6 +954,7 @@ def _render_codex_block(
*_kw_signature_lines(fork_fields),
" ) -> Thread:",
_approval_mode_assignment_line("_approval_mode_override_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadForkParams(",
" thread_id=thread_id,",
*_approval_mode_model_arg_lines(),
@@ -970,6 +988,7 @@ def _render_async_codex_block(
" ) -> AsyncThread:",
" await self._ensure_initialized()",
_approval_mode_assignment_line("_approval_mode_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadStartParams(",
*_approval_mode_model_arg_lines(),
*_model_arg_lines(thread_start_fields),
@@ -997,6 +1016,7 @@ def _render_async_codex_block(
" ) -> AsyncThread:",
" await self._ensure_initialized()",
_approval_mode_assignment_line("_approval_mode_override_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadResumeParams(",
" thread_id=thread_id,",
*_approval_mode_model_arg_lines(),
@@ -1014,6 +1034,7 @@ def _render_async_codex_block(
" ) -> AsyncThread:",
" await self._ensure_initialized()",
_approval_mode_assignment_line("_approval_mode_override_settings"),
_approval_mode_thread_sandbox_line(),
" params = ThreadForkParams(",
" thread_id=thread_id,",
*_approval_mode_model_arg_lines(),
@@ -1047,6 +1068,7 @@ def _render_thread_block(
" ) -> TurnHandle:",
" wire_input = _to_wire_input(input)",
_approval_mode_assignment_line("_approval_mode_override_settings"),
*_approval_mode_turn_sandbox_policy_lines(),
" params = TurnStartParams(",
" thread_id=self.id,",
" input=wire_input,",
@@ -1073,6 +1095,7 @@ def _render_async_thread_block(
" await self._codex._ensure_initialized()",
" wire_input = _to_wire_input(input)",
_approval_mode_assignment_line("_approval_mode_override_settings"),
*_approval_mode_turn_sandbox_policy_lines(),
" params = TurnStartParams(",
" thread_id=self.id,",
" input=wire_input,",

View File

@@ -0,0 +1,54 @@
from __future__ import annotations
from enum import Enum
from typing import NoReturn
from .generated.v2_all import (
ApprovalsReviewer,
AskForApproval,
AskForApprovalValue,
)
class ApprovalMode(str, Enum):
"""High-level approval behavior for escalated permission requests."""
deny_all = "deny_all"
auto_review = "auto_review"
dangerously_bypass_approvals_and_sandbox = "dangerously_bypass_approvals_and_sandbox"
def _approval_mode_settings(
approval_mode: ApprovalMode,
) -> tuple[AskForApproval, ApprovalsReviewer | None]:
"""Map the public approval mode to generated app-server start params."""
if not isinstance(approval_mode, ApprovalMode):
supported = ", ".join(mode.value for mode in ApprovalMode)
raise ValueError(f"approval_mode must be one of: {supported}")
match approval_mode:
case ApprovalMode.auto_review:
return (
AskForApproval(root=AskForApprovalValue.on_request),
ApprovalsReviewer.auto_review,
)
case ApprovalMode.deny_all:
return AskForApproval(root=AskForApprovalValue.never), None
case ApprovalMode.dangerously_bypass_approvals_and_sandbox:
return AskForApproval(root=AskForApprovalValue.never), None
case _:
return _assert_never_approval_mode(approval_mode)
def _assert_never_approval_mode(approval_mode: NoReturn) -> NoReturn:
"""Make approval mode mapping exhaustive for static type checkers."""
raise AssertionError(f"Unhandled approval mode: {approval_mode!r}")
def _approval_mode_override_settings(
approval_mode: ApprovalMode | None,
) -> tuple[AskForApproval | None, ApprovalsReviewer | None]:
"""Map an optional public approval mode to app-server override params."""
if approval_mode is None:
return None, None
return _approval_mode_settings(approval_mode)

View File

@@ -0,0 +1,54 @@
from __future__ import annotations
from .models import InitializeResponse, ServerInfo
def _split_user_agent(user_agent: str) -> tuple[str | None, str | None]:
raw = user_agent.strip()
if not raw:
return None, None
if "/" in raw:
name, version = raw.split("/", 1)
return (name or None), (version or None)
parts = raw.split(maxsplit=1)
if len(parts) == 2:
return parts[0], parts[1]
return raw, None
def validate_initialize_metadata(payload: InitializeResponse) -> InitializeResponse:
user_agent = (payload.userAgent or "").strip()
server = payload.serverInfo
server_name: str | None = None
server_version: str | None = None
if server is not None:
server_name = (server.name or "").strip() or None
server_version = (server.version or "").strip() or None
if (server_name is None or server_version is None) and user_agent:
parsed_name, parsed_version = _split_user_agent(user_agent)
if server_name is None:
server_name = parsed_name
if server_version is None:
server_version = parsed_version
normalized_server_name = (server_name or "").strip()
normalized_server_version = (server_version or "").strip()
if not user_agent or not normalized_server_name or not normalized_server_version:
raise RuntimeError(
"initialize response missing required metadata "
f"(user_agent={user_agent!r}, server_name={normalized_server_name!r}, server_version={normalized_server_version!r})"
)
if server is None:
payload.serverInfo = ServerInfo(
name=normalized_server_name,
version=normalized_server_version,
)
else:
server.name = normalized_server_name
server.version = normalized_server_version
return payload

View File

@@ -2,9 +2,14 @@ from __future__ import annotations
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import AsyncIterator, Iterator, NoReturn
from typing import AsyncIterator, Iterator
from ._approval_mode import (
ApprovalMode as ApprovalMode,
_approval_mode_override_settings,
_approval_mode_settings,
)
from ._initialize_metadata import validate_initialize_metadata
from ._inputs import (
ImageInput as ImageInput,
Input,
@@ -25,9 +30,7 @@ from ._run import (
from .async_client import AsyncAppServerClient
from .client import AppServerClient, AppServerConfig
from .generated.v2_all import (
ApprovalsReviewer,
AskForApproval,
AskForApprovalValue,
DangerFullAccessSandboxPolicy,
ModelListResponse,
Personality,
ReasoningEffort,
@@ -55,61 +58,31 @@ from .generated.v2_all import (
TurnStartParams,
TurnSteerResponse,
)
from .models import InitializeResponse, JsonObject, Notification, ServerInfo
from .models import InitializeResponse, JsonObject, Notification
def _split_user_agent(user_agent: str) -> tuple[str | None, str | None]:
raw = user_agent.strip()
if not raw:
return None, None
if "/" in raw:
name, version = raw.split("/", 1)
return (name or None), (version or None)
parts = raw.split(maxsplit=1)
if len(parts) == 2:
return parts[0], parts[1]
return raw, None
class ApprovalMode(str, Enum):
"""High-level approval behavior for escalated permission requests."""
deny_all = "deny_all"
auto_review = "auto_review"
def _approval_mode_settings(
approval_mode: ApprovalMode,
) -> tuple[AskForApproval, ApprovalsReviewer | None]:
"""Map the public approval mode to generated app-server start params."""
if not isinstance(approval_mode, ApprovalMode):
supported = ", ".join(mode.value for mode in ApprovalMode)
raise ValueError(f"approval_mode must be one of: {supported}")
match approval_mode:
case ApprovalMode.auto_review:
return (
AskForApproval(root=AskForApprovalValue.on_request),
ApprovalsReviewer.auto_review,
)
case ApprovalMode.deny_all:
return AskForApproval(root=AskForApprovalValue.never), None
case _:
return _assert_never_approval_mode(approval_mode)
def _assert_never_approval_mode(approval_mode: NoReturn) -> NoReturn:
"""Make approval mode mapping exhaustive for static type checkers."""
raise AssertionError(f"Unhandled approval mode: {approval_mode!r}")
def _approval_mode_override_settings(
def _thread_sandbox_for_approval_mode(
approval_mode: ApprovalMode | None,
) -> tuple[AskForApproval | None, ApprovalsReviewer | None]:
"""Map an optional public approval mode to app-server override params."""
if approval_mode is None:
return None, None
return _approval_mode_settings(approval_mode)
sandbox: SandboxMode | None,
) -> SandboxMode | None:
"""Apply approval-mode sandbox presets for thread operations."""
if approval_mode is not ApprovalMode.dangerously_bypass_approvals_and_sandbox:
return sandbox
if sandbox is not None:
raise ValueError("dangerous bypass approval_mode cannot be combined with sandbox")
return SandboxMode.danger_full_access
def _turn_sandbox_policy_for_approval_mode(
approval_mode: ApprovalMode | None,
sandbox_policy: SandboxPolicy | None,
) -> SandboxPolicy | None:
"""Apply approval-mode sandbox presets for turn operations."""
if approval_mode is not ApprovalMode.dangerously_bypass_approvals_and_sandbox:
return sandbox_policy
if sandbox_policy is not None:
raise ValueError("dangerous bypass approval_mode cannot be combined with sandbox_policy")
return SandboxPolicy(root=DangerFullAccessSandboxPolicy(type="dangerFullAccess"))
class Codex:
@@ -119,7 +92,7 @@ class Codex:
self._client = AppServerClient(config=config)
try:
self._client.start()
self._init = self._validate_initialize(self._client.initialize())
self._init = validate_initialize_metadata(self._client.initialize())
except Exception:
self._client.close()
raise
@@ -130,44 +103,6 @@ class Codex:
def __exit__(self, _exc_type, _exc, _tb) -> None:
self.close()
@staticmethod
def _validate_initialize(payload: InitializeResponse) -> InitializeResponse:
user_agent = (payload.userAgent or "").strip()
server = payload.serverInfo
server_name: str | None = None
server_version: str | None = None
if server is not None:
server_name = (server.name or "").strip() or None
server_version = (server.version or "").strip() or None
if (server_name is None or server_version is None) and user_agent:
parsed_name, parsed_version = _split_user_agent(user_agent)
if server_name is None:
server_name = parsed_name
if server_version is None:
server_version = parsed_version
normalized_server_name = (server_name or "").strip()
normalized_server_version = (server_version or "").strip()
if not user_agent or not normalized_server_name or not normalized_server_version:
raise RuntimeError(
"initialize response missing required metadata "
f"(user_agent={user_agent!r}, server_name={normalized_server_name!r}, server_version={normalized_server_version!r})"
)
if server is None:
payload.serverInfo = ServerInfo(
name=normalized_server_name,
version=normalized_server_version,
)
else:
server.name = normalized_server_name
server.version = normalized_server_version
return payload
@property
def metadata(self) -> InitializeResponse:
return self._init
@@ -195,6 +130,7 @@ class Codex:
thread_source: ThreadSource | None = None,
) -> Thread:
approval_policy, approvals_reviewer = _approval_mode_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadStartParams(
approval_policy=approval_policy,
approvals_reviewer=approvals_reviewer,
@@ -259,6 +195,7 @@ class Codex:
service_tier: str | None = None,
) -> Thread:
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadResumeParams(
thread_id=thread_id,
approval_policy=approval_policy,
@@ -293,6 +230,7 @@ class Codex:
thread_source: ThreadSource | None = None,
) -> Thread:
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadForkParams(
thread_id=thread_id,
approval_policy=approval_policy,
@@ -354,7 +292,7 @@ class AsyncCodex:
try:
await self._client.start()
payload = await self._client.initialize()
self._init = Codex._validate_initialize(payload)
self._init = validate_initialize_metadata(payload)
self._initialized = True
except Exception:
await self._client.close()
@@ -397,6 +335,7 @@ class AsyncCodex:
) -> AsyncThread:
await self._ensure_initialized()
approval_policy, approvals_reviewer = _approval_mode_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadStartParams(
approval_policy=approval_policy,
approvals_reviewer=approvals_reviewer,
@@ -463,6 +402,7 @@ class AsyncCodex:
) -> AsyncThread:
await self._ensure_initialized()
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadResumeParams(
thread_id=thread_id,
approval_policy=approval_policy,
@@ -498,6 +438,7 @@ class AsyncCodex:
) -> AsyncThread:
await self._ensure_initialized()
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox = _thread_sandbox_for_approval_mode(approval_mode, sandbox)
params = ThreadForkParams(
thread_id=thread_id,
approval_policy=approval_policy,
@@ -586,6 +527,10 @@ class Thread:
) -> TurnHandle:
wire_input = _to_wire_input(input)
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox_policy = _turn_sandbox_policy_for_approval_mode(
approval_mode,
sandbox_policy,
)
params = TurnStartParams(
thread_id=self.id,
input=wire_input,
@@ -670,6 +615,10 @@ class AsyncThread:
await self._codex._ensure_initialized()
wire_input = _to_wire_input(input)
approval_policy, approvals_reviewer = _approval_mode_override_settings(approval_mode)
sandbox_policy = _turn_sandbox_policy_for_approval_mode(
approval_mode,
sandbox_policy,
)
params = TurnStartParams(
thread_id=self.id,
input=wire_input,

View File

@@ -98,6 +98,11 @@ def response_approval_policy(response: Any) -> str:
return response.model_dump(by_alias=True, mode="json")["approvalPolicy"]
def response_sandbox_type(response: Any) -> str:
"""Return serialized sandbox policy type from a generated thread response."""
return response.model_dump(by_alias=True, mode="json")["sandbox"]["type"]
def agent_message_texts(events: list[Notification]) -> list[str]:
"""Extract completed agent-message text from SDK notifications."""
texts: list[str] = []

View File

@@ -1,12 +1,30 @@
from __future__ import annotations
import asyncio
import json
import shlex
from app_server_harness import AppServerHarness
from app_server_helpers import response_approval_policy
import pytest
from app_server_harness import (
AppServerHarness,
ev_completed,
ev_function_call,
ev_response_created,
sse,
)
from app_server_helpers import response_approval_policy, response_sandbox_type
from openai_codex import ApprovalMode, AsyncCodex, Codex
from openai_codex.generated.v2_all import AskForApprovalValue, ThreadResumeParams
from openai_codex.generated.v2_all import (
AskForApprovalValue,
DangerFullAccessSandboxPolicy,
ReadOnlySandboxPolicy,
SandboxMode,
SandboxPolicy,
ThreadResumeParams,
)
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE = DangerFullAccessSandboxPolicy(type="dangerFullAccess").type
def test_thread_resume_inherits_deny_all_approval_mode(tmp_path) -> None:
@@ -86,6 +104,261 @@ def test_thread_fork_can_override_approval_mode(tmp_path) -> None:
}
def test_dangerous_bypass_thread_lifecycle_persists_thread_settings(
tmp_path,
) -> None:
"""Thread lifecycle operations should preserve the explicit bypass preset."""
with AppServerHarness(tmp_path) as harness:
harness.responses.enqueue_assistant_message(
"bypass seeded",
response_id="bypass-thread",
)
with Codex(config=harness.app_server_config()) as codex:
source = codex.thread_start(
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
result = source.run("seed the bypass thread")
started_state = codex._client.thread_resume( # noqa: SLF001
source.id,
ThreadResumeParams(thread_id=source.id),
)
resumed = codex.thread_resume(
source.id,
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
resumed_state = codex._client.thread_resume( # noqa: SLF001
resumed.id,
ThreadResumeParams(thread_id=resumed.id),
)
forked = codex.thread_fork(
source.id,
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
forked_state = codex._client.thread_resume( # noqa: SLF001
forked.id,
ThreadResumeParams(thread_id=forked.id),
)
assert {
"final_response": result.final_response,
"forked_is_distinct": forked.id != source.id,
"started": (
response_approval_policy(started_state),
response_sandbox_type(started_state),
),
"resumed": (
response_approval_policy(resumed_state),
response_sandbox_type(resumed_state),
),
"forked": (
response_approval_policy(forked_state),
response_sandbox_type(forked_state),
),
} == {
"final_response": "bypass seeded",
"forked_is_distinct": True,
"started": (
AskForApprovalValue.never.value,
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE,
),
"resumed": (
AskForApprovalValue.never.value,
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE,
),
"forked": (
AskForApprovalValue.never.value,
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE,
),
}
def test_turn_dangerous_bypass_persists_thread_settings(tmp_path) -> None:
"""Turn-level bypass should persist approvals disabled and sandbox bypassed."""
with AppServerHarness(tmp_path) as harness:
harness.responses.enqueue_assistant_message(
"turn bypass",
response_id="bypass-turn",
)
with Codex(config=harness.app_server_config()) as codex:
thread = codex.thread_start(approval_mode=ApprovalMode.auto_review)
result = thread.run(
"bypass this turn",
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
after_turn = codex._client.thread_resume( # noqa: SLF001
thread.id,
ThreadResumeParams(thread_id=thread.id),
)
assert {
"final_response": result.final_response,
"thread_settings": (
response_approval_policy(after_turn),
response_sandbox_type(after_turn),
),
} == {
"final_response": "turn bypass",
"thread_settings": (
AskForApprovalValue.never.value,
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE,
),
}
def test_async_turn_dangerous_bypass_persists_thread_settings(tmp_path) -> None:
"""Async turn-level bypass should persist the same app-server settings."""
async def scenario() -> None:
with AppServerHarness(tmp_path) as harness:
harness.responses.enqueue_assistant_message(
"async turn bypass",
response_id="async-bypass-turn",
)
async with AsyncCodex(config=harness.app_server_config()) as codex:
thread = await codex.thread_start(approval_mode=ApprovalMode.auto_review)
result = await thread.run(
"bypass this async turn",
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
after_turn = await codex._client.thread_resume( # noqa: SLF001
thread.id,
ThreadResumeParams(thread_id=thread.id),
)
assert {
"final_response": result.final_response,
"thread_settings": (
response_approval_policy(after_turn),
response_sandbox_type(after_turn),
),
} == {
"final_response": "async turn bypass",
"thread_settings": (
AskForApprovalValue.never.value,
DANGER_FULL_ACCESS_SANDBOX_POLICY_TYPE,
),
}
asyncio.run(scenario())
def test_outside_workspace_write_rejected_for_deny_all_and_allowed_for_bypass(
tmp_path,
) -> None:
"""Dangerous bypass should be the mode that permits outside-workspace writes."""
rejected_path = tmp_path / "deny-all-outside-write.txt"
allowed_path = tmp_path / "dangerous-outside-write.txt"
with AppServerHarness(tmp_path) as harness:
rejected_args = json.dumps(
{
"command": (f"printf %s rejected > {shlex.quote(str(rejected_path))}"),
"login": False,
"timeout_ms": 1_000,
}
)
dangerous_args = json.dumps(
{
"command": (f"printf %s dangerous > {shlex.quote(str(allowed_path))}"),
"login": False,
"timeout_ms": 1_000,
}
)
harness.responses.enqueue_sse(
sse(
[
ev_response_created("deny-all-write"),
ev_function_call(
"deny-all-outside-write",
"shell_command",
rejected_args,
),
ev_completed("deny-all-write"),
]
)
)
harness.responses.enqueue_assistant_message(
"deny-all shell completed",
response_id="deny-all-final",
)
harness.responses.enqueue_sse(
sse(
[
ev_response_created("dangerous-write"),
ev_function_call(
"dangerous-outside-write",
"shell_command",
dangerous_args,
),
ev_completed("dangerous-write"),
]
)
)
harness.responses.enqueue_assistant_message(
"dangerous shell completed",
response_id="dangerous-final",
)
with Codex(config=harness.app_server_config()) as codex:
denied_thread = codex.thread_start(approval_mode=ApprovalMode.deny_all)
denied_result = denied_thread.run("write outside the workspace")
bypass_thread = codex.thread_start(
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
)
bypass_result = bypass_thread.run("write outside the workspace")
assert {
"denied_final_response": denied_result.final_response,
"denied_path_exists": rejected_path.exists(),
"bypass_final_response": bypass_result.final_response,
"bypass_file_contents": allowed_path.read_text(),
} == {
"denied_final_response": "deny-all shell completed",
"denied_path_exists": False,
"bypass_final_response": "dangerous shell completed",
"bypass_file_contents": "dangerous",
}
def test_dangerous_bypass_rejects_explicit_sandbox_conflicts_before_state_changes(
tmp_path,
) -> None:
"""Conflicting bypass presets should fail before mutating app-server state."""
with AppServerHarness(tmp_path) as harness:
with Codex(config=harness.app_server_config()) as codex:
with pytest.raises(ValueError, match="combined with sandbox"):
codex.thread_start(
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
sandbox=SandboxMode.read_only,
)
threads_after_invalid_start = codex.thread_list(archived=False)
thread = codex.thread_start()
with pytest.raises(ValueError, match="combined with sandbox_policy"):
thread.run(
"this should never reach app-server",
approval_mode=ApprovalMode.dangerously_bypass_approvals_and_sandbox,
sandbox_policy=SandboxPolicy(root=ReadOnlySandboxPolicy(type="readOnly")),
)
thread_state = thread.read()
assert {
"threads_after_invalid_start": [
existing.id for existing in threads_after_invalid_start.data
],
"turns_after_invalid_run": thread_state.thread.turns,
} == {
"threads_after_invalid_start": [],
"turns_after_invalid_run": [],
}
def test_turn_approval_mode_persists_until_next_turn(tmp_path) -> None:
"""A turn-level approval override should apply to later omitted-arg turns."""
with AppServerHarness(tmp_path) as harness:

View File

@@ -39,6 +39,61 @@ def test_thread_set_name_and_read(tmp_path) -> None:
}
def test_sync_and_async_initialization_round_trip_metadata(tmp_path) -> None:
"""Public clients should initialize and start threads through app-server."""
async def async_scenario(harness: AppServerHarness) -> dict[str, object]:
async with AsyncCodex(config=harness.app_server_config()) as codex:
thread = await codex.thread_start()
server = codex.metadata.serverInfo
return {
"thread_id": thread.id,
"user_agent": codex.metadata.userAgent,
"server_name": None if server is None else server.name,
"server_version": None if server is None else server.version,
}
with AppServerHarness(tmp_path) as harness:
with Codex(config=harness.app_server_config()) as codex:
thread = codex.thread_start()
server = codex.metadata.serverInfo
sync_summary = {
"thread_id": thread.id,
"user_agent": codex.metadata.userAgent,
"server_name": None if server is None else server.name,
"server_version": None if server is None else server.version,
}
async_summary = asyncio.run(async_scenario(harness))
assert {
"sync": {
"thread_id_present": bool(sync_summary["thread_id"]),
"user_agent_present": bool(sync_summary["user_agent"]),
"server_name_present": bool(sync_summary["server_name"]),
"server_version_present": bool(sync_summary["server_version"]),
},
"async": {
"thread_id_present": bool(async_summary["thread_id"]),
"user_agent_present": bool(async_summary["user_agent"]),
"server_name_present": bool(async_summary["server_name"]),
"server_version_present": bool(async_summary["server_version"]),
},
} == {
"sync": {
"thread_id_present": True,
"user_agent_present": True,
"server_name_present": True,
"server_version_present": True,
},
"async": {
"thread_id_present": True,
"user_agent_present": True,
"server_name_present": True,
"server_version_present": True,
},
}
def test_thread_list_filters_archived_threads(tmp_path) -> None:
"""Thread listing should reflect archive state through app-server."""
with AppServerHarness(tmp_path) as harness:

View File

@@ -149,12 +149,18 @@ def test_approval_modes_serialize_to_expected_start_params() -> None:
"approvalPolicy": "on-request",
"approvalsReviewer": "auto_review",
},
"dangerously_bypass_approvals_and_sandbox": {
"approvalPolicy": "never",
},
}
def test_unknown_approval_mode_is_rejected() -> None:
"""Invalid approval modes should fail before params are constructed."""
with pytest.raises(ValueError, match="deny_all, auto_review"):
with pytest.raises(
ValueError,
match="deny_all, auto_review, dangerously_bypass_approvals_and_sandbox",
):
public_api_module._approval_mode_settings("allow_all") # type: ignore[arg-type]

View File

@@ -18,6 +18,7 @@ from openai_codex import (
RunResult,
Thread,
)
from openai_codex._initialize_metadata import validate_initialize_metadata
from openai_codex.types import InitializeResponse
EXPECTED_ROOT_EXPORTS = [
@@ -128,6 +129,10 @@ def test_root_exports_approval_mode() -> None:
assert [(mode.name, mode.value) for mode in ApprovalMode] == [
("deny_all", "deny_all"),
("auto_review", "auto_review"),
(
"dangerously_bypass_approvals_and_sandbox",
"dangerously_bypass_approvals_and_sandbox",
),
]
@@ -444,7 +449,7 @@ def test_lifecycle_methods_are_codex_scoped() -> None:
def test_initialize_metadata_parses_user_agent_shape() -> None:
"""Initialize metadata should accept the legacy user-agent-only payload shape."""
payload = InitializeResponse.model_validate({"userAgent": "codex-cli/1.2.3"})
parsed = Codex._validate_initialize(payload)
parsed = validate_initialize_metadata(payload)
assert parsed is payload
assert parsed.userAgent == "codex-cli/1.2.3"
assert parsed.serverInfo is not None
@@ -455,7 +460,7 @@ def test_initialize_metadata_parses_user_agent_shape() -> None:
def test_initialize_metadata_requires_non_empty_information() -> None:
"""Initialize metadata should fail when the runtime gives no identity signal."""
try:
Codex._validate_initialize(InitializeResponse.model_validate({}))
validate_initialize_metadata(InitializeResponse.model_validate({}))
except RuntimeError as exc:
assert "missing required metadata" in str(exc)
else: